LCEL链调试：用RunnableLambda打造透明节点，轻松查看中间数据

调试 LCEL 链时，我犯过最经典的错误，就是在prompt | llm | parser中间加一行print()，然后看着报错发呆。LCEL 不是一行行命令式代码，它是一条声明式管道，中间产物没有“变量名”给你随时查看。后来我学乖了：把print()塞进一个既能接收输入、又能原样返回输入的节点里，再通过RunnableLambda挂到链上，这样中间数据随便看，链路行为完全不变。这套思路最终封装成了debug_runnable()，也就是今天这篇要聊的调试链方案。

如果你正在写 RAG 管道、多步 Agent 编排，或者只是想把一条链拆开看看每一步到底传了什么格式的数据，这篇应该能帮你在几分钟内给链条装上“仪表盘”。文里的核心实现可以直接复制使用，也会讲清楚为什么这样设计，以及实际踩过的坑。

1. 先弄明白：为什么 LCEL 链里看不见中间结果

1.1 声明式链和命令式思维是两回事

用 LangChain 写链，很多人会不自觉地把它当成函数串：

result = parser.invoke(llm.invoke(prompt.invoke(user_input)))

但换成 LCEL 写法后：

chain = prompt | llm | parser result = chain.invoke(user_input)

你没法在prompt和llm之间临时插一句打印，因为这两个 Runnable 已经通过|组合成了一个整体。链条内部每一步的输出，对外部来说就像流水线上被遮住的零件，你只知道最后出来成品，不知道中间到了哪一步、变成什么样。

这不是 LangChain 故意难为你。声明式组合的好处是：链可以被整体调度、并行执行、异步调用，甚至输出成图结构来分析。可调试性就要付出代价——你失去了随手写 print 的“命令式自由”。

1.2 直接 print 为什么不灵

你可能会想：那我封装一个函数，在函数里打印，再放到链里行不行？可以，但有个关键细节。

假设你写了：

def debug_print(x): print(x) return x chain = prompt | debug_print | llm | parser

只要函数返回x，这个思路就能跑通。真正的问题是，你会遇到四种更棘手的情况：

• 打印出来的东西既长又没重点，比如直接打印一个Document对象，几百行 metadata 混在一起；
• 多个分支并行时，打印顺序乱成一锅粥；
• 异步链里 print 的输出可能被缓冲，看不到实时日志；
• 调试完忘记删，生产环境里刷屏。

所以我需要的不是一个“能打印的 lambda”，而是一个能统一处理这些问题的工具函数。

1.3 debug_runnable() 要解决的核心诉求

自己动手封装debug_runnable()时，我心里列了一份需求清单：

• 不用手改原链，用一个函数把链条包起来，自动在每一步之间插入调试节点；
• 打印内容要结构化：字符串、字典、列表、Document、消息对象分别处理；
• 遵循“透传节点”的规矩：数据进去看一眼，原样返回，绝不影响链路；
• 支持开关，用环境变量控制，生产环境不刷屏。

这个函数名字是我自己起的，官方没有这个概念。但它的思想来自 LangChain 生态里最常见的调试组合拳：RunnableLambda、RunnablePassthrough、回调钩子，以及“副作用节点”的设计模式。

2. debug_runnable() 的两个基础配件：透传节点、副作用节点、透明节点

2.1 RunnablePassthrough 与透明节点的直觉

在 LangChain 里，RunnablePassthrough是一个特殊 Runnable：给它什么，它原样返回什么。它最常见的用法是在 Runnable 字典里透传某个键：

from langchain_core.runnables import RunnablePassthrough chain = { "context": retriever, "question": RunnablePassthrough() }

输入是{"question": "什么是向量数据库"}时，RunnablePassthrough()负责把question这个键原样带进输出字典，而retriever负责产出context。

这种“原样通过”的行为，是“透传”两个字最直白的解释。也是这一类调试节点成立的前提：如果你想在链路里塞一个观察点，又不希望它改变数据流，那么它必须是一个透传节点。

