LangChain Prompt 模板与 LCEL 链式调用

概述

在 LangChain 中，ChatPromptTemplate 负责组装输入，ChatOpenAI 负责调用模型。两者都实现了统一的 Runnable 接口，因此可以用 | 串联成链（LCEL）。本文从职责区分、invoke 差异、链式拼接、流式输出和链式前提五个角度，把这一套机制讲透。

一、ChatPromptTemplate 与 ChatOpenAI 的区别

两者是上下游关系，不存在替代关系。

维度	ChatPromptTemplate	ChatOpenAI
本质	提示词模板生成器	大模型 API 客户端
输入	变量字典（如 `{"input": "北京天气"}`）	messages 消息列表
输出	`ChatPromptValue`（可转成 messages）	`AIMessage`（模型真实回复）
是否发网络请求	否，纯本地字符串拼接	是，调用 OpenAI 接口
类比	填表的表格模板	接电话的客服

协作流程

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 第一步：造模板（不调用模型）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是翻译助手"),
    ("human", "{text}")
])

# 第二步：实例化模型客户端
model = ChatOpenAI(model="gpt-4o")

# 第三步：模板填变量 → 得到 messages
messages = prompt.invoke({"text": "今天天气不错"})

# 第四步：messages 丢给模型 → 得到回复
response = model.invoke(messages)

常见误区

以为 ChatPromptTemplate 会调模型 → 不会，它只是拼字符串
以为 ChatOpenAI 能直接吃变量 → 不能，它只认 messages
以为 ChatOpenAI 只能配 OpenAI → 不是，还有 ChatAnthropic、ChatZhipuAI、ChatTongyi 等，prompt 模板通用

二、两者都有 invoke，区别在哪里

两者都有 invoke，是因为都实现了 Runnable 接口（LangChain 的统一抽象），但 invoke 的输入、输出、内部动作完全不同。

维度	ChatPromptTemplate.invoke	ChatOpenAI.invoke
输入	变量字典 `{"text": "你好"}`	messages 消息列表
输出	`ChatPromptValue`	`AIMessage`
内部动作	字符串替换、组装消息对象	发 HTTP 请求到模型 API
是否联网	否，纯本地	是
耗时	微秒级	秒级（取决于模型）

用代码看差异

python

# ChatPromptTemplate.invoke —— 本地拼字符串
result1 = prompt.invoke({"text": "你好"})
print(type(result1))  # ChatPromptValue
print(result1.to_messages())
# [SystemMessage(content='你是翻译助手'), HumanMessage(content='你好')]
# 没有任何网络请求，只是把 {text} 替换成"你好"

# ChatOpenAI.invoke —— 调模型
result2 = model.invoke([HumanMessage(content="你好")])
print(type(result2))  # AIMessage
print(result2.content)
# "Hello"（模型真实返回，耗时几秒）

验证小实验

断开网络，prompt.invoke 照样能跑；model.invoke 立刻报连接错误。这就是最直观的区别：一个纯本地模板替换，一个远程模型推理。

三、LCEL 链式拼接

不用链式：手动拆两步

python

# 第一步：prompt 模板填变量，得到 messages
messages = prompt.invoke({
    "role": "资深架构师",
    "question": "什么是 Durable Execution？"
})

# 第二步：把 messages 丢给模型
resp = llm.invoke(messages)

print(resp.content)

等价关系：

python

chain.invoke(变量)  ≡  llm.invoke(prompt.invoke(变量))

链式写法 `prompt | llm`

python

chain = prompt | llm
resp = chain.invoke({"role": "资深架构师", "question": "什么是 Durable Execution？"})

| 是 Python 管道符重载，LangChain 用它实现了数据流自动传递：

变量字典
   ↓ （prompt.invoke）
ChatPromptValue / messages
   ↓ （llm.invoke）
AIMessage

前一个的输出，自动就是后一个的输入，不用手动接。

链式免费送的能力

1. 自动批处理 batch

python

# 不用链：要写循环
results = []
for q in ["问题1", "问题2", "问题3"]:
    results.append(llm.invoke(prompt.invoke({"question": q})))

