6. 链式结构（Chain）的奥秘：从LLMChain到RouterChain

引言：从“单兵作战”到“流水线革命”

2023年某电商平台客服系统因处理复杂咨询需手动串联多个AI模块，平均响应时间长达12秒。引入LangChain链式架构后，工单处理速度提升8倍，错误率下降45%。本文将深入解析链式编程范式，用4种核心Chain类型，教你像搭积木一样构建智能系统。

---

一、Chain的本质：AI开发的“乐高哲学”

1.1 链式编程核心价值

• 模块化：将复杂任务拆解为可复用单元（如“意图识别→信息抽取→生成回复”）

• 灵活组装：通过管道符|组合组件（prompt | llm | output_parser）

• 状态传递：自动管理模块间输入输出，开发者聚焦业务逻辑

1.2 链式类型全景图

类型	典型场景	性能特点
简单链	简单问答/文案生成	低延迟（<500ms）
顺序链	多步骤任务（如报告生成）	顺序执行，易调试
路由链	动态选择处理分支（如多语言）	需预定义路由逻辑
转化链	数据预处理/后处理	无模型调用，高吞吐

---

二、四大核心链式实战

2.1 简单链：智能客服基础应答链

from langchain_ollama import ChatOllama  
from langchain_core.prompts import ChatPromptTemplate  
from langchain_core.output_parsers import StrOutputParser  
​
# 构建产品问答链  
product_chain = (  ChatPromptTemplate.from_template("作为{role}，用{style}风格回答：{query}")  | ChatOllama(model="deepseek-r1")  | StrOutputParser()  
)  
​
response = product_chain.invoke({  "role": "电子产品导购",  "style": "专业严谨",  "query": "对比iPhone15和华为Mate60的摄像头性能"  
})   
print(response)

 输出为：

**iPhone 15 和华为 Mate 60 Pro 摄像头性能对比**
​
### 硬件规格对比
1. **主摄像头**- iPhone 15: 48MP 主摄像头（f/1.79 光圈，支持 ProRes 视频拍摄）- Mate 60 Pro: 50MP 主摄像头（f/1.82 光圈）
​
2. **超广角摄像头**- iPhone 15: 12MP 超广角摄像头（f/2.4 光圈，130° 视野）- Mate 60 Pro: 13MP 超广角摄像头（f/2.2 光圈）
​
3. **长焦摄像头**- iPhone 15: 12MP 长焦摄像头（f/2.8 光圈，支持 OIS）- Mate 60 Pro: 48MP 潜望式长焦摄像头（f/3.5 光圈，支持 OIS）
​
### 软件与算法
- **iPhone 15**: - 延续 AOD（Always-On Display）功能，实时调整 HDR。- 改进的计算摄影和夜景模式，提升低光表现。
​
- **Mate 60 Pro**:- XD Fusion Pro 技术优化图像质量，尤其在复杂光线环境下。- AI 超分辨率变焦增强长焦拍摄效果。
​
### 光学性能
- **iPhone 15**: 高透光率玻璃镜头和抗反射涂层减少眩光。
- **Mate 60 Pro**: 使用多层镀膜技术和优化的镜片结构，提升成像清晰度。
​
### 视频功能
- **iPhone 15**:- 支持 4K/60fps HDR 视频拍摄。- 增强动态范围和色彩保真度。
​
- **Mate 60 Pro**:- 支持 8K 视频录制（7680x4320），1.5 倍于 iPhone 的分辨率。- 慢动作视频支持更高的帧率，如 1080p/240fps。
​
### 总结
- **iPhone 15**: 在计算摄影、动态范围和 HDR 方面表现出色，适合需要稳定高质量照片和视频的用户。
- **Mate 60 Pro**: 拥有更专业的摄像系统，尤其在长焦和高分辨率视频方面更具优势，适合追求专业级拍摄体验的用户。
​
根据具体需求选择：日常使用与一致性表现选 iPhone；专业摄影或高要求视频拍摄考虑 Mate 60 Pro。

2.2 顺序链：工单处理流水线

