7. 使用 Pydantic 实现结构化输出

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
# 向 LLM 询问产品信息
message = HumanMessage(content="""
从以下文本中提取产品信息:
"UltraWidget Pro 售价 $299.99,目前有货。"
""")
 
response = llm.invoke([message])
print(response.content)

产品名称: UltraWidget Pro
价格: $299.99
库存状态: 有货

# 脆弱的解析方法
text = response.content
price_line = [line for line in text.split('\n') if '价格:' in line][0]
price_str = price_line.split('$')[1]
price = float(price_str)  # 脆弱 - 如果格式改变怎么办?

# 示例 1
"该产品是 UltraWidget Pro,售价 $299.99,有货。"
 
# 示例 2
"产品: UltraWidget Pro
成本: 299.99 美元
状态: 有货"
 
# 示例 3
"UltraWidget Pro - $299.99 (有货)"
 
# 示例 4
"我找到了 UltraWidget Pro。它售价 $299.99,目前可以购买。"

# 你期望这种结构
{
    "product_name": "UltraWidget Pro",
    "price": 299.99,
    "in_stock": true
}

from pydantic import BaseModel
 
class ProductInfo(BaseModel):
    product_name: str
    price: float
    in_stock: bool

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
 
# 步骤 1: 定义模式
class ProductInfo(BaseModel):
    product_name: str = Field(description="完整的产品名称")
    price: float = Field(description="以美元计的价格")
    in_stock: bool = Field(description="产品是否有货")
 
# 步骤 2: 将模式绑定到 LLM
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(ProductInfo)
 
# 步骤 3: 调用并获取类型化对象
message = HumanMessage(content="""
从以下文本中提取产品信息:
"UltraWidget Pro 售价 $299.99,目前有货。"
""")
 
result = structured_llm.invoke([message])
 
# result 现在是 ProductInfo 对象,而不是字符串
print(type(result))  # <class '__main__.ProductInfo'>
print(result.product_name)  # UltraWidget Pro
print(result.price)  # 299.99
print(result.in_stock)  # True

# 常规 Pydantic - 描述是给人类看的文档
class User(BaseModel):
    name: str = Field(description="用户名称")  # 没有它也能正常工作

# 使用 LLM - 描述决定 LLM 行为
class CustomerFeedback(BaseModel):
    sentiment: str = Field(
        description="整体情感: 'positive'、'negative' 或 'neutral'"
    )

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
 
class CustomerFeedback(BaseModel):
    sentiment: str = Field(
        description="整体情感: 'positive'、'negative' 或 'neutral'"
    )
    main_issue: str = Field(
        description="主要投诉或关注点(如果有)。如果没有提到问题,使用 'none'。"
    )
    urgency: str = Field(
        description="问题的紧急程度: 'low'、'medium' 或 'high'"
    )
 
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(CustomerFeedback)
 
message = HumanMessage(content="""
分析这条客户反馈:
"产品运行良好,但配送花了 3 周时间。我已经错过了项目截止日期。请立即回复。"
""")
 
result = structured_llm.invoke([message])
print(result.sentiment)    # negative
print(result.main_issue)   # Slow shipping
print(result.urgency)      # high

sentiment: str  # 没有描述

sentiment: str = Field(
    description="情感: 恰好是 'positive'、'negative' 或 'neutral'(小写)"
)

result.sentiment = "Positive"    # 首字母大写
result.sentiment = "NEGATIVE"    # 全大写
result.sentiment = "good"        # 完全不同的词

allowed = {"positive", "negative", "neutral"}
 
# 转换为小写并检查
sentiment = result.sentiment.lower()
 
if sentiment not in allowed:
    sentiment = "neutral"  # 对意外值使用默认值
 
# 现在 sentiment 保证是允许值之一

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
message = HumanMessage(content="""
从以下内容中提取产品名称、价格和库存状态:
"UltraWidget Pro 售价 $299.99,目前有货。"
格式: 名称 | 价格 | 库存状态
""")
 
response = llm.invoke([message])
text = response.content
 
