背景 06-结构化输出
第06章:结构化输出(Structured Output)
1、结构化输出概述 1.1 什么是结构化输出 LangChain 的结构化输出(Structured Output)指的是:
要求模型最终返回一个符合预定义结构的数据对象,例如固定字段的 JSON、Pydantic 模型、TypedDict,而不再是无格式的自然语言文本。
它的核心目标是把”自然语言回答 “变成”程序可以稳定消费的数据 “。
例如,不是让模型输出:
1 盗梦空间在2010年上映,导演是克里斯托弗·诺兰,评分9.3。
而是让它输出成类似这样的结构:
1 2 3 4 5 6 { "title" : "盗梦空间" , "year" : 2010 , "director" : "克里斯托弗·诺兰" , "rating" : 9.3 }
这样做的价值主要有三点:
更容易被代码处理 :下游系统可以直接读字段,而不是再从自然语言里做解析。
结果更稳定 :减少”模型说法变了但意思差不多”导致的解析失败。
更适合工程化 :适用于表单抽取、分类、路由、调用工具参数生成、工作流状态传递等场景。
1.2 传统方式 vs 结构化输出 1、传统的几种方式(繁琐、不推荐)
1 2 3 prompt = "以JSON格式返回:{name, age, occupation}" response = model.invoke(prompt)
1 2 3 import jsondata = json.loads(response.content)
1 2 3 if not isinstance(data['age' ], int): raise ValueError("age must be int" )
1 2 person = Person(**data)
2、结构化输出(简洁)
1 2 3 4 structured_llm = model.with_structured_output(Person) person = structured_llm.invoke("张三是一名 30 岁的软件工程师" )
为什么第 2 种结构化输出机制这么受欢迎?
在没有 Pydantic 等结构化方案之前,开发者需要写大量的 Prompt 苦口婆心地求大模型”请返回 JSON,不要带任何解释”,然后自己写繁琐的 json.loads() 和 try...except。
而有了 Pydantic 等结构化方案结合 .with_structured_output() 之后:
Prompt 变干净了 :字段的 description 直接充当了 Prompt 的一部分。
类型安全 :编辑器能自动补全,代码运行前就能做类型检查。
极其稳定 :依托大模型厂商底层的 JSON 模式,输出错误率降到了极低。
1.3 结构化输出模式 目前 LangChain 1.x 支持多种 Schema 与结构化输出方式:
Pydantic (字段校验、描述、嵌套结构,功能最丰富)
TypedDict (轻量类型约束)
JSON Schema (与前后端/跨语言接口最通用)
dataclass
模型对象可以调用 with_structured_output() 绑定输出模式(schema)。
只有 Pydantic 返回的是 Schema 类实例,其余三种方式返回的都是字典 ;也只有 Pydantic 在类型不匹配时会抛出异常。
问题:现在所有模型都支持”本章要讲解的结构化输出方式”吗?
大部分现代模型支持(通过函数调用):
✅ OpenAI (gpt-4, gpt-3.5-turbo)
✅ Anthropic (claude-3)
✅ Groq (llama-3)
❌ 某些旧模型不支持
如果不支持,LangChain 会回退到提示词 + JSON 解析。
2、四种模式的使用 2.1 模式 1:Pydantic 它通过在运行时强制执行类型提示,确保数据的正确性和一致性,是生产场景首选 。
2.1.1 基本使用 需要满足的几个要素:
所有结构化输出的数据模型都必须继承 BaseModel
使用类型提示 。Pydantic 支持丰富的字段类型:str、int、float、List[xxx]、Optional[xxx] 等
使用 Field() 添加字段默认值和描述,帮助 LLM 理解字段含义
举例 1:
1)大模型的初始化
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from langchain.chat_models import init_chat_modelfrom dotenv import load_dotenvimport osload_dotenv(override=True ) CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY" ) CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL" ) model = init_chat_model( model="gpt-5.4-mini" , model_provider="openai" , api_key=CLOSEAI_API_KEY, base_url=CLOSEAI_BASE_URL )
2)定义 Pydantic 模型
使用清晰的字段描述;没有描述,LLM 可能格式错误。
1 2 3 4 5 6 7 from pydantic import BaseModel, Fieldclass Person (BaseModel) : """人物信息""" name: str = Field(description="姓名" ) age: int = Field(description="年龄" ) occupation: str = Field(description="职业" )
LangChain 会要求 LLM 的输出必须能填充这些字段。
3)使用 with_structured_output 即可引导模型进行结构化输出:
1 2 3 4 5 6 7 8 9 10 11 12 13 structured_llm = model.with_structured_output(Person) result = structured_llm.invoke("张三是一名 30 岁的软件工程师" ) print(result) print(type(result)) print(result.name) print(result.age) print(result.occupation)
输出:
1 2 3 4 5 name='张三' age=30 occupation='软件工程师' <class '__main__.Person'> 张三 30 软件工程师
说明:没有描述,LLM 可能格式错误。
举例 2:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from pydantic import BaseModel, Field, SecretStrclass MovieModel (BaseModel) : """ 电影的详细信息 """ title: str = Field(description="电影标题" ) year: int = Field(description="电影上映年份" ) director: str = Field(description="导演" ) rating: float = Field(description="电影评分,满分十分" ) model_with_structure = model.with_structured_output(MovieModel) response = model_with_structure.invoke("给出盗梦空间的信息" ) print(response) print(type(response))
输出:
1 2 title='盗梦空间' year=2010 director='克里斯托弗·诺兰' rating=9.3 <class '__main__.MovieModel'>
举例 3:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 from pydantic import BaseModel, Fieldclass SentimentAnalysis (BaseModel) : """情感分析结果""" sentiment: str = Field(description="情感倾向:positive/negative/neutral" ) confidence: float = Field(description="置信度,0-1之间" ) keywords: list[str] = Field(description="关键词列表" ) structured_model = model.with_structured_output(SentimentAnalysis) text = "这个课程内容很实用,学到了很多知识,强烈推荐!" result = structured_model.invoke( f"分析以下文本的情感:\n{text} " ) print(f"类型: {type(result)} " ) print(f"情感: {result.sentiment} " ) print(f"置信度: {result.confidence} " ) print(f"关键词: {result.keywords} " )
输出:
1 2 3 4 类型: <class '__main__.SentimentAnalysis'> 情感: positive 置信度: 0.99 关键词: ['实用', '学到了很多知识', '强烈推荐']
2.1.2 高级特性 使用 CloseAI 平台的 gpt 模型:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from langchain.chat_models import init_chat_modelfrom dotenv import load_dotenvimport osload_dotenv(override=True ) CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY" ) CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL" ) model_with_closeai = init_chat_model( model="gpt-5.4-mini" , model_provider="openai" , api_key=CLOSEAI_API_KEY, base_url=CLOSEAI_BASE_URL )
使用 OpenRouter 平台的 gpt 模型:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 from langchain_openrouter import ChatOpenRouterfrom dotenv import load_dotenvimport osload_dotenv(override=True ) OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY" ) OPENROUTER_API_BASE = os.getenv("OPENROUTER_API_BASE" ) model_with_openrouter = ChatOpenRouter( model="openai/gpt-5.4-mini" , api_key=OPENROUTER_API_KEY, base_url=OPENROUTER_API_BASE, )
情况 1:可选字段
问题:LLM 未填充某些字段怎么办?
回答:使用 Optional 指定字段为可选的。
举例:
1 2 3 4 5 6 7 8 9 10 11 12 from pydantic import BaseModel, Fieldfrom typing import Optionalclass Person (BaseModel) : """人物信息""" name: str = Field(description="姓名" ) age: int = Field(description="年龄" ) occupation: str = Field(description="职业" ) structured_llm = model_with_closeai.with_structured_output(Person) structured_llm.invoke("张三是一名医生" )
输出:
1 Person(name='张三', age=0, occupation='医生')
作为对比:
1 2 3 4 5 6 7 8 9 10 11 12 from typing import Optionalfrom pydantic import BaseModel, Fieldclass Person (BaseModel) : """人物信息""" name: str = Field(description="姓名" ) age: Optional[int] = Field(description="年龄" ) occupation: str = Field(description="职业" ) structured_llm = model_with_closeai.with_structured_output(Person) structured_llm.invoke("张三是一名医生" )
输出:
1 Person(name='张三', age=None, occupation='医生')
情况 2:默认值
LLM 未提供的信息会使用默认值。格式如下:
1 Field(default="默认值" , description="描述" )
注意:不同模型提供商对 default 字段的支持是不同的。
举例 1:
使用 CloseAI 平台的 gpt 模型:
1 2 3 4 5 6 7 8 9 10 11 12 from typing import Optionalfrom pydantic import BaseModel, Fieldclass Person (BaseModel) : """人物信息""" name: str = Field(description="姓名" ) age: int = Field(1 , description="年龄" ) occupation: str = Field(description="职业" ) structured_llm = model_with_closeai.with_structured_output(Person) structured_llm.invoke("张三是一名医生" )
输出:
1 Person(name='张三', age=0, occupation='医生')
作为对比:
1 2 3 4 5 6 7 8 9 10 11 12 from typing import Optionalfrom pydantic import BaseModel, Fieldclass Person (BaseModel) : """人物信息""" name: str = Field(description="姓名" ) age: int = Field(1 , description="年龄" ) occupation: str = Field(description="职业" ) structured_llm = model_with_openrouter.with_structured_output(Person) structured_llm.invoke("张三是一名医生" )
输出:
1 Person(name='张三', age=1, occupation='医生')
举例 2:
1 2 3 4 5 6 7 8 class Config (BaseModel) : timeout: Optional[int] = Field(30 , description="超时时间(单位秒)" ) retry: bool = Field(False , description="是否支持重试" ) max_attempts: int = Field(6 , description="最大重试次数" ) structured_llm = model_with_closeai.with_structured_output(Config) structured_llm.invoke("配置要求:支持重试,最多重试5次" )
输出:
1 Config(timeout=None, retry=True, max_attempts=5)
作为对比:
1 2 3 4 5 6 7 8 class Config (BaseModel) : timeout: Optional[int] = Field(30 , description="超时时间(单位秒)" ) retry: bool = Field(False , description="是否支持重试" ) max_attempts: int = Field(6 , description="最大重试次数" ) structured_llm = model_with_openrouter.with_structured_output(Config) structured_llm.invoke("配置要求:支持重试,最多重试5次" )
输出:
1 Config(timeout=None, retry=True, max_attempts=5)
举例 3:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from typing import Optionalfrom pydantic import BaseModel, Fieldclass Product (BaseModel) : """产品信息""" name: str = Field(description="产品名称" ) price: float = Field(description="价格" ) description: Optional[str] = Field(description="产品描述" ) stock: int = Field(default=100 , description="库存" ) structured_llm = model_with_openrouter.with_structured_output(Product) print("\n场景1:完整信息" ) result1 = structured_llm.invoke("iPhone 15 售价 5999 元,最新款智能手机,库存 50 台" ) print(result1) print("\n场景2:缺少描述和库存" ) result2 = structured_llm.invoke("MacBook Pro 售价 12999 元" ) print(result2)
输出:
1 2 3 4 5 场景1:完整信息 name='iPhone 15' price=5999.0 description='最新款智能手机' stock=50 场景2:缺少描述和库存 name='MacBook Pro' price=12999.0 description=None stock=100
情况 3:枚举类型
问题:如何限制字段的可选值?
回答:使用枚举。
举例 1:
1 2 3 4 5 6 7 8 9 10 from enum import Enumclass Priority (str, Enum) : LOW = "低" MEDIUM = "中" HIGH = "高" class Task (BaseModel) : title: str priority: Priority
举例 2:
1 2 3 4 5 6 7 8 from enum import Enumclass Status (str, Enum) : ACTIVE = "激活" INACTIVE = "未激活" class User (BaseModel) : status: Status
举例 3:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 from enum import Enumfrom typing import Optionalfrom pydantic import BaseModel, Fieldclass Priority (str, Enum) : LOW = "低" MEDIUM = "中" HIGH = "高" class CustomerInfo (BaseModel) : """客户信息""" name: str = Field(description="客户姓名" ) phone: str = Field(description="电话号码" ) email: Optional[str] = Field(description="邮箱" ) issue: str = Field(description="问题描述" ) urgency: Priority = Field(description="紧急程度" ) structured_llm = model_with_openrouter.with_structured_output(CustomerInfo) conversation = """ 客服: 您好,请问有什么可以帮助您? 客户: 我是王小明,电话 138-1234-5678,我的订单一直没发货,很着急! 客服: 好的,我帮您查一下 """ result = structured_llm.invoke(f"从以下客服对话中提取客户信息:\n{conversation} " ) print(result) print("\n提取结果:" ) print(f" 客户: {result.name} " ) print(f" 电话: {result.phone} " ) print(f" 邮箱: {result.email or '未提供' } " ) print(f" 问题: {result.issue} " ) print(f" 紧急程度: {result.urgency.value} " )
输出:
1 2 3 4 5 6 7 8 name='王小明' phone='138-1234-5678' email=None issue='订单一直没发货,很着急' urgency=<Priority.HIGH: '高'> 提取结果: 客户: 王小明 电话: 138-1234-5678 邮箱: 未提供 问题: 订单一直没发货,很着急 紧急程度: 高
如果嫌单独定义一个 Enum 类太麻烦,也可以直接导入 typing 中的 Literal,直接在字段里把允许的值写死。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 from typing import Optional, Literalfrom pydantic import BaseModel, Fieldclass CustomerInfo (BaseModel) : """客户信息""" name: str = Field(description="客户姓名" ) phone: str = Field(description="电话号码" ) email: Optional[str] = Field("未提供" , description="邮箱" ) issue: str = Field(description="问题描述" ) urgency: Literal["低" , "中" , "高" ] = Field(description="紧急程度" ) structured_llm = model_with_openrouter.with_structured_output(CustomerInfo) conversation = """ 客服: 您好,请问有什么可以帮助您? 客户: 我是王小明,电话 138-1234-5678,我的订单一直没发货,很着急! 客服: 好的,我帮您查一下 """ result = structured_llm.invoke(f"从以下客服对话中提取客户信息:\n{conversation} " ) print(result) print("\n提取结果:" ) print(f" 客户: {result.name} " ) print(f" 电话: {result.phone} " ) print(f" 邮箱: {result.email} " ) print(f" 问题: {result.issue} " ) print(f" 紧急程度: {result.urgency} " )
输出:
1 2 3 4 5 6 7 8 name='王小明' phone='138-1234-5678' email='未提供' issue='订单一直没发货,很着急' urgency='高' 提取结果: 客户: 王小明 电话: 138-1234-5678 邮箱: 未提供 问题: 订单一直没发货,很着急 紧急程度: 高
应用场景 :
情况 4:列表提取
举例 1:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing import Listclass Person (BaseModel) : """人物信息""" name: str age: int class PersonList (BaseModel) : """人物列表信息""" people: List[Person] structured_llm = model.with_structured_output(PersonList) result = structured_llm.invoke("张三 30岁,李四 25岁" ) print(result)
输出:
1 people=[Person(name='张三', age=30), Person(name='李四', age=25)]
举例 2:产品评论分析
1 2 3 4 5 6 7 8 9 10 11 12 13 14 class Review (BaseModel) : """产品评论""" product: str rating: int = Field(description="评分 1-5" ) pros: List[str] = Field(description="优点列表" ) cons: List[str] = Field(description="缺点列表" ) structured_llm = model.with_structured_output(Review) review = structured_llm.invoke(""" iPhone 17 很棒!摄像头强大,手感好。但是价格贵,没有充电器。4分。 """ )print(review)
输出:
1 product='iPhone 17' rating=4 pros=['摄像头强大', '手感好'] cons=['价格贵', '没有充电器']
应用场景 :
批量处理用户评论
自动生成分析报告
发现产品改进点
举例 3:文档信息提取
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 class Invoice (BaseModel) : """发票信息""" invoice_number: str = Field(description="发票号" ) date: str = Field(description="日期" ) total_amount: float = Field(description="总金额" ) items: List[str] = Field(description="商品" ) structured_llm = model.with_structured_output(Invoice) invoice_text = """ 发票号: INV-2024-001 日期: 2024-01-15 总金额: 1299.00 商品: MacBook Pro, AppleCare+ """ invoice = structured_llm.invoke(f"提取发票信息:{invoice_text} " ) print(invoice)
输出:
1 invoice_number='INV-2024-001' date='2024-01-15' total_amount=1299.0 items=['MacBook Pro', 'AppleCare+']
应用场景 :
情况 5:嵌套结构
举例 1:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from pydantic import BaseModelclass Address (BaseModel) : """地点描述""" city: str district: str class Company (BaseModel) : """公司信息""" name: str address: Address structured_llm = model.with_structured_output(Company) result = structured_llm.invoke("阿里巴巴在杭州滨江区" ) print(result)
输出:
1 name='阿里巴巴' address=Address(city='杭州', district='滨江区')
举例 2:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 from pydantic import BaseModel, Fieldfrom typing import Listclass Actor (BaseModel) : """演员信息""" name: str = Field(description="演员姓名" ) role: str = Field(description="饰演的角色" ) class Movie (BaseModel) : """电影信息""" title: str = Field(description="电影标题" ) year: int = Field(description="上映年份" ) director: str = Field(description="导演" ) cast: List[Actor] = Field(description="演员列表" ) rating: float = Field(description="评分" ) structured_model = model.with_structured_output(Movie) response = structured_model.invoke("请介绍电影《盗梦空间》" ) print(f"电影名: {response.title} " ) print(f"上映年份: {response.year} " ) print(f"导演: {response.director} " ) print(f"演员列表: {response.cast} " ) print(f"评分: {response.rating} " )
以上代码输出结果如下:
1 2 3 4 5 电影名: 盗梦空间 上映年份: 2010 导演: 克里斯托弗·诺兰 演员列表: [Actor(name='莱昂纳多·迪卡普里奥', role='柯布'), Actor(name='约瑟夫·高登-莱维特', role='亚瑟'), Actor(name='艾伦·佩吉', role='阿里阿德涅'), Actor(name='汤姆·哈迪', role='艾姆斯'), Actor(name='渡边谦', role='斋藤'), Actor(name='玛丽昂·歌迪亚', role='梅尔')] 评分: 8.8
说明:LLM 能力有限,复杂嵌套结构可能会出错。所以建议:
1 2 3 4 5 class Bad (BaseModel) : user: User company: Company address: Address country: Country
使用清晰的 description
必要时拆分成多个调用
举例 3:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 from pydantic import BaseModelfrom typing import Listclass Aspect (BaseModel) : """评论维度""" name: str = Field(description="维度名称,如:质量、价格、服务" ) score: int = Field(description="评分,1-5" ) comment: str = Field(description="具体评价" ) class ProductReview (BaseModel) : """产品评论分析""" overall_sentiment: str = Field(description="整体情感:positive/negative/neutral" ) overall_score: int = Field(description="综合评分,1-5" ) aspects: List[Aspect] = Field(description="各维度评价" ) summary: str = Field(description="一句话总结" ) structured_model = model.with_structured_output(ProductReview) review_text = """ 这款笔记本电脑性能非常强大,运行大型软件毫无压力。 屏幕色彩鲜艳,看视频很舒服。 不过价格有点贵,而且风扇噪音较大。 客服态度很好,物流也快。 总体来说还是值得购买的。 """ result = structured_model.invoke( f"分析以下产品评论:\n{review_text} " ) print(f"整体情感: {result.overall_sentiment} " ) print(f"综合评分: {result.overall_score} /5" ) print(f"\n各维度评价:" ) for aspect in result.aspects: print(f" - {aspect.name} : {aspect.score} /5 - {aspect.comment} " ) print(f"\n总结: {result.summary} " )
输出:
1 2 3 4 5 6 7 8 9 10 11 整体情感: positive 综合评分: 4/5 各维度评价: - 性能: 5/5 - 性能非常强大,运行大型软件毫无压力。 - 屏幕: 5/5 - 屏幕色彩鲜艳,看视频很舒服。 - 价格: 2/5 - 价格有点贵,性价比略受影响。 - 噪音: 2/5 - 风扇噪音较大,影响使用体验。 - 服务: 5/5 - 客服态度很好,物流也快。 总结: 整体表现优秀,性能和屏幕突出,服务也好,但价格偏高且风扇噪音较大。
情况 6:限制条件
举例 1:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 from pydantic import ValidationErrorclass User (BaseModel) : name: str = Field(min_length=2 , max_length=20 ) age: int = Field(ge=0 , le=150 ) email: str print("\n有效数据:" ) try : user = User(name="张三" , age=30 , email="zhang@example.com" ) print(f"[OK] {user.name} , {user.age} , {user.email} " ) except ValidationError as e: print(f"[FAIL] {e} " ) print("\n无效数据(年龄超出范围):" ) try : user = User(name="李四" , age=200 , email="li@example.com" ) print(f"[OK] {user} " ) except ValidationError as e: print(f"[FAIL] 验证失败(符合预期): {e.errors()[0 ]['msg' ]} " )
输出:
1 2 3 4 5 有效数据: [OK] 张三, 30, zhang@example.com 无效数据(年龄超出范围): [FAIL] 验证失败(符合预期): Input should be less than or equal to 150
举例 2:
使用 CloseAI 平台模型:
1 2 3 4 5 6 7 8 9 10 11 12 class Product (BaseModel) : """产品信息(严格验证)""" name: str = Field(description="产品名称(字符串类型)" , min_length=2 ) price: float = Field(description="价格,数字类型" , gt=0 ) stock: int = Field(description="库存,整数类型" , ge=0 ) structured_llm = model_with_closeai.with_structured_output(Product) response = structured_llm.invoke("华为mate 80 promax 价格是-7999,当前库存-100" ) print(response)
输出:
1 name='华为mate 80 promax' price=7999.0 stock=100
使用 OpenRouter 平台模型:
1 2 3 4 5 6 7 8 9 10 11 12 class Product (BaseModel) : """产品信息(严格验证)""" name: str = Field(description="产品名称(字符串类型)" , min_length=2 ) price: float = Field(description="价格,数字类型" , gt=0 ) stock: int = Field(description="库存,整数类型" , ge=0 ) structured_llm = model_with_openrouter.with_structured_output(Product) response = structured_llm.invoke("华为mate 80 promax 价格是-7999,当前库存-100" ) print(response)
输出:
1 name='华为mate 80 promax' price=1.0 stock=0
2.1.3 工作流程图解 第 1 步:定义结构
比如:
1 2 3 4 5 6 from pydantic import BaseModel, Fieldclass BookInfo (BaseModel) : title: str = Field(description="书名" ) author: str = Field(description="作者名字" ) tags: list[str] = Field(description="书籍的标签或分类" )
第 2 步:协议转换
LangChain 内部会调用 Pydantic 的底层方法(如 model_json_schema()),将你写的 Python 代码自动转换成标准的 JSON Schema。
这个 JSON Schema 是一段严格的 JSON 文本,详细描述了有哪些字段、字段类型是什么(string、array 等)以及字段的描述(description)。
第 3 步:模型交互与强约束
LangChain 会将这个 JSON Schema 包装进给大模型的 API 请求中。
现代方法(.with_structured_output) :现代大模型(如 OpenAI、Anthropic、Gemini 等)普遍支持”函数调用/工具调用(Function/Tool Calling)”或”JSON Mode”。LangChain 会把 JSON Schema 作为 Tools 传入。
大模型侧的约束 :像 OpenAI 的 strict=True 参数,会启动模型的语法采样约束(Grammar-based sampling)。大模型在解码生成 token 时,不是瞎猜,而是严格按照 JSON Schema 的语法树进行选择,从而在模型底层级保证了输出格式绝不走样。
第 4 步:自动解析与验证
当大模型返回符合 JSON 规范的字符串后,LangChain 的 PydanticStructuredOutputParser(解析器)会接管工作:
解析(Parsing) :将字符串解析为 Python 字典。
验证(Validation) :将字典喂给你的 Pydantic 模型。Pydantic 会自动检查数据类型是否正确。如果模型漏掉了必填字段,或者类型错误,这里会直接抛出验证错误(或者触发 LangChain 的重试机制)。
返回(Return) :如果通过验证,你拿到的不再是冷冰冰的字符串,而是一个直接可以点出属性的 Python Pydantic 对象(例如 result.title)。
2.2 模式 2:TypedDict 2.2.1 什么是 TypedDict TypedDict 是 Python 3.8+ 引入的一种类型提示工具,即带有类型声明的字典结构。适合需要快速定义字典结构且无需 Pydantic 重量级功能的场景。
1、普通 dict 没有类型信息:
1 2 3 4 5 6 { "title" : "盗梦空间" , "year" : 2010 , "director" : "克里斯托弗·诺兰" , "rating" : 9.3 }
2、TypedDict 可以进一步说明:
TypedDict 主要是类型声明 ,不是运行时强校验器。
举例:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from typing_extensions import TypedDictclass MovieDict (TypedDict) : title: str year: int director: str rating: float movie: MovieDict = { "title1" : "盗梦空间" , "year" : 2010 , "director" : "克里斯托弗·诺兰" , "rating" : 8.8 , } print(movie)
实例化字典时给出的字段名称和 TypedDict 不完全一致,此时 IDE 的静态类型检查会标记。
但不会导致运行时异常,输出如下所示:
1 2 3 4 5 6 { 'title1': '盗梦空间', 'year': 2010, 'director': '克里斯托弗·诺兰', 'rating': 8.8 }
2.2.2 基本使用 Annotated 的使用
Annotated 用来在”类型”之外,再附加一些额外信息,即元数据。类似于 Pydantic 的 Field。
基本形式:
1 Annotated[类型, 附加信息1 , 附加信息2 , ...]
举例的前提 :
使用 CloseAI 平台的 gpt 模型:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from langchain.chat_models import init_chat_modelfrom dotenv import load_dotenvimport osload_dotenv(override=True ) CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY" ) CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL" ) model_with_closeai = init_chat_model( model="gpt-5.4-mini" , model_provider="openai" , api_key=CLOSEAI_API_KEY, base_url=CLOSEAI_BASE_URL )
使用 OpenRouter 平台的 gpt 模型:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 from langchain_openrouter import ChatOpenRouterfrom dotenv import load_dotenvimport osload_dotenv(override=True ) OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY" ) OPENROUTER_API_BASE = os.getenv("OPENROUTER_API_BASE" ) model_with_openrouter = ChatOpenRouter( model="openai/gpt-5.4-mini" , api_key=OPENROUTER_API_KEY, base_url=OPENROUTER_API_BASE, )
举例 1:返回简单结构
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 """ 使用 TypedDict 模型定义结构化输出 """ from typing_extensions import TypedDict, Annotatedclass MovieTypedDict (TypedDict) : """ 电影的详细信息 """ title: Annotated[str, "电影的正式名称,例如《盗梦空间》" ] year: Annotated[int, "电影的公映年份,使用四位数字表示" ] director: Annotated[str, "电影导演的全名" ] rating: Annotated[float, "电影在10分制下的评分,可包含一位小数" ] structured_llm = model_with_closeai.with_structured_output(MovieTypedDict) response = structured_llm.invoke("给我介绍下电影《星际穿越》" ) print(type(response)) print(response)
输出如下:
1 2 <class 'dict'> {'title': '星际穿越', 'year': 2014, 'director': '克里斯托弗·诺兰', 'rating': 8.6}
举例 2:返回嵌套结构
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 from typing import TypedDict, List, Annotatedclass Actor (TypedDict) : """演员情况""" name: Annotated[str, "演员姓名" ] role: Annotated[str, "饰演的角色" ] class Movie (TypedDict) : """电影情况""" title: Annotated[str, "电影标题" ] year: Annotated[int, "上映年份" ] director: Annotated[str, "导演" ] cast: Annotated[List[Actor], "演员列表" ] rating: Annotated[float, "评分" ] structured_llm = model_with_closeai.with_structured_output(Movie) resp = structured_llm.invoke("给我介绍下电影《盗梦空间》" ) print(f"电影名: {resp['title' ]} " ) print(f"上映年份: {resp['year' ]} " ) print(f"导演: {resp['director' ]} " ) print(f"演员列表:{resp['cast' ]} " ) print(f"评分: {resp['rating' ]} " )
输出如下:
1 2 3 4 5 电影名: 盗梦空间 上映年份: 2010 导演: 克里斯托弗·诺兰 演员列表:[{'name': '莱昂纳多·迪卡普里奥', 'role': '多姆·柯布'}, {'name': '约瑟夫·高登-莱维特', 'role': '亚瑟'}, {'name': '艾伦·佩吉', 'role': '阿里阿德涅'}, {'name': '汤姆·哈迪', 'role': '伊姆斯'}, {'name': '渡边谦', 'role': '斋藤'}] 评分: 8.8
举例 3:... 的使用
说明:与模型提供商有关系。
使用 CloseAI 平台 gpt 模型:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing_extensions import TypedDict, Annotatedclass MovieDict (TypedDict) : """ 电影的详细信息 """ title: Annotated[str, ..., "电影标题" ] year: Annotated[int, ..., "电影上映年份" ] director: Annotated[str, ..., "导演" ] rating: Annotated[float, ..., "电影评分,满分十分" ] model_with_structure = model_with_closeai.with_structured_output(MovieDict) response = model_with_structure.invoke("根据这段话抽取盗梦空间的信息,不包含的信息可以留空:盗梦空间在2010年上映,导演是克里斯托弗·诺兰。" ) print(response) print(type(response))
输出:
1 2 {'title': '盗梦空间', 'year': 2010, 'director': '克里斯托弗·诺兰'} <class 'dict'>
使用 OpenRouter 平台 gpt 模型:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing_extensions import TypedDict, Annotatedclass MovieDict (TypedDict) : """ 电影的详细信息 """ title: Annotated[str, ..., "电影标题" ] year: Annotated[int, ..., "电影上映年份" ] director: Annotated[str, ..., "导演" ] rating: Annotated[float, ..., "电影评分,满分十分" ] model_with_structure = model_with_openrouter.with_structured_output(MovieDict) response = model_with_structure.invoke("根据这段话抽取盗梦空间的信息,不包含的信息可以留空:盗梦空间在2010年上映,导演是克里斯托弗·诺兰。" ) print(response) print(type(response))
输出:
1 2 {'title': '盗梦空间', 'year': 2010, 'director': '克里斯托弗·诺兰', 'rating': 0} <class 'dict'>
说明:上述代码的 ... 是 Python 的字面量,等价于 Ellipsis,可以理解为占位符。下游框架(如 LangChain)可以对 ... 作定制化处理,如 LangChain 中 Annotated 的 ... 表示当前字段是必须存在的,不可省略,用来指示模型的输出。
再比如:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from typing_extensions import TypedDict, Annotatedclass MovieDict (TypedDict) : """ 电影的详细信息 """ title: Annotated[str, ..., "电影标题" ] year: Annotated[int, ..., "电影上映年份" ] director: Annotated[str, Ellipsis , "导演" ] rating: Annotated[float, "电影评分,满分十分" ] model_with_structure = model.with_structured_output(MovieDict) response = model_with_structure.invoke("根据这段话抽取盗梦空间的信息,不包含的信息必须留空:盗梦空间上映于2010年,我们并不知道它的导演是谁。" ) print(response) print(type(response))
输出:
1 2 {'title': '盗梦空间', 'year': 2010, 'director': ''} <class 'dict'>
director 和 rating 的信息都是缺失的,但是前者被标记为必填,因此输出的字典包含该字段但是值为空字符串,而 rating 字段被省略了。
2.3 模式 3:JSON Schema 这种方式需要按照 JSON Schema 规范拼接 JSON 字符串,比较繁琐,并且缺少校验机制。不推荐。
举例 1:返回简单结构
模型初始化:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from langchain.chat_models import init_chat_modelfrom dotenv import load_dotenvimport osload_dotenv(override=True ) CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY" ) CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL" ) model = init_chat_model( model="gpt-5.4-mini" , model_provider="openai" , api_key=CLOSEAI_API_KEY, base_url=CLOSEAI_BASE_URL )
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 json_schema = { "title" : "Movie" , "description" : "A movie with details" , "type" : "object" , "properties" : { "title" : { "type" : "string" , "description" : "The title of the movie" }, "year" : { "type" : "integer" , "description" : "The year the movie was released" }, "director" : { "type" : "string" , "description" : "The director of the movie" }, "rating" : { "type" : "number" , "description" : "The movie's rating out of 10" } }, "required" : ["title" , "year" , "director" , "rating" ] } structured_model = model.with_structured_output( json_schema, method="json_schema" ) response = structured_model.invoke("给出盗梦空间的信息" ) print(response) print(type(response))
输出如下:
1 2 {'title': '盗梦空间', 'year': 2010, 'director': '克里斯托弗·诺兰', 'rating': 8.8} <class 'dict'>
说明:
method:结构化输出的方式,但是否可用,依赖于模型供应商及 LangChain 适配器的具体实现。比如,DeepSeek 模型服务不支持 json_schema 模式。
json_schema:使用模型供应商提供的专用结构化输出功能。
以上代码中定义 json_schema 的时候指定的 title、description、type、properties、required 是遵循 JSON Schema 规范的标准关键字,是固定写法。几个关键字的解释如下:
title:为整个 Schema 或特定属性提供一个人类可读的标题,不能是中文,用于提高可读性。
description:提供更详细的文字描述,说明 Schema 或属性的用途等,和 title 一样,旨在帮助理解。
type:定义当前数据节点必须是什么数据类型。常见类型有 string, number, integer, boolean, object, array, null。object 即是 JSON 对象。
properties:用于定义 JSON 对象(Object)中可以包含哪些属性(键),以及每个属性对应的值类型和说明。
required:当 type 为 "object" 时使用,是一个数组,列出了对象中必须存在的属性名。
举例 2:返回嵌套结构
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 """ 使用 JSON Schema 定义嵌套结构 """ project_schema = { "title" : "MovieInfo" , "description" : "包含电影标题、上映年份、导演、演员和评分的电影对象" , "type" : "object" , "properties" : { "title" : {"type" : "string" , "description" : "电影标题" }, "year" : {"type" : "integer" , "description" : "上映年份" }, "director" : {"type" : "string" , "description" : "导演" }, "cast" : { "type" : "array" , "description" : "演员列表" , "items" : { "type" : "object" , "properties" : { "name" : {"type" : "string" , "description" : "演员姓名" }, "role" : {"type" : "string" , "description" : "演员角色" } }, "required" : ["name" , "role" ] } }, "rating" : {"type" : "number" , "description" : "评分(10分制)" } }, "required" : ["title" , "year" , "director" , "cast" , "rating" ] } structured_model = model.with_structured_output(project_schema) response = structured_model.invoke("生成一个关于《星际穿越》的电影信息,包含导演、演员、评分" ) print(response)
输出如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 { 'title': '星际穿越', 'year': 2014, 'director': '克里斯托弗·诺兰', 'rating': 8.6, 'cast': [{ 'name': '马修·麦康纳', 'role': '库珀' }, { 'name': '安妮·海瑟薇', 'role': '艾米莉亚·布兰德' }, { 'name': '杰西卡·查斯坦', 'role': '墨菲·库珀(成年)' }, { 'name': '迈克尔·凯恩', 'role': '布兰德教授' }, { 'name': '马特·达蒙', 'role': '曼恩博士' }, { 'name': '麦肯吉·弗依', 'role': '墨菲·库珀(10岁)' }, { 'name': '蒂莫西·柴勒梅德', 'role': '汤姆·库珀(15岁)' }, { 'name': '卡西·阿弗莱克', 'role': '汤姆·库珀(成年)' }] }
2.4 模式 4:@dataclass @dataclass 是 Python 标准库 dataclasses 提供的类装饰器,用于简化”以字段为核心”的数据类定义。
给类加上 @dataclass 后,Python 会根据字段声明自动生成常用方法,例如:
因此,从对象行为上看,@dataclass 创建的类,常常近似于 手写这些方法的普通类。
1 2 3 4 5 6 7 8 from dataclasses import dataclass@dataclass class Movie : title: str year: int director: str rating: float
这类写法的主要价值是:让”数据结构定义”更简洁、更清晰。
注意 :
虽然 @dataclass 在行为上接近于手写 __init__、__repr__、__eq__ 的普通类,但二者并不完全等价。@dataclass 修饰后的类仍然是普通的 Python 类,但它会被标准库标记为数据类,并携带 dataclass 的字段元信息,手写 __init__ 等方法的普通类不能替代 @dataclass 修饰的类。
前者可以作为 LangChain 的 Schema,而后者不行。
举例 :
模型的初始化:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from langchain.chat_models import init_chat_modelfrom dotenv import load_dotenvimport osload_dotenv(override=True ) CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY" ) CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL" ) model = init_chat_model( model="gpt-5.4-mini" , model_provider="openai" , api_key=CLOSEAI_API_KEY, base_url=CLOSEAI_BASE_URL )
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from dataclasses import dataclassfrom pydantic import Field@dataclass class Movie () : """ 电影的详细信息 """ title: str = Field(description="电影标题" ) year: int = Field(description="电影上映年份" ) director: str = Field(description="导演" ) rating: float = Field(description="电影评分,满分十分" ) structured_model = model.with_structured_output(Movie) response = structured_model.invoke("给出盗梦空间的信息" ) print(response) print(type(response))
输出:
1 2 {'title': '盗梦空间', 'year': 2010, 'director': '克里斯托弗·诺兰', 'rating': 8.8} <class 'dict'>
@dataclass 作为 schema,返回的是未经校验的字典。
3、关于类型校验 验证不同 schema 定义方式在数据格式不匹配时的处理方式。
3.1 Fake Server 我们构造一个虚拟的 DeepSeek 服务端,客户端接收到的响应是人为构造 的,主要目的是为了观察不同模式下 LangChain 会从响应中的哪些字段中抽取信息。
服务器端如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 import jsonimport timefrom http.server import BaseHTTPRequestHandler, HTTPServerclass FakeDeepSeekHandler (BaseHTTPRequestHandler) : def do_POST (self) : content_length = int(self.headers.get("Content-Length" , 0 )) raw_body = self.rfile.read(content_length).decode("utf-8" ) print("\n" + "=" * 100 ) json_body = None try : json_body = json.loads(raw_body) print("[JSON BODY]" ) print(json.dumps(json_body, ensure_ascii=False , indent=2 )) except Exception as e: print("[JSON PARSE ERROR]" ) print(repr(e)) response = { "id" : "chatcmpl-test" , "object" : "chat.completion" , "created" : int(time.time()), "model" : "any" , "choices" : [ { "index" : 0 , "message" : { "role" : "assistant" , "content" : "" , "tool_calls" : [ { "id" : "call_1" , "type" : "function" , "function" : { "name" : json_body["tools" ][0 ]["function" ]["name" ], "arguments" : json.dumps( {'title1' : '盗梦空间' , 'year2' : 2010 , 'director' : '克里斯托弗·诺兰' , 'rating' : 9.3 }, ensure_ascii=False ) } } ] }, "finish_reason" : "stop" } ], "usage" : { "prompt_tokens" : 1 , "completion_tokens" : 1 , "total_tokens" : 2 } } print("\n" + "=" * 100 ) print("[RESPONSE]" ) print(json.dumps(response, ensure_ascii=False , indent=2 )) body = json.dumps(response, ensure_ascii=False ).encode("utf-8" ) self.send_response(200 ) self.send_header("Content-Type" , "application/json; charset=utf-8" ) self.send_header("Content-Length" , str(len(body))) self.end_headers() self.wfile.write(body) def log_message (self, format, *args) : pass def main () : server = HTTPServer(("127.0.0.1" , 8889 ), FakeDeepSeekHandler) print("Fake DeepSeek server running at http://127.0.0.1:8889" ) server.serve_forever() if __name__ == "__main__" : main()
不必关注代码实现,只需要知道我们返回的核心内容是:
1 2 3 4 5 6 { 'title1' : '盗梦空间' , 'year2' : 2010 , 'director' : '克里斯托弗·诺兰' , 'rating' : 9.3 }
LangChain 客户端希望模型返回的是:
1 2 3 4 5 6 { 'title' : 'xxx' , 'year' : xxxx, 'director' : 'xxx' , 'rating' : xxx }
这里的前两个字段不匹配。
启动服务端,然后执行四种定义方式的客户端代码。
3.2 四种模式的校验 3.2.1 Pydantic 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 from pydantic import BaseModel, Field, SecretStrfrom langchain_deepseek import ChatDeepSeekmodel = ChatDeepSeek( model="deepseek-v4-flash" , api_base="http://localhost:8889" , api_key=SecretStr("<KEY>" ) ) class MovieModel (BaseModel) : """ 电影的详细信息 """ title: str = Field(description="电影标题" ) year: int = Field(description="电影上映年份" ) director: str = Field(description="导演" ) rating: float = Field(description="电影评分,满分十分" ) model_with_structure = model.with_structured_output(MovieModel) response = model_with_structure.invoke("给出盗梦空间的信息" ) print(response) print(type(response))
输出:
1 2 3 4 5 6 7 8 Traceback... ValidationError: 2 validation errors for MovieModel title Field required [type=missing, input_value={'title1': '盗梦空间'...诺兰', 'rating': 9.3}, input_type=dict] For further information visit https://errors.pydantic.dev/2.12/v/missing year Field required [type=missing, input_value={'title1': '盗梦空间'...诺兰', 'rating': 9.3}, input_type=dict] For further information visit https://errors.pydantic.dev/2.12/v/missing
因为 title 和 year 字段不匹配抛出异常。
3.2.2 TypedDict 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 from typing_extensions import TypedDict, Annotatedfrom langchain_deepseek import ChatDeepSeekmodel = ChatDeepSeek( model="deepseek-v4-flash" , api_base="http://localhost:8889" , api_key=SecretStr("<KEY>" ) ) class MovieDict (TypedDict) : """ 电影的详细信息 """ title: Annotated[str, "电影标题" ] year: Annotated[int, "电影上映年份" ] director: Annotated[str, "导演" ] rating: Annotated[float, "电影评分,满分十分" ] structured_model = model.with_structured_output(MovieDict) response = structured_model.invoke("给出盗梦空间的信息" ) print(response) print(type(response))
输出:
1 2 {'title1': '盗梦空间', 'year2': 2010, 'director': '克里斯托弗·诺兰', 'rating': 9.3} <class 'dict'>
按照字典形式输出,不报错。
3.2.3 JSON Schema 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 from langchain_deepseek import ChatDeepSeekmodel = ChatDeepSeek( model="deepseek-v4-flash" , api_base="http://localhost:8889" , api_key=SecretStr("<KEY>" ) ) json_schema = { "title" : "Movie" , "description" : "A movie with details" , "type" : "object" , "properties" : { "title" : { "type" : "string" , "description" : "The title of the movie" }, "year" : { "type" : "integer" , "description" : "The year the movie was released" }, "director" : { "type" : "string" , "description" : "The director of the movie" }, "rating" : { "type" : "number" , "description" : "The movie's rating out of 10" } }, "required" : ["title" , "year" , "director" , "rating" ] } structured_model = model.with_structured_output( json_schema, method="json_schema" ) response = structured_model.invoke("给出盗梦空间的信息" ) print(response) print(type(response))
输出如下:
1 2 {'title1': '盗梦空间', 'year2': 2010, 'director': '克里斯托弗·诺兰', 'rating': 9.3} <class 'dict'>
按照字典形式输出,不报错。
3.3.4 @dataclass 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 from dataclasses import dataclassfrom langchain_deepseek import ChatDeepSeekmodel = ChatDeepSeek( model="deepseek-v4-flash" , api_base="http://localhost:8889" , api_key=SecretStr("<KEY>" ) ) @dataclass class Movie () : """ 电影的详细信息 """ title: str = Field(description="电影标题" ) year: int = Field(description="电影上映年份" ) director: str = Field(description="导演" ) rating: float = Field(description="电影评分,满分十分" ) structured_model = model.with_structured_output(Movie) response = structured_model.invoke("给出盗梦空间的信息" ) print(response) print(type(response))
输出如下:
1 2 {'title1': '盗梦空间', 'year2': 2010, 'director': '克里斯托弗·诺兰', 'rating': 9.3} <class 'dict'>
按照字典形式输出,不报错。
3.2.5 小结 用 Pydantic 定义 schema,在接收到响应后会进行校验,字段不匹配则抛出异常,其余三种方式不校验。
4、获取结构化结果方式 以上定义输出结构的四种模式中,我们都是通过调用 with_structured_output 来获取结构化输出结果。除了这种方式外,还可以通过使用输出解析器 来获取结构化输出结果。下面介绍这两种获取结构化结果的方式。
4.1 使用 with_structured_output 这种方式是最新 、最简洁 的 API,直接让模型”理解”你需要的数据结构,并返回解析好的对象。
此外,我们可以在 with_structured_output 方法中传入 include_raw=True 参数,表示返回解析前的原始 AIMessage ,从而访问令牌用量等元数据。
举例:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from langchain.chat_models import init_chat_modelfrom dotenv import load_dotenvimport osload_dotenv(override=True ) CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY" ) CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL" ) model = init_chat_model( model="gpt-5.4-mini" , model_provider="openai" , api_key=CLOSEAI_API_KEY, base_url=CLOSEAI_BASE_URL )
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from pydantic import BaseModel, Fieldfrom rich import print as rprintclass Movie (BaseModel) : """电影信息""" title: str = Field(description="电影标题" ) year: int = Field(description="上映年份" ) director: str = Field(description="导演" ) rating: float = Field(description="评分(10分制)" ) model_with_structure = model.with_structured_output(Movie, include_raw=True ) resp = model_with_structure.invoke("给我介绍下电影《星际穿越》" ) print(type(resp)) rprint(resp)
以上代码运行结果如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 <class 'dict'> { 'raw': AIMessage(content = '', additional_kwargs = { 'refusal': None }, response_metadata = { 'token_usage': { 'completion_tokens': 93, 'prompt_tokens': 373, 'total_tokens': 466, 'completion_tokens_details': None, 'prompt_tokens_details': { 'audio_tokens': None, 'cached_tokens': 320 }, 'prompt_cache_hit_tokens': 320, 'prompt_cache_miss_tokens': 53 }, 'model_provider': 'deepseek', 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_eaab8d114b_prod0820_fp8_kvcache', 'id': '679228cd-956f-49bf-85b5-38ec684baca3', 'finish_reason': 'tool_calls', 'logprobs': None }, id = 'lc_run--019bbc7f-2c63-7611-a2e6-c413c2d11e52-0', tool_calls = [{ 'name': 'Movie', 'args': { 'title': '星际穿越', 'year': 2014, 'director': '克里斯托弗·诺兰', 'rating': 9.3 }, 'id': 'call_00_WCnHeO1OYKS1gjuJfvmB1QHz', 'type': 'tool_call' }], invalid_tool_calls = [], usage_metadata = { 'input_tokens': 373, 'output_tokens': 93, 'total_tokens': 466, 'input_token_details': { 'cache_read': 320 }, 'output_token_details': {} }), 'parsed': Movie(title = '星际穿越', year = 2014, director = '克里斯托弗·诺兰', rating = 9.3), 'parsing_error': None }
输出包含了完整的输出响应,包含三个字段:
raw:返回的原始 AIMessage。
parsed:解析后的输出。
parsing_error:解析错误,当前用的是 Pydantic,校验格式不符合 schema 会导致报错。其它三种方式不符合 schema 不会导致报错。
4.2 使用输出解析器(不推荐) 这种方法更传统,依赖于在提示词中明确指示模型输出特定格式的文本,然后使用解析器进行转换。
其流程是:提示词指导 (引导生成指定类型)→ 模型生成文本 → 解析器转换 。
举例:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 from langchain_core.output_parsers import JsonOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom pydantic import BaseModel, Fieldfrom langchain.chat_models import init_chat_modelfrom dotenv import load_dotenvimport osprompt_template = ChatPromptTemplate.from_messages([ ("system" , "回答用户问题,必须始终输出一个包含title(电影标题)和year(上映年份)的 JSON 对象" ), ("human" , "问题:{question}" ) ]) load_dotenv(override=True ) model = init_chat_model( model="gpt-5.4-mini" , model_provider="openai" , api_key=os.getenv("CLOSEAI_API_KEY" ), base_url=os.getenv("CLOSEAI_BASE_URL" ) ) class Movie (BaseModel) : """电影信息""" title: str = Field(description="电影标题" ) year: int = Field(description="上映年份" ) parser = JsonOutputParser(pydantic_object=Movie) chain = prompt_template | model | parser response = chain.invoke({"question" : "介绍电影《盗梦空间》" }) print(response)
输出:
1 {'title': '盗梦空间', 'year': 2010}