1. 项目概述从“单点工具”到“组合式工作流”的范式跃迁如果你在构建基于大语言模型的应用时还停留在“用户提问 - 调用模型 - 返回答案”的简单循环那么你很可能正在浪费模型90%的潜力并且亲手制造着无数难以维护的“一次性脚本”。我见过太多项目初期快速验证想法时代码写得飞起一个函数里塞满了提示词工程、模型调用、结果解析和后处理逻辑。两周后需求微调或者想增加一个简单的数据验证步骤就得在几百行混杂的代码里小心翼翼地“拆弹”稍有不慎就引发连锁崩溃。这正是“链”这个概念要解决的核心痛点将复杂、多步骤的智能任务拆解为标准化、可复用、可编排的模块化单元。“链”远不止是多个步骤的顺序执行。它本质上是一种设计模式和工作流引擎是对LLM应用开发范式的重新定义。想象一下乐高积木单个积木一个LLM调用或一个工具能力有限但通过标准的接口凸点将它们按设计图组合起来就能构建出从简单车辆到复杂城堡的任意结构。链就是那张设计图和那些标准接口。它让开发者从“写胶水代码”的泥潭中解放出来转而关注更高层次的业务逻辑编排。无论是处理一份PDF文档读取-总结-问答还是构建一个客服机器人意图识别-查询知识库-生成回复-情感安抚都可以通过链来清晰、健壮地实现。本章我们将深入“链”的肌理不仅学会如何使用现成的链更要掌握设计和构建自定义链的方法论让你能搭建出适应复杂业务场景的、可维护的AI工作流。2. 链的核心价值与设计哲学为什么顺序执行远远不够在深入代码之前我们必须先统一思想为什么要大费周章地引入“链”这一层抽象直接顺序调用几个函数不也一样能跑通吗这里面的区别就像用记事本写小说和用专业排版软件出版书籍一样巨大。2.1 超越“脚本”可维护性与可观测性临时脚本的核心问题是“一次性”。代码逻辑、提示词、API密钥、处理逻辑全部耦合在一起。当你想复用其中的“总结”步骤到另一个项目时只能复制粘贴然后面临调试的噩梦。而链将每个步骤封装为独立的“可运行体”Runnable每个Runnable有明确的输入/输出规范。这意味着独立测试你可以对链中的“总结链”进行单元测试用固定的输入验证其输出是否符合预期而不必运行整个流程。轻松替换如果觉得某个步骤的模型比如从GPT-4换成Claude或提示词效果不好你只需要更换那个Runnable其他部分完全不受影响。清晰观测在调试时你能清楚地看到数据流经每一个环节时的状态原始输入是什么经过A步骤后变成了什么B步骤的输入又接收到了什么。这对于排查“为什么最终答案错了”至关重要你可以精准定位是哪个环节的理解出现了偏差。2.2 动态路由与条件逻辑让工作流“活”起来简单的顺序链只是基础。真实世界的任务充满分支。例如一个用户输入“帮我订一张明天去北京的机票”你的链需要先进行“意图识别”如果识别为“订机票”则触发“查询航班”子链如果用户说“取消我昨天的订单”则应该路由到“订单管理”链。这种基于中间结果动态决定后续路径的能力是链的高级形态。它让AI应用不再是呆板的流水线而具备了初步的决策和上下文感知能力。在设计链时思考“这里是否需要分支”、“分支的判断条件是什么”是提升应用智能度的关键。2.3 组合复用积木化构建复杂系统这是链最强大的特性。你可以将一些验证有效的链作为“基础积木”。比如一个“提取实体链”从文本中提取人名、地点、时间一个“信息验证链”核对提取的信息是否合理一个“格式化输出链”将信息组织成JSON或表格。当你需要构建一个“从新闻中提取事件并生成简报”的复杂应用时你不需要从头开始而是将这些已有的链像搭积木一样组合起来可能只需要再写一个“生成摘要链”作为粘合剂。这种开发模式极大地提升了效率并保证了核心组件质量的一致性。注意不要陷入“为链而链”的陷阱。如果一个任务确实只有一步简单的LLM调用那么直接调用模型完全没问题。链的引入是为了管理复杂性。当你的代码里开始出现“如果上一步结果A则做X否则做Y然后还要把结果合并起来传给Z”这类逻辑时就是考虑使用链的最佳时机。3. 链的基石深入理解Runnable协议在绝大多数LLM应用框架中链的构建都围绕一个核心抽象Runnable可运行体。理解它就理解了链的DNA。3.1 什么是Runnable统一的接口约定Runnable是一个协议或接口它定义了一个对象必须实现的方法最常见的是invoke或run这个方法接收一个输入字典返回一个输出字典。为什么需要这个约定它是为了实现多态性。一个LLM模型是Runnable一个提示词模板是Runnable一个输出解析器是Runnable一个你写的自定义函数也可以包装成Runnable。当所有东西都是Runnable时它们就可以用统一的方式被连接、组合和调用。# 概念性代码展示Runnable的思想 class Runnable: def invoke(self, input: Dict) - Dict: 核心方法执行任务并返回结果 pass # LLM模型是一个Runnable class LLM(Runnable): def invoke(self, input): prompt input[prompt] response call_model_api(prompt) return {text: response} # 一个简单的函数也可以包装成Runnable class Tool(Runnable): def invoke(self, input): query input[query] result search_database(query) return {data: result}3.2 核心方法invoke,batch,stream一个完整的Runnable通常提供以下关键方法invoke(input) 同步调用处理单个输入。这是最常用的方法。batch(inputs) 批量处理输入列表。这对于提高吞吐量、处理数据集非常有用。框架底层可能会并行调用以提升效率。stream(input) 流式处理输入返回一个生成器可以逐词或逐块产生输出。这对于构建实时响应的聊天体验至关重要。3.3 连接符|与RunnableSequence链的优雅之处在于其声明式的组合方式。最常见的是使用管道符|它直观地表示了数据流的方向。# 假设我们已有一些Runnable实例 prompt_template ChatPromptTemplate.from_template(翻译以下中文到英文{text}) model ChatOpenAI(modelgpt-4) output_parser StrOutputParser() # 使用管道符 | 将它们组合成一个链 translation_chain prompt_template | model | output_parser # 调用链 result translation_chain.invoke({text: 今天天气真好}) print(result) # 输出The weather is really nice today.这段代码的可读性极高数据先流入prompt_template被格式化成提示词然后流入model生成回复最后流入output_parser提取出字符串。|操作符背后创建的是一个RunnableSequence对象它负责管理执行顺序和中间数据的传递。实操心得 当你调试链时如果最终输出不对可以尝试分步调用。例如先单独调用prompt_template.invoke(...)看看生成的提示词是否正确再手动将这个提示词喂给模型。这种“分而治之”的调试方法能快速定位问题环节。4. 构建各类链的实战详解了解了核心概念后我们进入实战环节。链有多种形态应对不同场景。4.1 顺序链 (RunnableSequence)这是最简单、最常用的链。如上例所示它将多个Runnable按顺序连接。关键在于前一个Runnable的输出结构必须与后一个Runnable的输入结构匹配。框架通常会自动处理将上一个的输出字典整体作为下一个的输入。如果遇到不匹配你可能需要使用RunnablePassthrough来保留特定字段或使用RunnableLambda一个将自定义函数包装成Runnable的工具来转换数据格式。场景示例文档问答链from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 假设我们已经有一个“检索器”能从知识库中找到相关文档 retriever get_retriever() # 1. 定义提示词模板 template 基于以下上下文回答用户问题。如果你不知道答案就说你不知道不要编造。 上下文 {context} 问题 {question} 请用中文给出答案 prompt ChatPromptTemplate.from_template(template) # 2. 定义模型和解析器 model ChatOpenAI(temperature0) parser StrOutputParser() # 3. 构建链 # 关键使用RunnablePassthrough将用户问题原封不动地传递下去同时用检索器获取上下文 qa_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | model | parser ) # 调用问题会先触发retriever结果与问题一起组成字典传给prompt answer qa_chain.invoke(你们公司的退货政策是什么)在这个链中输入是一个字符串问题。{context: retriever, question: RunnablePassthrough()}这一部分是一个特殊的映射结构它会并行执行retriever接收输入即问题并返回相关文档作为context而RunnablePassthrough()将原始输入作为question传递。两者结果合并成一个{context: ..., question: ...}的字典传递给后面的prompt。4.2 条件链与路由 (RunnableBranch)当你的工作流需要根据中间结果做出决策时就需要条件链。这通常通过RunnableBranch实现它接收一系列条件 Runnable对以及一个默认Runnable。from langchain_core.runnables import RunnableBranch # 定义几个处理不同意图的子链 greeting_chain ChatPromptTemplate.from_template(友好地回复问候{input}) | model | parser qa_chain ChatPromptTemplate.from_template(回答问题{input}) | model | parser default_chain ChatPromptTemplate.from_template(你说了{input}但我不知道怎么处理。) | model | parser # 定义条件判断函数也需要包装成Runnable def classify_intent(input_dict): user_input input_dict[input].lower() if any(word in user_input for word in [你好, 嗨, 早上好]): return greeting elif ? in user_input: return qa else: return default # 构建分支链 branch RunnableBranch( (lambda x: classify_intent(x) greeting, greeting_chain), (lambda x: classify_intent(x) qa, qa_chain), default_chain ) # 使用 result branch.invoke({input: 你好吗}) # 会走 greeting_chain result2 branch.invoke({input: 地球是圆的吗}) # 会走 qa_chain这里classify_intent函数是一个简单的基于规则的路由器。在实际生产中你可能会用一个更小的、专门训练的分类模型或另一个LLM调用来做更精准的意图识别。4.3 转换链 (RunnableLambda与自定义函数)不是所有步骤都需要LLM。数据清洗、格式转换、调用外部API等都需要自定义函数。RunnableLambda让你能将任何Python函数轻松集成到链中。from langchain_core.runnables import RunnableLambda import json # 自定义函数将模型输出的JSON字符串解析为字典并提取特定字段 def extract_price(response_dict): text response_dict[text] # 假设上游模型输出在text字段 try: data json.loads(text) # 假设模型返回 {product: xxx, price: 100} return {extracted_price: data.get(price), raw_response: text} except json.JSONDecodeError: return {extracted_price: None, error: Invalid JSON, raw_response: text} # 包装成Runnable extractor RunnableLambda(extract_price) # 组合进链 price_extraction_chain ( prompt_template | model | extractor # 这里接入自定义处理逻辑 ) # 现在链的最终输出是一个包含提取价格和原始响应的字典注意事项在RunnableLambda中的函数应尽量保持纯净无副作用和幂等输入输出最好是字典。如果函数可能抛出异常要考虑如何在链层面进行错误处理。4.4 并行与合并链 (RunnableParallel)有时你需要同时执行多个任务然后将结果合并。例如给定一个产品描述你需要同时让模型生成营销文案、提取关键特性和评估目标受众。from langchain_core.runnables import RunnableParallel # 定义三个并行的子链 generate_ad_chain ChatPromptTemplate.from_template(为以下产品写广告语{description}) | model | parser extract_features_chain ChatPromptTemplate.from_template(提取产品关键特性{description}) | model | parser analyze_audience_chain ChatPromptTemplate.from_template(分析产品目标受众{description}) | model | parser # 使用RunnableParallel组合 parallel_chain RunnableParallel( adgenerate_ad_chain, featuresextract_features_chain, audienceanalyze_audience_chain ) # 调用输入一个描述同时触发三个链 results parallel_chain.invoke({description: 一款续航达20小时的无线降噪耳机}) print(results) # 输出类似{ad: ..., features: [..., ...], audience: ...}RunnableParallel会并发地执行各个子链如果后端支持显著提升处理速度。它的输出是一个字典键是你在定义时指定的名字如ad,features值是对应子链的输出。5. 高级模式与最佳实践当链变得复杂时良好的设计和实践能避免项目失控。5.1 链的嵌套构建分层工作流复杂的业务逻辑可以通过链的嵌套来实现。一个“主链”的某个步骤本身可能就是一个复杂的“子链”。这符合软件工程的“分治”思想。# 子链处理用户反馈的情感分析和要点提取 feedback_processing_chain ( RunnableParallel( sentimentChatPromptTemplate.from_template(判断情感{feedback}) | model | parser, key_pointsChatPromptTemplate.from_template(总结要点{feedback}) | model | parser ) ) # 主链接收反馈处理然后根据情感生成不同回复 main_chain ( {feedback: RunnablePassthrough()} | feedback_processing_chain # 这里嵌套了并行链 | RunnableLambda(lambda x: { sentiment: x[sentiment], key_points: x[key_points], reply_prompt: f用户反馈情感为{x[sentiment]}要点是{x[key_points]}。请生成一份客服回复。 }) | ChatPromptTemplate.from_template({reply_prompt}) | model | parser )这种嵌套结构让代码模块清晰每个子链可以独立开发、测试和复用。5.2 状态管理与上下文传递在对话或多轮交互场景中链需要访问历史消息上下文。这通常通过将“消息历史”作为一个输入变量来实现。你需要设计链的输入结构使其包含input当前用户输入和chat_history历史消息列表。from langchain_core.messages import HumanMessage, AIMessage # 提示词模板设计为接收历史和当前输入 conversational_prompt ChatPromptTemplate.from_messages([ (system, 你是一个友好的助手。), MessagesPlaceholder(variable_namechat_history), # 历史消息占位符 (human, {input}) ]) conversation_chain conversational_prompt | model | parser # 模拟历史 chat_history [ HumanMessage(content你好), AIMessage(content你好有什么可以帮你的) ] # 调用时传入历史和当前输入 response conversation_chain.invoke({ chat_history: chat_history, input: 我上一句说了什么 })关键在于使用MessagesPlaceholder来动态插入历史消息。管理chat_history列表增删改查通常是应用层框架如LangGraph或你自己代码的责任。5.3 错误处理与重试机制网络请求、模型API调用都可能失败。一个健壮的链必须具备错误处理能力。重试对于暂时的网络错误可以使用带有退避策略的自动重试。许多LLM客户端库内置了重试功能。Fallback当主模型如GPT-4调用失败或返回不合理结果时可以降级到备用模型如GPT-3.5-Turbo或规则系统。验证与过滤在链的末端或关键节点加入输出验证。例如用一个小型验证链检查答案是否包含不适当内容或者格式是否正确。如果验证失败可以触发重生成或返回安全默认值。from tenacity import retry, stop_after_attempt, wait_exponential # 为模型调用添加重试装饰器示例 retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def reliable_model_invoke(prompt): return model.invoke(prompt) # 将包装后的函数集成到链中5.4 性能优化缓存与异步缓存对于确定性高的步骤如相同的提示词和参数下模型输出应相同可以引入缓存避免重复计算和API调用节省成本和时间。可以将缓存设置在LLM调用层或整个链的层面。异步对于I/O密集型的操作网络请求使用异步调用ainvoke,abatch,astream可以大幅提升并发性能特别是在处理批量请求或构建流式响应服务时。确保你的整个调用栈框架、HTTP客户端都支持异步。6. 调试、监控与可视化链的复杂性使得调试变得更具挑战性但也更有方法可循。6.1 链路追踪与日志最有效的调试方式是查看链中每一步的输入和输出。许多框架提供了回调Callbacks机制允许你在链执行的生命周期中注入钩子函数记录日志、发送到监控系统等。# 伪代码展示回调概念 from langchain_core.callbacks import BaseCallbackHandler class DebugCallbackHandler(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f[Chain Start] {serialized.get(name)} - Input: {inputs}) def on_chain_end(self, outputs, **kwargs): print(f[Chain End] Outputs: {outputs}) def on_llm_start(self, serialized, prompts, **kwargs): print(f[LLM Start] Prompt sent: {prompts[0][:100]}...) # 打印前100字符 # 在调用链时传入回调 result chain.invoke({input: test}, config{callbacks: [DebugCallbackHandler()]})通过回调你可以清晰地看到数据是如何流经每个组件的这对于排查“为什么提示词没生效”、“为什么解析器报错”等问题至关重要。6.2 可视化工具对于非常复杂的链尤其是包含分支和并行逻辑的图形化可视化能极大提升理解效率。一些框架或第三方工具支持将链结构自动生成流程图帮助你进行架构审查和向团队成员解释设计。6.3 评估与测试如何确保你构建的链是可靠的你需要一套评估体系。单元测试为每个独立的Runnable特别是自定义函数和简单的子链编写单元测试用固定输入断言输出。集成测试用一组有代表性的输入测试整个链检查最终输出是否符合业务预期。基于LLM的评估对于开放性任务可以设计另一个LLM作为“裁判”根据一些标准相关性、正确性、无害性来评估主链的输出质量。虽然成本较高但对于关键应用是必要的。构建链的过程是一个不断迭代和优化的过程。从最简单的顺序链开始逐步引入分支、并行、自定义逻辑并辅以完善的测试和监控你就能搭建出既强大又稳健的智能应用工作流。记住好的链设计就像好的代码架构追求的是高内聚、低耦合、可测试和可维护。当你习惯了这种思维模式开发复杂的AI应用将不再是一件令人畏惧的事情。