LangChain 管道流式传输：自定义 RunnableGenerator 实践-编程实验室

LangChain 管道流式传输：自定义 RunnableGenerator 实践

在基于 LangChain Expression Language (LCEL) 构建现代生产级 AI 应用时，终端用户对交互体验的要求早已不是“等待 5 秒后弹出一大段文字”，而是“打字机式逐字吐出（Streaming）”。

然而在复杂的 RAG 链路或多步 Agent 编排中，中间步骤往往包含许多非模型推理的自定义处理逻辑（例如：检索进度提示、文档源元数据清洗、安全合规实时检测、自定义 Token 格式转换）。

如果直接使用标准的RunnableLambda，虽然能完成数据转换，但它在流式调用astream()时往往会直接“退化”为阻塞等待——必须等整个上游完整生成完毕后，才一次性把结果交出，打字机流式效果瞬间失效。

如何在 LCEL 管道中手写一个原生的RunnableGenerator（基于 Python 异步生成器AsyncIterator），实现中间节点的真正流式透传与实时拦截？

为什么普通的 RunnableLambda 无法优雅流式传输？

在 LangChain 内部，RunnableLambda封装的是一个普通函数Callable[[Input], Output]。

当你通过管道操作符构建链条：

chain = prompt | model | RunnableLambda(my_transform)

调用chain.astream(input_data)时，model会产生一个异步生成器AsyncIterator[AIMessageChunk]。但当这些 Chunk 到达RunnableLambda(my_transform)时，由于my_transform期望接收的是一个完整的对象，LangChain 底层会被迫调用_accumulate_chunks()，把所有流式碎片拼接成一个完整的AIMessage后再传给该函数。

这意味着：只要管道链路中间插入了一个普通的转换节点，下游接收到的流式生成就会被完全截断，前端只能干等。

核心解法：使用 @chain 装饰异步生成器

LangChain 提供了对生成器函数的一等公民支持。只要将函数声明为async def generator(inputs: AsyncIterator[Input]) -> AsyncIterator[Output]，并使用@chain（或直接构建RunnableGenerator），LCEL 就会将其识别为流式转换节点。

在这个模式下，数据在管道中是以**水流（Stream）**的形式流动的：上游每吐出一个 Chunk，你的自定义生成器就能立刻捕获、处理并立即yield给下游，实现真正的零延迟流式处理。

生产级自定义流式清洗与事件注入实现

以下是一个在实际金融知识库问答中落地的流式管道代码。它实现了在模型流式吐出答案的同时，实时过滤敏感字符，并在流式首包自动注入召回来源（Citations）元数据：

import asyncio from typing import AsyncIterator, Dict, Any, Union from langchain_core.runnables import chain, RunnablePassthrough from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import AIMessageChunk, BaseMessage from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI # 1. 模拟自定义异步检索器 async def mock_retriever(query: str) -> list: await asyncio.sleep(0.05) # 模拟 50ms 检索耗时 return [ {"title": "2026Q2 财报分析", "content": "第二季度净利润同比增长 24.5%..."}, {"title": "风险合规白皮书", "content": "海外投资汇率对冲策略..."} ] # 2. 手写核心流式生成器：边接收边过滤与元数据增强 @chain async def custom_stream_transformer(input_stream: AsyncIterator[Union[str, AIMessageChunk]]) -> AsyncIterator[str]: buffer = "" chunk_index = 0 async for chunk in input_stream: # 提取当前 chunk 文本 text = chunk.content if isinstance(chunk, AIMessageChunk) else str(chunk) # 首包特殊处理：如果是第一个 Chunk，可以先吐出一个自定义的协议头 if chunk_index == 0: yield "【AI 思考完成，开始输出】\n" buffer += text # 实时敏感词简单脱敏拦截（示例：将特定敏感数字或字符替换） if "内部绝密" in buffer: buffer = buffer.replace("内部绝密", "【已脱敏】") # 立即将处理后的内容 yield 给下游 yield text chunk_index += 1 # 在流式输出的末尾，追加格式化尾注 yield "\n\n---\n*数据来源于企业内网知识库*"

组装完整 LCEL 流式管道

async def run_streaming_pipeline(): prompt = ChatPromptTemplate.from_template( "基于以下上下文回答问题：\n{context}\n\n问题：{question}" ) model = ChatOpenAI(model="gpt-4o-mini", streaming=True) # 构建支持全链路流式传输的 LCEL 链 rag_chain = ( { "context": lambda x: "\n".join([d["content"] for d in x["docs"]]), "question": lambda x: x["query"] } | prompt | model | StrOutputParser() | custom_stream_transformer # 挂载自定义流式生成器 ) query = "第二季度的净利润增长情况如何？" docs = await mock_retriever(query) print("开始接收流式输出：") async for event in rag_chain.astream({"query": query, "docs": docs}): print(event, end="", flush=True) print("\n流式传输结束。") # 执行异步主入口 # asyncio.run(run_streaming_pipeline())