很多人在命名上会把“透明节点”和“透传节点”混用。我的理解是：透传强调的是“不改数据”，透明强调的是“你能看到数据”。调试节点同时需要这两个特性：对外透传，对内透明，合起来就是一个“玻璃盒子”。东西穿过去，你能看见里面长什么样。

2.2 RunnableLambda 包出来的副作用节点

RunnableLambda可以把任意 Python 函数包装成 Runnable。这个能力特别适合做“副作用节点”。

“副作用”这个词来自函数式编程：函数的主要职责是“计算并返回值”，但如果你在函数里打印日志、计数、上报指标，这些动作就是副作用。副作用节点的特点在于：主数据流不变，只额外执行观察行为。

from langchain_core.runnables import RunnableLambda def debug_node(x): # 副作用：打印 print(f"当前数据类型: {type(x).__name__}") # 主要职责：原样返回 return x chain = prompt | RunnableLambda(debug_node) | llm | parser

注意顺序：先print，再return x。如果你把return x写在了 print 前面，或者漏写了 return，链条就会断掉。数据到这一步就变成了None，后面所有节点都会报错或者产生诡异输出。这是我在实际项目中见过最高频的翻车点。

2.3 为什么这两类节点容易混淆

不少人会把RunnablePassthrough和RunnableLambda(debug_node)记混。其实它们的定位不同：

• RunnablePassthrough本身没有副作用，它只是原样返回；
• RunnableLambda(debug_node)因为函数里写了 print，成了“带副作用的透传节点”。

debug_runnable()用的是后者。它不用依赖 RunnablePassthrough，只用 RunnableLambda 包装自己的打印函数，就能生成一个复用的调试节点。理解这个区别后，后面看代码就不会绕。

3. 手写一个通用 debug_runnable()：从拆链到智能打印

3.1 接口应该长什么样

我给debug_runnable()定了四个参数：

• chain：需要调试的 LCEL 链；
• prefix：打印日志的前缀，方便区分多条链；
• max_len：单条 preview 最大长度，防止刷屏；
• enabled：总开关，默认为 True，生产环境传入 False 或读环境变量。

chain = prompt | llm | parser debug_chain = debug_runnable(chain, prefix="qa", max_len=120, enabled=True)

返回的新链和原链的输入输出类型完全一致，只是多了一些打印节点。

3.2 把链拆开再嵌回去

LCEL 组合链在底层一般是RunnableSequence，它内部保存了每一步的 Runnable，通常可以通过chain.steps拿到步骤列表。debug_runnable()的核心逻辑就两步：

1. 把chain.steps拆出来；
2. 在每两个步骤之间插入“透明节点”，同时保留原步骤顺序。

插入的原则是：在步骤 A 输出之后、步骤 B 输入之前，放一个打印节点。这样既能打印 A 的结果，又不影响 B 的执行。

result = 打印节点(链输入) | 步骤0 result = result | 打印节点(步骤0输出) | 步骤1 result = result | 打印节点(步骤1输出) | 步骤2

这段逻辑翻译成代码就是：

def debug_runnable(chain, prefix="chain", max_len=300, enabled=True): if not enabled: return chain steps = getattr(chain, "steps", None) if steps is None: steps = [chain] else: steps = list(steps) result = _make_transparent(f"{prefix} 输入", max_len) | steps[0] for i, next_step in enumerate(steps[1:], start=1): result = ( result | _make_transparent(f"{prefix}[{i - 1}] 输出", max_len) | next_step ) result = result | _make_transparent(f"{prefix}[{len(steps) - 1}] 输出", max_len) return result

有几个边界情况要说明：如果链本身不是RunnableSequence，没有steps属性，那就把它当单一节点处理，只在入口和出口打印。如果链条已经被人手动包装过，steps数组里可能是子链，打印节点仍能正常工作，只是日志粒度会粗一些。

注意：chain.steps在 LangChain 内部属于“公开但未写入文档”的属性，不同版本可能调整。建议在使用前先打印type(chain)和chain.steps确认一下。我测试过常见 LCEL 组合链，这个做法是稳定的。

