AI Agent开发实战路径图：LangChain、LangGraph、RAG与MCP协同架构

1. 这不是一份“资料清单”，而是一张AI Agent开发者的实战地图

你搜过“AI Agent 学习资料整理”——页面刷出来几百个标题，PDF、GitHub仓库、知乎长文、B站合集，点开一个，开头是“什么是Agent？”，接着是“Agent = LLM + Tool + Memory + Planning”，再往下翻，突然跳到LangChain的AgentExecutor初始化代码，中间没解释为什么用这个类、不用那个类；又或者看到RAG部分，直接甩出Chroma和OpenAIEmbeddings两行代码，但没说清楚：为什么选Chroma而不是FAISS？Embedding维度怎么影响召回率？向量库建好后，query进来时到底发生了几层匹配？这些断层，就是初学者卡住的地方。我带过十几支从零起步的AI工程团队，发现90%的人不是学不会，而是被“资料堆”淹没了——资料本身没问题，问题在于它们默认你已经站在某个台阶上，而你其实连楼梯在哪都没看清。

这份整理，不叫“资料汇总”，它是一份按真实开发动线铺开的AI Agent能力成长路径图。核心关键词就五个：AI Agent、LangChain、LangGraph、RAG、MCP，但它们不是并列关系，而是层层嵌套的“能力模块”。比如LangChain不是独立框架，它是把LLM调用、工具绑定、记忆管理这些基础动作封装成可复用积木的“胶水层”；LangGraph则是在LangChain之上，解决“复杂决策流如何编排”的问题——当你需要让Agent先查天气、再比价、最后生成旅行建议时，LangChain的单步Agent就力不从心了，必须用LangGraph画状态机；RAG也不是一个插件，它是解决LLM“不知道你公司内部数据”的刚需方案，但直接套模板会掉进“召回不准、重排失效、上下文溢出”三重坑；MCP（Model Control Protocol）更不是新玩具，它是2024年突然冒头的协议级标准，目标是让不同厂商的Agent能像HTTP调用网页一样互相调用工具——就像当年REST API统一了服务间通信，MCP正在试图统一Agent间的协作语言。

适合谁看？如果你正卡在“看了十篇LangChain教程，还是写不出能自动订会议室的Agent”，或者“RAG知识库建好了，但用户问‘上季度华东区销售额’，它却返回了三年前的财报PDF”，又或者“听说LangGraph很厉害，但StateGraph里add_node和add_edge到底该在什么时机调用”，那这份整理就是为你写的。它不假设你懂Python异步、不预设你熟悉图数据库原理，所有技术点都从“这个东西解决什么具体问题”出发，配上我踩过的坑、调参时的真实截图、线上压测的QPS数据——不是教科书，是你的同事下班后给你发的那份带注释的笔记。

2. AI Agent的本质：从“问答机器人”到“数字员工”的能力跃迁

2.1 破除概念幻觉：Agent ≠ 更聪明的Chatbot

很多人第一次接触AI Agent时，下意识把它当成“升级版聊天机器人”——输入问题，输出答案，只是答案更准、更长。这是最大的认知陷阱。真正的AI Agent，核心差异不在“回答质量”，而在自主性（Autonomy）和目标导向性（Goal-directedness）。举个具体例子：

• 传统RAG应用：用户问“帮我写一封辞职信”，系统检索公司《员工手册》中“离职流程”章节，拼接模板生成文本。整个过程是被动响应，用户必须明确说出需求。
• AI Agent：用户只说“我想离职”，Agent立刻启动多步工作流：① 调用HR系统API查当前未完成的审批流程；② 检索《离职交接清单》确认需归还设备；③ 根据用户职级和入职时间，计算N+1补偿金；④ 生成包含法律条款、交接节点、补偿明细的完整辞职包，并邮件发送给直属领导和HRBP。

这里的关键转折点是：Agent自己判断下一步该做什么，而不是等用户指令。它背后有三个刚性能力支撑：

