LangChain工程化本质：从RAG到Agentic RAG的抽象演进

1. 为什么这个问题值得花一整篇长文来拆解“为什么很多 AI 应用最后都会绕回 LangChain”——这不是一个技术选型的闲聊而是过去三年我在交付 27 个企业级 AI 项目、亲手重构过 11 套生产环境 RAG/Agent 系统后被客户、同事、甚至自己反复问到的“灵魂拷问”。它背后藏着的不是某个库的文档写得有多好而是一整套AI 工程化落地过程中不可回避的抽象失配问题大模型原生能力自由生成、上下文理解和真实业务系统需状态管理、需工具调用、需错误恢复、需可观测、需权限隔离之间那道越来越宽的鸿沟。你可能已经试过直接用 OpenAI SDK 写一个问答接口也搭过基于 Chroma 的简易知识库甚至用 FastAPI 封装过几个 LLM 调用。但当需求从“查一份产品说明书”升级到“根据销售合同自动比对付款条款触发法务审批流生成风险摘要”你会发现代码开始失控——if-else 嵌套成迷宫重试逻辑散落在五六个函数里日志看不出是哪次检索失败导致了最终回答失真上线后运维连“当前这个请求走了几轮检索”都查不到。这时候LangChain 不是“又一个框架”而是你手边唯一能快速把混沌业务逻辑映射成可调试、可测试、可监控、可灰度发布的标准构件的“工程胶水”。它高频出现在热搜词里LangChain、Agent、RAG、工作流、知识库不是因为营销做得好而是因为所有绕开它的尝试最终都在生产环境的复杂性面前撞了墙。我见过最典型的案例一家做智能客服的团队用纯自研调度器硬扛了 8 个月直到某次大促期间因单个 API 超时未熔断导致整个对话状态机雪崩3000 用户会话卡死在“正在思考…”切换到 LangChain LangGraph 后仅用 3 天就完成了带超时、重试、降级、链路追踪的全链路重构。这不是框架的胜利是抽象层级对齐现实复杂度的必然结果。所以这篇文章不讲“LangChain 安装教程”也不堆砌 API 列表。我要带你钻进那些深夜 debug 的现场看清楚当一个真实业务需求比如“用户问‘上季度华东区销售额是多少’系统要自动查 BI 数据库聚合画图解释趋势”落到工程师桌上时LangChain 是如何用一套看似简单的概念Chain、Tool、Retriever、AgentExecutor、StateGraph一层层拆解掉隐藏在“智能”表象下的工程熵增难题。你会看到它真正不可替代的价值藏在Runnable的统一接口里在CallbackHandler的可插拔钩子里在StateGraph对有状态循环的原生支持中——这些设计不是为炫技而是为让 AI 应用像数据库连接池、HTTP 中间件一样成为现代软件栈里一块可信赖的、有明确行为边界的基础设施。2. 核心设计思路LangChain 解决的到底是什么问题2.1 从“调用模型”到“构建系统”的范式跃迁很多人初学 LangChain 时最大的困惑是“我直接openai.ChatCompletion.create()不就能调用模型了吗为什么还要多一层封装” 这个疑问本身就暴露了对 AI 应用本质的误判。我们来对比两个真实场景场景 A单次调用用户输入“帮我写一封辞职信”你调用一次 LLM返回文本。成功。场景 B生产系统用户输入“帮我写一封辞职信”系统需要① 先检查用户是否在职查 HR 系统② 若在职提取其入职日期、部门、直属领导查数据库③ 若已离职则返回提示④ 生成信件时必须引用公司《员工手册》第 3.2 条关于离职流程的原文RAG⑤ 最终输出前调用合规审查 API 检查敏感词Tool Call⑥ 整个过程需记录审计日志且任一环节失败需回滚并通知用户。提示场景 B 不是“更复杂的 Prompt”而是一个分布式事务系统。它涉及外部服务调用、状态保持、错误处理、可观测性——这些全是传统后端开发的常识却在 AI 项目初期被集体性忽视。LangChain 的核心价值正是把这套常识以声明式、组合式的方式重新带回 AI 开发流程。LangChain 的设计哲学本质上是在回答“当 LLM 只是一个能力组件类似 Redis 或 PostgreSQL而非应用主体时我们该如何构建围绕它的可靠系统” 它通过四个关键抽象完成了这场范式迁移Runnable统一的执行契约Runnable是 LangChain 的基石接口定义了invoke(input) - output、stream(input)、batch(inputs)三个方法。这意味着一个 PromptTemplate、一个 LLM、一个 RetrievalQA Chain、甚至你自定义的 Python 函数只要实现Runnable就能被同等对待、无缝组合。这种统一性直接解决了“不同模块调用方式五花八门”的混乱。例如你不再需要为每个工具写tool.run()、api_client.query()、db_session.execute()而是全部用chain.invoke({query: xxx})。这背后是工程领域经典的“依赖倒置原则”——高层模块业务逻辑不依赖低层模块具体实现二者都依赖抽象Runnable。Tool可编排的能力单元在 LangChain 之前“调用外部 API”常被写死在 Prompt 里如“请调用 weather_api 获取北京天气”导致模型幻觉、无法调试、无法监控。LangChain 的Tool抽象强制将能力与推理分离Tool是一个有明确定义名称、描述、参数 Schema、可独立测试、可被 Agent 动态选择的函数。当你定义WeatherTool时你同时定义了它的能力边界只能查天气、输入约束必须有 city 参数、失败模式网络超时抛异常。Agent 在运行时只负责决策“该不该用”和“传什么参数”而Tool自己负责“怎么执行”和“出错怎么办”。这种解耦让复杂工作流的可靠性陡增。Retriever语义检索的标准化管道RAG 不是简单地“把文档喂给向量库”。真实场景中你需要文档加载PDF/网页/数据库、清洗去页眉页脚/OCR 噪声、切块按语义而非固定长度、嵌入选择模型/批处理、存储向量库选型/索引优化、检索相似度阈值/重排序/Rerank、后处理去重/截断。LangChain 的Retriever接口将这一长串操作封装成一个黑盒get_relevant_documents(query)。你可以在不改上层业务代码的前提下把ChromaRetriever替换为MilvusRetriever或插入ParentDocumentRetriever父子文档检索提升长文档效果。这种可替换性是应对知识库规模从 100 页到 100 万页演进的关键。StateGraph有状态工作流的原生表达传统 RAG 是无状态的“查询-响应”而 Agentic RAG 必须支持循环、分支、状态持久化。LangGraphLangChain 的工作流引擎的StateGraph直接将工作流建模为有向图节点是Runnable边是条件判断如grade_node good ? answer : rewrite。它天然支持① 状态在节点间传递state[docs]② 循环重试rewrite节点返回route③ 并行执行all_of([node_a, node_b])④ 中断恢复状态可序列化存 DB。这比用 while 循环 字典手动管理状态安全性和可维护性高出两个数量级。2.2 为什么是 LangChain而不是其他框架市场上并非没有竞品LlamaIndex 专注 RAG 优化Haystack 偏向企业搜索Semantic Kernel 强推微软生态。LangChain 成为事实标准源于它对“工程适配性”的极致妥协——它不追求理论最优而追求“在 80% 场景下让工程师能用最少的认知成本写出可维护的代码”。与 LlamaIndex 对比LlamaIndex 的QueryEngine在纯文档问答场景下检索精度和速度常优于 LangChain。但它缺乏对 Tool Calling、多步骤 Agent、复杂状态流的原生支持。当你的需求从“查文档”扩展到“查文档调 API写数据库”LlamaIndex 需要大量胶水代码而 LangChain 的AgentExecutor和StateGraph已内置此能力。我曾帮一家金融客户做技术选型LlamaIndex 单轮问答快 15%但当加入风控规则校验需调用内部规则引擎后LangChain 方案的端到端稳定性99.98%远超 LlamaIndex 手动集成方案92.3%。与 Semantic Kernel 对比Semantic Kernel 的强项在于 C#/.NET 生态和 Azure 集成。但其 Python 版本长期滞后且设计理念偏向“LLM 为中心”Prompt 编排优先对异构系统集成如混合调用 MySQL、Kafka、SAP的支持较弱。LangChain 的Tool抽象则完全中立一个SQLDatabaseToolkit可以无缝接入任何 SQLAlchemy 支持的数据库无需关心底层是 PostgreSQL 还是 Oracle。与自研框架对比这是最痛的教训。我参与过三个“为性能自研调度器”的项目最终都回归 LangChain。原因很现实①人力成本维护一个支持重试、熔断、链路追踪、指标上报、配置热更新的工作流引擎需要 3-5 名资深后端工程师而 LangChain 的CallbackHandler体系已覆盖 90% 需求②生态成本LangChain 社区每天新增 20 个Tool如SlackTool、NotionTool、JiraTool自研意味着你要重复造所有轮子③升级成本当 LangChain 发布 v0.3 支持异步流式响应时你只需升级依赖而自研框架的每次架构升级都是一次停机发布。LangChain 的胜出不是技术碾压而是工程经济学的胜利它把 AI 应用开发中那些重复、高危、低附加值的基础设施工作打包成一个可共享、可验证、可演进的公共品。开发者得以聚焦在真正的业务差异点上——比如如何设计让销售合同解析准确率从 85% 提升到 99% 的 Prompt 工程而不是纠结于“怎么让 10 个 API 调用不互相阻塞”。3. 核心机制深度解析从 RAG 到 Agentic RAG 的进化路径3.1 传统 RAG 的“脆弱性”根源先看一个典型传统 RAG 流程的伪代码def rag_answer(query): # 步骤1检索 docs vectorstore.similarity_search(query, k4) # 步骤2拼接上下文 context \n\n.join([doc.page_content for doc in docs]) # 步骤3生成答案 prompt f根据以下信息回答问题{context}\n\n问题{query} return llm.invoke(prompt)这段代码在 Demo 环境中运行完美但在生产中会频繁暴雷。问题不在算法而在流程的刚性假设假设1检索必有结果当similarity_search返回空列表常见于 query 表达模糊、知识库覆盖不足context为空LLM 只能胡编乱造。LangChain 的Retriever本身不解决此问题但为后续增强留出接口——你可以轻松包装一个FallbackRetriever当主检索失败时自动触发关键词检索或默认文档。假设2检索结果质量恒定向量相似度得分如 0.72不等于语义相关性。一个得分高的 chunk 可能只是高频词匹配实际无关。传统 RAG 缺乏对检索结果的“质检”环节。LangChain 的ContextualCompressionRetriever可插入LlmRanker用 LLM 对检索结果重排序而 Agentic RAG 更进一步用grade_node节点做二元判断good/bad驱动工作流走向不同分支。假设3Query 无需预处理用户问“那个去年签的合同”query中的指代“那个”、“去年”无法被向量检索理解。传统方案靠 Prompt 工程让 LLM “猜”准确率极低。LangChain 的QueryRewritingRetriever可集成ConversationalRetrievalChain用 LLM 将模糊 query 重写为明确检索式如“2023年签署的销售合同”。假设4单次检索足够复杂问题常需多跳推理“华东区销售额” → “先查华东区有哪些城市” → “再查各城市销售额” → “最后求和”。传统 RAG 的单次检索无法支持此链式逻辑。LangGraph 的StateGraph天然支持循环retrieve节点输出城市列表后map节点可并行调用SalesTool查询各市数据再由reduce节点聚合。实操心得我在某政务知识库项目中发现 63% 的失败请求源于 Query 模糊。上线QueryRewriter后首检命中率从 41% 提升至 79%但仍有 12% 的请求因重写后仍无结果而失败。这时Agentic RAG 的router_node就发挥作用——它判断“此 query 无法重写且无检索结果”直接路由到direct_answer_node返回“未找到相关信息请提供更具体的合同编号或日期范围”。这比返回一个幻觉答案用户体验好十倍。3.2 Agentic RAG 的决策循环LangGraph 如何让 RAG “活”起来Agentic RAG 的核心是引入“智能体”作为流程控制器。LangGraph 的StateGraph让这个控制器的实现变得极其清晰。我们以标题中的热搜词agentic rag为蓝本拆解其工作流节点设计节点1router_node—— 智能分流器def router_node(state: dict) - dict: query state[query] # 规则1短 query10字大概率是闲聊直答 if len(query) 10: return {route: direct} # 规则2含明确实体合同号、ID、日期且知识库有索引直检 if re.search(r(合同|ID|编号)\s*[A-Z0-9\-], query): return {route: exact_lookup} # 默认走检索流 return {route: retrieve}这个节点的价值在于将模糊的“是否需要检索”决策转化为可测试、可配置的规则集。你可以随时添加新规则如“含‘最新版’字样的 query优先查版本号字段”而无需修改主流程。节点2retrieve_node—— 检索执行器def retrieve_node(state: dict) - dict: query state[query] # 主检索向量相似度 docs retriever.get_relevant_documents(query) # 辅助检索关键词匹配弥补向量检索对专有名词的不足 keyword_docs keyword_retriever.get_relevant_documents(query) # 合并去重 all_docs list(set(docs keyword_docs)) return {docs: all_docs[:4]} # 限制返回数防爆内存注意这里retriever和keyword_retriever是两个独立的Retriever实例体现了 LangChain 的组合能力。StateGraph的状态state是共享的但每个节点的实现完全解耦。节点3grade_node—— 质量守门员def grade_node(state: dict) - dict: docs state[docs] query state[query] # 方法1用 LLM 判断高精度有延迟 # grade_prompt f问题{query}\n文档{docs[0].page_content[:200]}\n该文档是否直接回答问题请只回答是或否 # result llm.invoke(grade_prompt).strip() # return {grade: good if result 是 else bad} # 方法2用规则低延迟需维护 if not docs: return {grade: bad} # 检查文档是否包含 query 中的核心名词 query_nouns extract_nouns(query) # 自定义函数 for doc in docs[:2]: if any(noun in doc.page_content for noun in query_nouns): return {grade: good} return {grade: bad}grade_node是 Agentic RAG 的“智能”体现。它不依赖单一指标而是融合规则与模型判断。实测中混合策略规则初筛 LLM 终审在准确率92%和 P99 延迟320ms间取得最佳平衡。节点4rewrite_node—— Query 优化师def rewrite_node(state: dict) - dict: query state[query] # 构建重写 Prompt强调“生成可用于向量检索的精确查询” rewrite_prompt f你是一个专业的检索优化师。请将用户问题改写为1-2个简洁、无指代、含具体实体的检索查询用于向量数据库搜索。 原始问题{query} 要求 - 移除这个、那个、最近等模糊指代 - 补充必要的时间、地点、对象等限定词 - 输出纯文本不要解释 示例 输入上季度华东区销售额 输出2024年第二季度华东地区销售总额报表 new_query llm.invoke(rewrite_prompt).strip() return {query: new_query}关键点rewrite_node的输出直接覆盖state[query]为下一轮retrieve_node提供新输入。这种状态驱动的循环是传统 RAG 无法实现的。节点5answer_node—— 终极生成器def answer_node(state: dict) - dict: query state[query] docs state.get(docs, []) # 构建上下文不仅拼接内容还标注来源提升可信度 context_parts [] for i, doc in enumerate(docs): source getattr(doc, metadata, {}).get(source, unknown) context_parts.append(f[来源 {i1}: {source}]\n{doc.page_content}) context \n\n.join(context_parts) # 最终 Prompt明确指令 final_prompt f你是一个严谨的AI助手。请严格基于以下提供的上下文信息回答问题禁止编造、推测或使用外部知识。 如果上下文信息不足以回答问题请明确说根据现有资料无法确定。 问题{query} 上下文 {context} answer llm.invoke(final_prompt).strip() return {answer: answer, sources: [d.metadata for d in docs]}answer_node的精妙在于它接收的是经过grade_node筛选、可能由rewrite_node优化过的docs因此生成质量显著提升。同时sources字段为审计和溯源提供依据。3.3 LangChain 与 LangGraph 的协同为什么不是“LangChain or LangGraph”常有人问“LangChain 和 LangGraph 有什么区别” 这是个误导性问题。LangGraph 不是 LangChain 的替代品而是其工作流能力的强化子集。类比关系如下LangChain≈Python 标准库requests,json,os提供基础构件LLM,Retriever,Tool,PromptTemplate。LangGraph≈Python 的asyncio或concurrent.futures提供在基础构件之上构建复杂执行模型异步、并发、有状态循环的能力。它们的关系是分层演进而非互斥竞争。一个典型的 Agentic RAG 系统代码结构是# 1. LangChain 层定义原子能力 llm ChatOpenAI(modelgpt-4-turbo) retriever MilvusRetriever(vectorstorevectorstore) weather_tool WeatherTool() # 继承 langchain_core.tools.BaseTool # 2. LangGraph 层编排原子能力 workflow StateGraph(AgentState) # AgentState 是自定义状态类 workflow.add_node(agent, agent_node) # agent_node 内部调用 llm workflow.add_node(retrieve, lambda state: {docs: retriever.invoke(state[query])}) workflow.add_node(weather, lambda state: {weather: weather_tool.invoke(state[location])}) # ... 添加边和条件LangGraph 的StateGraph本身就是一个Runnable可以被 LangChain 的Chain调用。这种设计保证了你可以在一个项目中对简单流程用SequentialChain对复杂流程用StateGraph所有组件共享同一套类型系统BaseMessage,Document,Tool和回调体系CallbackHandler。注意LangGraph 的StateGraph要求你显式定义状态类AgentState这看似增加复杂度实则是巨大优势。它强制你思考“哪些数据需要跨节点传递”避免了传统全局变量式的隐式状态极大提升了可测试性。我曾重构一个 2000 行的旧 Agent 代码将其状态抽象为AgentState后单元测试覆盖率从 32% 提升至 89%。4. 实操全流程从零搭建一个生产级 Agentic RAG 系统4.1 环境准备与依赖管理生产环境的首要原则是确定性。LangChain 生态更新频繁v0.1.x 和 v0.2.x 的 API 差异巨大。我的经验是永远锁定小版本拒绝^或~符号。# 创建隔离环境 python -m venv agentic_rag_env source agentic_rag_env/bin/activate # Linux/Mac # agentic_rag_env\Scripts\activate # Windows # 安装核心依赖经生产验证的稳定组合 pip install \ langchain0.2.12 \ langchain-core0.2.21 \ langchain-community0.2.12 \ langgraph0.2.45 \ pymilvus2.4.12 \ openai1.41.1 \ python-dotenv1.0.1 # 验证安装 python -c from langchain_core.runnables import Runnable; print(LangChain OK) python -c from langgraph.graph import StateGraph; print(LangGraph OK)关键点说明langchain-community包含所有第三方集成Milvus、PostgreSQL、Notion 等必须显式安装。pymilvus2.4.12是 Milvus 2.4.x 的稳定客户端与langchain-community的MilvusVectorStore兼容。openai1.41.1避免 v1.50 的 breaking change如ChatCompletion返回类型变更。提示在requirements.txt中务必添加注释说明每个包的用途例如# langchain-core: 提供 Runnable、CallbackHandler 等核心抽象 # langchain-community: 提供 Milvus、Chroma、SQLDatabase 等向量库和数据源集成 # langgraph: 提供 StateGraph、ConditionalEdge 等工作流编排能力4.2 数据准备从原始文档到可检索 Chunk真实业务数据绝非理想化的 Markdown。我们以某 SaaS 公司的客户支持知识库混合 PDF、网页、内部 Confluence 页面为例from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader, ConfluenceLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.documents import Document import re # 步骤1多源加载 loaders [ PyPDFLoader(docs/product_manual_v3.pdf), WebBaseLoader(https://help.example.com/guide/), ConfluenceLoader( urlhttps://confluence.example.com, usernameapi_user, api_keyxxx ) ] all_docs [] for loader in loaders: try: docs loader.load() # 清洗移除页眉页脚、多余空行、HTML 标签 cleaned_docs [] for doc in docs: content re.sub(rPage \d of \d, , doc.page_content) # 移除页码 content re.sub(r[^], , content) # 移除 HTML 标签 content re.sub(r\n\s*\n, \n\n, content) # 合并多余空行 cleaned_docs.append(Document(page_contentcontent, metadatadoc.metadata)) all_docs.extend(cleaned_docs) except Exception as e: print(f加载 {loader.__class__.__name__} 失败: {e}) # 步骤2智能切块关键 # 不同文档类型chunk_size 和 overlap 需差异化 splitter RecursiveCharacterTextSplitter( chunk_size600, # 技术文档适合 400-800 chunk_overlap100, separators[ # 按语义分割优先级从高到低 \n\n, # 段落 \n, # 换行 . , # 句号空格 , # 空格 # 最后按字符切兜底 ] ) chunks splitter.split_documents(all_docs) print(f原始文档数: {len(all_docs)}, 切块后: {len(chunks)}) # 步骤3增强元数据提升检索精度 for i, chunk in enumerate(chunks): # 添加 chunk ID 和顺序 chunk.metadata[chunk_id] fchunk_{i} chunk.metadata[chunk_order] i # 添加来源文档的标题若存在 if title in chunk.metadata: chunk.metadata[source_title] chunk.metadata[title]实操心得切块策略是 RAG 效果的天花板。我测试过 12 种组合结论是学术论文chunk_size1000, overlap200保留完整公式和图表说明。API 文档chunk_size300, overlap50确保每个 endpoint 描述独立。会议纪要chunk_size400, overlap150保留发言者上下文。法律合同必须用SemanticChunkerLangChain 新特性按条款分割chunk_size设为 0不限制长度因为“违约责任”条款不能被切断。4.3 向量入库Milvus 配置与性能调优Milvus 是开源向量库中对 LangChain 集成最成熟的选项。生产部署需关注三点索引、一致性、可观测性。from pymilvus import MilvusClient, DataType from langchain_community.vectorstores import Milvus from langchain_openai import OpenAIEmbeddings # 步骤1创建 Milvus Client本地轻量版适合中小规模 milvus_client MilvusClient(uri./milvus_demo.db) # 文件模式无需 Docker # 步骤2创建 Collection表关键参数 collection_name agentic_rag_kb milvus_client.create_collection( collection_namecollection_name, dimension1536, # OpenAI text-embedding-3-small 的维度 metric_typeCOSINE, # 余弦相似度最常用 consistency_levelStrong, # 强一致性确保读写实时可见 auto_idTrue # 自动生成主键 ) # 步骤3配置 HNSW 索引平衡精度与速度 milvus_client.create_index( collection_namecollection_name, field_namevector, # 向量字段名 index_params{ index_type: HNSW, metric_type: COSINE, params: {M: 32, efConstruction: 128} # M 越大精度越高efConstruction 越大建索引越慢但查询越准 } ) # 步骤4批量嵌入与入库生产必备 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 分批处理防内存溢出 batch_size 100 for i in range(0, len(chunks), batch_size): batch chunks[i:ibatch_size] texts [c.page_content for c in batch] vectors embeddings.embed_documents(texts) # 批量嵌入比单次快 5x # 构建 Milvus 数据格式 data [ {vector: vec, text: chunk.page_content, metadata: chunk.metadata} for vec, chunk in zip(vectors, batch) ] milvus_client.insert(collection_namecollection_name, datadata) print(f已入库 {min(ibatch_size, len(chunks))}/{len(chunks)} 个 chunk) # 步骤5构建 LangChain Retriever vectorstore Milvus( embedding_functionembeddings, collection_namecollection_name, connection_args{uri: ./milvus_demo.db}, ) retriever vectorstore.as_retriever( search_kwargs{ k: 4, # 返回 top-k 结果 param: {ef: 64} # HNSW 查询参数ef 越大召回率越高但越慢 } )注意事项consistency_levelStrong是生产必需。默认Bounded级别可能导致刚写入的文档查不到引发用户困惑。ef参数调优在search_kwargs中设置ef值建议为k*10如k4则ef40。我在线上环境将ef从 32 提升到 64召回率Recall4从 78% 提升至 91%P95 延迟仅增加 80ms。避免auto_idFalse手动管理 ID 易出错LangChain 的Milvus集成依赖auto_idTrue。4.4 工作流编排LangGraph StateGraph 实战现在我们将前面准备好的retriever、llm以及一个模拟的SalesTool组装成完整的 Agentic RAG 工作流。from langgraph.graph import StateGraph, END from typing import TypedDict, List, Dict, Any, Optional from langchain_core.messages import HumanMessage, AIMessage from langchain_core.tools import tool # 步骤1定义状态Type Safety 是 LangGraph 的核心优势 class AgentState(TypedDict): query: str docs: List[Document] grade: str answer: str sources: List[Dict[str, Any]] sales_data: Optional[Dict[str, float]] # 新增销售数据字段 # 步骤2定义 Tool模拟调用 BI 系统 tool def sales_tool(query: str) - Dict[str, float]: 查询销售数据的工具。生产中应对接真实 BI API。 # 模拟解析 query 中的城市和季度 import re city_match re.search(r(上海|北京|广州|深圳), query) quarter_match re.search(r(Q1|Q2|Q3|Q4|第一季度|第二季度), query) city city_match.group(1) if city_match else 全国 quarter quarter_match.group(1) if quarter_match else Q2 # 模拟返回数据 mock_data { 全国_Q2: 1250000.0, 上海_Q2: 320000.0, 北京_Q2: 280000.0, 广州_Q2: 210000.0, 深圳_Q2: 190000.0 } key f{city}_{quarter} return {sales: mock_data.get(key, 0.0)} # 步骤3定义所有节点 def router_node(state: AgentState) - Dict[str, str]: query state[query] # 简单规则含“销售额”、“销售”、“收入”等词且