3.3 智能打印 Document 与消息对象

直接print(document)会把 page_content 和整个 metadata 混在一起输出，内容稍微长一点就什么都看不清。更好的做法是分类处理。

我需要两个辅助函数：

• _fmt()：生成单行摘要；
• _dump()：递归展开字典、列表、文档、消息对象。

先看_fmt：

def _fmt(value, max_len=300): if isinstance(value, Document): content = value.page_content return ( f"Document(page_content_len={len(content)}, " f"metadata={list(value.metadata.keys())[:6]}, " f"preview={content[:max_len]!r})" ) if isinstance(value, BaseMessage): content = value.content if isinstance(value.content, str) else str(value.content) return ( f"{type(value).__name__}(content_len={len(content)}, " f"preview={content[:max_len]!r})" ) if isinstance(value, str): return f"str(len={len(value)}, preview={value[:max_len]!r})" if isinstance(value, dict): return f"dict(keys={list(value.keys())[:10]})" if isinstance(value, (list, tuple)): return f"{type(value).__name__}(len={len(value)})" if hasattr(value, "messages"): messages = value.messages return ( f"{type(value).__name__}(messages_len={len(messages)}, " f"types={[type(m).__name__ for m in messages]})" ) return f"{type(value).__name__}={str(value)[:max_len]!r}"

_dump则负责递归展开：

def _dump(value, indent="", max_len=300): if isinstance(value, dict): print(f"{indent}dict[{len(value)}]") for k, v in value.items(): secret_keys = ("key", "token", "secret", "password") if any(s in k.lower() for s in secret_keys): print(f"{indent} {k}: ***") continue print(f"{indent} {k}: {_fmt(v, max_len)}") if isinstance(v, (dict, list, tuple)) and not isinstance(v, Document): _dump(v, indent + " ", max_len) elif isinstance(value, (list, tuple)): print(f"{indent}{type(value).__name__}[{len(value)}]") for i, item in enumerate(value[:5]): print(f"{indent} [{i}] {_fmt(item, max_len)}") if isinstance(item, (dict, list, tuple, Document, BaseMessage)): _dump(item, indent + " ", max_len) elif hasattr(value, "messages"): print(f"{indent}{type(value).__name__}(messages_len={len(value.messages)})") for i, msg in enumerate(value.messages): print(f"{indent} [{i}] {_fmt(msg, max_len)}") else: print(f"{indent}{_fmt(value, max_len)}")

为什么费劲做递归展开？因为 RAG 链里最常见的数据结构就是“字典包列表，列表里是 Document”，如果只打印一层字典，你还是看不到文档内容。递归展开后，一条链从输入到输出的数据结构变化会非常直白。

3.4 透明节点本身：只打印，不触碰数据

透明节点用RunnableLambda生成：

def _make_transparent(name, max_len=300): def node(x): print(f"\n===== [DEBUG] {name} =====", flush=True) _dump(x, indent="| ", max_len=max_len) return x return RunnableLambda(node)

两个细节值得单独说。

第一，flush=True很重要。Python 的 print 默认走缓冲区，在异步链路里往往要等缓冲区满了才显示。加上flush=True才能保证每行日志实时出现，尤其是在 Jupyter、容器或者重定向日志文件的时候。

第二，函数必须return x。无论_dump打印了多少内容，节点的返回值仍然是你传入的原始 x。这就是副作用节点和普通节点的本质区别：打印是外部动作，返回是内部职责。

3.5 用环境变量做总开关

调试节点不能带进生产环境。最省事的办法是用环境变量控制：

import os def debug_runnable(chain, prefix="chain", max_len=300): enabled = os.environ.get("LCEL_DEBUG") == "1" if not enabled: return chain # ... 前面说的包装逻辑

这样命令行里跑：

LCEL_DEBUG=1 python my_chain.py

就能看到完整调试链；平时直接运行则完全无感。我自己的习惯是：默认关闭，需要排查时再开。调试完不开着，防止日志把真正的报错埋掉。

4. RAG 链实战：把 Document 与消息对象打印得明明白白

