from __future__ import annotations
from collections.abc import AsyncIterator, Mapping, Sequence
from dataclasses import dataclass, field
from math import isfinite
from typing import Any, Protocol, runtime_checkable
from ._compat import StrEnum
from .messages import AnyMessage, AssistantMessage
from .stop import StopInfo, classify_stop
[docs]
class ReasoningEffort(StrEnum):
"""Provider-neutral levels for controlling model reasoning effort.
The levels increase from ``minimal`` to ``ultra``. A provider maps the
levels it cannot express onto its closest supported value.
"""
OFF = "off"
MINIMAL = "minimal"
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
XHIGH = "xhigh"
MAX = "max"
ULTRA = "ultra"
[docs]
@dataclass(frozen=True, slots=True)
class ModelRequest:
"""Complete provider-neutral input for one model step.
The runtime rebuilds this object before every model call. ``messages``
therefore includes the current system instructions, restored context, and
completed tool round trips in chronological order.
Before-model hooks also receive this type. Each hook receives a fresh
shallow view of context at entry; the provider receives a final rebuilt
request after preprocessing. Frozen fields do not freeze nested messages.
Examples:
Inspect the shape received by a custom adapter::
request = ModelRequest(
messages=[
SystemMessage(content="Be concise."),
UserMessage(content="What is 20 + 22?"),
],
tools=[add.definition],
server_tools=[ServerToolDefinition(type="web_search")],
reasoning_effort=ReasoningEffort.LOW,
)
.. note::
Retry policy belongs to the model adapter, not to this request. One
``ModelRequest`` describes intent; it does not control transport attempts.
.. seealso::
:class:`~zett_agent.model.AgentModel` defines the receiving adapter, and
:class:`~zett_agent.model.ToolDefinition` defines exposed model tools.
"""
#: Fully assembled context, including instructions and tool round trips.
messages: Sequence[AnyMessage]
#: Tools exposed for this model step; empty means no tool definitions.
tools: Sequence[ToolDefinition] = ()
#: Provider-hosted tools that bypass the Agent's local tool executor.
server_tools: Sequence[ServerToolDefinition] = ()
#: Desired reasoning level; provider capabilities determine its mapping.
reasoning_effort: ReasoningEffort = ReasoningEffort.MEDIUM
#: Optional local function name or registered server-tool type the provider must call.
tool_choice: str | None = None
#: Whether the provider may return more than one local tool call in one response.
#:
#: Adapters map this preference when their protocol exposes an equivalent
#: switch. The Agent still decides whether returned calls are safe to run
#: concurrently; this flag alone never makes local execution concurrent.
parallel_tool_call: bool = True
#: Routing key that lets an endpoint reuse one cached prompt prefix.
#:
#: The Agent fills this from the run's session identity, so every step of one
#: conversation routes to the same cache. Adapters map it to their protocol;
#: the OpenAI Chat Completions and Responses APIs both call it
#: ``prompt_cache_key``. None declares no key and leaves routing to the
#: endpoint.
cache_key: str | None = None
def __post_init__(self) -> None:
if not isinstance(self.parallel_tool_call, bool):
raise ValueError("parallel_tool_call must be a boolean")
if self.cache_key is not None and not self.cache_key.strip():
raise ValueError("cache_key must be a non-empty string when supplied")
[docs]
@dataclass(frozen=True, slots=True)
class RetryOptions:
"""Model-owned exponential backoff, without jitter.
Wait base_delay seconds before the first retry, then double the delay
until max_delay. max_retries counts additional attempts, not the initial
request; zero disables retries. No sleep occurs after the final failure.
Examples:
Usage::
RetryOptions(base_delay=1.0, max_delay=5.0, max_retries=4)
# At most five requests, separated by 1, 2, 4, and 5 seconds.
"""
#: Initial retry delay in seconds; zero allows immediate retries.
base_delay: float = 0.25
#: Hard upper bound in seconds, including the first retry delay.
max_delay: float = 8.0
#: Maximum additional requests after transient failures.
max_retries: int = 3
def __post_init__(self) -> None:
for name in ("base_delay", "max_delay"):
value = getattr(self, name)
if isinstance(value, bool) or not isinstance(value, (int, float)) or not isfinite(value) or value < 0:
raise ValueError(f"{name} must be a finite non-negative number")
if isinstance(self.max_retries, bool) or not isinstance(self.max_retries, int) or self.max_retries < 0:
raise ValueError("max_retries must be a non-negative integer")
DEFAULT_RETRY_OPTIONS = RetryOptions()
[docs]
@dataclass(frozen=True, slots=True)
class TimeoutOptions:
"""How long a provider may stay silent before the request is given up on.
A model call has three stretches where nothing may arrive, and each gets its own
limit, because the right wait is very different for each:
* **connect**: opening the connection. A reachable host answers in milliseconds.
* **first byte**: from sending the request until the response *headers* come back.
Streaming APIs answer headers before the model has produced anything, so a long
wait here means a gateway or proxy that took the request and did nothing.
* **idle**: the longest gap between two chunks once the response has started. This
can legitimately be minutes, because a reasoning model may think silently.
There is deliberately no limit on the *total* time of a response: a long answer
that keeps streaming is healthy. A limit that is exceeded raises
:class:`~zett_agent.providers.base.ProviderTimeoutError` naming which stretch it
was. A timeout is not retried, except for a failed connection, because a retry
would wait the whole limit again and the first request may still be running (and
billed) on the other side.
``None`` disables a limit, which is how a provider behaved when none was set.
Examples:
Fail fast on a gateway that goes silent, and allow long silent reasoning::
OpenAIProvider("gpt-5", key, timeout=TimeoutOptions(first_byte=30, idle=900))
"""
#: Seconds to open the connection.
connect: float | None = 5.0
#: Seconds from sending the request to receiving the response headers.
first_byte: float | None = 60.0
#: Seconds of silence tolerated between two chunks of a response that has started.
idle: float | None = 600.0
def __post_init__(self) -> None:
for name in ("connect", "first_byte", "idle"):
value = getattr(self, name)
if value is None:
continue
if isinstance(value, bool) or not isinstance(value, (int, float)) or not isfinite(value) or value <= 0:
raise ValueError(f"{name} must be a finite positive number of seconds, or None")
DEFAULT_TIMEOUT_OPTIONS = TimeoutOptions()
def validate_retry(retry: RetryOptions) -> None:
"""Reject invalid model configuration before opening network clients."""
if not isinstance(retry, RetryOptions):
raise ValueError("retry must be a RetryOptions instance")
def validate_timeout(timeout: TimeoutOptions) -> None:
"""Reject an invalid timeout configuration before opening network clients."""
if not isinstance(timeout, TimeoutOptions):
raise ValueError("timeout must be a TimeoutOptions instance")
def validate_response(response: bool) -> None:
"""Require an explicit boolean for selecting the model transport API."""
if not isinstance(response, bool):
raise ValueError("response must be a boolean")
[docs]
@dataclass(frozen=True, slots=True)
class ModelUsage:
"""Normalized token and cache measurements reported by a provider adapter.
``input_tokens`` is the complete model input, including cache reads and
cache writes. ``output_tokens`` includes reasoning tokens. This invariant
makes totals and cache-hit ratios comparable across provider schemas.
``cache_reported`` separates a verified cache miss from a protocol that
never reports cache counters. Compatible gateways often relay one protocol
through another and drop those counters on the way, so a zero without
``cache_reported`` means unknown rather than measured.
"""
#: Total input tokens, including cached input and cache writes.
input_tokens: int = 0
#: Total generated tokens, including reasoning tokens when reported.
output_tokens: int = 0
#: Input tokens reused from the provider cache, not extra input tokens.
cache_read_tokens: int = 0
#: Input tokens written to the provider cache, not extra input tokens.
cache_write_tokens: int = 0
#: Whether the response carried cache counters at all; False means unknown.
cache_reported: bool = True
#: Reasoning subset of output_tokens; zero when not reported.
reasoning_tokens: int = 0
def __post_init__(self) -> None:
values = (
self.input_tokens,
self.output_tokens,
self.cache_read_tokens,
self.cache_write_tokens,
self.reasoning_tokens,
)
if any(value < 0 for value in values):
raise ValueError("Model usage counters cannot be negative")
if self.cache_read_tokens > self.input_tokens:
raise ValueError("cache_read_tokens cannot exceed input_tokens")
if self.cache_write_tokens > self.input_tokens:
raise ValueError("cache_write_tokens cannot exceed input_tokens")
if self.reasoning_tokens > self.output_tokens:
raise ValueError("reasoning_tokens cannot exceed output_tokens")
@property
def total_tokens(self) -> int:
"""Return input plus output without double-counting cache or reasoning."""
return self.input_tokens + self.output_tokens
@property
def cache_hit_rate(self) -> float | None:
"""Return cached input divided by total input, or None when unknown.
None means there is no ratio to report: the request had no input tokens,
or the provider did not report cache counters for it. A protocol that
reports cache usage still yields zero for a genuine miss.
"""
if self.input_tokens == 0 or not self.cache_reported:
return None
return self.cache_read_tokens / self.input_tokens
[docs]
@dataclass(frozen=True, slots=True)
class ModelResponse:
"""Final normalized response for one model step.
Exactly one response terminates a successful model stream. Its message must
contain complete tool calls even if partial tool-call deltas were emitted.
Examples:
Complete a text-only stream::
response = ModelResponse(
message=AssistantMessage(content="The answer is 42."),
finish_reason="stop",
usage=ModelUsage(input_tokens=18, output_tokens=6),
)
yield ModelEvent.completed(response)
.. seealso::
:class:`~zett_agent.model.ModelUsage` normalizes provider token counters.
"""
#: Complete assistant output, including any tool calls and replay blocks.
message: AssistantMessage
#: Provider completion reason, or None when unavailable. It is the provider's own
#: value; :attr:`stop` is the normalized form to branch on.
finish_reason: str | None = None
#: Normalized usage for this model call, not the entire conversation.
usage: ModelUsage = field(default_factory=ModelUsage)
#: Why the step stopped, normalized. An adapter that knows more than the bare
#: ``finish_reason`` says (a refusal category, an error code) supplies it. When it
#: does not, it is derived from ``finish_reason`` and the tool calls, so an adapter
#: written before this field existed still reports a refusal or a truncation.
stop: StopInfo | None = None
def __post_init__(self) -> None:
if self.stop is None:
object.__setattr__(
self,
"stop",
classify_stop(self.finish_reason, has_tool_calls=bool(self.message.tool_calls)),
)
[docs]
class ModelEventType(StrEnum):
"""Streaming events emitted by a model adapter."""
TEXT_DELTA = "text_delta"
REASONING_DELTA = "reasoning_delta"
TOOL_CALL_DELTA = "tool_call_delta"
SERVER_TOOL_STARTED = "server_tool_started"
SERVER_TOOL_INPUT_DELTA = "server_tool_input_delta"
SERVER_TOOL_COMPLETED = "server_tool_completed"
SERVER_TOOL_FAILED = "server_tool_failed"
RESPONSE = "response"
[docs]
@dataclass(frozen=True, slots=True)
class ModelEvent:
"""One provider-neutral event emitted by an :class:`AgentModel`.
.. zett-diagram:: model-stream
+---------------------+
| Text deltas |----------+
+---------------------+ |
|
+---------------------+ |
| Reasoning deltas |----------+
+---------------------+ |
| +------------------------------+
+---------------------+ +---->| terminal ModelResponse |
| Tool-call deltas |----------+ | message + usage + finish |
+---------------------+ | +------------------------------+
|
+---------------------+ |
| Server-tool events |----------+
+---------------------+
Examples:
A minimal deterministic model can stream text and then return the same
complete text in its terminal message::
async def stream(self, request):
yield ModelEvent.text("Hello")
yield ModelEvent.completed(
ModelResponse(AssistantMessage(content="Hello"))
)
.. warning::
Deltas are incremental display data. Do not append them as independent
conversation messages, and do not omit the terminal response.
.. seealso::
:class:`~zett_agent.events.AgentEvent` is the higher-level stream consumed by
applications after runtime lifecycle events are added.
"""
#: Discriminator indicating which optional event field is populated.
type: ModelEventType
#: Incremental answer or reasoning text for the corresponding delta type.
delta: str = ""
#: Partial tool identifier/name/arguments for TOOL_CALL_DELTA events.
tool_call_delta: ToolCallDelta | None = None
#: Hosted invocation identity for a server-tool start event.
server_tool_call: ServerToolCall | None = None
#: Hosted incremental input associated with an earlier start event.
server_tool_input_delta: ServerToolInputDelta | None = None
#: Hosted terminal result for server-tool completion or failure events.
server_tool_result: ServerToolResult | None = None
#: Complete final response, present only for RESPONSE events.
response: ModelResponse | None = None
[docs]
@classmethod
def text(cls, delta: str) -> ModelEvent:
"""Create one answer-text fragment event."""
return cls(ModelEventType.TEXT_DELTA, delta=delta)
[docs]
@classmethod
def reasoning(cls, delta: str) -> ModelEvent:
"""Create one provider-reported reasoning fragment event."""
return cls(ModelEventType.REASONING_DELTA, delta=delta)
[docs]
@classmethod
def completed(cls, response: ModelResponse) -> ModelEvent:
"""Create the single terminal response event for a successful model call."""
return cls(ModelEventType.RESPONSE, response=response)
[docs]
@runtime_checkable
class AgentModel(Protocol):
"""Protocol implemented by every provider or custom model adapter.
An adapter translates :class:`ModelRequest` into one provider request and
maps the provider stream back into :class:`ModelEvent` objects. The Agent
remains independent of SDK-specific chunks.
Examples:
Implement a complete offline adapter::
class EchoModel:
retry = RetryOptions(max_retries=0)
async def stream(self, request):
text = request.messages[-1].text
yield ModelEvent.text(text)
yield ModelEvent.completed(
ModelResponse(AssistantMessage(content=text))
)
model: AgentModel = EchoModel()
.. note::
``stream`` is declared as a regular protocol method returning an
``AsyncIterator``. Implementations normally use ``async def`` with
``yield``. Runtime protocol checks validate attribute presence, not the
implementation's async semantics; integration tests must consume it.
.. seealso::
:doc:`/extending/model-adapter` provides a complete adapter walkthrough;
:class:`~zett_agent.model.RetryOptions` defines the model-owned retry policy.
"""
#: Model-owned backoff policy; max_retries=0 disables retries.
#: This configuration never belongs to ModelRequest.
retry: RetryOptions = RetryOptions()
[docs]
def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
"""Stream deltas and finish with exactly one response event.
The model owns and applies its retry configuration for every request.
Transient failures may be retried only before the first emitted event.
"""
...
