LangChain 结构化输出：OutputParser 实战（JSON & Pydantic 一网打尽）

版本：Python 3.9+、LangChain 0.1.x
目标：定义好模型，一行代码返回类型安全的对象，告别正则切字符串

---

0. 前置安装

pip install -U langchain langchain-openai python-dotenv pydantic

.env 文件沿用：

OPENAI_API_KEY=sk-xxx

# 0_env.py
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

---

1. PydanticV2BaseModel · 最推荐写法

文件：1_pydantic_v2.py

from pydantic import BaseModel, Field
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import ChatPromptTemplate
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

class Book(BaseModel):
    title: str = Field(description="书名")
    price: float = Field(description="价格，单位元")
    in_stock: bool = Field(description="现货与否")

parser = PydanticOutputParser(pydantic_object=Book)

prompt = ChatPromptTemplate.from_messages([
    ("system", "用户说一句话，你提取出书名、价格、现货情况，按 JSON 返回。\n{format}"),
    ("user", "{sentence}")
]).partial(format=parser.get_format_instructions())

chain = prompt | llm | parser
print(chain.invoke({"sentence": "《Python 编程》现在 59.8 元，现货充足"}))
# 输出：Book(title='Python 编程', price=59.8, in_stock=True)

---

2. 裸 JSON 无 schema · 最快验证

文件：2_json_dict.py

from langchain.output_parsers import JsonOutputParser
from langchain.prompts import ChatPromptTemplate
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

parser = JsonOutputParser()
prompt = ChatPromptTemplate.from_messages([
    ("system", "返回纯 JSON，不要解释。\n{format}"),
    ("user", "用一句话介绍北京，字段：city, population, slogan")
]).partial(format=parser.get_format_instructions())

chain = prompt | llm | parser
print(chain.invoke({}))
# 输出：{'city': '北京', 'population': 2189万, 'slogan': '京彩之城'}

---

3. 带“修复”功能 · 输出不对自动重试

文件：3_retry_parser.py

from langchain.output_parsers import PydanticOutputParser, OutputFixingParser
from pydantic import BaseModel, Field
from langchain.prompts import ChatPromptTemplate
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

class Color(BaseModel):
    name: str
    hex_code: str = Field(regex=r'^#[0-9a-fA-F]{6}$')

parser = PydanticOutputParser(pydantic_object=Color)
# 包装一层自动修复
retry_parser = OutputFixingParser.from_llm(parser=parser, llm=llm)

prompt = ChatPromptTemplate.from_messages([
    ("system", "返回颜色信息，JSON 格式。\n{format}"),
    ("user", "随机一个颜色")
]).partial(format=parser.get_format_instructions())

chain = prompt | llm | retry_parser
print(chain.invoke({}))
# 如果模型返回 hex 少一位，retry_parser 会自动调用 llm 修复

---

4. 列表嵌套 · 一次返回多条记录

文件：4_list_nested.py

from typing import List
from pydantic import BaseModel, Field
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import ChatPromptTemplate
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

class Phone(BaseModel):
    brand: str
    price: float

class Catalog(BaseModel):
    phones: List[Phone] = Field(description="手机列表")

parser = PydanticOutputParser(pydantic_object=Catalog)
prompt = ChatPromptTemplate.from_messages([
    ("system", "返回 JSON，列表嵌套。\n{format}"),
    ("user", "列出 2 款热门手机，含品牌与价格")
]).partial(format=parser.get_format_instructions())

chain = prompt | llm | parser
print(chain.invoke({}))
# 输出：Catalog(phones=[Phone(brand='iPhone15', price=5999.0), Phone(brand='小米14', price=3999.0)])

---

5. 流式解析 · 大列表边返回边解析（实验）

文件：5_stream_list.py

from langchain.output_parsers import JsonOutputParser
from langchain.prompts import ChatPromptTemplate
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, streaming=True)

parser = JsonOutputParser()
prompt = ChatPromptTemplate.from_messages([
    ("system", "返回 JSON 数组，每个元素字段：id, task, done"),
    ("user", "生成 3 条今日待办")
])

chain = prompt | llm | parser
# 虽然底层是流式，但解析器会等完整 JSON 结束再返回
print(chain.invoke({}))

---

6. 一句话选型速览

需求	直接用
返回单个对象、字段强类型	PydanticOutputParser
只要裸 dict/list	JsonOutputParser
怕模型输出格式歪	OutputFixingParser 包一层
嵌套列表	Pydantic 里写 List[SubModel]

---

7. 常见坑

1. JsonOutputParser 要求模型返回纯 JSON，任何多余解释都会抛错
2. Pydantic 正则字段失败时，先让模型重试，再不行才手动改 schema
3. 流式场景下，目前解析器仍会等完整块结束，不会逐 token 解析
4. 如果模型总输出 Markdown 代码块，可在 prompt 里加“不要 ```json 包裹”

---

完结撒花！现在你可以

1. 定义 Pydantic 模型
2. 一行链式调用 chain.invoke(...)
   拿到类型安全的 Python 对象，彻底告别切字符串。
   点赞 + 收藏，CSDN 不迷路，下篇见！