Агентные ИИ с LangChain — Часть 6: Агентные RAG на основе MCP
Часть 6: Агентные ИИ с 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. «Концепция» сначала запишите контракт: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия соответствующим элементам, определите критерии успешности и не допускайте молчаливого частичного выполнения задачи. Оцените уровень воспроизводимости на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.
3.2 Архитектура MCP
При изучении раздела 3.2 «Архитектура MCP» сначала запишите условия взаимодействия: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Рядом с функциональными результатами записывайте время выполнения и стоимость токенов или запросов. Отслеживание затрат с самого начала предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды. Перед настройкой подсказок измерьте уровень воспроизведения ответов на фиксированный набор вопросов. Частая смена подсказок редко помогает улучшить качество поиска.
3.3. Режимы передачи данных MCP
При работе над разделом 3.3. Режимы передачи MCP сначала запишите условия взаимодействия: необходимые параметры входных данных, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Храните конфигурацию вне кода приложения. Файлы с настройками окружения, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Измеряйте уровень точности постоянного набора вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.
3.3.1. Режим STDIO
При работе над разделом 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 Client
Для версии 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 сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и последствия частичной неудачи. Такой список поможет сохранять честность при последующих изменениях кода. Рядом с функциональными результатами записывайте время выполнения и стоимость токена или запроса. Отслеживание затрат с самого начала предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды. Перед настройкой подсказок измерьте уровень воспроизведения ответов на фиксированный набор вопросов. Частая смена подсказок редко помогает улучшить качество поиска.
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 Client
При работе над разделом 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))