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 绝大多数链路都能拆解清楚。

串行和并行描述的是时间关系，分支描述的是路径选择，数据装配描述的是数据在节点之间怎么变形。真实链路里它们会嵌套使用。比如先做一个并行抽取，再对某个抽取结果做分支判断，最后串接到模型调用。

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": ...}传给 prompt，prompt 把消息模板格式化成消息列表；消息列表传给模型，模型返回一个包含文本内容的消息对象；消息对象传给 parser，parser 取出纯文本字符串返回。每一步的输入输出都是上一步产生的直接结果，没有中间变量。

这里有一个容易忽略的关键点：管道组合能够成立，前提是每个单元的输入类型能和上游输出类型对上。类型对不上时，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( intent=classify_intent, keywords=extract_keywords, sentiment=get_sentiment, ) result = parallel_step.invoke({"text": "应用崩溃了，恢复起来很慢，很糟糕。"}) print(result)

执行逻辑是：同一个输入字典{"text": ...}会分别传入三个分支，每个分支的输出挂到对应的字段名下，最终合并成一个{"intent": ..., "keywords": ..., "sentiment": ...}的新字典。三个分支内部是并行执行的，这也是 RunnableParallel 能比串行省时间的原因。

这里要提醒一句：RunnableParallel 的并行是“并发执行”，不是绝对意义上的“多线程/多进程并行”。它更适合处理 I/O 型等待任务，比如多个分支都在调用外部模型接口。如果分支里都是 CPU 密集型的纯计算，收益没那么大。还有一个隐蔽坑：分支之间不要共享可变状态，否则并发修改可能造成不可预期的结果。每个分支最好只依赖输入参数，不要偷偷读写同一个全局变量。

2.3 分支原语：RunnableBranch 与条件路由

业务里几乎没有一条链路能从头走到尾。用户说“我要退货”和说“我想问问价格”，后面的处理逻辑完全不同。这类条件分支如果写在一大段 Python 里，可读性会迅速劣化；如果组合原语里也有一个标准组件能代表“按条件选路由”，整条链就能保持图表化结构。

RunnableBranch 就是这样一个组件。它的构造方式是传入一组（条件函数，对应 Runnable）二元组，最后再给一个兜底 Runnable：

from 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_text=make_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( issue=classify_issue, keywords=extract_keywords, sentiment=get_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(reply=branch_step) )

这样 Runnable 的输出就会同时包含 issue、keywords、sentiment、reply，以及前面保留的 raw 和 source 字段。因为 normalize_step 已经保证输入里有 raw，extract_step 又把抽取字段挂了上去，branch_step 虽然输入是整个字典，但它只读自己需要的 issue、keywords、sentiment，最后通过 assign 把生成的 reply 追加回去，和分支输入并存在结果字典中。

注意，这里我用的是RunnablePassthrough.assign(reply=branch_step)，而不是再写一个合并字典的 Lambda，这是为了保持“不丢上下文”的语义。后续无论加多少字段，只要某个字段依赖已有字段，都放在 assign 里，链路结构就不会破坏。

3.3 LCEL 与显式 Runnable 的选择

上述代码用的是管道符，这是 LCEL 风格，优点是可读性好、嵌套自然。另一种写法是显式实例化 RunnableSequence：

from langchain_core.runnables import RunnableSequence chain = RunnableSequence(first=normalize_step, last=extract_step)

显式写法适合你需要在配置系统里动态指定链路、或者序列结构需要在运行时修改的场景。实际上 LCEL 底层也会编译成 RunnableSequence，两种风格性能没有差别，选择标准主要是团队可读性和业务需要。

4. 更复杂的组合技巧：参数注入、兜底重试、可观测性

4.1 用 bind 预绑定执行参数

在组合链路里，经常需要给某个步骤单独传一些参数，但这个参数不是上一个节点的输出。比如模型调用时想固定 temperature=0.2、max_tokens=500，这些参数如果写明在业务代码里，链路就不干净了。Runnable 提供了.bind()方法，可以提前把参数绑定到一个准备好的调用上：

llm = get_llm() summarize_chain = llm.bind(temperature=0.2, max_tokens=500) | StrOutputParser()

这里的 bind 可以理解成准备了一个“预设参数的模板调用”，它仍然是一个 Runnable，可以被正常组合。绑定后再和其他步骤相接，链路不关心参数是怎么注入的。如果你在做 A/B 测试，也可以从配置里动态决定绑定哪些参数，然后把不同版本分别放进回退列表。

还要区分 bind 和 with_config：bind 是改调用的运行时参数，对模型来说就是采样参数这类；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_start=on_start, on_end=on_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 一个速查表，方便对照

我把实际工作中见过的高频问题整理成一张表，不一定覆盖所有场景，但能帮你少走很多弯路。

先画数据流再写代码，是避免这些问题的根本办法。我写组合链的习惯是：动手前，先在白板上画数据流图，不画组件关系，只画每一跳的数据形态。哪一步输出字符串，哪一步输出字典，哪一步会新增字段，全部标出来。画完之后，代码基本就是照着图填空。很多新手纠结要不要用 RunnablePassthrough、要不要用 RunnableParallel，一追问数据流图就立刻清楚了：如果下一跳需要原始输入，就用透传或 assign；如果多个计算互不依赖，就用并行。

这个习惯帮我避开了大量返工，也让我在评审别人链路时能快速定位问题。组合原语的魅力，从来不是省几行代码，而是让复杂业务里的数据流动变得可见、可测、可复用。下次再搭一条流程时，先抽离几个小能力，再用原语把它们串起来，你会有完全不一样的体验。