调试 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, prefixqa, max_len120, enabledTrue)返回的新链和原链的输入输出类型完全一致只是多了一些打印节点。3.2 把链拆开再嵌回去LCEL 组合链在底层一般是RunnableSequence它内部保存了每一步的 Runnable通常可以通过chain.steps拿到步骤列表。debug_runnable()的核心逻辑就两步把chain.steps拆出来在每两个步骤之间插入“透明节点”同时保留原步骤顺序。插入的原则是在步骤 A 输出之后、步骤 B 输入之前放一个打印节点。这样既能打印 A 的结果又不影响 B 的执行。result 打印节点(链输入) | 步骤0 result result | 打印节点(步骤0输出) | 步骤1 result result | 打印节点(步骤1输出) | 步骤2这段逻辑翻译成代码就是def debug_runnable(chain, prefixchain, max_len300, enabledTrue): 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:], start1): 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()递归展开字典、列表、文档、消息对象。先看_fmtdef _fmt(value, max_len300): if isinstance(value, Document): content value.page_content return ( fDocument(page_content_len{len(content)}, fmetadata{list(value.metadata.keys())[:6]}, fpreview{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)}, fpreview{content[:max_len]!r}) ) if isinstance(value, str): return fstr(len{len(value)}, preview{value[:max_len]!r}) if isinstance(value, dict): return fdict(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)}, ftypes{[type(m).__name__ for m in messages]}) ) return f{type(value).__name__}{str(value)[:max_len]!r}_dump则负责递归展开def _dump(value, indent, max_len300): 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_len300): def node(x): print(f\n [DEBUG] {name} , flushTrue) _dump(x, indent| , max_lenmax_len) return x return RunnableLambda(node)两个细节值得单独说。第一flushTrue很重要。Python 的 print 默认走缓冲区在异步链路里往往要等缓冲区满了才显示。加上flushTrue才能保证每行日志实时出现尤其是在 Jupyter、容器或者重定向日志文件的时候。第二函数必须return x。无论_dump打印了多少内容节点的返回值仍然是你传入的原始 x。这就是副作用节点和普通节点的本质区别打印是外部动作返回是内部职责。3.5 用环境变量做总开关调试节点不能带进生产环境。最省事的办法是用环境变量控制import os def debug_runnable(chain, prefixchain, max_len300): enabled os.environ.get(LCEL_DEBUG) 1 if not enabled: return chain # ... 前面说的包装逻辑这样命令行里跑LCEL_DEBUG1 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_contentRAG 的核心流程是先检索后生成检索结果作为模型输入上下文。, 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, prefixrag, max_len100) answer debug_chain.invoke({question: 向量数据库是什么}) print(\n最终答案:, answer)控制台会输出类似下面这样一段日志 [DEBUG] rag 输入 | dict[1] | question: str(len12, preview向量数据库是什么) [DEBUG] rag[0] 输出 | dict[2] | question: str(len12, preview向量数据库是什么) | context: list[2] | [0] Document(page_content_len40, metadata[source, page], preview向量数据库存...) | [1] Document(page_content_len38, metadata[source, page], previewRAG 的核心...) [DEBUG] rag[1] 输出 | ChatPromptValue(messages_len2, types[SystemMessage, HumanMessage]) [DEBUG] rag[2] 输出 | AIMessage(content_len..., preview向量数据库是一种...) [DEBUG] rag[3] 输出 | str(len..., preview向量数据库是一种...)这段日志的价值不是看“打印出来了”而是看数据在每个环节的形状变化链输入是普通字符串在 Runnable 字典里变成了dictretriever输出是list[2]里面是两条Documentprompt把字典拼成了ChatPromptValue里面包含 SystemMessage 和 HumanMessage模型输出是AIMessage最后 parser 把它抽成纯字符串。如果哪一步报错比如模型拿到的是空 context你一眼就能从日志里发现context: list[0]后面 prompt 阶段只有 HumanMessage 里没有资料内容。这比看堆栈快多了。4.3 怎么看懂 Document 打印结果很多人在这一步会犯晕Document(page_content_len40, metadata[source, page], preview...)里为什么不直接打印全文因为默认 preview 截断到max_len我只在需要时把它调大debug_chain debug_runnable(base_chain, prefixrag, max_len500)对于 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, prefixrag, max_len2000)但要注意调大之后 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_lenDocument 内容被 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 一行代码切换 loggerprint 调试够用但如果团队成员多、链也多print 日志很难检索。可以把透明节点内部从print换成logging.debug改动很小import logging logger logging.getLogger(lcel.debug) def _make_transparent(name, max_len300): 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。你真正需要的不是更多日志而是“需要时看得见、不需要时完全消失”的观察口。这也是我把开关放在函数第一位的原因。