from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
​
# 定义子链：意图识别 → 解决方案生成
intent_chain = (ChatPromptTemplate.from_template("识别用户意图：{input}")| ChatOllama(model="deepseek-r1")| StrOutputParser()
)
​
solution_chain = (ChatPromptTemplate.from_template("针对'{intent}'生成解决方案")| ChatOllama(model="deepseek-r1")| StrOutputParser()
)
​
# 使用 Runnable 组合链并保留中间结果
full_chain = (intent_chain| RunnableParallel(solution=solution_chain,intent=RunnablePassthrough(),input=RunnablePassthrough(),)
)
​
result = full_chain.invoke({"input": "订单一直未发货"})
print(result)

输出为：

{'solution': '<think>\n好，我现在需要帮用户解决订单未发货的问题。首先，我要理解用户的情况：他们已经下单一段时间了，但订单还没发货，可能因此感到焦虑或不满。\n\n接下来，我应该先致歉，表达对他们不便的关心。然后，解释延迟的原因，比如可能是物流问题、库存短缺还是系统故障。接着，提供解决方案，比如查询最新状态或取消订单并退款。\n\n同时，考虑用户可能的需求，比如是否需要加快发货速度，或者是否有其他替代方案。最后，确保整个回应清晰明了，让用户感到被重视，并愿意进一步协助解决问题。\n</think>\n\n您的订单未发货给您带来了不便，对此我们深表歉意。目前，我们正在积极处理中，预计将在未来几天内完成发货。如果您有任何疑问或需要进一步的帮助，请随时联系我们的客服团队，我们将竭诚为您服务。感谢您的耐心等待和理解！', 'intent': '<think>\n好的，我现在需要解决的问题是用户的订单一直未发货的情况。首先，我得理解用户为什么会感到困扰，可能是因为他们急需商品或者对我们的服务产生了不满。\n\n接下来，我要考虑如何回应用户。应该先感谢他们的耐心等待，并表达歉意，这样可以缓和情绪。然后，解释订单延迟的原因，比如可能是供应链问题、物流影响还是系统故障等。同时，提供一些解决方案，比如查询最新状态或建议取消订单并退款。\n\n此外，我还需要思考是否有其他可能的解决办法，比如加快发货时间或者提供补偿措施，如优惠券或折扣，以弥补用户的不便。最后，确保整个回应清晰、有条理，并且让用户感到被重视和理解。\n</think>\n\n您的订单一直未发货给您带来了困扰，对此我们深表歉意。目前，我们正在积极处理中，预计将在未来几天内完成发货。如果情况没有改善，请随时联系我们的客服团队以获取进一步的帮助。感谢您的耐心等待和理解！', 'input': '<think>\n好的，我现在需要解决的问题是用户的订单一直未发货的情况。首先，我得理解用户为什么会感到困扰，可能是因为他们急需商品或者对我们的服务产生了不满。\n\n接下来，我要考虑如何回应用户。应该先感谢他们的耐心等待，并表达歉意，这样可以缓和情绪。然后，解释订单延迟的原因，比如可能是供应链问题、物流影响还是系统故障等。同时，提供一些解决方案，比如查询最新状态或建议取消订单并退款。\n\n此外，我还需要思考是否有其他可能的解决办法，比如加快发货时间或者提供补偿措施，如优惠券或折扣，以弥补用户的不便。最后，确保整个回应清晰、有条理，并且让用户感到被重视和理解。\n</think>\n\n您的订单一直未发货给您带来了困扰，对此我们深表歉意。目前，我们正在积极处理中，预计将在未来几天内完成发货。如果情况没有改善，请随时联系我们的客服团队以获取进一步的帮助。感谢您的耐心等待和理解！'}

2.3 路由链：多语言智能路由

from langchain_core.runnables import RunnableBranch
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
​
llm = ChatOllama(model="deepseek-r1")
​
chinese_chain = (ChatPromptTemplate.from_template("你是一名中文客服，回答：{input}")| llm| StrOutputParser()
)
​
english_chain = (ChatPromptTemplate.from_template("You are an English assistant, answer: {input}")| llm| StrOutputParser()
)
​
default_chain = (ChatPromptTemplate.from_template("未知语言：{input}")| llm| StrOutputParser()
)
​
# 定义路由功能
def route_input(input: str) -> str:if any('\u4e00' <= char <= '\u9fff' for char in input):  return "chinese"elif input.isascii(): return "english"return "default"
​
​
branch = RunnableBranch((lambda x: route_input(x["input"]) == "chinese", chinese_chain),(lambda x: route_input(x["input"]) == "english", english_chain),default_chain
)
​
full_chain = {"input": lambda x: x["input"]} | branch
​
print(full_chain.invoke({"input": "How to reset my password?"}))
print(full_chain.invoke({"input": "如何退换货？"}))