4.1 搭一条带检索的链

构造一条演示用 RAG 链。模型调用这里用一个示意占位model_client，实际换成你正在用的模型客户端即可：

from langchain_core.documents import Document from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough class DemoRetriever: def invoke(self, question): return [ Document( page_content="向量数据库存储的是非结构化数据的向量表达，常用于语义检索。", metadata={"source": "demo-1.txt", "page": 1}, ), Document( page_content="RAG 的核心流程是先检索后生成，检索结果作为模型输入上下文。", metadata={"source": "demo-2.txt", "page": 2}, ), ] async def ainvoke(self, question): return self.invoke(question)

然后构建链：

prompt = ChatPromptTemplate.from_messages([ ("system", "只根据资料回答问题，控制在两句话以内。"), ("human", "资料：\n{context}\n\n问题：{question}"), ]) retriever = DemoRetriever() model_client = ChatModel(...) # 示意，换成你正在用的模型调用 parser = StrOutputParser() base_chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt | model_client | parser )

这里已经用到了两处标题里的概念：RunnablePassthrough是透传节点，负责让question键原样穿过字典层；retriever.invoke()会返回 Document 对象列表。

4.2 套上 debug_runnable() 看输出

把整条链包进调试函数：

debug_chain = debug_runnable(base_chain, prefix="rag", max_len=100) answer = debug_chain.invoke({"question": "向量数据库是什么？"}) print("\n最终答案:", answer)

控制台会输出类似下面这样一段日志：

===== [DEBUG] rag 输入 ===== | dict[1] | question: str(len=12, preview='向量数据库是什么？') ===== [DEBUG] rag[0] 输出 ===== | dict[2] | question: str(len=12, preview='向量数据库是什么？') | context: list[2] | [0] Document(page_content_len=40, metadata=['source', 'page'], preview='向量数据库存...') | [1] Document(page_content_len=38, metadata=['source', 'page'], preview='RAG 的核心...') ===== [DEBUG] rag[1] 输出 ===== | ChatPromptValue(messages_len=2, types=['SystemMessage', 'HumanMessage']) ===== [DEBUG] rag[2] 输出 ===== | AIMessage(content_len=..., preview='向量数据库是一种...') ===== [DEBUG] rag[3] 输出 ===== | str(len=..., preview='向量数据库是一种...')

这段日志的价值，不是看“打印出来了”，而是看数据在每个环节的形状变化：

• 链输入是普通字符串，在 Runnable 字典里变成了dict；
• retriever输出是list[2]，里面是两条Document；
• prompt把字典拼成了ChatPromptValue，里面包含 SystemMessage 和 HumanMessage；
• 模型输出是AIMessage，最后 parser 把它抽成纯字符串。

如果哪一步报错，比如模型拿到的是空 context，你一眼就能从日志里发现：context: list[0]，后面 prompt 阶段只有 HumanMessage 里没有资料内容。这比看堆栈快多了。

4.3 怎么看懂 Document 打印结果

很多人在这一步会犯晕：Document(page_content_len=40, metadata=['source', 'page'], preview='...')里为什么不直接打印全文？因为默认 preview 截断到max_len，我只在需要时把它调大：

debug_chain = debug_runnable(base_chain, prefix="rag", max_len=500)

对于 Document 对象，page_content 是检索出来的正文，metadata 是来源、页码等依附信息。调试时先看 metadata 的 key 是否齐全，再看 page_content 的长度是否正常，通常就能定位大部分“检索为空”“上下文过于冗长”的问题。

5. 常见问题与排查速查：打印不出、乱序、截断、泄露

5.1 print 为什么完全没有出现

最常见的原因有三个：

• 你构造的是新链，但忘了调用debug_runnable()的返回值，实际跑的仍是原链；
• 链条本身没有被invoke()/ainvoke()，LCEL 是惰性求值，构建链并不会执行任何步骤；
• enabled参数或环境变量没有打开。

先把这三件事检查一遍，再去看代码。我建议在任何入口处先加一行print(debug_chain is base_chain)，确认你拿到的确实是被包装过的链。

