跳到主要内容

Chains 链

Chains（链）是 LangChain 中的核心概念，它允许我们将多个组件串联起来，形成完整的工作流程。本章将详细介绍各种类型的 Chains 及其使用方法。

什么是 Chain？​

Chain 是 LangChain 中的核心抽象，用于将多个组件组合成一条处理流水线。每个组件（如 LLM、Prompt、Tool 等）称为链的"链接（Link）"，它们按顺序执行，数据在链接之间流动。

Chain 的工作原理​

一个简单的 Chain 示例：

# 组成：PromptTemplate → LLM → OutputParser
用户输入 → 格式化 Prompt → 发送给 LLM → 解析输出

LCEL 语法​

LangChain v1 引入了 LCEL（LangChain Expression Language），一种声明式的链式组合语法。

基础语法​

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 简单的管道组合
chain = (
    ChatPromptTemplate.from_template("{question}")  # 第一个链接
    | model                                          # 第二个链接
)

# 调用
response = chain.invoke({"question": "Python 是什么?"})
print(response.content)

为什么使用 LCEL？​

  • 声明式：更直观地表达数据流
  • 可组合：易于组合和复用
  • 流式输出：天然支持流式处理
  • 异步支持：内置异步支持

LLMChain​

最基本的链，用于将 Prompt 模板和 LLM 结合：

from langchain.chat_models import init_chat_model
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

# 初始化模型
model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 创建 Prompt 模板
prompt = PromptTemplate.from_template(
    "请用一句话解释 {concept} 的核心概念"
)

# 创建 LLMChain
chain = LLMChain(llm=model, prompt=prompt)

# 调用
result = chain.invoke({"concept": "Python 装饰器"})
print(result["text"])

完整参数​

from langchain.chains import LLMChain

chain = LLMChain(
    llm=model,                  # LLM 实例
    prompt=prompt,               # Prompt 模板
    output_key="answer",         # 输出键名（默认 "text"）
    verbose=True,                # 是否打印详细信息
)

转换链（Transformation Chain）​

用于对输入进行预处理：

from langchain.chains import TransformChain
from langchain.schema import HumanMessage

# 定义转换函数
def transform_func(input_dict):
    text = input_dict["text"]
    # 简单的文本处理：转大写
    return {"processed_text": text.upper()}

# 创建转换链
transform_chain = TransformChain(
    input_variables=["text"],
    output_variables=["processed_text"],
    transform=transform_func
)

# 测试
result = transform_chain.invoke({"text": "hello world"})
print(result)  # {'processed_text': 'HELLO WORLD'}

实际应用：清理用户输入​

import re

def clean_input(inputs):
    text = inputs["text"]
    # 移除多余空白
    text = re.sub(r'\s+', ' ', text).strip()
    # 移除特殊字符
    text = re.sub(r'[^\w\s\u4e00-\u9fff]', '', text)
    return {"cleaned_text": text}

clean_chain = TransformChain(
    input_variables=["text"],
    output_variables=["cleaned_text"],
    transform=clean_input
)

Sequential Chain​

将多个链按顺序执行：

from langchain.chains import LLMChain, SimpleSequentialChain
from langchain.prompts import PromptTemplate

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# Chain 1：提取关键词
chain1 = LLMChain(
    llm=model,
    prompt=PromptTemplate.from_template(
        "从以下文本中提取3个关键词：{text}"
    ),
    output_key="keywords"
)

# Chain 2：生成解释
chain2 = LLMChain(
    llm=model,
    prompt=PromptTemplate.from_template(
        "用一句话解释以下关键词的含义：{keywords}"
    ),
    output_key="explanation"
)

# 组合成顺序链
sequential_chain = SimpleSequentialChain(
    chains=[chain1, chain2],
    verbose=True
)

# 调用
result = sequential_chain.invoke("Python 装饰器是一种高级函数")
print(result["output"])

使用 LCEL 实现​

# 使用 LCEL 实现相同功能
chain = (
    {"text": lambda x: x["text"]}
    | chain1
    | {lambda x: x["keywords"]}
    | chain2
)

Router Chain​

根据输入动态选择不同的处理路径：

from langchain.chains import LLMRouterChain
from langchain.prompts import PromptTemplate

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 定义目的地
destinations = {
    "python": "请解释Python相关问题...",
    "java": "请解释Java相关问题...",
    "go": "请解释Go相关问题..."
}

# 创建路由提示词
router_template = """
根据用户问题，选择最合适的处理方式。

问题：{input}

可用类别：python, java, go

请只返回类别名称，不要其他内容。
"""

router_chain = LLMRouterChain.from_llm(
    model,
    prompt=PromptTemplate.from_template(router_template),
    destinations=destinations
)

RouterChain 完整示例​

from langchain.chains import MultiRouteChain

# 定义多个处理链
python_chain = LLMChain(
    llm=model,
    prompt=PromptTemplate.from_template("你是一个Python专家。回答：{input}")
)
java_chain = LLMChain(
    llm=model,
    prompt=PromptTemplate.from_template("你是一个Java专家。回答：{input}")
)

# 路由配置
route_config = {
    "python": python_chain,
    "java": java_chain,
    "default": python_chain  # 默认链
}

