如何使用LangGraph搭建一个智能体应用，需要mcp和工具调用

下面是一份从零开始、基于LangGraph 框架开发智能问答系统的完整实战教程，涵盖API 工具调用和MCP 服务调用两大核心能力。

一、什么是 LangGraph？

LangGraph 是由 LangChain 团队开发的低级别编排框架，专门用于构建、管理和部署长期运行、有状态的 AI 智能体工作流。

核心设计理念

传统 LangChain 的 Chain 是线性链式调用（A→B→C），而 LangGraph 采用有向图结构，支持循环、条件分支、并行执行，更贴合真实业务中"反复推理-执行-判断"的 Agent 行为。

四大核心组件

为什么用 LangGraph 而不是 LangChain Agent？

• 精细控制：可以精确控制每一步的流转逻辑，而非黑盒执行
• 状态持久化：内置 Checkpoint 机制，支持断点续传和故障恢复
• 人机协同：支持在任意节点暂停等待人工审核
• 多智能体协作：支持子图嵌套，实现复杂的多 Agent 编排

二、整体架构设计

用户提问 │ ▼ ┌──────────────────────────────────────────┐ │ LangGraph 智能体（图工作流） │ │ │ │ ┌─────────┐ ┌──────────┐ │ │ │ LLM节点 │◄──►│ 工具节点 │ ← 循环调用 │ │ └─────────┘ └────┬─────┘ │ │ │ │ │ ┌────────────┼────────────┐ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ ┌────────┐ │ │ │自定义工具 │ │ RAG检索 │ │MCP工具 │ │ │ │(API调用) │ │(知识库) │ │(外部) │ │ │ └──────────┘ └──────────┘ └────────┘ │ └──────────────────────────────────────────┘ │ ▼ 输出回答

三、环境搭建

3.1 创建项目

mkdirlanggraph-qa-systemcdlanggraph-qa-system python-mvenv venv# Windowsvenv\Scripts\activate# macOS/Linuxsourcevenv/bin/activate

3.2 安装依赖

# LangGraph 核心框架pipinstalllanggraph langchain-core langchain-openai# MCP 集成pipinstalllangchain-mcp-adapters mcp fastmcp# 向量数据库（RAG）pipinstallchromadb langchain-chroma# 文档加载pipinstallpypdf python-docx# HTTP 请求pipinstallhttpx# 环境变量pipinstallpython-dotenv

3.3 配置环境变量

创建.env文件：

OPENAI_API_KEY="sk-xxx" # 如使用通义千问等国内模型 # DASHSCOPE_API_KEY="your_key"

四、接入大语言模型

importosfromdotenvimportload_dotenv load_dotenv()fromlangchain_openaiimportChatOpenAI# 方式一：OpenAIllm=ChatOpenAI(model="gpt-4o",temperature=0)# 方式二：通义千问（兼容 OpenAI 接口）# llm = ChatOpenAI(# model="qwen-plus",# api_key=os.getenv("DASHSCOPE_API_KEY"),# base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"# )

五、定义自定义工具（API 调用）

工具是智能体的"手脚"，让它能执行搜索、计算、调用外部 API 等操作。

fromlangchain_core.toolsimporttoolfromdatetimeimportdatetimeimporthttpx@tooldefget_current_time(city:str)->str:"""获取指定城市的当前时间。当用户询问时间、几点钟等问题时调用。"""now=datetime.now().strftime("%Y-%m-%d %H:%M:%S")returnf"{city}当前时间是{now}"@tooldefquery_weather_api(city:str)->str:"""通过 API 查询指定城市的天气信息。当用户询问天气时调用。"""geocode_url="https://geocoding-api.open-meteo.com/v1/search"weather_url="https://api.open-meteo.com/v1/forecast"withhttpx.Client(timeout=10)asclient:r=client.get(geocode_url,params={"name":city,"count":1,"language":"zh"})r.raise_for_status()data=r.json()["results"][0]lat,lon=data["latitude"],data["longitude"]r=client.get(weather_url,params={"latitude":lat,"longitude":lon,"current_weather":"true"})r.raise_for_status()weather=r.json()["current_weather"]returnf"{city}当前温度{weather['temperature']}°C，风速{weather['windspeed']}km/h"@tooldefcalculate(expression:str)->str:"""计算数学表达式。当用户需要数学计算时调用。"""try:result=eval(expression,{"__builtins__":{}},{})returnf"{expression}={result}"exceptExceptionase:returnf"计算出错:{e}"custom_tools=[get_current_time,query_weather_api,calculate]

