Агентскі ІІ з LangChain — Частка 6: Агентскі RAG на базе MCP
6-я частка: агентныя AI з LangChain — агентныя RAG, якія працуюць за дапамою MCP, дзе інструменты для запошуку дакладнай інфармацыі выклікаюцца з сервера MCP, а LangGraph керуе прабегам агента.
Крыніцы: Agentic AI with LangChain — Part 6: MCP Powered Agentic RAG with LangGraph
У гэтым керавану практычна перакладзець шлях ад сыр'ёў да рабочай сістэмы для: Агентнага ІІ з LangChain — Частка 6: Агентны RAG на базе MCP з LangGraph. Увага сфокусавана на кроках, якія можна выконваць, чыстах пераконтраляваннях і кодзе, які можна проста дадзіць у рэпазітарый без неабяснення намеру. Для адгледжэння, перш чым зменяць код, неабходна візначыць вхідныя даны, адпаведальнага за крок і крытэрыя завершэння. Аператары должны магчымае перадзваніць крок з вядомай точкі контролю без неабяснення схованага стану. Спрыятлівайце гэтам этапу як кантракту межа вхіднымі данымі і падтверджанымі выходнымі рэзультатамі. Дайце назвы артыфактам, візначыце пераконтралявання успеху і адмовіцеся ад тыхоўага частковага завершэння.
Тэхнічныя концэпты
Калі вы працуеце над Тэхнічнымія концэптамі, спачатку запісайце умовы працы: неабяжлівыя даныя, сігнал успеху і тое, што выходзіць пад частыя неудачы. Такі спіс дапамагае заліцварваць можлівыя змены ў кодзе. Запісваюце час выканання і кост токена або запыту праза функцыйнальныя рэзултаты. Відразувая ведамасць пра косты, вы ухіліваецеся ад неспакою, калі праця пераходзіць з дэмовай асяроддзя ў спакульную. Перад налаштаваннем запрошэнняяў пераканайцеся, як сістэма адпаведзя на фіксованы набор запытаў. Частае змінэ запрошэнняяў рэдка калі-небудзь выправляе слабую якасць адналічэння інформацыі.
1. Канфігурацыі сістэмы
Калі працюеце над раздзелам 1. Налашчэнні сістэмы, спачатку запісайце умовы: неабяжлівыя даны, сігнал успеху і тое, што выходзіць на частым неудачам. Такі спіс дапамагае заліцварыць будучыя змены ў кодзе. Зберагаюце налашчэнні пазыркам ад коду прыемлі. Файлы сераўіса, хранільнікі секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаць адбавіць аўдыт без неабяжлівага чытання всей структуры. Перад налашчэннем запітоў пераканайцеся ў робочым ступені воспамінання на фіксаванай сэтке запытаў. Частае змены запытаў рэдка калі вярнуюць слабую эфектыўнасць адзысквання дадзеных.
OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>
class BaseConfig(BaseSettings):
OPENAI_API_KEY: Optional[str]
PINECONE_API_KEY: Optional[str]
TAVILY_API_KEY: Optional[str]
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
2. Структура праекта
Калі працюеце над раздзелам 2. Структура праекта, спачатку запісайце умовы кантракту: неабяжлівыя даны, сігнал успеху і тое, што выканаецца у разе частковага нявыполнення. Такі спіс пераканаецца зберагчы чыстасць пазнейшых змян у кодзе. Документавайце адночасна шлях успеху і шлях вярнення. Перапрыбуткі, людзкіе контралі і обработка некорэктных паведамленняў ёсцю часткай продукту, а не пазнейшым дапрацоўкам. Змяржывайце рэткасць адпаведзенняў на фіксованы набор запітаў прычымо да налаштавання підказак. Частае змена підказак рэдка калі вярнайце слабкую базу адзыскання інфармацыі.
project/
├── rag_process/ # Pinecone RAG operations
│ ├── __init__.py
│ ├── extractor.py
| └── service.py
│
├── graphs # implement ReAct and Reflection StateGraph
│ ├── nodes.py # graph nodes. Defined as Python functions
│ ├── graph.py # graph workflow for MCP in SES mode
│ ├── graph_stdio.py # graph workflow for MCP in STDIO mode
│ ├── consts.py # define node names as constants
│ └── state.py # typed dictionary to store graph states
│
├── mcp_client_sse.py # mcp client in sse mode
├── mcp_client_stdio.py # mcp client in stdio mode
├── server_sse.py # mcp server in sse mode
│── server_stdio.py # mcp server in stdio mode
|── schemas.py # pydantic data models
|── upload.py # python functions to upload files asynchronously
├── main.py # fastAPI app
├── config.py # pydantic BaseConfig
│── utils.py # utility extracting question/docs from message list
├── .env # environment variable definition
├── client.py # streamlit client
├── .gitignore
└── requirements.txt # package requirements
3. MCP (Model Context Protocol)
Калі працюеце над 3. MCP (Model Context Protocol), спачатку запісайце умовы вікорыстоўвання: неабходныя даны, сігнал успеху і тое, што выходзіць на падчасныя адказы. Такі чэк-ліст дапамагае заставіць пазнейшыя змены коду застаўацца чыстымі. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выходзіць, адказ пра аблуку павінен вказваць на адзіную адпаведальнасць, а не на заплутаны ланцюг задач. Зберагаеце у кэшы стабільныя інструкцыі сістэмы і схемы інструментаў. Перадзесланне ідэнтычных прамаўляючых частак — частая прычына зайвага навантажэння.
3.1. Канцэпцыя
Калі працюеце над раздзелам 3.1. Concept, спачатку запісайце угоду: неабяжлівыя даны, сігнал успеху і тое, што выходзіць у разе частковага невыпання. Такі спіс пераконвае ў тым, што пазнейшыя змены коду будуць чыстымі. Спрэчвайце гэты этап як угоду межа данымі і перакананымі выходамі. Дайце назвы элементам, задаце крытэрыя успеху і не падзволяйце частковаму завершэнню без паведамлення. Змяркуйце ступень вярнага адналёгчэння на фіксаванай сэтке запытанняў прычым перад налашоўкай запрасаў. Частыя змены запрасаў рэдка калі вядома падлегчаюць слабую працэздатнась ціх сістэм.
3.2 Архітэктура MCP
Калі працюеце над 3.2 Архітэктура MCP, спачатку запісайце умовы вярбунка: неабяжлівыя даныя, сігнал успеху і тое, што выходзіць пад частыя неудачы. Такі спіс дапамагае заліцьвараць будучыя змены коду. Запісуйце час выканання і кост токена або запыту праз сума рэзультатаў функцый. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі працэс пераходзіць з дэмавайнага рэжыма ў спяльныя асяроддзя. Замерайце рэткісць адказаў на фіксаваны набор запытаў прычым перад налаштаваннем підказак. Частыя змены підказак рэдка калі вядуць да павышэння якасці адказаў.
3.3. Способы транспорту MCP
Калі працюеце над раздзелам 3.3. MCP Transport Modes, спачатку запісайце умовы вярбунка: неабяжлівыя даныя, сігнал успеху і тое, што выходзіць пад частковыя адказы. Такі спіс дапамагае заліцварваць будучыя змены ў кодзе. Храніце настройкі парадульна ад коду прыемліка. Файлы сераўіснага сэрвісу, базы секрэтных даных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаць пераглядаць іх без неабяжлівага чытання всей структуры. Перад налаштаваннем запітоў пераканайцеся, што сістэма правільна адзначае адказы на фіксаваныя запітанні. Частае змены запітоў рэдка калі дапамагае паспрабаваць выправіць слабую эфектыўнасць пошуку.
3.3.1. STDIO Mode
Калі працюеце над раздзелам 3.3.1. Рэжым STDIO, спачатку запісайце угоду: неабяжлівыя даныя, сігнал працявання і тое, што выходзіць на частым неяксамоце. Такі спіс контроля дапамагае залічыць пазнейшыя змены ў кодзе чыстымі. Документавайце як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія контралі і обработка некорэктных паведамленняў ёсць часткай продукту, а не пазнейшым дапрацоўкам. Змяркуйце рэткасць адпаведных адказоў на фіксованы набор запітаў прыштоям налаштавання падказак. Частае змена падказак рэдка калі вярнайць слабкую эфектыўнасць адзысквання інформацыі. Калі працюеце над раздзелам 3.3.1. Рэжым STDIO, спачатку запісайце угоду: неабяжлівыя даныя, сігнал працявання і тое, што выходзіць на частым неяксамоце. Такі спіс контроля дапамагае залічыць пазнейшыя змены ў кодзе чыстымі. Спрыймайце гэты этап як угоду межа данымі і перакананымі выходнымі рэзультатамі. Дайце назвы элементам, задаць крытэрыя успеху і адмовіцеся ад мовчанкавага частковага завершэння.
3.3.2. Рэжым SSE
3.3.2. Рэжым SSE работае наяўней, калі яго спрыяваць як мерыемую паверхню. Запісаце адна ідэальная транскрыпцыя, адзін прыклад неудачы і прыметку па абратанні ранейшага стану, прычаму расшыраеце дыямэт працы. Запісвайце часы выканання і косты токеноў або запытак праза функцыйнае рэзультат. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі працэс пераходзіць з дэмаверсіі ў спяльныя асяроддзя. Раздзеляйце правілы часткавання дадзеных і правілы ўтрымання іх. Змена аднаго з яных не павінна вымагаць перапісвання другога, калі зменяюцца паказнікі якосці.
4. Адкалічэнне MCP у рэжыме STDIO
- Адаптаванне MCP у рэжыме STDIO дае найлепыя рэзультаты, калі яго рассматрываць як параметр, які можна вымерыць. Зберагчы адны ідэальны прыклад работы, адну сітуацыю неудачы і прыметкі па поверненню да пачатковага стану, перш чым расшырваць сферу прыемлівання. Конфігурацыю трэба знаходзіць за межамі коду прыемлі. Файлы сераўса, хранільнікі секрэтных даных і пазнакі функцый должны быць у адном месца, куда аператары можуць аўдытаваць іх, не чытаючы весь код. Палітыку разбівання на часткі трэба аддзельваць ад палітыкы выявлення даных. Змена адной з іх не должна вымагаць перапісвання другой, калі зменяюцыся паказнікі якосці.
4.1. Сервер MCP
4.1. MCP Server работае наяўней, калі яго спрыяваць як меравальную плошчу. Зафіксавайце адны ідэальны прыклад працы, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расшырваць сферу дзеяння. Документавайце як шлях успеху, так і шлях вяснавання проблем. Перапрыбуткі, людзкі контроль і обработка некорэктных паведамленняў ёсць часткай продукту, а не наступным етапам дорабкі. Раздзеляйце правілы частковай обработкі дадзеных і правілы ўтрымання інформацыі. Змена адных не павінна вымагаць перапісвання іншых, калі змянююцыся паказчыкі якосці. 4.1. MCP Server работае наяўней, калі яго спрыяваць як меравальную плошчу. Зафіксавайце адны ідэальны прыклад працы, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расшырваць сферу дзеяння. Спрыявайце гэты этап як даговор між вхіднымі дадзенымі і перакананымі выходнымі рэзультатамі. Даўце назвы артыфактам, задайце крытэрыя успеху і не падтрымвайце беззвучнае частковае завершэння працы.
# initialize an STDIO FastMCP instance
mcp = FastMCP("Search Server")
# create mcp tools
@mcp.tool()
async def search_agentic_docs(query: str) -> str:
"""
Good for questions about generative agents, prompt engineering, and adversarial attacks.
Pass a natural language search query to retrieve context from the agentic ai database.
"""
# Simply invoke your existing vector service retriever
docs = await vector_service.retriever.ainvoke(query)
# Flatten the document content into a single string for the LLM
return "\n\n".join([d.page_content for d in docs])
# start MCP server in stdio mode
if __name__ == "__main__":
mcp.run(transport="stdio")
4.2. MCP Кліент
Для версіі 4.2 кліента MCP неабяжна практыка адзначыць вхідныя даныя, адпаведальнага за практыку і крэтарыя выходу пры перадзеіснаванні коду. Аператары должны магчымаць перзапуск практыкі з вядомага пункта контролю, не прымуячы заўважэнняў на адкулькі схованы стан. Неабяжна фіксаваць час выкарыстоўвання, а таксама косць токенаў чы запытанняў праза рэзультаты функцыйнасці. Відразлівае адображэнне косцаў запобегае неспакойным рашчыткам, калі практыка пераходзіць з дэмавай версіі ў спяльныя сераўысы. Неабяжна цітаваць тыя часткі тэксту, якія фактычна лежалі в основе адпаведнага адказу. Без цітатаў аператары не могуць разлічыць галюцынацію ад прасоўкі ў індэксаванні.
# define location of server file
BASE_DIR = Path(__file__).parent
SERVER_SCRIPT = BASE_DIR / "server.py"
# Local server running directly via Python over stdio
SERVER_CONFIG = {
"local_search_server": {
"transport": "stdio",
"command": "python",
"args": [str(SERVER_SCRIPT)],
},
}
# create a MCP client and return
def get_mcp_client() :
"""Works for stdio, sse, or a mix of both!"""
client = MultiServerMCPClient(SERVER_CONFIG)
return client
4.3. Існаванне інструментаў MCP у агенце
Для пункта 4.3. Прыткайчыкаў MCP у агенце, перад змянай коду неабходна задаць вхідныя даны, адпаведальнага за крок і крытэрыя завершэння. Аператары должны магчымае перзапускіць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Конфігурацыю трэба залічыць праз аплявацыйскі код. Файлы сераўнавання, хранілішчы секрэтных дадзеных і флагі функцый должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць, не чытаяўшы весь граф. Паказваць трэба тыя часткі, якія фактычна ляглі в основу адпаведнай адказы. Без цых цітатаў аператары не можаць разлічыць галюцынацію ад працягу індэксавання.
from mcp_client_stdio import get_mcp_client
async def build_graph():
"""
build StateGraph
:return:
"""
# initialize a MCP client instance
client = get_mcp_client()
# retrieve MCP tools
mcp_tools = await client.get_tools()
# create an llM
llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools
# bind tools to ToolNode and llm
tool_node = ToolNode(tools)
workflow = StateGraph(GraphState)
llm_with_tools = llm.bind_tools(tools)
4.4. Апробаванне рабочага прайсепту STDIO
Для пункту 4.4. Прабава рабочага прытэпу STDIO, перад змянайом коду неабходна якія-то вхідныя даны, адміністратар шагу і крэтыяры завершэння. Аперацыйныя працавнікі должны магчымае запускаць шаг з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Неабходна аддакуваць дакументацыю як пра успішны, так і пра патэнтны шляхы адрабоўкі. Перапрыбуткі, людзкія перакрыцця і апрацоўка некорэктных паведамленняў ёсць часткай продукту, а не чымсь, што дадаецца пазней. Паказваць часткі тэксту, якія фактычна лежаць у падставе адпаведнай адказы. Без ціх цытатаў аперацыйныя працавнікі не зможуць адразліць галюцынацыю ад працягу індэксавання. Для пункту 4.4. Прабава рабочага прытэпу STDIO, перад змянайом коду неабходна якія-то вхідныя даны, адміністратар шагу і крэтыяры завершэння. Аперацыйныя працавнікі должны магчымае запускаць шаг з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Спрыяць гэтаму этапу як кантракту між вхіднымі данымі і паўнастацэннымі выходнымі рэзультатамі. Назваць всі неабходныя элементы, задаць крэтыяры успеху і не прабоўваць прыймаць часткова завершаныя рэзультаты без паведамлення.
question_1 = "what is prompt engineer ?"
async def main():
print("hello ReAct agent by Langraph")
message = HumanMessage(content=question_1)
app = await build_graph()
res = await app.ainvoke({"messages": [message]})
if __name__ == "__main__":
asyncio.run(main())
5. Адаптаванне MCP у рэжыме SSE
Калі працуеце над пунктам 5. Адаптаванне MCP у рэжыме SSE, спачатку запісайце умовы вікорыстання: неабяжлівыя даны, сігнал працэздольнасці і тое, што выканаецца у разы частковай нявыходкі. Такі спіс дапамагае заліцвачыць змяны ў кодзе. Запісывайце час выканання аперацый, а таксама вартасць токенаў чы выкарыстоўваных запітаў разам з рэзультатамі ўжыцця. Відразлівае паказанне вартасцей запобегае неспадзяваным расходам, калі працэс пераходзіць з дэмавайнага рэжыма ў спяльныя асяроддзя. Перад налаштаваннем запрошэнняў (prompts) пераканайцеся, што сістэма правільна адпавядае на фіксованы набор запітаў. Частая зміна запрошэнняў рэдка калі-небудзь выправляе проблемы з якасцю адпаведзей.
5.1. Сервер MCP
Калі працуеце над раздзелам 5.1. MCP Server, спачатку запісайце умовы взаўмадзейнасці: неабходныя даны, сігнал пра успех і тое, што выходзіць на падчасныя адказы. Такі чарт дапамагае залічваць змяны ў кодзе праз прызма цых умов. Храніце настройкі парадульна ад коду прыемліка. Файлы сераўнавання, храненні секрэтных даных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаць пераглядаць іх без неабяжнай чытанняў усіх элементаў сістэмы. Перад налагоджэнням запітоў пераканайцеся, што сістэма правільна адпавядае на фіксованы набор запітаў. Частая зміна формулазавання запітоў рэдка калі дапамагае паспрабаваць выправіць слабкую эфектыўнасць адказоў.
# for SSE
mcp = FastMCP(
name="My MCP Server",
host="0.0.0.0",
port=8001
)
# implement mcp tools here
if __name__ == "__main__":
mcp.run(transport="sse")
5.2. MCP Кліент
Калі працуеце над раздзелам 5.2. MCP Client, спачатку запісайце умовы вярбунка: неабяжлівыя даныя, сігнал працэйскага успеху і тое, што выканаецца у разе частковага няўспэху. Такі спіс контроля дапамагае заставіць пазнейшыя змены коду быць чыстымі. Документавайце як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія контралі і обработка некорэктных паведамленняў ёсць часткай продукту, а не пазнейшым дапрацоўкам. Перад налаштаваннем запітаў пераканайцеся ў рэгулярнасці адпаведзенняў на фіксаваны набор запітанняў. Змена запітоў рэдка калі вярнайце слабкую якасць адпаведзенняў. Калі працуеце над раздзелам 5.2. MCP Client, спачатку запісайце умовы вярбунка: неабяжлівыя даныя, сігнал працэйскага успеху і тое, што выканаецца у разе частковага няўспэху. Такі спіс контроля дапамагае заставіць пазнейшыя змены коду быць чыстымі. Спрыятлівае ставленне да гэтага этапу значыць спрыятлівае ставленне да умов вярбунка і пераканання ў правамільнасці адпаведзенняў. Даўце назвы элементам, задайце крытэрыяі успеху і не прымайце частковыя адпаведзенні без паўтарэння інформацыі.
# specify config of the SSE MCP server that the client will connect to
SERVER_CONFIG = {
# config for mcp by sse transportation
"fastapi_mcp_server": {
"transport": "sse",
"url": "http://localhost:8001/sse",
}
}
# create a client
def get_mcp_client() :
client = MultiServerMCPClient(SERVER_CONFIG)
return client
5.3. Іспытанне інструментаў MCP у StateGraph
5.3. Іспытанне інструментаў MCP у StateGraph даходзіць до найкрашчых рэзультатаў, калі іх спрыяваць як мерыемую паверхню. Запісаць адна ідэальная транскрыпцыю, адзін прыклад неудачы і прыметкі па поверненню да пачатковага стану пры розшырэнні масштаба. Запісваць часы выканання і кост токеноў або запытаў разам з функцыйнальнымі рэзультатамі. Візуабілізацыя костаў з самага пачатку запобегае неспакойным рахункам, калі парадокс пераходзіць з дэмовай асяроддзя ў спяльнаныя сераўы. Раздзеляць політыку часткавага апрацоўвання дадзеных і політыку ўзяць дадзеныя. Змена адной з іх не должна вымагаць перапісвы другой, калі зменяюцца паказнікі якосці.
async def build_graph(mcp_tools: list):
"""
build StateGraph
:return:
"""
# create an llm
llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools
# bind tools to ToolNode and llm
tool_node = ToolNode(tools)
llm_with_tools = llm.bind_tools(tools)
5.4. Чаму рэалізацыя SSE разлічыцца ад STDIO
5.4. Чаму рэалізацыя SSE разлічаецца ад STDIO: яна працуе наякша, калі ўваходзіць у прыемнае паводзе для меры. Запісаўце адна ідэальная транскрыпцыя, адзін кейс неудачы і прыметкі па адвярненню перад расшырэнням масштаба. Храніце настройкі праза код аплікацыі. Файлы сераўіса, хранальнікі секрэтных дадзеных і флагі функцый должны знаходзіцца ў аднам месцы, куда аператары можу выконваць аўдыт без неабяжнага чытання всіх дадзеных.
6. Інтэграцыя кліента SSE MCP з FastAPI
from mcp_client_sse import get_mcp_client
# CREATES a manager for FastAPI startup/shutdown
@asynccontextmanager
async def lifespan(fastapi_app: FastAPI):
# create an MCP client, and retrieve MCP tools
mcp_client = get_mcp_client()
mcp_tools = await mcp_client.get_tools()
# Compile graph dynamically with tools and store it on app state
fastapi_app.state.agent = await build_graph(mcp_tools)
# PAUSES here while FastAPI runs and handles requests
yield
app = FastAPI(lifespan=lifespan)
7. Перабудова StateGraph у частцы 5
7.1. Канец FastAPI
@app.post("/generate_text", response_model=RAGResponse)
async def query_by_RAG_controller(request: Request, body: RAGRequest=Body(...)) -> RAGResponse:
try:
# wrap input query in a HumanMessage
message = HumanMessage(content=body.question)
# loads the StateGraph agent from the request
# this agent is created and maintained by lifespan context
agent = request.app.state.agent
# Async invokes the agent and returns the response
generation = await agent.ainvoke({"messages": [message]})
response = {"answer": generation.get("generation", "")}
return RAGResponse(**response)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))