Chains 链

Chains（链）是 LangChain 中组合多个组件的核心方式。在 LangChain v1 中，链的构建主要通过 LCEL（LangChain Expression Language） 实现，这是一种声明式的链式组合语法，让代码更简洁、更易维护。

什么是 Chain？

Chain 是将多个处理步骤串联成一个完整工作流的抽象。每个步骤接收输入、进行处理、产生输出，输出成为下一步的输入。

传统方式 vs LCEL

在 LangChain v1 之前，构建链需要使用各种 Chain 类（如 LLMChain、SimpleSequentialChain 等），这些类已被移至 langchain-classic 包。现在推荐使用 LCEL：

# 旧方式（已废弃，需要 langchain-classic）
from langchain.chains import LLMChain
chain = LLMChain(llm=model, prompt=prompt)

# 新方式（推荐）- 使用 LCEL
chain = prompt | model | output_parser

LCEL 的优势：

声明式语法：用管道操作符 | 直观表达数据流向
自动支持流式输出：无需额外配置即可流式返回结果
自动支持异步：所有同步方法都有对应的异步版本
内置批处理优化：并行处理多个输入时自动优化
易于组合：链可以嵌套、分支、合并

Runnable 接口

LCEL 的核心是 Runnable 接口。所有实现了这个接口的组件（模型、提示模板、输出解析器等）都可以用管道操作符连接。

Runnable 核心方法

方法	说明
`invoke(input)`	同步执行，返回单个输出
`stream(input)`	同步流式执行，返回输出迭代器
`batch([inputs])`	批量执行，返回输出列表
`ainvoke(input)`	异步执行
`astream(input)`	异步流式执行
`abatch([inputs])`	异步批量执行

基本用法

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

# 创建模型（使用统一的初始化函数）
model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 创建提示模板
prompt = ChatPromptTemplate.from_template("用一句话解释 {concept}")

# 使用 LCEL 组合成链
chain = prompt | model

# 同步调用
response = chain.invoke({"concept": "量子计算"})
print(response.content)

# 流式输出
for chunk in chain.stream({"concept": "机器学习"}):
    print(chunk.content, end="", flush=True)

# 批量处理
results = chain.batch([
    {"concept": "Python"},
    {"concept": "Java"},
    {"concept": "Go"}
])
for result in results:
    print(result.content)

解释：这里 prompt 和 model 都是 Runnable 对象。管道操作符 | 将它们连接成一个 RunnableSequence，数据从左到右流动：输入 → prompt 格式化 → model 生成 → 输出。

管道操作符

管道操作符 | 是 LCEL 的核心语法，它将两个 Runnable 连接成一个新的 Runnable。

工作原理

# 这两行等价
chain = runnable_a | runnable_b
chain = RunnableSequence([runnable_a, runnable_b])

当调用 chain.invoke(input) 时：

runnable_a.invoke(input) 得到 output_a
runnable_b.invoke(output_a) 得到 output_b
返回 output_b

多步骤链

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

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

# 三步链：提示 → 模型 → 解析
chain = (
    ChatPromptTemplate.from_template("将以下文本翻译成中文：{text}")
    | model
    | StrOutputParser()  # 提取字符串内容
)

result = chain.invoke({"text": "Hello, World!"})
print(result)  # 输出：你好，世界！

StrOutputParser 从 AIMessage 中提取文本内容，返回纯字符串。

RunnableParallel

当你需要将输入同时传递给多个组件，然后收集所有结果时，使用 RunnableParallel。

使用字典语法

在链中使用字典会自动创建 RunnableParallel：

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

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

# 并行执行：同时翻译成多种语言
translate_chain = ChatPromptTemplate.from_template(
    "将 '{text}' 翻译成{language}，只返回翻译结果"
) | model

# 并行调用
parallel_chain = {
    "chinese": translate_chain,
    "japanese": translate_chain,
    "korean": translate_chain
}

# 组合成完整链
chain = (
    ChatPromptTemplate.from_template("{text}")
    | {
        "chinese": ChatPromptTemplate.from_template(
            "将 '{text}' 翻译成中文，只返回翻译结果"
        ) | model | StrOutputParser(),
        "japanese": ChatPromptTemplate.from_template(
            "将 '{text}' 翻译成日语，只返回翻译结果"
        ) | model | StrOutputParser(),
        "korean": ChatPromptTemplate.from_template(
            "将 '{text}' 翻译成韩语，只返回翻译结果"
        ) | model | StrOutputParser(),
    }
)

