📊 LangChain 结构化输出

"""
结构化输出基础示例
"""
from pydantic import BaseModel
from langchain.agents import create_agent

# 定义输出结构
class ContactInfo(BaseModel):
    """联系人信息"""
    name: str
    email: str
    phone: str

# 创建 Agent（使用 response_format 指定输出格式）
agent = create_agent(
    model="gpt-4o",
    response_format=ContactInfo  # 自动选择最佳策略
)

# 调用 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "提取联系人信息：张三，邮箱 [email protected]，电话 13812345678"
    }]
})

# 获取结构化数据
contact = result["structured_response"]
print(contact.name)   # 张三
print(contact.email)  # [email protected]
print(contact.phone)  # 13812345678

"""
ProviderStrategy - 使用原生 API
"""
from langchain.agents.structured_output import ProviderStrategy
from pydantic import BaseModel

class ContactInfo(BaseModel):
    name: str
    email: str

# 显式使用 ProviderStrategy
response_format = ProviderStrategy(schema=ContactInfo)

agent = create_agent(
    model="gpt-4o",  # 支持原生结构化输出
    response_format=response_format
)

"""
ToolStrategy - 基于工具调用
"""
from langchain.agents.structured_output import ToolStrategy

# 使用 ToolStrategy
response_format = ToolStrategy(schema=ContactInfo)

agent = create_agent(
    model="any-model",  # 任何支持工具调用的模型
    response_format=response_format
)

"""
Auto-Selection - 推荐方式
"""
# ✅ 最简单的方式：直接传入 Pydantic 模型
agent = create_agent(
    model="gpt-4o",
    response_format=ContactInfo  # 自动选择最佳策略
)

# LangChain 自动判断：
# - GPT-4o 支持原生 → 使用 ProviderStrategy
# - 其他模型 → 使用 ToolStrategy

"""
Pydantic Models - 完整示例
"""
from pydantic import BaseModel, Field
from typing import Literal

class ProductReview(BaseModel):
    """产品评论分析"""

    # 带约束的字段
    rating: int | None = Field(
        description="评分 1-5 星",
        ge=1,  # 大于等于 1
        le=5   # 小于等于 5
    )

    # 使用 Literal 限制选项
    sentiment: Literal["positive", "negative", "neutral"] = Field(
        description="情感倾向"
    )

    # 列表类型
    key_points: list[str] = Field(
        description="关键要点列表"
    )

    # 可选字段
    would_recommend: bool | None = Field(
        default=None,
        description="是否推荐购买"
    )

# 使用
agent = create_agent(
    model="gpt-4o",
    response_format=ProductReview
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "分析评论：这个产品很棒，5星推荐！质量好，价格实惠。"
    }]
})

review = result["structured_response"]
print(f"评分：{review.rating}")
print(f"情感：{review.sentiment}")
print(f"要点：{review.key_points}")

"""
Dataclasses - 简单场景
"""
from dataclasses import dataclass

@dataclass
class EventDetails:
    event_name: str
    date: str
    location: str

agent = create_agent(
    model="gpt-4o",
    response_format=EventDetails
)

"""
TypedDict - 字典类型
"""
from typing import TypedDict, Literal

class MeetingAction(TypedDict):
    task: str
    assignee: str
    priority: Literal["low", "medium", "high"]

agent = create_agent(
    model="gpt-4o",
    response_format=MeetingAction
)

"""
Union 类型 - 智能路由
"""
from typing import Union
from pydantic import BaseModel
from langchain.agents.structured_output import ToolStrategy

class SendEmailRequest(BaseModel):
    """发送邮件请求"""
    recipient: str
    subject: str
    body: str

class ScheduleMeetingRequest(BaseModel):
    """安排会议请求"""
    attendees: list[str]
    title: str
    duration_minutes: int

# 使用 Union 类型
response_format = ToolStrategy(
    schema=Union[SendEmailRequest, ScheduleMeetingRequest]
)

agent = create_agent(
    model="gpt-4o",
    response_format=response_format
)

# 示例 1：发送邮件
result1 = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "发邮件给 [email protected]，主题是'项目更新'"
    }]
})

# Agent 自动选择 SendEmailRequest
email_request = result1["structured_response"]
print(type(email_request))  # SendEmailRequest

# 示例 2：安排会议
result2 = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "安排一个 30 分钟的会议，邀请 Alice 和 Bob"
    }]
})

# Agent 自动选择 ScheduleMeetingRequest
meeting_request = result2["structured_response"]
print(type(meeting_request))  # ScheduleMeetingRequest

"""
默认错误处理 - 捕获所有验证错误
"""
from langchain.agents.structured_output import ToolStrategy
from pydantic import BaseModel, Field

class ProductRating(BaseModel):
    rating: int = Field(ge=1, le=5)  # 必须在 1-5 之间
    comment: str

response_format = ToolStrategy(
    schema=ProductRating,
    handle_errors=True  # 默认值，捕获所有错误并自动重试
)

agent = create_agent(
    model="gpt-4o",
    response_format=response_format
)

# 当模型返回 rating=6 时，自动重试
result = agent.invoke({
    "messages": [{"role": "user", "content": "评分这个产品"}]
})

"""
自定义错误消息
"""
response_format = ToolStrategy(
    schema=ProductRating,
    handle_errors="评分必须在 1-5 之间，并且必须提供评论。请重新生成。"
)

# 验证失败时，模型会收到这条自定义消息