输出为：

<think>
Okay, so I need to figure out how to reset my password. Hmm, where do I start? Well, first off, the user wants to know the process for resetting their password, but they didn't specify which account or service they're referring to. That's a bit tricky because different platforms have different methods.
​
Let me think about this step by step. If someone is trying to reset their password on a website, usually there's a "Forgot Password" link somewhere. It might be on the login page or maybe in the account settings. Once they click that, they should enter the email associated with their account. Then, the system would send them a link via email to create a new password.
​
Wait, but what if they don't remember the email linked to their account? That could be a problem. They might need to contact customer support then. Also, sometimes when resetting a password, you have to answer security questions or go through two-factor authentication if that's enabled on their account.
​
Oh, and I should remind them about creating a strong new password. It shouldn't be something easy like "password123" because that's too common and insecure. They need to use a mix of letters, numbers, and symbols, maybe 8-12 characters long.
​
Another thing to consider is what happens if they don't have access to their email anymore or if they can't remember the security answers. In such cases, contacting support would be the way to go because they might need additional steps to verify identity before resetting the password.
​
I should also make sure they know not to share their new password with anyone and to keep it secure. Maybe suggest using a password manager to help them manage all their passwords safely.
​
Oh, and sometimes after resetting the password, it's a good idea to check all account settings just in case there were unauthorized changes. Security is important, so ensuring everything looks normal on the account is a good step.
​
Wait, did I cover all the bases? Let me recap: locate the reset option, enter email or username, follow instructions which might include security questions, create a strong password, and maybe contact support if there are issues. Yeah, that seems comprehensive. I think I have a solid plan here to explain how to reset a password.
</think>
​
To reset your password, follow these organized steps:
​
1. **Locate the Reset Option**: - Go to the login page of the website or service you're using and look for a "Forgot Password" link.
​
2. **Enter Your Email/Username**:- Click on the "Forgot Password" link and enter the email address or username associated with your account.
​
3. **Follow Instructions**:- Check your email for a reset link sent by the service.- If prompted, answer any security questions or go through two-factor authentication if enabled.
​
4. **Create a Strong Password**:- When prompted, create a new password that is secure. It should be 8-12 characters long and include a mix of letters, numbers, and symbols. Avoid common words or simple sequences like "password123".
​
5. **Contact Support if Needed**:- If you can't access your email or don't remember security answers, contact the service's customer support for assistance.
​
6. **Secure Your Account**:- After resetting, avoid sharing your password and consider using a password manager to securely manage it.- Check your account settings to ensure everything appears normal and secure.
​
By following these steps, you can effectively reset your password while maintaining security.
<think>
好的，我现在需要解决用户关于“如何退换货”的问题。首先，我要理解这个问题的具体要求是什么。用户可能刚刚购买了商品，遇到了一些问题，比如商品质量问题、不合适或者收到了错误的商品，所以他们想了解退货或换货的流程。
​
接下来，我应该考虑用户的使用场景。他们可能是第一次网购，不太清楚退换货的具体步骤，或者是遇到了问题需要帮助解决。因此，我的回答需要详细且易懂，避免使用过于专业的术语，让用户能够轻松跟随指引完成操作。
​
然后，我要分析用户的身份。可能是一个普通消费者，对电商流程不熟悉，所以需要详细的指导。此外，用户的真实需求不仅仅是了解流程，还包括在遇到问题时如何快速有效地解决问题，减少他们的麻烦和时间成本。
​
接下来，我应该考虑用户的深层需求。他们可能希望整个过程简单、透明，不会被繁琐的步骤困扰，同时希望得到及时的帮助和支持。因此，在回答中不仅要提供步骤，还要给出联系方式，让用户在遇到问题时能够寻求帮助。
​
现在，我要构建一个清晰的回答结构。首先，说明退换货通常需要联系卖家或平台客服，然后列出可能的几种情况：质量问题、不合适或错误商品。接着详细描述每种情况下的处理步骤，包括检查订单信息、联系卖家、提供证据、寄回商品以及退款或换货的处理。
​
此外，我还需要提醒用户注意一些关键点，比如保留商品原貌、及时联系客服、了解平台规则等。这些提示可以帮助用户避免常见错误，确保退换货过程顺利进行。
​
最后，我要确保整个回答逻辑清晰，分步骤明确，语言简洁明了。这样用户可以按照指引一步步操作，不会感到困惑或不知所措。
</think>
​
您好！关于如何退换货，您可以按照以下步骤进行操作：
​
1. **联系卖家**：首先，请您确认商品是否存在质量问题或者是否符合您的需求。如果需要退换货，请及时与卖家取得联系。
​
2. **提供相关信息**：- **订单号**：请准备好您的订单号以便快速查询。- **商品信息**：描述清楚商品的状况，例如是否有损坏、瑕疵等。- **照片或视频**（如有）：如果商品有质量问题，请尽量提供清晰的照片或视频作为证据。
​
3. **寄回商品**：- 在寄回商品前，请确保商品处于原始状态，包装完整，并附上相关单据（如发票、保修卡等）。- 请使用可靠的快递方式寄回，并保存好快递单号以便查询物流信息。
​
4. **退款或换货处理**：- 如果您选择退货，卖家会在收到商品并确认无误后为您办理退款手续。- 如果您选择换货，请与卖家沟通明确换货的具体要求和流程。
​
5. **注意事项**：- 请保留好商品的原包装、说明书及相关附件，以免影响退换货的顺利进行。- 退换货通常有一定的时效性，请务必在规定时间内完成相关操作。- 如果在退换货过程中遇到任何问题，请及时联系卖家或平台客服寻求帮助。
​
如有其他疑问，欢迎随时咨询！祝您购物愉快！

