★ FEATURED ARTICLE 接手一个基于大模型的业务功能时我最先看的往往不是 prompt 写得漂不漂亮而是调用这一层是不是被硬编码在了业务代码里。见过太多项目第一版 demo 只有三行代码——拼字符串、发请求、取返回跑得飞快等到第二个需求进来要加历史对话第三个需求进来要输出 JSON第四个需求进来要换模型、要统计 token、要做重试这时候那段三行代码就变成了一坨谁都不敢动的面条。Chain 这一层抽象存在的意义就是把这些迟早要加的东西提前收拢到一个可复用的结构里。这篇东西我会从 Chain 这一层的设计动机讲起把它内部那几个关键方法拆开看一遍再用一份完整的 LLMChain 示例把模板、模型、解析器三段式跑通最后聊聊实际编排时最容易翻车的几个地方。适合已经能调通模型 API、准备把 demo 往生产推的人看也适合正在纠结到底要不要上这套框架的人。1. 从裸调API的返工说起Chain这一层到底在兜什么底1.1 第一版能跑第三版就崩裸调代码的宿命我印象比较深的是一个商品文案生成的场景。第一版需求很简单给一个商品名和三个卖点返回一段两百字的文案。裸调代码大概是这样的把模板字符串写死在一个函数里占位符用 f-string 替换发一次请求取choices[0]的文本返回。这版代码二十行测试也过了上线之后没人管它。第二版需求是文案要能选风格正式一点、活泼一点、价格敏感一点。这时候 f-string 里的模板开始分叉函数里冒出三四个if。第三版需求是输出要带标题、正文、标签三个字段前端要解析。于是又得在里面塞 prompt 约束、塞 JSON 解析、塞解析失败的重试。第四版要加同一批商品批量生成控制并发不超过五路第五版要记录每次调用的 token 消耗做成本核算。到这一步那个函数已经三百多行改一个模板要小心翼翼地确认没影响到别的分支想复用到另一个业务线更是天方夜谭。问题的根源不在于写代码的人水平不行而在于把提示词组织模型调用结果解析流程编排这四件完全不同的事揉进了一个函数。Chain 的价值本质上就是把这四件事拆成四个可以独立替换的零件。1.2 拆开之后四件事各自可替换拆开的好处很直接。提示词模板变成独立对象之后改文案不用动代码逻辑甚至可以让非技术同事来维护模板文本模型调用被封装之后从一家服务商换到另一家、从大模型换到小模型只改一个构造参数解析器独立之后输出格式从纯文本改成结构化字段只需要换一个 parser前面的逻辑一行不动编排层独立之后加缓存、加重试、加并发、加追踪都是在链的外层套装饰器业务代码本身保持干净。这个思路在用 LangChain 这类框架时体现得特别明显。它提供的Chain基类约定了几件事输入是什么键、输出是什么键、核心执行逻辑写在哪。你只要遵守这个契约就能免费得到一批能力——同步调用、异步调用、批量调用、流式调用、回调追踪、序列化保存。这些能力你自己从头写一遍少说也得几百行而且容易写出微妙的并发 bug。一个判断标准如果你的业务代码里出现了if 模型 A或者if 输出格式 json这类分支说明抽象没做够。这些差异应该被收敛到构造参数里而不是扩散到执行路径上。1.3 什么时候其实不需要Chain反过来讲也不是所有场景都值得上这层。如果你只是偶尔跑一次翻译、一次摘要全项目就这一处调用那老老实实写十行代码反而是最优解引入一整套框架只会增加依赖体积和心智负担。我一般用这几个信号来判断要不要引入一是同一类调用在项目里出现了三次以上二是需求方已经开始提换个模型试试输出改成表格这类变更三是需要观测比如要统计每次调用的耗时和 token。三条里中两条就该考虑上 Chain 了。至于用框架自带的还是自己写一层薄封装取决于团队对这套 API 的熟悉程度框架的优势在于生态齐全劣势是版本变动比较频繁这点后面会细说。2. Chain的骨架长什么样两代接口的差别与迁移判断2.1 老一代Chain的三个约定老版本的Chain是一个抽象基类子类必须实现_call方法并且声明input_keys和output_keys。它对外暴露的入口是__call__内部流程大致是先做输入校验检查传入的字典里有没有缺 key然后处理 memory 相关的读写接着调用子类的_call拿到原始结果再包装成字典返回。这个设计里有一个细节值得注意__call__的返回值永远是字典而_call的返回值可以是字符串。包装这一步由prep_outputs完成——如果子类声明了output_key就用它作为键如果没有就用output_keys[0]。这个设计是为了让链能嵌进另一条链里因为链与链之间传递的必须是命名参数不能是裸字符串。还有一个历史包袱是run方法。它的作用是让你直接拿到字符串结果不用从字典里取值写起来更顺手比如chain.run(namexxx)。但它有个硬限制只允许链有且只有一个输出键。多输出的链调run会直接抛异常。这个限制坑过不少人尤其是从一个简单的单输出链改成多输出链之后原来满地的.run()全挂了。2.2 Runnable出现之后一个接口打通所有组件后来框架引入了Runnable协议把提示词模板、模型、解析器、检索器全部统一成同一套接口。核心就三个方法invoke单次调用、batch批量、stream流式加上对应的ainvoke、abatch、astream异步版本。所有组件实现同一套协议之后就能用管道符|串起来。from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一名资深电商文案输出的文案要具体、有画面感禁止使用‘品质之选’这类空话。), (human, 商品{product}\n卖点{selling_points}\n字数控制在{word_limit}字以内。), ]) chain prompt | llm | StrOutputParser() result chain.invoke({ product: 便携折叠水壶, selling_points: 硅胶材质、可压扁收纳、容量500ml, word_limit: 120, })这段代码里prompt | llm会返回一个RunnableSequence再| StrOutputParser()还是RunnableSequence。管道符的实现很朴素__or__方法判断右操作数返回一个新的序列对象把左右两边依次塞进steps列表里。真正执行时invoke遍历steps把上一步的输出当作下一步的输入。也就是说prompt收到的是字典吐出的是PromptValuellm收到PromptValue吐出AIMessageStrOutputParser收到AIMessage吐出字符串。每一步的输入输出类型都是明确的串错了在类型层面就能发现。2.3 两代接口的对照与选择维度老式 ChainLLMChain 那一套Runnable / LCEL组合方式构造函数里传chains[...]或继承Chain重写_call管道符 输入输出字典进、字典出靠input_keys、output_key约束任意类型进出类型由上下游组件决定流式支持基本拿不到中间 token只能等整段返回原生支持stream、astream逐 token 返回异步支持有_acall但很多子类没实现会退化成同步全链路原生异步ainvoke不会阻塞事件循环批量能力需要自己写循环或调applybatch内置线程池配max_concurrency控制并发生态维护逐步迁移新特性基本不再往这边加主线方向新组件默认实现这套接口表格里最后一行是选择的关键。如果你的项目是新建的直接用管道符这套如果是老项目在维护也不需要急着全量重写因为老的 Chain 对象通常也实现了Runnable接口可以和新组件混着用比如把一整个老的LLMChain塞进管道里当一个步骤这样能渐进式迁移。迁移时最常见的坑是.run()和字典解包的写法。老代码里大量出现chain.run(**kwargs)和result[text]换成新接口之后要改成chain.invoke(kwargs)和直接取返回值。我一般会在迁移期写一个薄适配函数把新接口的返回值包成老格式先让调用方不动等功能稳定了再逐步清理。3. 把LLMChain拆开看一次调用到底走了哪几步3.1 入口处的输入校验与memory读写LLMChain是老式 Chain 里最典型的一个实现它的构造参数就三个核心prompt、llm、output_parser外加一个可选的output_key。理解它的执行链路基本上就理解了所有老式链的套路。调用从__call__开始。第一步是prep_inputs做两件事一是把 memory 里存的历史变量合并进来比如对话场景下的history二是校验输入键是否齐全。校验这一步看似啰嗦实际上是排错的关键——如果你在模板里写了{product}但调用时传的是product_name这里会直接抛出缺键异常而不是让模型收到一个带着{product}字面量的畸形 prompt。后者才是最可怕的情况不报错但模型输出莫名其妙你得盯着日志看半天。memory 的处理还有一个容易忽略的细节它在调用前后各执行一次。调用前load_memory_variables把历史读出来塞进输入调用后save_context把这一轮的输入和输出写回历史。如果你的链被套在一个循环里而 memory 是共享的就会出现历史越滚越长、逐步把上下文窗口撑爆的情况。这时候要么换用带窗口截断的 memory要么把 memory 从链里拆出来显式控制在什么时机读写。3.2 _call里的四步核心逻辑进入_call之后整个流程可以概括成四步self.prompt.format_prompt(**inputs)—— 把输入变量填进模板得到一个PromptValue对象。这一步做的是字符串替换加上消息格式组装如果模板里有多轮消息会组装成消息列表。self.llm.generate_prompt([prompt_value], ...)—— 把PromptValue交给模型。注意这里用的是generate_prompt而不是invoke前者返回的是LLMResult里面除了生成文本还带着llm_output包含 token 用量、模型名、结束原因等元信息。这个元信息是做成本核算的唯一来源很多人只取文本把它扔了后面要算账的时候追悔莫及。self.output_parser.parse_result(result, ...)—— 把LLMResult交给解析器。如果没配解析器默认行为是从result.generations[0][0].text取第一段文本。配了解析器的话这里会做 JSON 解析、字段校验、类型转换等。prep_outputs(...)—— 把解析后的结果包装成字典。如果链声明了output_key就用它作键否则用output_keys[0]。一份简化后的结构大概是这样细节会随版本变化但主干就是这个形状class LLMChain(Chain): prompt: BasePromptTemplate llm: BaseLanguageModel output_parser: BaseOutputParser None output_key: str text property def input_keys(self): return self.prompt.input_variables property def output_keys(self): return [self.output_key] def _call(self, inputs, run_managerNone): # 第一步模板填充 prompt_value self.prompt.format_prompt(**inputs) # 第二步模型生成保留完整 LLMResult response self.llm.generate_prompt( [prompt_value], callbacksrun_manager.get_child() if run_manager else None, ) # 第三步解析 if self.output_parser is not None: return self.output_parser.parse_result( response, callbacksrun_manager.get_child() if run_manager else None )[0] # 没配解析器时的兜底取第一段文本 return response.generations[0][0].text3.3 读源码时值得盯住的三个点第一是input_keys的来源。LLMChain没有单独维护输入键列表而是直接读prompt.input_variables。这意味着模板里出现了什么变量链就要求传什么变量两者天然同步。这也是为什么模板里的花括号处理必须小心——你写的每一个{}框架都会当成变量来要值。第二是回调的传递。run_manager.get_child()这一句看起来不起眼但它是整条追踪链路能串起来的原因。每一次子调用都会从父级 run manager 派生一个子级最终形成一棵调用树耗时、token、中间输入输出都挂在这棵树上。你看到的那些漂亮的调用链路图源头就是这个父子关系。第三是错误处理的位置。_call内部基本不做异常捕获模型超时、网络错误、解析失败都会直接往上抛。真正做兜底的地方在外层也就是__call__这一级或者你套在链外面的重试装饰器。知道这一点很重要不要在_call里写try/except把异常吞掉那样重试机制就失效了因为外层根本感知不到失败。4. 从零搭一份能跑的LLMChain示例三段式与迁移写法4.1 依赖固定与最小骨架版本这块我的建议很明确一定要锁版本。这套框架迭代速度不慢小版本之间改动导入路径是常有的事尤其老式 Chain 相关的类从主包挪到单独的包、再挪到另一个子模块不同版本写法不一样。项目里用requirements.txt或者锁文件把版本钉死比事后排查 ImportError 省事得多。pip install langchain0.2.* langchain-core0.2.0 langchain-community0.2.0最小可用版本走老式LLMChain这条路from langchain.chains import LLMChain from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser prompt PromptTemplate( template( 你是一名资深客服质检员。请判断下面这通对话是否出现了承诺赔付。\n 对话内容\n{transcript}\n 只回答 是 或 否不要解释。 ), input_variables[transcript], ) llm ChatOpenAI( modelgpt-4o-mini, temperature0, max_tokens8, timeout30, ) chain LLMChain( promptprompt, llmllm, output_parserStrOutputParser(), output_keyverdict, ) result chain.invoke({transcript: 客户你们赔不赔客服这个我们会处理的您放心。}) print(result) # {verdict: 否}这里有几个选择值得解释。temperature0是因为质检任务是判断类不是创作类需要结果稳定、可复现温度调高只会让同一段对话有时判是、有时判否没法做回归测试。max_tokens8是给输出加硬约束任务只要两个字给它留 8 个 token 足够了多了反而给模型自由发挥的空间输出个根据对话内容答案是否这种解析又得额外处理。output_keyverdict是显式命名比默认的text可读性好后面拼多链的时候也不容易撞名。4.2 换成结构化输出从字符串到对象判断类任务还好办真正麻烦的是需要返回多个字段的场景。比如从工单文本里抽出问题分类、紧急程度、涉及产品、建议处理人四个字段。这时候纯文本返回就没法用了得让模型按固定结构吐内容。from typing import Literal from pydantic import BaseModel, Field from langchain_core.output_parsers import PydanticOutputParser class TicketInfo(BaseModel): category: Literal[账号, 支付, 物流, 产品故障, 其他] Field(description工单问题分类) urgency: Literal[低, 中, 高] Field(description紧急程度) product: str Field(description涉及的产品名称未提及则填 未知) owner: str Field(description建议处理团队如 客服一组、技术支持组) parser PydanticOutputParser(pydantic_objectTicketInfo) prompt PromptTemplate( template( 从下面的工单文本中抽取结构化信息。\n {format_instructions}\n 工单原文\n{raw_text}\n ), input_variables[raw_text], partial_variables{format_instructions: parser.get_format_instructions()}, ) chain LLMChain(promptprompt, llmllm, output_parserparser, output_keyticket) info chain.invoke({raw_text: 买的水壶用了一周盖子就裂了急着出差要用赶紧给我换个新的})[ticket] print(info.category, info.urgency) # 产品故障 高这里的关键技巧是把format_instructions作为partial_variables注入。partial_variables的意思是这个变量不用每次调用时传构造模板时就固定下来。解析器生成的格式说明是一大段 JSON Schema 描述每次调用都手动传一遍既啰嗦又容易漏用 partial 绑定是标准做法。这段格式说明里包含了字段名、类型、枚举值和示例模型照着填的准确率会明显高于你自己手写请返回 JSON。Literal类型的用处也值得说一下。它把可选值限定死了模型如果吐了产品问题这种不在枚举里的值Pydantic 校验会当场失败而不是让脏数据流进下游。失败比脏数据好这是我在数据管道里的一贯态度——脏数据会静默污染报表校验失败至少能在日志里看见。4.3 同一件事用LCEL写一遍上面那段用管道符重写逻辑是一样的但代码形状变了from langchain_core.runnables import RunnablePassthrough chain ( { raw_text: RunnablePassthrough(), format_instructions: lambda _: parser.get_format_instructions(), } | prompt | llm | parser ) info chain.invoke(买的水壶用了一周盖子就裂了急着出差要用赶紧给我换个新的)这段里出现了一个字典它的作用是并行组装输入。字典的每个键对应下游prompt期望的变量名值可以是任何 Runnable 或者函数。RunnablePassthrough()表示把原始输入原样传过去后面的 lambda 忽略输入直接返回格式说明。整个字典会被自动包装成一个RunnableParallel各分支同时执行虽然这里都是瞬时完成然后合并成一个字典交给prompt。两种写法怎么选我自己的习惯是新代码一律用管道符原因是三点。一是流式支持完整老式链在流式场景下基本拿不到中间 token用户盯着空白屏幕等十秒体验很差二是组合更自由想在中间插一个数据清洗、插一个条件分支、插一个并发调用都是加一个组件的事三是类型信息更清楚每一步的输入输出是什么类型从代码上能直接读出来。但老式LLMChain也不是一无是处。它的input_keys机制在做复杂输入校验时反而更省事而且很多存量代码和第三方集成还在用它。我的做法是业务主链路用管道符需要对接老组件的地方把老 Chain 当成管道里的一个步骤塞进去两者可以共存。5. 链编排里最容易翻车的几个点5.1 花括号和变量名模板解析的第一道坎这个坑几乎每个人都会踩一次。你的 prompt 里想让模型输出 JSON 示例于是写了{name: 张三, age: 30}结果一运行就报变量缺失说找不到name这个变量。原因是模板引擎把花括号当成变量占位符双引号里的内容被当成了变量名。解决办法有两个。一是在模板里把示例的花括号转义成双花括号{{name: 张三}}渲染后会变成单花括号。二是干脆不在模板里写 JSON 示例改用解析器的get_format_instructions()来生成它在内部已经处理好了转义。我更倾向第二种因为手写示例还得自己维护格式和字段名容易和 Pydantic 模型对不上。变量名不匹配是另一个高频问题。模板里写{product}调用时传product_name如果链做了输入校验会直接报错这是好事。但如果你用的是裸的format_prompt调用或者传了个**kwargs展开的字典缺的变量可能被静默替换成空字符串prompt 里就留了个洞模型看到的是商品\n卖点...输出自然驴唇不对马嘴。所以我坚持一件事模板的变量名和调用处的键名用一个常量或者数据类统一管理别靠人肉记忆对齐。5.2 output_key撞车导致的缺键异常多链串联的时候每条链的输出都会合并进同一个字典。如果两条链都用了默认的output_keytext后一条会覆盖前一条等到下游需要前一条的结果时就会抛出Missing some input keys。这个报错信息看不出是覆盖导致的排查起来很费时间。我的做法是给每条链的输出显式命名命名规则用业务域_字段名比如ticket_category、summary_text一眼能看出是谁产出的。另外在拼装之前先把整条链的输入输出键画成一张表确认每一环的产出都有人在用、每一环的消耗都有人提供中间没有断点。这张表看着土但能省下大量调试时间。链路环节输入键输出键备注文本清洗raw_textcleaned_text去掉多余空白与乱码分类判断cleaned_textticket_category五分类枚举约束摘要生成cleaned_textsummary_text限 80 字汇总组装ticket_category, summary_textfinal_payload输出给前端5.3 流式输出拿不到token很多人第一次做流式会发现前端一直不吐字等好几秒之后整段文字突然出现。原因通常是中间用了不支持流式的组件或者用了invoke而不是stream。老式的LLMChain.run()就是典型代表它内部走的是同步生成必须等模型完整返回。# 想流式就得这么写 async for chunk in chain.astream({raw_text: text}): print(chunk, end, flushTrue)还有一点要注意如果链的末尾是结构化解析器流式就没意义了。因为PydanticOutputParser必须拿到完整文本才能解析中间 chunk 是半截 JSON解析必然失败。要做流式输出到前端末端的解析器得换成StrOutputParser或者自己写一个增量解析逻辑。这属于设计上的取舍得在需求阶段就定下来要结构化就不要流式要流式就让模型直接吐文本。5.4 缓存和重试的隐性失效缓存这块有个很隐蔽的问题。框架层的缓存一般是按完整的 prompt 内容 模型参数做键的这意味着只要 prompt 里有一个变量变了缓存就失效。如果你的 prompt 里带了时间戳、带了随机 ID、带了用户昵称那缓存命中率会接近零加了等于没加。我见过一个项目抱怨缓存不生效翻出来一看模板里有一句当前时间{now}每次都不一样。重试也是类似。外层套的重试装饰器只有在异常抛出时才会触发如果模型返回了一个格式错误但请求成功的结果重试根本不会启动。结构化输出场景下这种成功但无效的情况很常见模型返回了{category: 产品问题}字段值不在枚举里解析器抛异常这时候你希望的是重新生成一次而不是直接失败。解决办法是套一层修复解析器它在解析失败时会带着错误信息再问模型一次让它自己改。from langchain.output_parsers import OutputFixingParser robust_parser OutputFixingParser.from_llm(parserparser, llmllm)这一层的代价是多一次调用、多一点延迟值不值得看业务。分类判断这种场景失败一次重试的成本远低于把脏数据放进去我觉得值。但如果是对延迟极度敏感的实时接口可能更适合直接失败、走降级逻辑。5.5 并发一开就超时批量处理的时候batch接口很顺手但max_concurrency不给限制默认会按输入条数开线程。一百条数据同时发出去服务商那边大概率开始限流返回 429 或者超时。我一般的做法是根据服务商的配额来定先从小值开始试比如 5观察成功率和平均延迟再往上调。results chain.batch( [{raw_text: t} for t in texts], config{max_concurrency: 5}, )还有一个容易忽略的点是异步场景下的连接复用。用ainvoke的时候如果每次调用都新建一个客户端连接池会迅速耗尽。正确做法是把模型对象创建一次全局复用它内部的 HTTP 客户端会自动管理连接池。6. 我在实际项目里坚持的几条Chain实践准则6.1 一条链只干一件事这个原则听起来像废话但真正做到的项目不多。我见过把抽取、分类、打分、生成回复塞进一条链的prompt 有八百多字输出是四段文本靠分隔符切。这种链的问题是任何一个环节要调整整条链都得重新测某一段输出格式飘了后面全崩还定位不到是哪一段的问题。拆成四条链之后每条的输入输出都很窄出错时一眼能看出是哪一环改 prompt 也只影响那一环。代价是调用次数变多延迟增加。如果延迟敏感可以用并行编排把没有依赖关系的链并发起来比如分类和抽取其实可以同时跑。这种拆法还有个额外好处每一条链都可以单独做评测集跑回归测试时能精确定位是哪条链退化了。6.2 把可变的都外置到配置模板文本、模型名、温度、最大 token、超时时间、重试次数这些东西全部放配置文件代码里只留读取逻辑。理由很实际调 prompt 是个高频操作如果每次都要改代码、走一遍发布流程迭代速度会被拖死。放到配置里之后改完热加载就能生效甚至可以做个简单的后台界面让运营同学自己调。有个细节要提醒配置变更必须留痕。我一般会记录每次调用用的是哪个版本的模板出问题时能快速定位是不是昨天改的那句话导致的。最土的做法是在 prompt 里塞一个隐藏的版本号标注好一点的做法是在调用配置里带上run_name和metadata框架的追踪系统会把这些信息一起记下来。6.3 观测这件事从第一天就要做结构化输出的失败率、平均延迟、token 消耗、缓存命中率这四个指标我在任何一条上线的链上都会看。它们的实现都不复杂失败率靠异常计数延迟靠计时器token 消耗从返回的llm_output里取缓存命中率看缓存层自己的统计。response llm.generate_prompt([prompt_value]) usage response.llm_output.get(token_usage, {}) # usage 里通常有 prompt_tokens、completion_tokens、total_tokens顺手提一句run_name。在调用时传config{run_name: 工单分类_v3}追踪系统里就能按这个名字筛选比在一堆无名调用里翻找高效太多。这东西加一行代码的事但排查线上问题时能救命。6.4 版本升级前先跑一遍评测集前面说过这套框架迭代快老式 Chain 相关的导入路径变过不止一次。我的做法是维护一个小评测集五十到一百条覆盖典型场景的样本带上期望输出。每次升级依赖版本先跑一遍看通过率和之前比有没有掉。这一步花的时间不多但能挡掉大部分升级事故。对于已经被标记为逐步迁移的老组件我的建议是别急着动。能跑就先跑着等哪天真的要改那块逻辑了顺手迁到新接口上比专门排一个迁移任务划算得多。技术债要还但不必一次还清挑影响面最小的时机还就行。最后分享一个我在调试期常用的小技巧用假模型替代真实调用。框架里有个FakeListLLM你预先给它一串固定返回它会按顺序吐出来完全不发网络请求。用它来测链的编排逻辑、测解析器的容错、测异常分支速度快到可以做到每次提交都跑一遍。等编排逻辑稳定了再换成真模型跑端到端能省下大量等待和调用费用。这个习惯我保持了挺久尤其是在改复杂编排的时候先让假模型把流程跑通比直接对着真模型瞎试效率高得多。
阅读完成 · 觉得有帮助？
