LangChain 的学习与应用（二）：输出解析与链式编排

在上一篇里，我们解决了“怎么调模型”：初始化、消息格式、同步异步、流式、批量。但模型返回的始终是一段自由文本，而真实的程序需要的是能直接使用的数据结构；并且一个稍复杂的任务往往要经过“拼提示词 → 调模型 → 解析结果”好几步。这一篇就解决这两个问题：让输出结构化，以及把步骤编排成链。

一、为什么需要输出解析

上一篇里所有的调用都是这样收尾的：

response = model.invoke("...")
print(response.content)  # 一段自由文本

打印出来给人看没问题，但程序没法消费自由文本。想象一个场景：让模型从一段话里提取日程信息，你希望拿到的是 event.name、event.date 这样能直接访问的字段，而不是“好的！这段话中提到的日程是……”这样一段话，还得自己写正则去抠。

让模型输出结构化数据，主流有两条路：

提示词约束：在提示词里明确要求模型“只输出 JSON，格式如下”，拿到文本后再解析成字典——对任何模型都有效；
厂商原生结构化输出：利用厂商 API 自带的结构化能力，由服务端保证输出符合给定 schema——更可靠，但要求模型支持。

下面分别看这两种。

二、方式一：JsonOutputParser——提示词约束 JSON

自己在提示词里手写“请输出 JSON，包含 xx 字段……”当然可以，但格式说明写得不严谨，模型就容易跑偏。JsonOutputParser 帮你把这段提示词生成出来：先用 Pydantic 类描述目标结构，解析器会据此生成一份严格的格式说明。

from typing import List
from pydantic import BaseModel
from langchain_core.output_parsers import JsonOutputParser

# 1. 用 Pydantic 类描述你想要的结构
class Star(BaseModel):
    name: str
    age: int

class StarList(BaseModel):
    stars: List[Star]

# 2. 创建解析器，并让它生成格式说明提示词
parser = JsonOutputParser(pydantic_object=StarList)
format_instructions = parser.get_format_instructions()
# 内容大意：输出必须是符合以下 JSON schema 的实例……（附上完整 schema）

# 3. 把格式说明放进 system 消息，正常提问
messages = [
    {"role": "system", "content": format_instructions},
    {"role": "user", "content": "请列出几位著名的足球明星"},
]
response = model.invoke(messages)

# 4. 把返回的 JSON 文本解析成 Python 字典
result = parser.parse(response.content)
names = [star["name"] for star in result["stars"]]
print(names)  # ['梅西', 'C罗', ...]

整个流程是“生成提示词 → 模型照做 → 解析文本”。它的优点是通用——哪怕模型不支持任何结构化输出特性，只要它听得懂提示词就能用；缺点是本质上靠模型自觉，偶尔会输出不合法的 JSON，生产环境需要做好解析失败的兜底。

三、方式二：with_structured_output——厂商原生结构化

如果模型厂商本身支持结构化输出（OpenAI、Anthropic、Gemini 等主流厂商都支持），更推荐用 with_structured_output()。它把 schema 直接传给厂商 API，由服务端约束输出，返回的不再是消息对象，而是直接可用的 Pydantic 对象：

from pydantic import BaseModel

class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]

# 用 schema 包装模型，得到一个新的可调用对象
structured_model = model.with_structured_output(CalendarEvent)

event = structured_model.invoke([
    {"role": "system", "content": "请从以下内容中提取日程信息"},
    {"role": "user", "content": "李明和韩梅梅星期五要去博物馆"},
])

# 直接访问字段，带类型校验，不需要再 parse
print(event.name)          # 博物馆参观
print(event.date)          # 星期五
print(event.participants)  # ['李明', '韩梅梅']

两种方式怎么选，一张表说清楚：

	JsonOutputParser	with_structured_output
原理	提示词约束，客户端解析	厂商 API 原生支持，服务端约束
可靠性	靠模型自觉，可能输出坏 JSON	高，schema 由服务端保证
返回类型	Python 字典	Pydantic 对象（带类型校验）
适用范围	任何模型	需要模型支持结构化输出

结论：模型支持就用 with_structured_output，不支持（比如一些本地小模型）再退回 JsonOutputParser。

四、一切皆 Runnable：链的前置知识

进入链式编排之前，先看一个贯穿 LangChain 的设计：所有核心组件都实现了统一的 Runnable 接口，都有 invoke 方法。

上一篇的提示词模板有 invoke，模型有 invoke，这一篇的输出解析器也有 invoke。于是一个“生成提示词 → 调模型 → 解析结果”的完整流程，可以写成三次接力：

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt_template = PromptTemplate.from_template("写一段关于{keyword}的介绍")
parser = StrOutputParser()  # 最简单的解析器：从消息中取出纯文本

# 三步接力：每一步的输出是下一步的输入
prompt = prompt_template.invoke({"keyword": "人工智能"})
response = model.invoke(prompt)
text = parser.invoke(response)
print(text)

能跑，但样板味很重：三个中间变量只是把数据从上一步搬到下一步。既然每个组件的接口都一样，这种“搬运”完全可以交给框架——这就是链。

