Chains 链式调用

LangChain V1.x

---

1、Chain的核心概念

什么是Chain？

简单来说，Chain是将多个组件按特定顺序组合起来完成复杂任务的工作流或管道（Pipeline）。

你可以把它想象成一个流水线：原材料（输入）经过一系列加工步骤（LLM调用、工具使用、数据转换等），最终成为成品（输出）。

为什么需要Chain？

虽然LLM本身很强大，但大多数实际应用需要多个步骤：

1. 接收用户输入
2. 根据输入构建提示（Prompt）
3. 将提示发送给LLM
4. 解析LLM的输出
5. 根据输出可能再执行其他操作（如调用API、查询数据库等）
6. 将最终结果返回给用户

Chain的本质是自动化和封装——让我们能像搭积木一样构建复杂AI应用，使用起来却像调用一个函数一样简单。

2、Runnable接口

LangChain所有核心组件都实现了统一的 Runnable接口。这是LangChain最底层的抽象，代表一个"可以被调用、批处理、流式传输和组合的工作单元"。

• 定位：LangChain中的抽象基类（ABC）
• 核心理念："一切可执行的对象都应该有统一的调用方式"

Runnable核心方法

方法	作用	适用场景
invoke	同步调用单个输入	简单单次请求
batch	批量处理多个输入	批量数据处理
stream	流式处理输入	实时输出展示
ainvoke	异步调用单个输入	异步编程环境
abatch	异步批量处理	高性能批量处理

哪些组件是Runnable？

几乎所有的LangChain核心组件都实现了Runnable接口：

组件类型	示例
提示模板	PromptTemplate, ChatPromptTemplate
语言模型	ChatOpenAI, ChatOllama
输出解析器	StrOutputParser, JsonOutputParser
工具	Tool
整个链	任何LCEL链

统一调用的价值

有了统一的Runnable接口，所有组件都可以用相同的方式调用：

prompt.invoke({"topic": "AI"})        # 提示模板
model.invoke(prompt_value)            # 语言模型
parser.invoke(ai_message)             # 输出解析器
chain.invoke({"question": "你好"})    # 整个链

本质：接口统一让组件具备了"即插即用"的能力，这直接催生了LCEL。

3、LCEL表达式语言

什么是LCEL？

LCEL（LangChain Expression Language）是LangChain提供的声明式组合语言，专门用于组合Runnable组件。

• 核心操作符：管道符 |
• 核心思想：使用 | 将多个Runnable像拼积木一样组合起来

# 典型的LCEL链式写法
chain = prompt | model | output_parser

# Chain本身也是Runnable，可以继续调用
result = chain.invoke({"topic": "编程"})# 1 prompt.invoke  2. model.invoke 3. output_parser.invoke

基础语法示例

两个组件连接：

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

prompt = ChatPromptTemplate.from_template("请用{tone}风格回答：{question}")
model = init_chat_model(model="gpt-3.5-turbo")

# LCEL组合 - 像拼积木一样简单
chain = prompt | model

chain.invoke({
    "tone": "幽默",
    "question": "什么是人工智能？"
})

三个组件连接：

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()

# LCEL组合
chain = prompt | model | parser

chain.invoke({
    "tone": "幽默",
    "question": "什么是人工智能？"
})

LCEL的核心优势

优势	说明
组合后的链自动拥有所有Runnable能力	支持invoke/stream/batch等全部方法
链本身也是Runnable，可以继续组合	可以将链作为组件，构建更复杂的链
声明式方式，可读性更高	管道符连接，逻辑一目了然

4、Runnable组合器

LCEL之所以强大，是因为它背后有丰富的Runnable组合器。这些组合器让我们能够构建各种复杂逻辑。

4.1 RunnableSequence - 顺序流水线

作用：将多个Runnable按顺序连接成处理管道，前一个的输出作为后一个的输入。

from langchain_core.runnables import RunnableSequence, RunnableLambda

# 显式创建序列
sequence_chain = RunnableSequence(
    first=RunnableLambda(lambda x: x.upper()),           # 第一步：转换大写
    middle=[RunnableLambda(lambda x: f"HELLO {x} !")],   # 第二步：加装饰
    last=RunnableLambda(lambda x: f"最终：{x}")          # 第三步：加前缀
)