2.4 转化链：数据清洗流水线

from langchain.chains import TransformChain
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama
​
​
# 1. 定义文本清洗转换链
def clean_text(inputs: dict) -> dict:"""清洗文本：移除特殊字符和多余空格"""text = inputs["raw_text"]# 移除#号和反斜杠，并去除首尾空格cleaned = text.replace("#", "").replace("\\", "").strip()return {"cleaned_text": cleaned}
​
​
# 创建清洗链
clean_chain = TransformChain(input_variables=["raw_text"],  # 输入变量名output_variables=["cleaned_text"],  # 输出变量名transform=clean_text  # 清洗函数
)
​
# 2. 定义产品信息分析链
# 使用中文提示模板
product_prompt = ChatPromptTemplate.from_template("请分析以下产品查询：{cleaned_text}\n""要求：\n""1. 识别产品类型\n""2. 列出关键参数\n""3. 给出购买建议"
)
​
# 初始化大语言模型
llm = ChatOllama(model="deepseek-r1:32B")
​
# 创建产品分析链
product_chain = product_prompt | llm
​
# 3. 正确组合处理管道
pipeline = clean_chain | product_chain
​
​
# 4. 创建处理函数
def process_text(input_text: str):"""处理文本的完整流程"""result = pipeline.invoke({"raw_text": input_text})return result
​
​
# 5. 测试管道
if __name__ == "__main__":# 测试用例：包含特殊字符的产品查询test_input = "##预算5000元\\的最佳#游戏笔记本#"
​print("=== 测试开始 ===")print("\n【原始输入】", test_input)result = process_text(test_input)
​print("\n【分析报告】")print(result.content)print("=== 测试结束 ===")

输出为：

