第3章：链（Chains）基础

本章将带你深入了解 LangChain 最核心的概念——链（Chain）。学会使用链，你就能像搭积木一样组合各种组件，构建出强大的 AI 应用。

3.1 什么是 Chain

Chain（链）是 LangChain 框架的核心概念，它代表了一系列组件的有序组合，形成一个处理数据的管道（Pipeline）。

Chain 的基本结构

┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│   输入数据   │───▶│  组件处理1   │───▶│  组件处理2   │───▶│   输出结果   │
│   (Input)   │    │ (Component) │    │ (Component) │    │  (Output)   │
└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘

一个最简单的 Chain 通常包含：

• 输入（Input）：用户提供的原始数据或问题
• Prompt 模板：格式化输入，构建给 LLM 的指令
• LLM 模型：执行推理，生成回复
• 输出解析器（可选）：将模型输出转换为结构化数据
• 输出（Output）：最终返回给用户的结果

3.2 为什么使用 Chain

使用 Chain 有以下几个核心优势：

3.3 LLMChain（基础链）

LLMChain是最基础的链类型，它将 Prompt + LLM + OutputParser 组合在一起，是最常用的链模式。

LLMChain 的组成

┌─────────────────────────────────────────────────────────────┐
│                      LLMChain                                │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│   ┌──────────────┐    ┌──────────┐    ┌──────────────┐      │
│   │ PromptTemplate│───▶│   LLM    │───▶│OutputParser  │      │
│   │  (格式化输入) │    │ (推理)   │    │ (解析输出)   │      │
│   └──────────────┘    └──────────┘    └──────────────┘      │
│          │                                            │      │
│          ▼                                            ▼      │
│    {问题} ─────────────────────────────────────▶ {答案}      │
│                                                              │
└─────────────────────────────────────────────────────────────┘

完整代码示例

# 导入必要的库
from langchain import OpenAI, PromptTemplate, LLMChain
from langchain.output_parsers import CommaSeparatedListOutputParser
import os

# 设置 OpenAI API 密钥
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# ==================== 基础 LLMChain 示例 ====================

# 步骤1：创建 Prompt 模板
# 使用 {product} 作为占位符，运行时会被替换为实际值
template = """
你是一位专业的产品经理。请为以下产品起一个吸引人的中文名字，并写一段简短的宣传语。

产品描述：{product}

请按以下格式输出：
产品名：xxx
宣传语：xxx
"""

prompt = PromptTemplate(
    input_variables=["product"],  # 定义输入变量
    template=template             # 定义模板内容
)

# 步骤2：创建 LLM 实例
# temperature=0.7 控制创造性，值越高输出越多样
llm = OpenAI(temperature=0.7)

# 步骤3：创建 LLMChain
# 将 prompt 和 llm 组合成链
llm_chain = LLMChain(
    llm=llm,
    prompt=prompt
)

# 步骤4：运行链
# 传入实际的产品描述
product_description = "一款可以自动记录宠物健康状况和饮食情况的智能项圈"
result = llm_chain.predict(product=product_description)

print("=== 基础 LLMChain 结果 ===")
print(result)
print()

# ==================== 带输出解析器的 LLMChain ====================

# 步骤1：创建输出解析器
# 这个解析器会将逗号分隔的文本转换为 Python 列表
output_parser = CommaSeparatedListOutputParser()

# 步骤2：获取格式说明
# 输出解析器会自动生成格式说明，告诉 LLM 如何输出
format_instructions = output_parser.get_format_instructions()

# 步骤3：创建新的 Prompt 模板，包含格式说明
template_with_parser = """
请列出{topic}的5个主要优点。

{format_instructions}
"""

prompt_with_parser = PromptTemplate(
    input_variables=["topic"],
    partial_variables={"format_instructions": format_instructions},  # 预设格式说明
    template=template_with_parser
)

# 步骤4：创建带解析器的 LLMChain
chain_with_parser = LLMChain(
    llm=llm,
    prompt=prompt_with_parser,
    output_parser=output_parser  # 添加输出解析器
)

# 步骤5：运行并获取结构化输出
result_list = chain_with_parser.predict(topic="远程办公")

print("=== 带输出解析器的 LLMChain 结果 ===")
print(f"结果类型: {type(result_list)}")
print(f"结果内容: {result_list}")

# 现在 result_list 是一个 Python 列表，可以方便地处理
for i, item in enumerate(result_list, 1):
    print(f"{i}. {item}")

3.4 Sequential Chain（顺序链）

当需要多个步骤依次处理时，可以使用顺序链（Sequential Chain）。它允许你将多个链串联起来，前一个链的输出作为后一个链的输入。

SimpleSequentialChain：简单串联

SimpleSequentialChain 是最简单的顺序链，它按顺序执行多个链，每个链只有一个输入和一个输出，前一个输出自动作为后一个输入。

SimpleSequentialChain 执行流程：

输入 ──▶ Chain1 ──▶ 中间结果1 ──▶ Chain2 ──▶ 中间结果2 ──▶ Chain3 ──▶ 最终结果

每个链只有一个输入和一个输出，自动串联

from langchain import OpenAI, PromptTemplate, LLMChain, SimpleSequentialChain
import os

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

llm = OpenAI(temperature=0.7)

# ==================== Chain 1: 生成文章标题 ====================
template1 = """你是一位科技博主。请根据以下主题，创作一个吸引人的文章标题。

主题：{topic}

只输出标题，不要其他内容。"""

prompt1 = PromptTemplate(input_variables=["topic"], template=template1)
chain1 = LLMChain(llm=llm, prompt=prompt1, output_key="title")

# ==================== Chain 2: 根据标题写文章大纲 ====================
template2 = """你是一位资深编辑。请为以下标题创作一个详细的文章大纲。

标题：{title}

大纲格式：
一、引言
二、主要内容（分3-4点）
三、结论

只输出大纲。"""

prompt2 = PromptTemplate(input_variables=["title"], template=template2)
chain2 = LLMChain(llm=llm, prompt=prompt2, output_key="outline")

# ==================== Chain 3: 根据大纲写完整文章 ====================
template3 = """你是一位专业作家。请根据以下大纲，撰写一篇完整的文章。

大纲：
{outline}

要求：
1. 文章长度约800字
2. 语言流畅，逻辑清晰
3. 适合普通读者阅读

开始写作："""

prompt3 = PromptTemplate(input_variables=["outline"], template=template3)
chain3 = LLMChain(llm=llm, prompt=prompt3, output_key="article")

# ==================== 组合成 SimpleSequentialChain ====================
# 注意：SimpleSequentialChain 要求每个链只有一个输入和一个输出
# 输出会自动传递给下一个链的输入
overall_chain = SimpleSequentialChain(
    chains=[chain1, chain2, chain3],  # 按顺序执行
    verbose=True  # 显示执行过程，方便调试
)

# 运行链
result = overall_chain.run("人工智能对未来教育的影响")

print("\n=== 最终结果 ===")
print(result)

SequentialChain：多输入多输出

当需要处理多个输入和输出时，使用 SequentialChain。它可以明确指定每个链的输入和输出关系。

SequentialChain 执行流程：

输入A ──┐
       ├──▶ Chain1 ──┬──▶ 输出X ──┐
输入B ──┘              └──▶ 输出Y ──┼──▶ Chain2 ──▶ 最终结果
                                   │
                              输入C ──┘

支持多输入多输出，可以灵活组合

from langchain import OpenAI, PromptTemplate, LLMChain, SequentialChain
import os

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

llm = OpenAI(temperature=0.7)

# ==================== 实战示例：写文章 → 翻译 → 评价 ====================

# Chain 1: 根据主题和风格写文章
# 输入：topic, style
# 输出：article
template1 = """你是一位专业作家。请根据以下要求写一篇文章：

主题：{topic}
风格：{style}
字数：约500字

文章内容："""

prompt1 = PromptTemplate(
    input_variables=["topic", "style"], 
    template=template1
)
chain1 = LLMChain(
    llm=llm, 
    prompt=prompt1, 
    output_key="article"  # 明确指定输出名称
)

