三件套之所以能用
|无缝拼接,是因为它们遵守同一份 Runnable 接口契约:上一环的输出类型恰好是下一环的输入类型。Prompt 吃 dict、吐 PromptValue;ChatModel 吃 PromptValue/消息、吐 AIMessage;OutputParser 吃 AIMessage、吐目标 Python 类型。类型对齐,管道就成立。
环境准备:一次装齐
# 核心包 + OpenAI 集成(任选一家模型厂商,这里以 OpenAI 为例)
pip install -U langchain langchain-core langchain-openai
# 设置 API key(也可写进 .env 后用 python-dotenv 加载)
export OPENAI_API_KEY="sk-..."
# 如果用 Anthropic:pip install -U langchain-anthropic,并 export ANTHROPIC_API_KEY
Prompt:从 PromptTemplate 到 ChatPromptTemplate
PromptTemplate 面向「单段补全文本」,ChatPromptTemplate 面向「带角色的多条消息」。现代 ChatModel 几乎都是对话式的,所以默认用 ChatPromptTemplate.from_messages 这条路。下面把单模板、多角色模板、MessagesPlaceholder 历史占位、few-shot 一次讲透。
from langchain_core.prompts import (
PromptTemplate,
ChatPromptTemplate,
MessagesPlaceholder,
)
from langchain_core.messages import HumanMessage, AIMessage
# 1) 最朴素:单段文本模板(补全式模型时代的写法,了解即可)
tpl = PromptTemplate.from_template("把下面这句话翻译成{lang}:{text}")
print(tpl.format(lang="英文", text="今天天气不错"))
# -> 把下面这句话翻译成英文:今天天气不错
# 2) 推荐:多角色对话模板
# 每个元组是 (role, template_string),role ∈ system/human/ai
chat_tpl = ChatPromptTemplate.from_messages([
("system", "你是一名严谨的翻译,只输出译文,不要解释。"),
MessagesPlaceholder(variable_name="history"), # 多轮历史的占位槽
("human", "把这句话翻译成{lang}:{text}"),
])
# 渲染时 history 传入一个消息列表(可为空 [])
messages = chat_tpl.invoke({
"lang": "英文",
"text": "今天天气不错",
"history": [
HumanMessage("上一句叫我翻成日文"),
AIMessage("今日はいい天気ですね"),
],
})
for m in messages.to_messages():
print(m.type, "->", m.content)
# system -> 你是一名严谨的翻译...
# human -> 上一句叫我翻成日文
# ai -> 今日はいい天気ですね
# human -> 把这句话翻译成英文:今天天气不错
from langchain_core.prompts import (
ChatPromptTemplate,
FewShotChatMessagePromptTemplate,
)
# few-shot:把几组示范 (输入->输出) 注入对话,让模型照葫芦画瓢
examples = [
{"input": "2 + 2", "output": "4"},
{"input": "2 + 3", "output": "5"},
]
# 单条示范的渲染模板:一个 human + 一个 ai
example_prompt = ChatPromptTemplate.from_messages([
("human", "{input}"),
("ai", "{output}"),
])
few_shot = FewShotChatMessagePromptTemplate(
example_prompt=example_prompt,
examples=examples,
)
final_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个只回答最终数字的计算器。"),
few_shot, # 示范会被展开成多条 human/ai 消息
("human", "{question}"),
])
print(final_prompt.invoke({"question": "3 + 5"}).to_messages())
# 你会看到 system + 两组示范 + 最后的 human,共 6 条消息
ChatModel:消息类型与关键参数
ChatModel 接收消息列表(SystemMessage / HumanMessage / AIMessage),返回一条 AIMessage。最常调的两个参数:temperature 控制随机性(0 趋向确定、稳定可复现;越高越发散),max_tokens 限制生成的最大 token 数(不含输入)。
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
model = ChatOpenAI(
model="gpt-4o-mini", # 模型名按厂商实际可用值填
temperature=0, # 0 = 尽量确定性输出,适合抽取/翻译类任务
max_tokens=512, # 限制回复长度,控成本、防失控
timeout=30, # 单次请求超时秒数
max_retries=2, # 失败自动重试次数
)
# 直接传消息列表即可调用,返回一条 AIMessage
resp = model.invoke([
SystemMessage("你是简洁的助手,一句话作答。"),
HumanMessage("用一句话解释什么是 LangChain。"),
])
print(type(resp).__name__) # AIMessage
print(resp.content) # -> LangChain 是一个用于构建 LLM 应用的开发框架。
print(resp.usage_metadata) # -> {'input_tokens':..., 'output_tokens':..., ...}
| 需求 | 用法 | 返回类型 | 适用场景 |
|---|---|---|---|
| 要原始文本 | model.invoke(...).content | str | 自由问答、闲聊、翻译 |
| 要强类型对象 | model.with_structured_output(Schema) | Pydantic / dict | 抽取字段、分类打标、表单填充 |
| 要边出边显示 | model.stream(...) | AIMessageChunk 流 | 聊天 UI、长文生成 |
| 要批量并发 | model.batch([...]) | list | 离线跑一批样本 |
OutputParser:把回复收敛成可用类型
模型默认吐 AIMessage,下游代码却想要干净的 str、dict 或强类型对象。OutputParser 就是这层「收敛器」。常用三档:StrOutputParser(取纯文本)、JsonOutputParser(解析成 dict)、PydanticOutputParser(解析并校验成 Pydantic 模型,配 format_instructions 把字段约束注入 prompt)。
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import (
StrOutputParser,
PydanticOutputParser,
)
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# --- 档位 A:StrOutputParser,最常见,链尾接它直接拿字符串 ---
str_chain = (
ChatPromptTemplate.from_messages([("human", "一句话介绍{topic}")])
| model
| StrOutputParser()
)
print(str_chain.invoke({"topic": "向量数据库"})) # 直接是 str,不用再 .content
# --- 档位 C:PydanticOutputParser,要强类型 + 字段校验时用 ---
class Person(BaseModel):
name: str = Field(description="人物姓名")
age: int = Field(description="年龄,整数")
skills: list[str] = Field(description="技能列表")
parser = PydanticOutputParser(pydantic_object=Person)
prompt = ChatPromptTemplate.from_messages([
("system", "你是信息抽取器。严格按格式输出。\n{format_instructions}"),
("human", "{text}"),
]).partial(
# 把 parser 生成的 JSON Schema 说明注入 system prompt
format_instructions=parser.get_format_instructions()
)
pydantic_chain = prompt | model | parser
result = pydantic_chain.invoke(
{"text": "张三今年28岁,擅长 Python、SQL 和数据分析。"}
)
print(type(result).__name__) # Person
print(result.name, result.age, result.skills)
# -> 张三 28 ['Python', 'SQL', '数据分析']
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
class Sentiment(BaseModel):
"""对一段评论的情感分析结果。"""
label: str = Field(description="positive / negative / neutral 之一")
score: float = Field(description="置信度 0~1")
reason: str = Field(description="判定理由,一句话")
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# with_structured_output 直接返回一个「会吐 Sentiment 对象」的 Runnable
structured_model = model.with_structured_output(Sentiment)
prompt = ChatPromptTemplate.from_messages([
("system", "你是评论情感分析器。"),
("human", "分析这条评论:{review}"),
])
# 注意:这里链尾不再需要任何 OutputParser,模型本身已结构化
chain = prompt | structured_model
out = chain.invoke({"review": "物流很慢,但客服态度不错,整体一般。"})
print(type(out).__name__) # Sentiment
print(out.label, out.score) # -> neutral 0.7
print(out.reason) # -> 物流差但服务好,整体中性
✓推荐做法
- 默认用 ChatPromptTemplate.from_messages 写多角色 prompt
- 结构化输出首选 with_structured_output,把字段含义写进 Pydantic 的 Field(description=...)
- 抽取/分类/翻译类任务把 temperature 设 0,保证可复现
- 链尾根据下游需要选 parser:要 str 就 StrOutputParser,要对象就走结构化
✗不推荐
- 不要把对话历史拼成一坨字符串塞进 system prompt
- 不要在已用 with_structured_output 的链尾再接 OutputParser(重复且会报错)
- 不要用正则手撕模型返回的 JSON——交给 JsonOutputParser / Pydantic
- 不要忘了把 parser.get_format_instructions() 注入 prompt(用 PydanticOutputParser 时)
⚠常见误区
- format_instructions 没注入 prompt 时,PydanticOutputParser 几乎必然解析失败
- temperature 偏高会让 JSON 结构偶发漂移,结构化任务务必压低
- max_tokens 设太小会把 JSON 截断成非法片段,导致 parser 抛错
一条 prompt | model | parser 链能稳定 invoke 返回目标 Python 类型(str/dict/Pydantic),跑 10 次结构不漂移,即视为三件套打通。
OutputParserException:无法解析为 JSON
- 典型表现
- 用 PydanticOutputParser 时报
OutputParserException,模型回了带解释的文本而非纯 JSON - 判断标准
- parser.parse(模型原始输出) 能成功返回对象
- 解决方向
- 把 parser.get_format_instructions() 注入 system prompt,并在 system 里强调「只输出 JSON,不要任何额外文字」;更稳的做法是改用 with_structured_output。
KeyError / 缺变量
- 典型表现
- invoke 时报模板变量缺失,如缺 history 或 format_instructions
- 判断标准
- prompt.invoke(输入 dict) 不抛 KeyError
- 解决方向
- MessagesPlaceholder 的变量要在 invoke 时给(空对话传 []);固定不变的变量(如 format_instructions)用 prompt.partial(...) 预先绑定。
拿到的是 AIMessage 不是字符串
- 典型表现
- 下游想要 str,却拿到 AIMessage 对象,访问报错
- 判断标准
- chain.invoke(...) 直接返回 str
- 解决方向
- 链尾接 StrOutputParser(),即 prompt | model | StrOutputParser();它会自动取 AIMessage.content。
三件套打通后,你已经能写出任意「输入 → 提示 → 模型 → 结构化输出」的可运行链。下一章把这条链接上外部知识:通过文档加载、分块、向量检索,做出能回答私有资料的 RAG 问答系统。