result = sequence_chain.invoke("world")
print(result)  # 最终：HELLO WORLD !

LCEL等价写法（推荐）：

chain = (
    RunnableLambda(lambda x: x.upper())
    | RunnableLambda(lambda x: f"HELLO {x} !")
    | RunnableLambda(lambda x: f"最终：{x}")
)

说明：LCEL的 | 运算符底层就是创建 RunnableSequence，推荐使用更简洁的LCEL写法。

4.2 RunnableParallel - 并行分叉

作用：同时执行多个Runnable，将结果合并为一个字典。

from langchain_core.runnables import RunnableParallel, RunnableLambda

# 创建并行任务
parallel_chain = RunnableParallel({
    "length": RunnableLambda(lambda x: len(x)),           # 计算长度
    "uppercase": RunnableLambda(lambda x: x.upper()),     # 转大写
    "reversed": RunnableLambda(lambda x: x[::-1]),        # 反转字符串
    "word_count": RunnableLambda(lambda x: len(x.split())) # 单词计数
})

result = parallel_chain.invoke("Hello World LangChain")
# 输出：
# {
#   'length': 24,
#   'uppercase': 'HELLO WORLD LANGCHAIN',
#   'reversed': 'niahCgnaL dlroW olleH',
#   'word_count': 3
# }

工作原理：

1. 同一份输入广播 调用 invoke(x) 时，它把同一个输入 x 传给字典里每个 runnable（length/uppercase/...）。

2. 并发执行 这些 runnable 会被调度为并行/并发运行（在 LangChain 的 runnable 运行时里实现；对纯 RunnableLambda 这种本地计算，通常是并发调度；对网络 I/O（LLM/检索）收益更明显）。

3. 收集并合并结果 等全部分支都完成后，把每个分支的输出按 key 组装成一个 dict

4. {
     "length": <length的结果>,
     "uppercase": <uppercase的结果>,
     ...
    }

LCEL等价写法（推荐）：

# 字典语法自动创建RunnableParallel
chain = {
    "length": RunnableLambda(lambda x: len(x)),
    "uppercase": RunnableLambda(lambda x: x.upper()),
    "reversed": RunnableLambda(lambda x: x[::-1]),
    "word_count": RunnableLambda(lambda x: len(x.split()))
}
# 不能单独用，单独用实际是个字典不能invoke,如果想使用一定要是一个链chain

LCEL 字典语法为什么等价？ 在 LCEL 中，链里出现一个字典就会被自动"编译"为 RunnableParallel；也就是说 {key: runnable} 是一种语法糖，运行时机制与 RunnableParallel({...}) 相同。

说明：每个分支是相互独立的；如果你想"某个分支依赖另一个分支的结果"，就不适合并行分叉，而要用顺序链（|）或先并行后再用一个 RunnableLambda 做汇总/计算。

实际应用：先并行生成多个结果，再汇总：

from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

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

# 两个并行的赏析链
paragraph_1_chain = (
    PromptTemplate.from_template("对这首诗做赏析，分析含义：{poem}")
    | llm | StrOutputParser()
)
paragraph_2_chain = (
    PromptTemplate.from_template("对这首诗做赏析，分析意境：{poem}")
    | llm | StrOutputParser()
)

# 汇总链
summary_chain = (
    PromptTemplate.from_template(
        "第一种赏析：{paragraph_1}\n\n第二种赏析：{paragraph_2}\n\n请比较哪个更好，为什么"
    )
    | llm | StrOutputParser()
)

# 先并行，后汇总
full_chain = {
    "paragraph_1": paragraph_1_chain,
    "paragraph_2": paragraph_2_chain,
} | summary_chain

resp = full_chain.invoke({"poem": "菩提本无树，明镜亦非台，本来无一物，何处惹尘埃。"})
print(resp)

4.3 RunnablePassthrough - 输入传递

作用：将输入原样传递到输出，常用于保留原始输入的同时添加新字段。

from langchain_core.runnables import RunnablePassthrough, RunnableParallel

# 场景1：直接传递输入
chain = RunnablePassthrough()
result = chain.invoke({"key": "value"})
# 输出: {"key": "value"}

# 场景2：保留原始输入 + 添加新字段
chain = RunnableParallel(
    original=RunnablePassthrough(),            # 保留原始输入
    uppercase=lambda x: x["text"].upper()      # 添加转换后的字段
)
result = chain.invoke({"text": "hello"})
# 输出: {"original": {"text": "hello"}, "uppercase": "HELLO"}