result = chain.invoke({"text": "Good morning"})
# result = {
#     "chinese": "早上好",
#     "japanese": "おはようございます",
#     "korean": "좋은 아침"
# }

显式创建 RunnableParallel

from langchain_core.runnables import RunnableParallel

# 显式创建
parallel = RunnableParallel(
    summary=ChatPromptTemplate.from_template("总结：{text}") | model,
    keywords=ChatPromptTemplate.from_template("提取关键词：{text}") | model,
    sentiment=ChatPromptTemplate.from_template("分析情感：{text}") | model
)

chain = parallel
result = chain.invoke({"text": "今天天气真不错，阳光明媚！"})

解释：RunnableParallel 会将相同的输入同时传递给所有子 Runnable，然后收集结果到一个字典中。这在需要并行分析、多角度处理时非常有用。

RunnablePassthrough

RunnablePassthrough 是一个特殊的 Runnable，它直接传递输入，不做任何修改。常用于在链中保留原始数据。

基本用法

from langchain_core.runnables import RunnablePassthrough

# 直接传递输入
passthrough = RunnablePassthrough()
result = passthrough.invoke({"a": 1, "b": 2})
# result = {"a": 1, "b": 2}

在链中保留原始数据

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

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

# 需求：生成解释并保留原始问题
chain = (
    # 并行：保留原始问题，同时生成解释
    {
        "question": RunnablePassthrough(),  # 原样传递
        "answer": ChatPromptTemplate.from_template(
            "简要回答：{question}"
        ) | model | StrOutputParser()
    }
)

result = chain.invoke("什么是递归？")
# result = {
#     "question": "什么是递归？",
#     "answer": "递归是函数调用自身的编程技术..."
# }

RunnablePassthrough.assign

动态添加新字段：

from langchain_core.runnables import RunnablePassthrough

# 在现有数据上添加新字段
chain = RunnablePassthrough.assign(
    # 添加一个计算字段
    length=lambda x: len(x["text"]),
    # 添加一个转换字段
    upper=lambda x: x["text"].upper()
)

result = chain.invoke({"text": "hello"})
# result = {"text": "hello", "length": 5, "upper": "HELLO"}

实际应用：在 RAG 场景中，同时保留问题和检索到的文档：

# RAG 链示例
rag_chain = (
    {
        "context": retriever,  # 检索相关文档
        "question": RunnablePassthrough()  # 保留原始问题
    }
    | ChatPromptTemplate.from_template("""
基于以下上下文回答问题：
上下文：{context}
问题：{question}
""")
    | model
    | StrOutputParser()
)

RunnableLambda

RunnableLambda 将普通 Python 函数包装成 Runnable，使其可以参与链式调用。

基本用法

from langchain_core.runnables import RunnableLambda

# 使用函数创建 Runnable
def to_uppercase(text: str) -> str:
    return text.upper()

upper_runnable = RunnableLambda(to_uppercase)

# 直接调用
result = upper_runnable.invoke("hello")
# result = "HELLO"

# 在链中使用
chain = RunnableLambda(lambda x: x.lower()) | RunnableLambda(lambda x: x + "!")
result = chain.invoke("HELLO")
# result = "hello!"

在复杂链中使用

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda

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

def format_output(response):
    """格式化模型输出"""
    return f"【AI回复】{response.content}"

def add_metadata(result):
    """添加元数据"""
    return {
        "content": result,
        "length": len(result),
        "timestamp": "2024-01-01"
    }

chain = (
    ChatPromptTemplate.from_template("用一句话解释：{topic}")
    | model
    | RunnableLambda(format_output)
    | RunnableLambda(add_metadata)
)

result = chain.invoke({"topic": "区块链"})
# result = {
#     "content": "【AI回复】区块链是一种分布式账本技术...",
#     "length": 30,
#     "timestamp": "2024-01-01"
# }

解释：RunnableLambda 让你可以在链中插入任意 Python 逻辑，非常适合数据转换、格式化、验证等操作。