"""
只处理特定异常
"""
response_format = ToolStrategy(
    schema=ProductRating,
    handle_errors=(ValueError, TypeError)  # 只重试这些异常
)

# 其他异常会传播到调用方

"""
自定义错误处理函数
"""
from langchain.agents.structured_output import (
    StructuredOutputValidationError,
    MultipleStructuredOutputsError
)

def custom_error_handler(error: Exception) -> str:
    """自定义错误处理逻辑"""

    if isinstance(error, StructuredOutputValidationError):
        # 字段验证失败
        return f"格式错误：{error}。请按照正确的格式重新生成。"

    elif isinstance(error, MultipleStructuredOutputsError):
        # 返回了多个输出
        return "你返回了多个输出，但只需要一个。请选择最相关的输出。"

    else:
        # 其他错误
        return f"发生错误：{str(error)}。请重试。"

response_format = ToolStrategy(
    schema=Union[SendEmailRequest, ScheduleMeetingRequest],
    handle_errors=custom_error_handler
)

agent = create_agent(
    model="gpt-4o",
    response_format=response_format
)

"""
禁用错误处理 - 所有异常直接抛出
"""
response_format = ToolStrategy(
    schema=ProductRating,
    handle_errors=False  # 不自动重试
)

# 验证失败时立即抛出异常

"""
实战用例 1：发票数据提取
"""
from pydantic import BaseModel, Field
from langchain.agents import create_agent

class InvoiceData(BaseModel):
    """发票数据结构"""
    invoice_id: str = Field(description="发票编号")
    amount: float = Field(description="总金额", gt=0)
    vendor: str = Field(description="供应商名称")
    date: str = Field(description="日期，格式 YYYY-MM-DD")
    items: list[str] = Field(description="商品列表")

agent = create_agent(
    model="gpt-4o",
    response_format=InvoiceData
)

# 从文本中提取发票信息
invoice_text = """
发票编号：INV-2024-001
日期：2024-01-15
供应商：ABC 公司
商品：笔记本电脑 x1、鼠标 x2、键盘 x1
总金额：5,800 元
"""

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": f"提取发票信息：\n{invoice_text}"
    }]
})

invoice = result["structured_response"]
print(f"发票号：{invoice.invoice_id}")
print(f"金额：{invoice.amount}")
print(f"供应商：{invoice.vendor}")
print(f"商品：{invoice.items}")

# 直接插入数据库
# db.invoices.insert(invoice.model_dump())

"""
实战用例 2：用户注册表单自动填充
"""
from pydantic import BaseModel, Field, EmailStr

class UserProfile(BaseModel):
    """用户资料"""
    first_name: str = Field(description="名")
    last_name: str = Field(description="姓")
    email: EmailStr = Field(description="有效的邮箱地址")
    age: int = Field(description="年龄", ge=18, le=120)
    preferences: list[str] = Field(description="兴趣爱好")

agent = create_agent(
    model="gpt-4o",
    response_format=UserProfile
)

# 从对话中提取用户信息
result = agent.invoke({
    "messages": [
        {"role": "user", "content": "我叫李华"},
        {"role": "assistant", "content": "你好李华！"},
        {"role": "user", "content": "我 25 岁，邮箱是 [email protected]"},
        {"role": "user", "content": "我喜欢编程、阅读和旅游"}
    ]
})

profile = result["structured_response"]
# 自动填充表单
form_data = profile.model_dump()
print(form_data)
# {
#     "first_name": "华",
#     "last_name": "李",
#     "email": "[email protected]",
#     "age": 25,
#     "preferences": ["编程", "阅读", "旅游"]
# }

"""
实战用例 3：基于意图的 API 路由
"""
from typing import Union
from pydantic import BaseModel

class SearchQuery(BaseModel):
    """搜索请求"""
    query: str
    filters: dict = {}

class CreateOrder(BaseModel):
    """创建订单"""
    product_id: str
    quantity: int

class CustomerSupport(BaseModel):
    """客服咨询"""
    category: str
    message: str

# Union 类型自动路由
agent = create_agent(
    model="gpt-4o",
    response_format=Union[SearchQuery, CreateOrder, CustomerSupport]
)

# 根据用户意图自动选择合适的 API
def handle_user_request(user_input: str):
    result = agent.invoke({
        "messages": [{"role": "user", "content": user_input}]
    })

    request = result["structured_response"]

    # 根据类型调用不同的 API
    if isinstance(request, SearchQuery):
        return search_api(request.query, request.filters)
    elif isinstance(request, CreateOrder):
        return order_api(request.product_id, request.quantity)
    elif isinstance(request, CustomerSupport):
        return support_api(request.category, request.message)

# 示例
handle_user_request("搜索蓝色的衬衫")  # → SearchQuery
handle_user_request("购买 2 件 产品ID-123")  # → CreateOrder
handle_user_request("我的订单还没到")  # → CustomerSupport

"""
OpenAI Strict Mode - 更严格的 Schema 约束
"""
response_format = ProviderStrategy(
    schema=ProductReview,
    strict=True  # 强制严格遵守 Schema
)

agent = create_agent(
    model="gpt-4o",
    response_format=response_format
)

# 1. 查看完整的消息历史
result = agent.invoke({...})
for msg in result["messages"]:
    print(msg)

# 2. 启用 LangSmith 追踪
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"

class Address(BaseModel):
    street: str
    city: str
    country: str

class Person(BaseModel):
    name: str
    addresses: list[Address]  # 嵌套模型

agent = create_agent(
    model="gpt-4o",
    response_format=Person
)