Class StructuredOutputCollector

src/workflow/structured.py:41–72

Accumulates a schema subagent's ``StructuredOutput`` emissions. The injected tool calls :meth:`offer` for each emission. A valid object is captured and ends the run; an invalid one returns the validation error (fed back to the model as a tool error so it retries) until the retry cap is

@dataclass
class StructuredOutputCollector:
    """Accumulates a schema subagent's ``StructuredOutput`` emissions.

    The injected tool calls :meth:`offer` for each emission. A valid object is
    captured and ends the run; an invalid one returns the validation error (fed
    back to the model as a tool error so it retries) until the retry cap is hit.
    """

    schema: Mapping[str, Any]
    max_retries: int = MAX_STRUCTURED_OUTPUT_RETRIES
    attempts: int = 0
    value: Any = None
    succeeded: bool = False
    last_error: Optional[str] = None

    def offer(self, obj: Any) -> tuple[bool, Optional[str]]:
        """Validate one emission. Returns ``(accepted, error_message)``."""
        if self.succeeded:
            return True, None
        self.attempts += 1
        ok, error = validate_structured(obj, self.schema)
        if ok:
            self.value = obj
            self.succeeded = True
            return True, None
        self.last_error = error
        return False, error

    @property
    def exhausted(self) -> bool:
        """True once retries are spent without a valid object."""
        return not self.succeeded and self.attempts >= self.max_retries


def make_structured_output_tool(collector: StructuredOutputCollector) -> Tool: