LangChain Model I/O:Prompt、模型、输出解析
每个 LLM 应用都在做三件事:构造 Prompt、调用模型、处理输出。听起来简单,但当项目里有 15 个 AI 调用点,你就会发现这三件事有多容易失控。
Prompt 开始是简单的 f-string,然后越改越长,然后有人加了 if/else,然后有多语言需求,然后有人复制了一份改了几行但忘了同步……最后没人知道哪个版本是"正确的"。
模型输出更难搞。你让它返回 JSON,它今天返回 {"name": "张三"},明天突发善意在前面加了"以下是分析结果:\n",后天把所有 key 从英文改成了中文。你的 json.loads() 会在最不合时宜的时候崩溃。
LangChain 的 Model I/O 就是专门管这三件事的。
整体流程
你的变量输入
│
▼
PromptTemplate ← 把变量填进模板,生成标准化 Prompt
│
▼
BaseChatModel ← 统一接口调用任意模型(OpenAI / Claude / Gemini)
│
▼
OutputParser ← 把模型返回的字符串转成你需要的 Python 对象
│
▼
你的代码逻辑
这三步完全独立。换模型不影响 Prompt 模板,换输出格式不影响模型调用。这是 LangChain 和直接拼字符串的根本区别。
PromptTemplate:别再写 f-string 了
用 f-string 拼 Prompt 没什么问题,直到某天你的 Prompt 需要出现在 10 个地方,需要支持 A/B 测试,或者需要版本管理。那时候你会想要一个专门管理 Prompt 的东西。
ChatPromptTemplate 就是这个东西:
from langchain_core.prompts import ChatPromptTemplate
template = ChatPromptTemplate.from_messages([
("system", "你是一位专业的{role},回答要{style}。"),
("user", "{question}"),
])
# 填入变量,返回可以直接传给模型的 messages
messages = template.invoke({
"role": "Python 老师",
"style": "简洁,附带代码示例",
"question": "什么是生成器?",
})
注意这里有个坑:PromptTemplate(单条)和 ChatPromptTemplate(多条 messages)是不同的类。多轮对话一定要用 ChatPromptTemplate,用错了会有奇怪的格式报错。
如果你做的是多轮对话,还需要在模板里留一个历史记录的位置:
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
template = ChatPromptTemplate.from_messages([
("system", "你是一个友好的技术助手。"),
MessagesPlaceholder(variable_name="chat_history"), # 历史消息注入点
("user", "{question}"),
])
MessagesPlaceholder 就是一个槽位,运行时把历史对话列表填进去。Memory 那一页会详细讲怎么管理这个历史。
另外一个有用的功能:如果某些变量是固定的,可以用 partial() 提前填好,避免每次调用都传同样的东西:
# format_instructions 是 OutputParser 自动生成的,提前填进去
simplified_template = template.partial(
role="Python 专家",
format_instructions="输出必须是合法 JSON"
)
# 之后只需要传 question
result = simplified_template.invoke({"question": "什么是闭包?"})
统一模型接口:换模型改一行的秘密
LangChain 最早吸引我的功能就是这个——所有模型用同一套调用方式。
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI
# 三家不同的模型
llm_gpt = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_claude = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)
llm_gemini = ChatGoogleGenerativeAI(model="gemini-1.5-flash", temperature=0)
# 调用方式完全一样
for llm in [llm_gpt, llm_claude, llm_gemini]:
response = llm.invoke("Python 的 GIL 是什么?")
print(response.content)
之前那个帮客户换模型的故事,如果用了 LangChain,真的只需要改一行:把 ChatOpenAI(...) 换成 ChatAnthropic(...),完成。
几个常用的初始化参数:
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0, # 0 = 稳定输出,0.7 = 有创意,1.0 = 最随机
max_tokens=1000, # 限制输出长度,直接控制成本
timeout=30, # 超时秒数,别设太短
max_retries=3, # 失败自动重试
)
关于 temperature:做信息抽取、分类、解析结构化数据,一定设 temperature=0——你需要结果稳定复现,不需要创意。写作、头脑风暴类任务才调高。
还有一个功能我在生产环境用过:Fallbacks,主模型挂了自动切备用:
primary = ChatOpenAI(model="gpt-4o")
backup = ChatAnthropic(model="claude-3-5-sonnet-20241022")
# gpt-4o 超时或报错时,自动试 Claude
llm_with_fallback = primary.with_fallbacks([backup])
OpenAI 偶尔会有服务中断,这个设置让你的应用不至于完全挂掉。
OutputParser:解决 JSON 解析崩溃问题
我个人最推荐的 Parser 组合是 JsonOutputParser + Pydantic 类。
为什么不直接用 StrOutputParser 然后手动 json.loads()?因为当模型心情不好时,它会在 JSON 前后加 markdown 代码块(````json...```)、解释性文字、甚至直接换一种格式。json.loads() 不能处理这些情况,而 JsonOutputParser 可以。
但 JsonOutputParser 也不是万能的。如果你想让模型知道它应该输出什么字段,最好用 Pydantic 类定义数据结构,让 Parser 自动生成格式说明:
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List, Optional
# 定义你期望的数据结构
class JobPosting(BaseModel):
company: str = Field(description="公司名称")
position: str = Field(description="职位名称")
salary_range: str = Field(description="薪资范围,如 '15k-25k'")
requirements: List[str] = Field(description="岗位要求,3-5条")
is_remote: bool = Field(description="是否支持远程")
parser = JsonOutputParser(pydantic_object=JobPosting)
# parser.get_format_instructions() 会自动生成 JSON schema 说明
prompt = ChatPromptTemplate.from_messages([
("system", "分析招聘信息并按格式输出。\n{format_instructions}"),
("user", "{job_text}"),
]).partial(format_instructions=parser.get_format_instructions())
chain = prompt | ChatOpenAI(model="gpt-4o-mini", temperature=0) | parser
result = chain.invoke({
"job_text": "字节跳动招高级前端工程师,3年React经验,薪资25k-40k,支持远程"
})
print(result["company"]) # 字节跳动
print(result["is_remote"]) # True
常用 Parser 对比,直接给结论:
| Parser | 输出类型 | 什么时候用 | 一句话 |
|---|---|---|---|
StrOutputParser | str | 普通问答、对话 | 最简单,默认用它 |
JsonOutputParser | dict | 需要结构化数据 | 加上 Pydantic 类会更稳 |
PydanticOutputParser | Pydantic 对象 | 需要 Python 类型,而不是 dict | 类型安全要求高时 |
CommaSeparatedListOutputParser | list[str] | 让模型输出简单列表 | 适合标签、关键词提取 |
LCEL:用 | 把它们串起来
chain = prompt | llm | parser
这是 LangChain 的核心语法,叫 LCEL(LangChain Expression Language)。
第一次见到 | 这个写法我以为是 Python 的黑魔法,其实就是运算符重载——LangChain 的组件都实现了 __or__ 方法,让你可以用管道符把它们连起来。效果是把三个独立对象变成一个可以统一调用的 Chain。
这个 Chain 自动支持四种调用方式,不用额外代码:
# 普通调用
result = chain.invoke({"role": "老师", "question": "什么是闭包?"})
# 流式输出(适合前端实时显示)
for chunk in chain.stream({"role": "老师", "question": "什么是闭包?"}):
print(chunk, end="", flush=True)
# 异步调用(配合 FastAPI)
result = await chain.ainvoke({"role": "老师", "question": "什么是闭包?"})
# 批量并发处理(比 for 循环 invoke 快多了)
results = chain.batch([
{"role": "老师", "question": "什么是闭包?"},
{"role": "老师", "question": "什么是装饰器?"},
{"role": "老师", "question": "什么是生成器?"},
])
做 RAG 时经常需要把检索出来的文档和原始问题一起传给模型,这时候用 RunnablePassthrough 透传原始输入:
from langchain_core.runnables import RunnablePassthrough
rag_chain = (
{
"context": retriever | format_docs, # 检索相关文档
"question": RunnablePassthrough(), # 原始问题原样传过去
}
| prompt
| llm
| StrOutputParser()
)
几个常踩的坑
JSON 解析失败:报 json.decoder.JSONDecodeError 时,十有八九是模型在 JSON 外面加了 markdown 代码块。在 Prompt 里加一句"只输出 JSON,不要 Markdown 格式"能解决 90% 的情况。JsonOutputParser 比裸的 json.loads() 容错性好一些,但也不是百分百可靠。
换模型后输出格式变了:不同模型对同一个 Prompt 的理解有差异,Claude 和 GPT 有时对格式要求的遵守程度不同。换模型后要重新测试所有格式要求严格的 Prompt。
batch() 没有并发效果:LangChain 的 batch() 默认是并发的,但如果 API 速率限制触发,反而比串行慢。可以设 config={"max_concurrency": 5} 控制并发数。
ChatPromptTemplate 报变量未找到:报错信息会告诉你缺哪个变量(missing required variable: 'xxx'),对照检查一下拼写和 invoke() 时传的 key 是否一致。
动手练习
写一个简历信息提取器:输入一段非结构化的简历文本,输出结构化的 JSON:
from pydantic import BaseModel, Field
from typing import List, Optional
class ResumeInfo(BaseModel):
name: str = Field(description="候选人姓名")
years_exp: int = Field(description="工作年限(整数)")
skills: List[str] = Field(description="技术技能列表")
expected_salary: Optional[str] = Field(description="期望薪资,没有则为 null")
# 测试文本:
resume_text = """
张三,5 年软件开发经验。
熟练掌握 Python、FastAPI、PostgreSQL、Redis。
有 AI 应用开发经验,用过 LangChain 和 OpenAI API。
期望薪资 25-35k,可远程。
"""
# TODO: 用 JsonOutputParser + ChatPromptTemplate + ChatOpenAI 组成 Chain
# 提示:用 parser.get_format_instructions() 把格式说明注入 Prompt
进阶:改造这个 Chain,批量处理 10 份简历,统计哪些技能出现频次最高。
小结
- LangChain 把 LLM 应用的三步标准化:PromptTemplate 统一管理 Prompt,BaseChatModel 统一调用任意模型,OutputParser 自动解析输出——三者独立可替换。
- 多轮对话用
ChatPromptTemplate,不要用PromptTemplate,用错了格式会出问题。用MessagesPlaceholder留历史消息注入点。 JsonOutputParser+ Pydantic 类是目前最稳健的结构化输出方案——Pydantic 定义结构,Parser 自动生成格式说明。temperature=0用于信息抽取、分类等需要稳定输出的任务,不要在结构化输出任务上用高 temperature。- LCEL 的
|语法让 Chain 自动支持同步/流式/异步/批量四种调用——不用额外代码,改方法名就行。
下一步:Memory 记忆系统 — 让你的 Chain 记住上下文,变成真正的多轮对话助手
相关参考:LCEL 官方文档 | 所有 Output Parsers