# Chain 2: 翻译文章
# 输入：article
# 输出：translation
template2 = """请将以下文章翻译成英文：

{article}

英文翻译："""

prompt2 = PromptTemplate(
    input_variables=["article"], 
    template=template2
)
chain2 = LLMChain(
    llm=llm, 
    prompt=prompt2, 
    output_key="translation"
)

# Chain 3: 评价文章质量
# 输入：article（同时使用原文）
# 输出：review
template3 = """请对以下文章进行评价：

原文：
{article}

请从以下几个方面评价：
1. 内容质量（1-10分）
2. 语言表达（1-10分）
3. 结构逻辑（1-10分）
4. 总体建议

评价："""

prompt3 = PromptTemplate(
    input_variables=["article"], 
    template=template3
)
chain3 = LLMChain(
    llm=llm, 
    prompt=prompt3, 
    output_key="review"
)

# Chain 4: 生成摘要
# 输入：article, translation
# 输出：summary
template4 = """请根据原文和英文翻译，生成一个简短的双语摘要。

原文摘要（50字以内）：
英文翻译摘要（30词以内）：

原文：
{article}

英文翻译：
{translation}

摘要："""

prompt4 = PromptTemplate(
    input_variables=["article", "translation"], 
    template=template4
)
chain4 = LLMChain(
    llm=llm, 
    prompt=prompt4, 
    output_key="summary"
)

# ==================== 组合成 SequentialChain ====================
overall_chain = SequentialChain(
    chains=[chain1, chain2, chain3, chain4],
    input_variables=["topic", "style"],           # 初始输入
    output_variables=["article", "translation", "review", "summary"],  # 最终输出
    verbose=True  # 显示执行过程
)

# 运行链
result = overall_chain({
    "topic": "远程工作的利与弊",
    "style": "客观理性"
})

# 输出各个结果
print("\n" + "="*50)
print("生成的文章：")
print("="*50)
print(result["article"])

print("\n" + "="*50)
print("英文翻译：")
print("="*50)
print(result["translation"])

print("\n" + "="*50)
print("文章评价：")
print("="*50)
print(result["review"])

print("\n" + "="*50)
print("双语摘要：")
print("="*50)
print(result["summary"])

3.5 Router Chain（路由链）

Router Chain 是 LangChain 中非常强大的功能，它可以根据输入内容自动路由到不同的子链进行处理。这非常适合构建智能客服、多意图识别等场景。

路由链的工作原理

Router Chain 执行流程：

用户输入
    │
    ▼
┌─────────────┐
│  路由链     │───▶ 分析输入，确定意图/类别
│ (Router)    │
└─────────────┘
    │
    ▼
选择对应的处理链
    │
    ├─▶ 技术问题 ──▶ 技术客服链
    ├─▶ 销售咨询 ──▶ 销售支持链
    ├─▶ 售后服务 ──▶ 售后处理链
    └─▶ 其他问题 ──▶ 通用客服链

根据路由结果，执行对应的子链

from langchain import OpenAI, PromptTemplate, LLMChain
from langchain.chains.router import MultiPromptChain
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
from langchain.chains import ConversationChain
import os

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

llm = OpenAI(temperature=0)

# ==================== 定义多个专业领域的 Prompt ====================

# 技术支持模板
tech_template = """你是某软件公司的技术支持专家。你擅长解决各种技术问题，
包括软件安装、配置、故障排查等。

请用专业但易懂的语言回答用户的技术问题。如果问题不够明确，请询问更多细节。

用户问题：{input}"""

# 销售咨询模板
sales_template = """你是某公司的销售顾问。你熟悉公司的各种产品和服务，
能够根据客户需求推荐合适的解决方案。

请热情、专业地回答客户的销售咨询，突出产品优势，同时诚实说明产品限制。

客户咨询：{input}"""

# 售后服务模板
service_template = """你是某公司的售后服务专员。你负责处理客户的退换货、
投诉、使用指导等售后问题。

请耐心、友善地处理客户问题，尽力让客户满意。对于投诉，要认真倾听并道歉。

客户反馈：{input}"""

