纲要

• 链的基本生成方式
  • 使用管道操作符|快速串联组件
  • 使用.pipe()方法构建链
• 流式调用
  • 同步流式stream()
  • 异步流式astream()
  • 事件流astream_events()与过滤（按 name、tag、事件阶段）
• 并行执行多条链
  • RunnableParallel同时运行多个分支并合并结果
• 链的可视化调试
  • get_graph()绘制链路图
  • get_prompts()查看所有提示词
• 从旧版预制链迁移到 LCEL
  • LLMChain→ LCEL
  • StuffDocumentsChain→create_stuff_documents_chain
• 完整可运行代码：使用模拟模型演示管道、并行与流式

引言

LangChain 的核心魅力在于“链”（Chain），即通过简洁的表达式将提示词、模型、解析器等组件串联成完整的处理流水线。从 v0.2 开始，LCEL（LangChain Expression Language）和Runnable接口成为构建链的推荐方式，彻底取代了旧版本中固化且难以定制的预制链。本文将通过可运行的代码，展示如何使用管道符、流式输出、并行执行以及如何从旧版链平滑迁移到现代 LCEL。

链的快速生成：管道操作符与.pipe()

使用|串联组件

|操作符可以将任意Runnable对象按顺序连接，数据从左向右流动，天然匹配“输入 → 提示词 → 模型 → 输出解析器”的经典 IO 流程。

fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_community.chat_models.fakeimportFakeListChatModel# 模拟模型，预设回答model=FakeListChatModel(responses=["为什么狗喜欢圆形骨头？因为不想骨头太累！"])prompt=ChatPromptTemplate.from_template("讲一个关于{topic}的笑话，不要任何解释。")# 使用管道符串联chain=prompt|model|StrOutputParser()print(chain.invoke({"topic":"狗"}))

使用.pipe()方法

.pipe()与管道符完全等价，下面的代码与上面的链效果相同。

chain=prompt.pipe(model).pipe(StrOutputParser())print(chain.invoke({"topic":"猫"}))

流式调用：同步流、异步流与事件过滤

流式输出能将生成结果逐块返回，适合构建打字机效果。

同步流式stream()

forchunkinchain.stream({"topic":"程序员"}):print(chunk,end="|")# 每个 chunk 后加分隔符方便观察

异步流式astream()

异步流式需使用async for，这里给出概念代码（在线 IDE 可能受限）。实际使用中只需替换模型即可。

asyncforchunkinchain.astream({"topic":"AI"}):print(chunk,end="")

事件流与过滤

astream_events()可以获取更细粒度的事件（如on_chat_model_start、on_chat_model_stream等），支持按 name、tag 或事件阶段过滤。例如仅捕获模型流式输出：

asyncforeventinchain.astream_events({"topic":"天气"},version="v2"):ifevent["event"]=="on_chat_model_stream":print(event["data"]["chunk"].content,end="")

若按 tag 过滤，可在构建链时添加配置：

fromlangchain_core.runnablesimportRunnableConfig chain_with_tag=chain.with_config(RunnableConfig(tags=["my_chain"]))

并行执行多条链

RunnableParallel能让多个分支同时处理同一输入，最后合并结果。

fromlangchain_core.runnablesimportRunnableParallel joke_prompt=ChatPromptTemplate.from_template("讲一个关于{topic}的笑话。")poem_prompt=ChatPromptTemplate.from_template("写一首关于{topic}的诗。")joke_chain=joke_prompt|model|StrOutputParser()poem_chain=poem_prompt|model|StrOutputParser()parallel_chain=RunnableParallel(joke=joke_chain,poem=poem_chain)result=parallel_chain.invoke({"topic":"程序员"})print("笑话:",result["joke"])print("诗:",result["poem"])

链的可视化调试

LCEL 内置了调试方法，可将链的结构导出为图或列出所有提示词。

# 打印链路图（需要安装 langchain 可视化扩展，此处仅示意）graph=chain.get_graph()print(graph.draw_ascii())# ASCII 图# 获取所有提示词forpinchain.get_prompts():print(p)

从旧版预制链迁移到 LCEL

LLMChain → LCEL

旧版：

fromlangchain.chainsimportLLMChainfromlangchain_core.promptsimportPromptTemplatefromlangchain_community.llms.fakeimportFakeListLLM llm=FakeListLLM(responses=["北京是中国的首都。"])old_chain=LLMChain(llm=llm,prompt=PromptTemplate.from_template("介绍一下{city}"))print(old_chain.run("北京"))

新版 LCEL：

fromlangchain_core.promptsimportPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_community.llms.fakeimportFakeListLLM llm=FakeListLLM(responses=["北京是中国的首都。"])new_chain=PromptTemplate.from_template("介绍一下{city}")|llm|StrOutputParser()print(new_chain.invoke({"city":"北京"}))

StuffDocumentsChain →create_stuff_documents_chain

旧版嵌套复杂，新版官方提供了create_stuff_documents_chain函数，接收 LLM 和提示词即可生成总结链，代码更简洁。

完整可运行代码

以下代码整合管道、并行和流式，使用模拟模型，无需任何 API Key。

安装依赖：

pipinstalllangchain langchain-core langchain-community

核心代码：

fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableParallelfromlangchain_community.chat_models.fakeimportFakeListChatModel# 模拟模型预设回答列表（分别用于笑话、诗）responses=["程序员为什么总在深夜工作？因为他们喜欢安静的代码。","代码如诗夜未央，键盘轻响伴月光。"]model=FakeListChatModel(responses=responses)# 构建笑话链和诗链joke_chain=(ChatPromptTemplate.from_template("讲一个关于{topic}的笑话。")|model|StrOutputParser())poem_chain=(ChatPromptTemplate.from_template("写一首关于{topic}的诗。")|model|StrOutputParser())# 并行执行parallel_chain=RunnableParallel(joke=joke_chain,poem=poem_chain)result=parallel_chain.invoke({"topic":"程序员"})print("并行结果：")print("笑话:",result["joke"])print("诗:",result["poem"])# 流式输出演示print("\n流式输出：")forchunkinjoke_chain.stream({"topic":"狗"}):print(chunk,end="|")

总结

LCEL 用|和.pipe()让链式调用变得像写表达式一样自然，搭配stream()、RunnableParallel以及内置的可视化调试工具，极大地提升了 LLM 应用开发的灵活性和可维护性。

对于旧版预制链，只需按照“提示词 + 模型 + 解析器”的模式重构，即可享受全部新特性。掌握这些技巧，你将能从容构建从简单对话到复杂并行的各类 AI 应用。