组合模式

顺序链（Sequential Chain）

将多个步骤按顺序执行，前一步的输出是后一步的输入：

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

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

# 三步顺序链
chain = (
    # Step 1: 生成大纲
    ChatPromptTemplate.from_template("为'{topic}'生成一个简短的大纲")
    | model
    | StrOutputParser()
    # Step 2: 扩展内容
    | (lambda outline: {"outline": outline})
    | ChatPromptTemplate.from_template("基于大纲扩展内容：{outline}")
    | model
    | StrOutputParser()
    # Step 3: 添加总结
    | (lambda content: {"content": content})
    | ChatPromptTemplate.from_template("为以下内容添加一句总结：{content}")
    | model
    | StrOutputParser()
)

result = chain.invoke({"topic": "人工智能"})

分支链（Branching）

根据条件选择不同的处理路径：

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch

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

# 定义不同的处理链
tech_chain = ChatPromptTemplate.from_template(
    "作为技术专家回答：{question}"
) | model

general_chain = ChatPromptTemplate.from_template(
    "简洁回答：{question}"
) | model

detailed_chain = ChatPromptTemplate.from_template(
    "详细解释：{question}"
) | model

# 创建分支
def route_question(input_dict):
    """根据问题类型路由"""
    question = input_dict["question"].lower()
    if any(word in question for word in ["代码", "编程", "算法"]):
        return "tech"
    elif any(word in question for word in ["是什么", "为什么"]):
        return "detailed"
    else:
        return "general"

branch_chain = RunnableBranch(
    (lambda x: route_question(x) == "tech", tech_chain),
    (lambda x: route_question(x) == "detailed", detailed_chain),
    general_chain  # 默认分支
)

chain = (
    {"question": RunnablePassthrough()}
    | branch_chain
    | StrOutputParser()
)

result = chain.invoke("什么是 Python 装饰器？")

循环处理（Map-Reduce）

对列表中的每个元素应用相同的处理：

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda

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

# 处理单个项目的链
process_chain = (
    ChatPromptTemplate.from_template("用一句话总结：{text}")
    | model
    | StrOutputParser()
)

# 使用 map 对列表进行处理
def process_list(items):
    """处理列表中的每一项"""
    results = []
    for item in items:
        result = process_chain.invoke({"text": item})
        results.append(result)
    return results

# 或者使用 batch 方法更高效
chain = RunnableLambda(lambda items: process_chain.batch([{"text": item} for item in items]))

texts = [
    "Python 是一种高级编程语言，以其简洁的语法著称。",
    "JavaScript 是网页开发的核心语言。",
    "Rust 是一种注重安全和性能的系统编程语言。"
]

summaries = chain.invoke(texts)
# summaries = [
#     "Python 是语法简洁的高级编程语言。",
#     "JavaScript 是网页开发核心语言。",
#     "Rust 是注重安全性能的系统语言。"
# ]

Runnable 高级模式

LCEL 提供了多种高级模式来处理复杂的业务场景，包括条件分支、重试、回退等。

RunnableBranch 条件分支

RunnableBranch 根据条件选择不同的处理路径，类似于编程中的 if-elif-else 结构：

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch
from langchain.output_parsers import StrOutputParser

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

# 定义不同类型的处理链
tech_chain = (
    ChatPromptTemplate.from_template("作为技术专家回答：{question}")
    | model
    | StrOutputParser()
)

creative_chain = (
    ChatPromptTemplate.from_template("用创意的方式回答：{question}")
    | model
    | StrOutputParser()
)

general_chain = (
    ChatPromptTemplate.from_template("简洁回答：{question}")
    | model
    | StrOutputParser()
)

# 定义路由函数
def route_by_topic(input_dict):
    """根据问题类型路由到不同的链"""
    question = input_dict["question"].lower()
    
    # 技术类问题关键词
    tech_keywords = ["代码", "编程", "算法", "python", "java", "api", "函数", "调试"]
    # 创意类问题关键词
    creative_keywords = ["故事", "创意", "想象", "设计", "创作", "有趣"]
    
    if any(keyword in question for keyword in tech_keywords):
        return tech_chain
    elif any(keyword in question for keyword in creative_keywords):
        return creative_chain
    else:
        return general_chain

