symfonic.agent.structured

structured

Structured output — typed, schema-validated final answers.

symfonic-core has always been able to constrain the model's tool choice (supports_forced_tool_choice / ForcedToolChoiceResolver), but until now the framework offered no first-class way to make the agent's final answer conform to a caller-supplied schema. Adopters hand-rolled a forced "respond" tool or post-parsed free text.

This module closes that gap with a single, provider-agnostic extraction pass. The agent runs its normal tool loop to completion; when the caller passes a response_model to :meth:SymfonicAgent.run, the engine hands the final conversation to :func:extract_structured_output, which binds the Pydantic schema onto the chat model via LangChain's with_structured_output and returns a validated instance.

Design notes

• Post-loop, not mid-loop. Extraction is a terminal pass over the already-produced conversation, so tools run exactly as they do without structured output. This mirrors how Strands Agents layers structured output on top of the agent loop and avoids fighting the react topology.
• Provider-neutral. with_structured_output dispatches to whatever the underlying BaseChatModel supports (tool-calling for Anthropic / Bedrock Converse / OpenAI-compatible gateways, JSON mode where offered). The framework does not author the schema payload — the model does.
• Fail-loud. A provider whose chat model cannot do structured output, or a validation failure, raises :class:StructuredOutputError rather than silently returning None. Callers opted in explicitly by passing a schema; a quiet None would hide a real contract break.

StructuredOutputError

StructuredOutputError(message: str)

Bases: SymfonicAgentError

Raised when a structured-output extraction cannot be satisfied.

Carries code="unprocessable" so the FastAPI layer maps it to a 422 rather than a generic 500 — the request was well-formed, but the model could not produce a value matching the requested schema.

Source code in src/symfonic/agent/structured.py

def __init__(self, message: str) -> None:
    super().__init__(message, code="unprocessable")

extract_structured_output async

extract_structured_output(
    chat_model: Any,
    messages: list[BaseMessage],
    response_model: type[BaseModel],
    *,
    instruction: str | None = None,
) -> BaseModel

Return a validated response_model instance from the conversation.

Parameters:

Name	Type	Description	Default
chat_model	Any	The resolved BaseChatModel for the agent's action role. Must support with_structured_output.	required
messages	list[BaseMessage]	The accumulated conversation (LangChain messages) produced by the agent loop. Used verbatim as context for the extraction.	required
response_model	type[BaseModel]	A Pydantic BaseModel subclass describing the required shape of the final answer.	required
instruction	str | None	Optional override for the terminal instruction appended to messages before extraction.	None

Raises:

Type	Description
StructuredOutputError	When the provider cannot do structured output, the extraction call fails, or the result fails schema validation.

Source code in src/symfonic/agent/structured.py

async def extract_structured_output(
    chat_model: Any,
    messages: list[BaseMessage],
    response_model: type[BaseModel],
    *,
    instruction: str | None = None,
) -> BaseModel:
    """Return a validated ``response_model`` instance from the conversation.

    Args:
        chat_model: The resolved ``BaseChatModel`` for the agent's action
            role. Must support ``with_structured_output``.
        messages: The accumulated conversation (LangChain messages) produced
            by the agent loop. Used verbatim as context for the extraction.
        response_model: A Pydantic ``BaseModel`` subclass describing the
            required shape of the final answer.
        instruction: Optional override for the terminal instruction appended
            to ``messages`` before extraction.

    Raises:
        StructuredOutputError: When the provider cannot do structured output,
            the extraction call fails, or the result fails schema validation.
    """
    if not isinstance(response_model, type) or not issubclass(
        response_model, BaseModel
    ):
        raise StructuredOutputError(
            "response_model must be a Pydantic BaseModel subclass, got "
            f"{response_model!r}"
        )

    if not supports_structured_output(chat_model):
        raise StructuredOutputError(
            "The configured model provider does not support structured "
            "output (its chat model has no with_structured_output). Use a "
            "provider backed by a tool-calling or JSON-mode model."
        )

    convo: list[BaseMessage] = list(messages)
    convo.append(HumanMessage(content=instruction or _DEFAULT_INSTRUCTION))

    try:
        structured_model = chat_model.with_structured_output(response_model)
        result = await structured_model.ainvoke(convo)
    except StructuredOutputError:
        raise
    except NotImplementedError as exc:
        raise StructuredOutputError(
            "The configured model provider does not implement structured "
            "output."
        ) from exc
    except Exception as exc:  # noqa: BLE001 — normalise provider errors
        raise StructuredOutputError(
            f"Structured-output extraction failed: {exc}"
        ) from exc

    # ``with_structured_output`` normally returns the validated instance
    # directly, but some providers return a dict (JSON mode) — coerce and
    # re-validate so the caller always gets a typed instance.
    if isinstance(result, response_model):
        return result
    try:
        if isinstance(result, BaseModel):
            return response_model.model_validate(result.model_dump())
        if isinstance(result, dict):
            return response_model.model_validate(result)
    except ValidationError as exc:
        raise StructuredOutputError(
            f"Model output did not match {response_model.__name__}: {exc}"
        ) from exc

    raise StructuredOutputError(
        f"Unexpected structured-output result type {type(result).__name__!r}; "
        f"expected {response_model.__name__} or dict."
    )

supports_structured_output

supports_structured_output(chat_model: Any) -> bool

True when chat_model exposes a usable with_structured_output.

BaseChatModel declares the method but raises NotImplementedError for models that do not support it. A best-effort hasattr check plus a "not the base sentinel" guard is the cheapest reliable probe; the actual call is still wrapped so a late NotImplementedError degrades to a clean :class:StructuredOutputError.

Source code in src/symfonic/agent/structured.py

def supports_structured_output(chat_model: Any) -> bool:
    """True when ``chat_model`` exposes a usable ``with_structured_output``.

    ``BaseChatModel`` declares the method but raises ``NotImplementedError``
    for models that do not support it. A best-effort ``hasattr`` check plus a
    "not the base sentinel" guard is the cheapest reliable probe; the actual
    call is still wrapped so a late ``NotImplementedError`` degrades to a
    clean :class:`StructuredOutputError`.
    """
    return callable(getattr(chat_model, "with_structured_output", None))