1. 规划能力（Planning）：能将模糊目标（“我想离职”）拆解为可执行子任务（查审批→核清单→算补偿→发邮件）；
2. 工具调用能力（Tool Use）：能动态选择并调用HR系统、邮件服务、计算器等外部工具，且理解每个工具的输入约束（如邮件API要求收件人字段非空）；
3. 状态记忆能力（Stateful Memory）：在执行“查审批”后，结果（“待办：离职面谈预约”）必须被记住，才能在后续步骤中触发面谈提醒。

提示：很多初学者用LangChain写了个“调用天气API+翻译结果”的小Demo，就以为掌握了Agent。这其实是Tool Calling，离真正Agent还差两层：一是没有动态规划（步骤固定），二是没有状态流转（每次调用都是无状态的）。真正的Agent开发，80%精力花在设计状态机和错误恢复逻辑上，而非写prompt。

2.2 四层能力架构：为什么LangChain/LangGraph/RAG/MCP必须协同工作

把AI Agent想象成一辆自动驾驶汽车，它的能力不是单一模块堆砌，而是分层协作的系统工程。我们按“从底层到顶层”拆解这四层：

层级	名称	核心职责	典型技术栈	关键风险
L1 基础设施层	大模型与工具接入	提供LLM推理能力、连接数据库/API/文件系统	OpenAI API、Ollama、FastAPI、SQLAlchemy	模型响应延迟高、工具调用超时、认证密钥泄露
L2 能力编排层	LangChain	封装LLM调用、Prompt模板、记忆管理、工具绑定等通用操作	LLMChain,Tool,ConversationBufferMemory	过度依赖AgentExecutor导致调试困难、内存泄漏、工具链路不可见
L3 流程控制层	LangGraph	定义Agent决策逻辑的状态机，处理分支、循环、异常回滚	StateGraph,add_node,add_edge,interrupt	状态定义混乱（如把用户输入和工具结果混在一个dict）、边条件逻辑耦合、缺乏可视化调试
L4 协同协议层	MCP	定义Agent间交互标准，实现跨系统工具发现与调用	MCP Server、mcp-server-python、mcp-client-js	协议版本不兼容（v0.1 vs v0.2）、工具描述缺失关键参数、安全策略未配置

为什么不能只用LangChain？
LangChain的create_react_agent确实能快速跑通一个带搜索功能的Demo，但它本质是单线程状态机：用户问→Agent思考→调用工具→返回结果。一旦需求变成“如果搜索结果为空，则尝试换关键词重试，最多3次；若仍失败，则转人工客服”，LangChain原生API就捉襟见肘了。这时必须用LangGraph定义search_node、retry_node、fallback_node三个节点，并设置should_retry边条件——这就是L3层的价值。

RAG为何必须嵌入L3层？
常见误区是把RAG当做一个独立模块，先建好知识库，再“塞进”Agent。但实际中，RAG的检索策略必须随Agent状态动态调整。例如：

• 用户问“报销流程”，Agent处于onboarding状态（新员工），应优先检索《新人指南》；
• 同样问题，若用户状态是finance_manager，则应检索《财务审批权限细则》。
  LangGraph的状态对象（State）天然支持携带用户角色、历史交互、当前任务类型等元信息，RAG检索器就能基于这些字段动态选择向量库或调整k值——这才是Agentic RAG的精髓。

MCP解决什么终极问题？
设想一个政务场景：市民App的Agent需要调用“社保查询”、“公积金提取”、“户籍迁移”三个服务。如果每个服务都用私有API对接，当新增“生育津贴申领”时，App侧要重新开发、测试、上线。MCP协议让所有服务提供方统一暴露/tools端点，返回标准化的工具描述（含名称、参数、认证方式）。App Agent只需一次发现（Discovery），就能动态加载新工具——这直接把跨系统集成周期从2周压缩到2小时。

3. LangChain：从胶水层到生产级Agent的避坑指南

3.1 别再用AgentExecutor！生产环境必须重构的三大陷阱

LangChain官方文档里，AgentExecutor是入门首选，但我在三个政务项目中亲眼见过它引发的线上事故：