# 方式一：使用 RunnableBranch
branch = RunnableBranch(
    (lambda x: "代码" in x["question"].lower() or "编程" in x["question"].lower(), tech_chain),
    (lambda x: "故事" in x["question"].lower() or "创意" in x["question"].lower(), creative_chain),
    general_chain  # 默认分支
)

# 使用
result = branch.invoke({"question": "Python 如何写一个快速排序？"})
# 会使用 tech_chain

result = branch.invoke({"question": "给我讲个有趣的故事"})
# 会使用 creative_chain

result = branch.invoke({"question": "今天天气怎么样？"})
# 会使用 general_chain

RunnableBranch 的工作原理：

按顺序检查每个条件（第一个参数）
如果条件返回 True，执行对应的 Runnable（第二个参数）
如果所有条件都不满足，执行默认的 Runnable（最后一个参数）

RunnableRetry 重试机制

RunnableRetry 为链添加自动重试能力，处理临时性错误：

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableRetry
from langchain.output_parsers import StrOutputParser

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

# 基础链
chain = (
    ChatPromptTemplate.from_template("回答：{question}")
    | model
    | StrOutputParser()
)

# 添加重试包装
chain_with_retry = RunnableRetry(
    bound=chain,
    max_attempt_number=3,  # 最多尝试 3 次
    wait_exponential_jitter=True,  # 指数退避 + 抖动
    retry_on=(ConnectionError, TimeoutError),  # 只重试这些异常
)

# 使用
result = chain_with_retry.invoke({"question": "什么是机器学习？"})

使用 with_retry 方法（更简洁）：

# 使用 with_retry 方法
chain_with_retry = chain.with_retry(
    stop_after_attempt=3,  # 最多重试 3 次
    wait_exponential_jitter=True,  # 指数退避
    retry_on=(ConnectionError, TimeoutError),
)

# 指定自定义等待策略
from tenacity import wait_random_exponential

chain_with_retry = chain.with_retry(
    stop_after_attempt=5,
    wait=wait_random_exponential(multiplier=1, min=2, max=60),
)

什么时候需要重试：

网络请求超时或连接失败
API 限流（429 错误）
服务暂时不可用（503 错误）
结构化输出解析失败

不应该重试的情况：

认证错误（401）- 重试无法解决
参数错误（400）- 需要修复代码
配额耗尽 - 需要等待或升级

RunnableFallback 回退机制

RunnableFallback 在主链失败时自动切换到备用链：

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableFallback
from langchain.output_parsers import StrOutputParser

# 主模型
primary_model = init_chat_model("openai:gpt-4o-mini")
# 备用模型
fallback_model = init_chat_model("anthropic:claude-sonnet-4-5-20250929")

# 主链
primary_chain = (
    ChatPromptTemplate.from_template("回答：{question}")
    | primary_model
    | StrOutputParser()
)

# 备用链
fallback_chain = (
    ChatPromptTemplate.from_template("回答：{question}")
    | fallback_model
    | StrOutputParser()
)

# 组合：主链失败时使用备用链
chain_with_fallback = RunnableFallback(
    bound=primary_chain,
    fallbacks=[fallback_chain],
)

# 或使用 with_fallbacks 方法
chain_with_fallback = primary_chain.with_fallbacks([fallback_chain])

# 使用
result = chain_with_fallback.invoke({"question": "什么是深度学习？"})
# 如果 primary_chain 失败，会自动尝试 fallback_chain

典型应用场景：

模型容错：主模型不可用时切换到备用模型
降级策略：复杂模型失败时使用简单模型
本地回退：API 不可用时使用本地模型

from langchain_community.chat_models import ChatOllama

# API 模型 + 本地模型回退
api_chain = ChatPromptTemplate.from_template("{question}") | api_model
local_chain = ChatPromptTemplate.from_template("{question}") | ChatOllama(model="llama3")

# API 失败时使用本地模型
robust_chain = api_chain.with_fallbacks([local_chain])

RunnableEach 批量处理

RunnableEach 对列表中的每个元素应用相同的 Runnable：

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableEach
from langchain.output_parsers import StrOutputParser

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

# 单个文本处理链
single_chain = (
    ChatPromptTemplate.from_template("将以下文本翻译成英文：{text}")
    | model
    | StrOutputParser()
)

