ag2-structured-output

用途

AG2 结构化输出用于从 AG2 的 Agent 获取带类型的 Python 值，而不是自由文本。当用户希望返回 Pydantic 模型、dataclass、dict、基础类型或 Union，而不只是字符串时使用。典型场景包括分类、抽取、打分、归一化，以及任何由下游代码解析模型回复的流程。它也支持校验失败时的自动重试。

主要能力

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）。

可传入的 Schema 类型

类型	得到的结果
基础类型（`int`、`float`、`bool`）	裸值，框架会为 API 包装为 `{"data": ...}`
`dataclass`	dataclass 实例
Pydantic `BaseModel`	模型实例
Union（`int \| str`、`(int, str)`）	其中一个备选项
`dict[K, V]`、`TypedDict`	已校验的 dict
`ResponseSchema(...)`	同上，并为提供方显式指定 `name` / `description`
`@response_schema` 可调用对象	自定义校验/解析逻辑
`PromptedSchema(inner)`	对无原生结构化输出能力的提供方，将 schema 注入系统提示

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 对象 schema：

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 schema 中。

支持异步校验器：

import json

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

PromptedSchema：面向无原生结构化输出的提供方

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

from ag2 import Agent, PromptedSchema

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

可包裹任意内部 schema（类型、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.")     # 回到默认（无 schema）

传入 response_schema=None 可在单次调用中移除 agent 上已设置的 schema。

重试

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

result = await reply.content(retries=3)   # 初始 + 最多 3 次重新询问
result = await reply.content(retries=math.inf)   # 仅交互使用——可能无限循环

校验错误会作为后续消息发回模型，使其自行纠正。

基础类型嵌入（embed）

裸基础类型（int、float、bool、list[T]、基础类型 Union）默认会被包装为 {"data": ...}——多数结构化输出 API 处理对象比处理裸值更可靠。content() 会透明地解包。可选择退出：

ResponseSchema(int, name="RawInt", embed=False)              # 模型必须产出裸 42
@response_schema(embed=False)
def parse_rating(value: int) -> int: ...

使用方式

通过 response_schema= 在构造 Agent 时设置，或在单次 ask() 调用时覆盖。调用后使用 await reply.content() 获取解析后的类型化值；原始文本仍可通过 reply.body 获取。可通过 retries= 参数控制校验失败时的重试次数。

适用场景

需要 Pydantic 模型、dataclass、dict、基础类型或 Union 返回值，而非字符串。
进行分类、抽取、打分、归一化等下游代码需要解析回复的任务。
需要在校验失败时自动重试。
使用不支持原生结构化输出的提供方时，可通过 PromptedSchema 注入 schema。

限制与注意事项

读取 reply.body 而非类型化输出：reply.body 是原始文本，需 await reply.content() 才进行解析。
忘记对 content() 使用 await：它是异步的，会得到协程而非值。
Pydantic 字段缺少 description：模型可能猜测每个字段填什么，应为每个不明显的键添加 Field(description=...)。
提供方不支持原生结构化输出时，应使用 PromptedSchema(...) 而非强行对抗 API。
生产环境中使用 retries=math.inf：面对无法满足要求的模型会无限循环，应使用有限次数。
单轮覆盖仅限单轮：在一次 ask() 中传入 response_schema=int 不会改变 agent 的默认值，下一轮会回到构造函数中设置的内容。

技能介绍

用途

主要能力

使用方式

适用场景

限制与注意事项