首页 / 文章

mcplangchainragagents

LangChain驱动的智能体AI——第6部分：基于MCP的智能体RAG

LangChain驱动的代理型AI第六部分：基于MCP的代理型RAG，检索工具来自MCP服务器，而LangGraph负责协调代理流程。

发布者：REACTAPP.TOP阅读约 12 分钟2686 词

来源: Agentic AI with LangChain — Part 6: MCP Powered Agentic RAG with LangGraph

本指南将逐步构建从原始材料到可运行系统的完整流程，内容涵盖：使用 LangChain 的智能体 AI —— 第 6 部分：基于 LangGraph 和 MCP 的智能体 RAG。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。 在修改代码之前，应先明确输入内容、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需推测隐藏状态。 可将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检查标准，并拒绝默许的半完成状态。

技术概念

在研究技术概念时，首先列出相关约定：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先用固定的问题集测试召回率。仅仅更换提示词往往无法改善较差的检索效果。

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（模型上下文协议）

在研究 3. MCP（模型上下文协议）时，首先需明确相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元而非庞大的脚本。当某个步骤失败时，故障应能指向具体的责任模块，而非复杂的流程链。 应对稳定的系统指令和工具结构进行缓存。重复发送相同的开头信息是导致资源浪费的常见原因。

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. 在STDIO模式下实现MCP

  1. 在STDIO模式下实现MCP时，若将其视为可度量的对象会更为有效。在扩大范围之前，先记录一份理想的操作示例、一个失败案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。

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部分：在Agent中使用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. 在SSE模式下实现MCP

在处理“5. 在SSE模式下实现MCP”这一部分时，首先需明确相关规范：所需的输入参数、成功信号以及出现部分故障时的处理方式。这样的清单能确保后续的代码修改始终符合预期。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定的问题集测试检索效果。仅仅更换提示词往往无法改善较差的检索性能。

5.1. MCP服务器

在处理5.1节MCP服务器相关内容时，首先需明确接口规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改不会偏离原有设计。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，这样操作人员无需查看整个系统结构即可进行审计。 在调整提示词之前，先使用固定的问题集测试系统的召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。

# 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客户端时，首先需明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。 在处理5.2节MCP客户端时，首先需明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并杜绝无声的半完成状态。

# 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. 在StateGraph中使用MCP工具

5.3. 在StateGraph中使用MCP工具时，若将其视为可度量的表面来处理效果最佳。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。在功能结果旁还需记录执行时间以及令牌或查询成本。提前了解这些成本信息，可避免在系统从演示环境过渡到共享环境时出现意外费用。应将分块策略与检索策略分开，当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。

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. 重新构建第5部分中的StateGraph

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))

7.2. Streamlit前端

8. MCP管理的工具与本地LangChain工具的对比

9. 总结

操作检查清单

← 全部文章发送简报