💡 工具的docstring 非常关键，大模型依赖它来判断何时调用哪个工具。

六、构建 RAG 知识库

fromlangchain_community.document_loadersimportPyPDFLoaderfromlangchain.text_splitterimportRecursiveCharacterTextSplitterfromlangchain_openaiimportOpenAIEmbeddingsfromlangchain_chromaimportChroma# 加载文档loader=PyPDFLoader("your_document.pdf")documents=loader.load()# 切割文本text_splitter=RecursiveCharacterTextSplitter(chunk_size=500,chunk_overlap=50)chunks=text_splitter.split_documents(documents)# 向量化存储embeddings=OpenAIEmbeddings()vectorstore=Chroma.from_documents(documents=chunks,embedding=embeddings,persist_directory="./chroma_db")retriever=vectorstore.as_retriever(search_kwargs={"k":3})# 封装为工具@tooldefsearch_knowledge_base(query:str)->str:"""从私有知识库中检索相关信息。当用户询问专业知识、文档内容时调用。"""docs=retriever.invoke(query)ifnotdocs:return"知识库中未找到相关信息。"return"\n\n".join([doc.page_contentfordocindocs])

七、编写 MCP Server（外部服务调用）

7.1 创建 MCP Server

创建文件news_mcp_server.py：

frommcp.server.fastmcpimportFastMCP mcp=FastMCP("NewsService")@mcp.tool()defsearch_news(keyword:str)->str:"""根据关键词搜索最新新闻。当用户询问新闻资讯时调用。"""# 示例：实际项目中替换为真实 API 调用mock_news=[{"title":f"关于{keyword}的最新报道1","summary":"这是第一条新闻摘要..."},{"title":f"关于{keyword}的最新报道2","summary":"这是第二条新闻摘要..."},]return"\n".join([f"📰{n['title']}:{n['summary']}"forninmock_news])@mcp.tool()defget_stock_price(symbol:str)->str:"""查询股票实时价格。当用户询问股价、行情时调用。"""returnf"{symbol}当前价格为 150.25 元，涨幅 +2.3%"if__name__=="__main__":# stdio 模式（本地开发）mcp.run(transport="stdio")# 生产环境改用 HTTP 模式：# mcp.run(transport="streamable-http", host="0.0.0.0", port=8005)

八、用 LangGraph 构建智能体（核心）

这是整个教程的核心部分。LangGraph 提供两种构建方式：预构建快捷方式和手动图编排。

方式一：使用预构建的create_react_agent（推荐入门）

importasynciofromlangchain_openaiimportChatOpenAIfromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlanggraph.prebuiltimportcreate_react_agentasyncdefmain():# 1. 加载 MCP 工具mcp_config={"news-server":{"transport":"stdio","command":"python","args":["/你的绝对路径/news_mcp_server.py"]# 必须用绝对路径！}}client=MultiServerMCPClient(mcp_config)mcp_tools=awaitclient.get_tools()# 2. 合并所有工具all_tools=custom_tools+[search_knowledge_base]+mcp_toolsprint(f"共加载{len(all_tools)}个工具:{[t.namefortinall_tools]}")# 3. 创建 ReAct 智能体（LangGraph 预构建）agent=create_react_agent(model=llm,tools=all_tools,prompt="你是一个智能问答助手，能够查询天气、计算数学、搜索新闻和检索知识库。")# 4. 执行问答result=awaitagent.ainvoke({"messages":[{"role":"user","content":"北京现在天气怎么样？"}]})print(result["messages"][-1].content)if__name__=="__main__":asyncio.run(main())

create_react_agent内部就是一个 LangGraph 图，遵循 ReAct（推理-行动-观察）循环。

方式二：手动构建 StateGraph（完全掌控流程）

这是 LangGraph 的精髓——你可以精确控制每一步的流转逻辑：

importasyncioimportjsonfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportToolMessagefromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.prebuiltimportToolNode,tools_conditionfromlangchain_mcp_adapters.clientimportMultiServerMCPClient# ===== 1. 定义状态（State） =====classState(TypedDict):messages:Annotated[list,add_messages]# 消息列表，自动追加# ===== 2. 定义节点函数 =====defchatbot(state:State,llm_with_tools):"""LLM 节点：接收消息，决定是否调用工具"""response=llm_with_tools.invoke(state["messages"])return{"messages":[response]}# ===== 3. 构建图 =====asyncdefbuild_graph():# 加载 MCP 工具mcp_config={"news-server":{"transport":"stdio","command":"python","args":["/你的绝对路径/news_mcp_server.py"]}}client=MultiServerMCPClient(mcp_config)mcp_tools=awaitclient.get_tools()# 合并所有工具all_tools=custom_tools+[search_knowledge_base]+mcp_tools# 绑定工具到 LLMllm_with_tools=llm.bind_tools(all_tools)# 创建图graph=StateGraph(State)# 添加节点# 节点1：LLM 推理节点graph.add_node("chatbot",lambdastate:chatbot(state,llm_with_tools))# 节点2：工具执行节点（LangGraph 预构建）tool_node=ToolNode(tools=all_tools)graph.add_node("tools",tool_node)# 添加边（定义流转逻辑）# 入口 → LLM 节点graph.add_edge(START,"chatbot")# LLM 节点 → 条件判断：# - 如果 LLM 决定调用工具 → 跳转到 tools 节点# - 如果 LLM 直接回答 → 跳转到 ENDgraph.add_conditional_edges("chatbot",tools_condition)# 工具执行完 → 回到 LLM 节点（形成循环）graph.add_edge("tools","chatbot")# 编译图app=graph.compile()returnapp# ===== 4. 运行 =====asyncdefmain():app=awaitbuild_graph()# 交互式对话messages=[]whileTrue:user_input=input("\n👨‍💻: ")ifuser_input.lower()in["quit","exit","q"]:breakmessages.append({"role":"user","content":user_input})result=awaitapp.ainvoke({"messages":messages})messages=result["messages"]print(f"🤖:{messages[-1].content}")if__name__=="__main__":asyncio.run(main())

图的执行流程解析

START │ ▼ ┌─────────┐ │ chatbot │ ← LLM 推理，决定是否调工具 └────┬────┘ │ ├── 需要调工具？──► tools 节点（执行工具）──► 回到 chatbot（循环） │ └── 不需要？──► END（输出最终回答）

这就是 LangGraph 的核心价值：循环 + 条件分支，让 Agent 能反复推理直到得出最终答案。

九、添加多轮记忆（状态持久化）

fromlanggraph.checkpoint.memoryimportMemorySaver# 创建记忆存储memory=MemorySaver()# 编译时注入 checkpointerapp=graph.compile(checkpointer=memory)# 通过 thread_id 区分不同会话config={"configurable":{"thread_id":"user_001"}}# 第一轮awaitapp.ainvoke({"messages":[{"role":"user","content":"我叫小明，帮我查一下北京天气"}]},config=config)# 第二轮（智能体会记住"小明"）result=awaitapp.ainvoke({"messages":[{"role":"user","content":"我叫什么名字？"}]},config=config)

十、添加人机协同（Human-in-the-Loop）

# 在工具执行前暂停，等待人工确认app=graph.compile(checkpointer=memory,interrupt_before=["tools"]# 在执行工具前中断)# 执行到工具节点前会暂停result=awaitapp.ainvoke({"messages":[{"role":"user","content":"帮我发邮件给老板"}]},config=config)# 人工审核后，更新状态并继续awaitapp.aupdate_state(config,{"messages":[ToolMessage(...)]})awaitapp.ainvoke(None,config=config)# 继续执行

十一、生产部署建议

• 容器化：使用langgraph build构建 Docker 镜像
• 持久化存储：生产环境用PostgresSaver替代MemorySaver
• 监控追踪：集成 LangSmith 可视化追踪每一步推理过程
• MCP Server 独立部署：使用streamable-http传输模式，独立部署为微服务
• 异常处理：设置max_iterations防止无限循环

📌 常见踩坑点总结

整体开发流程总结

环境搭建 → 模型接入 → 自定义工具（API调用）→ RAG知识库构建 → MCP Server编写 → LangGraph图编排（节点+边+状态） → 工具合并注入 → 记忆集成 → 人机协同 → 测试部署

建议从create_react_agent预构建方式开始跑通流程，理解原理后再切换到手动StateGraph编排，逐步掌握 LangGraph 的图控制能力。