1. 先搞清楚LangGraph和MCP到底解决什么问题如果你正在接触AI智能体开发特别是用Python做工具调用、多步骤任务编排或复杂工作流设计LangGraph和MCP这两个工具组合值得先了解清楚。LangGraph不是LangChain的替代品而是专门解决有状态工作流的库。简单说当你需要让AI智能体记住之前做了什么、根据中间结果决定下一步、或者多个智能体之间需要协作时用LangGraph比直接用LangChain更直接。它最核心的能力是把任务流程变成一张图每个节点可以是一个工具调用、一个条件判断或一个子任务节点之间通过状态传递信息。MCPModel Context Protocol则是解决工具调用标准化的协议。传统AI智能体调用外部工具时每个工具都要写适配代码而MCP定义了一套标准接口让工具开发者和智能体开发者可以解耦。用MCP后你只需要关注工具本身的功能实现智能体通过标准协议发现和调用这些工具。实际开发中这两个技术经常结合使用LangGraph负责编排工作流MCP负责提供标准化工具库。比如一个数据分析智能体用LangGraph定义获取数据→清洗数据→分析数据→生成报告的流程每个步骤调用的工具都通过MCP协议标准化接入。2. 环境准备从零开始配好开发环境开始前先确认你的基础环境。我建议用Python 3.9或以上版本太低可能会有兼容性问题。虚拟环境不是必须但如果你经常切换不同项目用conda或venv隔离依赖会更稳妥。核心依赖就这几个包pip install langgraph langchain-coreMCP相关的包要看具体使用方式。如果你只是调用现成的MCP服务器可能只需要安装客户端pip install mcp-client如果要自己开发MCP工具还需要服务端SDKpip install mcp这里有个容易混淆的点LangGraph本身不强制要求MCP你可以用传统方式定义工具。但如果你看到项目里同时提到这两个技术大概率是要用MCP协议来标准化工具调用。验证安装是否成功最简单的方法是跑一个导入测试# 测试基础环境 try: from langgraph import StateGraph print(✓ LangGraph 导入成功) except ImportError as e: print(fLangGraph 导入失败: {e}) try: from mcp import ClientSession print(✓ MCP 客户端导入成功) except ImportError: print(MCP 客户端未安装如需使用MCP功能请单独安装)如果只是学习LangGraph的工作流设计MCP部分可以先跳过。等把状态图的基本概念搞明白后再加入MCP工具调用会更顺畅。3. LangGraph核心概念状态图和节点编排LangGraph最核心的概念是状态图StateGraph。和普通函数调用不同状态图维护一个共享的状态对象每个节点都可以读取和修改这个状态。先看一个最简单的例子聊天对话的回合制流程。from typing import Dict, Any, List from langgraph import StateGraph # 定义状态结构 class ChatState: messages: List[Dict[str, Any]] current_step: str # 创建图 builder StateGraph(ChatState) # 定义节点函数 def llm_node(state: ChatState): # 模拟LLM调用 last_message state.messages[-1][content] response f回复: {last_message} state.messages.append({role: assistant, content: response}) return state def human_input_node(state: ChatState): # 模拟用户输入 user_input input(用户: ) state.messages.append({role: user, content: user_input}) return state # 添加节点 builder.add_node(llm, llm_node) builder.add_node(human, human_input_node) # 设置入口点 builder.set_entry_point(human) # 定义边节点之间的流转条件 builder.add_edge(human, llm) builder.add_edge(llm, human) # 编译图 graph builder.compile()这个例子虽然简单但包含了LangGraph的几个关键要素状态定义ChatState定义了工作流中需要维护的数据结构节点函数每个节点接收状态处理后再返回更新后的状态边定义决定工作流的执行顺序图编译把定义好的节点和边编译成可执行的工作流实际开发中节点函数通常会更复杂可能包含工具调用、条件判断、错误处理等。但基本模式都是定义状态→添加节点→连接边→编译执行。4. MCP工具集成标准化工具调用MCP的核心价值在于工具调用的标准化。传统方式下每个工具都要写特定的适配代码而MCP通过协议定义了一套标准接口。一个典型的MCP工具开发流程# mcp_tools.py - MCP工具定义 from mcp import Server, Tool # 定义工具 Tool def search_web(query: str) - str: 搜索网页获取信息 # 实际实现会调用搜索API return f搜索结果: {query} Tool def calculate(expression: str) - str: 计算数学表达式 try: result eval(expression) # 生产环境不要用eval这里只是示例 return f计算结果: {result} except: return 计算错误 # 创建MCP服务器 server Server(tools[search_web, calculate])在LangGraph中集成MCP工具from langgraph import StateGraph from mcp_client import MCPSession class AgentState: task: str tools_used: List[str] results: List[str] async def tool_node(state: AgentState): async with MCPSession(http://localhost:8000) as session: # 根据任务类型选择合适的工具 if 计算 in state.task: result await session.call_tool(calculate, {expression: 22}) elif 搜索 in state.task: result await session.call_tool(search_web, {query: state.task}) state.results.append(result) state.tools_used.append(计算工具 if 计算 in state.task else 搜索工具) return stateMCP的优势在这里体现得很明显工具开发者只需要关注工具功能实现智能体开发者通过标准协议调用不需要关心具体工具的内部实现。5. 多智能体协作实战任务分解与协调多智能体协作是LangGraph的强项。通过定义不同的智能体角色和协作规则可以处理复杂任务。假设我们要开发一个数据分析智能体系统包含三个角色from langgraph import StateGraph from typing import Dict, List class AnalysisState: raw_data: str cleaned_data: str analysis_result: str current_agent: str steps: List[str] def data_collector(state: AnalysisState): 数据收集智能体 # 模拟数据收集 state.raw_data 收集到的原始数据... state.current_agent data_cleaner state.steps.append(数据收集完成) return state def data_cleaner(state: AnalysisState): 数据清洗智能体 if not state.raw_data: state.current_agent data_collector # 需要重新收集数据 return state # 模拟数据清洗 state.cleaned_data state.raw_data.replace(原始, 清洗后) state.current_agent analyzer state.steps.append(数据清洗完成) return state def analyzer(state: AnalysisState): 分析智能体 if not state.cleaned_data: state.current_agent data_cleaner # 需要重新清洗 return state # 模拟分析 state.analysis_result f分析报告: 基于{state.cleaned_data} state.current_agent end state.steps.append(分析完成) return state # 构建多智能体工作流 builder StateGraph(AnalysisState) builder.add_node(collector, data_collector) builder.add_node(cleaner, data_cleaner) builder.add_node(analyzer, analyzer) builder.set_entry_point(collector) # 定义条件流转 def route_after_collector(state: AnalysisState): return state.current_agent def route_after_cleaner(state: AnalysisState): return state.current_agent builder.add_conditional_edges(collector, route_after_collector) builder.add_conditional_edges(cleaner, route_after_cleaner) builder.add_edge(analyzer, end) graph builder.compile()这个例子展示了多智能体协作的几个关键点角色分工每个智能体专注特定任务状态传递通过共享状态传递工作成果条件流转根据当前状态决定下一步执行哪个智能体错误处理检查前置条件必要时回退到前一个步骤实际项目中每个智能体节点可能会集成MCP工具比如数据收集调用网络爬虫工具数据分析调用统计计算工具。6. 实战项目开发一个智能研究助手现在我们把前面学的概念整合成一个完整项目智能研究助手。这个助手能根据用户问题自动搜索资料、分析信息、生成报告。6.1 项目结构设计research_assistant/ ├── mcp_servers/ # MCP工具服务器 │ ├── web_search.py # 网络搜索工具 │ ├── document_ai.py # 文档分析工具 │ └── report_gen.py # 报告生成工具 ├── agents/ # 智能体定义 │ ├── researcher.py # 研究智能体 │ ├── analyzer.py # 分析智能体 │ └── writer.py # 写作智能体 ├── workflows/ # 工作流定义 │ └── research_flow.py # 研究流程 └── main.py # 主程序6.2 MCP工具实现先实现核心的MCP工具# mcp_servers/web_search.py from mcp import Server, Tool import requests Tool async def search_web(query: str, max_results: int 5) - str: 搜索网络获取相关信息 # 实际项目会调用搜索引擎API # 这里用模拟数据演示 results [ f结果1: 关于{query}的权威资料, f结果2: {query}的最新研究, f结果3: {query}实践指南 ] return \n.join(results[:max_results]) Tool async def get_page_content(url: str) - str: 获取网页内容 try: response requests.get(url, timeout10) return response.text[:5000] # 限制内容长度 except: return 获取页面内容失败 # 更多工具...6.3 智能体节点实现定义研究流程中的各个智能体# agents/researcher.py from typing import Dict, Any from mcp_client import MCPSession class ResearchState: question: str search_results: List[str] analysis: str report: str current_step: str async def research_agent(state: ResearchState): 研究智能体负责信息收集 async with MCPSession(http://localhost:8001) as session: # 搜索相关信息 search_results await session.call_tool( search_web, {query: state.question, max_results: 3} ) state.search_results search_results.split(\n) state.current_step analyze return state6.4 工作流整合用LangGraph把各个智能体连接起来# workflows/research_flow.py from langgraph import StateGraph from agents.researcher import research_agent, ResearchState from agents.analyzer import analyze_agent from agents.writer import write_agent def create_research_workflow(): builder StateGraph(ResearchState) # 添加节点 builder.add_node(research, research_agent) builder.add_node(analyze, analyze_agent) builder.add_node(write, write_agent) # 设置流程 builder.set_entry_point(research) builder.add_edge(research, analyze) builder.add_edge(analyze, write) return builder.compile() # 使用工作流 async def run_research(question: str): workflow create_research_workflow() initial_state ResearchState(questionquestion) result await workflow.ainvoke(initial_state) return result[report]6.5 运行和测试启动MCP服务器和测试工作流# 启动MCP工具服务器 python mcp_servers/web_search.py --port 8001 python mcp_servers/document_ai.py --port 8002 python mcp_servers/report_gen.py --port 8003 # 运行研究助手 python main.py --question 人工智能的最新发展趋势7. 常见问题排查与性能优化实际开发中会遇到各种问题这里总结几个典型场景的排查思路。7.1 工作流卡住或循环执行现象工作流一直在几个节点间循环无法结束。排查步骤检查状态对象的current_step或类似控制字段是否正确更新确认条件边的判断逻辑是否覆盖所有可能情况添加调试日志输出每个节点执行前后的状态变化# 添加调试信息 def debug_node(state): print(f当前节点: {state.current_step}) print(f状态内容: {state.__dict__}) # ... 正常处理逻辑 return state7.2 MCP工具调用失败现象工具调用返回错误或超时。排查顺序确认MCP服务器是否正常启动和监听检查工具参数格式是否符合MCP协议要求验证网络连接和端口访问查看MCP服务器日志了解具体错误# MCP调用错误处理 async def safe_tool_call(session, tool_name, params): try: result await session.call_tool(tool_name, params) return result except Exception as e: print(f工具调用失败: {tool_name}, 错误: {e}) return None7.3 内存或性能问题现象处理大量数据时内存占用过高或速度慢。优化建议对于大文件处理使用流式处理而不是一次性加载全部数据合理设置工作流超时时间避免长时间阻塞考虑异步执行耗时工具调用定期清理状态对象中不再需要的历史数据# 内存优化示例 class OptimizedState: def cleanup_old_data(self): 清理历史数据释放内存 if len(self.search_results) 10: self.search_results self.search_results[-5:] # 只保留最近5条7.4 工作流调试技巧开发阶段可以使用可视化工具查看工作流执行过程# 生成工作流图 graph builder.compile() graph.write_png(workflow.png) # 需要安装graphviz # 逐步执行调试 state ResearchState(questiontest) for step in range(10): # 限制最大步数防止无限循环 result graph.invoke(state) print(f步骤{step}: {result.current_step}) if result.current_step end: break state result8. 生产环境部署建议当项目从开发转向生产时有几个关键点需要注意。8.1 配置管理不要硬编码服务器地址、API密钥等配置# config.py import os from typing import Optional class Config: MCP_SEARCH_HOST: str os.getenv(MCP_SEARCH_HOST, localhost) MCP_SEARCH_PORT: int int(os.getenv(MCP_SEARCH_PORT, 8001)) MCP_ANALYSIS_HOST: str os.getenv(MCP_ANALYSIS_HOST, localhost) MCP_ANALYSIS_PORT: int int(os.getenv(MCP_ANALYSIS_PORT, 8002)) property def search_server_url(self) - str: return fhttp://{self.MCP_SEARCH_HOST}:{self.MCP_SEARCH_PORT}8.2 错误处理和重试机制生产环境必须有完善的错误处理from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) async def robust_tool_call(session, tool_name, params): 带重试的工具调用 try: return await session.call_tool(tool_name, params) except Exception as e: logger.error(f工具调用失败: {e}) raise8.3 监控和日志添加详细的日志记录工作流执行情况import logging import json logger logging.getLogger(research_assistant) class LoggingState(ResearchState): def to_log_dict(self): 转换为可日志记录格式 return { question: self.question, current_step: self.current_step, steps_count: len(self.steps) } async def logged_agent(state: LoggingState): logger.info(f开始执行节点: {state.current_step}) logger.debug(f状态: {json.dumps(state.to_log_dict())}) # ... 正常处理逻辑 logger.info(f节点完成: {state.current_step}) return state8.4 性能考量根据实际负载考虑部署方案低并发场景单进程部署使用异步处理中等并发使用进程池或容器化部署高并发场景考虑分布式工作流引擎或消息队列对于资源消耗大的工具调用可以添加限流机制from asyncio import Semaphore # 限制并发工具调用数量 tool_semaphore Semaphore(5) async def limited_tool_call(session, tool_name, params): async with tool_semaphore: return await session.call_tool(tool_name, params)LangGraph和MCP的组合为智能体开发提供了强大的工作流编排和工具标准化能力。实际项目中建议先从小规模原型开始确保单任务流程稳定后再扩展复杂功能。最关键的是理解状态管理的思想和MCP协议的价值而不是盲目追求技术栈的复杂度。