AI Agent白手起家40: LangChain 链式调用实战：管道、流式、并行与迁移
纲要
- 链的基本生成方式
- 使用管道操作符
|快速串联组件 - 使用
.pipe()方法构建链
- 使用管道操作符
- 流式调用
- 同步流式
stream() - 异步流式
astream() - 事件流
astream_events()与过滤（按 name、tag、事件阶段）
- 同步流式
- 并行执行多条链
RunnableParallel同时运行多个分支并合并结果
- 链的可视化调试
get_graph()绘制链路图get_prompts()查看所有提示词
- 从旧版预制链迁移到 LCEL
LLMChain→ LCELStuffDocumentsChain→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 应用。