带有 RAG 和 MCP 的 AI 代理：外部工具集成

## 通过 MCP 使用 RAG 和外部工具构建 AI 代理

现代 AI 代理已超越单纯的文本生成——它们能与外部 API 交互、管理状态，并基于上下文做出决策。本文将构建一个结合 RAG 处理内部文档和 Model Context Protocol (MCP) 访问外部服务（天气和 GitHub Issues）的代理。该架构基于 LangGraph 和 FastMCP 构建，无需编写自定义适配器即可实现扩展。

架构组件：ReAct、RAG 和 MCP

我们的 AI 代理基于 ReAct 循环运行：模型分析查询 → 选择工具 → 获取结果 → 重复循环直至得出最终答案。这与线性链不同，后者步骤序列是严格预定义的。LangGraph 用于支持决策过程——它提供显式状态管理和转换图。

这里的 RAG 并非独立模块，而是代理工具之一。当用户询问内部文档时，代理会调用 search_documentation，而非依赖模型的参数记忆。这能降低幻觉风险，提高响应准确性。

Model Context Protocol (MCP) 标准化了外部工具的连接。我们无需为每个 API 手动编写包装器，只需运行 MCP 服务器，通过 JSON-RPC 提供工具。代理通过客户端连接服务器，所有工具即自动可用。这在生产环境中特别有用，因为集成数量会快速增长。

技术栈及选择理由

技术选型基于可扩展性、异步支持和 MCP 兼容性要求：

• Python 3.11+：支持现代 async/await，生态稳定。
• LangGraph + LangChain：LangGraph 提供状态控制和 ReAct 循环，LangChain 提供模型与工具抽象。
• FastMCP：用于快速创建 MCP 服务器的 Python 库，通过装饰器实现。2.0 版开发活跃，并包含客户端功能。
• ChromaDB：轻量级嵌入式向量数据库，适合原型开发。
• OpenAI text-embedding-3-small：嵌入向量的性价比最优。
• Claude 或 GPT-4：任何支持函数调用的模型。

实现外部工具的 MCP 服务器

第一步是创建 MCP 服务器，提供两个工具：获取天气和创建 GitHub Issue。该服务器基于 FastMCP 构建，以子进程形式运行，使用 stdio 传输。

# tools_server.py
import os
import httpx
from fastmcp import FastMCP

mcp = FastMCP("ExternalTools")

@mcp.tool()
async def get_weather(city: str) -> str:
    async with httpx.AsyncClient() as client:
        geo_response = await client.get(
            "https://geocoding-api.open-meteo.com/v1/search",
            params={"name": city, "count": 1, "language": "en", "format": "json"},
            timeout=10.0
        )
        geo_data = geo_response.json()
        if not geo_data.get("results"):
            return f"Gorod '{city}' not found"
        
        location = geo_data["results"][0]
        lat, lon = location["latitude"], location["longitude"]
        
        weather_response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={"latitude": lat, "longitude": lon, "current_weather": True},
            timeout=10.0
        )
        current = weather_response.json()["current_weather"]
        return f"Pogoda in {location['name']}: {current['temperature']}°C, veter {current['windspeed']} km/ch"

@mcp.tool()
async def create_github_issue(repo: str, title: str, body: str, labels: list[str] | None = None) -> str:
    github_token = os.environ.get("GITHUB_TOKEN")
    if not github_token:
        return "Error: GITHUB_TOKEN not ustanovlen"
    
    url = f"https://api.github.com/repos/{repo}/issues"
    headers = {"Authorization": f"Bearer {github_token}", "Accept": "application/vnd.github+json"}
    payload = {"title": title, "body": body}
    if labels: payload["labels"] = labels
    
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload, headers=headers)
        if response.status_code == 201:
            return f"Issue sozdan: {response.json()['html_url']}"
        else:
            return f"Error: {response.status_code}"

if __name__ == "__main__":
    mcp.run(transport="stdio")

实现的关键特性：

• @mcp.tool() 装饰器根据类型注解和文档字符串自动生成参数 JSON 模式。
• 使用异步 HTTP 客户端实现非阻塞请求。
• 每个工具内置错误处理——代理收到人类可读消息，而非异常。
• Stdio 传输允许客户端将服务器作为子进程运行，通过 stdin/stdout 通信。

将 RAG 模块设置为代理工具

RAG 系统索引 PDF 文档，并通过 LangChain 兼容工具提供搜索功能。重要的是，该工具与 MCP 工具注册在同一池中，代理可与外部 API 一起选择它。

# rag_tool.py
import os
from pathlib import Path
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.tools import tool

CHROMA_PATH = "./chroma_db"
DATA_PATH = "./data"

@tool
def search_documentation(query: str) -> str:
    """Ischet relevantnye fragmenty in documentation by request."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma(persist_directory=CHROMA_PATH, embedding_function=embeddings)
    docs = vectorstore.similarity_search(query, k=3)
    return "\n\n".join([f"Withtranitsa {d.metadata.get('page', 'N/A')}: {d.page_content[:500]}..." for d in docs])

# Function initsializatsii database (vyzyvaetsya when starte agenta)
def initialize_vector_store(force_reload=False):
    if os.path.exists(CHROMA_PATH) and not force_reload:
        return
    
    documents = []
    for pdf_file in Path(DATA_PATH).glob("*.pdf"):
        loader = PyPDFLoader(str(pdf_file))
        documents.extend(loader.load())
    
    if not documents:
        return
    
    splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    chunks = splitter.split_documents(documents)
    
    Chroma.from_documents(chunks, OpenAIEmbeddings(model="text-embedding-3-small"), persist_directory=CHROMA_PATH)

在 LangGraph 中集成 MCP 客户端并组装代理

最后阶段是组装结合 MCP 工具和 RAG 的代理。我们使用 langchain-mcp-adapters 将 MCP 工具转换为 LangChain 格式，然后与 RAG 工具一起注册到图中。

# agent.py
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage
from langchain_mcp_adapters import McpToolNode
from rag_tool import search_documentation
import subprocess
import asyncio

# Run MCP-server how podprotsessa
mcp_process = subprocess.Popen(
    ["python", "tools_server.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

# Withbudynek MCP-client and adaptera narzędzieov
mcp_node = McpToolNode(
    server_command=["python", "tools_server.py"],
    transport="stdio"
)

# Receiving list narzędzieov from MCP-server
mcp_tools = asyncio.run(mcp_node.get_tools())
all_tools = [*mcp_tools, search_documentation]

# Initialization modeli and graph
model = ChatAnthropic(model="claude-3-5-sonnet-20240620").bind_tools(all_tools)

# Withstoyanie graph
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

# Uzly graph
def call_model(state):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state):
    last_message = state["messages"][-1]
    return "continue" if last_message.tool_calls else "end"

# Postroenie graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", mcp_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"continue": "tools", "end": END})
workflow.add_edge("tools", "agent")

app = workflow.compile()

# Example run
result = app.invoke({"messages": [HumanMessage(content="Uznay pogodu in Londone and sozday issue in repo/my-project with etim rezultatom")]})
print(result["messages"][-1].content)

重要特性：

• MCP 无需为每个新工具编写自定义适配器——只需在服务器上添加 @mcp.tool() 装饰器。
• RAG 作为普通工具集成，代理可在搜索文档和调用外部 API 之间选择。
• LangGraph 提供透明状态管理和 ReAct 循环控制，这对复杂多步任务至关重要。
• 异步工具实现避免阻塞代理主线程。
• Stdio 传输简化部署——服务器和代理可在同一容器或机器上运行，无需网络依赖。

— Editorial Team