1. 为什么我放弃了“拼积木”式的现成Chain接触LangChain的人大概率都经历过这样一个过程刚开始觉得这东西太方便了LLMChain、RetrievalQA、ConversationChain文档里示例一抄一个能跑的回答机器人就出来了。但真到了业务场景里尤其是那种需要跨多个数据源、走多步判断、还要根据中间结果动态决定下一步怎么做的时候你会发现现成组件就像乐高基础块——拼个简单房子绰绰有余但要搭一个带升降电梯、太阳能板、还得会自己开关门的复杂建筑光靠基础块是拼不出来的。我印象很深的一次是做一个内部知识库的智能问答系统单轮问答用RetrievalQA很顺利但业务方提了一个需求用户问一个项目整体进度系统需要先查数据库中某个项目的当前阶段再去读取这个阶段的验收文档把文档里的关键结论和数据库的更新日期做对比如果发现文档更新滞后还要自动提醒。这个流程涉及一次检索增强、两次工具调用、一个条件判断还有一条提醒规则。用现成的Chain硬拼代码很快会变成一大坨if-else包着LLMChain的意大利面维护成本直接爆表。后来我彻底转向了自定义Chain和自定义Agent现在回头看这是LangChain真正开始发挥威力的地方。本文就把我在实际项目里沉淀下来的设计思路、代码骨架和一些坑按实战的顺序完整拆一遍希望能帮到那些已经过了Hello World阶段、正在为复杂任务发愁的同学。1.1 现成工具链的边界在哪先说清楚一个问题LangChain的现成Chain本质上是“预设好的数据流模板”。RetrievalQA就是“问题→向量检索→拼上下文→丢给模型→输出”ConversationChain就是“历史新问题→模型→更新记忆”。它帮你把80%的常规调用封装好了你只需要填参数。但复杂任务往往意味着三件事这三件事恰恰是现成模板的死穴。第一数据流不是线性的。复杂任务经常需要“先判断条件再决定走哪条分支”或者是“循环执行直到满足某个条件”。现成Chain大多是单向管道A到B到C中间想插一个动态分支要么靠RunnableBranch硬写要么就只能拆成多个Chain在外层用Python代码串。拆到最后框架反而成了累赘。第二状态需要跨步骤共享。举个具体的例子假设一个客服工单处理Agent第一步要读工单内容识别问题类型第二步要根据问题类型查不同文档或者调用不同接口第三步要把查到的信息整理成回复草稿第四步还要判断是否需要转人工。整个流程里“问题类型”这个状态是在第一步产生、在第二三四步都要用到的。现成Chain做这件事非常别扭你得用RunnablePassthrough小心翼翼地把变量传下去中间一旦忘了传模型就只能凭幻觉瞎猜。第三模型需要动态决定用什么工具。这也是最核心的一点。复杂任务里“下一步调用什么”往往不是开发者在写代码时就能确定的而是要根据模型的推理结果动态决定。比如用户说“把上周的销售数据和本周的做个对比然后查一下有哪些大客户流失了”这个任务最合理的执行路径是什么是先查两周数据还是先查客户流失名单如果让开发者硬编码场景一变代码就得改但如果让模型动态决策系统就有了自适应不同表达方式的能力。这正是Agent的用武之地。所以我的判断标准很简单如果任务流程是固定的、输入输出结构是可以预定义的用现成Chain没问题但只要出现“分支”“循环”“动态选工具”“多步状态共享”这四个词里的任何一个就该考虑自己撸袖子了。1.2 需要自定义Chain的三种典型信号什么样的信号出现就意味着你该停止拼积木开始写自定义代码了我总结了三个高频信号基本覆盖了绝大多数复杂任务场景。第一个信号输出必须遵循特定结构且这个结构要作为下一步的输入。比如你要让模型输出一个严格的JSON带字段校验和类型转换同时这个JSON要直接作为下一个模块的入参。现成Chain输出的是原始字符串你得在中间手动加解析和校验逻辑。每加一层代码的可复用性就差一层。这时不如把这个“解析校验转换”整体写成一个自定义Chain对外暴露一个干净的接口。第二个信号流程中存在条件分支或者循环重试。举个例子一个合同审查系统先让模型审查合同文本找风险条款如果发现高风险条款要触发第二个专门处理高风险条款的Chain如果没发现就走常规摘要Chain。这种“根据中间结果决定下一步”的逻辑LangChain的RunnableBranch能写但写出来可读性极差而且一旦分支条件变多、每个分支还要调用不同工具那基本就是灾难现场。把整个审查流程封装成一个自定义Chain内部用普通Python逻辑控制分支外观上仍然是一个可被调用的单元清晰得多。第三个信号你需要融合外部工具且工具的输入输出需要模型来协调。比如一个数据分析Agent它需要调用数据库查询工具、Python代码执行工具、文件读写工具每个工具的输入参数都依赖模型在前一步的推理结果。这时候真正的主角是那个“决策循环”而不是某个单一的Chain。现成Agent跑通简单场景没问题但一旦工具数量超过三五个工具描述长度超过了模型窗口的一半默认的ReAct循环就开始频繁出错你必须自定义Agent的提示词、解析逻辑、甚至循环控制。看到这三个信号别犹豫直接进入自定义开发。接下来两章分别讲Chain和Agent的自定义方法这也是LangChain进阶之路上的两个核心抓手。2. 自定义Chain的核心机制从协议到实现在LangChain 0.2/0.3时代官方主推的Chain自定义方式是继承Runnable协议而不是老版本的Chain基类。这是一个很重要的演进Runnable协议的核心思想是标准化“一个可执行的单元应该长什么样”这个单元的结构有四个核心方法invoke(input) - output同步执行给一个输入拿一个输出。ainvoke(input) - output异步执行适用于IO密集场景。stream(input) - Iterator[output_chunk]流式输出适用于逐token返回的场景。batch(inputs: list) - list批量执行多个输入并行处理。你可以把Runnable想象成一个“标准插座”——不管内部逻辑多复杂对外的接口统一这样一来不同类型的Chain、不同来源的工具、甚至一个自定义的Python函数都可以用|管道操作符串在一起LangChain称之为LCELLangChain Expression Language。2.1 Runnable协议Everything is a pipeline理解LCEL的管道思想是自定义Chain的前提。一条LCEL表达式的核心就是左侧的输出自动变成右侧的输入。比如chain prompt | llm | StrOutputParser()这一行的意思就是给chain一个字典格式的输入prompt格式化出消息llm生成回复StrOutputParser把消息对象转成字符串。写自定义Chain本质上就是写一个能插在这个管道中间的“兼容单元”——既可以当管道的一部分被调用也可以自己内部再嵌套一个完整管道。我自己设计自定义Chain时的第一个原则是对外统一用Runnable接口对内随意编排。内部不管你有几个模型调用、几层工具请求、多少逻辑判断外部只暴露一个invoke方法。这样带来的好处是——整个复杂任务可以拆成多个独立的自定义Chain每个Chain都保持简单的输入输出契约再通过LCEL把它们组装起来。有一个细节新手容易忽略RunnableLambda可以让普通Python函数变成Runnable但它的输出类型不会被自动记录。如果你希望自定义逻辑能参与管道类型推断最好直接用Runnable的子类或者tool装饰器而不是到处滥用RunnableLambda套娃。2.2 手写一个能在复杂流程里复用的Chain来看一个具体的例子。假设你要做一个风险条款审查模块它要做三件事解析输入文本、调用模型提取风险条款并判断风险等级、根据等级决定走哪个后续分支。拆成自定义Chain的话我会这么写from langchain_core.runnables import Runnable, RunnableConfig from typing import Any, Dict, Optional import json class RiskAssessmentChain(Runnable): def __init__(self, llm, high_risk_chain, low_risk_chain): self.llm llm self.high_risk_chain high_risk_chain self.low_risk_chain low_risk_chain self._prompt self._build_prompt() def _build_prompt(self): # 实际项目中这里通常是PromptTemplate这里简化说明 pass def invoke(self, input: Dict[str, Any], config: Optional[RunnableConfig] None) - Dict[str, Any]: # 第一步解析并提取风险条款 raw_text input[text] extraction_result self._extract_risks(raw_text) # 第二步根据风险等级选择分支 if extraction_result[has_high_risk]: return self.high_risk_chain.invoke({ text: raw_text, risks: extraction_result[risks] }) else: return self.low_risk_chain.invoke({ text: raw_text, risks: extraction_result[risks] })这段代码揭示了自定义Chain的骨架构造时接收依赖项调用时根据输入执行内部逻辑最后返回结构化的输出。比起用RunnableBranch硬拼这种写法的可读性强得多而且你可以直接在invoke里打日志、做监控、加缓存这是框架自带写法做不到的。有一点特别值得注意Runnable协议的方法签名是固定的input和config这两个参数最好严格按规范来因为LangChain内部会对这些方法做动态调度比如ainvoke默认会把参数传给invoke的异步版本如果你强行改签名可能出现调用链断裂的诡异问题。别问我怎么知道的踩过。3. 自定义Agent的编排逻辑让模型学会调用你的工具如果说Chain解决的是“固定流程如何封装”那Agent解决的就是“不固定流程如何决策”。复杂任务的终点几乎都是Agent——因为它要应对的是用户千变万化的表达方式以及任务路径的多种可能性。3.1 Agent不是Chatbot是一个决策循环很多人对Agent有误解觉得就是“一个会调用工具的聊天机器人”。实际上Agent的核心是一个循环这个循环可以拆成四步思考Thought模型根据当前状态用户问题、已有观察结果、工具列表决定下一步该做什么。动作Action模型输出一个结构化的动作指令一般包含工具名和入参。观察Observation系统执行这个动作拿到工具返回的结果。循环Repeat把观察结果追加回上下文再次交给模型思考直到模型认为任务完成输出最终答案。这个循环结构是ReAct范式的核心也是LangChain Agent的底层机制。理解了循环你就理解了一个关键问题为什么Agent的效果高度依赖工具描述写得好不好因为工具的输入参数、功能边界、返回值格式全部要靠模型在“思考”阶段自行推断如果描述含糊模型决策的正确率就直线下降。所以在自定义Agent之前我会把全部精力放在一个事情上把工具定义成让模型一看就懂的东西。3.2 工具定义决定了Agent效果的上限我之前做过一个内部系统工具列表里有六个工具其中有一个工具query_weekly_report第一次写描述时我写的是“查询周报数据”。结果模型经常在用户问“八月的营收”时调query_weekly_report传参却是乱的后来我把描述改成了这个效果提升非常明显查询指定时间范围内的周报汇总数据。 仅适用于以下问题周/月度营收、客户数、订单数等统计类问题。 输入必须包含 date_from 和 date_to 两个参数格式为 YYYY-MM-DD。 如果用户没有给出明确的起始和结束日期不要调用本工具先向用户询问日期范围。这里的关键不是描述写得多详细而是“约束条件”写得多清晰什么情况能调用、什么情况不能、参数格式是什么、缺参数时怎么办。我把工具描述当成“写给一个聪明但没经验的实习生的操作手册”来写每次写完后我会问自己如果一个人从来没看过系统代码光看这个描述能不能准确调用定义工具本身也用LangChain的tool装饰器值得注意的一个细节是函数的docstring会被自动解析成工具的description参数的类型注解和注释会被解析成工具的JSON Schema。我通常会再手动指定args_schema用Pydantic模型精确控制参数结构from langchain_core.tools import tool from pydantic import BaseModel, Field class WeeklyReportInput(BaseModel): date_from: str Field(description起始日期格式 YYYY-MM-DD) date_to: str Field(description结束日期格式 YYYY-MM-DD) tool(query_weekly_report, args_schemaWeeklyReportInput) def query_weekly_report(date_from: str, date_to: str) - str: 查询指定日期范围内的周报汇总数据。 仅适用于统计类问题如营收、订单数、客户数等。 若日期参数缺失不要调用本工具。 # 实际查询逻辑 return ...很多人在这一步偷懒让模型自己从prompt里猜参数格式这是Agent效果差的头号原因。工具的JSON Schema写得不严格模型解析参数时就容易产生幻觉——比如把2024-08-01写成八月一日把product_id猜成product_name一旦入参错了后面全盘皆输。3.3 用LangGraph写可控的Agent循环虽然LangChain自带create_react_agent和AgentExecutor可以快速搭一个Agent但它们在处理真正复杂的任务时有两个致命缺陷一是循环次数和停止条件不可控模型容易陷入死循环二是中间状态难以观察出了问题你根本不知道是哪一步错了。我在实际项目中更倾向于用LangGraph来构建Agent循环因为LangGraph把“图”的概念引入了工作流节点和边都是显式定义的你可以在任意节点之间插入回调、条件判断、甚至人工干预。拿前面提到的项目周报系统举例它的Agent循环我定义为四个节点from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated class AgentState(TypedDict): user_query: str tool_results: dict intermediate_answers: list[str] final_answer: str def decide_next(state: AgentState) - str: # 根据当前状态决定进入工具调用节点还是输出节点 if state.get(need_more_tools): return call_tool return finalize graph StateGraph(AgentState) graph.add_node(initial, initial_node) graph.add_node(call_tool, call_tool_node) graph.add_node(finalize, finalize_node) graph.set_entry_point(initial) graph.add_conditional_edges( initial, decide_next, {call_tool: call_tool, finalize: finalize} ) graph.add_edge(call_tool, initial) graph.add_edge(finalize, END) app graph.compile()写到这里你会发现LangGraph和自定义Chain其实是绝配自定义Chain作为“节点执行体”解决单步内部逻辑LangGraph作为“调度器”控制整体流程走向。两者结合复杂任务的工程化程度会高一大截。4. 实战一个多数据源项目周报系统的完整落地理论说了这么多下面用一个我实际做过的需求来串一遍。背景是这样的某虚拟团队每周要给管理层发一份项目周报周报内容包括本周完成事项、关键指标变化、风险预警、下周计划四个板块。管理层的需求是用自然语言提问系统自动从数据库、项目文档、风险登记表三个数据源拉数据汇总成结构化周报。这个任务拆开来看至少涉及两个数据源查询、一次文档检索、两次模型调用先分析后汇总、一个风险判断分支。用现成Chain拼过一次失败后我直接按自研工作流来设计。4.1 任务拆解与模块边界划分动工之前先把任务拆成独立的模块。我画了一张很粗糙的功能清单不画图了口头描述模块A指标查询Chain。输入是“本周日期区间”输出是结构化的指标变化列表数据来自数据库。模块B项目事项检索Chain。输入是“本周日期区间”加用户问题输出是本周完成事项列表数据来自项目文档向量库。模块C风险分类Agent。输入是模块A和模块B的结果输出是风险判断和预警建议数据来自风险登记表。模块D周报聚合Chain。输入是ABC的结果输出是最终的四板块周报。边界划分的原则是“数据获取”和“逻辑判断”隔离、能独立测试的模块单独成Chain。这样任何一个模块出了问题都能单独替换和修复。4.2 模块A和B的实现用Runnable协议保持接口统一模块A的实现思路是这样的先用tool把数据库查询函数封装成工具再用Runnable管道串起来。注意这个模块虽然是“Chain”但它内部使用的其实是LangGraph的Agent节点——因为“从用户问题提取日期区间”这件事最好还是交给模型判断。我当时用了比较取巧的方式模块A不直接接受用户问题而接受一个已经标准化好的日期区间对象由上游Agent负责从用户问题里解析出date_from和date_to。这样模块A就变成一个纯函数式的Chain不需要模型参与可以做单元测试、也可以直接缓存。模块B则是一个典型的RAG流程区别在于我自定义了一个“检索后处理步骤”从向量库拿回的文档块不能直接全部丢给模型要先把文档块的元数据比如文档更新时间提取出来和模块A的数据库更新时间做对比这一步我用一个RunnableLambda解决。这部分让我学到的教训是模块之间的数据接口类型越严格越好。我都用Pydantic定义了输入输出模型宁可写的时候多两行也不要跑到线上才发现某个模块返回了None导致下游炸掉。4.3 模块C和D的编排让Agent自己决定风险判断模块C本来是整个系统里最“智能”的部分它要根据A和B的数据结合风险登记表的历史条目判断当前哪些项目存在风险。风险判断的规则本来藏在业务方的脑子里写死在代码里的话一旦规则变整个模块重写。所以这里我用了一个独立Agent把风险登记表查询工具、A和B的输出、以及一段“什么是风险、什么不是风险”的判断标准给到Agent让它自己决定要不要调用工具、调用几次、最终输出什么。模块D更简单一些是一个纯汇总Chain输入是A、B、C三块的结构化输出输出是Markdown格式的周报。这个模块写成了自定义Chain内部其实就是一个大号的PromptTemplate加LLM调用再加输出格式化。读到这里你应该发现我整个系统的结构其实是“一个Agent编排多个Chain”Agent负责上游调度和决策Chain负责下游具体的功能执行。这是我在LangChain进阶后最推荐的一种架构。5. 跑通之后我踩过的坑和最终建议这部分全是血泪。项目从一个Demo到真正能稳定跑起来我踩了下面几个最典型的坑每个都浪费了不少时间。5.1 第一个坑状态传递靠“隐式约定”导致神秘丢数据最初我写Agent时AgentState里定义了一些“可选项”字段比如review_result只有在某个分支节点才会写入。结果某次在某一分支里读这个字段时它变成了None原因是有个上游节点没有初始化这个字段就跑到了下一个节点。好在LangGraph的StateGraph允许节点之间自由读写state不强制类型这也意味着你少写一个字段就是少写一个字段没有任何报错直到下游实际用到时才爆发。解决方法是把所有state字段都定义为必填的TypedDict字段并对可能为空的字段显式初始化。你可能觉得这是小事但在Agent这种多节点、多分支的流程里这种“隐式约定”是最容易出事的地方。5.2 第二个坑工具返回的JSON被模型二次加工某位同事在工具里返回了一个JSON字符串里面是查询好的数据但他没在工具描述里说明“返回的是标准JSON只需要原样粘贴给用户”。结果模型拿到这个结果后在生成最终答案时自作主张把JSON重新格式化了一遍字段顺序变了、转义符变了下游解析直接失败。这个坑的本质是模型不会按照你“返回什么我就用什么”的预期工作它倾向于在生成文本时“美化”一切。对策有两个要么在工具描述里明确标注“返回数据为标准JSON禁止二次格式化必须原样输出”要么让工具返回的不是文本而是直接写入state由下游节点按结构化方式读取彻底绕开模型生成环节。我后来全部改成了后者干净利落。5.3 排查思路从“模型抽风”到“找元凶”复杂Agent跑出奇怪结果时第一反应别怪模型而是按这个顺序排查检查工具的输入参数。模型传入的date_from真是YYYY-MM-DD格式吗还是它自己编了一个“上个月”检查工具的输出。返回的数据结构符合下游期望吗有没有空列表套空列表检查Agent的循环历史。模型是经过几次思考才走到的这个结果中间有没有误调用别的工具最后才检查模型本身。如果前三步都正常那多半是提示词里缺少了对某种边界情况的约束。按这个顺序排查十次里有八次能在工具层找到问题——模型的“幻觉”其实很多是上层给的条件不清晰导致的。这也能解释为什么工具定义和模块边界在整个系统中如此重要。5.4 给后来者的三条建议如果你正打算上手自定义Chain和Agent我的建议有三条。第一先小后大。不要一上来就设计一个巨大的Agent图先写一个小Chain跑通再慢慢加上工具和分支。复杂的编排能力是在迭代中生长的不是一开始设计出来的。第二给Agent设计“紧急出口”。每个Agent循环我至少留一个手动干预节点或者一个“当循环超过某次数时强制终止并给出部分结果”的保护逻辑。生产环境里一个死循环Agent造成的资源浪费是肉眼可见的。第三模块的标准接口比代码本身更值钱。把时间和精力花在定义好“每个模块的输入输出长什么样”上远比在那里调prompt更有长期收益。因为模块之间一旦用清晰的Pydantic模型约束住后续加功能、换实现、跑测试都会非常舒服。我在实际项目中最终沉淀下来的这套“Agent编排Chain执行严格数据契约”的架构已经稳定跑了几个月期间经历了业务规则多次调整基本都是改某个Chain的内部逻辑就能完成不用动编排层。最后再多说一句LangChain这框架版本迭代快文中展示的API写法在你读到这篇文章时可能已经有所变化但核心的思路——把流程拆成标准单元、用协议统一调用、用图来控制走向——这一点是稳定适用的。把思路吃透了具体写法照着对应版本文档改就行这才是真正能沉淀下来的东西。
1. 为什么我放弃了“拼积木”式的现成Chain接触LangChain的人大概率都经历过这样一个过程刚开始觉得这东西太方便了LLMChain、RetrievalQA、ConversationChain文档里示例一抄一个能跑的回答机器人就出来了。但真到了业务场景里尤其是那种需要跨多个数据源、走多步判断、还要根据中间结果动态决定下一步怎么做的时候你会发现现成组件就像乐高基础块——拼个简单房子绰绰有余但要搭一个带升降电梯、太阳能板、还得会自己开关门的复杂建筑光靠基础块是拼不出来的。我印象很深的一次是做一个内部知识库的智能问答系统单轮问答用RetrievalQA很顺利但业务方提了一个需求用户问一个项目整体进度系统需要先查数据库中某个项目的当前阶段再去读取这个阶段的验收文档把文档里的关键结论和数据库的更新日期做对比如果发现文档更新滞后还要自动提醒。这个流程涉及一次检索增强、两次工具调用、一个条件判断还有一条提醒规则。用现成的Chain硬拼代码很快会变成一大坨if-else包着LLMChain的意大利面维护成本直接爆表。后来我彻底转向了自定义Chain和自定义Agent现在回头看这是LangChain真正开始发挥威力的地方。本文就把我在实际项目里沉淀下来的设计思路、代码骨架和一些坑按实战的顺序完整拆一遍希望能帮到那些已经过了Hello World阶段、正在为复杂任务发愁的同学。1.1 现成工具链的边界在哪先说清楚一个问题LangChain的现成Chain本质上是“预设好的数据流模板”。RetrievalQA就是“问题→向量检索→拼上下文→丢给模型→输出”ConversationChain就是“历史新问题→模型→更新记忆”。它帮你把80%的常规调用封装好了你只需要填参数。但复杂任务往往意味着三件事这三件事恰恰是现成模板的死穴。第一数据流不是线性的。复杂任务经常需要“先判断条件再决定走哪条分支”或者是“循环执行直到满足某个条件”。现成Chain大多是单向管道A到B到C中间想插一个动态分支要么靠RunnableBranch硬写要么就只能拆成多个Chain在外层用Python代码串。拆到最后框架反而成了累赘。第二状态需要跨步骤共享。举个具体的例子假设一个客服工单处理Agent第一步要读工单内容识别问题类型第二步要根据问题类型查不同文档或者调用不同接口第三步要把查到的信息整理成回复草稿第四步还要判断是否需要转人工。整个流程里“问题类型”这个状态是在第一步产生、在第二三四步都要用到的。现成Chain做这件事非常别扭你得用RunnablePassthrough小心翼翼地把变量传下去中间一旦忘了传模型就只能凭幻觉瞎猜。第三模型需要动态决定用什么工具。这也是最核心的一点。复杂任务里“下一步调用什么”往往不是开发者在写代码时就能确定的而是要根据模型的推理结果动态决定。比如用户说“把上周的销售数据和本周的做个对比然后查一下有哪些大客户流失了”这个任务最合理的执行路径是什么是先查两周数据还是先查客户流失名单如果让开发者硬编码场景一变代码就得改但如果让模型动态决策系统就有了自适应不同表达方式的能力。这正是Agent的用武之地。所以我的判断标准很简单如果任务流程是固定的、输入输出结构是可以预定义的用现成Chain没问题但只要出现“分支”“循环”“动态选工具”“多步状态共享”这四个词里的任何一个就该考虑自己撸袖子了。1.2 需要自定义Chain的三种典型信号什么样的信号出现就意味着你该停止拼积木开始写自定义代码了我总结了三个高频信号基本覆盖了绝大多数复杂任务场景。第一个信号输出必须遵循特定结构且这个结构要作为下一步的输入。比如你要让模型输出一个严格的JSON带字段校验和类型转换同时这个JSON要直接作为下一个模块的入参。现成Chain输出的是原始字符串你得在中间手动加解析和校验逻辑。每加一层代码的可复用性就差一层。这时不如把这个“解析校验转换”整体写成一个自定义Chain对外暴露一个干净的接口。第二个信号流程中存在条件分支或者循环重试。举个例子一个合同审查系统先让模型审查合同文本找风险条款如果发现高风险条款要触发第二个专门处理高风险条款的Chain如果没发现就走常规摘要Chain。这种“根据中间结果决定下一步”的逻辑LangChain的RunnableBranch能写但写出来可读性极差而且一旦分支条件变多、每个分支还要调用不同工具那基本就是灾难现场。把整个审查流程封装成一个自定义Chain内部用普通Python逻辑控制分支外观上仍然是一个可被调用的单元清晰得多。第三个信号你需要融合外部工具且工具的输入输出需要模型来协调。比如一个数据分析Agent它需要调用数据库查询工具、Python代码执行工具、文件读写工具每个工具的输入参数都依赖模型在前一步的推理结果。这时候真正的主角是那个“决策循环”而不是某个单一的Chain。现成Agent跑通简单场景没问题但一旦工具数量超过三五个工具描述长度超过了模型窗口的一半默认的ReAct循环就开始频繁出错你必须自定义Agent的提示词、解析逻辑、甚至循环控制。看到这三个信号别犹豫直接进入自定义开发。接下来两章分别讲Chain和Agent的自定义方法这也是LangChain进阶之路上的两个核心抓手。2. 自定义Chain的核心机制从协议到实现在LangChain 0.2/0.3时代官方主推的Chain自定义方式是继承Runnable协议而不是老版本的Chain基类。这是一个很重要的演进Runnable协议的核心思想是标准化“一个可执行的单元应该长什么样”这个单元的结构有四个核心方法invoke(input) - output同步执行给一个输入拿一个输出。ainvoke(input) - output异步执行适用于IO密集场景。stream(input) - Iterator[output_chunk]流式输出适用于逐token返回的场景。batch(inputs: list) - list批量执行多个输入并行处理。你可以把Runnable想象成一个“标准插座”——不管内部逻辑多复杂对外的接口统一这样一来不同类型的Chain、不同来源的工具、甚至一个自定义的Python函数都可以用|管道操作符串在一起LangChain称之为LCELLangChain Expression Language。2.1 Runnable协议Everything is a pipeline理解LCEL的管道思想是自定义Chain的前提。一条LCEL表达式的核心就是左侧的输出自动变成右侧的输入。比如chain prompt | llm | StrOutputParser()这一行的意思就是给chain一个字典格式的输入prompt格式化出消息llm生成回复StrOutputParser把消息对象转成字符串。写自定义Chain本质上就是写一个能插在这个管道中间的“兼容单元”——既可以当管道的一部分被调用也可以自己内部再嵌套一个完整管道。我自己设计自定义Chain时的第一个原则是对外统一用Runnable接口对内随意编排。内部不管你有几个模型调用、几层工具请求、多少逻辑判断外部只暴露一个invoke方法。这样带来的好处是——整个复杂任务可以拆成多个独立的自定义Chain每个Chain都保持简单的输入输出契约再通过LCEL把它们组装起来。有一个细节新手容易忽略RunnableLambda可以让普通Python函数变成Runnable但它的输出类型不会被自动记录。如果你希望自定义逻辑能参与管道类型推断最好直接用Runnable的子类或者tool装饰器而不是到处滥用RunnableLambda套娃。2.2 手写一个能在复杂流程里复用的Chain来看一个具体的例子。假设你要做一个风险条款审查模块它要做三件事解析输入文本、调用模型提取风险条款并判断风险等级、根据等级决定走哪个后续分支。拆成自定义Chain的话我会这么写from langchain_core.runnables import Runnable, RunnableConfig from typing import Any, Dict, Optional import json class RiskAssessmentChain(Runnable): def __init__(self, llm, high_risk_chain, low_risk_chain): self.llm llm self.high_risk_chain high_risk_chain self.low_risk_chain low_risk_chain self._prompt self._build_prompt() def _build_prompt(self): # 实际项目中这里通常是PromptTemplate这里简化说明 pass def invoke(self, input: Dict[str, Any], config: Optional[RunnableConfig] None) - Dict[str, Any]: # 第一步解析并提取风险条款 raw_text input[text] extraction_result self._extract_risks(raw_text) # 第二步根据风险等级选择分支 if extraction_result[has_high_risk]: return self.high_risk_chain.invoke({ text: raw_text, risks: extraction_result[risks] }) else: return self.low_risk_chain.invoke({ text: raw_text, risks: extraction_result[risks] })这段代码揭示了自定义Chain的骨架构造时接收依赖项调用时根据输入执行内部逻辑最后返回结构化的输出。比起用RunnableBranch硬拼这种写法的可读性强得多而且你可以直接在invoke里打日志、做监控、加缓存这是框架自带写法做不到的。有一点特别值得注意Runnable协议的方法签名是固定的input和config这两个参数最好严格按规范来因为LangChain内部会对这些方法做动态调度比如ainvoke默认会把参数传给invoke的异步版本如果你强行改签名可能出现调用链断裂的诡异问题。别问我怎么知道的踩过。3. 自定义Agent的编排逻辑让模型学会调用你的工具如果说Chain解决的是“固定流程如何封装”那Agent解决的就是“不固定流程如何决策”。复杂任务的终点几乎都是Agent——因为它要应对的是用户千变万化的表达方式以及任务路径的多种可能性。3.1 Agent不是Chatbot是一个决策循环很多人对Agent有误解觉得就是“一个会调用工具的聊天机器人”。实际上Agent的核心是一个循环这个循环可以拆成四步思考Thought模型根据当前状态用户问题、已有观察结果、工具列表决定下一步该做什么。动作Action模型输出一个结构化的动作指令一般包含工具名和入参。观察Observation系统执行这个动作拿到工具返回的结果。循环Repeat把观察结果追加回上下文再次交给模型思考直到模型认为任务完成输出最终答案。这个循环结构是ReAct范式的核心也是LangChain Agent的底层机制。理解了循环你就理解了一个关键问题为什么Agent的效果高度依赖工具描述写得好不好因为工具的输入参数、功能边界、返回值格式全部要靠模型在“思考”阶段自行推断如果描述含糊模型决策的正确率就直线下降。所以在自定义Agent之前我会把全部精力放在一个事情上把工具定义成让模型一看就懂的东西。3.2 工具定义决定了Agent效果的上限我之前做过一个内部系统工具列表里有六个工具其中有一个工具query_weekly_report第一次写描述时我写的是“查询周报数据”。结果模型经常在用户问“八月的营收”时调query_weekly_report传参却是乱的后来我把描述改成了这个效果提升非常明显查询指定时间范围内的周报汇总数据。 仅适用于以下问题周/月度营收、客户数、订单数等统计类问题。 输入必须包含 date_from 和 date_to 两个参数格式为 YYYY-MM-DD。 如果用户没有给出明确的起始和结束日期不要调用本工具先向用户询问日期范围。这里的关键不是描述写得多详细而是“约束条件”写得多清晰什么情况能调用、什么情况不能、参数格式是什么、缺参数时怎么办。我把工具描述当成“写给一个聪明但没经验的实习生的操作手册”来写每次写完后我会问自己如果一个人从来没看过系统代码光看这个描述能不能准确调用定义工具本身也用LangChain的tool装饰器值得注意的一个细节是函数的docstring会被自动解析成工具的description参数的类型注解和注释会被解析成工具的JSON Schema。我通常会再手动指定args_schema用Pydantic模型精确控制参数结构from langchain_core.tools import tool from pydantic import BaseModel, Field class WeeklyReportInput(BaseModel): date_from: str Field(description起始日期格式 YYYY-MM-DD) date_to: str Field(description结束日期格式 YYYY-MM-DD) tool(query_weekly_report, args_schemaWeeklyReportInput) def query_weekly_report(date_from: str, date_to: str) - str: 查询指定日期范围内的周报汇总数据。 仅适用于统计类问题如营收、订单数、客户数等。 若日期参数缺失不要调用本工具。 # 实际查询逻辑 return ...很多人在这一步偷懒让模型自己从prompt里猜参数格式这是Agent效果差的头号原因。工具的JSON Schema写得不严格模型解析参数时就容易产生幻觉——比如把2024-08-01写成八月一日把product_id猜成product_name一旦入参错了后面全盘皆输。3.3 用LangGraph写可控的Agent循环虽然LangChain自带create_react_agent和AgentExecutor可以快速搭一个Agent但它们在处理真正复杂的任务时有两个致命缺陷一是循环次数和停止条件不可控模型容易陷入死循环二是中间状态难以观察出了问题你根本不知道是哪一步错了。我在实际项目中更倾向于用LangGraph来构建Agent循环因为LangGraph把“图”的概念引入了工作流节点和边都是显式定义的你可以在任意节点之间插入回调、条件判断、甚至人工干预。拿前面提到的项目周报系统举例它的Agent循环我定义为四个节点from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated class AgentState(TypedDict): user_query: str tool_results: dict intermediate_answers: list[str] final_answer: str def decide_next(state: AgentState) - str: # 根据当前状态决定进入工具调用节点还是输出节点 if state.get(need_more_tools): return call_tool return finalize graph StateGraph(AgentState) graph.add_node(initial, initial_node) graph.add_node(call_tool, call_tool_node) graph.add_node(finalize, finalize_node) graph.set_entry_point(initial) graph.add_conditional_edges( initial, decide_next, {call_tool: call_tool, finalize: finalize} ) graph.add_edge(call_tool, initial) graph.add_edge(finalize, END) app graph.compile()写到这里你会发现LangGraph和自定义Chain其实是绝配自定义Chain作为“节点执行体”解决单步内部逻辑LangGraph作为“调度器”控制整体流程走向。两者结合复杂任务的工程化程度会高一大截。4. 实战一个多数据源项目周报系统的完整落地理论说了这么多下面用一个我实际做过的需求来串一遍。背景是这样的某虚拟团队每周要给管理层发一份项目周报周报内容包括本周完成事项、关键指标变化、风险预警、下周计划四个板块。管理层的需求是用自然语言提问系统自动从数据库、项目文档、风险登记表三个数据源拉数据汇总成结构化周报。这个任务拆开来看至少涉及两个数据源查询、一次文档检索、两次模型调用先分析后汇总、一个风险判断分支。用现成Chain拼过一次失败后我直接按自研工作流来设计。4.1 任务拆解与模块边界划分动工之前先把任务拆成独立的模块。我画了一张很粗糙的功能清单不画图了口头描述模块A指标查询Chain。输入是“本周日期区间”输出是结构化的指标变化列表数据来自数据库。模块B项目事项检索Chain。输入是“本周日期区间”加用户问题输出是本周完成事项列表数据来自项目文档向量库。模块C风险分类Agent。输入是模块A和模块B的结果输出是风险判断和预警建议数据来自风险登记表。模块D周报聚合Chain。输入是ABC的结果输出是最终的四板块周报。边界划分的原则是“数据获取”和“逻辑判断”隔离、能独立测试的模块单独成Chain。这样任何一个模块出了问题都能单独替换和修复。4.2 模块A和B的实现用Runnable协议保持接口统一模块A的实现思路是这样的先用tool把数据库查询函数封装成工具再用Runnable管道串起来。注意这个模块虽然是“Chain”但它内部使用的其实是LangGraph的Agent节点——因为“从用户问题提取日期区间”这件事最好还是交给模型判断。我当时用了比较取巧的方式模块A不直接接受用户问题而接受一个已经标准化好的日期区间对象由上游Agent负责从用户问题里解析出date_from和date_to。这样模块A就变成一个纯函数式的Chain不需要模型参与可以做单元测试、也可以直接缓存。模块B则是一个典型的RAG流程区别在于我自定义了一个“检索后处理步骤”从向量库拿回的文档块不能直接全部丢给模型要先把文档块的元数据比如文档更新时间提取出来和模块A的数据库更新时间做对比这一步我用一个RunnableLambda解决。这部分让我学到的教训是模块之间的数据接口类型越严格越好。我都用Pydantic定义了输入输出模型宁可写的时候多两行也不要跑到线上才发现某个模块返回了None导致下游炸掉。4.3 模块C和D的编排让Agent自己决定风险判断模块C本来是整个系统里最“智能”的部分它要根据A和B的数据结合风险登记表的历史条目判断当前哪些项目存在风险。风险判断的规则本来藏在业务方的脑子里写死在代码里的话一旦规则变整个模块重写。所以这里我用了一个独立Agent把风险登记表查询工具、A和B的输出、以及一段“什么是风险、什么不是风险”的判断标准给到Agent让它自己决定要不要调用工具、调用几次、最终输出什么。模块D更简单一些是一个纯汇总Chain输入是A、B、C三块的结构化输出输出是Markdown格式的周报。这个模块写成了自定义Chain内部其实就是一个大号的PromptTemplate加LLM调用再加输出格式化。读到这里你应该发现我整个系统的结构其实是“一个Agent编排多个Chain”Agent负责上游调度和决策Chain负责下游具体的功能执行。这是我在LangChain进阶后最推荐的一种架构。5. 跑通之后我踩过的坑和最终建议这部分全是血泪。项目从一个Demo到真正能稳定跑起来我踩了下面几个最典型的坑每个都浪费了不少时间。5.1 第一个坑状态传递靠“隐式约定”导致神秘丢数据最初我写Agent时AgentState里定义了一些“可选项”字段比如review_result只有在某个分支节点才会写入。结果某次在某一分支里读这个字段时它变成了None原因是有个上游节点没有初始化这个字段就跑到了下一个节点。好在LangGraph的StateGraph允许节点之间自由读写state不强制类型这也意味着你少写一个字段就是少写一个字段没有任何报错直到下游实际用到时才爆发。解决方法是把所有state字段都定义为必填的TypedDict字段并对可能为空的字段显式初始化。你可能觉得这是小事但在Agent这种多节点、多分支的流程里这种“隐式约定”是最容易出事的地方。5.2 第二个坑工具返回的JSON被模型二次加工某位同事在工具里返回了一个JSON字符串里面是查询好的数据但他没在工具描述里说明“返回的是标准JSON只需要原样粘贴给用户”。结果模型拿到这个结果后在生成最终答案时自作主张把JSON重新格式化了一遍字段顺序变了、转义符变了下游解析直接失败。这个坑的本质是模型不会按照你“返回什么我就用什么”的预期工作它倾向于在生成文本时“美化”一切。对策有两个要么在工具描述里明确标注“返回数据为标准JSON禁止二次格式化必须原样输出”要么让工具返回的不是文本而是直接写入state由下游节点按结构化方式读取彻底绕开模型生成环节。我后来全部改成了后者干净利落。5.3 排查思路从“模型抽风”到“找元凶”复杂Agent跑出奇怪结果时第一反应别怪模型而是按这个顺序排查检查工具的输入参数。模型传入的date_from真是YYYY-MM-DD格式吗还是它自己编了一个“上个月”检查工具的输出。返回的数据结构符合下游期望吗有没有空列表套空列表检查Agent的循环历史。模型是经过几次思考才走到的这个结果中间有没有误调用别的工具最后才检查模型本身。如果前三步都正常那多半是提示词里缺少了对某种边界情况的约束。按这个顺序排查十次里有八次能在工具层找到问题——模型的“幻觉”其实很多是上层给的条件不清晰导致的。这也能解释为什么工具定义和模块边界在整个系统中如此重要。5.4 给后来者的三条建议如果你正打算上手自定义Chain和Agent我的建议有三条。第一先小后大。不要一上来就设计一个巨大的Agent图先写一个小Chain跑通再慢慢加上工具和分支。复杂的编排能力是在迭代中生长的不是一开始设计出来的。第二给Agent设计“紧急出口”。每个Agent循环我至少留一个手动干预节点或者一个“当循环超过某次数时强制终止并给出部分结果”的保护逻辑。生产环境里一个死循环Agent造成的资源浪费是肉眼可见的。第三模块的标准接口比代码本身更值钱。把时间和精力花在定义好“每个模块的输入输出长什么样”上远比在那里调prompt更有长期收益。因为模块之间一旦用清晰的Pydantic模型约束住后续加功能、换实现、跑测试都会非常舒服。我在实际项目中最终沉淀下来的这套“Agent编排Chain执行严格数据契约”的架构已经稳定跑了几个月期间经历了业务规则多次调整基本都是改某个Chain的内部逻辑就能完成不用动编排层。最后再多说一句LangChain这框架版本迭代快文中展示的API写法在你读到这篇文章时可能已经有所变化但核心的思路——把流程拆成标准单元、用协议统一调用、用图来控制走向——这一点是稳定适用的。把思路吃透了具体写法照着对应版本文档改就行这才是真正能沉淀下来的东西。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