# 通用咨询模板
general_template = """你是某公司的客服代表。你友好、专业，能够回答各种一般性问题。

如果用户问题需要转接给专业部门，请礼貌地告知用户并简要说明原因。

用户咨询：{input}"""

# 创建 Prompt 信息列表
prompt_infos = [
    {
        "name": "技术支持", 
        "description": "回答软件安装、配置、故障排查等技术问题",
        "prompt_template": tech_template
    },
    {
        "name": "销售咨询", 
        "description": "回答产品价格、功能、购买方式等销售相关问题",
        "prompt_template": sales_template
    },
    {
        "name": "售后服务", 
        "description": "处理退换货、投诉、使用指导等售后问题",
        "prompt_template": service_template
    },
    {
        "name": "通用咨询", 
        "description": "回答其他一般性问题",
        "prompt_template": general_template
    }
]

# ==================== 构建目标链 ====================
# 为每个 Prompt 创建一个 LLMChain

destination_chains = {}
for p_info in prompt_infos:
    name = p_info["name"]
    prompt = PromptTemplate(
        template=p_info["prompt_template"],
        input_variables=["input"]
    )
    chain = LLMChain(llm=llm, prompt=prompt)
    destination_chains[name] = chain

# 默认链：当路由无法确定时使用
default_chain = ConversationChain(llm=llm, output_key="text")

# ==================== 构建路由链 ====================

# 路由模板：告诉 LLM 如何选择
destinations = [f"{p['name']}: {p['description']}" for p in prompt_infos]
destinations_str = "\n".join(destinations)

router_template = """根据用户输入，选择最适合处理该问题的客服类型。

可选类型：
{destinations}

用户输入：{{input}}

请分析用户意图，输出一个 JSON 对象：
{{{{
    "destination": "选择的类型名称",
    "next_inputs": "处理后的输入"
}}}}

如果不确定，请输出 "DEFAULT" 作为 destination。"""

router_prompt = PromptTemplate(
    template=router_template.format(destinations=destinations_str),
    input_variables=["input"],
    output_parser=RouterOutputParser(),  # 解析路由决策
)

router_chain = LLMRouterChain.from_llm(llm, router_prompt)

# ==================== 组合成 MultiPromptChain ====================
chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=destination_chains,
    default_chain=default_chain,
    verbose=True  # 显示路由决策过程
)

# ==================== 测试不同场景 ====================

test_questions = [
    "我的软件安装到一半报错，错误代码 0x80070005，怎么办？",  # 技术问题
    "你们的企业版多少钱？有什么功能？",  # 销售问题
    "我想退货，订单号是 12345，可以吗？",  # 售后问题
    "你们公司成立多久了？",  # 通用问题
]

for question in test_questions:
    print("\n" + "="*60)
    print(f"用户问题：{question}")
    print("="*60)
    result = chain.run(question)
    print(f"\n回复：\n{result}")

3.6 LCEL（LangChain Expression Language）

LCEL 是 LangChain 表达式语言，它提供了一种更简洁、更直观的方式来组合链。使用管道操作符 |，你可以像写 Unix 管道命令一样构建链。

LCEL 的优势

• 简洁直观：代码更易读，像流水线一样清晰
• 自动类型推断：更好的 IDE 支持和类型检查
• 流式支持：更容易实现流式输出
• 并行执行：支持自动并行化
• 可观测性：更好的调试和日志支持

基础语法

# LCEL 基础语法
chain = component1 | component2 | component3

# 等价于传统写法
chain = Chain(components=[component1, component2, component3])

用 LCEL 重写链

from langchain import OpenAI, PromptTemplate
from langchain.output_parsers import CommaSeparatedListOutputParser
from langchain.schema import StrOutputParser
import os

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# ==================== 传统写法 vs LCEL 写法 ====================

# ---------------- 传统写法 ----------------
from langchain import LLMChain

template = "请列出{topic}的5个主要优点。"
prompt = PromptTemplate.from_template(template)
llm = OpenAI(temperature=0.7)

