当前位置：首页 > news >正文

LangChain 实战第 4 章：让模型输出“程序能直接用“的数据（结构化输出）

news 2026/7/31 4:15:23

LangChain 实战第 4 章：让模型输出"程序能直接用"的数据（结构化输出）

📌 系列第 4 篇。前三章我们把 Prompt 写得明明白白，模型也能乖乖回答了。但模型默认吐出来的是自然语言——人读着舒服，程序读着犯难。这一章解决的就是这个问题：把模型输出变成程序能直接resume.name、result.priority这样用的结构化数据。

一、本章目标

学完本章你应该能：

理解自然语言输出和结构化输出的区别
用StrOutputParser拿到字符串结果
用 Pydantic 定义输出结构
用PydanticOutputParser解析模型输出
用with_structured_output直接拿结构化对象
处理结构化输出解析失败的情况
完成简历信息抽取和商品评论分析两个实战案例

二、为什么需要结构化输出

假设系统要从一份简历里提取：姓名、工作年限、技能、目标岗位。

如果模型返回自然语言：

候选人姓名是张三，工作 3 年，熟悉 Python、FastAPI 和 MySQL，希望应聘后端开发工程师。

程序还得再写一堆正则/规则去"抠"字段。

如果模型返回结构化数据：

{ "name": "张三", "years_of_experience": 3, "skills": ["Python", "FastAPI", "MySQL"], "target_position": "后端开发工程师" }

程序就能直接读：

resume.name resume.skills

💡 结构化输出常见用途：信息抽取、文本分类、情感分析、工单分类、内容审核、数据入库、调用其他业务接口。一句话——凡是模型结果还要进业务流程的，都该用结构化输出。

三、Output Parser 是什么

Output Parser 专门负责"处理模型输出"。可以理解成一条流水线：

模型原始输出 → Output Parser → 程序可用的数据

本章重点讲三种方式：

方式	作用
`StrOutputParser`	把模型回复转换成字符串
`PydanticOutputParser`	把模型回复解析成Pydantic 对象
`with_structured_output`	让模型直接按指定结构返回

四、StrOutputParser：拿到纯文本

模型调用返回的通常是一个消息对象，不是裸字符串：

response = model.invoke("请介绍 LangChain") print(response.content) # 从 AIMessage 里取文本

StrOutputParser可以把这个消息对象转换成普通字符串：

from langchain_core.output_parsers import StrOutputParser parser = StrOutputParser() text = parser.invoke(response) print(text)

它适合：文案生成、内容总结、普通问答、翻译——只要业务只需要文本，不一定要上复杂的结构化。

🤔 思考一个问题：print(response.content)和parser.invoke(response)打印出来都是字符串，有啥区别？

图片无法显示

关键区别在这里：

model.invoke()返回的是AIMessage对象；StrOutputParser是 LangChain 标准解析器，传入AIMessage时等价于取.content，但它还支持多种输入类型，不止AIMessage。
最重要的实战差异：支持链式（Pipeline）拼接。LangChain 推崇用|把组件串成管道，管道里每个组件都得遵守统一协议（都继承自Runnable）。model、prompt、parser都是Runnable。

❌ 错误写法（管道里不能直接写.content，因为它是属性不是Runnable）：

chain = model | response.content

✅ 正确写法：

from langchain_core.output_parsers import StrOutputParser parser = StrOutputParser() chain = model | parser # Runnable 链式，官方标准 result = chain.invoke("请介绍 LangChain") print(result) # result 直接就是字符串

📌 这就是StrOutputParser最大的价值：作为 Runnable 参与链式拼接。如果只是单独调model.invoke()再打印，两者几乎没差别；一旦要搭链路，必须用解析器。

五、案例一：文本总结

创建01_text_summary.py：

from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from utils.model_factory import get_deepSeek_model model = get_deepSeek_model() chat_prompt = ChatPromptTemplate.from_messages([ ( "system", "你是一个内容编辑工程师，擅长提炼文本重点", ), ("human", """请将下面内容总结成一句话，不超过 50 字。 内容：{question}"""), ]) prompt = chat_prompt.invoke({ "question": "LangChain 是一个用于开发大模型应用的框架，" "提供模型调用、Prompt 管理、文档处理、检索和工具调用等能力。" }) resp = model.invoke(prompt) # 第一种写法 print(resp.content) # 第二种写法（也是建议写法） parser = StrOutputParser() text = parser.invoke(resp) print(text)

运行：

python 01_text_summary.py

这个案例最终得到的是普通字符串。

六、用 Pydantic 定义输出结构

Pydantic 用来定义"模型应该返回哪些字段、各是什么类型"。

