LangChain链式编排：从脚本思维到AI工作流设计的实战指南

1. 项目概述从“单点调用”到“链式编排”的思维跃迁如果你已经跟着这个系列走过了前面的十四天那么恭喜你你已经掌握了从环境搭建、模型调用、提示词工程到RAG检索增强生成等一系列AI应用开发的“单兵技能”。你可能已经能写出一个调用大模型API的脚本或者构建一个简单的问答应用。但当你试图构建一个更复杂、更贴近真实业务场景的应用时比如一个能自动分析数据、生成报告、并发送邮件的智能助手你会发现一个棘手的问题这些“单兵技能”如何协同作战这就是我们第十五天也是这个系列收官之战的焦点使用LangChain封装AI执行链。这不仅仅是学习一个新工具更是一种开发范式的转变——从面向过程的“脚本思维”升级为面向流程的“编排思维”。LangChain提供的“链”Chain正是将多个离散的AI操作如调用模型、处理输入、解析输出以及工具如计算器、网络搜索、代码执行粘合在一起的“胶水”。它让你能够以声明式的方式定义复杂的工作流而无需陷入繁琐的状态管理和错误处理中。网络上关于“LangChain入门”、“LangChain组件”的搜索热度很高但很多教程停留在“Hello World”式的简单链演示。我们今天的目的是深入一步理解链的设计哲学掌握构建健壮、可维护执行链的核心方法并最终让你有能力将之前学到的所有知识点串联成一个真正可用的AI应用。无论是应对“AI应用开发面试题”中关于工作流设计的问题还是为你“儿子学了前端开发如今公司裁员现在想继续学AI应用与智能体开发”探索一条扎实的路径掌握链式编排都是不可或缺的核心能力。简单来说今天之后你的代码将不再是一堆零散的函数调用而是一个个清晰定义、可复用、可测试的“智能流程单元”。2. LangChain链的核心哲学为何是“链”而非“脚本”在深入代码之前我们必须先统一思想为什么需要“链”直接写Python脚本调用不同模块不行吗当然可以但对于AI应用尤其是涉及大模型这种非确定性产生的结果可能有变化组件的应用直接写脚本会迅速导致代码混乱且难以维护。让我们通过一个对比来理解。假设我们要实现一个“翻译并总结”的功能将一段中文文本翻译成英文然后总结英文文本的要点。“脚本思维”下的实现可能长这样import openai def translate_chinese_to_english(text): # 这里需要处理API密钥、构造提示词、处理响应、错误重试... prompt f将以下中文翻译成英文{text} response openai.ChatCompletion.create(...) # 需要解析response提取出翻译结果 translated_text response.choices[0].message.content return translated_text def summarize_english(text): # 同样需要处理一整套模型调用逻辑 prompt f总结以下英文文本的要点{text} response openai.ChatCompletion.create(...) summary response.choices[0].message.content return summary # 主程序 input_text 今天天气很好我们决定去公园野餐。公园里人很多孩子们在玩耍大人们在聊天。 try: translated translate_chinese_to_english(input_text) summary summarize_english(translated) print(summary) except Exception as e: print(f出错啦{e}) # 还需要决定是重试、回滚还是记录日志这段代码的问题显而易见重复代码每个函数都有相似的API调用、错误处理模板。硬编码逻辑流程先翻译后总结被固化在主程序里想调整顺序或插入新步骤比如先检查输入是否为空很麻烦。错误处理脆弱一个步骤失败整个流程就中断缺乏灵活的故障恢复或降级策略。难以测试和复用translate_chinese_to_english函数很难单独测试因为它和具体的API调用强耦合。“链式思维”下的LangChain实现LangChain的“链”将流程中的每一步抽象为一个“可链接的组件”。每个组件有明确的输入和输出规范。链负责将这些组件按顺序组合起来并自动处理数据在组件间的传递。它的核心优势在于声明式编排你关注的是“做什么”先翻译后总结而不是“怎么做”具体的API调用细节。流程变得清晰可见。组件化与复用翻译器和总结器可以被定义为独立的RunnableLangChain中的可运行单元可以在不同的链中被复用。内置的健壮性链可以配置重试逻辑、回退策略如主模型失败时自动切换备用模型、以及输出解析让整个流程更稳定。流式与异步支持链天然支持流式输出和异步执行这对于构建响应式的Web应用至关重要。所以当你搜索“langchain和langgraph的区别”时其本质是编排复杂度的不同。简单链处理线性流程而LangGraph基于LangChain用于处理有循环、分支等复杂状态的图工作流。今天我们先攻克最基础的链。3. 构建你的第一条执行链从PromptTemplate到LLMChain理论说再多不如动手。让我们从LangChain中最经典、最基础的链——LLMChain开始。它串联了一个提示词模板PromptTemplate和一个大语言模型LLM。首先确保你的环境已安装LangChain和对应的模型接入包如openai。pip install langchain langchain-openai3.1 定义组件提示词模板与模型在LangChain中一切皆可链接。我们首先创建两个基本组件。from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI # 1. 创建提示词模板 # 注意这里的 {topic} 是一个变量会在运行时被传入的具体值替换 prompt_template PromptTemplate.from_template( 请用简单易懂的方式为一位初学者解释什么是{topic}。 ) # 2. 创建大语言模型实例 # 替换your_api_key为你的实际密钥建议通过环境变量管理 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.5, openai_api_keyyour_api_key)这里有几个关键点PromptTemplate.from_template这是创建模板的便捷方法。模板中的花括号{}定义了输入变量。链在运行时会将实际值绑定到这些变量上。ChatOpenAI这是LangChain为OpenAI ChatGPT模型封装的类。temperature参数控制输出的随机性0.0更确定1.0更随机。对于解释类任务0.5是一个不错的起点平衡了准确性和可读性。API密钥管理永远不要将密钥硬编码在代码中应该使用os.environ[OPENAI_API_KEY]从环境变量读取。这是开发中的基本安全规范也是面试中常被考察的点。3.2 组装链LLMChain现在我们将这两个组件“链”起来。from langchain.chains import LLMChain # 3. 创建链 explanation_chain LLMChain(llmllm, promptprompt_template)是的就这么简单。LLMChain接收一个llm和一个prompt参数将它们封装成一个可执行单元。此时explanation_chain就是一个完整的、可复用的“解释器”。3.3 运行链与解析输出运行链有两种主要方式run和invoke(在较新版本中推荐使用invoke)。# 方式一使用 run 方法较旧但直观 result_v1 explanation_chain.run(topic神经网络) print(使用 run 方法的结果) print(result_v1) print(- * 50) # 方式二使用 invoke 方法新版推荐更统一 # invoke 方法接收一个字典键是提示词模板中的变量名 result_v2 explanation_chain.invoke({topic: 机器学习}) print(使用 invoke 方法的结果) # invoke 返回一个字典其中 text 或 output 字段是模型的文本回复 # 对于LLMChain输出通常在 text 字段 print(result_v2.get(text, result_v2)) print(- * 50) # 方式三流式输出适合Web应用 print(流式输出演示) for chunk in explanation_chain.stream({topic: 深度学习}): # chunk 可能是一个字典我们需要提取其中的文本增量 if hasattr(chunk, content): print(chunk.content, end, flushTrue) elif isinstance(chunk, dict) and text in chunk: print(chunk[text], end, flushTrue) # 在某些版本中chunk可能就是字符串 elif isinstance(chunk, str): print(chunk, end, flushTrue) print(\n - * 50)实操心得与避坑指南runvsinvokevsstreamrun方法简单但功能单一。invoke是LangChain新版Runnable接口的标准方法它返回结构化的字典便于获取元数据如token用量。stream用于流式响应在构建聊天界面时体验极佳。建议在新项目中统一使用invoke和stream。输出解析invoke返回的是包含input、output等键的字典。直接打印整个字典会很乱。通常你需要的是result[output]或result[text]。理解返回数据结构是调试的第一步。如果你遇到“langchain 打印invoke发送的内容”这类问题其实就是用print(result)查看这个字典的全貌。错误处理实际应用中一定要用try...except包裹链的调用处理可能的网络超时、API限额、模型错误等。LangChain也提供了RunnableWithFallbacks来配置备用模型。至此你已经完成了第一条链的构建和运行。它虽然简单但包含了链的所有核心要素输入、处理组件、输出。4. 构建复杂链SequentialChain与TransformChain现实任务很少一步到位。更多时候我们需要将多个步骤串联起来。这就是SequentialChain的用武之地。同时我们可能需要在模型调用前后进行一些纯Python的数据处理这时就需要TransformChain。让我们构建一个更实用的链“智能书评生成器”。它的工作流程是输入一本书的名称和作者。步骤一根据书名和作者生成一段该书的情节简介。步骤二基于生成的情节简介撰写一篇简短的书评包含亮点和不足。输出最终的书评。4.1 创建子链首先我们为每个步骤创建独立的子链。from langchain.chains import LLMChain, SequentialChain from langchain.prompts import PromptTemplate llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 子链1生成情节简介 synopsis_prompt PromptTemplate.from_template( 你是一位专业的图书编辑。请根据以下书籍信息撰写一段约150字的情节简介。 书籍名称《{book_title}》 作者{author} 请只输出情节简介不要添加其他评论。 ) synopsis_chain LLMChain(llmllm, promptsynopsis_prompt, output_keysynopsis) # 注意 output_key它定义了此链输出的变量名供后续链使用 # 子链2基于简介生成书评 review_prompt PromptTemplate.from_template( 你是一位资深的书评人。请基于以下书籍的情节简介撰写一篇约200字的书评。 书评应包含本书的核心亮点和可能存在的不足之处保持客观中立。 情节简介 {synopsis} 请开始你的书评 ) review_chain LLMChain(llmllm, promptreview_prompt, output_keyreview)关键设计点output_key这是SequentialChain正常工作的关键。它给每个子链的输出起了个名字。在review_chain的提示词模板中我们引用了{synopsis}这个变量名必须和synopsis_chain的output_key值完全一致。LangChain会自动完成数据传递。4.2 使用SequentialChain组装总链现在我们用SequentialChain把两个子链按顺序组装起来。# 定义总链 book_review_chain SequentialChain( chains[synopsis_chain, review_chain], # 按顺序执行 input_variables[book_title, author], # 总链的输入变量 output_variables[synopsis, review], # 总链的输出变量即各个子链的output_key verboseTrue # 设为True可以打印详细的执行日志调试时非常有用 ) # 运行总链 result book_review_chain.invoke({ book_title: 三体, author: 刘慈欣 }) print( 生成的情节简介 ) print(result[synopsis]) print(\n 生成的书评 ) print(result[review])当verboseTrue时你会在控制台看到类似下面的日志这非常有助于理解链的执行过程和数据流 Entering new SequentialChain chain... Entering new LLMChain chain... Prompt after formatting: 你是一位专业的图书编辑。请根据以下书籍信息撰写一段约150字的情节简介。 书籍名称《三体》 作者刘慈欣 请只输出情节简介不要添加其他评论。 Finished chain. Entering new LLMChain chain... Prompt after formatting: 你是一位资深的书评人。请基于以下书籍的情节简介撰写一篇约200字的书评。 书评应包含本书的核心亮点和可能存在的不足之处保持客观中立。 情节简介 这里会是上一步生成的简介内容 请开始你的书评 Finished chain. Finished chain.4.3 引入TransformChain进行数据清洗有时我们需要在模型调用之间进行一些非AI的数据处理比如格式化字符串、过滤信息、调用某个纯函数。TransformChain就是干这个的。假设我们在生成书评后想自动计算一下书评的字数并和简介字数一起输出。from langchain.chains import TransformChain from typing import Dict, Any # 定义一个纯Python的转换函数 def length_calculator(inputs: Dict[str, Any]) - Dict[str, Any]: 计算文本长度 synopsis inputs.get(synopsis, ) review inputs.get(review, ) return { synopsis_length: len(synopsis), review_length: len(review), synopsis: synopsis, # 原样传递供后续使用 review: review } # 创建TransformChain transform_chain TransformChain( input_variables[synopsis, review], # 它需要哪些输入 output_variables[synopsis_length, review_length, synopsis, review], # 它会产生哪些输出 transformlength_calculator # 核心转换函数 ) # 重新组装链简介链 - 书评链 - 转换链 enhanced_review_chain SequentialChain( chains[synopsis_chain, review_chain, transform_chain], input_variables[book_title, author], output_variables[synopsis, review, synopsis_length, review_length], verboseFalse ) final_result enhanced_review_chain.invoke({book_title: 百年孤独, author: 加西亚·马尔克斯}) print(f简介字数{final_result[synopsis_length]}) print(f书评字数{final_result[review_length]})注意事项输入输出变量必须匹配TransformChain的input_variables必须能从上游链的输出中获取到output_variables必须包含所有transform函数返回的字典键。这是链式数据流中最容易出错的地方务必仔细检查。保持函数纯净transform函数应该是无副作用的纯函数给定相同输入总是产生相同输出。避免在其中进行网络请求或修改全局状态。调试技巧当链不按预期工作时第一件事就是把verboseTrue打开看数据到底在哪一步变成了什么样。这也是解决“langchain流式输出吞掉reasoning-content字段”之类问题的基础方法——先看清楚完整的中间状态。通过SequentialChain和TransformChain你已经可以构建出绝大多数线性的AI工作流了。它们将复杂的业务逻辑分解为一个个可测试、可维护的小单元极大地提升了开发效率和应用可靠性。5. 高级链与实战技巧RouterChain与自定义链当你的应用需要根据输入内容动态选择不同的处理分支时简单的顺序链就不够用了。这就需要用到RouterChain。此外为了应对更特殊的场景我们可能需要创建完全自定义的链。5.1 使用RouterChain实现动态路由假设我们有一个客服系统需要根据用户问题类型路由到不同的专业处理链技术问题链、账单问题链和普通咨询链。from langchain.chains.router import MultiPromptChain from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser from langchain.prompts import PromptTemplate # 1. 定义不同目的地的提示词模板和信息 destinations [ { name: technical, description: 擅长回答关于产品技术细节、API使用、错误代码的问题, prompt_template: 你是一位资深技术专家。请专业且清晰地回答以下技术问题\n\n{input} }, { name: billing, description: 擅长处理账单、订阅、支付和退款相关的问题, prompt_template: 你是一位耐心细致的客服专员。请友好地帮助用户解决以下账单问题\n\n{input} }, { name: general, description: 擅长回答一般性的产品咨询、功能介绍等非技术非账单问题, prompt_template: 你是一位热情的产品顾问。请解答以下一般性咨询\n\n{input} } ] # 2. 为每个目的地创建对应的LLMChain destination_chains {} for dest in destinations: prompt PromptTemplate.from_template(dest[prompt_template]) chain LLMChain(llmllm, promptprompt) destination_chains[dest[name]] chain # 3. 创建一个默认链当路由器无法确定时使用 default_chain LLMChain( llmllm, promptPromptTemplate.from_template(请直接回答以下问题\n\n{input}) ) # 4. 构建路由提示词模板 router_template 根据用户的问题将其分类到最合适的处理类别。 可选的类别有 {destinations} 请只输出类别名称不要输出任何其他文字。 用户问题 {input} router_prompt PromptTemplate.from_template( router_template, partial_variables{destinations: \n.join([f{d[name]}: {d[description]} for d in destinations])} ) # 5. 创建路由链 router_chain LLMRouterChain.from_llm( llmllm, promptrouter_prompt, output_parserRouterOutputParser() ) # 6. 组装成MultiPromptChain multi_chain MultiPromptChain( router_chainrouter_chain, destination_chainsdestination_chains, default_chaindefault_chain, verboseTrue ) # 测试 questions [ 我的API调用返回了502错误怎么办, 我这个月的账单金额不对能查一下吗, 你们产品的主要功能有哪些, 今天天气怎么样 # 这个问题不属于任何预设类别应走默认链 ] for q in questions: print(f\n问题{q}) response multi_chain.run(q) print(f回答{response}) print(-*40)当运行上述代码时verboseTrue会显示路由链是如何工作的它首先调用LLM根据问题内容选择technical、billing、general中的一个然后将问题传递给对应的子链处理。对于“今天天气怎么样”由于不匹配任何预设类别描述路由器可能无法做出明确选择从而交给default_chain处理。实战技巧路由描述的准确性destinations列表中的description字段至关重要。它直接决定了路由LLM的判断依据需要清晰、无歧义地定义每个类别的边界。默认链的必要性一定要设置一个default_chain作为兜底处理未能路由或路由错误的情况保证系统的鲁棒性。输出解析RouterOutputParser负责将路由LLM的输出如字符串technical解析为路由链能理解的目标名称。5.2 创建自定义链应对复杂逻辑如果LangChain内置的链类型都无法满足你的需求你可以通过继承Chain基类来创建自定义链。这是最灵活的方式。假设我们需要一个链它先调用模型生成一段文本然后自动调用另一个翻译API非LLM将其翻译成目标语言。from langchain.chains.base import Chain from typing import Dict, List, Any, Optional from pydantic import Field class TranslationChain(Chain): 一个自定义链生成内容并翻译。 # 定义链的组件 generator_chain: LLMChain # 用于生成内容的子链 target_language: str Field(default英文) # 翻译目标语言 # 假设我们有一个简单的翻译函数这里用伪代码模拟 # 真实场景可能是调用Google Translate API, DeepL API等 translation_function: Any Field(defaultNone) # 定义输入键 property def input_keys(self) - List[str]: return self.generator_chain.input_keys # 定义输出键 property def output_keys(self) - List[str]: return [original_text, translated_text] def _call(self, inputs: Dict[str, Any]) - Dict[str, Any]: 链的核心执行逻辑 # 1. 调用生成链 gen_result self.generator_chain.invoke(inputs) original_text gen_result[text] # 2. 调用翻译函数 # 这里简化处理真实情况需要处理API调用、错误等 if self.translation_function: translated_text self.translation_function(original_text, self.target_language) else: # 如果没有提供翻译函数就用LLM模拟翻译仅作演示 translate_prompt f将以下内容翻译成{self.target_language}\n\n{original_text} translated_response self.generator_chain.llm.invoke(translate_prompt) translated_text translated_response.content # 3. 返回结果 return { original_text: original_text, translated_text: translated_text } property def _chain_type(self) - str: return translation_chain # 使用自定义链 generator_prompt PromptTemplate.from_template(写一首关于{theme}的五言绝句。) generator_llm_chain LLMChain(llmllm, promptgenerator_prompt) # 模拟一个翻译函数 def mock_translator(text, target_lang): return f[模拟翻译到{target_lang}]{text} my_custom_chain TranslationChain( generator_chaingenerator_llm_chain, target_language法语, translation_functionmock_translator ) result my_custom_chain.invoke({theme: 秋天}) print(f原始文本{result[original_text]}) print(f翻译文本{result[translated_text]})创建自定义链的关键在于继承Chain类。定义组件使用Pydantic的Field声明链内部需要的属性其他链实例、配置参数等。实现input_keys和output_keys属性明确告诉框架这个链需要什么输入会产生什么输出。实现_call方法这里是业务逻辑的核心。调用其他链或工具处理数据并返回一个包含output_keys中所有键的字典。实现_chain_type属性可选用于标识。自定义链让你能够将任意复杂的业务逻辑封装成一个标准的、可组合的LangChain单元这是构建高度定制化AI应用的终极武器。6. 链的调试、优化与生产化部署构建链只是第一步让链在生产环境中稳定、高效、可观测地运行才是更大的挑战。6.1 调试verbose、回调与中间结果verboseTrue如前所述这是最直接的调试工具能打印出链的每一步执行和输入输出。在开发阶段务必开启。回调函数CallbacksLangChain提供了强大的回调系统允许你在链执行的各个生命周期开始、结束、出错等注入自定义逻辑用于日志记录、监控、流式传输等。from langchain.callbacks.base import BaseCallbackHandler class MyCustomCallbackHandler(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f[链开始] 输入{inputs}) def on_chain_end(self, outputs, **kwargs): print(f[链结束] 输出{outputs}) def on_llm_start(self, serialized, prompts, **kwargs): print(f[LLM调用开始] 提示词{prompts[:50]}...) # 只打印前50字符 def on_llm_end(self, response, **kwargs): print(f[LLM调用结束] 响应长度{len(response.generations[0][0].text)}字符) # 在调用链时传入回调 result book_review_chain.invoke( {book_title: 小王子, author: 安托万·德·圣-埃克苏佩里}, config{callbacks: [MyCustomCallbackHandler()]} # 新版LangChain通过config传递 )获取中间结果对于SequentialChain如果你设置了output_variables包含中间链的output_key那么最终结果字典里就能拿到所有中间值这对于调试和分析数据流非常有用。6.2 优化缓存、重试与限流缓存相同的提示词和参数多次调用模型是巨大的浪费。LangChain支持内存缓存InMemoryCache和更持久的SQLite、Redis缓存。from langchain.globals import set_llm_cache from langchain.cache import InMemoryCache, SQLiteCache import sqlite3 # 使用内存缓存简单但进程重启后失效 set_llm_cache(InMemoryCache()) # 使用SQLite缓存持久化 set_llm_cache(SQLiteCache(database_path.langchain.db)) # 之后的所有LLM调用如果输入相同将直接返回缓存结果极大节省成本和时间。重试与回退网络和API服务是不稳定的。RunnableWithFallbacks可以为链配置备用方案。from langchain.chains import LLMChain from langchain_openai import ChatOpenAI, AzureChatOpenAI from langchain.schema.runnable import RunnableWithFallbacks primary_llm ChatOpenAI(modelgpt-4, temperature0) # 主模型 fallback_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 备用模型 # 或者使用不同供应商的模型作为回退如Azure OpenAI # fallback_llm AzureChatOpenAI(...) chain_with_fallback RunnableWithFallbacks( runnableprimary_llm, # 主执行体也可以是复杂的链 fallbacks[fallback_llm], exceptions_to_handle(Exception,) # 捕获哪些异常时触发回退 ) # 使用这个带回退的runnable来创建链 robust_chain LLMChain(llmchain_with_fallback, promptprompt_template)超时与限流在生产中必须为外部API调用设置超时并考虑实施限流以避免触发服务的速率限制。这通常在初始化LLM时配置或者使用像tenacity这样的重试库进行更精细的控制。6.3 生产化部署序列化与LangServe序列化链你可以将调试好的链保存到磁盘以便在其他环境加载避免重复的初始化代码。# 保存链注意并非所有组件都可序列化如自定义函数 import json chain_dict book_review_chain.dict() # 导出为字典 with open(book_review_chain.json, w) as f: json.dump(chain_dict, f, indent2) # 加载链 from langchain.chains.loading import load_chain loaded_chain load_chain(book_review_chain.json)使用LangServe部署为API这是将LangChain应用产品化的官方推荐方式。LangServe可以快速将你的链包装成REST API。# 首先安装 langserve pip install langserve[all]创建一个app.py文件from fastapi import FastAPI from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langserve import add_routes # 1. 创建链示例 prompt ChatPromptTemplate.from_template(请用{tone}的语气回答{question}) model ChatOpenAI() chain prompt | model # 这是LangChain新版的管道语法非常简洁 # 2. 创建FastAPI应用 app FastAPI( title我的AI助手API, version1.0, description一个简单的LangChain服务 ) # 3. 为链添加路由 # 这会自动生成 /invoke, /stream, /batch 等端点 add_routes( app, chain, path/assistant, ) if __name__ __main__: import uvicorn uvicorn.run(app, hostlocalhost, port8000)运行python app.py你就拥有了一个完整的API服务。访问http://localhost:8000/docs可以看到自动生成的交互式API文档。这正是“fastapi llm基础知识 langchain”组合的威力体现。通过本章的调试、优化和部署技巧你的LangChain应用将从一个实验脚本蜕变为一个真正可服务于用户的、健壮的生产级系统。这十五天的旅程从零开始到如今能够设计、构建、调试并部署一个完整的AI应用链你已经掌握了AI应用开发的核心脉络。记住工具LangChain是手段解决实际问题才是目的。不断用这些技能去探索和创造吧。