# 批量处理链
batch_chain = RunnableEach(bound=single_chain)

# 使用
texts = [
    {"text": "你好世界"},
    {"text": "机器学习很有趣"},
    {"text": "Python 是一种编程语言"},
]

results = batch_chain.invoke(texts)
# results = [
#     "Hello World",
#     "Machine learning is interesting",
#     "Python is a programming language",
# ]

RunnableEach vs batch 的区别：

batch：并行处理，一次性发送所有请求
RunnableEach：可以配置为顺序处理，适合有依赖关系的场景

组合使用高级模式

将多种模式组合使用，构建健壮的处理流程：

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch, RunnablePassthrough
from langchain.output_parsers import StrOutputParser

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

# 定义多个处理链
simple_chain = (
    ChatPromptTemplate.from_template("简洁回答：{question}")
    | model
    | StrOutputParser()
)

detailed_chain = (
    ChatPromptTemplate.from_template("详细解释：{question}")
    | model
    | StrOutputParser()
)

code_chain = (
    ChatPromptTemplate.from_template("提供代码示例回答：{question}")
    | model
    | StrOutputParser()
)

# 条件分支
branch_chain = RunnableBranch(
    (lambda x: any(kw in x["question"].lower() for kw in ["代码", "实现", "写一个"]), code_chain),
    (lambda x: any(kw in x["question"].lower() for kw in ["为什么", "如何理解", "原理"]), detailed_chain),
    simple_chain,
)

# 添加重试和回退
fallback_model = init_chat_model("anthropic:claude-sonnet-4-5-20250929")
fallback_chain = (
    ChatPromptTemplate.from_template("回答：{question}")
    | fallback_model
    | StrOutputParser()
)

# 完整链：分支 + 重试 + 回退
robust_chain = (
    branch_chain
    .with_retry(stop_after_attempt=3)
    .with_fallbacks([fallback_chain])
)

# 使用
result = robust_chain.invoke({"question": "Python 如何实现单例模式？"})

动态链构建

根据运行时条件动态构建链：

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain.output_parsers import StrOutputParser

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

def create_dynamic_chain(config):
    """根据配置动态创建链"""
    
    # 根据配置选择提示模板
    if config.get("style") == "formal":
        prompt = ChatPromptTemplate.from_template(
            "请以正式、专业的方式回答：{question}"
        )
    elif config.get("style") == "casual":
        prompt = ChatPromptTemplate.from_template(
            "请用轻松、友好的方式回答：{question}"
        )
    else:
        prompt = ChatPromptTemplate.from_template("回答：{question}")
    
    return prompt | model | StrOutputParser()

# 使用 RunnableLambda 动态选择链
def dynamic_router(input_dict):
    config = input_dict.get("config", {})
    chain = create_dynamic_chain(config)
    return chain.invoke({"question": input_dict["question"]})

dynamic_chain = RunnableLambda(dynamic_router)

# 使用
result = dynamic_chain.invoke({
    "question": "什么是人工智能？",
    "config": {"style": "formal"}
})

链的调试和检查

在开发过程中，检查链的结构和数据流：

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

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

chain = (
    ChatPromptTemplate.from_template("回答：{question}")
    | model
)

# 获取链的结构信息
print(f"链类型: {type(chain)}")
print(f"输入类型: {chain.InputType}")
print(f"输出类型: {chain.OutputType}")

# 查看链的步骤
if hasattr(chain, 'steps'):
    for i, step in enumerate(chain.steps):
        print(f"步骤 {i+1}: {type(step).__name__}")

# 使用 get_graph 可视化链
from langchain_core.runnables.graph import MermaidDrawMethod

# 生成流程图
graph = chain.get_graph()
print(graph.draw_ascii())

添加输出解析

输出解析器将模型输出转换为结构化格式：

StrOutputParser

提取字符串内容：

from langchain.output_parsers import StrOutputParser

chain = prompt | model | StrOutputParser()
result = chain.invoke({"topic": "AI"})  # 返回字符串而非 AIMessage

CommaSeparatedListOutputParser

解析逗号分隔的列表：

from langchain.output_parsers import CommaSeparatedListOutputParser
from langchain.prompts import ChatPromptTemplate