# 手动解析
parts = text.split('|')
product_name = parts[0].strip()
price_str = parts[1].strip().replace('$', '')
price = float(price_str)
availability = parts[2].strip().lower()
in_stock = 'in stock' in availability or 'available' in availability
 
print(f"名称: {product_name}")
print(f"价格: ${price}")
print(f"有货: {in_stock}")

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
 
class ProductInfo(BaseModel):
    product_name: str = Field(description="完整的产品名称")
    price: float = Field(description="以美元计的价格")
    in_stock: bool = Field(description="产品是否有货")
 
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(ProductInfo)
 
message = HumanMessage(content="""
从以下内容中提取产品信息:
"UltraWidget Pro 售价 $299.99,目前有货。"
""")
 
result = structured_llm.invoke([message])
 
print(f"名称: {result.product_name}")
print(f"价格: ${result.price}")
print(f"有货: {result.in_stock}")

from pydantic import BaseModel
 
class ProductInfo(BaseModel):
    product_name: str
    price: float
    in_stock: bool

from typing import Optional
 
class ProductReview(BaseModel):
    rating: int
    review_text: str
    reviewer_name: Optional[str] = None  # 匿名评论没有评论者姓名

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from typing import Optional
 
class ProductInfo(BaseModel):
    product_name: str
    price: float
    in_stock: bool
    discount_percentage: Optional[float] = None
    warranty_years: Optional[int] = None
 
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(ProductInfo)
 
message = HumanMessage(content="""
提取产品信息: "UltraWidget Pro 售价 $299.99,有货。"
""")
 
result = structured_llm.invoke([message])
print(result.product_name)  # UltraWidget Pro
print(result.price)  # 299.99
print(result.in_stock)  # True
print(result.discount_percentage)  # None (未提及)
print(result.warranty_years)  # None (未提及)

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, ValidationError
 
class ProductInfo(BaseModel):
    product_name: str
    price: float
    in_stock: bool
 
llm = ChatAnthropic(model='claude-sonnet-4-5')
structured_llm = llm.with_structured_output(ProductInfo)
 
# 输入缺少必需信息
message = HumanMessage(content="""
从以下内容中提取产品信息: "这个小部件很棒!强烈推荐。"
""")
 
try:
    result = structured_llm.invoke([message])
    print(result)
except ValidationError as e:
    print("发生 ValidationError")

class ProductInfo(BaseModel):
    product_name: str
    price: float
    in_stock: bool
 
message = HumanMessage(content="""
从以下内容中提取产品信息: "这个小部件很棒!"
""")
 
# 当一些 LLM 为 product_name、price 和 in_stock 虚构值时
result = structured_llm.invoke([message])
# 没有引发错误!
print(result.product_name)  # "widget" (从文本中提取)
print(result.price)  # 0.0 (虚构的!)
print(result.in_stock)  # False (虚构的!)
 
# 问题: 你无法分辨哪些值是真实的 vs 虚构的

from typing import Optional
from pydantic import BaseModel, model_validator, ValidationError
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
 
class ProductInfo(BaseModel):
    # 这些实际上是必需的,但声明为 Optional
    # 真正的验证发生在下面的验证器中
    product_name: Optional[str] = None
    price: Optional[float] = None
    in_stock: Optional[bool] = None
    
    @model_validator(mode='after')
    def check_required_fields(self):
        """验证所有必要字段都存在"""
        if self.product_name is None or self.price is None or self.in_stock is None:
            raise ValueError("必须提供所有字段 (product_name, price, in_stock)")
        return self
 
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(ProductInfo)
 
# 使用不完整数据测试
message = HumanMessage(content="""
从以下内容中提取产品信息: "这个小部件很棒!"
""")
 
try:
    result = structured_llm.invoke([message])
    # 如果到达这里,所有字段都保证存在
    print(f"产品: {result.product_name}")
    print(f"价格: ${result.price}")
except ValidationError as e:
    # 必需字段缺失 - 提取失败
    print(f"提取不完整: {e}")