LangChain组合原语实战：Runnable如何构建可复用LLM工作流-北京尧图网络

LangChain 学到一个阶段很多人会卡在一道坎单独调模型已经很顺手但要把“识别意图—抽取要素—组合提示—生成回复—兜底重试”串成一条稳定链路时代码立刻乱成一锅粥。我最近重构内部客服工单处理的脚本时体会特别深真正难的不是某个模型调得好不好而是组合原语——把一次 LLM 调用和前后一堆小任务拼成一条完整流水线的最底层连接工具。这篇作为系列第二篇重点拆组合原语。适合已经会用 LangChain 的提示模板和模型对象、想在链路设计上更进一步的人也适合被一堆 chain 文档绕晕想知道 Runnable、LCEL、RunnableParallel、分支路由到底怎么配合的人。读完你能拿到一套清晰的设计套路外加我实际踩过的坑。1. 组合原语先搞明白到底在拆什么1.1 从脚本式调用到流水线式装配大多数人的第一版 LangChain 代码长得像脚本用户输入进来先用 Python 写 if-else 判断再手动拼提示词然后调用一次模型拿到结果后自己写正则清洗再用另一个 if-else 决定是不是要再调一次。这种写法的问题不在函数写得不好而在每个环节都是“命令式”地粘在代码里谁先谁后、怎么分支、怎么复用全都写在业务循环里。改动一个分支要全局考虑想给某一步做并行、加缓存、接日志都得侵入原有流程。组合原语解决的是这个层面的问题它让你先把一个个能力抽成标准单元再用固定的连接方式把它们装起来。单元之间不关心对方内部怎么实现只约定输入输出连接方式也不负责理解业务只负责数据的流转方向。这种思维切换很像做菜脚本式是“洗菜、切菜、炒菜”全部写进同一个菜谱流程组合式是先准备好几个独立的小工序再用“串行流水线”“并行同时处理”“按条件进不同灶台”这些标准机制把它们接起来。工序可以单测接法可以替换整体链路还能整条复用。1.2 Runnable所有组合的本质接口在 LangChain 里几乎所有可以被组合的东西都实现了同一个基类接口Runnable。提示模板是 Runnable语言模型对象是 Runnable输出解析器是 Runnable甚至连你自己写的一个普通函数只要包进 RunnableLambda也能变成 Runnable。为什么这一点重要因为组合原语关心的是“接口是否统一”而不是“你是模型还是函数”。只要接口统一我就可以把一个自定义函数和官方模型对象串在同一条链子里而不需要为每一种类型单独写一套拼接逻辑。Runnable 定义了几种主流执行方式invoke输入一个值得到一个输出batch输入一批值批量处理stream输入一个值输出流式结果ainvoke / abatch / astream对应的异步版本。组合后的整条链路也自动继承同样的接口。这是组合原语最漂亮的地方一条由 prompt、llm、parser 拼起来的链对外依然是一个 Runnable。你可以把整条链再作为别的链的一个分支继续往下组合这种递归式组合正是这套设计的核心价值。1.3 四类基础原语的职责划分我习惯把组合原语分成四类串行、并行、分支、数据装配。理解这四类LangChain 绝大多数链路都能拆解清楚。类型常见实现职责典型场景串行管道符、RunnableSequence上一个单元的输出作为下一个单元的输入并行RunnableParallel多个分支同时执行并合并成一个结果字典同时抽取关键词、判断情感、识别实体分支RunnableBranch按条件选择下一步执行哪个单元按用户意图走不同处理链路数据装配RunnablePassthrough、assign保留或追加字段控制数据形态把中间结果原样透传或临时计算新字段串行和并行描述的是时间关系分支描述的是路径选择数据装配描述的是数据在节点之间怎么变形。真实链路里它们会嵌套使用。比如先做一个并行抽取再对某个抽取结果做分支判断最后串接到模型调用。2. 核心原语逐个拆解2.1 管道连接串行执行的精髓LangChain 表达式语言LCEL里两个 Runnable 用竖线连接a | b表达式的含义是执行b(a(input))。这不是 Python 原生的某种魔法而是 LangChain 给 Runnable 协议实现的运算符重载。你写prompt | llm | parser实际就是构造了一个 RunnableSequence。from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt ChatPromptTemplate.from_template(请把下面这句话的语气改得更温和{raw}) llm get_llm() # 这里替换成你接好的模型实例 parser StrOutputParser() chain prompt | llm | parser result chain.invoke({raw: 立刻马上把报告交给我}) print(result)执行过程是字典{raw: ...}传给 promptprompt 把消息模板格式化成消息列表消息列表传给模型模型返回一个包含文本内容的消息对象消息对象传给 parserparser 取出纯文本字符串返回。每一步的输入输出都是上一步产生的直接结果没有中间变量。这里有一个容易忽略的关键点管道组合能够成立前提是每个单元的输入类型能和上游输出类型对上。类型对不上时LangChain 通常会在运行时报错而且经常要到数据传到那一刻才会暴露所以我建议在搭建链路后先用最小样本跑一遍。你可以用chain.input_schema.schema()查看链路期望的输入结构用chain.output_schema.schema()查看最终输出结构。2.2 并行原语RunnableParallel 与结果合并串行管线之外另一类刚需是“同一份输入同时喂给多个分支”。比如分析一条用户反馈时既要判断意图又要提取关键词还要看情绪倾向三个任务互不依赖串行做会浪费时间代码也绕。RunnableParallel 就是干这个的from langchain_core.runnables import RunnableLambda, RunnableParallel def classify_intent(info): text info[text] if 退货 in text: return 退款 if 崩溃 in text or 报错 in text: return 技术 return 咨询 def extract_keywords(info): words info[text].replace(, ).split() return [w for w in words if len(w) 1][:5] def get_sentiment(info): negative_words [崩溃, 糟糕, 慢, 气] return negative if any(w in info[text] for w in negative_words) else positive parallel_step RunnableParallel( intentclassify_intent, keywordsextract_keywords, sentimentget_sentiment, ) result parallel_step.invoke({text: 应用崩溃了恢复起来很慢很糟糕。}) print(result)执行逻辑是同一个输入字典{text: ...}会分别传入三个分支每个分支的输出挂到对应的字段名下最终合并成一个{intent: ..., keywords: ..., sentiment: ...}的新字典。三个分支内部是并行执行的这也是 RunnableParallel 能比串行省时间的原因。这里要提醒一句RunnableParallel 的并行是“并发执行”不是绝对意义上的“多线程/多进程并行”。它更适合处理 I/O 型等待任务比如多个分支都在调用外部模型接口。如果分支里都是 CPU 密集型的纯计算收益没那么大。还有一个隐蔽坑分支之间不要共享可变状态否则并发修改可能造成不可预期的结果。每个分支最好只依赖输入参数不要偷偷读写同一个全局变量。2.3 分支原语RunnableBranch 与条件路由业务里几乎没有一条链路能从头走到尾。用户说“我要退货”和说“我想问问价格”后面的处理逻辑完全不同。这类条件分支如果写在一大段 Python 里可读性会迅速劣化如果组合原语里也有一个标准组件能代表“按条件选路由”整条链就能保持图表化结构。RunnableBranch 就是这样一个组件。它的构造方式是传入一组条件函数对应 Runnable二元组最后再给一个兜底 Runnablefrom langchain_core.runnables import RunnableBranch, RunnableLambda def is_refund(info): return info.get(intent) 退款 def is_technical(info): return info.get(intent) 技术 refund_chain RunnableLambda(lambda info: {reply: 退款流程已记录请提供订单号。}) tech_chain RunnableLambda(lambda info: {reply: 技术问题已反馈请尝试清除缓存。}) normal_chain RunnableLambda(lambda info: {reply: 已收到您的咨询稍后回复。}) branch RunnableBranch( (is_refund, refund_chain), (is_technical, tech_chain), normal_chain # 兜底 ) result branch.invoke({intent: 退款}) print(result)RunnableBranch 会按顺序逐个调用条件函数返回 True 的分支被选中并执行如果所有条件都不满足则执行兜底 Runnable。注意条件的顺序是有意义的先匹配的先生效这要求你在写条件时把更具体的规则放在前面。比如“退款”和“想退但还在咨询”同时触发时应该先判断更明确的“退款”。分支本身也是 Runnable这意味着分支组件内部还可以嵌套分支或管道。举例来说tech_chain 可以先通过管道调用一个模型生成回复再在后半段做一个是否需要转人工的分支。把分支当成普通节点看待整条链就不会变成不可维护的蜘蛛网。2.4 数据透传与字段装配RunnablePassthrough 与 assign最后一类基础原语不改变数据流向只负责数据形态。RunnablePassthrough 最简单的用法是“原样透传”也就是不修改输入直接把输入作为输出返回。单独看没什么用但在链路的中间环节非常常见你希望保留原始用户消息同时往下游追加一些计算字段而不是每次都要手动拼一个新的完整字典。配合.assign()方法用法会更灵活from langchain_core.runnables import RunnablePassthrough def make_upper(info): return info[text].upper() chain RunnablePassthrough.assign(upper_textmake_upper) result chain.invoke({text: 组合原语有意思}) print(result) # 输出形如 {text: 组合原语有意思, upper_text: 组合原语有意思}这句代码的含义是先保留输入字典里的原有字段再把 make_upper 的计算结果挂到 upper_text 字段下。原字段不丢新字段追加。这种“保留旧数据、叠加新数据”的写法特别适合组合链路。比如你一边把原始消息透传到下游一边用并行分支计算情感最后在生成回复时既能看到原文又能看到情感结果。RunnablePassthrough 和 RunnableLambda 很容易被混淆。区别在于RunnableLambda 的返回值会成为链路上唯一的下一步输入而 RunnablePassthrough.assign 会保留并合并原输入字典。前者是“替换”后者是“扩充”。用错了最典型的现象就是下一步突然丢失了之前字段。3. 实操构建一套可复用的工单响应链3.1 场景设计与方案映射光讲概念容易飘我拿一个真实场景来串。假设要给某个内部客服系统搭一条自动响应链路用户提交一段工单文本系统需要从文本里同时判断问题类型、提取关键词、识别情绪倾向最后根据不同的类型生成不同风格的回复。这类需求如果写脚本能写出来但后续要加“合并多个工单”“批量跑历史数据”“对回复做质量抽检”都会很痛苦。用组合原语重写后每个环节是独立的随时可以替换或测试。先把需求拆成五个小能力标准化输入去掉首尾空格统一字段名并行抽取识别问题类型、提取关键词、判断情绪条件分支按照问题类型选择回复模板回复生成根据抽取结果拼出最终回复输出保留最终阶段仍保留原始文本方便复盘。3.2 完整代码实现下面是完整可运行的示例我用 RunnableLambda 模拟了规则判断核心目的是展示原语怎么组合模型对象可以原样替换成你自己接好的实例。from langchain_core.runnables import ( RunnableLambda, RunnableParallel, RunnablePassthrough, RunnableBranch, ) def normalize(raw): return { raw: raw.strip(), source: 工单系统, } def classify_issue(info): text info[raw] if 崩溃 in text or 报错 in text or 卡顿 in text: return 技术问题 if 退货 in text or 退款 in text: return 退款问题 return 普通咨询 def extract_keywords(info): text info[raw] words text.replace(, ).replace(。, ).split() return [w for w in words if len(w) 1][:5] def get_sentiment(info): negative_words [崩溃, 卡顿, 慢, 糟糕, 气] positive_words [好用, 感谢, 顺利] if any(w in info[raw] for w in negative_words): return negative if any(w in info[raw] for w in positive_words): return positive return neutral def refund_reply(info): return { reply: f退款诉求已记录。关键词{info[keywords]}情绪{info[sentiment]}。后续会由专员跟进订单号。, } def technical_reply(info): return { reply: f技术问题已提交。关键词{info[keywords]}情绪{info[sentiment]}。建议先清除缓存并重启应用。, } def normal_reply(info): return { reply: f您的咨询已收到。关键词{info[keywords]}情绪{info[sentiment]}。感谢您的反馈。, } normalize_step RunnableLambda(normalize) extract_step RunnableParallel( issueclassify_issue, keywordsextract_keywords, sentimentget_sentiment, ) branch_step RunnableBranch( (lambda info: info[issue] 技术问题, RunnableLambda(technical_reply)), (lambda info: info[issue] 退款问题, RunnableLambda(refund_reply)), RunnableLambda(normal_reply), ) chain ( normalize_step | extract_step | branch_step ) res chain.invoke({raw: 应用崩溃了速度特别慢很糟糕。 }) print(res)这里我刻意没有用 RunnablePassthrough 保留原始字段是想先展示最简单的组装。执行流程非常清晰normalize_step 把用户文本标准化extract_step 同时算出 issue、keywords、sentiment 三个字段branch_step 再根据 issue 决定用哪个回复模板。如果你希望最终输出里保留原始文本、关键词和情绪方便后续人工审核可以把链路改成chain ( normalize_step | extract_step | RunnablePassthrough.assign(replybranch_step) )这样 Runnable 的输出就会同时包含 issue、keywords、sentiment、reply以及前面保留的 raw 和 source 字段。因为 normalize_step 已经保证输入里有 rawextract_step 又把抽取字段挂了上去branch_step 虽然输入是整个字典但它只读自己需要的 issue、keywords、sentiment最后通过 assign 把生成的 reply 追加回去和分支输入并存在结果字典中。注意这里我用的是RunnablePassthrough.assign(replybranch_step)而不是再写一个合并字典的 Lambda这是为了保持“不丢上下文”的语义。后续无论加多少字段只要某个字段依赖已有字段都放在 assign 里链路结构就不会破坏。3.3 LCEL 与显式 Runnable 的选择上述代码用的是管道符这是 LCEL 风格优点是可读性好、嵌套自然。另一种写法是显式实例化 RunnableSequencefrom langchain_core.runnables import RunnableSequence chain RunnableSequence(firstnormalize_step, lastextract_step)显式写法适合你需要在配置系统里动态指定链路、或者序列结构需要在运行时修改的场景。实际上 LCEL 底层也会编译成 RunnableSequence两种风格性能没有差别选择标准主要是团队可读性和业务需要。4. 更复杂的组合技巧参数注入、兜底重试、可观测性4.1 用 bind 预绑定执行参数在组合链路里经常需要给某个步骤单独传一些参数但这个参数不是上一个节点的输出。比如模型调用时想固定 temperature0.2、max_tokens500这些参数如果写明在业务代码里链路就不干净了。Runnable 提供了.bind()方法可以提前把参数绑定到一个准备好的调用上llm get_llm() summarize_chain llm.bind(temperature0.2, max_tokens500) | StrOutputParser()这里的 bind 可以理解成准备了一个“预设参数的模板调用”它仍然是一个 Runnable可以被正常组合。绑定后再和其他步骤相接链路不关心参数是怎么注入的。如果你在做 A/B 测试也可以从配置里动态决定绑定哪些参数然后把不同版本分别放进回退列表。还要区分 bind 和 with_configbind 是改调用的运行时参数对模型来说就是采样参数这类with_config 是改 Runnable 的执行配置包括运行名称、标签、元数据。两个侧重点完全不同混了会很难排查。4.2 给链路加兜底现实里没有一次就成功的链路。模型接口会超时、解析器会遇到意料之外的格式、中间步骤会抛异常。组合原语里对应的做法是把兜底逻辑也当成可组合的一部分用.with_fallbacks()给某个节点挂备用节点或挂一组按顺序尝试的备用链from langchain_core.runnables import RunnableLambda def unstable_step(data): raise RuntimeError(模拟不稳定服务错误) def fallback_step(data): return {status: ok, message: 兜底成功} unstable RunnableLambda(unstable_step) fallback RunnableLambda(fallback_step) robust_chain unstable.with_fallbacks([fallback]) result robust_chain.invoke({query: 测试}) print(result)这段代码不会抛异常unstable_step 失败后LangChain 会自动用 fallback 顶替。业务上常见的做法是主模型超时先用缓存结果兜底缓存没有再用简化提示词兜底简化提示词也失败最后返回固定话术。每个兜底节点自身也可以是一条复杂链路只要它是 Runnable。这里要控制一个度兜底节点太多会把链路变成“错误吞噬器”。如果所有错误都被兜底静默吞掉你根本不知道系统已经降级了多少次。建议在兜底节点里通过日志记录“当前使用了哪个 fallback”后面审计时会发现非常有价值。4.3 执行配置与追踪可观测性是组合链的生命线链路越长越需要知道每一步发生了什么。Runnable 的.with_config()可以给链条或某个节点绑定运行名称、标签、元数据chain ( normalize_step | extract_step | branch_step ).with_config( run_name工单自动回复, tags[客服, demo], metadata{version: v1.2, owner: ticket-bot}, )配合监听器可以拿到每一步的输入输出事件。一个朴素的监听实现是这样的def on_start(record): # record 里包含当前 run 的名称和输入 pass def on_end(record): # record 里包含当前 run 的输出 pass chain.with_listeners(on_starton_start, on_endon_end)监听器是诊断组合链问题的关键工具特别是当链里有并行分支时你会在日志里看到多条交织的事件但每条事件都带有自己的 run_name能够准确还原执行路径。我见过的绝大多数“链路输出不对”问题都是靠从第一步开始追踪输入、逐步比较每个中间节点输出解决的。与其猜不如把链路拆开逐节点验证。4.4 循环组合让模型自我修正组合原语不仅能表达直线或树形结构也能表达循环。LangChain 里你可以用 RunnableLambda 加递归函数来实现“多轮检查修正”from langchain_core.runnables import RunnableLambda def revise_step(info): if info.get(accepted): return {**info, final: info[draft]} attempt info.get(attempt, 0) 1 if attempt 2: return {**info, final: info[draft], accepted: True} # 假设这里会调用一次模型对 draft 做修正 revised_draft f修正版{attempt}: info[draft] (已润色) return {**info, draft: revised_draft, attempt: attempt} revise_step RunnableLambda(revise_step)这种循环没法用普通的管道表达需要把“是否继续循环”的判断写在函数内部。对这类逻辑我的建议是给循环设定最大次数并且每次循环保留 attempt 字段避免模型陷入无穷重试。真实团队里常犯的错是把循环写成 while True一旦模型没有收敛生产环境就出现超时和资源白白烧掉。5. 常见问题与排查实录5.1 字段匹配错误为什么这么难查组合链最常见的故障并不是“代码崩溃”而是“链路静默跑完但输出结果和预期差很远”。原因八成出在数据装配某个节点返回的是字符串但下游期望字典或者下游读的键名和上游写的键名只差一个下划线。排查工具我按顺序用三个。第一步看chain.input_schema和chain.output_schema先确认链路对外暴露的输入输出结构。第二步用最小输入样本逐步调用单个节点把每个节点的输出打印出来确认数据形态。第三步用监听器记录事件比较完全一样的输入在两次运行中的中间差异。绝大多数情况下问题在第二步就已经暴露。chain normalize_step | extract_step | branch_step inputs {raw: 测试一下 } print(标准化结果, normalize_step.invoke(inputs)) print(抽取结果, extract_step.invoke(normalize_step.invoke(inputs))) print(最终结果, chain.invoke(inputs))这种逐节点打印的方式虽然土但非常有效。尤其当你怀疑是某个分支的问题时直接调用分支组件单独执行能立刻确认是条件写错还是分支内部逻辑错。5.2 RunnableLambda 的副作用别乱搞RunnableLambda 只是把你写的普通函数包装成 Runnable它不会自动帮你处理线程安全。如果函数内部修改了全局列表、读写共享计数器在 RunnableParallel 并发执行时就会出现数据竞争。结果可能时好时坏且非常难复现。我遇到过一个案例两个并行分支都往全局日志列表里追加记录跑单条数据没问题跑批量数据时偶发丢日志。后来改成每个分支返回独立的日志字段最后统一合并问题彻底消失。这个教训就是分支函数应该是纯函数尽量只依赖传入参数所有中间产物都放进返回值里。5.3 并行并不是万能的RunnableParallel 虽然名字里有 parallel但它默认实现的并发度受 Python 线程和 GIL全局解释器锁的限制。如果分支里是 CPU 密集型的文本解析并发提升有限如果分支是调用外部接口路径上的等待时间才是主要耗时并发就有效。另外RunnableParallel.invoke 是并发执行了多个分支但如果你用异步循环同时调用整条大链就要注意内部是否还有嵌套的线程池。层数多了以后线程池资源会互相争抢性能反而下降。生产环境下优先用异步接口ainvoke控制高并发不要在单线程里大量开 RunnableParallel 去硬扛。5.4 链路不是越细越好组合原语鼓励拆分但拆得太碎也会有成本。每一步 Runnable 都带来一次函数调用、一次数据转换如果链路有十几层可读性和调试成本都会上去。我自己的习惯是把“有独立业务含义、可能需要单独替换或单独测试”的环节作为节点把纯粹为了多包一层的代码合并进去。比如“把字符串去掉空格”这种操作合并进标准化节点就行没必要单独造一个 Runnable。而“调用模型生成摘要”这种可能被其他地方复用的能力才值得抽成一个独立节点。判断标准很简单如果这段逻辑在另一条链里也能直接用就单独成节点如果只是当前链路上的临时格式处理就并进相邻节点。5.5 一个速查表方便对照我把实际工作中见过的高频问题整理成一张表不一定覆盖所有场景但能帮你少走很多弯路。现象常见原因排查方向链路静默跑通输出却少了某个字段某节点返回了新的字典上游字段没保留检查是否用了 RunnableLambda 而不是 assign并行分支结果时好时坏分支间共享了可变状态去掉全局变量让分支变成纯函数报错提示“Expected Runnable”把普通函数或非 Runnable 对象直接接进了管道用 RunnableLambda 包装后再接入同样的输入中间结果不稳定某个步骤内部有随机性或者并发竞争先单节点调用定位再考虑加缓存链路能跑但整体很慢串行步骤过多或并行分支里混入 CPU 密集计算把可并行的 I/O 抽出来异步化调用fallback 没有生效异常发生在条件函数里而不是 Runnable 内部条件函数不要依赖可能抛错的逻辑提前做防御先画数据流再写代码是避免这些问题的根本办法。我写组合链的习惯是动手前先在白板上画数据流图不画组件关系只画每一跳的数据形态。哪一步输出字符串哪一步输出字典哪一步会新增字段全部标出来。画完之后代码基本就是照着图填空。很多新手纠结要不要用 RunnablePassthrough、要不要用 RunnableParallel一追问数据流图就立刻清楚了如果下一跳需要原始输入就用透传或 assign如果多个计算互不依赖就用并行。这个习惯帮我避开了大量返工也让我在评审别人链路时能快速定位问题。组合原语的魅力从来不是省几行代码而是让复杂业务里的数据流动变得可见、可测、可复用。下次再搭一条流程时先抽离几个小能力再用原语把它们串起来你会有完全不一样的体验。

LangChain组合原语实战：Runnable如何构建可复用LLM工作流

相关推荐

Spring多数据源路由失效的四大根源与契约验证

POJ基础题刷题指南：AC代码答案包与踩坑总结

短剧SaaS系统源码：多租户隔离与微信分账实战解析

更多资讯

Nemea 高速网络流量分析：从抓包瓶颈到流式异常检测架构

临床试验估计目标：如何精准定义你真正想问的科学问题

低代码平台四大运行时引擎深度解析

Linux终端入门：从命令行交互到C程序编译运行

设计模式实战：SOLID原则、UML类图与五个高频C++模式解析

Python爬虫实战：requests+parsel抓取招聘数据并写入CSV

今日更新

MATLAB缩进快捷键与批量格式化实战指南

用Poco实现Qt风格信号与槽的跨平台C++方案

ESXi裸机虚拟化安装指南：从硬件兼容到虚拟机部署

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？