五、串行链：一个管道符搞定

LangChain 重载了 | 运算符（这套语法叫 LCEL，LangChain Expression Language），把上面三步压缩成一行声明：

chain = prompt_template | model | parser

text = chain.invoke({"keyword": "人工智能"})
print(text)

数据像水流过管道：输入字典流进模板变成提示词，提示词流进模型变成消息，消息流进解析器变成字符串。提示词模板、模型、输出解析器就是链的三要素，而且组合出来的链本身也是一个 Runnable——有 invoke，也有上一篇讲过的 stream、batch、ainvoke，还能继续参与更大的链的编排。这是理解后面并行链、混合链的关键。

六、并行链：一份输入，多路处理

有时候需要对同一份输入做多种独立处理，比如把一句话同时翻译成英文和韩文。两条链彼此无依赖，串行执行就浪费了。RunnableParallel 让它们并发跑：

from langchain_core.runnables import RunnableParallel

prompt_en = PromptTemplate.from_template("把这句话翻译成英文：{topic}")
prompt_ko = PromptTemplate.from_template("把这句话翻译成韩文：{topic}")

chain_en = prompt_en | model | parser
chain_ko = prompt_ko | model | parser

# 两条链并行执行，结果按 key 汇总成字典
parallel = RunnableParallel(en=chain_en, ko=chain_ko)

result = parallel.invoke({"topic": "我爱你"})
print(result["en"])  # I love you
print(result["ko"])  # 사랑해요

输入被同时分发给两条子链，返回值是一个字典，key 就是你给每条子链起的名字。

七、混合链：串行与并行的组合

真正体现编排能力的是把串行和并行组合起来。一个经典场景：让两个不同的模型分别完成同一任务，再让第三个模型对比总结两者的结果——多模型交叉验证就是这么搭的：

model_a = init_chat_model("openai:gpt-4o-mini")
model_b = init_chat_model("anthropic:claude-sonnet-4-5")
model_judge = init_chat_model("openai:gpt-4o")

prompt_analyze = PromptTemplate.from_template("请赏析这首诗：{poem}")
prompt_judge = PromptTemplate.from_template(
    "以下是对同一首诗的两种赏析，请对比并总结。\n第一种：{result_a}\n第二种：{result_b}"
)

chain_a = prompt_analyze | model_a | parser
chain_b = prompt_analyze | model_b | parser
chain_judge = prompt_judge | model_judge | parser

# 字典部分并行执行，产出的结果字典正好填进下游模板的两个槽位
chain = {"result_a": chain_a, "result_b": chain_b} | chain_judge

summary = chain.invoke({"poem": "白日依山尽，黄河入海流。"})
print(summary)

注意中间那个字典：在管道中，字典会被自动当作 RunnableParallel 处理——两条赏析链并行执行，输出 {"result_a": ..., "result_b": ...}，恰好对上 prompt_judge 需要的两个槽位，无缝流入下游。整个数据流是“一进、二分、再合一”，而代码只有一行编排逻辑。

八、换个口味：接入本地模型

链编排好之后，“模型”这个环节是可以随意替换的——包括换成跑在自己电脑上的本地模型。用 Ollama 在本地起一个模型服务后，接入方式和云端厂商没有任何区别：

pip install -U langchain-ollama

from langchain_ollama import ChatOllama

local_model = ChatOllama(
    model="qwen3:8b",                  # 本地已下载的模型
    base_url="http://127.0.0.1:11434", # Ollama 默认端口
)

# 依然是熟悉的 invoke，也能直接替换掉链中的云端模型
chain = prompt_template | local_model | parser

这再次印证了统一接口的价值：无论模型在云端还是本地、来自哪家厂商，在链里都只是一个可插拔的 Runnable。数据不出本机的隐私场景、离线环境、零成本实验，都可以走这条路。

九、小结与下一步

这一篇补上了 Model I/O 的最后一块拼图（输出解析），并正式进入了第二大模块 Chains：

JsonOutputParser：生成格式约束提示词 + 客户端解析，任何模型都能用；
with_structured_output：厂商原生结构化输出，直接返回 Pydantic 对象，支持就优先用它；
Runnable 统一接口：模板、模型、解析器都有 invoke，这是链的基石；
串行链：prompt | model | parser，数据顺管道流动，链本身也是 Runnable；
并行链：RunnableParallel 让无依赖的子链并发执行，结果按 key 汇总；
混合链：管道中的字典自动并行，与下游模板的槽位衔接，串并自由组合；
本地模型：ChatOllama 无缝替换云端模型，链的结构不用动。

到这里，“输入构建 → 模型调用 → 输出解析 → 流程编排”这条链路已经完整。但模型还有一个绕不开的局限：它只知道训练时见过的东西，不知道你的私有文档，知识也有截止日期。解决方案就是四大模块的第三个——RAG（检索增强生成）：把你的文档切分、向量化、存入向量库，回答问题前先检索出最相关的内容喂给模型。这将是本系列下一篇的主题。

老规矩，以官方文档为准，把每段代码亲手跑一遍。