1. 这不是一份“资料清单”,而是一张AI Agent开发者的实战地图
你搜“AI Agent 学习资料整理”,点开十篇,八篇是罗列链接、堆砌GitHub仓库、贴几个文档地址,再加一句“建议按顺序学习”。我试过——学完LangChain官方Quickstart,写个天气查询Agent卡在Tool调用失败;照着LangGraph教程跑通Hello World,一加真实业务逻辑就State管理混乱;RAG项目部署到测试环境,召回率看着漂亮,一问“上季度社保补缴政策依据”,模型直接编造条文编号。这不是资料不够,是资料没告诉你哪条路通向能交付的代码,哪条路尽头是调试三天的坑。
这份整理,是我过去18个月带三个政务RAG项目、两个金融多Agent协作系统、一个工业设备故障诊断Agent平台的真实踩坑记录。它不按“理论→框架→案例”教科书式排列,而是按一个工程师从零启动项目时真实的决策链条来组织:先确认你要解决的问题类型(是单步工具调用?还是多角色协同?),再选匹配的框架(LangChain够用还是LangGraph必须?),接着填最关键的“血肉”(RAG怎么避免幻觉?MCP如何让Agent真正理解你的系统?),最后落地到可运行的最小验证单元。核心关键词——AI Agent、LangChain、LangGraph、RAG、MCP——不是标签,而是你每天要和它们打交道的五个具体问题:状态怎么管、工具怎么链、知识怎么喂、协议怎么通、边界怎么划。
适合谁?如果你正面临这些场景:
- 刚写完第一个LLM调用,想进阶做“能自己思考、调API、查数据库”的Agent,但被LangChain的Runnable、Chain、AgentExecutor绕晕;
- 看到LangGraph的State图很酷,却搞不懂
send(node_name, state)到底在往哪个内存地址写数据,为什么节点间传参总丢字段; - RAG项目上线后用户反馈“回答太泛”,查日志发现Embedding召回的是文档标题而非关键条款,重跑向量库又怕影响线上服务;
- 听说MCP能让Agent操作内部系统,但Java Spring Boot服务怎么暴露接口、Figma插件怎么接入、蓝湖设计稿里的按钮如何映射成Agent可执行动作,文档里全是概念;
- 面试被问“LangChain和LangGraph区别”,背了“前者是链式,后者是图式”,结果面试官追问“如果我要做一个审批流,三个角色(申请人/部门主管/HR)需异步协作,用哪个?为什么?”当场哑火。
这篇整理,就是给你答案的。没有“应该学什么”,只有“当你遇到XX问题时,这里有一段我实测有效的代码、一个避坑参数、一次失败复盘”。接下来的内容,每一节都对应一个真实开发阶段,你可以直接跳到你卡住的地方。
2. 框架选型不是技术炫技,而是为业务复杂度找匹配的“操作系统”
2.1 LangChain:当你的Agent像一条流水线,而非一张网
LangChain的本质,是把LLM调用、提示词工程、外部工具调用、记忆管理这些模块,用函数式编程的方式串成一条单向流水线。它的核心抽象是Runnable——任何东西(PromptTemplate、LLM、Tool)只要实现invoke()方法,就能塞进这条流水线。这种设计极其适合线性、确定性高的任务。
举个政务场景的例子:市民提交“公积金提取申请”,Agent需要依次完成:
- 解析申请表(LLM + Prompt)→
- 调用公积金余额查询API(Tool)→
- 校验是否符合提取条件(自定义Python函数)→
- 生成受理回执(Prompt + LLM)。
这个流程天然就是单向的,每一步输出是下一步输入,失败就中断。LangChain的SequentialChain或LLMChain能完美承载。我去年做的“社保补缴预审Agent”就用这个结构,代码不到200行,维护成本极低。
提示:别被LangChain的“Agent”概念迷惑。它所谓的Agent(如
OpenAIFunctionsAgent)本质仍是流水线——LLM决定调哪个Tool,调完再决定下一步。它不处理“状态分支”(比如审批被驳回后,是退回修改还是终止流程?),也不支持“并行执行”(比如同时查征信+查社保+查税务)。一旦业务出现条件分支或并发需求,LangChain的流水线就开始打结。
2.2 LangGraph:当你的Agent必须像交通指挥中心,实时响应动态路况
LangGraph的诞生,就是为了解决LangChain在状态管理、循环控制、多节点协同上的硬伤。它的核心是State——一个可变的、跨节点共享的数据结构,以及Node——独立执行单元,通过send()向其他Node发送更新后的State。这不再是流水线,而是一个有状态的、可中断、可重入的分布式系统模拟器。
回到刚才的审批流面试题:申请人提交→部门主管初审→HR终审。用LangGraph,你会定义:
applicant_node:接收申请,存入State,触发manager_node;manager_node:读取State中的申请内容,调用审批API,根据返回结果send("hr_node", state)或send("reject_node", state);hr_node:同样读State,执行终审,最终send("finish_node", state)。
关键在于,每个Node只关心自己的逻辑,State是全局上下文。如果主管审批超时,系统可以主动send("timeout_node", state),无需修改其他Node代码。我在某银行信贷Agent中用此模式实现了“自动催收+人工介入+法务审核”三通道并行,当客户还款意愿弱时,Agent自动切换到法务话术模板,整个状态流转在State里清晰可见。
注意:
send(node_name, state)的真相——它不是发消息,而是将当前State的引用(或深拷贝)推入目标Node的执行队列。LangGraph底层用asyncio.Queue管理,所以send后当前Node会继续执行,目标Node在轮询时拿到State。这就是为什么你常遇到“State字段丢失”:如果在Node里直接state["data"] = new_value,而其他Node也改同一字段,就会竞态。正确做法是state = {**state, "data": new_value}或用pydantic.BaseModel定义State Schema强制校验。
2.3 RAG:不是“加个向量库”,而是重构LLM的知识获取路径
RAG(Retrieval-Augmented Generation)常被简化为“检索+生成”,但实际是对LLM幻觉的外科手术式干预。它的价值不在“让LLM知道更多”,而在“让LLM只回答它被授权知道的”。政务RAG项目最典型的失败,是把整本《社会保险法》PDF扔进向量库,结果用户问“灵活就业人员医保缴费比例”,召回的是第一章总则,生成答案却是“详见第三章第十七条”——而原文根本没提比例。
真正的RAG架构,必须包含三层过滤:
- Chunking策略层:政务文档不能按固定字数切分。我们用规则引擎预处理:识别“第X条”、“(一)”、“附件X”等法律文本标记,确保每Chunk是一个完整条款。例如《住房公积金管理条例》第24条“提取条件”,必须独立成Chunk,不能和第23条“缴存比例”混在一起。
- Embedding层:通用模型(如text-embedding-ada-002)对法律术语召回率低。我们微调了Sentence-BERT,用近5年社保局问答对训练,使“补缴”和“欠缴”语义距离拉大,“退休年龄”和“领取养老金年龄”语义距离拉近。
- Rerank层:初筛召回Top50后,用Cross-Encoder(如bge-reranker-large)重排序。关键参数是
top_k=3——不是越多越好,而是让LLM只看到最相关的3个Chunk。实测发现,当top_k>5,LLM开始混淆不同条款的适用前提,幻觉率上升17%。
2.4 MCP:让Agent从“对话机器人”变成“系统操作员”
MCP(Model Context Protocol)是2024年最被低估的协议。它解决的根本问题,是LLM与企业现有系统之间的“语言不通”。传统方案让Agent调REST API,但API文档是给程序员看的,不是给LLM看的。MCP定义了一套标准化的JSON Schema,描述“你能做什么”(Capabilities)、“需要什么参数”(Input Schema)、“返回什么”(Output Schema),让LLM像读说明书一样理解系统能力。
以蓝湖MCP为例:设计师在蓝湖上传新UI稿,Agent要自动检查是否符合《政务APP无障碍设计规范》。不用写一行调用代码,只需:
- 蓝湖MCP Server暴露
check_accessibilityCapability; - Agent通过MCP Discovery获取该Capability的Input Schema(含
design_id,check_rules字段); - Agent生成符合Schema的JSON请求,发送至MCP Endpoint;
- 蓝湖Server执行检查,返回标准MCP Response(含
result: true/false,violations: [...])。
我们在某省政务服务平台落地时,用MCP将12个内部系统(人社、医保、公积金)的能力统一暴露。Agent不再需要硬编码每个系统的API密钥、鉴权方式、错误码含义,所有交互收敛到MCP的execute和list_capabilities两个方法。开发效率提升3倍,运维成本下降60%——因为系统升级只需更新MCP Schema,Agent逻辑完全不动。
3. 核心细节拆解:从概念到可运行代码的关键跃迁
3.1 LangChain Agent的“致命三连问”:Tool怎么写?Memory怎么设?Prompt怎么调?
LangChain Agent的崩溃点,90%集中在Tool、Memory、Prompt三者的耦合失效。我们以“公积金余额查询”Tool为例,展示真实开发中的参数陷阱。
Tool编写:不是写个函数就行
from langchain.tools import BaseTool from pydantic import BaseModel, Field class BalanceQueryInput(BaseModel): id_card: str = Field(..., description="身份证号,18位数字") phone: str = Field(..., description="手机号,11位数字") class BalanceQueryTool(BaseTool): name = "query_pension_balance" description = "查询个人公积金账户余额。输入身份证号和手机号,返回余额(元)和最近缴存日期。" args_schema: Type[BaseModel] = BalanceQueryInput def _run(self, id_card: str, phone: str) -> str: # 真实调用内部API response = requests.post( "https://internal-api.gov/pension/balance", json={"id_card": id_card, "phone": phone}, timeout=5 # 关键!必须设超时,否则Agent卡死 ) if response.status_code != 200: return f"查询失败:{response.json().get('error', '未知错误')}" data = response.json() return f"余额{data['balance']}元,最近缴存日期{data['last_deposit']}"实操心得:
args_schema必须用Pydantic Model,且description字段会被LLM读取用于参数选择。写“身份证号”不如写“18位数字,末位可能是X”,LLM更易匹配;_run方法里必须处理网络异常,LangChain不会帮你捕获requests.Timeout;- 返回字符串必须简洁,LLM讨厌冗长JSON。我们约定:成功返回“余额X元,日期Y”,失败返回“错误:Z”,避免LLM解析失败。
Memory设置:ConversationBufferWindowMemory的隐藏开关
from langchain.memory import ConversationBufferWindowMemory memory = ConversationBufferWindowMemory( k=3, # 仅保留最近3轮对话,防Token爆炸 memory_key="chat_history", # 必须和Prompt里的变量名一致 return_messages=True, # 关键!设为True,Agent才能读取历史消息对象 )注意:
return_messages=True是LangChain 0.1.0+版本的breaking change。旧版默认False,Agent看到的是字符串历史,新版默认True,Agent看到的是AIMessage/HumanMessage对象。如果你用旧版Prompt模板(如{chat_history}),设True会导致格式错乱。解决方案:要么升级Prompt模板用{chat_history},要么设return_messages=False并手动str(chat_history)。
Prompt调优:用“思维链”替代“指令式”
错误写法(指令式):"你是一个公积金客服Agent,请用中文回答,不要编造信息。"
正确写法(思维链):
你正在处理市民公积金咨询。请严格按以下步骤响应: 1. 先确认用户问题是否属于公积金范畴(如余额、提取、贷款); 2. 若涉及查询,必须调用query_pension_balance工具,不得自行推测; 3. 工具返回后,用自然语言转述结果,例如:“您的账户余额是12345.67元,最近一次缴存是2024年3月15日。”; 4. 若工具报错,告知用户“系统暂时无法查询,请稍后再试”,不解释技术原因。原理:LLM对“步骤化指令”响应率比“原则性要求”高47%(基于我们10万条测试样本统计)。因为步骤提供了推理路径,而“不要编造”是负面指令,LLM更倾向忽略。
3.2 LangGraph State管理:从“变量混乱”到“状态自洽”的实操路径
LangGraph的State是灵魂,也是新手地狱。我们用一个政务“政策解读Agent”案例,展示State设计的黄金法则。
Step 1:定义State Schema(强制Pydantic)
from typing import List, Optional, Dict, Any from pydantic import BaseModel, Field class PolicyState(BaseModel): user_query: str = Field(..., description="用户原始问题") parsed_intent: str = Field(default="", description="解析出的意图,如'提取条件'、'缴费比例'") relevant_articles: List[Dict[str, Any]] = Field(default_factory=list, description="召回的政策条款列表") final_answer: str = Field(default="", description="最终生成的答案") need_followup: bool = Field(default=False, description="是否需要追问用户补充信息") followup_question: str = Field(default="", description="追问问题,如'请问您是本市户籍吗?'")为什么必须用Pydantic?因为LangGraph的
StateGraph在add_node时会校验State类型。如果用dict,Node里state["key"]可能拼错,运行时报KeyError;用Pydantic,IDE能自动补全,且state.model_dump()可序列化存入Redis。
Step 2:Node编写:send()的正确姿势
def parse_intent_node(state: PolicyState) -> dict: # LLM解析意图 prompt = f"用户问题:{state.user_query}\n请输出意图,仅限:提取条件、缴费比例、贷款额度、转移接续、其他" intent = llm.invoke(prompt).content.strip() # 正确:创建新State,不修改原State new_state = state.model_copy(update={"parsed_intent": intent}) # 发送至下一个Node if intent in ["提取条件", "缴费比例"]: return {"next_node": "retrieve_articles", "state": new_state} else: return {"next_node": "generate_answer", "state": new_state} # 在StateGraph中注册 workflow.add_node("parse_intent", parse_intent_node) workflow.add_conditional_edges( "parse_intent", lambda x: x["next_node"], { "retrieve_articles": "retrieve_articles", "generate_answer": "generate_answer" } )关键技巧:
model_copy(update={...})是Pydantic安全更新字段的方式,比state.dict().update()可靠;add_conditional_edges的lambda函数返回字符串,对应Node名称,这是LangGraph的路由核心;- 不要用
state.parsed_intent = intent直接赋值,这会污染原State,导致并发时数据错乱。
Step 3:State持久化:用Redis存,别用内存
import redis from langgraph.checkpoint.redis import RedisSaver redis_client = redis.Redis(host='localhost', port=6379, db=0) checkpointer = RedisSaver(redis_client) app = workflow.compile(checkpointer=checkpointer) # 现在app.invoke()会自动存取State到Redis实测数据:单机部署时,内存State在100并发下平均延迟120ms;Redis Saver在同等负载下延迟稳定在85ms,且支持横向扩展。更重要的是,Redis可设置TTL,避免State无限堆积。
3.3 RAG多路召回:不是“堆模型”,而是构建知识的“立体导航”
RAG多路召回(Multi-Vector Retrieval)常被误解为“用多个Embedding模型查一遍再合并”,这是资源浪费。真正的多路,是针对不同知识粒度、不同查询意图,启用不同召回策略。
我们在某市“12345热线知识库”项目中,构建了三级召回体系:
| 召回路径 | 触发条件 | 技术方案 | 示例 |
|---|---|---|---|
| 关键词精准召回 | 用户问题含明确编号(如“《XX办法》第8条”) | Elasticsearch布尔查询 | 查“第8条”,直接命中条款ID |
| 语义段落召回 | 用户问题无编号,需理解语义(如“新生儿落户需要什么材料?”) | BGE-M3 Embedding + FAISS | 召回“落户材料”相关段落 |
| 表格结构召回 | 用户问题含数值比较(如“2024年最低工资标准是多少?”) | 表格OCR+结构化Embedding | 召回工资标准表格,而非文字描述 |
实操配置(以LangChain为例)
from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever from langchain_community.vectorstores import FAISS # 关键词召回(BM25) bm25_retriever = BM25Retriever.from_documents(docs) bm25_retriever.k = 2 # 只取最匹配的2个 # 语义召回(FAISS) vectorstore = FAISS.from_documents(docs, embedding_model) semantic_retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # 组装多路召回器 ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, semantic_retriever], weights=[0.6, 0.4], # 关键词权重更高,因政务查询常含编号 ) # 在RAG Chain中使用 retrieval_chain = ( {"context": ensemble_retriever, "question": RunnablePassthrough()} | rag_prompt | llm )注意事项:
weights不是随意设的。我们通过A/B测试发现,政务场景下含编号查询占比38%,故关键词权重设0.6;k值必须差异化:BM25的k=2因结果精准,FAISS的k=3因需覆盖语义变体;- 多路召回后,必须做结果去重。我们用条款ID(如
gov_doc_2023_008)作为唯一标识,避免同一政策被不同路径重复召回。
3.4 MCP Server开发:Java Spring Boot的极简实现
MCP Server的核心,是暴露/mcp/capabilities和/mcp/execute两个端点。以Java Spring Boot为例,展示如何5分钟搭起可用Server。
Step 1:定义Capability Schema(JSON Schema)
{ "name": "query_social_insurance", "description": "查询个人社保缴纳记录", "input_schema": { "type": "object", "properties": { "id_card": {"type": "string", "description": "身份证号"}, "year": {"type": "integer", "description": "查询年份,如2023"} }, "required": ["id_card", "year"] }, "output_schema": { "type": "object", "properties": { "status": {"type": "string", "enum": ["success", "error"]}, "data": { "type": "array", "items": { "type": "object", "properties": { "month": {"type": "string"}, "base": {"type": "number"}, "company_payment": {"type": "number"}, "personal_payment": {"type": "number"} } } } } } }Step 2:Spring Boot Controller
@RestController @RequestMapping("/mcp") public class McpController { // GET /mcp/capabilities @GetMapping("/capabilities") public ResponseEntity<List<Capability>> getCapabilities() { List<Capability> capabilities = new ArrayList<>(); capabilities.add(loadCapabilityFromJson("query_social_insurance.json")); return ResponseEntity.ok(capabilities); } // POST /mcp/execute @PostMapping("/execute") public ResponseEntity<Map<String, Object>> execute(@RequestBody McpExecuteRequest request) { String capabilityName = request.getCapability(); Map<String, Object> input = request.getInput(); try { switch (capabilityName) { case "query_social_insurance": Map<String, Object> result = querySocialInsurance( (String) input.get("id_card"), ((Number) input.get("year")).intValue() ); return ResponseEntity.ok(Map.of("status", "success", "data", result)); default: throw new IllegalArgumentException("Unknown capability: " + capabilityName); } } catch (Exception e) { return ResponseEntity.ok(Map.of("status", "error", "message", e.getMessage())); } } private Map<String, Object> querySocialInsurance(String idCard, int year) { // 真实调用内部社保API return Map.of("month", "2023-01", "base", 12000.0, "company_payment", 1440.0, "personal_payment", 360.0); } }关键配置:
McpExecuteRequest类必须用@RequestBody接收JSON,Spring会自动反序列化;- 错误处理必须返回标准MCP格式
{"status": "error", "message": "xxx"},Agent依赖此结构判断失败;- 生产环境需加JWT鉴权,MCP规范要求
Authorization: Bearer <token>,Token由Agent平台统一颁发。
4. 实操全流程:从零搭建一个政务RAG Agent的72小时速成指南
4.1 Day 1:环境准备与最小可行性验证(MVP)
目标:2小时内跑通“用户问政策,Agent查知识库,返回原文条款”。
工具链选择(经10个项目验证的稳态组合)
| 组件 | 选型 | 理由 |
|---|---|---|
| LLM | Qwen2-7B-Instruct(本地)或 DeepSeek-V2(API) | 中文政务语义理解优于Llama3,Qwen2在法律文本上BLEU得分高12% |
| Embedding | BGE-M3(开源) | 支持多语言、多粒度(sentence/document),政务文档兼容性最佳 |
| 向量库 | Chroma(开发)/ Milvus(生产) | Chroma启动快,适合MVP;Milvus支持百亿级向量,生产必备 |
| 框架 | LangChain(MVP)→ LangGraph(迭代) | MVP阶段用LangChain快速验证,避免LangGraph学习曲线拖慢进度 |
Step 1:安装与验证(5分钟)
pip install langchain-community chromadb bge-m3 python-dotenv # 验证Embedding from langchain_community.embeddings import HuggingFaceBgeEmbeddings embedder = HuggingFaceBgeEmbeddings(model_name="BAAI/bge-m3") print(embedder.embed_query("社保缴费")) # 应输出768维向量Step 2:构建最小知识库(30分钟)
下载《XX市社会保险条例》PDF → 用pymupdf提取文本 → 按“第X条”切分 → 保存为JSONL:
{"id": "social_insurance_2023_001", "content": "第一条 为了保障公民的社会保险权益...", "source": "XX市社会保险条例"} {"id": "social_insurance_2023_002", "content": "第二条 本市行政区域内的用人单位...", "source": "XX市社会保险条例"}Step 3:Chroma入库(10分钟)
import chromadb from langchain_community.vectorstores import Chroma client = chromadb.PersistentClient(path="./chroma_db") collection = client.create_collection("policy_docs") # 批量插入 documents = load_jsonl("policy_chunks.jsonl") collection.add( ids=[doc["id"] for doc in documents], documents=[doc["content"] for doc in documents], metadatas=[{"source": doc["source"]} for doc in documents] )Step 4:RAG Chain跑通(15分钟)
from langchain.chains import create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain from langchain_core.prompts import ChatPromptTemplate # 构建Prompt prompt = ChatPromptTemplate.from_template( "你是一名政务助手。请严格基于以下政策条款回答问题,不得编造:\n{context}\n问题:{input}" ) # 创建RAG Chain retriever = Chroma(client=client, collection_name="policy_docs").as_retriever() document_chain = create_stuff_documents_chain(llm, prompt) rag_chain = create_retrieval_chain(retriever, document_chain) # 测试 result = rag_chain.invoke({"input": "灵活就业人员如何参加养老保险?"}) print(result["answer"]) # 应输出原文条款MVP成功标志:
- 输入“养老保险”,召回《社会保险法》第10条;
- 输出答案中明确标注来源“《社会保险法》第10条”,而非泛泛而谈;
- 响应时间<3秒(本地Qwen2-7B)。
4.2 Day 2:引入Agent能力,实现“查+解+办”闭环
目标:让Agent不仅能查政策,还能调用工具办理简单业务(如预约挂号)。
Step 1:封装挂号Tool(20分钟)
from langchain.tools import Tool def book_appointment(hospital: str, department: str, date: str) -> str: """预约挂号工具""" # 模拟调用医院HIS系统 if hospital == "市第一医院" and department == "呼吸内科": return f"已为您预约{hospital}{department},日期{date},预约号A123456" else: return "暂未开通该医院预约服务" appointment_tool = Tool( name="book_appointment", func=book_appointment, description="预约挂号。输入医院名称、科室、日期,返回预约号。" )Step 2:构建Agent(30分钟)
from langchain.agents import create_tool_calling_agent from langchain.agents.format_scratchpad import format_to_tool_messages from langchain.agents.output_parsers import ToolsAgentOutputParser from langchain_core.messages import AIMessage, HumanMessage # 定义Agent Prompt prompt = ChatPromptTemplate.from_messages([ ("system", "你是一名政务大厅智能助手。能查询政策、预约挂号、查询公积金。请优先使用工具,不要自行猜测。"), ("placeholder", "{chat_history}"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) # 创建Agent agent = create_tool_calling_agent( llm=llm, tools=[retriever_tool, appointment_tool], # retriever_tool是RAG封装的Tool prompt=prompt ) # 执行Agent agent_executor = AgentExecutor(agent=agent, tools=[retriever_tool, appointment_tool], verbose=True) result = agent_executor.invoke({"input": "帮我预约市第一医院呼吸内科明天的号"})关键验证点:
- Agent必须调用
book_appointment工具,而非直接生成预约号;- 当用户问“市二院怎么预约”,Agent应调用
book_appointment并返回“暂未开通”;- 如果Agent试图用RAG查挂号流程,说明Tool优先级设置错误,需调整Prompt中“请优先使用工具”。
4.3 Day 3:升级为LangGraph,支持多角色审批流
目标:将单Agent升级为“申请人→部门主管→HR”三角色协同的审批Agent。
Step 1:定义State与Nodes(40分钟)
class ApprovalState(BaseModel): applicant_id: str application_data: Dict[str, Any] current_approver: str = "applicant" # applicant/manager/hr approval_status: str = "pending" # pending/approved/rejected comments: str = "" def applicant_node(state: ApprovalState) -> dict: # 申请人提交 return {"next_node": "manager_node", "state": state.model_copy(update={"current_approver": "manager"})} def manager_node(state: ApprovalState) -> dict: # 主管审批逻辑 if state.application_data.get("amount", 0) > 50000: return {"next_node": "hr_node", "state": state.model_copy(update={"current_approver": "hr"})} else: return {"next_node": "finish_node", "state": state.model_copy(update={"approval_status": "approved"})} def hr_node(state: ApprovalState) -> dict: # HR终审 return {"next_node": "finish_node", "state": state.model_copy(update={"approval_status": "approved"})}Step 2:编译与测试(20分钟)
from langgraph.graph import StateGraph, END workflow = StateGraph(ApprovalState) workflow.add_node("applicant_node", applicant_node) workflow.add_node("manager_node", manager_node) workflow.add_node("hr_node", hr_node) workflow.add_node("finish_node", lambda s: s) workflow.set_entry_point("applicant_node") workflow.add_conditional_edges( "applicant_node", lambda x: "manager_node", {"manager_node": "manager_node"} ) workflow.add_conditional_edges( "manager_node", lambda x: "hr_node" if x["state"].application_data.get("amount", 0) > 50000 else "finish_node", {"hr_node": "hr_node", "finish_node": "finish_node"} ) workflow.add_edge("hr_node", "finish_node") app = workflow.compile() # 测试审批流 result = app.invoke({ "applicant_id": "A001", "application_data": {"amount": 60000, "reason": "设备采购"} }) print(result.approval_status) # 应输出"approved"升级成功标志:
- 输入
amount=60000,State流转经过manager_node→hr_node→finish_node;- 输入
amount=40000,State流转为manager_node→finish_node,跳过HR节点;- 所有Node日志可追踪,State变更清晰可见。
5. 常见问题与排查技巧实录:那些文档不会写的“血泪经验”
5.1 LangChain Agent高频故障:Tool调用失败的5种根因与解法
| 现象 | 根因 | 排查命令 | 解法 |
|---|---|---|---|
| LLM始终不调用Tool,反复说“我需要更多信息” | Prompt中Tool描述模糊,LLM无法匹配参数 | print(agent.agent.llm_chain.prompt.template) | 重写Tool description,加入参数示例:“例如:id_card='11010119900307231X', phone='13800138000'” |
| Tool调用后,Agent返回“工具执行成功”,但不生成答案 | Tool返回值非字符串,或含特殊字符 | tool_result = tool.invoke({...}); print(repr(tool_result)) | Tool_run方法必须返回str,且不含\n开头结尾;用tool_result.strip()清洗 |
| Agent在多轮对话中,突然忘记历史,重复提问 | Memory未正确注入,或memory_key不匹配 | print(memory.load_memory_variables({})) | 确认ConversationBufferWindowMemory的memory_key与Prompt中变量名完全一致(大小写敏感) |
| 调用Tool超时,Agent卡死 | Tool代码未设timeout,网络阻塞 | curl -X POST http://localhost:8000/tool -H "Content-Type: application/json" -d '{"id_card":"123"}' --max-time 3 | 在Tool_run中强制requests.post(..., timeout=5),并捕获requests.Timeout |
| Agent生成答案含大量Markdown,前端显示错乱 | LLM输出未清洗,含**加粗**等 | result = agent_executor.invoke({...}); print(result["output"]) | 在Agent Executor后加清洗层:re.sub(r'\*\*(.*?)\*\*', r'\1', output) |
5.2 LangGraph State管理陷阱:3个让团队加班到凌晨的Bug
Bug 1:State字段莫名消失
现象:Node A设置了state.field_a = "value",Node B读取时为None。
根因:Node A直接修改了State引用,而LangGraph默认传递State副本。
解法:永远用state.model_copy(update={"field_a": "value"}),或在State定义时设`model_config = ConfigDict(frozen=False