# 传统方式
chain_traditional = LLMChain(llm=llm, prompt=prompt)
result = chain_traditional.predict(topic="远程办公")

# ---------------- LCEL 写法 ----------------
from langchain.chat_models import ChatOpenAI

# 使用 ChatOpenAI 替代 OpenAI（推荐用于聊天场景）
chat_llm = ChatOpenAI(temperature=0.7)

# LCEL 写法：使用 | 管道符连接
chain_lcel = prompt | chat_llm

# 运行链
result = chain_lcel.invoke({"topic": "远程办公"})
print(result.content)  # ChatOpenAI 返回的是消息对象

# ==================== 完整的 LCEL 链 ====================

# 步骤1：定义 Prompt
prompt = PromptTemplate.from_template("""
请分析以下主题，列出5个关键点。

主题：{topic}

要求：简洁明了，每点不超过20字。
""")

# 步骤2：定义输出解析器
output_parser = StrOutputParser()  # 字符串输出解析器

# 步骤3：使用 LCEL 组合链
# prompt | llm | output_parser
chain = (
    prompt 
    | chat_llm 
    | output_parser  # 从消息对象中提取文本内容
)

# 步骤4：运行链
result = chain.invoke({"topic": "人工智能的应用"})
print("=== LCEL 链结果 ===")
print(result)

# ==================== 更复杂的 LCEL 示例 ====================

from langchain.schema.runnable import RunnableLambda

# 定义一个自定义处理函数
def add_header(text: str) -> str:
    """为结果添加标题"""
    return f"【分析结果】\n\n{text}"

def count_lines(text: str) -> str:
    """统计行数"""
    lines = text.strip().split('\n')
    return f"{text}\n\n共 {len(lines)} 行"

# 将普通函数转换为 Runnable
add_header_runnable = RunnableLambda(add_header)
count_lines_runnable = RunnableLambda(count_lines)

# 组合更复杂的链
complex_chain = (
    prompt 
    | chat_llm 
    | output_parser
    | add_header_runnable      # 添加自定义处理
    | count_lines_runnable     # 再添加一个处理
)

result = complex_chain.invoke({"topic": "机器学习"})
print("\n=== 复杂 LCEL 链结果 ===")
print(result)

LCEL 中的实用技巧

from langchain import PromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain.schema.runnable import RunnableParallel, RunnablePassthrough
from langchain.schema import StrOutputParser
import os

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

llm = ChatOpenAI(temperature=0.7)

# ==================== 技巧1: 传递原始输入 ====================
# 使用 RunnablePassthrough 保留原始输入

template = """根据用户问题生成回答。

用户问题：{question}
上下文：{question}  

回答："""

prompt = PromptTemplate.from_template(template)

# RunnablePassthrough 可以让输入原样通过
chain = (
    {"question": RunnablePassthrough()}  # 保留原始输入
    | prompt 
    | llm 
    | StrOutputParser()
)

result = chain.invoke("什么是深度学习？")
print(result)

# ==================== 技巧2: 并行执行 ====================
# 同时执行多个任务

# 定义两个并行的分析任务
sentiment_template = """分析以下文本的情感倾向（积极/消极/中性）：

{text}

情感："""

summary_template = """总结以下文本的主要内容（50字以内）：

{text}

总结："""

sentiment_prompt = PromptTemplate.from_template(sentiment_template)
summary_prompt = PromptTemplate.from_template(summary_template)

# 创建两个独立的链
sentiment_chain = sentiment_prompt | llm | StrOutputParser()
summary_chain = summary_prompt | llm | StrOutputParser()

# 使用 RunnableParallel 并行执行
parallel_chain = RunnableParallel(
    sentiment=sentiment_chain,  # 情感分析
    summary=summary_chain       # 内容总结
)

# 注意：这里需要为两个链提供相同的输入
text = """
人工智能正在深刻改变我们的生活方式。从智能手机的语音助手到自动驾驶汽车，
从医疗诊断到金融风控，AI技术无处不在。这些创新带来了巨大的便利，
但也引发了人们对隐私和就业的担忧。
"""

