三件套之所以能用 | 无缝拼接,是因为它们遵守同一份 Runnable 接口契约:上一环的输出类型恰好是下一环的输入类型。Prompt 吃 dict、吐 PromptValueChatModel 吃 PromptValue/消息、吐 AIMessageOutputParser 吃 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(...).contentstr自由问答、闲聊、翻译
要强类型对象model.with_structured_output(Schema)Pydantic / dict抽取字段、分类打标、表单填充
要边出边显示model.stream(...)AIMessageChunk 流聊天 UI、长文生成
要批量并发model.batch([...])list离线跑一批样本
口诀文本用 content,结构用 structured,体验用 stream,吞吐用 batch

OutputParser:把回复收敛成可用类型

模型默认吐 AIMessage,下游代码却想要干净的 strdict 或强类型对象。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 问答系统。