=== 测试开始 ===
​
【原始输入】 ##预算5000元\的最佳#游戏笔记本#
​
【分析报告】
<think>
嗯，用户问的是预算5000元左右的最佳游戏笔记本。首先，我需要明确产品类型是游戏本，所以重点应该放在性能上。
​
接下来，考虑用户可能的需求。他们可能是在校学生或者刚工作的年轻人，希望在有限的预算内买到一款能流畅运行主流游戏的笔记本。除了玩游戏，可能还会有日常办公、多媒体娱乐的需求，所以续航和便携性也可能重要。
​
关键参数方面，处理器肯定是要好的，i5或者R5这种中端型号应该合适。显卡的话，RTX 3060或RX 6600M这样的独立显卡比较适合，能够应对1080p高画质游戏。内存和存储建议至少16GB RAM和512GB SSD，这样运行大型游戏更流畅，加载时间也更快。
​
屏幕的话，144Hz刷新率是基本要求，IPS面板能带来更好的色彩表现，对长时间使用眼睛更好。电池容量要足够支撑一段时间的游戏或者日常使用，所以至少50Wh以上比较好。
​
品牌方面，联想、惠普、戴尔、华硕和微星都是不错的选择，这些品牌的售后服务也比较完善，用户可以放心购买。另外，还要提醒用户注意散热设计，毕竟游戏本容易发热，影响性能发挥。
​
最后，根据不同的侧重点推荐几款具体的型号，比如适合主流游戏的拯救者Y7000P、追求性价比的暗影精灵9以及喜欢高刷新率屏幕的华硕天选3。这样用户可以根据自己的具体需求选择最适合的那一款。
​
总的来说，预算5000元的游戏本市场有很多不错的选择，重点是要在性能和价格之间找到平衡点，确保能够满足用户的主要使用场景。
</think>
​
### 1. **产品类型识别**- **游戏笔记本**：专注于高性能硬件配置，适合运行主流3D游戏、图形处理等高负载任务。
​
---
​
### 2. **关键参数分析**
以下是选择预算5000元游戏笔记本时需要关注的关键参数：
- **处理器（CPU）**：  - 推荐选择Intel i5或AMD R5系列，如i5-13500H/R5-7600H。- 避免选择性能较低的低压处理器（如U系列）。
​
- **显卡（GPU）**：  - 独立显卡是游戏笔记本的核心配置，推荐NVIDIA RTX 3060或AMD Radeon RX 6600M/6700M。- 显存容量建议至少8GB GDDR6。
​
- **内存（RAM）**：  - 至少16GB DDR4/DDR5双通道内存，满足大型游戏和多任务处理需求。- 避免选择8GB内存的配置。
​
- **存储（SSD）**：  - 推荐512GB NVMe SSD，支持快速启动和加载游戏。- 避免选择机械硬盘（HDD）或仅有256GB SSD的配置。
​
- **屏幕（Display）**：  - 分辨率建议为1920x1080（FHD），刷新率至少144Hz，适合流畅游戏体验。- 颜色校准、高亮度和低延迟也是加分项。
​
- **电池与续航**：  - 推荐大容量电池（≥50Wh），支持长时间使用。- 游戏本通常以性能优先，续航可能较短，但需满足日常轻度使用需求。
​
- **散热设计**：  - 需要高效的散热系统，避免因高温导致降频，影响游戏体验。
​
---
​
### 3. **购买建议**
#### **推荐型号（预算5000元左右）**
以下是一些性价比高且适合主流游戏的笔记本推荐：
1. **联想拯救者Y7000P**  - CPU：i5-13500H/R5-7600H- GPU：RTX 3060/AMD Radeon RX 7600M- 屏幕：144Hz FHD IPS- 适合主流游戏和日常使用。
​
2. **惠普暗影精灵9**  - CPU：i5-13500H/R5-7600H- GPU：RTX 3060/AMD Radeon RX 7600M- 屏幕：144Hz FHD IPS- 性价比高，散热表现优秀。
​
3. **戴尔G15/G16**  - CPU：i5-13500H/R5-7600H- GPU：RTX 3060/AMD Radeon RX 7600M- 屏幕：可选144Hz或165Hz FHD IPS- 品牌售后有保障，适合长时间使用。
​
4. **华硕天选3**  - CPU：i5-13500H/R7-7730H- GPU：RTX 3060/AMD Radeon RX 7700M- 屏幕：可选144Hz或240Hz FHD IPS- 性能强劲，适合高刷新率游戏。
​
5. **微星冲锋TITAN GT**  - CPU：i5-13500H/R7-7730H- GPU：RTX 3060/AMD Radeon RX 7700M- 屏幕：144Hz FHD IPS- 散热设计优秀，适合高性能需求。
​
---
​
#### **注意事项**
1. **关注散热设计**：游戏笔记本需要高效的散热系统，避免因高温导致性能下降。
2. **选择大品牌**：联想、惠普、戴尔、华硕等品牌的售后服务和质量更有保障。
3. **验证配置真实性**：部分商家可能会虚标配置或使用二手翻新机，建议从官网或授权渠道购买。
4. **性价比优先**：在预算范围内尽量选择性能更强的显卡和更高的刷新率屏幕。
​
---
​
#### **总结**
预算5000元左右的游戏笔记本市场中，联想拯救者Y7000P、惠普暗影精灵9和华硕天选3是性价比较高的选择。它们均搭载了i5-13500H/R5-7600H处理器和RTX 3060显卡，能够流畅运行《英雄联盟》、《原神》等主流游戏，适合大多数游戏玩家的需求。
=== 测试结束 ===