📘 小知识：Pydantic里的Pydantic源自pedantic（/pɪˈdæntɪk/，意为"严谨教条的、拘泥规则的"）——正好契合它"严格校验数据"的脾气。

以简历信息为例：

from pydantic import BaseModel, Field class ResumeInfo(BaseModel): name: str = Field(description="候选人姓名") years_of_experience: int = Field(description="工作年限") skills: list[str] = Field(description="掌握的技术技能") target_position: str = Field(description="目标岗位")

这个模型既描述了字段类型，也描述了字段含义（通过description）。如果模型返回的数据不符合字段类型，解析时就会报错——这反而是好事，能帮你挡住脏数据。

七、PydanticOutputParser：让模型按格式返回

PydanticOutputParser能根据 Pydantic 模型自动生成格式要求，并解析模型的返回结果。

创建 Parser 并拿到格式说明：

from langchain_core.output_parsers import PydanticOutputParser parser = PydanticOutputParser(pydantic_object=ResumeInfo) format_instructions = parser.get_format_instructions()

把格式说明塞进 Prompt 的 system 里：

prompt_template = ChatPromptTemplate.from_messages([ ( "system", "你是一名招聘信息分析助手。\n{format_instructions}", ), ("human", "请从下面简历中提取信息：\n{resume_text}"), ])

最后解析：

result = parser.parse(response.content)

八、完整案例：简历信息抽取

创建02_resume_extractor.py：

from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field from utils.model_factory import get_deepSeek_model model = get_deepSeek_model() class ResumeInfo(BaseModel): name: str = Field(description="姓名") years_of_experience: int = Field(description="工作年限") skills: list[str] = Field(description="掌握的技术技能") target_position: str = Field(description="目标岗位") parser = PydanticOutputParser(pydantic_object=ResumeInfo) format_instructions = parser.get_format_instructions() template = ChatPromptTemplate.from_messages([ ( "system", """ 你是一名招聘信息分析助手。 请严格按照指定格式返回结果。 {format_instructions} """, ), ("human", "{resume_content}"), ]) resume_content = """ 我叫张三，我干大模型开发10年了，我擅长的技术是 python,langchain,fastapi等，我比较喜欢养猫养狗，我想找一份智能体开发的工作。 """ prompt = template.invoke({ "format_instructions": format_instructions, "resume_content": resume_content, }) # 调用大模型，拿到 AIMessage response = model.invoke(prompt) # 把大模型返回值解析成 Pydantic 对象 result = parser.invoke(response) print(result) print(result.name) print(result.years_of_experience) print(result.skills) print(result.target_position)

预期得到类似：

name='张三' years_of_experience=10 skills=['python', 'langchain', 'fastapi'] target_position='智能体开发'

九、with_structured_output：更简洁的方式

较新的 LangChain 模型组件直接提供了with_structured_output，让模型按 Pydantic 模型返回结构化结果：

structured_model = model.with_structured_output(ResumeInfo) result = structured_model.invoke("从简历中提取信息")

比"手动拿格式说明 + 调 Parser"简洁得多。是否支持、底层用哪种方式，取决于模型服务能力。

⚠️用 DeepSeek 的 OpenAI 兼容接口时，要加method="json_mode"：

structured_model = model.with_structured_output( ResumeInfo, method="json_mode", )

Prompt 里要明确要求模型返回 JSON。否则会报这个错：

openai.BadRequestError: Error code: 400 - {'error': {'message': 'This response_format type is unavailable now', ...}}

十、案例三：商品评论分析

本案例用with_structured_output分析一条商品评论。创建03_review_analyzer.py：

from typing import Literal from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field from utils.model_factory import get_deepSeek_model class ReviewAnalysis(BaseModel): sentiment: Literal["正面", "负面", "中性"] = Field(description="情感分析的值") keywords: list[str] = Field(description="评论中的关键词") summary: str = Field(description="对评论的简短总结") needs_reply: bool = Field(description="商家是否需要回复") model = get_deepSeek_model() template = ChatPromptTemplate.from_messages([ ( "system", """你是专业商品评论分析助手，严格遵守以下规则，仅输出纯JSON，无任何多余文字、解释、markdown： 1. 输出JSON必须包含4个字段：sentiment、keywords、summary、needs_reply，缺一不可； 2. sentiment 仅允许三个中文值：「正面」「中性」「负面」，绝对不能使用 mixed / positive / negative 等英文； 3. keywords 是字符串数组，提取评论核心描述词； 4. summary 用一句话概括整条评论优缺点； 5. needs_reply：商品存在质量问题、故障、严重不满设为true，单纯好评设为false。 """, ), ("human", """ 请分析下面的商品评论： {review} """), ]) prompt = template.invoke({ "review": "鼠标手感不错，也很安静，但是用了两周滚轮就有异响。" }) structured_model = model.with_structured_output( ReviewAnalysis, method="json_mode", ) response = structured_model.invoke(prompt) print(response) print(response.sentiment) print(response.keywords) print(response.summary) print(response.needs_reply)