parser = CommaSeparatedListOutputParser()

prompt = ChatPromptTemplate.from_template(
    "列出5种编程语言。\n{format_instructions}"
).partial(format_instructions=parser.get_format_instructions())

chain = prompt | model | parser
result = chain.invoke({})
# result = ["Python", "JavaScript", "Java", "Go", "Rust"]

PydanticOutputParser

解析为 Pydantic 模型：

from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

# 定义数据模型
class Person(BaseModel):
    name: str = Field(description="人名")
    age: int = Field(description="年龄")
    occupation: str = Field(description="职业")

parser = PydanticOutputParser(pydantic_object=Person)

prompt = ChatPromptTemplate.from_template(
    "提取人物信息：{text}\n\n{format_instructions}"
).partial(format_instructions=parser.get_format_instructions())

chain = prompt | model | parser

result = chain.invoke({"text": "张三今年28岁，是一名软件工程师"})
# result = Person(name="张三", age=28, occupation="软件工程师")

JsonOutputParser

解析 JSON 格式输出：

from langchain.output_parsers import JsonOutputParser

parser = JsonOutputParser()

prompt = ChatPromptTemplate.from_template(
    "以JSON格式返回以下信息：{text}\n\n{format_instructions}"
).partial(format_instructions=parser.get_format_instructions())

chain = prompt | model | parser
result = chain.invoke({"text": "苹果公司成立于1976年"})
# result = {"company": "苹果公司", "founded": 1976}

配置与调试

传递配置

# 运行时配置
result = chain.invoke(
    {"topic": "AI"},
    config={
        "tags": ["demo", "test"],  # 用于追踪的标签
        "metadata": {"user": "test_user"},  # 元数据
        "max_concurrency": 5  # 并发限制
    }
)

重试机制

# 添加重试逻辑
chain_with_retry = chain.with_retry(
    stop_after_attempt=3,  # 最多重试3次
    wait_exponential_jitter=True  # 指数退避
)

调试输出

# 方式1：全局调试
from langchain_core.globals import set_debug
set_debug(True)

# 方式2：使用回调
from langchain_core.tracers import ConsoleCallbackHandler

result = chain.invoke(
    {"topic": "AI"},
    config={"callbacks": [ConsoleCallbackHandler()]}
)

流式事件

# 获取详细的执行事件
async for event in chain.astream_events({"topic": "AI"}, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

从旧版 Chain 迁移

如果你有使用旧版 Chain API 的代码，需要迁移到 LCEL 或 langchain-classic 包：

LLMChain 迁移

# 旧代码（已废弃）
from langchain.chains import LLMChain
chain = LLMChain(llm=model, prompt=prompt)
result = chain.run("question")

# 新代码（推荐）
chain = prompt | model | StrOutputParser()
result = chain.invoke({"question": "question"})

SequentialChain 迁移

# 旧代码（已废弃）
from langchain.chains import SimpleSequentialChain
chain = SimpleSequentialChain(chains=[chain1, chain2])

# 新代码（推荐）
chain = chain1 | chain2

RetrievalQA 迁移

# 旧代码（已废弃）
from langchain.chains import RetrievalQA
qa = RetrievalQA.from_chain_type(llm=model, retriever=retriever)

# 新代码（推荐）
from langchain_core.runnables import RunnablePassthrough

qa_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | ChatPromptTemplate.from_template("""
基于以下上下文回答问题：
上下文：{context}
问题：{question}
""")
    | model
    | StrOutputParser()
)

使用 langchain-classic

如果暂时无法迁移，可以安装 langchain-classic 继续使用旧 API：

pip install langchain-classic

# 更新导入路径
from langchain_classic.chains import LLMChain
from langchain_classic.chains import RetrievalQA

注意：langchain-classic 仅用于过渡，新项目请直接使用 LCEL。

实战示例

智能问答系统

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

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

def classify_question(question):
    """问题分类"""
    classify_chain = (
        ChatPromptTemplate.from_template(
            "判断问题类型，只返回一个词：technical/general/opinion。\n问题：{question}"
        )
        | model
        | StrOutputParser()
    )
    return classify_chain.invoke({"question": question})

