ag2-structured-output

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.

60-second recipe

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`).

## Schema types you can pass

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

## `ResponseSchema` — name your payload

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)

@response_schema — custom validation

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

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:

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:

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

Injects the JSON schema into the system prompt instead of using response_format:

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:

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

Per-turn override

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.

Retries

When validation fails, automatically re-ask the model:

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.

Primitive embedding (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:

ResponseSchema(int, name="RawInt", embed=False)              # model must produce a bare 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.

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.