# 用链：一行搞定
chain.batch([
    {"role": "架构师", "question": "问题1"},
    {"role": "架构师", "question": "问题2"},
    {"role": "架构师", "question": "问题3"},
])

2. 自动流式输出 stream

python

for chunk in chain.stream({"question": "..."}):
    print(chunk.content, end="", flush=True)

3. 自动异步 ainvoke / abatch

python

await chain.ainvoke({"question": "..."})
await chain.abatch([...])

4. 自由组合任意数量

python

from langchain_core.output_parsers import StrOutputParser

chain = prompt | llm | StrOutputParser()
# 最终直接输出字符串，不用 .content
text = chain.invoke({"question": "..."})
print(type(text))  # str

5. 并行分支（RunnableParallel）

python

from langchain_core.runnables import RunnableParallel

chain = RunnableParallel({
    "answer": prompt | llm | StrOutputParser(),
    "length": lambda x: len(x["question"])
})
# 一次调用，同时得到回答和问题长度

管道符的本质

python

# 你写的：
chain = prompt | llm

# 实际等价于：
chain = prompt.__or__(llm)
# 返回一个新的 Sequence 类型 Runnable

这个 Sequence 对象内部记住了"先跑 prompt，再跑 llm"，调用 invoke 时按顺序执行。

什么时候不用链式

场景	建议
简单一问一答	用链，简洁
工具调用循环（要判断 tool_calls）	手动写循环，链不够灵活
中间有条件分支	手动控制，或用 `RunnableBranch`
学习理解底层原理	先拆开写，再用链

四、流式输出 stream 完整代码

python

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

MODEL = "gpt-4o"

# 1. 构造模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位{role}，回答要简洁专业。"),
    ("human", "{question}"),
])

# 2. 实例化模型
llm = ChatOpenAI(model=MODEL)

# 3. 拼成链
chain = prompt | llm

# 4. 流式输出：逐 token 打印
print("[流式回答] ", end="", flush=True)
for chunk in chain.stream({
    "role": "资深架构师",
    "question": "什么是 Durable Execution？用三句话解释。"
}):
    # chunk 是 AIMessageChunk，只取 content 部分
    print(chunk.content, end="", flush=True)

print("\n")

关键细节

chunk 是什么：每一段返回的是 AIMessageChunk，不是完整的 AIMessage，只包含当前这一小段文字增量。
为什么 end="" + flush=True：end="" 让所有 chunk 连在一起不换行；flush=True 强制立即输出，否则 Python 会缓存，看不到打字机效果。

不用链的等价写法

python

messages = prompt.invoke({"role": "资深架构师", "question": "..."})

print("[流式回答] ", end="", flush=True)
for chunk in llm.stream(messages):
    print(chunk.content, end="", flush=True)

工具调用场景下的 stream

如果模型可能返回 tool_calls，chunk 里也会带：

python

for chunk in chain.stream({...}):
    if chunk.tool_call_chunks:  # 工具调用增量
        print("工具调用片段:", chunk.tool_call_chunks)
    if chunk.content:
        print(chunk.content, end="", flush=True)

"自己处理"到底处理什么

不用链和用链，chunk 遍历代码完全一样。区别在衔接环节：

步骤	不用链	用链
1. 变量 → messages	手动调 `prompt.invoke()`	chain 内部自动做
2. messages → 模型	手动把结果传给 `llm.stream()`	chain 内部自动做
3. 遍历 chunk	一样	一样
4. 打印 content	一样	一样

所以"自己处理"指的是上一个组件的输出要手动递给下一个组件，不是 chunk 本身。链路越长，手动衔接越繁琐，链的价值越大。

五、链式的前提：Runnable 协议

不是任何两个对象都能用 | 串起来，必须满足 LangChain 的 Runnable 协议。

硬性要求

两个对象必须都是 Runnable 实例，并且前一个的输出类型 = 后一个的输入类型。

python

prompt | llm
#     ↑          ↑
# 输出是 messages   输入要的是 messages
# 类型正好对上 ✅

