LangGraph与MCP协议：构建可编排的大模型工作流实战指南 1. 先搞清楚 LangGraph MCP 到底解决什么问题如果你正在接触大模型应用开发大概率遇到过这些问题单次对话无法处理复杂任务链、工具调用不稳定、多步骤任务状态容易丢失、不同模型或服务之间的切换成本高。LangGraph 加上 MCPModel Context Protocol就是针对这类问题的一套组合方案。LangGraph 的核心是让大模型任务变成可编排、有状态的工作流。它不是简单封装 API而是把对话、工具调用、条件分支、循环这些动作组织成图结构。MCP 则更具体——它定义了一套标准协议让不同模型、工具、数据源之间能安全、可控地交换上下文信息。实际落地时这套组合最适合两类场景一是需要多步骤决策的智能助手比如根据用户需求自动查数据、做分析、生成报告二是需要混合多个模型或工具的长任务链比如先用一个模型做摘要再用另一个模型做校对最后调用外部 API 推送结果。和单纯用 LangChain 相比LangGraph 更强调“状态持久化”和“流程控制”。如果你做过一些简单的聊天机器人或单次问答工具但一遇到需要“记住前面步骤结果”或“根据中间结果决定下一步”的场景就头疼那 LangGraph 的设计思路会直接解决这个痛点。2. 环境准备别急着装包先确认你的资源边界开始写代码前最容易被忽略的是环境资源评估。LangGraph 工作流一旦跑起来可能会同时占用多个模型实例、工具服务或外部 API如果没提前规划好资源配额和网络条件很容易在调试阶段卡住。基础环境清单Python 3.9建议 3.11兼容性最稳至少 4GB 可用内存复杂工作流会同时加载多个组件稳定的网络如果需要调用在线模型或工具本地开发时预留 2GB 磁盘空间缓存模型、日志、临时文件模型资源准备如果你用 OpenAI GPT 系列、Claude 或国内大模型 API先确认账号余额和速率限制如果本地部署模型比如用 Ollama、vLLM需要检查 GPU 显存7B 模型至少 8GB13B 模型建议 16GB测试阶段建议先用小参数模型如 Llama 3 8B、Qwen 7B跑通流程再换大模型关键依赖版本# 核心包 pip install langgraph0.0.40 pip install langchain0.1.0 # 如果用到 MCP 协议工具 pip install mcp1.0.0 # 常用工具包按需安装 pip install requests # 调用外部 API pip install pydantic # 数据验证 pip install numpy # 数值计算注意LangGraph 和 LangChain 版本兼容性很重要别直接装最新版。建议先锁版本langgraph0.0.40、langchain0.1.0跑通后再升级。3. 最小可运行示例从单轮对话到工作流LangGraph 的学习曲线有点陡直接看复杂示例容易懵。我建议从最基础的“对话记忆”工作流开始再逐步加入工具调用和多步逻辑。3.1 先实现一个能记住上下文的聊天助手from langgraph.graph import StateGraph, END from typing import Dict, Any import operator # 定义状态结构记录对话历史和最新查询 class ChatState(Dict[str, Any]): messages: list [] current_query: str # 初始化图 builder StateGraph(ChatState) # 添加节点处理用户输入 def input_node(state: ChatState): state[messages].append({role: user, content: state[current_query]}) return state # 添加节点调用模型生成回复 def model_node(state: ChatState): # 这里简化处理实际替换为你的大模型调用 last_message state[messages][-1][content] response f收到了你的消息{last_message}。我正在处理中... state[messages].append({role: assistant, content: response}) return state # 构建流程 builder.add_node(input, input_node) builder.add_node(model, model_node) # 设置边从 input 到 model然后结束 builder.set_entry_point(input) builder.add_edge(input, model) builder.add_edge(model, END) # 编译图 graph builder.compile() # 测试运行 initial_state ChatState(current_query你好请帮我查天气) result graph.invoke(initial_state) print(result[messages])这个示例虽然简单但包含了 LangGraph 核心概念状态管理、节点定义、流程编排。运行后会看到消息历史被完整保留这是后续扩展的基础。3.2 加入条件判断让工作流能“做决定”真实场景中我们经常需要根据模型回复决定下一步动作。比如用户问“查天气”助手应该判断是否需要调用天气 API。from langgraph.graph import StateGraph, END from typing import Dict, Any class AgentState(Dict[str, Any]): messages: list [] needs_tool: bool False tool_result: str # 判断是否需调用工具 def should_use_tool(state: AgentState): last_msg state[messages][-1][content] # 简单关键词判断实际可用模型分类 if 天气 in last_msg or weather in last_msg: state[needs_tool] True return state # 工具调用模拟 def call_weather_api(state: AgentState): if state[needs_tool]: state[tool_result] 北京晴25℃ # 模拟API返回 return state # 生成最终回复 def generate_response(state: AgentState): if state[needs_tool]: response f根据查询结果{state[tool_result]} else: response 我暂时无法处理这个请求。 state[messages].append({role: assistant, content: response}) return state builder StateGraph(AgentState) builder.add_node(check_tool, should_use_tool) builder.add_node(call_tool, call_weather_api) builder.add_node(respond, generate_response) builder.set_entry_point(check_tool) builder.add_edge(check_tool, call_tool) builder.add_edge(call_tool, respond) builder.add_edge(respond, END) graph builder.compile() # 测试 test_state AgentState(messages[{role: user, content: 北京天气怎么样}]) result graph.invoke(test_state) print(result[messages][-1])这个示例展示了条件路由先判断用户意图再决定是否调用工具。LangGraph 的强项就在这里——你可以把复杂决策流程可视化、可调试。4. 集成 MCP 协议让工具调用更标准可控MCP 的核心价值是标准化工具接入方式。在没有 MCP 之前每个工具都需要写自定义封装代码有了 MCP只要工具方提供符合协议的 Server就能快速集成。4.1 理解 MCP 的基本工作模式MCP 协议下工具被抽象为独立的 Server通过 JSON-RPC 与主程序通信。比如天气查询工具、数据库连接工具、文件读写工具都可以作为 MCP Server 运行。典型交互流程主程序LangGraph 工作流向 MCP Server 发送请求MCP Server 执行具体操作如查询数据库、调用 APIServer 返回结构化结果主程序将结果融入上下文继续工作流4.2 本地运行一个 MCP Server 示例先创建一个简单的 MCP Server模拟天气查询# weather_mcp_server.py import json import asyncio from mcp.server import Server from mcp.server.models import InitializationOptions import mcp.server.stdio server Server(weather-server) server.list_tools() async def handle_list_tools(): return [{ name: get_weather, description: 获取指定城市的天气信息, inputSchema: { type: object, properties: { city: {type: string, description: 城市名称} }, required: [city] } }] server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name get_weather: city arguments.get(city, 北京) # 模拟天气数据 return {content: [{type: text, text: f{city}晴25℃}]} raise ValueError(f未知工具: {name}) async def main(): async with mcp.server.stdio.stdio_server() as (read, write): await server.run( read, write, InitializationOptions( server_nameweather-server, server_version1.0.0, capabilitiesserver.get_capabilities( notification_optionsNone, experimental_capabilitiesNone ) ) ) if __name__ __main__: asyncio.run(main())4.3 在 LangGraph 中集成 MCP Clientfrom langgraph.graph import StateGraph, END from typing import Dict, Any import mcp.client as mcp_client import asyncio class MCPState(Dict[str, Any]): messages: list [] tool_results: list [] async def create_mcp_session(): # 连接本地 MCP Server实际需要启动上述 weather_mcp_server.py session await mcp_client.create_session( python, [weather_mcp_server.py] ) await session.initialize() return session def tool_call_node(state: MCPState): # 简化示例实际需要异步处理 async def call_weather_tool(): session await create_mcp_session() result await session.call_tool( get_weather, {city: 北京} ) return result # 在同步环境中运行异步代码生产环境建议用异步框架 loop asyncio.new_event_loop() asyncio.set_event_loop(loop) try: tool_result loop.run_until_complete(call_weather_tool()) state[tool_results].append(tool_result) finally: loop.close() return state def response_node(state: MCPState): if state[tool_results]: result_text state[tool_results][-1][content][0][text] state[messages].append({ role: assistant, content: f天气信息{result_text} }) return state builder StateGraph(MCPState) builder.add_node(tool_call, tool_call_node) builder.add_node(response, response_node) builder.set_entry_point(tool_call) builder.add_edge(tool_call, response) builder.add_edge(response, END) graph builder.compile()这个示例展示了 LangGraph 如何通过 MCP 协议调用外部工具。实际生产中MCP Server 可以部署为独立服务提供更复杂的工具能力。5. 构建智能小秘书从单功能到多功能组合现在我们把前面学的组合起来做一个真正可用的智能小秘书原型。这个秘书能处理多种请求查天气、记笔记、做计算。5.1 定义完整的状态结构from typing import Dict, Any, List, Optional from datetime import datetime from langgraph.graph import StateGraph, END class SecretaryState(Dict[str, Any]): messages: List[Dict] [] # 对话历史 current_intent: str # 当前意图 tool_calls: List[Dict] [] # 工具调用记录 user_data: Dict {} # 用户数据如笔记 timestamp: str # 当前时间戳 # 意图识别节点 def intent_detection(state: SecretaryState): last_message state[messages][-1][content] if state[messages] else if any(word in last_message for word in [天气, weather]): state[current_intent] weather elif any(word in last_message for word in [笔记, 记一下, note]): state[current_intent] note elif any(word in last_message for word in [计算, 算一下, calculate]): state[current_intent] calculation else: state[current_intent] chat state[timestamp] datetime.now().isoformat() return state5.2 实现多工具路由# 工具调用路由 def tool_router(state: SecretaryState): intent state[current_intent] if intent weather: return weather_tool elif intent note: return note_tool elif intent calculation: return calc_tool else: return general_chat # 天气工具节点 def weather_tool(state: SecretaryState): # 模拟天气查询 state[tool_calls].append({ tool: weather, timestamp: state[timestamp], result: 北京晴25℃ # 实际调用MCP Server }) return state # 笔记工具节点 def note_tool(state: SecretaryState): last_message state[messages][-1][content] # 提取笔记内容简化处理 note_content last_message.replace(记一下, ).strip() state[user_data][notes] state[user_data].get(notes, []) [note_content] state[tool_calls].append({ tool: note, content: note_content, timestamp: state[timestamp] }) return state # 计算工具节点 def calc_tool(state: SecretaryState): last_message state[messages][-1][content] # 简单数学计算实际可用eval或数学库 if 11 in last_message: result 2 else: result 无法计算 state[tool_calls].append({ tool: calculation, expression: last_message, result: result }) return state5.3 组装完整工作流# 通用聊天节点无需工具时使用 def general_chat(state: SecretaryState): last_message state[messages][-1][content] response f我理解您说的是{last_message}。请问需要我具体做什么吗 state[messages].append({role: assistant, content: response}) return state # 响应生成节点整合工具结果 def response_generator(state: SecretaryState): if state[tool_calls]: last_tool state[tool_calls][-1] if last_tool[tool] weather: response f天气信息{last_tool[result]} elif last_tool[tool] note: response f已记录笔记{last_tool[content]} elif last_tool[tool] calculation: response f计算结果{last_tool[result]} else: response 任务完成 else: response 我暂时无法处理这个请求。 state[messages].append({role: assistant, content: response}) return state # 构建完整图 builder StateGraph(SecretaryState) builder.add_node(intent_detect, intent_detection) builder.add_node(weather_tool, weather_tool) builder.add_node(note_tool, note_tool) builder.add_node(calc_tool, calc_tool) builder.add_node(general_chat, general_chat) builder.add_node(generate_response, response_generator) builder.set_entry_point(intent_detect) # 条件路由 builder.add_conditional_edges( intent_detect, tool_router, { weather_tool: weather_tool, note_tool: note_tool, calc_tool: calc_tool, general_chat: general_chat } ) # 所有工具节点最终都流向响应生成 builder.add_edge(weather_tool, generate_response) builder.add_edge(note_tool, generate_response) builder.add_edge(calc_tool, generate_response) builder.add_edge(general_chat, generate_response) builder.add_edge(generate_response, END) secretary_graph builder.compile() # 测试多个功能 test_cases [ 今天天气怎么样, 记一下明天开会时间下午三点, 算一下11等于多少, 随便聊聊天 ] for query in test_cases: state SecretaryState(messages[{role: user, content: query}]) result secretary_graph.invoke(state) print(f用户: {query}) print(f助手: {result[messages][-1][content]}) print(- * 50)6. 生产环境注意事项从原型到可用的关键步骤上面示例能在本地跑通但要真正投入使用还需要解决几个实际问题。6.1 错误处理和重试机制工作流中任何一个节点失败都可能让整个流程中断。必须添加错误捕获和恢复逻辑。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def safe_tool_call(state: SecretaryState, tool_func): try: return tool_func(state) except Exception as e: state[errors] state.get(errors, []) [str(e)] # 记录日志 print(f工具调用失败: {e}) # 返回降级结果 state[tool_calls].append({ tool: fallback, error: str(e), result: 服务暂时不可用 }) return state # 在节点函数中包装工具调用 def robust_weather_tool(state: SecretaryState): return safe_tool_call(state, weather_tool)6.2 状态持久化LangGraph 默认在内存中维护状态服务重启后数据会丢失。生产环境需要持久化存储。import pickle import os class PersistentSecretary: def __init__(self, user_id: str, storage_path: str ./sessions): self.user_id user_id self.storage_path storage_path os.makedirs(storage_path, exist_okTrue) self.session_file f{storage_path}/{user_id}.pkl def load_state(self) - SecretaryState: if os.path.exists(self.session_file): with open(self.session_file, rb) as f: return pickle.load(f) return SecretaryState() def save_state(self, state: SecretaryState): with open(self.session_file, wb) as f: pickle.dump(state, f) def process_query(self, query: str) - str: # 加载历史状态 state self.load_state() state[messages].append({role: user, content: query}) # 执行工作流 result secretary_graph.invoke(state) # 保存更新后的状态 self.save_state(result) return result[messages][-1][content] # 使用示例 secretary PersistentSecretary(user123) response secretary.process_query(北京天气怎么样) print(response)6.3 性能监控和限流复杂工作流可能消耗大量资源需要监控和限制。import time from functools import wraps def monitor_performance(func): wraps(func) def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) execution_time time.time() - start_time # 记录性能数据 print(f{func.__name__} 执行时间: {execution_time:.2f}秒) if execution_time 5.0: # 超时阈值 print(警告: 节点执行超时) return result return wrapper # 应用到关键节点 monitor_performance def monitored_weather_tool(state: SecretaryState): return weather_tool(state)7. 常见问题排查指南实际开发中一定会遇到各种问题这是我自己踩坑后整理的排查顺序。7.1 工作流卡住或无限循环现象程序运行不结束CPU 占用高但无输出。排查步骤检查条件边界确认每个循环节点都有退出条件简化测试先用最小输入测试单个节点添加日志在每个节点入口出口打印状态检查状态更新确认每个节点都正确修改了状态对象# 调试用日志装饰器 def debug_logging(func): wraps(func) def wrapper(state, *args, **kwargs): print(f 进入节点 {func.__name__}) print(f 当前状态: {state.keys()}) result func(state, *args, **kwargs) print(f 离开节点 {func.__name__}) print(f 更新状态: {result.keys()}) return result return wrapper7.2 MCP 连接失败现象工具调用时报连接错误或超时。排查步骤确认 MCP Server 已启动检查进程是否运行验证通信协议用简单客户端测试 MCP Server 是否响应检查防火墙和端口确认没有网络隔离查看 MCP Server 日志通常会有详细的错误信息7.3 状态数据混乱现象多次调用后状态数据出现异常或重复。排查步骤检查状态初始化每次调用前是否正确重置或加载状态验证数据序列化持久化存储时数据格式是否正确审查状态修改逻辑确认没有意外的状态污染添加数据验证使用 Pydantic 模型验证状态结构from pydantic import BaseModel, validator class ValidatedSecretaryState(BaseModel): messages: List[Dict] [] current_intent: str tool_calls: List[Dict] [] validator(current_intent) def validate_intent(cls, v): allowed_intents [weather, note, calculation, chat, ] if v not in allowed_intents: raise ValueError(f无效意图: {v}) return v这套 LangGraph MCP 的方案确实能解决复杂智能助手的开发痛点但真正落地时需要耐心调试。建议先在一个小项目上跑通全流程再逐步增加复杂度。