Sequential Chain详解 - 元素码农

发布时间: 2025-03-29 19:07

↑

# Sequential Chain详解

本文将详细介绍LangChain中的Sequential Chain，帮助你理解如何构建和使用顺序链来处理复杂的任务流程。

## Sequential Chain简介

Sequential Chain是一种特殊的Chain类型，它允许我们将多个Chain按照特定顺序串联起来，形成一个处理流水线。每个Chain的输出可以作为下一个Chain的输入。

## Chain类型

### 1. SimpleSequentialChain

最简单的顺序链，每个Chain只有一个输入和一个输出：

```python
from langchain.chains import SimpleSequentialChain, LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

# 创建第一个Chain：生成故事大纲
outline_template = """请为一个{theme}主题的故事写一个简单大纲"""
outline_prompt = PromptTemplate(input_variables=["theme"], template=outline_template)
outline_chain = LLMChain(llm=OpenAI(), prompt=outline_prompt)

# 创建第二个Chain：根据大纲写故事
story_template = """根据以下大纲写一个完整的故事：
{outline}
"""
story_prompt = PromptTemplate(input_variables=["outline"], template=story_template)
story_chain = LLMChain(llm=OpenAI(), prompt=story_prompt)

# 创建顺序链
overall_chain = SimpleSequentialChain(
    chains=[outline_chain, story_chain],
    verbose=True
)

# 运行链
result = overall_chain.run("科幻")
print(result)
```

### 2. SequentialChain

更灵活的顺序链，支持多输入多输出：

```python
from langchain.chains import SequentialChain

# 创建第一个Chain：分析产品特点
feature_template = """分析{product}的主要特点和优势"""
feature_prompt = PromptTemplate(input_variables=["product"], template=feature_template)
feature_chain = LLMChain(
    llm=OpenAI(),
    prompt=feature_prompt,
    output_key="features"
)

# 创建第二个Chain：生成营销文案
marketing_template = """根据以下特点，为{product}写一段营销文案：
{features}

目标受众：{audience}
"""
marketing_prompt = PromptTemplate(
    input_variables=["product", "features", "audience"],
    template=marketing_template
)
marketing_chain = LLMChain(
    llm=OpenAI(),
    prompt=marketing_prompt,
    output_key="marketing_copy"
)

# 创建顺序链
chain = SequentialChain(
    chains=[feature_chain, marketing_chain],
    input_variables=["product", "audience"],
    output_variables=["features", "marketing_copy"],
    verbose=True
)

# 运行链
result = chain({
    "product": "智能手表",
    "audience": "年轻运动人群"
})
print("产品特点：", result["features"])
print("营销文案：", result["marketing_copy"])
```

## 高级用法

### 1. 条件分支

```python
from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate

def create_branching_chain():
    # 分析情感Chain
    sentiment_template = """分析以下文本的情感倾向（正面/负面）：{text}"""
    sentiment_prompt = PromptTemplate(
        input_variables=["text"],
        template=sentiment_template
    )
    sentiment_chain = LLMChain(
        llm=OpenAI(),
        prompt=sentiment_prompt,
        output_key="sentiment"
    )
    
    # 根据情感选择回复模板
    def select_response_template(sentiment):
        if "正面" in sentiment:
            return "感谢您的好评！{text}"
        else:
            return "抱歉给您带来不好的体验。{text}"
    
    # 生成回复Chain
    response_chain = LLMChain(
        llm=OpenAI(),
        prompt=PromptTemplate(
            input_variables=["text", "sentiment"],
            template=lambda x: select_response_template(x["sentiment"])
        ),
        output_key="response"
    )
    
    # 组合Chain
    return SequentialChain(
        chains=[sentiment_chain, response_chain],
        input_variables=["text"],
        output_variables=["sentiment", "response"]
    )
```

### 2. 错误处理

```python
from typing import Dict, Any

def safe_run_sequential_chain(chain, inputs: Dict[str, Any]) -> Dict[str, Any]:
    try:
        return chain(inputs)
    except Exception as e:
        print(f"Chain执行错误：{e}")
        return {
            output_key: None
            for output_key in chain.output_variables
        }
```

### 3. 中间结果处理

```python
from langchain.chains import TransformChain

def create_processing_chain():
    # 数据处理Chain
    def process_data(inputs: dict) -> dict:
        text = inputs["text"].lower()  # 转小写
        word_count = len(text.split())  # 计算词数
        return {
            "processed_text": text,
            "word_count": word_count
        }
    
    process_chain = TransformChain(
        input_variables=["text"],
        output_variables=["processed_text", "word_count"],
        transform=process_data
    )
    
    # 主Chain
    main_chain = LLMChain(
        llm=OpenAI(),
        prompt=PromptTemplate(
            input_variables=["processed_text", "word_count"],
            template="处理后的文本({word_count}词)：{processed_text}"
        )
    )
    
    # 组合Chain
    return SequentialChain(
        chains=[process_chain, main_chain],
        input_variables=["text"],
        output_variables=["processed_text", "word_count", "text"]
    )
```

## 应用示例

### 1. 文档处理流水线

```python
from langchain.chains import SequentialChain
from langchain.chains.summarize import load_summarize_chain

def create_document_pipeline():
    # 摘要Chain
    summarize_chain = load_summarize_chain(
        llm=OpenAI(),
        chain_type="map_reduce",
        output_key="summary"
    )
    
    # 翻译Chain
    translate_chain = LLMChain(
        llm=OpenAI(),
        prompt=PromptTemplate(
            input_variables=["summary"],
            template="将以下文本翻译成中文：{summary}"
        ),
        output_key="translation"
    )
    
    # 关键词提取Chain
    keyword_chain = LLMChain(
        llm=OpenAI(),
        prompt=PromptTemplate(
            input_variables=["translation"],
            template="从以下文本中提取关键词：{translation}"
        ),
        output_key="keywords"
    )
    
    # 组合Chain
    return SequentialChain(
        chains=[summarize_chain, translate_chain, keyword_chain],
        input_variables=["input_documents"],
        output_variables=["summary", "translation", "keywords"]
    )
```

### 2. 对话生成器

```python
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

def create_conversation_generator():
    # 主题生成Chain
    topic_chain = LLMChain(
        llm=OpenAI(),
        prompt=PromptTemplate(
            input_variables=["context"],
            template="根据上下文生成一个有趣的对话主题：{context}"
        ),
        output_key="topic"
    )
    
    # 对话生成Chain
    conversation_chain = ConversationChain(
        llm=OpenAI(),
        memory=ConversationBufferMemory(),
        output_key="dialogue"
    )
    
    # 组合Chain
    return SequentialChain(
        chains=[topic_chain, conversation_chain],
        input_variables=["context"],
        output_variables=["topic", "dialogue"]
    )
```

## 总结

Sequential Chain提供了强大的任务流程处理能力：
- 支持简单和复杂的链式处理
- 灵活的输入输出控制
- 条件分支支持
- 错误处理机制
- 中间结果处理

掌握Sequential Chain的使用，将帮助你构建更复杂和强大的AI应用。