Skip to content

消息、提示模板与 Runnable 数据流 ​

LangChain core 定义消息、Runnable 和工具协议,供应商包负责模型适配,LangGraph 负责状态编排,LangSmith 负责追踪与评估。统一接口减少接线工作,但不抹平不同模型的工具、图片、上下文窗口和计费能力。

环境与第一条真实链 ​

bash
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install 'langchain==1.2.10' 'langchain-core==1.6.4' 'langchain-openai==1.1.10' 'langgraph==1.0.8' 'langgraph-prebuilt==1.0.7' 'langchain-classic==1.0.1'

完整实验不调用模型。RunnableLambda 包装普通函数,| 将左侧输出传给右侧;ChatPromptTemplate 输出消息结构,不能把它误当普通字符串。

python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda, RunnableParallel
from langchain_core.messages import HumanMessage
prompt = ChatPromptTemplate.from_messages([
    ("system","使用{language}回答。"),
    MessagesPlaceholder("history"),
    ("human","{question}")])
render = prompt | RunnableLambda(lambda p:[(m.type,m.content) for m in p.to_messages()])
value = {"language":"中文","history":[HumanMessage(content="上下文")],"question":"幂等是什么"}
r = render.invoke(value)
assert r[0] == ("system","使用中文回答。") and r[-1][1] == "幂等是什么"
parallel = RunnableParallel(length=RunnableLambda(len), upper=RunnableLambda(str.upper))
assert parallel.invoke("ab") == {"length":2,"upper":"AB"}
assert render.batch([value,value]) == [r,r]
print(r)

变量不存在会在模板阶段报错;历史必须是消息列表,不能把工具结果和用户消息合成一段失去角色的信息。文字、图像等内容块是模型输入的一部分,供应商支持不同;响应的 metadata 与 usage 单独用于记账,不能从文本长度反推真实 token。

invoke、batch 与流式的区别 ​

invoke 处理一项,ainvoke 供异步服务等待,batch 多项并发但受 Runnable config 与下游配额约束。stream 输出块,是否逐 token 取决于组件;某个中间函数必须等完整输入时,会缓冲并打断端到端流式。不要看到方法叫 stream 就保证首 token 延迟改善。

RunnableParallel 给各分支同一输入并以键收集结果,不负责跨系统事务。RunnablePassthrough 保留输入,assign 添加派生字段时应避免覆盖身份字段。重试应包住会失败的外部调用,不要把带副作用的整条链随意重跑;fallback 需要保持相同输出契约并标注降级来源。

LCEL 是数据流组合方式,不是新的模型能力。复杂暂停、循环和检查点进入 LangGraph;一个普通请求若只调用一次模型,裸 SDK 可能更容易维护。排错先单独 invoke 每个组件,观察实际类型,再检查配置与回调传播。

练习 ​

在parallel增加reverse分支,输入abc应返回长度3、大写ABC、反转cba;再给prompt遗漏question,确认发生KeyError而非静默生成空问题。

核对本章答案

下一章

用心记录每一步成长