📌 小贴士：Literal["正面", "负面", "中性"]是"字面量类型约束"，意思是这个字段只能取这三个值之一，类似枚举，模型瞎写别的字符串会被拦下。

预期结果：

sentiment='负面' keywords=['手感', '静音', '滚轮异响'] summary='用户认可鼠标手感和静音效果，但反馈滚轮出现质量问题。' needs_reply=True

💡 你可能会问："method='json_mode'我代码里没看到任何 JSON 啊？"——json_mode不是让你手动处理 JSON 字符串，底层流程是：① 模型输出一段合法 JSON 文本；② LangChain 内部自动把它解析成你的ReviewAnalysis对象；③ 你拿到手直接就是对象，所以直观上看不到原始 JSON。

十一、两种结构化方式怎么选

方式	特点	适合场景
`PydanticOutputParser`	通过 Prompt 要求格式，再解析文本	想弄清 Parser 工作原理时
`with_structured_output`	调用更简洁	模型服务支持结构化输出时

简单建议：

先掌握PydanticOutputParser（理解格式要求怎么传、解析为什么失败）
实际项目优先考虑with_structured_output
使用前先确认模型服务是否支持对应方式

一句话区分：PydanticOutputParser是"让模型按提示输出 JSON，我再本地解析"；with_structured_output是"把结构化能力直接绑在模型调用上，让模型/API 尽量按 schema 生成"。

十二、案例四：工单分类

智能体客服接到用户反馈后，自动把问题分到不同类别、并标优先级。创建04_ticket_classifier.py：

from typing import Literal from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field from utils.model_factory import get_deepSeek_model class TicketResult(BaseModel): category: Literal["订单", "物流", "退款", "产品", "其他"] = Field( description="工单分类" ) priority: Literal["低", "中", "高"] = Field(description="工单优先级") reason: str = Field(description="分类原因") model = get_deepSeek_model() parser = PydanticOutputParser(pydantic_object=TicketResult) format_instructions = parser.get_format_instructions() template = ChatPromptTemplate.from_messages([ ( "system", """你是一名客服工单分类助手。 请根据用户问题完成分类。 {format_instructions}""", ), ("human", "用户问题：{question}"), ]) prompt = template.invoke({ "format_instructions": format_instructions, "question": "订单显示已签收，但我没有收到商品，请尽快处理。", }) response = model.invoke(prompt) result = parser.invoke(response) print(result.category) print(result.priority) print(result.reason) if result.priority == "高": print("转人工客服优先处理")

这个结果能继续进业务流程：

if result.priority == "高": print("转人工客服优先处理")

这就是结构化输出的实际价值：模型结果能直接驱动后面的程序逻辑。

十三、处理解析错误

模型输出不稳定时，可能解析失败。可以捕获OutputParserException：

from langchain_core.exceptions import OutputParserException try: result = parser.parse(response.content) print(result) except OutputParserException as exc: print("结构化输出解析失败") print("模型原始输出：", response.content) print("错误信息：", exc)

处理建议：

保存模型原始输出（方便排查）
记录错误日志
优化 Prompt
必要时重新请求模型
重要业务不能完全依赖模型自行保证格式

下面是带异常处理的完整版（在第十二节基础上包一层try/except）：

from typing import Literal from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field from utils.model_factory import get_deepSeek_model class TicketResult(BaseModel): category: Literal["订单", "物流", "退款", "产品", "其他"] = Field( description="工单分类" ) priority: Literal["低", "中", "高"] = Field(description="工单优先级") reason: str = Field(description="分类原因") model = get_deepSeek_model() parser = PydanticOutputParser(pydantic_object=TicketResult) format_instructions = parser.get_format_instructions() template = ChatPromptTemplate.from_messages([ ( "system", """你是一名客服工单分类助手。 请根据用户问题完成分类。 {format_instructions}""", ), ("human", "用户问题：{question}"), ]) prompt = template.invoke({ "format_instructions": format_instructions, "question": "订单显示已签收，但我没有收到商品，请尽快处理。", }) response = model.invoke(prompt) try: result = parser.invoke(response) print(result.category) print(result.priority) print(result.reason) if result.priority == "高": print("转人工客服优先处理") except Exception as exception: print("失败的原因是：", exception) print("大模型响应的内容是：", response.content)