result = parallel_chain.invoke({"text": text})
print("\n=== 并行执行结果 ===")
print(f"情感分析：{result['sentiment']}")
print(f"内容总结：{result['summary']}")

# ==================== 技巧3: 条件分支（简单版） ====================

def route_by_length(data: dict) -> dict:
    """根据文本长度选择不同的处理方式"""
    text = data["text"]
    if len(text) < 100:
        return {"route": "short", "text": text}
    else:
        return {"route": "long", "text": text}

# 这种路由需要结合其他组件使用
# 实际项目中可以使用 RouterChain 或自定义 Runnable

3.7 自定义 Chain

当内置的链无法满足需求时，你可以创建自定义 Chain。这在需要特殊处理逻辑或与外部系统集成时非常有用。

from langchain.chains.base import Chain
from langchain import OpenAI, PromptTemplate
from typing import Dict, List
import os

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# ==================== 创建自定义 Chain ====================

class ValidationChain(Chain):
    """
    自定义验证链：先验证输入，再决定是否调用 LLM
    """
    
    # 定义链使用的组件
    llm: OpenAI = None
    prompt: PromptTemplate = None
    
    # 验证规则
    min_length: int = 5
    max_length: int = 500
    forbidden_words: List[str] = None
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.forbidden_words = self.forbidden_words or ["敏感词1", "敏感词2"]
        if self.llm is None:
            self.llm = OpenAI(temperature=0.7)
    
    @property
    def input_keys(self) -> List[str]:
        """定义输入变量"""
        return ["query"]
    
    @property
    def output_keys(self) -> List[str]:
        """定义输出变量"""
        return ["response", "validation_passed", "validation_message"]
    
    def _validate(self, query: str) -> tuple:
        """
        验证输入
        返回: (是否通过, 错误信息)
        """
        # 检查长度
        if len(query) < self.min_length:
            return False, f"输入太短，至少需要 {self.min_length} 个字符"
        
        if len(query) > self.max_length:
            return False, f"输入太长，最多 {self.max_length} 个字符"
        
        # 检查敏感词
        for word in self.forbidden_words:
            if word in query:
                return False, f"输入包含敏感词: {word}"
        
        return True, "验证通过"
    
    def _call(self, inputs: Dict[str, str]) -> Dict[str, str]:
        """
        执行链的核心逻辑
        """
        query = inputs["query"]
        
        # 步骤1: 验证输入
        is_valid, message = self._validate(query)
        
        if not is_valid:
            # 验证失败，返回错误信息
            return {
                "response": "",
                "validation_passed": "False",
                "validation_message": message
            }
        
        # 步骤2: 调用 LLM（如果没有设置 prompt，使用默认处理）
        if self.prompt:
            formatted_prompt = self.prompt.format(query=query)
            response = self.llm.predict(formatted_prompt)
        else:
            response = self.llm.predict(f"请回答以下问题：{query}")
        
        return {
            "response": response,
            "validation_passed": "True",
            "validation_message": message
        }

# ==================== 使用自定义 Chain ====================

# 创建验证链实例
validation_chain = ValidationChain(
    min_length=3,
    max_length=200,
    forbidden_words=["垃圾", "诈骗"]
)

# 测试验证通过的情况
print("=== 测试1: 正常输入 ===")
result = validation_chain.run({"query": "请介绍一下Python语言"})
print(f"验证结果: {result}")

# 测试验证失败的情况（太短）
print("\n=== 测试2: 输入太短 ===")
result = validation_chain.run({"query": "Hi"})
print(f"验证结果: {result}")

# 测试验证失败的情况（包含敏感词）
print("\n=== 测试3: 包含敏感词 ===")
result = validation_chain.run({"query": "这是一个垃圾信息"})
print(f"验证结果: {result}")

# ==================== 更复杂的自定义 Chain ====================