5.2 打印顺序乱掉

多条链并行、或者使用.batch()批量执行时，print 日志会线程交错。这是 print 调试的天然劣势。处理办法有两种：

• 给每个节点命名时带上链前缀，比如rag[0]，抓日志时用关键字过滤；
• 如果必须严格保序，考虑用 Python 标准库logging并指定线程编号。

对大多数本地调试场景，我觉得再加一个chain_id就行，不用一开始就上日志体系。

5.3 内容被截断到看不见重点

max_len太小，preview 看不出上下文。调大即可：

debug_chain = debug_runnable(base_chain, prefix="rag", max_len=2000)

但要注意，调大之后 LLM 的完整输出可能很长，终端翻页也麻烦。我更推荐先用小参数确认“类型和长度对不对”，再针对可疑节点单独调大。

5.4 打印 Document 变成了超大 JSON

如果你看到的是“metadata 非常多、page_content 淹没在 JSON 里”，大概率是没用_fmt/_dump辅助函数，直接对对象做了repr。正确的做法是：永远不要直接print(document)，要走_dump的分类路径。

同理，消息对象也不要直接打印。AIMessage可能除了 content 之外还有tool_calls、additional_kwargs，这些在调试阶段看一眼摘要就够，全部展开反而干扰判断。

5.5 密钥被打印出来怎么办

字典里如果出现api_key、token、secret这类键，_dump已经做了脱敏：直接显示***。但你自己的 prompt 模板里可能已经拼了密钥，那就不在脱敏范围内。

我另外建议一条铁律：不要在生产链路里开启调试开关；不要在日志里记录完整请求参数。debug_runnable()的开关设计就是为了这个，别把它当成常驻日志工具。

下面是一张速查表，方便今后直接对号入座：

现象	常见原因	处理方式
没有任何打印输出	未调用返回的新链	重新赋值给新变量并确认 invoke
只有“输入”没有中间节点	链不是 RunnableSequence	检查chain.steps是否存在
日志顺序错乱	并行分支 / batch 调用	用前缀过滤或改日志框架
preview 过短	max_len 太小	调大 max_len
Document 内容被 metadata 淹没	直接 print 对象	改用 _fmt / _dump 分类打印
密钥泄露	字典 key 含敏感词	使用内置脱敏逻辑，生产不开调试

5.6 和get_graph().print_ascii()有什么区别

LangChain 自带chain.get_graph().print_ascii()，能打印链的结构图，但它只显示“有哪些节点”，看不到“节点之间流动的数据”。debug_runnable()看的是数据，二者互补。

我通常先跑print_ascii()确认结构，再上debug_runnable()观察数据。结构错了时看数据只会更晕，先治好结构问题，再调数据问题。

6. 再进一步：把打印升级成日志源

6.1 一行代码切换 logger

print 调试够用，但如果团队成员多、链也多，print 日志很难检索。可以把透明节点内部从print换成logging.debug，改动很小：

import logging logger = logging.getLogger("lcel.debug") def _make_transparent(name, max_len=300): def node(x): logger.debug("===== %s ===== 数据: %s", name, _custom_preview(x, max_len)) return x return RunnableLambda(node)

logging的好处是可以配 handler、加时间戳、按模块过滤。坏处是调试反馈不如 print 直接，你还要先配置 logging 等级。我的建议是：个人排查用 print 版本，团队需要日志归档时再切 logging 版本。

6.2 调试节点只服务于“人”

最后提一句设计观念上的体会。LCEL 链本身是为了让程序高效执行，调试节点则是为了让人看懂执行过程。这两件事的优先级会冲突：调试节点太多，链会变慢；数据全都打印，眼睛会疲劳。

所以debug_runnable()的正确用法是：平时关掉，排查时打开；打开时先看摘要，锁定可疑节点后再放大。别把调试逻辑写进业务代码里，更别让一条干净的生产链背负一堆 print。你真正需要的不是更多日志，而是“需要时看得见、不需要时完全消失”的观察口。这也是我把开关放在函数第一位的原因。