---

三、企业级案例：电商工单系统架构

3.1 核心流程设计

3.2 性能优化技巧

• 异步加速：非依赖任务并行执行

import asyncio
from langchain_community.llms import FakeListLLM
from langchain.prompts import PromptTemplate
​
​
async def parallel_process(inputs):# 模拟库存API的chainchain1 = create_mock_inventory_chain()
​# 模拟促销信息的chainchain2 = create_mock_promotion_chain()
​# 并行执行两个chainresults = await asyncio.gather(chain1.ainvoke(inputs),chain2.ainvoke(inputs))
​return format_responses(*results)
​
​
def create_mock_inventory_chain():"""创建模拟库存API的chain"""responses = ["库存状态: 产品ID {} 有 150 件库存".format(i) for i in range(1, 6)]llm = FakeListLLM(responses=responses)
​prompt = PromptTemplate(input_variables=["product_id"],template="查询产品 {product_id} 的库存状态")return prompt | llm
​
​
def create_mock_promotion_chain():"""创建模拟促销信息的chain"""responses = ["促销信息: 产品ID {} 享受8折优惠".format(i) for i in range(1, 4)] + ["促销信息: 产品ID {} 无促销活动".format(i) for i in range(4, 6)]llm = FakeListLLM(responses=responses)
​prompt = PromptTemplate(input_variables=["product_id"],template="获取产品 {product_id} 的促销信息")return prompt | llm
​
​
def format_responses(inventory_resp, promotion_resp):"""格式化两个chain的响应结果"""return {"inventory_info": inventory_resp,"promotion_info": promotion_resp,"combined_info": f"{inventory_resp}\n{promotion_resp}"}
​
​
# 测试代码
async def main():# 测试产品ID 3result = await parallel_process({"product_id": 3})print("测试结果:")print(result)
​
​
if __name__ == "__main__":asyncio.run(main())

输出为：

测试结果:
{'inventory_info': '库存状态: 产品ID 1 有 150 件库存', 'promotion_info': '促销信息: 产品ID 1 享受8折优惠', 'combined_info': '库存状态: 产品ID 1 有 150 件库存\n促销信息: 产品ID 1 享受8折优惠'}

---

四、避坑指南：链式开发六大陷阱

1. 循环依赖：链A依赖链B的结果，链B又依赖链A → 死锁

2. 变量覆盖：多个链使用同名变量导致数据丢失

3. 超时失控：复杂链式未设置超时阈值 → 服务雪崩

4. 错误传播：未捕获单个链的错误 → 全流程崩溃

5. 资源泄漏：未及时释放模型资源 → 内存溢出

6. 监控缺失：无法定位性能瓶颈 → 优化困难

---

五、下期预告

《记忆（Memory）机制：让AI拥有“短期记忆”与“长期记忆”》

• 揭秘：如何让Deepseek-R1突破“7秒记忆”，实现跨会话对话历史留存？

• 实战：基于FAISS本地存储构建用户画像知识库

• 陷阱：长期记忆中的隐私泄露与合规风险

---

链式架构是LangChain的灵魂，掌握它意味着掌握了复杂AI系统的构建密码。记住：优秀的链式设计，既要像钟表般精密，又要像乐高般灵活！