# 根据类型选择提示模板
def get_prompt_by_type(question_type):
    templates = {
        "technical": "作为技术专家，详细回答：{question}",
        "opinion": "从多个角度分析：{question}",
        "general": "简洁回答：{question}"
    }
    return ChatPromptTemplate.from_template(templates.get(question_type, templates["general"]))

# 完整链
def answer_question(question):
    # 分类
    q_type = classify_question(question)
    # 选择提示并回答
    chain = (
        get_prompt_by_type(q_type)
        | model
        | StrOutputParser()
    )
    return chain.invoke({"question": question})

result = answer_question("Python 装饰器的工作原理是什么？")

文档处理管道

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel

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

# 文档处理链
document_chain = (
    RunnableParallel(
        # 并行执行多个分析
        summary=(
            ChatPromptTemplate.from_template("用一句话总结：{text}")
            | model
            | StrOutputParser()
        ),
        keywords=(
            ChatPromptTemplate.from_template("提取3个关键词：{text}")
            | model
            | StrOutputParser()
        ),
        sentiment=(
            ChatPromptTemplate.from_template("分析情感倾向（正面/负面/中性）：{text}")
            | model
            | StrOutputParser()
        ),
        topics=(
            ChatPromptTemplate.from_template("识别主题领域：{text}")
            | model
            | StrOutputParser()
        )
    )
)

result = document_chain.invoke({
    "text": "人工智能技术发展迅速，深度学习在图像识别领域取得重大突破。"
})
# result = {
#     "summary": "AI深度学习在图像识别领域取得突破。",
#     "keywords": "人工智能, 深度学习, 图像识别",
#     "sentiment": "正面",
#     "topics": "科技, 人工智能"
# }

下一步

LCEL 是 LangChain 的基础，但构建复杂的智能代理需要更强大的工具。

Agents 代理 - 使用 create_agent 构建能够自主决策的智能代理
RAG 检索增强 - 构建基于私有知识的问答系统
Tools 工具 - 为代理添加工具调用能力

什么是 Chain？​

传统方式 vs LCEL​

Runnable 接口​

Runnable 核心方法​

基本用法​

管道操作符​

工作原理​

多步骤链​

RunnableParallel​

使用字典语法​

显式创建 RunnableParallel​

RunnablePassthrough​

基本用法​

在链中保留原始数据​

RunnablePassthrough.assign​

RunnableLambda​

基本用法​

在复杂链中使用​

组合模式​

顺序链（Sequential Chain）​

分支链（Branching）​

循环处理（Map-Reduce）​

Runnable 高级模式​

RunnableBranch 条件分支​

RunnableRetry 重试机制​

RunnableFallback 回退机制​

RunnableEach 批量处理​

组合使用高级模式​

动态链构建​

链的调试和检查​

添加输出解析​

StrOutputParser​

CommaSeparatedListOutputParser​

PydanticOutputParser​

JsonOutputParser​

配置与调试​

传递配置​

重试机制​

调试输出​

流式事件​

从旧版 Chain 迁移​

LLMChain 迁移​

SequentialChain 迁移​

RetrievalQA 迁移​

使用 langchain-classic​

实战示例​

智能问答系统​

文档处理管道​

下一步​

什么是 Chain？

传统方式 vs LCEL

Runnable 接口

Runnable 核心方法

基本用法

管道操作符

工作原理

多步骤链

RunnableParallel

使用字典语法

显式创建 RunnableParallel

RunnablePassthrough

基本用法

在链中保留原始数据

RunnablePassthrough.assign

RunnableLambda

基本用法

在复杂链中使用

组合模式

顺序链（Sequential Chain）

分支链（Branching）

循环处理（Map-Reduce）

Runnable 高级模式

RunnableBranch 条件分支

RunnableRetry 重试机制

RunnableFallback 回退机制

RunnableEach 批量处理

组合使用高级模式

动态链构建

链的调试和检查

添加输出解析

StrOutputParser

CommaSeparatedListOutputParser

PydanticOutputParser

JsonOutputParser

配置与调试

传递配置

重试机制

调试输出

流式事件

从旧版 Chain 迁移

LLMChain 迁移

SequentialChain 迁移

RetrievalQA 迁移

使用 langchain-classic

实战示例

智能问答系统

文档处理管道

下一步