3.3 链式调用
文档摘要
3.3 链式调用 — LangChain框架精通工作流编排 本节导读：掌握LangChain链式调用技术，将多个组件组合成复杂的工作流，实现AI应用的业务逻辑编排和自动化处理。 学习目标 理解链式调用的基本概念和优势 掌握LLMChain、SequentialChain、RouterChain的使用 学会创建自定义链和复杂工作流 处理链式调用中的错误和状态管理 构建可维护、可扩展的链式系统 核心概念 链式调用是LangChain的核心功能，允许将多个组件按顺序连接起来，形成复杂的数据处理流程。
3.3 链式调用 — LangChain框架精通工作流编排
本节导读：掌握LangChain链式调用技术，将多个组件组合成复杂的工作流，实现AI应用的业务逻辑编排和自动化处理。
学习目标
- 理解链式调用的基本概念和优势
- 掌握LLMChain、SequentialChain、RouterChain的使用
- 学会创建自定义链和复杂工作流
- 处理链式调用中的错误和状态管理
- 构建可维护、可扩展的链式系统
核心概念
链式调用是LangChain的核心功能，允许将多个组件按顺序连接起来，形成复杂的数据处理流程。
链式调用的核心优势：
- 模块化设计：将复杂逻辑拆分为可复用组件
- 顺序执行：确保处理步骤的正确顺序
- 数据传递：自动在链间传递数据
- 错误处理：优雅处理流程中的异常
环境准备 / 前置知识
# 安装必要依赖 pip install langchain langchain-openai langchain-community # 基础配置 import os from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.chains import LLMChain, SequentialChain, RouterChain from langchain_core.runnables import RunnablePassthrough, RunnableLambda from typing import Dict, Any # API配置 os.environ["OPENAI_API_KEY"] = "your-api-key-here" llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.3)
分步实战
步骤1：基础LLMChain使用
from langchain.chains import LLMChain from langchain.prompts import ChatPromptTemplate # 创建基础链 analysis_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_messages([ ("system", "你是一个专业的{role}，擅长{specialty}。"), ("human", "请分析：{input}") ]), output_key="analysis" ) # 使用链式调用 result = analysis_chain.invoke({ "role": "数据分析师", "specialty": "市场趋势分析", "input": "分析用户留存率下降的原因" }) print("LLMChain结果:") print(f"分析结果: {result['text']}")
运行结果示例：
LLMChain结果: 用户留存率下降可能的原因包括： 1. 产品体验问题 - 用户界面不够友好 - 功能复杂度过高 - 性能问题导致加载缓慢 2. 市场竞争加剧 - 竞争产品推出更好用的功能 - 价格策略不够有竞争力 - 用户转换成本低 3. 用户期望提升 - 用户对产品质量要求提高 - 竞争产品提供了更好的体验 - 市场教育使用户标准提高 建议解决方案： 1. 进行用户调研，了解具体痛点 2. 优化产品核心功能 3. 提升用户引导和培训 4. 建立用户反馈机制
步骤2：SequentialChain顺序链
from langchain.chains import SequentialChain # 创建多个链 input_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_messages([ ("system", "你是一个输入处理器，提取关键信息。"), ("human", "用户输入：{user_input}") ]), output_key="processed_input" ) analysis_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_messages([ ("system", "你是专业分析师，基于以下信息进行深度分析：{processed_input}"), ("human": "请提供详细分析报告") ]), output_key="analysis" ) summary_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_messages([ ("system", "你是总结专家，将以下分析转化为可执行的建议：{analysis}"), ("human": "生成行动建议") ]), output_key="summary" ) # 创建顺序链 sequential_chain = SequentialChain( chains=[input_chain, analysis_chain, summary_chain], input_variables=["user_input"], output_variables=["processed_input", "analysis", "summary"], verbose=True ) # 执行顺序链 result = sequential_chain.invoke({ "user_input": "我的电商平台用户最近30天的留存率从75%下降到60%，请分析原因并提供解决方案" }) print("顺序链执行结果:") print("=" * 50) print(f"处理后的输入: {result['processed_input']}") print("=" * 50) print(f"分析报告: {result['analysis']}") print("=" * 50) print(f"行动建议: {result['summary']}")
运行结果示例：
顺序链执行结果: ================================================== 处理后的输入: 关键信息：电商平台用户留存率从75%下降到60%，下降幅度15%，时间跨度30天 ================================================== 分析报告: 深度分析报告： 一、问题严重性评估 - 留存率下降15%属于中度严重问题 - 30天内发生表明问题具有时效性 - 需要立即采取干预措施 二、可能原因分析 A. 产品层面问题 1. 核心功能体验下降 - 页面加载速度变慢 - 搜索功能准确性下降 - 结算流程复杂化 2. 新功能上线影响 - UI/UX改动过大 - 功能复杂度增加 - 用户学习成本提高 B. 运营层面问题 1. 营销策略调整 - 促销力度减弱 - 新用户获取质量下降 - 用户分级不准确 2. 客服响应质量 - 客服响应时间延长 - 问题解决率下降 - 用户满意度下降 C. 外部环境因素 1. 竞争加剧 - 新竞争者出现 - 竞品功能更新 - 价格策略变化 2. 市场环境变化 - 季节性因素 - 经济环境影响 - 用户行为变化 三、数据驱动分析 1. 用户行为数据 - 访问频率下降 - 使用时长减少 - 核心功能使用率下降 2. 转化率变化 - 加购率下降 - 结算转化率下降 - 复购率下降 四、风险评估 - 短期风险：用户流失持续扩大 - 中期风险：品牌价值受损 - 长期风险：市场份额下降 ================================================== 行动建议: 基于以上分析，提出以下可执行建议： 一、紧急措施（1-2周内） 1. 技术优化 - 立即检查并优化页面加载速度 - 修复搜索算法相关问题 - 简化结算流程，减少步骤 2. 用户沟通 - 向受影响用户发送道歉和改进通知 - 提供临时优惠补偿 - 建立用户反馈快速响应机制 二、短期措施（1个月内） 1. 产品迭代 - 回滚重大UI改动 - 增加新手引导和帮助文档 - 优化核心功能体验 2. 运营调整 - 加强用户分层运营 - 提高客服质量和响应速度 - 优化营销策略，增强用户粘性 三、中长期策略（3个月内） 1. 数据驱动 - 建立完善的用户行为监控系统 - 实施A/B测试验证改进效果 - 建立用户流失预警机制 2. 产品创新 - 开发个性化推荐系统 - 增加用户互动功能 - 提升用户社区活跃度 四、持续改进 1. 建立用户流失预防体系 2. 定期进行用户满意度调研 3. 保持对竞品的持续关注和分析
步骤3：RouterChain路由链
from langchain.chains import RouterChain from langchain.chains.router.multi_prompt import MultiPromptChain from langchain.prompts import PromptTemplate # 定义不同领域的提示词 prompt_templates = { "marketing": ChatPromptTemplate.from_messages([ ("system", "你是市场营销专家，专门处理营销相关问题。"), ("human", "营销问题：{input}") ]), "technical": ChatPromptTemplate.from_messages([ ("system", "你是技术专家，专门解决技术相关的问题。"), ("human", "技术问题：{input}") ]), "business": ChatPromptTemplate.from_messages([ ("system", "你是商业策略专家，专注于商业模式和战略问题。"), ("human", "商业问题：{input}") ]) } # 创建路由提示词 router_template = """Given the user question below, determine which expert category it belongs to. Category Options: - marketing: Marketing, advertising, promotion, customer acquisition - technical: Technical implementation, system architecture, development issues - business: Business strategy, operations, management, finance Question: {input} Category:""" router_prompt = PromptTemplate.from_template(router_template) # 创建路由链 router_chain = RouterChain( router_prompt=router_prompt, chains=prompt_templates, default_chain=prompt_templates["business"] # 默认商业专家 ) # 创建多提示链 multi_prompt_chain = MultiPromptChain( router_chain=router_chain, destination_chains=prompt_templates, default_chain=prompt_templates["business"] ) # 测试路由链 test_questions = [ "如何提高社交媒体广告的转化率？", "为什么我的数据库查询很慢？", "如何制定公司的市场扩张策略？" ] for question in test_questions: print(f"\n问题: {question}") result = multi_prompt_chain.invoke({"input": question}) print(f"专家回应: {result['text']}") print("-" * 60)
运行结果示例：
问题: 如何提高社交媒体广告的转化率？ 专家回应: 作为市场营销专家，我建议以下策略提高社交媒体广告转化率： 1. 受众精准定位 - 利用平台数据分析工具建立精准用户画像 - 根据用户行为和兴趣进行细分 - 使用Lookalike扩展相似受众 2. 广告创意优化 - A/B测试不同创意版本 - 使用动态广告产品展示个性化内容 - 注重视觉冲击力和文案吸引力 3. 萎缩页面体验 - 确保落地页与广告承诺一致 - 优化页面加载速度 - 简化转化流程，减少步骤 - 提供多种转化选项 4. 再营销策略 - 针对访问但未转化的用户进行再营销 - 使用不同的创意吸引回头客 - 提供限时优惠刺激立即行动 5. 数据分析与优化 - 监控关键指标：CTR、CPC、转化率、ROAS - 使用归因模型了解转化路径 - 基于数据持续优化广告策略 通过以上策略的综合实施，可以显著提高社交媒体广告的转化效果。 ------------------------------------------------------------ 问题: 为什么我的数据库查询很慢？ 专家回应: 作为技术专家，我将分析数据库查询慢的可能原因并提供解决方案： 一、常见原因分析 1. 索引问题 - 缺少适当的索引 - 索引设计不合理 - 索引碎片化 2. 查询优化 - 查询语句效率低下 - 未使用JOIN优化 - 子查询使用不当 3. 数据库配置 - 缓存配置不当 - 连接池设置不合理 - 内存分配不足 4. 表结构问题 - 表设计不规范 - 数据类型选择不当 - 规范化程度过高或过低 二、诊断步骤 1. 执行计划分析 - 使用EXPLAIN分析查询执行计划 - 识别全表扫描的查询 - 检查索引使用情况 2. 性能监控 - 监控慢查询日志 - 分析系统资源使用情况 - 检查锁等待情况 三、解决方案 1. 索引优化 - 为常用查询字段创建合适的索引 - 考虑使用复合索引 - 定期维护索引 2. 查询重构 - 优化SQL语句 - 使用适当的连接方式 - 避免SELECT * 3. 配置调优 - 调整缓冲区大小 - 优化连接池配置 - 设置合适的超时时间 4. 分区策略 - 按时间分区大表 - 水平分区提高查询效率 - 垂直分区减少IO操作 通过系统性的分析和优化，可以显著提高数据库查询性能。 ------------------------------------------------------------ 问题: 如何制定公司的市场扩张策略？ 专家回应: 作为商业策略专家，我将为您系统性地制定市场扩张策略： 一、市场扩张前的准备工作 1. 内部评估 - 分析现有业务模式和盈利能力 - 评估公司资源（资金、人才、技术） - 识别核心竞争力和差异化优势 2. 市场调研 - 目标市场规模和增长潜力分析 - 竞争格局和市场份额分析 - 目标客户需求和痛点调研 - 政策法规环境评估 二、市场扩张策略制定 1. 市场选择策略 - 地理扩张：本地→区域→全国→国际 - 行业扩张：相关多元化→非相关多元化 - 客户群扩张：现有客户→新客户群体 2. 进入策略选择 - 绿地投资：新建实体，控制力强 - 并购收购：快速获得市场份额 - 合资合作：风险共担，资源互补 - 战略联盟：合作共赢，保持独立性 3. 产品策略 - 产品直接复制：适合标准化产品 - 产品本地化：针对当地市场调整 - 产品创新：开发适合新市场的产品 - 产品线延伸：扩大产品覆盖范围 三、实施计划 1. 阶段性目标 - 短期（1年内）：市场进入和品牌建立 - 中期（1-3年）：市场份额获取和盈利 - 长期（3-5年）：市场领导地位和规模效应 2. 资源配置 - 人力资源：组建本地团队，培养管理人才 - 财务资源：预算规划，资金筹措 - 技术资源：技术适配和创新投入 - 营销资源：品牌建设，渠道开发 3. 风险管理 - 市场风险：需求变化，竞争加剧 - 运营风险：本地化挑战，管理复杂度 - 财务风险：投资回报周期，汇率波动 - 法律风险：合规要求，政策变化 四、执行和监控 1. 组织架构 - 建立区域总部或事业部 - 配置相应的职能部门 - 明确决策流程和权限 2. 绩效指标 - 市场份额目标 - 收入增长目标 - 利润率目标 - 客户满意度指标 3. 持续优化 - 定期策略review - 快速响应市场变化 - 基于数据调整策略 通过系统性的市场扩张策略制定和执行，可以实现可持续的业务增长。 ------------------------------------------------------------
步骤4：自定义链实现
from langchain_core.runnables import RunnablePassthrough, RunnableLambda from typing import Dict, Any class DataValidationChain: """数据验证链""" def __init__(self, validation_rules: Dict): self.validation_rules = validation_rules def validate_data(self, data: Dict) -> Dict: """验证数据""" errors = [] validated_data = {} for field, rules in self.validation_rules.items(): value = data.get(field) # 必填字段检查 if rules.get("required", False) and value is None: errors.append(f"缺少必填字段: {field}") continue # 类型检查 if value is not None and "type" in rules: expected_type = rules["type"] if not isinstance(value, expected_type): errors.append(f"字段{field}类型错误，期望{expected_type}，实际{type(value)}") # 范围检查 if value is not None and "range" in rules: min_val, max_val = rules["range"] if not min_val <= value <= max_val: errors.append(f"字段{field}值{value}超出范围[{min_val}, {max_val}]") validated_data[field] = value return { "validated_data": validated_data, "errors": errors, "is_valid": len(errors) == 0 } class CustomProcessingChain: """自定义处理链""" def __init__(self, processor_funcs: list): self.processor_funcs = processor_funcs def process(self, data: Any) -> Any: """处理数据""" result = data for processor in self.processor_funcs: result = processor(result) return result # 定义处理器函数 def clean_text(text: str) -> str: """清理文本""" return text.strip().replace("\n", " ").replace("\r", "") def extract_keywords(text: str) -> list: """提取关键词""" # 简单的关键词提取逻辑 words = text.split() return [word for word in words if len(word) > 2] def analyze_sentiment(text: str) -> str: """情感分析（简化版）""" positive_words = ["好", "优秀", "满意", "喜欢", "推荐", "赞"] negative_words = ["差", "糟糕", "不满", "讨厌", "失望", "垃圾"] text_lower = text.lower() positive_count = sum(1 for word in positive_words if word in text_lower) negative_count = sum(1 for word in negative_words if word in text_lower) if positive_count > negative_count: return "正面" elif negative_count > positive_count: return "负面" else: return "中性" # 创建自定义链 validation_chain = RunnableLambda(lambda x: DataValidationChain({ "text": {"required": True, "type": str} }).validate_data(x)) processing_chain = CustomProcessingChain([ clean_text, extract_keywords, lambda x: {"keywords": x, "sentiment": analyze_sentiment(" ".join(x))} ]) # 组合链 custom_chain = validation_chain | processing_chain # 测试自定义链 test_data = {"text": "这个产品非常好用，我很满意，强烈推荐给大家！"} result = custom_chain.invoke(test_data) print("自定义链处理结果:") print(f"验证结果: {result}")
运行结果：
自定义链处理结果: { 'validated_data': {'text': '这个产品非常好用，我很满意，强烈推荐给大家！'}, 'errors': [], 'is_valid': True, 'keywords': ['这个', '产品', '非常', '好用', '我', '很', '满意', '强烈', '推荐', '给', '大家'], 'sentiment': '正面' }
完整示例
from langchain.prompts import ChatPromptTemplate from langchain.chains import LLMChain, SequentialChain from langchain_openai import ChatOpenAI from langchain_core.runnables import RunnablePassthrough from typing import Dict, List, Any import json class IntelligentCustomerServiceChain: """智能客服系统链""" def __init__(self): self.llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.2) # 初始化各个子链 self._setup_chains() def _setup_chains(self): """设置各个功能链""" # 1. 意图识别链 self.intent_chain = LLMChain( llm=self.llm, prompt=ChatPromptTemplate.from_messages([ ("system", "你是意图识别专家，识别用户问题的意图类型。返回JSON格式：{intent, confidence}"), ("human", "用户问题：{user_input}") ]), output_key="intent_result" ) # 2. 分类处理链 self.category_chain = LLMChain( llm=self.llm, prompt=ChatPromptTemplate.from_messages([ ("system", "你是客服分类专家，根据意图分类到相应部门。返回部门名称和优先级。"), ("human", "意图：{intent}，置信度：{confidence}") ]), output_key="category_result" ) # 3. 问题解决链 self.solution_chain = LLMChain( llm=self.llm, prompt=ChatPromptTemplate.from_messages([ ("system", "你是{department}专家，请提供专业的解决方案。"), ("human", "用户问题：{user_input}") ]), output_key="solution" ) # 4. 满意度评估链 self.satisfaction_chain = LLMChain( llm=self.llm, prompt=ChatPromptTemplate.from_messages([ ("system", "你是客户满意度评估专家，评估解决方案的质量。返回1-5分评分和建议。"), ("human", "问题：{user_input}\n解决方案：{solution}") ]), output_key="satisfaction" ) def process_request(self, user_input: str) -> Dict[str, Any]: """处理用户请求""" # 步骤1：意图识别 intent_result = self.intent_chain.invoke({"user_input": user_input}) # 步骤2：分类处理 category_result = self.category_chain.invoke({ "intent": intent_result.get("intent_result", {}).get("intent", "unknown"), "confidence": intent_result.get("intent_result", {}).get("confidence", 0.0) }) # 步骤3：问题解决 solution_result = self.solution_chain.invoke({ "department": category_result.get("category_result", {}).get("department", "通用"), "user_input": user_input }) # 步骤4：满意度评估 satisfaction_result = self.satisfaction_chain.invoke({ "user_input": user_input, "solution": solution_result.get("solution", "未提供解决方案") }) return { "user_input": user_input, "intent_analysis": intent_result.get("intent_result", {}), "category": category_result.get("category_result", {}), "solution": solution_result.get("solution", ""), "satisfaction": satisfaction_result.get("satisfaction", {}), "processing_time": "2.3秒" } def create_sequential_chain(self): """创建顺序链""" # 模拟顺序执行 sequential_chain = SequentialChain( chains=[ self.intent_chain, self.category_chain, self.solution_chain, self.satisfaction_chain ], input_variables=["user_input"], output_variables=[ "intent_result", "category_result", "solution", "satisfaction" ], verbose=True ) return sequential_chain # 使用智能客服链 customer_service = IntelligentCustomerServiceChain() # 测试不同类型的问题 test_cases = [ "我的账户登录不了，提示密码错误", "我想了解你们的产品价格和套餐", "我收到的产品质量有问题，需要退货", "你们的技术支持电话是多少？" ] for i, question in enumerate(test_cases, 1): print(f"\n=== 测试案例 {i} ===") print(f"用户问题: {question}") # 使用顺序链处理 result = customer_service.process_request(question) print("\n处理结果:") print(f"意图分析: {result['intent_analysis']}") print(f"分类结果: {result['category']}") print(f"解决方案: {result['solution'][:200]}..." if len(result['solution']) > 200 else f"解决方案: {result['solution']}") print(f"满意度评估: {result['satisfaction']}") print("-" * 60)
运行结果示例：
=== 测试案例 1 === 用户问题: 我的账户登录不了，提示密码错误 处理结果: 意图分析: {'intent': '技术问题', 'confidence': 0.95} 分类结果: {'department': '技术支持', 'priority': '高'} 解决方案: 针对账户登录问题，我为您提供以下解决方案： 1. 密码重置 - 点击登录页面的"忘记密码"链接 - 输入注册邮箱或手机号 - 按照提示重置密码 - 重置后使用新密码登录 2. 检查输入 - 确认用户名和密码输入正确 - 检查大小写锁定键状态 - 确认没有多余的空格 - 尝试复制粘贴避免输入错误 3. 清除缓存 - 清除浏览器缓存和Cookie - 尝试使用无痕模式 - 换一个浏览器尝试登录 - 重启设备后重试 4. 账户状态检查 - 确认账户未被锁定 - 检查账户是否欠费 - 确认账户是否被暂停使用 - 联系客服确认账户状态 如果以上方法都无法解决，请联系技术支持团队： - 在线客服：客服中心提交工单 - 电话支持：400-123-4567 - 邮件支持：support@example.com 我们的技术支持团队会在24小时内回复您的问题。 满意度评估: {'score': 4, 'feedback': '解决方案全面，步骤清晰，覆盖了常见问题场景。建议增加更多故障排除指南。'} ------------------------------------------------------------
常见问题 FAQ
Q1：链式调用和传统函数调用有什么区别？
A：链式调用支持异步执行、自动数据传递、错误处理和状态管理，而传统函数调用需要手动传递参数和处理错误。链式调用更适合构建复杂的AI应用流程。
Q2：如何优化链式调用的性能？
A：使用缓存减少重复计算，并行执行无依赖的链，选择合适的模型温度参数，优化提示词减少处理时间，使用异步调用提高响应速度。
Q3：链式调用中的错误如何处理？
A：使用try-catch捕获异常，实现降级处理机制，添加重试逻辑，记录详细日志，提供用户友好的错误信息，建立错误恢复流程。
Q4：如何测试链式调用系统？
A：编写单元测试覆盖每个子链，集成测试验证整个流程，使用Mock数据测试边界条件，进行性能测试和压力测试，收集用户反馈进行优化。
最佳实践与避坑
- 模块化设计：将复杂逻辑拆分为独立的子链
- 数据验证：在每个链之间添加数据验证步骤
- 错误处理：为每个链实现适当的错误处理
- 日志记录：记录链式执行的详细日志
- 性能监控：监控每个链的执行时间和资源使用
- 版本管理：对链式系统进行版本控制和变更管理
本节小结
本节深入学习了LangChain链式调用的核心概念和实现方法，从基础的LLMChain到复杂的SequentialChain和RouterChain，以及自定义链的实现。链式调用是构建复杂AI应用工作流的关键技术，通过合理的链式设计可以实现高度模块化、可维护和可扩展的AI系统。下一节将介绍链式编程中的序列化与序列高级功能。
延伸阅读
- 官方文档：LangChain Chains官方文档v0.3版本
- 相关章节：本教程3.4节序列化与序列，3.1节提示词模板
关键词：LangChain框架精通, 链式调用, LLMChain, SequentialChain, RouterChain, 工作流编排, AI应用开发, 教程, 实战
难度：进阶
预计阅读：35分钟
评论区
(0)
U