class MultiStepChain(Chain):
    """
    多步骤处理链：先分析问题，再生成回答，最后格式化输出
    """
    
    llm: OpenAI = None
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        if self.llm is None:
            self.llm = OpenAI(temperature=0.7)
    
    @property
    def input_keys(self) -> List[str]:
        return ["question"]
    
    @property
    def output_keys(self) -> List[str]:
        return ["analysis", "answer", "formatted_response"]
    
    def _call(self, inputs: Dict[str, str]) -> Dict[str, str]:
        question = inputs["question"]
        
        # 步骤1: 分析问题
        analysis_prompt = f"""请分析以下问题的关键点和用户的潜在需求：

问题：{question}

分析："""
        analysis = self.llm.predict(analysis_prompt)
        
        # 步骤2: 基于分析生成回答
        answer_prompt = f"""基于以下分析，详细回答用户的问题。

问题：{question}
分析：{analysis}

请提供详细、准确的回答："""
        answer = self.llm.predict(answer_prompt)
        
        # 步骤3: 格式化输出
        formatted = f"""【问题分析】
{analysis}

【详细回答】
{answer}

---
希望以上回答对您有帮助！"""
        
        return {
            "analysis": analysis,
            "answer": answer,
            "formatted_response": formatted
        }

# 使用多步骤链
multi_chain = MultiStepChain()
print("\n=== 多步骤链测试 ===")
result = multi_chain({"question": "如何提高编程能力？"})
print(result["formatted_response"])

3.8 完整实战：构建智能客服链

现在我们将综合运用本章所学知识，构建一个完整的智能客服链。

需求分析

• 用户输入问题 → 系统自动分类（技术/销售/售后/其他）
• 根据分类 → 路由到对应的处理链
• 生成专业回答 → 格式化输出
• 记录处理日志 → 便于后续分析

"""
智能客服系统 - 完整实战
功能：
1. 问题分类
2. 路由到专业客服
3. 生成回答
4. 格式化输出
5. 记录日志
"""

from langchain import OpenAI, PromptTemplate
from langchain.chains import LLMChain, SequentialChain, TransformChain
from langchain.output_parsers import CommaSeparatedListOutputParser
import os
from datetime import datetime
import json

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# ==================== 配置 ====================
llm = OpenAI(temperature=0.5)

# ==================== Chain 1: 问题分类器 ====================
# 输入：user_question
# 输出：category（技术/销售/售后/其他）

classify_template = """你是智能客服系统的分类器。请分析用户问题，将其分类为以下类别之一：

可选类别：
- 技术问题：软件安装、配置、故障排查、功能使用等技术相关问题
- 销售咨询：产品价格、功能对比、购买方式、授权许可等销售相关问题
- 售后服务：退换货、退款、投诉、账户问题等售后相关问题
- 其他问题：不属于以上类别的一般性问题

用户问题：{user_question}

请只输出类别名称（技术问题/销售咨询/售后服务/其他问题），不要其他内容。
类别："""

classify_prompt = PromptTemplate(
    input_variables=["user_question"],
    template=classify_template
)

classify_chain = LLMChain(
    llm=llm,
    prompt=classify_prompt,
    output_key="category"
)

# ==================== Chain 2: 生成专业回答 ====================
# 输入：user_question, category
# 输出：raw_response

def get_specialist_prompt(category):
    """根据类别获取专业 Prompt"""
    prompts = {
        "技术问题": """你是一位资深技术支持工程师。请专业、耐心地回答用户的技术问题。
如果问题信息不足，请礼貌地询问更多细节。

用户问题：{user_question}

请提供详细的解决方案：""",
        
        "销售咨询": """你是一位热情的销售顾问。请专业地介绍产品优势，同时诚实说明产品限制。
突出产品如何满足客户需求。

客户咨询：{user_question}

请提供详细的回复：""",
        
        "售后服务": """你是一位耐心的售后服务专员。请认真倾听客户问题，尽力解决。
对于投诉，要真诚道歉并提供补救方案。

客户反馈：{user_question}

请提供处理方案：""",
        
        "其他问题": """你是一位友好的客服代表。请热情、专业地回答用户问题。
如果问题需要转接其他部门，请说明原因。

用户咨询：{user_question}

请提供回复："""
    }
    return prompts.get(category, prompts["其他问题"])

