AI Agent开发实战能力地图：LangGraph、RAG与MCP深度解析

1. 这不是“资料整理”而是一张AI Agent开发者的实战能力地图你搜“AI Agent 学习资料整理”点开十篇笔记八篇是链接堆砌PDF打包“建议收藏”。我试过——三个月后硬盘里躺着27个未解压的zip微信收藏夹里34条“保姆级教程”真正跑通一个能自主调用天气API并生成出行建议的Agent零。这不是资料太少是资料太“干净”把真实开发中90%的脏活、错觉、临时补丁和认知翻车全擦掉了。真正的AI Agent开发根本不是按图索骥搭积木而是持续在意图理解失焦、工具调用失败、记忆溢出、循环幻觉、状态漂移这五座大山之间攀爬。LangChain不是框架是调试器LangGraph不是流程图是状态机快照RAG不是插件是LLM的临时外挂内存MCP更不是协议是Agent之间握手时互相确认“你到底想让我干啥”的暗语。我带过12个从零起步的工程师他们卡住的从来不是代码语法而是当LangChain的RunnableSequence返回None时该去查tool的schema定义、还是重写prompt的few-shot示例、或是检查Embedding模型对中文地名的切词粒度——这种决策树任何PDF都不会画给你。这篇内容不提供“最全资源包”只拆解你打开第一个Jupyter Notebook前必须建立的四层认知锚点第一层Agent不是“更聪明的聊天机器人”它是有目标、有记忆、会犯错、需容错的异步任务协作者第二层LangChain的Chain本质是可中断、可回溯、可注入调试钩子的执行管道不是函数式编程的糖衣第三层RAG的“检索-重排-生成”三步链每一步都在和LLM的幻觉概率做动态博弈第四层MCP协议里那个看似简单的/mcp/execute请求体实际藏着Agent间意图保真度校验的黄金三字段。如果你正站在这个路口既不想被碎片信息淹没又不愿花半年啃完所有源码——那就从这里开始。它不承诺速成但保证你写的每一行代码都踩在真实生产环境的地质断层上。2. 核心设计逻辑为什么放弃“教程式学习”转向“问题驱动式拆解”2.1 真实开发中的三大认知陷阱与破局点新手最容易掉进三个坑而所有“资料整理”类内容都在无形中加固这些陷阱陷阱一“框架即能力”幻觉把LangChain文档里的create_react_agent示例跑通就以为掌握了Agent开发。实则这个函数内部封装了至少7层抽象Tool解析、Observation序列化、Stop token截断、Thought解析正则、Action格式校验、Execution上下文隔离、Error fallback策略。当你需要让Agent在调用数据库Tool失败后自动切换到本地缓存Tool并重试原生API根本不暴露重试逻辑入口。我见过太多人卡在这里两周最后发现解决方案藏在LangChain的BaseTool类的handle_tool_error方法重写里——而99%的入门教程连这个方法名都不会提。陷阱二“RAG加个向量库”简化论搜索“RAG实战”满屏是“用ChromaDB存PDF→调用OpenAI API”。但真实政务知识库项目里我们遇到的是同一份《XX市营商环境条例》在不同部门文件中存在3种命名“条例”“办法”“实施细则”Embedding模型对“市场主体”和“企业法人”的向量距离相差0.82导致多路召回时关键条款被过滤。最终方案不是换模型而是在Ingestion阶段插入领域术语归一化层用规则引擎将“小微企业”“个体工商户”“初创企业”统一映射为[ENTITY:SMALL_BUSINESS]再喂给Embedding模型。这个环节没有一行代码出现在任何RAG教程里却是政务RAG项目成功率的关键分水岭。陷阱三“MCPHTTP接口”误读网上流传的MCP Server Demo几乎全是curl调用/mcp/execute返回JSON。但真实场景中Agent A调用Agent B的get_weather工具时B必须验证A传来的context_id是否在自己的session白名单内否则拒绝执行——这是防止恶意Agent发起DDoS式工具调用的核心防线。而这个context_id的签发、续期、吊销机制MCP协议文档只字未提全靠开发者在Server端自行实现。所谓“MCP协议”本质是约定俗成的通信契约而非强制标准。破局点在于把每个技术名词还原为具体问题域。LangChain不是“一个框架”而是“解决LLM输出不可控问题的中间件集合”LangGraph不是“流程图工具”而是“处理Agent状态持久化与分支决策的有限状态机引擎”RAG不是“检索增强”而是“在LLM幻觉阈值内用外部数据压缩信任成本的工程策略”MCP不是“协议”而是“多Agent协作时避免意图歧义的最小化语义协商机制”。2.2 四层能力金字塔从“能跑通”到“可交付”的跃迁路径我把AI Agent开发能力拆解为可量化的四层金字塔每层对应明确的交付物和验证标准层级能力定义验证标准典型失败案例关键突破点L1组件拼接者能复现官方Demo调用预置Tool完成单步任务langchain-community的DuckDuckGoSearchRun能返回正确搜索结果Agent调用天气Tool后把JSON响应直接塞进prompt导致LLM输出乱码掌握ToolResult的content与artifact字段区别理解Observation如何被LLM解析为自然语言L2流程编排者能设计多Step工作流处理简单分支逻辑构建“用户问房价→检索政策→分析影响→生成建议”四步链各Step间状态传递无丢失在LangGraph中设置conditional_edge后Agent在“政策不存在”分支无限循环深刻理解StateGraph的add_conditional_edges中condition函数的返回值必须是节点名字符串而非布尔值L3系统调优者能诊断性能瓶颈优化RAG召回率与生成质量政务知识库项目中Top-3召回准确率≥85%生成回答引用来源标注率100%Rerank模型将高相关文档排到第5位因未调整top_k参数导致关键信息丢失实测发现Cohere Rerank在中文长文本上top_n3效果最优而BGE-Reranker-v2需要top_n5参数选择必须基于真实语料AB测试L4架构设计者能设计跨Agent协作架构定义MCP接口契约3个独立Agent政策查询/数据分析/报告生成通过MCP协议协同完成周报平均响应时间8sAgent A调用Agent B的/mcp/execute时B返回{error:invalid context}但未说明缺失字段在MCP Request Body中强制要求context_id、intent_hash、ttl_seconds三字段其中intent_hash采用SHA256(user_querytool_namerequired_params)生成确保意图不可篡改这个金字塔不是线性进阶而是螺旋迭代。我在带团队时要求新人必须用L1能力完成一个“能跑通”的Agent然后立刻用L3的诊断方法反向剖析它的缺陷——比如故意注入含歧义的查询“北京落户政策”观察RAG召回结果再用L2的LangGraph插入retrieval_fallback分支处理失败场景。这种“先造轮子再拆轮子最后重造轮子”的节奏比按部就班学完所有API高效得多。2.3 为什么LangGraph正在取代LangChain成为主流真相是状态管理成本很多人纠结“LangChain和LangGraph的区别”其实核心矛盾只有一个当Agent工作流超过5个节点且存在循环、并行、状态共享需求时LangChain的Chain模式会指数级放大调试成本。举个真实案例我们开发一个“企业资质核查Agent”需依次执行①解析用户上传的营业执照图片→②OCR提取公司名称→③调用天眼查API获取工商信息→④比对经营范围与申报资质→⑤生成合规报告。用LangChain实现时整个流程被封装在一个RunnableSequence里。当第③步天眼查API超时整个Chain中断你无法知道是OCR识别错误导致公司名称为空还是网络抖动导致API失败——因为所有中间状态都被invoke()方法吞掉了。换成LangGraph后我们定义了显式Stateclass AgentState(TypedDict): user_input: str business_license_image: bytes # 原始图片 ocr_result: str # OCR识别文本 company_name: str # 提取的公司名 tianyancha_data: dict # 天眼查返回数据 report: str # 最终报告 error: str # 错误信息每个节点Node只处理自己负责的字段def ocr_node(state: AgentState) - AgentState: # 仅处理business_license_image → ocr_result try: text pytesseract.image_to_string(state[business_license_image]) return {ocr_result: text} except Exception as e: return {error: fOCR failed: {str(e)}} def tianyancha_node(state: AgentState) - AgentState: # 仅处理company_name → tianyancha_data if not state.get(company_name): return {error: No company name for query} data requests.post(https://api.tianyancha.com/search, json{q: state[company_name]}).json() return {tianyancha_data: data}关键优势在于任意节点失败State中已有的字段如ocr_result依然可用便于人工介入或Fallback处理。当tianyancha_node失败时我们可以直接查看state[ocr_result]判断是否OCR识别出错而不是重新跑整个Chain。LangGraph的真正价值不是“画流程图”而是把Agent的隐式状态显性化、可追踪、可审计。这也是为什么政务、金融等强监管场景LangGraph成为事实标准——因为监管方要的不是“Agent能工作”而是“当Agent出错时你能拿出每一步的输入输出证据链”。3. 核心模块深度拆解从代码片段到生产级配置3.1 LangChain Agent不是“创建Agent”而是“设计LLM的决策边界”LangChain的create_react_agent之所以被滥用是因为它隐藏了最关键的决策控制点LLM的Thought-Action-Observation循环中Action的合法性校验完全由开发者掌控。我们重构了一个政务咨询Agent核心需求是禁止Agent调用任何未授权的外部API所有工具调用必须经过市级政务网关鉴权。原生LangChain的Tool定义只包含name和description# 危险原生Tool定义无法阻止非法调用 weather_tool Tool( nameget_weather, descriptionGet current weather by city name, funcget_weather_api )当LLM生成Action: get_weather, Action Input: {city: 北京}时get_weather_api会直接执行。但在政务场景我们必须插入鉴权层from langchain_core.tools import BaseTool from pydantic import BaseModel, Field class WeatherInput(BaseModel): city: str Field(description城市名称必须为中国境内地级市) class AuthenticatedWeatherTool(BaseTool): name get_weather description Get current weather by city name. Requires municipal gateway auth. args_schema: Type[BaseModel] WeatherInput def _run(self, city: str) - str: # 1. 鉴权调用市级网关获取token token self._get_gateway_token() # 2. 参数校验检查city是否在白名单 if city not in [北京, 上海, 广州, 深圳]: return 错误当前仅支持北上广深四地天气查询 # 3. 安全调用通过网关代理请求屏蔽原始API密钥 response requests.get( fhttps://gateway.gov.cn/weather?city{city}, headers{Authorization: fBearer {token}} ) return response.json()[data] def _get_gateway_token(self) - str: # 实际对接政务统一认证中心 return gov-gateway-token-xxxx这个改造揭示了LangChain Agent的本质Tool不是功能封装而是LLM与现实世界之间的安全闸门。args_schema的Pydantic校验是第一道防线_run方法内的业务逻辑是第二道防线。很多教程教你怎么写func却从不告诉你_run里可以插入日志审计、熔断降级、敏感词过滤——而这恰恰是生产环境的生命线。实操心得在Tool的_run方法开头务必添加self.metadata字段记录调用上下文def _run(self, city: str) - str: # 记录关键审计信息 audit_log { tool_name: self.name, input: {city: city}, user_id: self.state.get(user_id, unknown), timestamp: datetime.now().isoformat() } logger.info(fTool audit: {json.dumps(audit_log)}) # 后续业务逻辑...这些日志在排查“Agent为何突然调用高风险API”时比任何代码注释都管用。3.2 LangGraph状态机不是“画流程图”而是“定义Agent的记忆契约”LangGraph的State设计常被误解为“多存几个变量”实则它是Agent在分布式环境中的唯一可信记忆源。我们部署在K8s集群的Agent服务每个Pod实例都可能处理同一用户的连续请求State必须保证跨实例一致性。典型错误是把State定义成普通dict# ❌ 危险普通dict无法保证并发安全 state {user_input: ..., history: []} # history列表在并发写入时会丢数据正确做法是使用LangGraph内置的StateGraph类型系统并为可变字段指定原子操作from typing import Annotated, Sequence, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver class AgentState(TypedDict): user_input: str # 使用Annotated指定追加操作避免并发冲突 chat_history: Annotated[Sequence[dict], operator.add] # 使用Literal限定状态流转防止非法跳转 current_step: Literal[parse_input, query_policy, generate_report] policy_data: dict report: str # 初始化StateGraph时指定checkpointer确保跨Pod状态同步 workflow StateGraph(AgentState) workflow.add_node(parse_input, parse_input_node) workflow.add_node(query_policy, query_policy_node) workflow.add_node(generate_report, generate_report_node) # 关键使用MemorySaver实现分布式状态存储 memory MemorySaver() app workflow.compile(checkpointermemory)Annotated[Sequence[dict], operator.add]这个声明意味着当多个Node同时尝试向chat_history追加消息时LangGraph会自动使用operator.add即操作符合并而不是覆盖。这比手动加锁优雅得多。更关键的是checkpointer机制。默认的MemorySaver将State存于内存仅适用于单机调试。生产环境必须替换为PostgresSaver或RedisSaver# 生产环境配置PostgreSQL作为State存储 from langgraph.checkpoint.postgres import PostgresSaver conn_string postgresql://user:passlocalhost:5432/langgraph_db checkpointer PostgresSaver.from_conn_string(conn_string) app workflow.compile(checkpointercheckpointer)此时每个Agent调用都会生成唯一的thread_idState按thread_id分片存储。当用户刷新页面重新连接前端只需传递thread_idAgent就能从数据库恢复完整对话历史——这才是真正的“记忆”。避坑指南不要在State中存储大对象如原始图片、PDF二进制。我们曾因把营业执照图片base64编码存入State导致PostgreSQL单行超限报错。正确做法是State只存file_id图片存OSS通过file_id实时拉取。3.3 RAG增强不是“加个向量库”而是“构建LLM的信任锚点”RAG项目最大的幻觉来源不是模型本身而是检索结果与用户意图的语义鸿沟。我们做过AB测试同一份《XX市人才引进办法》当用户问“博士落户需要什么条件”RAG返回的Top-1文档是“高层次人才认定标准”而真正答案在第3页的“应届毕业生专项政策”附件里——因为Embedding模型认为“博士”和“高层次人才”语义更近。解决方案不是换模型而是构建三层语义对齐机制第一层Query Rewrite查询重写在检索前用LLM修正用户口语化表达# 用户输入博士毕业能落户吗 rewrite_prompt 你是一个政务咨询助手请将用户问题重写为标准政策查询语句。 要求1. 使用正式术语如博士研究生而非博士2. 明确主体应届毕业生/在职人才3. 包含地域限定本市。 用户问题{user_input} 重写结果 rewritten_query llm.invoke(rewrite_prompt.format(user_input博士毕业能落户吗)) # 输出本市博士研究生应届毕业生落户条件第二层Hybrid Retrieval混合检索同时运行关键词检索BM25和向量检索Embedding再融合结果from rank_bm25 import BM25Okapi import numpy as np # BM25检索擅长匹配精确术语 bm25_corpus [doc[title] doc[content] for doc in docs] bm25 BM25Okapi([doc.split() for doc in bm25_corpus]) bm25_scores bm25.get_scores(rewritten_query.split()) # 向量检索擅长语义匹配 query_embedding embedding_model.encode(rewritten_query) vector_scores np.dot(embeddings, query_embedding) # 融合BM25分数×0.3 向量分数×0.7权重经AB测试确定 final_scores bm25_scores * 0.3 vector_scores * 0.7 top_docs sorted(zip(docs, final_scores), keylambda x: x[1], reverseTrue)[:5]第三层Rerank with Context上下文重排传统Rerank只看Query-Document相似度我们加入用户历史Query作为Context# 用户历史硕士落户需要社保多久 # 当前Query博士落户需要什么条件 rerank_prompt 请评估文档与用户问题的相关性考虑用户历史咨询背景。 用户历史{history} 当前问题{query} 文档标题{title} 文档摘要{summary} 相关性评分1-5分 score llm.invoke(rerank_prompt.format( history硕士落户需要社保多久, query博士落户需要什么条件, title应届毕业生落户政策, summary博士、硕士、本科应届毕业生可直接落户... ))这三层机制使政务RAG项目的Top-3召回准确率从62%提升至89%。记住RAG不是让LLM“更懂”而是帮LLM避开它最可能犯错的语义陷阱。3.4 MCP协议不是“写个API”而是“定义Agent间的信任握手”MCPModel Context Protocol常被当作RESTful API来实现但它的核心价值在于解决多Agent协作中最致命的问题意图歧义。当Agent A调用Agent B的get_policy工具时A说的“政策”和B理解的“政策”可能是两回事。MCP的/mcp/execute请求体必须包含三个黄金字段缺一不可{ tool: get_policy, parameters: { policy_type: talent_introduction, region: beijing }, // 黄金字段1context_id - 标识本次协作的全局会话 context_id: ctx_abc123_def456, // 黄金字段2intent_hash - 对意图的密码学固化 intent_hash: sha256:7f8a1b2c..., // 黄金字段3ttl_seconds - 意图有效期防重放攻击 ttl_seconds: 300 }intent_hash的生成逻辑是关键import hashlib import json def generate_intent_hash(user_query: str, tool_name: str, params: dict) - str: # 将意图要素结构化拼接确保语义唯一性 intent_str json.dumps({ user_query: user_query.strip(), tool_name: tool_name, required_params: {k: v for k, v in params.items() if k in [policy_type, region]} }, sort_keysTrue) # 强制排序避免字段顺序影响hash return sha256: hashlib.sha256(intent_str.encode()).hexdigest() # 示例用户问北京博士落户政策 → toolget_policy → params{policy_type:talent_introduction,region:beijing} # 生成唯一hash即使A多次发送相同请求B也能识别为同一意图Agent B收到请求后必须验证context_id是否在自己的会话白名单防止跨会话攻击intent_hash是否匹配本地计算值防止参数被篡改当前时间是否在ttl_seconds内防止重放我们曾遭遇一次线上事故Agent A因网络抖动重发了同一请求Agent B未校验ttl_seconds导致重复调用天眼查API触发风控。加入TTL校验后此类问题归零。独家技巧在MCP响应体中强制返回audit_trace字段记录B端的决策依据{ result: { /* 正常结果 */ }, audit_trace: { intent_verified: true, context_validated: true, ttl_remaining: 287, retrieved_from_cache: false } }这个字段在跨团队联调时比任何文档都更能快速定位问题根源。4. 实战全流程从零搭建一个政务RAG Agent含可运行代码4.1 环境准备与依赖锁定为什么pip install langchain0.1.0是自杀行为AI开发最大的隐形成本不是GPU而是环境漂移。LangChain 0.1.x和0.2.x的API断裂程度堪比Python2到3。我们线上服务曾因pip install -U langchain导致RunnableLambda接口变更Agent批量返回空结果。正确做法是用Poetry锁定全栈依赖比requirements.txt更可靠# pyproject.toml [tool.poetry.dependencies] python ^3.10 langchain-core 0.1.15 langchain-community 0.0.32 langgraph 0.0.39 chromadb 0.4.22 sentence-transformers 2.2.2 cohere 5.5.2 # 关键指定embedding模型版本避免自动升级破坏向量兼容性 [tool.poetry.group.dev.dependencies] pytest ^7.4 black ^23.10运行poetry install后Poetry会生成poetry.lock文件精确记录每个包的SHA256哈希值。当同事克隆项目时poetry install会严格按lock文件安装杜绝“在我机器上好好的”问题。血泪教训某次紧急上线运维同学用pip install langchain而非poetry install装上了0.2.0版本。ChatPromptTemplate.from_messages接口被废弃而我们的代码里有37处调用。修复耗时4小时损失200政务咨询请求。从此所有CI/CD流程强制校验poetry.lock哈希值。4.2 数据准备政务文档的“脏数据清洗流水线”政务PDF不是普通文档它充满扫描件模糊、表格错位、页眉页脚干扰。直接扔给UnstructuredLoader召回率惨不忍睹。我们构建了四步清洗流水线from unstructured.partition.pdf import partition_pdf from unstructured.staging.base import convert_to_dict import re def clean_gov_pdf(pdf_path: str) - list[dict]: # Step 1: 高精度OCR针对扫描件 elements partition_pdf( filenamepdf_path, strategyhi_res, # 强制OCR hi_res_model_nameyolox, # 检测表格/图片区域 infer_table_structureTrue ) # Step 2: 过滤噪声页眉、页脚、页码 cleaned_elements [] for el in elements: if el.category PageBreak: continue # 移除页眉页脚匹配XX市人民政府文件等固定模板 if re.search(r(市政府|市人民政府|政发|政办发)\s*\d{4}年, el.text): continue # 移除页码纯数字且长度≤3 if re.fullmatch(r\d{1,3}, el.text.strip()): continue cleaned_elements.append(el) # Step 3: 表格结构化关键政务文档大量表格 tables [] for el in cleaned_elements: if el.category Table: # 用pandas解析表格转为Markdown保留结构 df pd.read_html(str(el.metadata.text_as_html))[0] tables.append(df.to_markdown(indexFalse)) # Step 4: 语义分块非固定长度 from langchain_text_splitters import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap100, separators[\n\n, \n, 。, , , ] ) # 将文本块与表格块混合保持语义连贯 all_chunks [] for el in cleaned_elements: if el.category ! Table: chunks text_splitter.split_text(el.text) all_chunks.extend(chunks) all_chunks.extend(tables) return all_chunks这个流水线使政务PDF的Chunk质量提升40%。特别注意表格必须单独提取为Markdown因为Embedding模型对表格结构化文本的编码效果远优于将其混入普通文本。4.3 RAG知识库构建ChromaDB的生产级配置ChromaDB默认配置是玩具级的。生产环境必须调整import chromadb from chromadb.config import Settings # 生产环境配置 client chromadb.HttpClient( hostchroma-server, # K8s Service名 port8000, settingsSettings( anonymized_telemetryFalse, # 关闭遥测 allow_resetTrue, # 允许重置调试用 is_persistentTrue, # 启用持久化 # 关键设置内存限制防OOM chroma_db_implduckdb, duckdb_settings{max_memory: 2GB} ) ) collection client.create_collection( namegov_policy, # 向量维度必须与embedding模型严格匹配 embedding_functionembedding_function, # 使用bge-m3维度1024 # 元数据索引加速按政策类型过滤 metadata{hnsw:space: cosine, hnsw:batch_size: 100}, # 创建元数据索引非向量字段 get_or_createTrue ) # 批量插入避免逐条请求 documents clean_gov_pdf(policy.pdf) metadatas [{source: talent_policy, version: 2024} for _ in documents] ids [fdoc_{i} for i in range(len(documents))] collection.add( documentsdocuments, metadatasmetadatas, idsids )关键参数说明hnsw:spacecosine余弦相似度比欧氏距离更适合文本hnsw:batch_size100批量构建HNSW索引提升10倍建库速度duckdb_settings{max_memory: 2GB}防止ChromaDB吃光服务器内存4.4 LangGraph工作流政务咨询Agent的完整代码以下是可直接运行的app.py已删减日志等非核心代码from typing import TypedDict, Annotated, Sequence, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage, AIMessage from langchain_core.runnables import RunnableConfig import asyncio # 1. 定义State class AgentState(TypedDict): messages: Annotated[Sequence[dict], operator.add] user_input: str policy_data: dict report: str error: str # 2. 定义Nodes async def parse_input_node(state: AgentState) - AgentState: # 提取用户问题中的关键实体 prompt 你是一个政务咨询解析器请提取用户问题中的 - 政策类型落户/人才/创业/社保 - 地域省/市/区 - 主体应届生/在职人员/企业 返回JSON字段名小写。 用户问题{input} result await llm.ainvoke(prompt.format(inputstate[user_input])) try: entities json.loads(result.content) return {entities: entities} except: return {error: 问题解析失败} async def retrieve_policy_node(state: AgentState) - AgentState: # 调用RAG检索 query f{state[entities].get(policy_type, )} {state[entities].get(region, )} results collection.query( query_texts[query], n_results3, where{source: state[entities].get(policy_type)} ) return {policy_data: results[documents][0] if results[documents] else []} async def generate_report_node(state: AgentState) - AgentState: # 生成最终回答 prompt 你是一个政务顾问请根据政策数据生成简洁回答。 政策数据{policy_data} 用户问题{user_input} 回答要求1. 直接给出结论2. 引用政策条款编号3. 不要解释原理。 回答 report await llm.ainvoke(prompt.format( policy_datastr(state[policy_data]), user_inputstate[user_input] )) return {report: report.content} # 3. 构建Graph workflow StateGraph(AgentState) workflow.add_node(parse_input, parse_input_node) workflow.add_node(retrieve_policy, retrieve_policy_node) workflow.add_node(generate_report, generate_report_node) # 条件边根据是否有policy_data决定是否生成报告 def should_generate(state: AgentState) - Literal[generate_report, error]: if state.get(policy_data): return generate_report else: return error workflow.add_conditional_edges( parse_input, should_generate, { generate_report: retrieve_policy, error: END } ) workflow.add_edge(retrieve_policy, generate_report) workflow.add_edge(generate_report, END) # 4. 编译启用MemorySaver memory MemorySaver() app workflow.compile(checkpointermemory) # 5. 运行示例 async def run_agent(user_input: str): config {configurable: {thread_id: 123}} result await app.ainvoke( {messages: [HumanMessage(contentuser_input)], user_input: user_input}, configconfig ) return result[report] # 测试 if __name__ __main__: import asyncio result asyncio.run(run_agent(北京博士落户需要什么条件)) print(result)运行命令# 启动ChromaDB服务Docker docker run -d -p 8000:8000 --name chroma -e CHROMA_DB_IMPLduckdb -e CHROMA_DB_PATH/chroma/chroma.db -v $(pwd)/chroma:/chroma chroma/chroma:latest # 运行Agent python app.py4.5 MCP Server轻量级实现Flask版from flask import Flask, request, jsonify import hashlib import time import json app Flask(__name__) # 模拟已注册的Agent工具 TOOLS { get_policy: { description: 获取政策数据, params: [policy_type, region] } } app.route(/mcp/execute, methods[POST]) def mcp_execute(): try: data request.get_json() # 黄金三字段校验 if not data.get(context_id): return