chain链
「将组件串联,上一个组件的输出作为下一个组件的输入」是Langchain链(尤其是 | 管道链)的核心工作原理,这也是链式调用的核心价值:实现数据的自动化流转与组件的协同工作,如下。
chain = prompt_template | model
(左边的输出作为右边的输入)
核心前提:即Runnable子类对象才能入链(以及callable、Mapping接口子类对象也可加入)。

image.png
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.runnables.base import RunnableSerializable
chat_prompt_template = ChatPromptTemplate.from_messages([
("system", "你是一个边塞诗人,可以创作诗词"),
MessagesPlaceholder(variable_name="history"),
("human", "请再来一首唐诗,无需额外输出"),
])
history_data = [
("human", "你来写一个唐诗"),
("ai", "床前明月光,疑是地上霜。举头望明月,低头思故乡。"),
("human", "好诗!请再写一首"),
("ai", "锄禾日当午,汗滴禾下土。谁知盘中餐,粒粒皆辛苦。"),
]
model = ChatTongyi(model="qwen3-max")
chain: RunnableSerializable = chat_prompt_template | model
print(type(chain))
# Runnable接口,invoke执行
response = chain.invoke({
"history": history_data,
"input": "再写一首!",
})
print(response.content + "\n")
# Runnable接口,stream执行
for chunk in chain.stream({
"history": history_data,
"input": "再写一首!",
}):
print(chunk.content, end="")
print("\n")
<class 'langchain_core.runnables.base.RunnableSequence'>
大漠孤烟直,长河落日圆。
萧关逢候骑,都护在燕然。
烽火照西京,心中自不平。
牙璋辞凤阙,铁骑绕龙城。
雪暗凋旗画,风多杂鼓声。
宁为百夫长,胜作一书生。
- 通过 | 链接提示词模板对象和模型对象
- 返回值chain对象是RunnableSerializable对象
** 是Runnable接口的直接子类
** 也是绝大多数组件的父类 - 通过invoke或stream进行阻塞执行或流式执行
组成的链在执行上有:上一个组件的输出作为下一个组件的输入的特性。
所以有如下执行流程:

image.png