# 创建多路由链
multi_chain = MultiRouteChain(
    router_chain=router_chain,
    destination_chains=route_config,
    default_chain=python_chain
)

Map-Reduce Chain​

对多个输入并行处理，然后合并结果：

from langchain.chains import MapReduceChain
from langchain.text_splitter import RecursiveCharacterTextSplitter

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 文本分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=100
)

# Map 链：对每个chunk进行处理
map_template = "请总结以下内容：{chunk}"
map_chain = LLMChain(
    llm=model,
    prompt=PromptTemplate.from_template(map_template)
)

# Reduce 链：合并所有结果
reduce_template = "将以下摘要合并成一个完整的摘要：{summaries}"
reduce_chain = LLMChain(
    llm=model,
    prompt=PromptTemplate.from_template(reduce_template)
)

# Map-Reduce 链
map_reduce_chain = MapReduceChain(
    text_splitter=text_splitter,
    map_chain=map_chain,
    reduce_chain=reduce_chain
)

# 使用
long_text = "..."  # 长文本
result = map_reduce_chain.invoke(long_text)

Retrieval Chain​

用于基于检索的问答：

from langchain.chains import RetrievalQA
from langchain.retrievers import WikipediaRetriever

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 创建检索器
retriever = WikipediaRetriever()

# 创建问答链
qa_chain = RetrievalQA.from_chain_type(
    llm=model,
    retriever=retriever,
    chain_type="stuff"
)

# 提问
result = qa_chain.invoke("Python 装饰器是什么?")
print(result["result"])

不同的 Chain Type​

类型说明适用场景
stuff将所有相关文档拼接后输入文档数量少
map_reduce并行处理每个文档再合并大量文档
refine逐个处理，结果逐步优化需要逐步完善
map_rerank并行处理并选择最佳结果需要精准答案

自定义 Chain​

创建自定义 Chain​

from langchain.chains import LLMChain
from langchain.schema import BaseRetriever

class MyCustomChain(LLMChain):
    """自定义 Chain 示例"""
    
    @property
    def input_keys(self):
        return ["query", "context"]
    
    @property
    def output_keys(self):
        return ["answer", "sources"]
    
    def _call(self, inputs, run_manager):
        # 自定义处理逻辑
        context = inputs["context"]
        query = inputs["query"]
        
        # 调用 LLM
        response = self.llm.invoke(f"基于以下内容回答：{context}\n问题：{query}")
        
        return {
            "answer": response.content,
            "sources": context
        }

使用自定义 Chain​

custom_chain = MyCustomChain(
    llm=model,
    prompt=prompt
)

result = custom_chain.invoke({
    "query": "Python 有什么优势?",
    "context": "Python 是一种高级编程语言..."
})

实际应用示例​

1. 多步骤问答​

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# Step 1: 分析问题类型
analyze_template = ChatPromptTemplate.from_template("""
分析用户问题的类型：
问题：{question}

返回以下类型之一：fact, opinion, howto, other
""")

# Step 2: 生成回答
answer_template = ChatPromptTemplate.from_template("""
类型：{question_type}
问题：{question}

请给出恰当的回答
""")

# Step 3: 优化回答
refine_template = ChatPromptTemplate.from_template("""
原始回答：{answer}

请优化以上回答，使其更加完善
""")

# 组合链
chain = (
    {"question": lambda x: x["question"]}
    | {"question_type": (analyze_template | model | lambda x: x.content.strip())}
    | {"question": lambda x: x["question"], "question_type": lambda x: x["question_type"]}
    | answer_template
    | model
    | {"answer": lambda x: x.content}
    | {"answer": lambda x: x["answer"]}
    | refine_template
    | model
)

result = chain.invoke({"question": "如何学习Python?"})
print(result.content)

2. 数据处理管道​

# 场景：处理用户提交的数据，提取信息、验证、格式化

from langchain.chains import TransformChain, LLMChain
from langchain.prompts import PromptTemplate

# Step 1: 提取结构化数据
extract_chain = LLMChain(
    llm=model,
    prompt=PromptTemplate.from_template("""
从以下文本提取JSON格式的信息：
{text}

提取字段：姓名、年龄、职业
""")
)

# Step 2: 验证数据
def validate(inputs):
    data = inputs["data"]
    # 简单验证
    if "年龄" in data and int(data["年龄"]) < 0:
        return {"valid": False, "error": "年龄不能为负"}
    return {"valid": True, "data": data}

validate_chain = TransformChain(
    input_variables=["data"],
    output_variables=["valid", "error", "data"],
    transform=validate
)

# Step 3: 格式化输出
format_chain = LLMChain(
    llm=model,
    prompt=PromptTemplate.from_template("""
将以下JSON数据格式化输出为友好的文本：
{data}
""")
)

# 组合
pipeline = extract_chain | validate_chain | format_chain

result = pipeline.invoke({"text": "张三今年28岁，是一名软件工程师"})

Chain 调试​

使用 verbose​

chain = LLMChain(
    llm=model,
    prompt=prompt,
    verbose=True  # 打印详细执行信息
)

使用 LangSmith​

# 开启 LangSmith 追踪
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="ls-..."
from langsmith import traceable

@traceable
def my_chain(input_data):
    return chain.invoke(input_data)

下一步​

现在你已经掌握了 Chains 的核心概念，接下来学习：