AI Agent白手起家40: LangChain 链式调用实战：管道、流式、并行与迁移

纲要链的基本生成方式使用管道操作符|快速串联组件使用.pipe()方法构建链流式调用同步流式stream()异步流式astream()事件流astream_events()与过滤按 name、tag、事件阶段并行执行多条链RunnableParallel同时运行多个分支并合并结果链的可视化调试get_graph()绘制链路图get_prompts()查看所有提示词从旧版预制链迁移到 LCELLLMChain→ LCELStuffDocumentsChain→create_stuff_documents_chain完整可运行代码使用模拟模型演示管道、并行与流式引言LangChain 的核心魅力在于“链”Chain即通过简洁的表达式将提示词、模型、解析器等组件串联成完整的处理流水线。从 v0.2 开始LCELLangChain Expression Language和Runnable接口成为构建链的推荐方式彻底取代了旧版本中固化且难以定制的预制链。本文将通过可运行的代码展示如何使用管道符、流式输出、并行执行以及如何从旧版链平滑迁移到现代 LCEL。链的快速生成管道操作符与.pipe()使用|串联组件|操作符可以将任意Runnable对象按顺序连接数据从左向右流动天然匹配“输入 → 提示词 → 模型 → 输出解析器”的经典 IO 流程。fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_community.chat_models.fakeimportFakeListChatModel# 模拟模型预设回答modelFakeListChatModel(responses[为什么狗喜欢圆形骨头因为不想骨头太累])promptChatPromptTemplate.from_template(讲一个关于{topic}的笑话不要任何解释。)# 使用管道符串联chainprompt|model|StrOutputParser()print(chain.invoke({topic:狗}))使用.pipe()方法.pipe()与管道符完全等价下面的代码与上面的链效果相同。chainprompt.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:天气},versionv2):ifevent[event]on_chat_model_stream:print(event[data][chunk].content,end)若按 tag 过滤可在构建链时添加配置fromlangchain_core.runnablesimportRunnableConfig chain_with_tagchain.with_config(RunnableConfig(tags[my_chain]))并行执行多条链RunnableParallel能让多个分支同时处理同一输入最后合并结果。fromlangchain_core.runnablesimportRunnableParallel joke_promptChatPromptTemplate.from_template(讲一个关于{topic}的笑话。)poem_promptChatPromptTemplate.from_template(写一首关于{topic}的诗。)joke_chainjoke_prompt|model|StrOutputParser()poem_chainpoem_prompt|model|StrOutputParser()parallel_chainRunnableParallel(jokejoke_chain,poempoem_chain)resultparallel_chain.invoke({topic:程序员})print(笑话:,result[joke])print(诗:,result[poem])链的可视化调试LCEL 内置了调试方法可将链的结构导出为图或列出所有提示词。# 打印链路图需要安装 langchain 可视化扩展此处仅示意graphchain.get_graph()print(graph.draw_ascii())# ASCII 图# 获取所有提示词forpinchain.get_prompts():print(p)从旧版预制链迁移到 LCELLLMChain → LCEL旧版fromlangchain.chainsimportLLMChainfromlangchain_core.promptsimportPromptTemplatefromlangchain_community.llms.fakeimportFakeListLLM llmFakeListLLM(responses[北京是中国的首都。])old_chainLLMChain(llmllm,promptPromptTemplate.from_template(介绍一下{city}))print(old_chain.run(北京))新版 LCELfromlangchain_core.promptsimportPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_community.llms.fakeimportFakeListLLM llmFakeListLLM(responses[北京是中国的首都。])new_chainPromptTemplate.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[程序员为什么总在深夜工作因为他们喜欢安静的代码。,代码如诗夜未央键盘轻响伴月光。]modelFakeListChatModel(responsesresponses)# 构建笑话链和诗链joke_chain(ChatPromptTemplate.from_template(讲一个关于{topic}的笑话。)|model|StrOutputParser())poem_chain(ChatPromptTemplate.from_template(写一首关于{topic}的诗。)|model|StrOutputParser())# 并行执行parallel_chainRunnableParallel(jokejoke_chain,poempoem_chain)resultparallel_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 应用。