链式调用 (Chains)

使用 LCEL 构建复杂的工作流

什么是 LCEL？

LCEL (LangChain Expression Language) 是 LangChain 1.0 的声明式链式语法，使用 | 操作符组合组件。

python 
# LCEL 基本语法
chain = prompt | model | parser

# 等价于传统写法：
# chain = prompt | model | parser

Runnable 接口

所有 LCEL 组件都实现 Runnable 接口，提供统一的方法：

方法 说明
stream() 流式输出
invoke() 同步调用
ainvoke() 异步调用
batch() 批量处理
abatch() 异步批量处理
map() 并行映射

基础链构建

简单链

python 
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 创建组件
prompt = ChatPromptTemplate.from_template("讲个关于{topic}的笑话")
llm = ChatOpenAI(model="gpt-4o")
parser = StrOutputParser()

# 用 | 组合成链
joke_chain = prompt | llm | parser

# 调用
result = joke_chain.invoke({"topic": "程序员"})
print(result)

并行执行

python 
from langchain_core.runnables import RunnableParallel

# 并行链：同时执行多个操作
chain = RunnableParallel(
    joke=prompt | llm | parser,
    count=prompt | llm | StrOutputParser() | len
)

# 同时获取笑话和字数
result = chain.invoke({"topic": "AI"})
print(result)
# 输出: {'joke': '...', 'count': 150}

路由链

python 
from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 定义不同的处理函数
def handle_code(input):
    return "这是代码问题"

def handle_general(input):
    return "这是一般问题"

# 创建路由
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "代码" in x, RunnableLambda(handle_code)),
    (lambda x: True, RunnableLambda(handle_general))
)

# 使用路由
result = branch.invoke("我的代码有问题")
print(result)  # "这是代码问题"

复杂链式组合

顺序执行

python 
from langchain_core.runnables import RunnablePassthrough

# 复杂链：先总结，再翻译，再格式化
summary_prompt = ChatPromptTemplate.from_template("总结以下内容：\n{content}")
translate_prompt = ChatPromptTemplate.from_template("翻译成英文：\n{summary}")
format_prompt = ChatPromptTemplate.from_template("格式化为列表：\n{translation}")

llm = ChatOpenAI(model="gpt-4o")

chain = (
    {"content": lambda x: x["content"]}
    | summary_prompt | llm
    | {"summary": lambda x: x.content}
    | translate_prompt | llm
    | {"translation": lambda x: x.content}
    | format_prompt | llm
)

中间步骤访问

python 
from langchain_core.runnables import RunnablePassthrough

# RunnablePassthrough 传递输入，同时添加新字段
chain = {
    "original": RunnablePassthrough(),  # 保留原始输入
    "processed": prompt | llm | StrOutputParser()  # 处理后的输出
}

result = chain.invoke({"query": "Python 是什么"})
# 输出: {'original': {'query': '...'}, 'processed': '...'}

错误处理

重试机制

python 
from langchain_core.runnables import RunnableRetryWithMessage
from langchain_openai import ChatOpenAI

# 创建带重试的链
llm = ChatOpenAI(model="gpt-4o")

retry_chain = RunnableRetryWithMessage(
    llm,
    max_retries=3,
    retry_error_message="请重新生成"
)

result = retry_chain.invoke("讲个笑话")

回退策略

python 
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# 主模型和回退模型
primary = ChatOpenAI(model="gpt-4o")
fallback = ChatAnthropic(model="claude-sonnet-4-5-20250514")

# 创建回退链
from langchain_core.runnables import RunnableWithFallbacks

chain_with_fallback = RunnableWithFallbacks(
    primary,
    fallbacks=[fallback]
)

# 主模型失败时自动使用回退
result = chain_with_fallback.invoke("Hello")

LCEL 流程图

graph TD
    A[输入] --> B[PromptTemplate]
    B --> C[ChatModel]
    C --> D[OutputParser]
    D --> E[输出]

    C -.->|错误| F[Fallback Model]
    F --> D

动态链构建

python 
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 动态选择模型
def create_chain(model_name: str):
    if model_name == "openai":
        llm = ChatOpenAI(model="gpt-4o")
    elif model_name == "anthropic":
        from langchain_anthropic import ChatAnthropic
        llm = ChatAnthropic(model="claude-sonnet-4-5-20250514")
    else:
        raise ValueError(f"Unknown model: {model_name}")

    prompt = ChatPromptTemplate.from_template("回答：{question}")
    return prompt | llm

# 使用
chain = create_chain("openai")
result = chain.invoke({"question": "什么是 LCEL？"})

✏️ 练习题

选择题 1. LCEL 中的 | 操作符的作用是？
A. 逻辑或运算
B. 链式组合组件
C. 位运算
D. 字符串拼接
编程题 2. 创建一个链：输入问题 → 获取答案 → 翻译成英文 → 计算字数
查看参考答案
python 
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# 问题 -> 答案
qa_prompt = ChatPromptTemplate.from_template("回答：{question}")
qa_chain = qa_prompt | llm

# 翻译
translate_prompt = ChatPromptTemplate.from_template("翻译成英文：{text}")
translate_chain = translate_prompt | llm

# 组合完整链
chain = (
    qa_chain
    | {"answer": lambda x: x.content}
    | (lambda x: translate_chain.invoke({"text": x["answer"]}))
    | {"translation": lambda x: x.content, "count": lambda x: len(x.content)}
)

result = chain.invoke({"question": "什么是 Python？"})
print(result)  # {'translation': '...', 'count': ...}