ag2-structured-output

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Structured output

结构化输出

When to use

适用场景

The user wants a Pydantic model, dataclass, dict, primitive, or union back — not a string.
They're doing classification, extraction, scoring, normalisation, or anything where downstream code parses the reply.
They want automatic retry on validation failure.

用户需要返回Pydantic模型、数据类、字典、基本类型或联合类型，而非字符串。
进行分类、提取、评分、标准化或任何需要下游代码解析回复的操作。
需要在验证失败时自动重试。

60-second recipe

60秒快速上手

python

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

is still the raw model text;

await reply.content()

runs validation and returns the parsed value. If validation fails,

content()

raises (e.g.

pydantic.ValidationError

python

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()      # → 类型化的TicketTriage实例
print(triage.category, triage.urgency)

reply.body

仍然是原始模型文本；

await reply.content()

会执行验证并返回解析后的值。如果验证失败，

content()

会抛出异常（例如

pydantic.ValidationError

）。

Schema types you can pass

可传入的Schema类型

Type	What you get
Primitive ( `int` , `float` , `bool` )	Bare value, framework wraps in `{"data": ...}` for the API
`dataclass`	Instance of the dataclass
Pydantic `BaseModel`	Instance of the model
Union ( `int \| str` , `(int, str)` )	One of the alternatives
`dict[K, V]` , `TypedDict`	Validated dict
`ResponseSchema(...)`	Same as above, with explicit `name` / `description` for the provider
`@response_schema` callable	Custom validation/parsing logic
`PromptedSchema(inner)`	Schema injected into the system prompt for providers without native structured output

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

ResponseSchema

— name your payload

ResponseSchema

— 为负载命名

Helps the provider treat the structured output as a named contract:

python

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)

帮助服务商将结构化输出视为一个命名契约：

python

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

— custom validation

@response_schema

— 自定义验证

For clamping, regex cleanup, decoding wrapped JSON, or combining fields:

python

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)

Multi-parameter form synthesises a JSON object schema from the parameter names:

python

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}

The function also participates in dependency injection —

Context

Variable

Inject

Depends

work the same way as in tools (and don't appear in the JSON schema).

Async validators are supported:

python

import json

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

用于限制范围、正则清理、解码包装的JSON或合并字段：

python

from ag2 import Agent, response_schema

@response_schema
def parse_rating(content: str) -> int:
    """解析评分并限制在1–5范围内。"""
    return max(1, min(5, int(content)))

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

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

python

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

支持异步验证器：

python

import json

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

PromptedSchema

— for providers without native structured output

PromptedSchema

— 针对不支持原生结构化输出的服务商

Injects the JSON schema into the system prompt instead of using

response_format

python

from ag2 import Agent, PromptedSchema

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

Wraps any inner schema (type,

ResponseSchema

@response_schema

callable). The validation logic stays the same; only the wire format changes.

Custom prompt template:

python

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

将JSON Schema注入系统提示词，而非使用

response_format

：

python

from ag2 import Agent, PromptedSchema

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

可包装任何内部Schema（类型、

ResponseSchema

、

@response_schema

可调用对象）。验证逻辑保持不变，仅传输格式会改变。

自定义提示词模板：

python

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

Per-turn override

单轮覆盖

python

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)

Pass

response_schema=None

to drop a schema set on the agent for one call.

python

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。

Retries

重试机制

When validation fails, automatically re-ask the model:

python

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

The validation error is sent back to the model as a follow-up so it can correct itself.

当验证失败时，自动重新请求模型：

python

result = await reply.content(retries=3)   # 初始请求 + 最多3次重试
result = await reply.content(retries=math.inf)   # 仅适用于交互场景 — 可能无限循环

验证错误会作为后续请求发送给模型，以便模型自我修正。

Primitive embedding (

embed

)

基本类型嵌入（

embed

）

Bare primitives (

int

float

bool

list[T]

, primitive unions) get wrapped in

{"data": ...}

by default — most structured-output APIs handle objects more reliably than bare values.

content()

transparently unwraps. Opt out:

python

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

默认情况下，纯基本类型（

int

、

float

、

bool

、

list[T]

、基本类型联合）会被包装为

{"data": ...}

格式 — 大多数结构化输出API处理对象比处理原始值更可靠。

content()

会自动解包。可选择关闭该功能：

python

ResponseSchema(int, name="RawInt", embed=False)              # 模型必须返回纯数字42
@response_schema(embed=False)
def parse_rating(value: int) -> int: ...

Going deeper

深入学习

Working starter:

assets/recipe_builder.py

(mirrors

code_examples/02

) — Pydantic model +

@tool

response_schema=

Full reference:
```
website/docs/user-guide/structured_output.mdx
```
— covers every schema type, multi-param
```
@response_schema
```
,
```
Field
```
constraints,
```
PromptedSchema
```
, retries, embedding semantics.

入门示例：

assets/recipe_builder.py

（对应

code_examples/02

）—— Pydantic模型 +

@tool

response_schema=

。

完整参考文档：
```
website/docs/user-guide/structured_output.mdx
```
— 涵盖所有Schema类型、多参数
```
@response_schema
```
、
```
Field
```
约束、
```
PromptedSchema
```
、重试机制、嵌入语义。

Common pitfalls

常见陷阱

Reading
reply.body
when you wanted typed output —
```
reply.body
```
is the raw text.
```
await reply.content()
```
does the parsing.
Forgetting
await
on
content()
— it's async; you'll get a coroutine, not the value.
No
description
in the Pydantic field — the LLM may guess what to put in each field. Add a
```
Field(description=...)
```
for every non-obvious key.
Provider doesn't support native structured output — wrap with
```
PromptedSchema(...)
```
rather than fighting the API.
retries=math.inf
in production — will loop forever on a model that can't comply. Use a finite count.
Per-turn override is single-turn — passing
```
response_schema=int
```
to one
```
ask()
```
doesn't change the agent's default. The next turn returns to whatever was set on the constructor.

需要类型化输出却读取
reply.body
—
```
reply.body
```
是原始文本。
```
await reply.content()
```
才会执行解析。
忘记在
content()
前加
await
— 它是异步方法，直接调用会得到协程对象而非结果值。
Pydantic字段未添加
description
— LLM可能会猜测字段内容。为每个非显而易见的键添加
```
Field(description=...)
```
。
服务商不支持原生结构化输出 — 使用
```
PromptedSchema(...)
```
包装，而非强行调用API。
生产环境中使用
retries=math.inf
— 若模型始终无法符合要求，会无限循环。请使用有限次数。
单轮覆盖仅生效一次 — 为某次
```
ask()
```
传入
```
response_schema=int
```
不会改变Agent的默认设置。下一轮会恢复为构造函数中设置的Schema。

ag2-structured-output

Original

Translation

Structured output

结构化输出

When to use

适用场景

60-second recipe

60秒快速上手

Schema types you can pass

可传入的Schema类型

`ResponseSchema`
— name your payload

`ResponseSchema`
— 为负载命名

`@response_schema`
— custom validation

`@response_schema`
— 自定义验证

`PromptedSchema`
— for providers without native structured output

`PromptedSchema`
— 针对不支持原生结构化输出的服务商

Per-turn override

单轮覆盖

Retries

重试机制

Primitive embedding (
`embed`
)

基本类型嵌入（
`embed`
）

Going deeper

深入学习

Common pitfalls

常见陷阱

ag2-structured-output

Original

Translation

Structured output

结构化输出

When to use

适用场景

60-second recipe

60秒快速上手

Schema types you can pass

可传入的Schema类型

ResponseSchema — name your payload

ResponseSchema — 为负载命名

@response_schema — custom validation

@response_schema — 自定义验证

PromptedSchema — for providers without native structured output

PromptedSchema — 针对不支持原生结构化输出的服务商

Per-turn override

单轮覆盖

Retries

重试机制

Primitive embedding (embed)

基本类型嵌入（embed）

Going deeper

深入学习

Common pitfalls

常见陷阱

`ResponseSchema`
— name your payload

`ResponseSchema`
— 为负载命名

`@response_schema`
— custom validation

`@response_schema`
— 自定义验证

`PromptedSchema`
— for providers without native structured output

`PromptedSchema`
— 针对不支持原生结构化输出的服务商

Primitive embedding (
`embed`
)

基本类型嵌入（
`embed`
）