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 就是这个东西：

python
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，用错了会有奇怪的格式报错。

如果你做的是多轮对话，还需要在模板里留一个历史记录的位置：

python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

template = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的技术助手。"),
    MessagesPlaceholder(variable_name="chat_history"),  # 历史消息注入点
    ("user", "{question}"),
])

MessagesPlaceholder 就是一个槽位，运行时把历史对话列表填进去。Memory 那一页会详细讲怎么管理这个历史。

另外一个有用的功能：如果某些变量是固定的，可以用 partial() 提前填好，避免每次调用都传同样的东西：

python
# format_instructions 是 OutputParser 自动生成的，提前填进去
simplified_template = template.partial(
    role="Python 专家",
    format_instructions="输出必须是合法 JSON"
)

# 之后只需要传 question
result = simplified_template.invoke({"question": "什么是闭包？"})

#统一模型接口：换模型改一行的秘密

LangChain 最早吸引我的功能就是这个——所有模型用同一套调用方式。

python
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(...)，完成。

几个常用的初始化参数：

python
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，主模型挂了自动切备用：

python
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 自动生成格式说明：

python
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：用 `|` 把它们串起来

python
chain = prompt | llm | parser

这是 LangChain 的核心语法，叫 LCEL（LangChain Expression Language）。

第一次见到 | 这个写法我以为是 Python 的黑魔法，其实就是运算符重载——LangChain 的组件都实现了 __or__ 方法，让你可以用管道符把它们连起来。效果是把三个独立对象变成一个可以统一调用的 Chain。

这个 Chain 自动支持四种调用方式，不用额外代码：

python
# 普通调用
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 透传原始输入：

python
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：

python
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↗

LangChain Model I/O：Prompt、模型、输出解析

#整体流程

#PromptTemplate：别再写 f-string 了

#统一模型接口：换模型改一行的秘密

#OutputParser：解决 JSON 解析崩溃问题

#LCEL：用 `|` 把它们串起来

#几个常踩的坑

#动手练习

#小结

Core system design concepts and practical case studies

Related Guides

Related Roadmaps

LangChain Model I/O：Prompt、模型、输出解析

#整体流程

#PromptTemplate：别再写 f-string 了

#统一模型接口：换模型改一行的秘密

#OutputParser：解决 JSON 解析崩溃问题

#LCEL：用 | 把它们串起来

#几个常踩的坑

#动手练习

#小结

Core system design concepts and practical case studies

Related Guides

Related Roadmaps

#LCEL：用 `|` 把它们串起来