输入输出类型匹配表

组件	invoke 输入	invoke 输出
`ChatPromptTemplate`	变量字典	`ChatPromptValue`（可转 messages）
`ChatOpenAI`（聊天模型）	messages 列表	`AIMessage`
`StrOutputParser`	`AIMessage`	`str`
`Tool` / 自定义函数	对应类型	对应类型

所以能串成：

python

prompt | llm | StrOutputParser()
# 字典 → messages → AIMessage → str ✅

哪些能用 `|`

只要是 LangChain 的 Runnable 都可以：

ChatPromptTemplate / PromptTemplate
所有 ChatModel（ChatOpenAI、ChatAnthropic 等）
所有 OutputParser（StrOutputParser、JsonOutputParser 等）
Tool / @tool 装饰的函数
Retriever
自己继承 Runnable 写的类
甚至上一条链本身（链也是 Runnable）

哪些不能用 `|`

普通 Python 对象没有 __or__ 重载，直接串会报错：

python

"你好" | llm   # ❌ str 不是 Runnable
123 | llm      # ❌ int 不是 Runnable

# 但可以用 RunnableLambda 包一下
from langchain_core.runnables import RunnableLambda
RunnableLambda(lambda x: x["question"].upper()) | llm  # ✅

验证方法

python

from langchain_core.runnables import Runnable

print(isinstance(prompt, Runnable))  # True
print(isinstance(llm, Runnable))     # True
print(isinstance("hello", Runnable)) # False

统一接口

所有 Runnable 都实现：

invoke(input) → output
batch(list_of_inputs) → list_of_outputs
stream(input) → iterator_of_chunks
ainvoke / abatch / astream（异步版）
__or__(other) → 拼成新的 Sequence Runnable

一句话总结

ChatPromptTemplate 负责本地拼消息，ChatOpenAI 负责远程调模型；两者都实现 Runnable 接口，所以能用 | 自动衔接成链，白送 batch / stream / 异步 / 多步组合能力。能用 | 的前提是两边都是 Runnable 且输入输出类型对得上。

LangChain Prompt 模板与 LCEL 链式调用 ​

概述 ​

一、ChatPromptTemplate 与 ChatOpenAI 的区别 ​

协作流程 ​

常见误区 ​

二、两者都有 invoke，区别在哪里 ​

用代码看差异 ​

验证小实验 ​

三、LCEL 链式拼接 ​

不用链式：手动拆两步 ​

链式写法 prompt | llm ​

链式免费送的能力 ​

1. 自动批处理 batch ​

2. 自动流式输出 stream ​

3. 自动异步 ainvoke / abatch ​

4. 自由组合任意数量 ​

5. 并行分支（RunnableParallel） ​

管道符的本质 ​

什么时候不用链式 ​

四、流式输出 stream 完整代码 ​

关键细节 ​

不用链的等价写法 ​

工具调用场景下的 stream ​

"自己处理"到底处理什么 ​

五、链式的前提：Runnable 协议 ​

硬性要求 ​

输入输出类型匹配表 ​

哪些能用 | ​

哪些不能用 | ​

验证方法 ​

统一接口 ​

一句话总结 ​

LangChain Prompt 模板与 LCEL 链式调用

概述

一、ChatPromptTemplate 与 ChatOpenAI 的区别

协作流程

常见误区

二、两者都有 invoke，区别在哪里

用代码看差异

验证小实验

三、LCEL 链式拼接

不用链式：手动拆两步

链式写法 `prompt | llm`

链式免费送的能力

1. 自动批处理 batch

2. 自动流式输出 stream

3. 自动异步 ainvoke / abatch

4. 自由组合任意数量

5. 并行分支（RunnableParallel）

管道符的本质

什么时候不用链式

四、流式输出 stream 完整代码

关键细节

不用链的等价写法

工具调用场景下的 stream

"自己处理"到底处理什么

五、链式的前提：Runnable 协议

硬性要求

输入输出类型匹配表

哪些能用 `|`

哪些不能用 `|`

验证方法

统一接口

一句话总结