AgentModel

Implement the model contract and normalize usage and stream output. See the guide.

class zett_agent.model.AgentModel(*args, **kwargs)

Protocol implemented by every provider or custom model adapter.

An adapter translates ModelRequest into one provider request and maps the provider stream back into 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.

See also

Implement a model adapter provides a complete adapter walkthrough; RetryOptions defines the model-owned retry policy.

retry: RetryOptions = RetryOptions(base_delay=0.25, max_delay=8.0, max_retries=3)

Model-owned backoff policy; max_retries=0 disables retries. This configuration never belongs to ModelRequest.

stream(request)

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.

Parameters:

request (ModelRequest)

Return type:

AsyncIterator[ModelEvent]