• 事故1（内存泄漏）：某市12345热线Agent使用ConversationBufferMemory，运行72小时后内存占用达16GB，原因是AgentExecutor内部缓存了所有中间步骤的完整prompt和tool call日志，且未提供清理接口；
• 事故2（超时雪崩）：天气API响应慢（>5s），AgentExecutor默认重试3次，导致单次请求耗时15s+，QPS从200骤降至30；
• 事故3（调试黑洞）：当Agent返回错误结果时，日志只显示“Execution failed”，无法定位是LLM解析失败、tool参数错误，还是网络超时。

替代方案：用RunnableSequence手写执行链
核心思路是放弃“黑盒执行”，把每一步拆成显式可监控的节点。以“查天气+翻译”为例：

from langchain_core.runnables import RunnableSequence, RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 步骤1：LLM生成结构化查询（避免自由发挥） weather_query_chain = ( {"input": RunnablePassthrough()} | prompt_template # 精心设计的few-shot prompt，强制输出{"city": "北京", "unit": "celsius"} | llm | JsonOutputParser() # 解析为dict，非字符串 ) # 步骤2：调用天气API（带熔断和重试） def call_weather_api(state): try: # 使用tenacity库实现指数退避重试 @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1)) def _fetch(): response = requests.get( f"https://api.weather.com/v3/weather/forecast/daily", params={"geocode": state["city"], "format": "json"}, timeout=3 # 强制3秒超时 ) return response.json() return _fetch() except Exception as e: logger.error(f"Weather API failed: {e}") return {"error": "service_unavailable"} # 步骤3：翻译结果（用轻量级模型，非LLM） def translate_result(state): # 调用本地fasttext模型，毫秒级响应 return fasttext_translate(state["weather_data"]["forecast"], "zh->en") # 组装执行链（每步可单独打点监控） agent_chain = ( weather_query_chain | RunnableLambda(call_weather_api) | RunnableLambda(translate_result) | StrOutputParser() )

优势对比表：

维度	AgentExecutor	手写RunnableSequence
可观测性	日志仅显示“Agent executed”，无中间态	每步可打点（start/end/time/error），接入Prometheus
容错性	任一环节失败即终止，无降级策略	可为每步定制fallback（如API失败时返回缓存数据）
性能	额外序列化/反序列化开销，平均延迟+120ms	直接内存传递，延迟降低至原生水平
调试成本	需阅读源码才能理解执行逻辑	逻辑直白，IDE可逐行debug

实操心得：在政务项目中，我们强制要求所有Agent必须通过RunnableSequence实现，且每个节点需满足：① 输入输出类型明确（Pydantic Model）；② 有超时和重试配置；③ 错误时返回结构化error code（非抛异常）。这看似增加初期工作量，但上线后故障排查时间减少70%。

3.2 记忆管理：别让ConversationBufferMemory毁掉你的Agent

LangChain的ConversationBufferMemory是新手最爱，因为它“开箱即用”——但正是这种便利，埋下了最隐蔽的雷。问题根源在于：它把所有对话历史无差别拼接进prompt，而LLM的上下文窗口是有限的（GPT-4 Turbo 128K已是奢侈）。当用户聊了20轮，buffer里存了5000字，再加一个1000字的RAG检索结果，prompt直接爆满，LLM开始胡言乱语。

真实案例：某银行理财顾问Agent，用户连续询问“产品A收益”、“产品B风险”、“产品C手续费”，第5轮时Agent突然开始推荐不存在的“产品Z”。日志显示：prompt长度已达127K tokens，LLM被迫截断历史，丢失了关键约束条件（“只推荐持牌产品”）。

生产级记忆方案：分层存储+智能摘要
我们采用三级记忆架构：

1. 短期记忆（Short-term）：用Redis Hash存储最近3轮对话（key:agent:{session_id}:short），每次请求只加载这部分，确保prompt可控；
2. 长期记忆（Long-term）：将用户关键偏好（如“厌恶本金损失”、“偏好年化4%以上”）存入PostgreSQL，用向量相似度检索；
3. 上下文摘要（Context Summary）：用专用LLM（如Phi-3-mini）对短期记忆做摘要，生成50字内核心意图（例：“用户关注三款产品收益率对比，倾向保本型”），注入prompt头部。

# 摘要生成函数（轻量模型，100ms内完成） def generate_context_summary(history: List[Dict]) -> str: # 构造极简prompt，避免LLM自由发挥 prompt = f"""请用1句话总结用户核心诉求，不超过50字。历史对话： {history[-3:]}""" summary = phi3_mini.invoke(prompt) return summary.strip() # 注入prompt时 final_prompt = f"""【当前摘要】{generate_context_summary(short_history)} 【RAG结果】{rag_content} 【用户最新输入】{user_input} 请严格按以下格式回复：..."""

注意：摘要模型必须用小参数量模型（<3B），大模型做摘要是资源浪费。我们实测Phi-3-mini在A10 GPU上吞吐达120 QPS，而GPT-4 Turbo仅12 QPS，成本相差10倍。

4. LangGraph：用状态机思维重构Agent复杂逻辑

4.1 State设计：90%的LangGraph故障源于状态定义错误

LangGraph的StateGraph强大，但它的威力完全取决于State的设计。很多团队直接用dict作为state，很快陷入混乱：

# ❌ 危险示例：无结构的dict state state = { "input": "查北京天气", "intermediate_steps": [...], # 工具调用历史 "memory": {...}, # 对话记忆 "user_profile": {...} # 用户画像 }

问题在于：

• 类型不安全：state["input"]可能是str、None、甚至list，运行时才报错；
• 职责不清：intermediate_steps既存工具结果，又存LLM输出，调试时无法区分来源；
• 扩展困难：新增“多语言支持”需改所有节点，因为state结构被硬编码。

正确做法：用Pydantic V2定义强类型State

from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any class AgentState(BaseModel): # 用户原始输入（不可变） input: str = Field(..., description="用户原始提问") # 当前任务目标（由规划节点动态生成） goal: str = Field(..., description="当前需达成的具体目标") # 工具调用结果（只存成功结果，失败走error字段） tool_results: Dict[str, Any] = Field(default_factory=dict) # LLM生成的结构化输出（非自由文本） llm_output: Optional[Dict[str, Any]] = None # 错误信息（结构化，非字符串） error: Optional[Dict[str, str]] = None # {"type": "tool_timeout", "message": "..."} # 业务上下文（如用户ID、会话ID） context: Dict[str, Any] = Field(default_factory=dict) # 初始化state（强制校验） initial_state = AgentState( input="查北京天气", goal="获取北京未来24小时温度和降水概率", context={"user_id": "u123", "session_id": "s456"} )

优势：

• IDE自动补全state.tool_results，避免拼写错误；
• FastAPI可直接用此Model做请求体校验；
• 新增字段时，旧节点无需修改（Pydantic默认忽略未知字段）；
• 序列化时自动过滤None值，减小网络传输体积。

4.2 节点设计：每个Node必须是纯函数，且有明确副作用

LangGraph的add_node注册函数，但很多开发者把DB操作、API调用、LLM调用全塞进一个node，导致：

• 无法单元测试（依赖外部服务）；
• 重试逻辑混乱（LLM失败重试 vs DB失败重试策略不同）；
• 性能瓶颈难定位（是LLM慢，还是DB慢？）。

黄金法则：一个Node只做一件事，且副作用可预测
以“查询天气”为例，拆分为三个Node：

# Node 1：规划（纯逻辑，无副作用） def plan_weather_search(state: AgentState) -> Dict[str, str]: # 基于input和context，生成结构化查询参数 city = extract_city(state.input) or state.context.get("last_known_city", "北京") return {"goal": f"获取{city}未来24小时天气", "city": city} # Node 2：工具调用（有副作用，但可重试） def call_weather_api(state: AgentState) -> Dict[str, Any]: try: # 调用API，返回结构化数据 data = requests.get(f"https://api.example.com/weather?city={state.city}").json() return {"tool_results": {"weather": data}} except Exception as e: return {"error": {"type": "api_failure", "message": str(e)}} # Node 3：生成回复（纯LLM，无IO） def generate_response(state: AgentState) -> Dict[str, str]: # 用RAG结果+tool_results生成最终文本 prompt = f"""你是一个天气助手，请用中文简洁回复。 城市：{state.city} 天气数据：{state.tool_results.get('weather', {})} 请直接输出温度和降水概率，不要解释。""" response = llm.invoke(prompt) return {"llm_output": {"text": response.content}} # 注册节点（清晰分离） workflow.add_node("plan", plan_weather_search) workflow.add_node("call_weather", call_weather_api) workflow.add_node("respond", generate_response)

边条件（Edge）设计技巧：
不要用lambda x: x["error"] is not None这种模糊判断。定义明确的error type：

# 明确的边条件 def should_retry(state: AgentState) -> str: if state.error and state.error["type"] in ["api_failure", "network_timeout"]: return "retry" elif state.error and state.error["type"] == "invalid_input": return "fallback" else: return "next" workflow.add_conditional_edges( "call_weather", should_retry, { "retry": "call_weather", # 自循环重试 "fallback": "respond_fallback", # 降级节点 "next": "respond" # 正常流程 } )

实操心得：我们在政务项目中规定，所有Node必须通过@traceable装饰器接入LangSmith，记录输入/输出/耗时。当某节点平均耗时突增，能立即定位是LLM响应慢（需切模型），还是API超时（需调优重试策略）。

5. RAG实战：从“知识库能搜到”到“答案精准可靠”的质变

5.1 多路召回：为什么单靠向量检索必然失败？

RAG效果差，90%的原因不是embedding模型不好，而是召回策略太单一。向量检索本质是“语义相似度匹配”，但用户提问方式千奇百怪：

用户提问	向量检索问题	关键词检索优势	结构化查询优势
“去年Q3华东区销售额”	“Q3”和“华东”在embedding空间距离远	“华东”、“Q3”、“销售额”精确匹配	直接查sales WHERE region='华东' AND quarter='2023-Q3'
“报销发票抬头错了怎么办？”	“抬头”和“发票”语义相关，但“错了”是负面词，削弱相似度	“发票抬头”、“错误”、“怎么办”组合匹配	匹配FAQ文档中“发票信息错误”章节

解决方案：三路召回+重排序（Multi-Stage Retrieval）
我们在线上系统采用三级召回：

1. 关键词召回（BM25）：用Elasticsearch，覆盖精确术语（如“增值税专用发票”、“工单号”）；
2. 向量召回（Hybrid Search）：用Chroma的hybrid_search，同时计算BM25分数和向量相似度，加权融合；
3. 结构化召回（SQL Query）：对表格型知识（如政策文件、审批流程），用LLM生成SQL，直接查数据库。

# 三路召回整合函数 def multi_retrieve(query: str, top_k: int = 5) -> List[Document]: # 路径1：BM25关键词召回 bm25_docs = es_client.search( index="policy_docs", body={"query": {"multi_match": {"query": query, "fields": ["title^3", "content"]}}} ) # 路径2：向量+BM25混合召回 hybrid_docs = chroma_collection.query( query_texts=[query], n_results=top_k, where={"source": "manual"} # 限定知识库范围 ) # 路径3：结构化查询（LLM生成SQL） sql = llm.invoke(f"将'{query}'转为SQL查询，表名：policy_rules，字段：id,title,effect_date") sql_docs = db.execute(sql).fetchall() # 合并去重，按综合分数排序 all_docs = bm25_docs + hybrid_docs + sql_docs return sorted(all_docs, key=lambda x: x.score, reverse=True)[:top_k]

重排序（Rerank）为何不可省略？
向量召回返回的Top5，可能前2个是语义相近但无关的文档（如“华东区”匹配到“华东分公司成立公告”）。我们用Cross-Encoder模型（如bge-reranker-base）对候选文档做精细化打分：

from sentence_transformers import CrossEncoder reranker = CrossEncoder('BAAI/bge-reranker-base') scores = reranker.predict([(query, doc.page_content) for doc in candidates]) # scores是float数组，对应candidates顺序 reranked_docs = [doc for _, doc in sorted(zip(scores, candidates), key=lambda x: x[0], reverse=True)]

实测数据：某政务知识库，单路向量召回准确率62%，三路召回+重排序后达89%。关键提升来自结构化召回——用户问“退休年龄”，向量检索返回一堆政策解读，而SQL直接查出《国发〔2024〕X号》文件中“男性60岁，女性55岁”的原文段落。

5.2 Agentic RAG：让RAG成为Agent的“感官”，而非“附件”

传统RAG是静态的：用户问→检索→生成。Agentic RAG则是动态的：Agent根据当前状态决定是否检索、检索什么、如何使用结果。

典型场景：多跳问答（Multi-hop QA）
用户问：“上海张江科学城的生物医药企业，2023年获得多少笔市级专项资金？”

• 第一跳：检索“张江科学城 生物医药企业名录” → 得到企业列表；
• 第二跳：对每个企业，检索“XX公司 2023年专项资金” → 得到各企业金额；
• 第三跳：汇总计算总金额。

LangGraph实现：

# 定义多跳状态 class RAGState(AgentState): rag_results: List[Dict] = Field(default_factory=list) # 存储各跳结果 current_hop: int = Field(default=0, description="当前是第几跳") # 跳转节点 def rag_hop(state: RAGState) -> Dict[str, Any]: if state.current_hop == 0: # 第一跳：找企业名录 docs = vector_retriever.invoke("张江科学城 生物医药企业名录") return {"rag_results": docs, "current_hop": 1} elif state.current_hop == 1: # 第二跳：对每个企业查资金 companies = [d.metadata["name"] for d in state.rag_results] all_funds = [] for company in companies[:3]: # 限制并发数 funds = vector_retriever.invoke(f"{company} 2023年专项资金") all_funds.extend(funds) return {"rag_results": all_funds, "current_hop": 2} else: # 第三跳：汇总 total = sum(float(d.metadata.get("amount", "0")) for d in state.rag_results) return {"llm_output": {"text": f"总计{total}万元"}} workflow.add_node("rag_hop", rag_hop)

注意：多跳RAG必须设置hop上限（如max_hops=3），否则可能无限循环。我们在政务系统中强制加入hop_count字段，每次调用递增，超过阈值自动fallback。

6. MCP协议：Agent协作的“HTTP协议”正在诞生

6.1 MCP不是新框架，而是Agent世界的“TCP/IP”

MCP（Model Control Protocol）常被误解为另一个LLM框架，实则它是Agent间通信的底层协议规范，类比互联网发展史：

• 1970年代：各大学有自己的网络（ARPANET、CYCLADES），互不联通；
• 1983年：TCP/IP协议统一，互联网诞生；
• 2024年：MCP协议出现，目标是让不同厂商的Agent能互相调用工具。

MCP的核心价值：解耦工具提供方与使用方
传统方式：

• Agent A想用“社保查询”工具 → 开发者需研究社保局API文档 → 写适配代码 → 测试认证 → 上线。
  MCP方式：
• 社保局部署MCP Server，暴露/tools端点；
• Agent A发起HTTP GET/tools，得到标准化工具列表：

{ "tools": [ { "name": "get_social_security_info", "description": "查询用户社保缴纳记录", "parameters": { "type": "object", "properties": { "id_card": {"type": "string", "description": "身份证号"}, "year": {"type": "integer", "description": "查询年份"} }, "required": ["id_card"] } } ] }

• Agent A直接调用/tool/get_social_security_info，传参即可。

为什么MCP比REST API更进一步？
REST API定义的是“如何调用”，MCP定义的是“如何发现和理解工具”。它强制要求：

• 工具必须有机器可读的description（非人类可读）；
• 参数必须有type和description，支持LLM自动生成调用代码；
• 支持capabilities字段声明工具能力（如"requires_auth": true）。

6.2 在LangGraph中集成MCP：让Agent学会“上网找工具”

MCP Client不是独立模块，而是LangGraph的一个Node。我们封装了一个MCPToolNode：

from mcp.client import ClientSession from mcp.types import ToolResult class MCPToolNode: def __init__(self, server_url: str): self.client = ClientSession(server_url) async def invoke(self, state: AgentState) -> Dict[str, Any]: # 1. 发现可用工具 tools = await self.client.list_tools() # 2. 基于用户问题，LLM选择最匹配的工具 tool_selection_prompt = f"""从以下工具中，选择最匹配用户问题的1个。只返回工具名。 用户问题：{state.input} 工具列表：{[t.name for t in tools]} """ selected_tool_name = await llm.ainvoke(tool_selection_prompt) # 3. 生成工具调用参数（LLM填充） tool = next(t for t in tools if t.name == selected_tool_name) param_prompt = f"""为工具'{selected_tool_name}'生成调用参数。用户问题：{state.input} 工具参数定义：{tool.parameters} 只返回JSON，不要解释。""" params = await llm.ainvoke(param_prompt) # 4. 调用MCP Server result = await self.client.call_tool(selected_tool_name, params) return {"tool_results": {selected_tool_name: result}} # 在LangGraph中使用 workflow.add_node("mcp_tool", MCPToolNode("https://mcp-social-security.gov.cn")) workflow.add_edge("plan", "mcp_tool") # 规划后自动调用MCP工具

生产注意事项：

• 安全隔离：MCP Server必须部署在DMZ区，Agent调用需经API网关鉴权；
• 协议版本：强制检查MCP Server的/health端点返回的protocol_version，不兼容则拒绝调用；
• 降级策略：当MCP Server不可用时，自动切换到本地缓存的工具描述（mcp-tools-cache.json）。

实操心得：我们在某省政务平台试点MCP，接入社保、公积金、户籍三个部门的MCP Server。上线后，新业务上线周期从平均14天缩短至2天——因为Agent开发者不再需要对接每个部门的私有API，只需理解MCP协议。

7. 常见问题与排查技巧实录：那些文档里不会写的坑

7.1 LangChain常见问题速查表

问题现象	根本原因	排查步骤	解决方案
AgentExecutor卡死，CPU 100%	ConversationBufferMemory在长对话中反复拼接字符串，触发Python GC风暴	1.ps aux | grep python看进程；2.py-spy record -p <pid>生成火焰图	改用ConversationSummaryMemory，或手写RunnableSequence
工具调用返回None，但API实际成功	LangChain工具装饰器@tool未正确处理异步函数	1. 检查工具函数是否加了async；2. 查看tool.py源码中_run方法是否支持await	同步工具用@tool，异步工具用@tool(aio=True)
RAG检索结果为空，但文档存在	Chroma collection未设置embedding_function，或embedding维度与模型不匹配	1.collection.get()看文档是否存入；2.collection.peek()检查embedding长度	创建collection时显式传入embedding_function=OpenAIEmbeddings()
LLM输出格式错乱（如JSON缺逗号）	Prompt未强制LLM输出JSON Schema，或temperature过高	1. 用JsonOutputParser包装LLM；2. 设置temperature=0.1	在prompt末尾加：“请严格按以下JSON Schema输出：{schema}”

7.2 LangGraph调试三板斧

第一斧：可视化状态流
LangGraph自带get_graph().draw_mermaid_png()，但生产环境需实时监控。我们在每个Node入口加日志：

def debug_node(state: AgentState) -> Dict[str, Any]: logger.info(f"Node 'plan' entered. State keys: {list(state.model_dump().keys())}") logger.debug(f"Full state: {state.model_dump_json()}") # ...业务逻辑 logger.info(f"Node 'plan' exited. New goal: {state.goal}")

第二斧：状态快照回放
当线上出现诡异行为，我们保存state.model_dump_json()到S3，用脚本重放：

# replay.py from langgraph.graph import StateGraph from my_workflow import workflow, AgentState # 加载故障时的状态快照 with open("state_snapshot.json") as f: state_dict = json.load(f) state = AgentState(**state_dict) # 从指定节点开始重放 result = workflow.invoke(state, {"recursion_limit": 10}) print(result)

第三斧：边条件断点
add_conditional_edges的lambda函数难调试，改为命名函数：

def edge_condition(state: AgentState) -> str: logger.debug(f"Edge condition check: error={state.error}, tool_results={state.tool_results}") if state.error: return "handle_error" return "next" workflow.add_conditional_edges("call_weather", edge_condition, {...})

7.3 RAG效果差的5个隐藏原因

1. 文档切分不当：用\n\n切分PDF，导致表格被割