RAG典型用法——保留问题 + 检索上下文：

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template(
    "请回答以下问题：{question}\n\n相关背景：{context}"
)

chain = (
    {
        "question": RunnablePassthrough(),           # 用户问题原样传递
        "context": lambda x: retrieve_context(x)     # 检索相关上下文
    }
    | prompt
    | llm
    | StrOutputParser()
)

4.4 RunnableLambda - 自定义逻辑

作用：将普通Python函数包装为Runnable，使其可以在LCEL链中使用。

from langchain_core.runnables import RunnableLambda

def extract_domain(url):
    """从URL中提取域名"""
    return url.split('//')[-1].split('/')[0]

def add_protocol(domain):
    """添加协议前缀"""
    return f"http://{domain}"

# 包装成Runnable
domain_extractor = RunnableLambda(extract_domain)
protocol_adder = RunnableLambda(add_protocol)

# 在链中使用
url_processor = domain_extractor | protocol_adder
result = url_processor.invoke("https://www.example.com/path")
# 输出："http://www.example.com"

5、Runnable方法一览

Runnable接口提供的完整方法集：

方法	说明	示例
invoke	同步单次调用	chain.invoke({"topic": "AI"})
stream	流式输出	for chunk in chain.stream({...})
batch	批量处理	chain.batch([{"topic": "AI"}, {"topic": "猫"}])
ainvoke	异步调用	await chain.ainvoke({...})
astream	异步流式	async for chunk in chain.astream({...})
abatch	异步批量	await chain.abatch([...])

from langchain_openai import ChatOpenAI

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

# 同步调用
response = llm.invoke("你好")

# 流式调用 —— 逐token输出
for chunk in llm.stream("讲一个故事"):
    print(chunk.content, end="")

# 批量调用 —— 同时处理多个输入
responses = llm.batch(["你好", "再见", "谢谢"])

# 异步调用
import asyncio
response = await llm.ainvoke("你好")

6、添加对话历史

构建对话系统时，需要让链"记住"之前的对话内容：

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

# 创建会话存储（以session_id为key）
store = {}


def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]


# 创建基础链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是AI助手"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm

# 包装为带历史记录的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 使用时指定session_id
response_1 = chain_with_history.invoke(
    {"input": "我叫张三"},
    config={"configurable": {"session_id": "user123"}}
)
print(response_1.content)

# 后续对话会自动携带历史
response_2 = chain_with_history.invoke(
    {"input": "我叫什么名字？"},
    config={"configurable": {"session_id": "user123"}}
)
print(response_2.content)
# AI会回答"你叫张三"，因为历史记录中有这个信息

---

7、高级模式

7.1 错误处理与重试

# 使用with_retry自动重试
chain_with_retry = prompt | llm.with_retry(stop_after_attempt=3) | parser

7.2 回退机制

主模型失败时自动切换备用模型：

primary_llm = ChatOpenAI(model="gpt-4o")
fallback_llm = ChatOpenAI(model="gpt-4o-mini")

chain_with_fallback = prompt | primary_llm.with_fallbacks([fallback_llm]) | parser

---

8、小结

8.1 核心概念总结

主题	核心要点	关键API
Runnable接口	统一的调用方式	invoke, stream, batch
LCEL	管道符组合组件	`chain = prompt
Runnable组合器	构建复杂逻辑	RunnableParallel, RunnablePassthrough

8.2 代码模板速查

# 基础链
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"topic": "AI"})

# 结构化输出
class Result(BaseModel):
    answer: str
    confidence: float

structured_llm = llm.with_structured_output(Result)

# 并行链
parallel = {
    "en": translate_chain_en,
    "kr": translate_chain_kr
}

# 带对话历史
chain_with_history = RunnableWithMessageHistory(chain, get_session_history, ...)

8.3 选型建议

场景	推荐
简单文本生成	StrOutputParser
结构化数据	with_structured_output
多轮对话	ChatPromptTemplate + RunnableWithMessageHistory
批量处理	batch()

---

参考资料

• LangChain Prompts 官方文档
• LangChain Output Parsers 文档
• LangChain Expression Language (LCEL)
• Runnable 接口文档