ag2-structured-output

结构化输出

适用场景

• 用户想要的是 Pydantic 模型、dataclass、dict、基本类型或联合类型，而不是字符串。
• 用户在做分类、提取、评分、规范化，或任何需要下游代码解析回复的任务。
• 用户希望在验证失败时自动重试。

60 秒快速上手

from pydantic import BaseModel, Field
from typing import Annotated

from ag2 import Agent
from ag2.config import OpenAIConfig

class TicketTriage(BaseModel):
    category: Annotated[str, Field(description="e.g. billing, bug, account_access")]
    urgency: Annotated[str, Field(description="low, medium, or high")]
    summary_one_line: Annotated[str, Field(description="Max 120 characters", max_length=120)]

agent = Agent(
    "triage",
    prompt="You triage support messages. Be conservative with urgency.",
    config=OpenAIConfig(model="gpt-4o-mini"),
    response_schema=TicketTriage,
)

reply = await agent.ask("I was charged twice and can't export reports. Quarter close blocked.")
triage = await reply.content()      # → typed TicketTriage
print(triage.category, triage.urgency)

reply.body 仍是原始模型文本；await reply.content() 会执行验证并返回解析后的值。如果验证失败，content() 会抛出异常（例如 pydantic.ValidationError）。

可传入的架构类型

类型	返回结果
基本类型（int、float、bool）	原始值；框架会为 API 将其包装为 {"data": ...}
dataclass	dataclass 实例
Pydantic BaseModel	模型实例
联合类型（int | str、(int, str)）	其中一种类型的值
dict[K, V]、TypedDict	经过验证的 dict
ResponseSchema(...)	同上，并为提供方显式指定 name / description
@response_schema 可调用对象	自定义验证／解析逻辑
PromptedSchema(inner)	将架构注入系统提示词，供不支持原生结构化输出的提供方使用

ResponseSchema — 为载荷命名

帮助提供方将结构化输出视为一个具有名称的约定：

from ag2 import Agent, ResponseSchema

schema = ResponseSchema(int | str, name="ByteWidth", description="Number of bits in one byte.")
agent = Agent("assistant", config=config, response_schema=schema)

@response_schema — 自定义验证

适用于钳制数值、正则清理、解码包装后的 JSON，或组合字段：

from ag2 import Agent, response_schema

@response_schema
def parse_rating(content: str) -> int:
    """Parse a rating and clamp to 1–5."""
    return max(1, min(5, int(content)))

agent = Agent("assistant", config=config, response_schema=parse_rating)

多参数形式会根据参数名称合成 JSON 对象架构：

from typing import Annotated
from pydantic import Field
from ag2 import response_schema

@response_schema
def extract_listing(
    title: Annotated[str, Field(description="Product name")],
    price_usd: Annotated[float, Field(description="Price in USD", ge=0)],
    in_stock: Annotated[bool, Field(description="True if it ships now")],
) -> dict:
    return {"title": title, "price_usd": price_usd, "in_stock": in_stock}

该函数也支持依赖注入——Context、Variable、Inject、Depends 的用法与工具中的相同（且不会出现在 JSON 架构中）。

支持异步验证器：

import json

@response_schema
async def fetch_and_validate(content: str) -> dict:
    data = json.loads(content)
    data["validated"] = True
    return data

PromptedSchema — 用于不支持原生结构化输出的提供方

将 JSON 架构注入系统提示词，而不是使用 response_format：

from ag2 import Agent, PromptedSchema

agent = Agent("assistant", config=config, response_schema=PromptedSchema(int))

可包装任何内部架构（类型、ResponseSchema、@response_schema 可调用对象）。验证逻辑保持不变；只有线上的格式会改变。

自定义提示词模板：

PromptedSchema(int, prompt_template="Reply with JSON matching this schema:\n{schema}")

单轮覆盖

agent = Agent("assistant", config=config)

turn = await agent.ask("How many seconds in a minute?", response_schema=int)
print(await turn.content())   # 60

turn2 = await turn.ask("Say hello.")     # back to default (no schema)

传入 response_schema=None，可在一次调用中取消为 Agent 设置的架构。

重试

验证失败时，自动重新询问模型：

result = await reply.content(retries=3)   # initial + up to 3 re-asks
result = await reply.content(retries=math.inf)   # interactive only — could loop forever

验证错误会作为后续消息发送回模型，以便模型自行修正。

基本类型嵌入（embed）

默认情况下，原始基本类型（int、float、bool、list[T]、基本类型联合）会被包装为 {"data": ...}——大多数结构化输出 API 处理对象比处理原始值更可靠。content() 会透明地解除包装。若要禁用此行为：

ResponseSchema(int, name="RawInt", embed=False)              # model must produce a bare 42
@response_schema(embed=False)
def parse_rating(value: int) -> int: ...

深入了解

• 可运行的入门示例：assets/recipe_builder.py（对应 code_examples/02）——包含 Pydantic 模型、@tool 和 response_schema=。
• 完整参考：website/docs/user-guide/structured_output.mdx——涵盖所有架构类型、多参数 @response_schema、Field 约束、PromptedSchema、重试和嵌入语义。

常见问题

• 想要类型化输出，却读取了 reply.body — reply.body 是原始文本。使用 await reply.content() 进行解析。
• 忘记对 content() 使用 await — 它是异步的；否则会得到协程，而不是值。
• Pydantic 字段没有 description — LLM 可能会猜测每个字段应填入什么。为每个含义不明显的键添加 Field(description=...)。
• 提供方不支持原生结构化输出 — 使用 PromptedSchema(...) 包装，而不是试图绕过 API 的限制。
• 在生产环境中使用 retries=math.inf — 对于无法遵循要求的模型，这会导致无限循环。请使用有限次数。
• 单轮覆盖仅对单轮生效 — 将 response_schema=int 传给一次 ask() 不会更改 Agent 的默认设置。下一轮会恢复为构造函数中设置的值。