1. 从“零散调用”到“组装流水线”为什么我们需要LangChain的链如果你刚开始接触LangChain可能会觉得它概念繁多有点无从下手。Agent、Tool、Memory、Chain……这些名词堆在一起很容易让人迷失。但如果你已经尝试过用LangChain的LLM类调用过几次大模型写过几个简单的提示词你可能会产生一个疑问我直接调用API不就行了吗为什么还要费劲去学这个框架这个问题的答案就藏在“链”Chain这个概念里。想象一下你正在搭建一个智能客服系统。用户问“我上周买的那个蓝色的保温杯现在能退货吗” 要回答这个问题一个简单的LLM.invoke()调用是远远不够的。你需要从用户问题中提取关键实体产品保温杯属性蓝色时间上周。根据这些信息去查询订单数据库找到对应的订单记录。查询该订单的退货政策可能涉及商品类别、购买时长、商品状态等。将订单信息和退货政策组合成一个完整的上下文。最后让大模型基于这个上下文生成友好、准确的回复。这个过程涉及至少5个步骤每个步骤都可能失败并且步骤之间有严格的依赖关系。如果你用最原始的代码来写可能就是一连串的函数调用中间穿插着大量的错误处理和结果传递逻辑。代码会迅速变得冗长、脆弱且难以维护。而LangChain的“链”就是为了解决这种“多步骤、有状态、可复用”的任务流程而生的。它不是一个魔法黑盒而是一种声明式的编程范式。你把每个步骤定义成一个独立的“环节”如LLM调用、工具调用、数据转换然后用“链”把它们像乐高积木一样组装起来。链负责管理这些环节的执行顺序、数据流转和错误处理。LLMChain是最基础、最核心的一种链它完美诠释了“提示词模板 大模型调用 输出解析”这一标准化流程。理解了它你就拿到了打开LangChain自动化世界大门的钥匙。2. LLMChain深度拆解不止是prompt llm很多人对LLMChain的理解停留在表面认为它就是LLMChain(promptprompt_template, llmllm)然后chain.invoke({topic: AI})。这没错但这只是冰山一角。LLMChain的精髓在于它标准化并封装了一个完整LLM交互周期的三个核心阶段并提供了极大的灵活性。2.1 核心三要素Prompt、LLM与Parser的协同一个功能完整的LLMChain其核心通常包含三个部分而不仅仅是两个PromptTemplate提示词模板这是任务的蓝图。它定义了输入的变量如{topic}和给模型的指令。好的模板是成功的一半。LangChain的模板支持多种格式f-string, jinja2并可以包含少样本示例FewShotPromptTemplate。from langchain.prompts import PromptTemplate # 一个简单的模板 prompt PromptTemplate.from_template( “请用中文为{product}写一段不超过50字的广告文案。要求突出其{feature}特性。” )LLM大语言模型这是执行引擎。LangChain的LLM或ChatModel类是对不同供应商APIOpenAI, Anthropic, 本地模型等的抽象。通过llm ChatOpenAI(model“gpt-4”, temperature0.7)这样的初始化你可以灵活切换模型而无需重写业务逻辑。注意temperature参数控制创造性对于广告文案可以设为0.7-0.9对于事实性问答则应设为0-0.2。OutputParser输出解析器这是最容易被忽略但至关重要的环节。大模型的输出是自由文本而我们的程序需要结构化的数据。OutputParser负责将文本输出转换为Python对象。from langchain.output_parsers import CommaSeparatedListOutputParser, StructuredOutputParser, ResponseSchema # 解析为列表 parser CommaSeparatedListOutputParser() # 或者解析为复杂的结构强烈推荐 response_schemas [ ResponseSchema(name“ad_copy”, description“生成的广告文案”), ResponseSchema(name“keywords”, type“array”, description“文案中的核心关键词列表”), ResponseSchema(name“tone”, description“文案的语气如激昂、温馨、专业”, enum[“激昂”, “温馨”, “专业”]) ] parser StructuredOutputParser.from_response_schemas(response_schemas) # 将格式指令自动加入到提示词中 format_instructions parser.get_format_instructions() prompt PromptTemplate( template“...你的任务...\\n{format_instructions}\\n用户输入{input}”, input_variables[“input”], partial_variables{“format_instructions”: format_instructions} )当链执行后你可以直接result parser.parse(output)得到一个字典里面包含ad_copy、keywords等字段可以直接用于后续逻辑。2.2 链的“输入/输出”模式invoke, stream, batchLLMChain提供了多种调用方式适配不同场景invoke(input_dict)最常用的同步调用。输入一个字典键对应提示词模板中的变量返回一个字典默认包含‘text’键如果配置了parser则包含解析后的结果。result chain.invoke({“product”: “智能手表”, “feature”: “超长续航”}) print(result[“ad_copy”]) # 直接访问结构化字段stream(input_dict)流式调用。当你需要实时显示模型生成的内容如聊天应用时使用。它返回一个生成器逐词或逐块产出结果。for chunk in chain.stream({“product”: “智能手表”}): print(chunk, end“”, flushTrue) # 模拟打字机效果batch(input_list)批量调用。一次性处理多个输入通常比循环调用invoke更高效因为可能涉及底层的批处理优化。inputs [{“product”: “手表”, “feature”: “续航”}, {“product”: “耳机”, “feature”: “降噪”}] results chain.batch(inputs)2.3 实战中的高级配置与陷阱在实际项目中直接使用基础的LLMChain可能会遇到问题需要一些高级配置超时与重试网络不稳定或模型服务偶发性错误是常态。LangChain内置了重试逻辑。from langchain.callbacks.manager import AsyncCallbackManager from langchain.llms import OpenAI from tenacity import retry, stop_after_attempt, wait_exponential # 更推荐使用LangChain的retry装饰器或配置 llm ChatOpenAI( ..., max_retries2, # 自动重试次数 request_timeout30 # 单次请求超时 )回调与日志为了调试和监控你需要知道链内部发生了什么。回调系统Callbacks是答案。from langchain.callbacks import StdOutCallbackHandler handler StdOutCallbackHandler() # 会将详细日志打印到标准输出 chain LLMChain(llmllm, promptprompt, callbacks[handler]) # 调用时你会看到类似“ Entering LLMChain...”的详细执行日志。这对于理解提示词的实际渲染内容、模型的原始响应耗时至关重要。内存Memory的集成LLMChain本身是无状态的。但很多场景如聊天需要记忆历史。虽然LLMChain可以搭配ConversationBufferMemory等使用但这通常是通过ConversationChain它是LLMChain的子类或特定组合来实现的。理解这一点可以避免混淆。一个常见的陷阱直接拼接字符串而非使用PromptTemplate。比如prompt_text f“写一个关于{topic}的故事”然后直接传给LLM。这放弃了模板的管理、变量验证和后期复用性。始终使用PromptTemplate。3. 超越LLMChainSequentialChain与RouterChain当你掌握了单个LLMChain自然会想如何把多个链串联起来完成更复杂的任务这就是SequentialChain的用武之地。而当你需要根据输入内容动态选择执行哪条链时RouterChain就登场了。3.1 SequentialChain构建确定性的工作流SequentialChain用于按预定顺序执行一系列链。它有两种主要模式SimpleSequentialChain最简单的形式每个链只有一个输入和一个输出上一个链的输出直接作为下一个链的输入。适合线性管道。from langchain.chains import SimpleSequentialChain # 假设 chain1 写大纲chain2 根据大纲写文章 overall_chain SimpleSequentialChain(chains[chain1, chain2], verboseTrue) result overall_chain.invoke(“人工智能的未来”) # chain1的输入是“人工智能的未来”输出大纲传给chain2chain2输出最终文章。SequentialChain功能强大得多也是实际项目中最常用的。它允许每个链有多个输入和多个输出并可以精确指定整个工作流的输入变量和输出变量。from langchain.chains import SequentialChain # 定义三个子链 critique_chain LLMChain(llmllm, promptcritique_prompt, output_key“critique”) revision_chain LLMChain(llmllm, promptrevision_prompt, output_key“revised_text”) scoring_chain LLMChain(llmllm, promptscoring_prompt, output_key“score”) # 组装成顺序链 overall_chain SequentialChain( chains[critique_chain, revision_chain, scoring_chain], input_variables[“original_text”], # 整个工作流的初始输入 output_variables[“revised_text”, “score”], # 我们最终要的输出 verboseTrue ) result overall_chain.invoke({“original_text”: “这是一段待改进的文本...”}) print(f“修订后文本{result[‘revised_text’]}\\n评分{result[‘score’]}”)这里critique_chain接收original_text产出critique。revision_chain需要两个输入original_text和critique产出revised_text。SequentialChain会自动处理这种依赖关系只要你在子链的prompt模板中声明了对应的变量名。关键点SequentialChain通过output_key和变量名映射来传递数据。务必确保子链的输入变量名与上游链的输出output_key或其他输入变量名一致。3.2 RouterChain实现动态任务分发有些时候流程不是确定的。例如一个用户输入可能是“写首诗”、“翻译这句话”或“总结这篇文章”。你需要先判断意图再路由到不同的处理链。这就是RouterChain。RouterChain通常与MultiPromptChain结合使用。其核心思想是一个LLMChain路由链先分析输入决定将其分发给哪个“专家链”。每个“专家链”对应一个特定的提示词模板和处理逻辑。一个默认链处理无法路由的情况。from langchain.chains.router import MultiPromptChain from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser from langchain.prompts import PromptTemplate # 1. 定义不同“专家”的提示词模板 poet_template “你是一位诗人。请根据以下主题创作一首诗{input}” translator_template “你是一位翻译家。请将以下中文翻译成英文{input}” summarizer_template “你是一位总结助手。请用一句话总结以下文本{input}” prompt_infos [ { “name”: “poetry”, “description”: “适合创作诗歌的请求”, “prompt_template”: poet_template }, { “name”: “translation”, “description”: “适合进行中英翻译的请求”, “prompt_template”: translator_template }, { “name”: “summarization”, “description”: “适合总结文本的请求”, “prompt_template”: summarizer_template } ] # 2. 创建目标链专家链 destination_chains {} for p_info in prompt_infos: prompt PromptTemplate(templatep_info[“prompt_template”], input_variables[“input”]) chain LLMChain(llmllm, promptprompt) destination_chains[p_info[“name”]] chain # 3. 创建默认链 default_prompt PromptTemplate(template“请直接回复以下内容{input}”, input_variables[“input”]) default_chain LLMChain(llmllm, promptdefault_prompt) # 4. 组装MultiPromptChain router_chain MultiPromptChain( router_chainLLMRouterChain.from_prompts(llm, prompt_infos, default_chaindefault_chain), destination_chainsdestination_chains, default_chaindefault_chain, verboseTrue ) # 5. 使用 result router_chain.invoke(“写一首关于春天的诗”) # 路由链会判断这是“poetry”请求将其交给诗人链处理。RouterChain非常适合构建插件化、可扩展的AI应用比如智能客服中的意图识别和转接。4. 与工具和记忆的集成TransformChain与自定义链LLMChain和SequentialChain处理的是LLM之间的协作。但真实世界任务需要LLM与外部工具API、数据库、计算器交互也需要处理非LLM的数据转换。这就需要TransformChain和自定义链。4.1 TransformChain处理纯数据转换TransformChain是一个特殊的链它不包含LLM只包含一个输入/输出处理函数。它用于在LLM链之间进行数据清洗、格式转换或调用纯函数逻辑。from langchain.chains import TransformChain import json def transform_func(inputs: dict) - dict: “””将字符串化的JSON转换为Python对象并提取特定字段。””” text inputs[“text”] data json.loads(text) # 假设我们只关心data字段下的items列表 return {“extracted_items”: data.get(“data”, {}).get(“items”, [])} # 创建转换链 transform_chain TransformChain( input_variables[“text”], output_variables[“extracted_items”], transformtransform_func ) # 可以将其插入SequentialChain中 # 假设 chain1 输出一个JSON字符串chain2 需要处理其中的items overall_chain SequentialChain( chains[chain1, transform_chain, chain2], input_variables[“query”], output_variables[“final_answer”] )4.2 构建自定义链实现工具调用Agent的基石最强大的模式是创建自定义链让LLM能够决定何时、如何调用外部工具。这其实就是LangChain中Agent的核心思想。虽然Agent已经封装得很好但理解其底层链的构造有助于深度定制。一个简化的“工具调用链”可能包含以下步骤思考链分析用户输入决定是否需要调用工具以及调用哪个工具。动作链如果需调用工具则生成工具调用所需的参数一个格式化的字符串或JSON。执行在链外执行工具函数、API调用。观察链将工具执行的结果反馈给LLM。最终回答链综合原始问题、思考和工具结果生成最终回答。你可以用SequentialChain来粗略模拟这个过程但LangChain的Agent和Tool框架提供了更优雅、更强大的抽象。理解这一点后你再去看initialize_agent函数就会明白它内部其实是在组装一个由LLMChain、Tool和特定Agent类如ReActAgent构成的复杂工作流。实操心得在项目早期如果任务流程非常固定且已知优先使用SequentialChain。它更稳定、可预测、调试方便。只有当任务需要LLM动态决策比如“上网搜索”“查数据库”“进行计算”三选一或组合时才考虑引入Agent。Agent虽然灵活但成本更高更多Token消耗、更不可控可能做出错误决策。5. 生产环境下的链调试、监控与最佳实践在开发环境跑通链只是第一步。将其部署到生产环境你需要关注可靠性、性能和可观测性。5.1 调试与日志如前所述verboseTrue是你的第一调试工具。但对于复杂链你可能需要更细粒度的控制。使用LangSmith这是LangChain官方提供的追踪和调试平台。它能可视化链的每一步执行查看输入输出分析延迟和Token消耗是调试复杂工作流的神器。只需设置环境变量LANGCHAIN_TRACING_V2true和LANGCHAIN_API_KEY所有调用会自动记录。自定义回调你可以编写自己的回调处理器将日志发送到你的监控系统如ELK、Datadog。from langchain.callbacks.base import BaseCallbackHandler class MyCustomHandler(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f“ 链 {serialized.get(‘name’)} 开始输入{inputs}”) def on_chain_end(self, outputs, **kwargs): print(f“ 链结束输出{outputs}”) chain LLMChain(..., callbacks[MyCustomHandler()])5.2 性能优化缓存相同的输入往往产生相同的输出。对LLM调用进行缓存可以大幅节省成本和时间。LangChain支持内存缓存(InMemoryCache)、数据库缓存(SQLiteCache)等。from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache()) # 首次调用会访问API result1 chain.invoke({“topic”: “AI”}) # 相同的输入第二次会直接从缓存读取 result2 chain.invoke({“topic”: “AI”})异步支持大多数链都支持异步调用(ainvoke,astream,abatch)。在Web服务器等异步环境中使用可以避免阻塞提高吞吐量。import asyncio async def process_requests(queries): tasks [chain.ainvoke({“input”: q}) for q in queries] results await asyncio.gather(*tasks) return results批处理与速率限制利用batch方法并配合模型的批处理能力。同时注意为API设置合理的速率限制避免被限流。5.3 错误处理与鲁棒性结构化输出的验证即使使用了StructuredOutputParser模型偶尔也可能返回不符合格式的文本。在解析后添加数据验证逻辑如使用Pydantic模型。from pydantic import BaseModel, ValidationError class AdOutput(BaseModel): ad_copy: str keywords: list[str] tone: str try: parsed_dict parser.parse(chain_output) validated_data AdOutput(**parsed_dict) except (ValidationError, Exception) as e: # 处理解析失败例如记录日志、使用默认值、重试等 validated_data AdOutput(ad_copy“生成失败”, keywords[], tone“专业”)降级策略当主要模型如GPT-4调用失败或超时时应有备选方案如切换到GPT-3.5-Turbo或返回一个预定义的友好错误消息。5.4 测试与版本控制对链进行单元测试像测试普通函数一样测试你的链。使用固定的输入断言预期的输出结构或关键内容。Mock LLM的响应使测试快速且不依赖外部API。提示词版本化提示词模板是代码的一部分。将重要的提示词模板存储在版本控制系统如Git中并考虑使用LangChain Hub来管理和共享提示词。配置管理将模型参数temperature, max_tokens、API密钥、链的配置等放在环境变量或配置文件中不要硬编码在代码里。从LLMChain这个简单的“积木”开始通过SequentialChain进行串联利用RouterChain实现分支最后通过自定义逻辑和TransformChain与外部世界连接——你就能搭建出任意复杂的AI智能体工作流。这个过程中核心思想始终是声明式组装和关注点分离。每个链只做好一件事通过清晰的接口组合在一起。当你掌握了这些预置链的用法和思想再去探索LangChain中更高级的Agent和Retrieval模块就会发现它们不过是这些基础模式在特定领域的精妙应用和封装。 1. 从“零散调用”到“组装流水线”为什么我们需要LangChain的链如果你刚开始接触LangChain可能会觉得它概念繁多有点无从下手。Agent、Tool、Memory、Chain……这些名词堆在一起很容易让人迷失。但如果你已经尝试过用LangChain的LLM类调用过几次大模型写过几个简单的提示词你可能会产生一个疑问我直接调用API不就行了吗为什么还要费劲去学这个框架这个问题的答案就藏在“链”Chain这个概念里。想象一下你正在搭建一个智能客服系统。用户问“我上周买的那个蓝色的保温杯现在能退货吗” 要回答这个问题一个简单的LLM.invoke()调用是远远不够的。你需要从用户问题中提取关键实体产品保温杯属性蓝色时间上周。根据这些信息去查询订单数据库找到对应的订单记录。查询该订单的退货政策可能涉及商品类别、购买时长、商品状态等。将订单信息和退货政策组合成一个完整的上下文。最后让大模型基于这个上下文生成友好、准确的回复。这个过程涉及至少5个步骤每个步骤都可能失败并且步骤之间有严格的依赖关系。如果你用最原始的代码来写可能就是一连串的函数调用中间穿插着大量的错误处理和结果传递逻辑。代码会迅速变得冗长、脆弱且难以维护。而LangChain的“链”就是为了解决这种“多步骤、有状态、可复用”的任务流程而生的。它不是一个魔法黑盒而是一种声明式的编程范式。你把每个步骤定义成一个独立的“环节”如LLM调用、工具调用、数据转换然后用“链”把它们像乐高积木一样组装起来。链负责管理这些环节的执行顺序、数据流转和错误处理。LLMChain是最基础、最核心的一种链它完美诠释了“提示词模板 大模型调用 输出解析”这一标准化流程。理解了它你就拿到了打开LangChain自动化世界大门的钥匙。2. LLMChain深度拆解不止是prompt llm很多人对LLMChain的理解停留在表面认为它就是LLMChain(promptprompt_template, llmllm)然后chain.invoke({topic: AI})。这没错但这只是冰山一角。LLMChain的精髓在于它标准化并封装了一个完整LLM交互周期的三个核心阶段并提供了极大的灵活性。2.1 核心三要素Prompt、LLM与Parser的协同一个功能完整的LLMChain其核心通常包含三个部分而不仅仅是两个PromptTemplate提示词模板这是任务的蓝图。它定义了输入的变量如{topic}和给模型的指令。好的模板是成功的一半。LangChain的模板支持多种格式f-string, jinja2并可以包含少样本示例FewShotPromptTemplate。from langchain.prompts import PromptTemplate # 一个简单的模板 prompt PromptTemplate.from_template( “请用中文为{product}写一段不超过50字的广告文案。要求突出其{feature}特性。” )LLM大语言模型这是执行引擎。LangChain的LLM或ChatModel类是对不同供应商APIOpenAI, Anthropic, 本地模型等的抽象。通过llm ChatOpenAI(model“gpt-4”, temperature0.7)这样的初始化你可以灵活切换模型而无需重写业务逻辑。注意temperature参数控制创造性对于广告文案可以设为0.7-0.9对于事实性问答则应设为0-0.2。OutputParser输出解析器这是最容易被忽略但至关重要的环节。大模型的输出是自由文本而我们的程序需要结构化的数据。OutputParser负责将文本输出转换为Python对象。from langchain.output_parsers import CommaSeparatedListOutputParser, StructuredOutputParser, ResponseSchema # 解析为列表 parser CommaSeparatedListOutputParser() # 或者解析为复杂的结构强烈推荐 response_schemas [ ResponseSchema(name“ad_copy”, description“生成的广告文案”), ResponseSchema(name“keywords”, type“array”, description“文案中的核心关键词列表”), ResponseSchema(name“tone”, description“文案的语气如激昂、温馨、专业”, enum[“激昂”, “温馨”, “专业”]) ] parser StructuredOutputParser.from_response_schemas(response_schemas) # 将格式指令自动加入到提示词中 format_instructions parser.get_format_instructions() prompt PromptTemplate( template“...你的任务...\\n{format_instructions}\\n用户输入{input}”, input_variables[“input”], partial_variables{“format_instructions”: format_instructions} )当链执行后你可以直接result parser.parse(output)得到一个字典里面包含ad_copy、keywords等字段可以直接用于后续逻辑。2.2 链的“输入/输出”模式invoke, stream, batchLLMChain提供了多种调用方式适配不同场景invoke(input_dict)最常用的同步调用。输入一个字典键对应提示词模板中的变量返回一个字典默认包含‘text’键如果配置了parser则包含解析后的结果。result chain.invoke({“product”: “智能手表”, “feature”: “超长续航”}) print(result[“ad_copy”]) # 直接访问结构化字段stream(input_dict)流式调用。当你需要实时显示模型生成的内容如聊天应用时使用。它返回一个生成器逐词或逐块产出结果。for chunk in chain.stream({“product”: “智能手表”}): print(chunk, end“”, flushTrue) # 模拟打字机效果batch(input_list)批量调用。一次性处理多个输入通常比循环调用invoke更高效因为可能涉及底层的批处理优化。inputs [{“product”: “手表”, “feature”: “续航”}, {“product”: “耳机”, “feature”: “降噪”}] results chain.batch(inputs)2.3 实战中的高级配置与陷阱在实际项目中直接使用基础的LLMChain可能会遇到问题需要一些高级配置超时与重试网络不稳定或模型服务偶发性错误是常态。LangChain内置了重试逻辑。from langchain.callbacks.manager import AsyncCallbackManager from langchain.llms import OpenAI from tenacity import retry, stop_after_attempt, wait_exponential # 更推荐使用LangChain的retry装饰器或配置 llm ChatOpenAI( ..., max_retries2, # 自动重试次数 request_timeout30 # 单次请求超时 )回调与日志为了调试和监控你需要知道链内部发生了什么。回调系统Callbacks是答案。from langchain.callbacks import StdOutCallbackHandler handler StdOutCallbackHandler() # 会将详细日志打印到标准输出 chain LLMChain(llmllm, promptprompt, callbacks[handler]) # 调用时你会看到类似“ Entering LLMChain...”的详细执行日志。这对于理解提示词的实际渲染内容、模型的原始响应耗时至关重要。内存Memory的集成LLMChain本身是无状态的。但很多场景如聊天需要记忆历史。虽然LLMChain可以搭配ConversationBufferMemory等使用但这通常是通过ConversationChain它是LLMChain的子类或特定组合来实现的。理解这一点可以避免混淆。一个常见的陷阱直接拼接字符串而非使用PromptTemplate。比如prompt_text f“写一个关于{topic}的故事”然后直接传给LLM。这放弃了模板的管理、变量验证和后期复用性。始终使用PromptTemplate。3. 超越LLMChainSequentialChain与RouterChain当你掌握了单个LLMChain自然会想如何把多个链串联起来完成更复杂的任务这就是SequentialChain的用武之地。而当你需要根据输入内容动态选择执行哪条链时RouterChain就登场了。3.1 SequentialChain构建确定性的工作流SequentialChain用于按预定顺序执行一系列链。它有两种主要模式SimpleSequentialChain最简单的形式每个链只有一个输入和一个输出上一个链的输出直接作为下一个链的输入。适合线性管道。from langchain.chains import SimpleSequentialChain # 假设 chain1 写大纲chain2 根据大纲写文章 overall_chain SimpleSequentialChain(chains[chain1, chain2], verboseTrue) result overall_chain.invoke(“人工智能的未来”) # chain1的输入是“人工智能的未来”输出大纲传给chain2chain2输出最终文章。SequentialChain功能强大得多也是实际项目中最常用的。它允许每个链有多个输入和多个输出并可以精确指定整个工作流的输入变量和输出变量。from langchain.chains import SequentialChain # 定义三个子链 critique_chain LLMChain(llmllm, promptcritique_prompt, output_key“critique”) revision_chain LLMChain(llmllm, promptrevision_prompt, output_key“revised_text”) scoring_chain LLMChain(llmllm, promptscoring_prompt, output_key“score”) # 组装成顺序链 overall_chain SequentialChain( chains[critique_chain, revision_chain, scoring_chain], input_variables[“original_text”], # 整个工作流的初始输入 output_variables[“revised_text”, “score”], # 我们最终要的输出 verboseTrue ) result overall_chain.invoke({“original_text”: “这是一段待改进的文本...”}) print(f“修订后文本{result[‘revised_text’]}\\n评分{result[‘score’]}”)这里critique_chain接收original_text产出critique。revision_chain需要两个输入original_text和critique产出revised_text。SequentialChain会自动处理这种依赖关系只要你在子链的prompt模板中声明了对应的变量名。关键点SequentialChain通过output_key和变量名映射来传递数据。务必确保子链的输入变量名与上游链的输出output_key或其他输入变量名一致。3.2 RouterChain实现动态任务分发有些时候流程不是确定的。例如一个用户输入可能是“写首诗”、“翻译这句话”或“总结这篇文章”。你需要先判断意图再路由到不同的处理链。这就是RouterChain。RouterChain通常与MultiPromptChain结合使用。其核心思想是一个LLMChain路由链先分析输入决定将其分发给哪个“专家链”。每个“专家链”对应一个特定的提示词模板和处理逻辑。一个默认链处理无法路由的情况。from langchain.chains.router import MultiPromptChain from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser from langchain.prompts import PromptTemplate # 1. 定义不同“专家”的提示词模板 poet_template “你是一位诗人。请根据以下主题创作一首诗{input}” translator_template “你是一位翻译家。请将以下中文翻译成英文{input}” summarizer_template “你是一位总结助手。请用一句话总结以下文本{input}” prompt_infos [ { “name”: “poetry”, “description”: “适合创作诗歌的请求”, “prompt_template”: poet_template }, { “name”: “translation”, “description”: “适合进行中英翻译的请求”, “prompt_template”: translator_template }, { “name”: “summarization”, “description”: “适合总结文本的请求”, “prompt_template”: summarizer_template } ] # 2. 创建目标链专家链 destination_chains {} for p_info in prompt_infos: prompt PromptTemplate(templatep_info[“prompt_template”], input_variables[“input”]) chain LLMChain(llmllm, promptprompt) destination_chains[p_info[“name”]] chain # 3. 创建默认链 default_prompt PromptTemplate(template“请直接回复以下内容{input}”, input_variables[“input”]) default_chain LLMChain(llmllm, promptdefault_prompt) # 4. 组装MultiPromptChain router_chain MultiPromptChain( router_chainLLMRouterChain.from_prompts(llm, prompt_infos, default_chaindefault_chain), destination_chainsdestination_chains, default_chaindefault_chain, verboseTrue ) # 5. 使用 result router_chain.invoke(“写一首关于春天的诗”) # 路由链会判断这是“poetry”请求将其交给诗人链处理。RouterChain非常适合构建插件化、可扩展的AI应用比如智能客服中的意图识别和转接。4. 与工具和记忆的集成TransformChain与自定义链LLMChain和SequentialChain处理的是LLM之间的协作。但真实世界任务需要LLM与外部工具API、数据库、计算器交互也需要处理非LLM的数据转换。这就需要TransformChain和自定义链。4.1 TransformChain处理纯数据转换TransformChain是一个特殊的链它不包含LLM只包含一个输入/输出处理函数。它用于在LLM链之间进行数据清洗、格式转换或调用纯函数逻辑。from langchain.chains import TransformChain import json def transform_func(inputs: dict) - dict: “””将字符串化的JSON转换为Python对象并提取特定字段。””” text inputs[“text”] data json.loads(text) # 假设我们只关心data字段下的items列表 return {“extracted_items”: data.get(“data”, {}).get(“items”, [])} # 创建转换链 transform_chain TransformChain( input_variables[“text”], output_variables[“extracted_items”], transformtransform_func ) # 可以将其插入SequentialChain中 # 假设 chain1 输出一个JSON字符串chain2 需要处理其中的items overall_chain SequentialChain( chains[chain1, transform_chain, chain2], input_variables[“query”], output_variables[“final_answer”] )4.2 构建自定义链实现工具调用Agent的基石最强大的模式是创建自定义链让LLM能够决定何时、如何调用外部工具。这其实就是LangChain中Agent的核心思想。虽然Agent已经封装得很好但理解其底层链的构造有助于深度定制。一个简化的“工具调用链”可能包含以下步骤思考链分析用户输入决定是否需要调用工具以及调用哪个工具。动作链如果需调用工具则生成工具调用所需的参数一个格式化的字符串或JSON。执行在链外执行工具函数、API调用。观察链将工具执行的结果反馈给LLM。最终回答链综合原始问题、思考和工具结果生成最终回答。你可以用SequentialChain来粗略模拟这个过程但LangChain的Agent和Tool框架提供了更优雅、更强大的抽象。理解这一点后你再去看initialize_agent函数就会明白它内部其实是在组装一个由LLMChain、Tool和特定Agent类如ReActAgent构成的复杂工作流。实操心得在项目早期如果任务流程非常固定且已知优先使用SequentialChain。它更稳定、可预测、调试方便。只有当任务需要LLM动态决策比如“上网搜索”“查数据库”“进行计算”三选一或组合时才考虑引入Agent。Agent虽然灵活但成本更高更多Token消耗、更不可控可能做出错误决策。5. 生产环境下的链调试、监控与最佳实践在开发环境跑通链只是第一步。将其部署到生产环境你需要关注可靠性、性能和可观测性。5.1 调试与日志如前所述verboseTrue是你的第一调试工具。但对于复杂链你可能需要更细粒度的控制。使用LangSmith这是LangChain官方提供的追踪和调试平台。它能可视化链的每一步执行查看输入输出分析延迟和Token消耗是调试复杂工作流的神器。只需设置环境变量LANGCHAIN_TRACING_V2true和LANGCHAIN_API_KEY所有调用会自动记录。自定义回调你可以编写自己的回调处理器将日志发送到你的监控系统如ELK、Datadog。from langchain.callbacks.base import BaseCallbackHandler class MyCustomHandler(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f“ 链 {serialized.get(‘name’)} 开始输入{inputs}”) def on_chain_end(self, outputs, **kwargs): print(f“ 链结束输出{outputs}”) chain LLMChain(..., callbacks[MyCustomHandler()])5.2 性能优化缓存相同的输入往往产生相同的输出。对LLM调用进行缓存可以大幅节省成本和时间。LangChain支持内存缓存(InMemoryCache)、数据库缓存(SQLiteCache)等。from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache()) # 首次调用会访问API result1 chain.invoke({“topic”: “AI”}) # 相同的输入第二次会直接从缓存读取 result2 chain.invoke({“topic”: “AI”})异步支持大多数链都支持异步调用(ainvoke,astream,abatch)。在Web服务器等异步环境中使用可以避免阻塞提高吞吐量。import asyncio async def process_requests(queries): tasks [chain.ainvoke({“input”: q}) for q in queries] results await asyncio.gather(*tasks) return results批处理与速率限制利用batch方法并配合模型的批处理能力。同时注意为API设置合理的速率限制避免被限流。5.3 错误处理与鲁棒性结构化输出的验证即使使用了StructuredOutputParser模型偶尔也可能返回不符合格式的文本。在解析后添加数据验证逻辑如使用Pydantic模型。from pydantic import BaseModel, ValidationError class AdOutput(BaseModel): ad_copy: str keywords: list[str] tone: str try: parsed_dict parser.parse(chain_output) validated_data AdOutput(**parsed_dict) except (ValidationError, Exception) as e: # 处理解析失败例如记录日志、使用默认值、重试等 validated_data AdOutput(ad_copy“生成失败”, keywords[], tone“专业”)降级策略当主要模型如GPT-4调用失败或超时时应有备选方案如切换到GPT-3.5-Turbo或返回一个预定义的友好错误消息。5.4 测试与版本控制对链进行单元测试像测试普通函数一样测试你的链。使用固定的输入断言预期的输出结构或关键内容。Mock LLM的响应使测试快速且不依赖外部API。提示词版本化提示词模板是代码的一部分。将重要的提示词模板存储在版本控制系统如Git中并考虑使用LangChain Hub来管理和共享提示词。配置管理将模型参数temperature, max_tokens、API密钥、链的配置等放在环境变量或配置文件中不要硬编码在代码里。从LLMChain这个简单的“积木”开始通过SequentialChain进行串联利用RouterChain实现分支最后通过自定义逻辑和TransformChain与外部世界连接——你就能搭建出任意复杂的AI智能体工作流。这个过程中核心思想始终是声明式组装和关注点分离。每个链只做好一件事通过清晰的接口组合在一起。当你掌握了这些预置链的用法和思想再去探索LangChain中更高级的Agent和Retrieval模块就会发现它们不过是这些基础模式在特定领域的精妙应用和封装。 
：从核心原理到工程实践)