# 使用动态 Prompt 的 Chain
response_template = """根据以下类别处理用户问题：

类别：{category}
用户问题：{user_question}

{specialist_instructions}"""

# 这里我们使用一个转换链来根据类别选择提示词
def generate_response(inputs):
    """根据分类生成专业回答"""
    category = inputs["category"].strip()
    question = inputs["user_question"]
    
    # 获取对应类别的 Prompt
    specialist_template = get_specialist_prompt(category)
    specialist_prompt = PromptTemplate(
        input_variables=["user_question"],
        template=specialist_template
    )
    
    # 创建临时 Chain 生成回答
    response_chain = LLMChain(llm=llm, prompt=specialist_prompt)
    response = response_chain.predict(user_question=question)
    
    return {"raw_response": response}

response_transform_chain = TransformChain(
    input_variables=["user_question", "category"],
    output_variables=["raw_response"],
    transform=generate_response
)

# ==================== Chain 3: 格式化输出 ====================
# 输入：user_question, category, raw_response
# 输出：formatted_response

format_template = """请将以下客服回复整理成标准格式：

原始回复：
{raw_response}

请按以下格式输出（不要包含思考过程）：

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
【客服回复】

{raw_response}

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
问题类型：{category}
回复时间：{timestamp}
如有其他问题，欢迎继续咨询！
"""

format_prompt = PromptTemplate(
    input_variables=["raw_response", "category"],
    template=format_template
)

# 使用当前时间
from langchain.chains import TransformChain

def add_timestamp(inputs):
    """添加时间戳"""
    inputs["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return inputs

timestamp_chain = TransformChain(
    input_variables=["raw_response", "category"],
    output_variables=["raw_response", "category", "timestamp"],
    transform=add_timestamp
)

format_chain = LLMChain(
    llm=llm,
    prompt=format_prompt,
    output_key="formatted_response"
)

# ==================== 组合成完整客服链 ====================
customer_service_chain = SequentialChain(
    chains=[classify_chain, response_transform_chain, timestamp_chain, format_chain],
    input_variables=["user_question"],
    output_variables=["category", "raw_response", "timestamp", "formatted_response"],
    verbose=True
)

# ==================== 日志记录功能 ====================

def log_interaction(question, result):
    """记录客服交互日志"""
    log_entry = {
        "timestamp": result["timestamp"],
        "question": question,
        "category": result["category"],
        "response": result["raw_response"]
    }
    
    # 写入日志文件
    with open("customer_service_log.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
    
    print(f"\n[日志已记录] 类别: {result['category']}")

# ==================== 运行测试 ====================

test_questions = [
    "软件安装时出现错误代码 0x80070005，如何解决？",
    "企业版和个人版有什么区别？价格分别是多少？",
    "我购买的软件激活失败，订单号是 12345，能退款吗？",
    "你们公司的地址在哪里？"
]

print("="*70)
print("智能客服系统 - 开始测试")
print("="*70)

for question in test_questions:
    print(f"\n{'='*70}")
    print(f"用户问题：{question}")
    print(f"{'='*70}")
    
    # 运行客服链
    result = customer_service_chain({"user_question": question})
    
    # 显示格式化输出
    print(f"\n{result['formatted_response']}")
    
    # 记录日志
    log_interaction(question, result)

print("\n" + "="*70)
print("测试完成！日志已保存到 customer_service_log.jsonl")
print("="*70)

# ==================== 统计功能 ====================

def show_statistics():
    """显示客服统计信息"""
    try:
        with open("customer_service_log.jsonl", "r", encoding="utf-8") as f:
            logs = [json.loads(line) for line in f if line.strip()]
        
        # 统计各类问题数量
        from collections import Counter
        categories = [log["category"] for log in logs]
        category_count = Counter(categories)
        
        print("\n" + "="*70)
        print("客服数据统计")
        print("="*70)
        print(f"总咨询量：{len(logs)}")
        print("\n问题分类统计：")
        for category, count in category_count.most_common():
            print(f"  - {category}: {count}次")
        print("="*70)
        
    except FileNotFoundError:
        print("暂无日志记录")

# 显示统计
show_statistics()

本章小结