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 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, repair_attempts: int = 0) -> 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
repair_attempts	int	Additional regeneration attempts after an invalid result. A retry sees the original conversation and a generic repair instruction, never the invalid value or exception text.	0

Raises:

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

Source code in symfonic/agent/structured.py

async def extract_structured_output(
    chat_model: Any,
    messages: list[BaseMessage],
    response_model: type[BaseModel],
    *,
    instruction: str | None = None,
    repair_attempts: int = 0,
) -> 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.
        repair_attempts: Additional regeneration attempts after an invalid
            result. A retry sees the original conversation and a generic
            repair instruction, never the invalid value or exception text.

    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 (
        isinstance(repair_attempts, bool)
        or not isinstance(repair_attempts, int)
        or not 0 <= repair_attempts <= 3
    ):
        raise StructuredOutputError(
            "repair_attempts must be an integer between 0 and 3"
        )

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

    last_error: StructuredOutputError | None = None
    for attempt in range(repair_attempts + 1):
        convo: list[BaseMessage] = list(messages)
        convo.append(HumanMessage(content=instruction or _DEFAULT_INSTRUCTION))
        if attempt:
            convo.append(HumanMessage(content=_REPAIR_INSTRUCTION))

        try:
            structured_model = chat_model.with_structured_output(response_model)
            result = await structured_model.ainvoke(convo)
            if isinstance(result, response_model):
                return result
            if isinstance(result, BaseModel):
                return response_model.model_validate(result.model_dump())
            if isinstance(result, dict):
                return response_model.model_validate(result)
            raise StructuredOutputError(
                "Unexpected structured-output result type "
                f"{type(result).__name__!r}; expected "
                f"{response_model.__name__} or dict."
            )
        except NotImplementedError as exc:
            raise StructuredOutputError(
                "The configured model provider does not implement structured "
                "output."
            ) from exc
        except StructuredOutputError as exc:
            last_error = exc
        except ValidationError as exc:
            last_error = StructuredOutputError(
                f"Model output did not match {response_model.__name__}: {exc}"
            )
        except Exception as exc:  # noqa: BLE001 — normalise provider errors
            last_error = StructuredOutputError(
                f"Structured-output extraction failed: {exc}"
            )

    assert last_error is not None
    raise last_error

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