Agent#

Understand request ownership, phases, and runtime limits. See the guide.

class zett_agent.agent.Agent(model, *, system_prompt='You are a helpful assistant.', tools=(), extensions=None, reasoning_effort=ReasoningEffort.MEDIUM, parallel_tool_call=True, max_iterations=36, max_internal_messages=8)[源代码]#

Run a model, execute requested tools, and repeat until an answer is ready.

Agent is the low-level runtime. Most applications can use create_agent(), which initializes it and optionally dispatches events. Direct construction is useful when initialization must happen later.

        flowchart TD
    user[UserMessage] --> model[model]
    model -- ToolCall --> tool[tool]
    tool -- ToolMessage --> model
    model -- final answer --> assistant[AssistantMessage]

The model-tool loop#

Tools are parallel by default. When parallel calling is enabled, parallel tools from one model response run together before explicitly serial tools. Hooks, Raw Log appends, and completion events remain ordered. Streaming deltas are display data; only the final ModelResponse is appended as an assistant message.

示例

Prefer the asynchronous factory when using the runtime directly:

agent = await Agent.create(
    model,
    config=AgentRunConfig(session_id="chat-42"),
    tools=[read_file, grep],
    reasoning_effort=ReasoningEffort.HIGH,
)
answer = await agent.run("Find where retries are configured")
print(answer.content)

Stream visible events and always close an abandoned iterator:

from contextlib import aclosing

async with aclosing(agent.stream("Inspect the tests")) as events:
    async for event in events:
        if event.type is AgentEventType.TEXT_DELTA:
            print(event.delta, end="", flush=True)

警告

An Agent must be initialized before run or stream. Use create() or await initialize() after direct construction.

参见

create_agent() is the convenient initialized entry point. AgentClient adds application event dispatch, and 请求生命周期 explains the complete loop.

Create an agent with its own default extensions.

Omit extensions to enable InMemoryMessageAccumulator and ToolGuidelinesExtension. An explicit sequence replaces those defaults. InternalMessageExtension and SteeringExtension are Agent-owned built-ins and remain enabled even when an empty extension sequence is supplied. Steering checks run after every completed model/tool operation, before internal input or remaining tool calls. It never interrupts in-flight work.

Extension names must be unique, including built-ins. max_iterations limits model calls for each UserMessage or AgentMessage; tool round trips consume that message's budget. max_internal_messages separately limits internal inputs per run/stream. Zero disables internal processing. Exceeding either limit raises AgentIterationLimitError; pending messages are discarded on failure.

参数:

model (AgentModel | None) -- Default provider adapter. It may be None when every request supplies an explicit model to run() or stream().
system_prompt (str) -- Initial instructions rebuilt for each request.
tools (Sequence[AgentTool]) -- Static tool definitions with unique names.
extensions (Sequence[AgentExtension] | None) -- Unique named extensions sorted by ascending priority. None installs defaults; an empty sequence omits optional defaults. Their inherited middleware hooks wrap model requests and local tool handlers in the same priority order.
reasoning_effort (ReasoningEffort) -- Default provider-neutral reasoning level.
parallel_tool_call (bool) -- Allow providers to return multiple tool calls and concurrently execute tools not explicitly marked serial.
max_iterations (int) -- Model-call budget per user/internal message.
max_internal_messages (int) -- Internal input budget per request; zero disables it.

备注

Construction alone does not initialize the runtime. Await initialize() or use the asynchronous create() factory before running.

示例

Usage:

agent = await Agent.create(
    model,
    config=config,
    tools=[read_file],
    reasoning_effort=ReasoningEffort.HIGH,
)

max_pause_continuations: int = 5#: How many times in a row a paused stop is continued without asking anyone. A provider returns paused when it hands control back mid-turn (Anthropic does so when a server-tool loop reaches its limit) and expects the turn to be sent again. At this many in a row the run ends and MODEL_STOPPED reports the pause. Zero never continues. Set it on an instance to change it.

model#: The default provider adapter for model calls. A request may bring its own; a request that has neither fails. None when the agent was built without one.

tools#: The tools registered on the agent, by name. Each request's context starts with a copy of these, and its extensions may add more to that copy.

extensions#: Every extension, built-ins included, in ascending priority. Hooks are called in this order.

system_prompt#: The system prompt used when a request does not bring its own.

reasoning_effort#: The default reasoning effort, used when a request does not choose one.

parallel_tool_call#: The default for whether a model may ask for several tools at once, and whether tools marked parallel run together.

state#: The most recently started request's state; use get_state to select by session.

max_iterations#: How many model calls one input may use before the request fails with an iteration-limit error. Steering input starts its own count.

max_internal_messages#: How many internal messages (instructions an extension queues for the model) one request may take up. Zero turns them off.

async initialize(*, config=None)[源代码]#

Bind the default session used when a request omits config.

示例

Usage:

agent = Agent(model)
await agent.initialize(config=AgentRunConfig(session_id="session-42"))
reply = await agent.run("Hello")

参数:: config (AgentRunConfig | None)
返回类型:: None

get_state(session_id)[源代码]#

Return one session's latest state, or None before its first request.

Concurrent callers must select by session rather than reading state, which is only the most recently started request's convenience view.

参数:: session_id (str)
返回类型:: AgentState | None

property pending: bool#

Return whether this Agent owns at least one unfinished request.

A request becomes pending when stream execution claims its session and remains pending through model calls, tools, external waits, and terminal hooks. Success, failure, cancellation, and explicit stream closure all clear it. Concurrent sessions keep the value true until the last one exits.

is_conflict(session_id)[源代码]#

Return whether another request currently owns session_id.

run() and stream() use the same check while atomically claiming the session, so concurrent callers cannot both start work for one conversation. Requests for different sessions remain independent.

参数:: session_id (str) -- Non-empty conversation identity to inspect.
抛出:: ValueError -- If session_id is empty.
返回类型:: bool

emit_external_event(event, *, config=None)[源代码]#

Broadcast config and event separately; return accepting extension names.

Without explicit config, use the sole active request, then the initialized session, or None before initialization. Extensions own routing and acceptance decisions. The original event is forwarded without copying, interpreting, or modifying its payload.

Every registered extension receives the event in priority order, including extensions after the first one that accepts it. This keeps the caller independent from the extension that owns a protocol.

示例

Usage:

accepted = agent.emit_external_event(
    ExternalEvent(
        name="ask_user_response",
        payload={"tool_call_id": "call-1", "answer": "Yes"},
    ),
    config=AgentRunConfig(session_id="session-42", request_id="request-1"),
)

参数:

event (ExternalEvent)
config (AgentRunConfig | None)

返回类型:

list[str]

async classmethod create(model, *, config=None, system_prompt='You are a helpful assistant.', tools=(), extensions=None, reasoning_effort=ReasoningEffort.MEDIUM, parallel_tool_call=True, max_iterations=36, max_internal_messages=8)[源代码]#

Construct and initialize an Agent before returning it.

参数:

model (AgentModel | None) -- Provider-neutral streaming model adapter.
config (AgentRunConfig | None) -- Initial session identity used when requests omit config. None generates a UUIDv7 session during initialization.
system_prompt (str) -- Initial instructions for every fresh request state.
tools (Sequence[AgentTool]) -- Static tools registered before extension setup.
extensions (Sequence[AgentExtension] | None) -- Optional lifecycle and middleware extensions; None keeps defaults.
reasoning_effort (ReasoningEffort) -- Default reasoning level for future requests.
parallel_tool_call (bool) -- Enable provider and local parallel tool calls.
max_iterations (int) -- Maximum model calls per user/internal input.
max_internal_messages (int) -- Maximum internal continuations per request.

返回:

An initialized instance of the class on which create() was called.

返回类型:

Self

示例

Usage:

agent = await Agent.create(model, config=AgentRunConfig(session_id="session-42"))
reply = await agent.run("Hello")

async run(message: UserMessage, *, config: AgentRunConfig | None = None, model: AgentModel | None = None, reasoning_effort: ReasoningEffort | None = None, parallel_tool_call: bool | None = None, metadata: Mapping[str, JsonValue] | None = None, tags: Mapping[str, JsonValue] | None = None) → AssistantMessage[源代码]#

async run(message: str, *, config: AgentRunConfig | None = None, model: AgentModel | None = None, reasoning_effort: ReasoningEffort | None = None, parallel_tool_call: bool | None = None, metadata: Mapping[str, JsonValue] | None = None, tags: Mapping[str, JsonValue] | None = None) → AssistantMessage

Collect one run and return its final answer.

Omit reasoning_effort to use the default configured on this Agent.

参数:

message (UserMessage | str) -- User text or a typed message containing text/image blocks.
config (AgentRunConfig | None) -- Request identity; omitted to reuse the initialized session.
model (AgentModel | None) -- Request-specific model; omitted to use the Agent default.
reasoning_effort (ReasoningEffort | None) -- Optional reasoning override for this request.
parallel_tool_call (bool | None) -- Optional parallel-call override for this request.
metadata (Mapping[str, JsonValue] | None) -- JSON data for extensions and Raw Log persistence, not a prompt.
tags (Mapping[str, JsonValue] | None) -- JSON classifications for extensions and Raw Log persistence.

返回:

The final assistant answer, after all stream cleanup completes.

抛出:

AgentProtocolError -- If uninitialized or no terminal answer is produced.
AgentIterationLimitError -- If the model/internal-message budget is exhausted.

返回类型:

AssistantMessage

示例

Usage:

agent = await Agent.create(model, config=AgentRunConfig(session_id="session-42"))
reply = await agent.run(
    "Summarize this conversation.",
    config=AgentRunConfig(session_id="session-42"),
    metadata={"source": "editor"},
    tags={"domain": "notes"},
)

async resume_point(*, config=None, model=None)[源代码]#

Report where a session was interrupted, without changing it.

A request can end without finishing: the connection drops, the user stops it, the process is killed. Nothing records that, so the session is restored exactly as a request restores it and its last message is read. Nothing is written and no model is called.

参数:

config (AgentRunConfig | None) -- Session identity; omitted to use the initialized session.
model (AgentModel | None) -- Only needed when the Agent has no default model.

返回:

Where it was interrupted and what resume() would do, or None when there is nothing to resume: the session is empty, its last request finished, or it ended in a refusal, a full context window, or a stop this library does not know. resume() raises for those.

返回类型:

ResumePoint | None

示例

Offer a resume only when there is one:

point = await agent.resume_point()
if point is not None:
    print(point.description)

async resume(*, config=None, model=None, reasoning_effort=None, parallel_tool_call=None, metadata=None, tags=None)[源代码]#

Pick an interrupted session up where it stopped and return the final answer.

This is run() without a new user message. The session is restored, its last message says how it was interrupted, and the request carries on from there; see ResumeKind for what is done in each case. Pass the same model, tools and settings the interrupted request had: none of them is stored.

参数:

config (AgentRunConfig | None) -- Session identity; omitted to use the initialized session.
model (AgentModel | None) -- Request-specific model; omitted to use the Agent default.
reasoning_effort (ReasoningEffort | None) -- Optional reasoning override for this request.
parallel_tool_call (bool | None) -- Optional parallel-call override for this request.
metadata (Mapping[str, JsonValue] | None) -- JSON data for extensions and Raw Log persistence, not a prompt.
tags (Mapping[str, JsonValue] | None) -- JSON classifications for extensions and Raw Log persistence.

返回:

The final assistant answer.

抛出:

AgentResumeError -- If there is nothing to resume. Ask resume_point() first to find out without an exception.
AgentIterationLimitError -- If the model/internal-message budget is exhausted.

返回类型:

AssistantMessage

示例

Carry on after a failure:

try:
    reply = await agent.run("Refactor the parser")
except ProviderTimeoutError:
    reply = await agent.resume()

async stream_resume(*, config=None, model=None, reasoning_effort=None, parallel_tool_call=None, metadata=None, tags=None)[源代码]#

Stream resume(); the events are those of stream().

参数:

config (AgentRunConfig | None) -- Session identity; omitted to use the initialized session.
model (AgentModel | None) -- Request-specific model; omitted to use the Agent default.
reasoning_effort (ReasoningEffort | None) -- Reasoning override limited to this request.
parallel_tool_call (bool | None) -- Parallel-call override limited to this request.
metadata (Mapping[str, JsonValue] | None) -- Request data visible to extensions and persistence.
tags (Mapping[str, JsonValue] | None) -- Request classifications visible to extensions and persistence.

生成器:

AgentEvent objects in execution order, ending in RUN_COMPLETED on success.

返回类型:

AsyncIterator[AgentEvent]

备注

The session is only known once it is restored, so a session with nothing to resume raises AgentResumeError as the request fails, and runs the error hooks like any failed request. Use resume_point() to ask first. The on_message hook is not called, because there is no new input for it to transform, and a goal that was armed for the interrupted request is not re-armed.

sync(*, runtime=None)#

Return a context-managed synchronous view; optionally share a runtime.

Omitting runtime creates a new background thread for this view. Passing an existing runtime reuses its loop and leaves its shutdown to the caller. This API is for service entry points, not Extension lifecycle hooks.

示例

Use a model or tool from a plain function:

with read_file.sync() as read:
    result = read({"path": "README.md"})

参数:: runtime (SyncRuntime | None)
返回类型:: SyncObject[Self]

async stream(message: UserMessage, *, config: AgentRunConfig | None = None, model: AgentModel | None = None, reasoning_effort: ReasoningEffort | None = None, parallel_tool_call: bool | None = None, metadata: Mapping[str, JsonValue] | None = None, tags: Mapping[str, JsonValue] | None = None) → AsyncIterator[AgentEvent][源代码]#

async stream(message: str, *, config: AgentRunConfig | None = None, model: AgentModel | None = None, reasoning_effort: ReasoningEffort | None = None, parallel_tool_call: bool | None = None, metadata: Mapping[str, JsonValue] | None = None, tags: Mapping[str, JsonValue] | None = None) → AsyncIterator[AgentEvent]

Stream one run while retaining messages in the agent state.

Extensions receive the same mutable state and may add, remove, or replace messages at each lifecycle hook. Omit reasoning_effort to use the Agent's configured default; an explicit value overrides it for this request only.

参数:

message (UserMessage | str) -- Plain user text or a typed multimodal UserMessage.
config (AgentRunConfig | None) -- Session/request identity; defaults to the initialized configuration.
model (AgentModel | None) -- Request-specific model; omitted to use the Agent default.
reasoning_effort (ReasoningEffort | None) -- Reasoning override limited to this request.
parallel_tool_call (bool | None) -- Parallel-call override limited to this request.
metadata (Mapping[str, JsonValue] | None) -- Request data visible to extensions and persistence.
tags (Mapping[str, JsonValue] | None) -- Request classifications visible to extensions and persistence.

生成器:

AgentEvent objects in execution order, ending in RUN_COMPLETED on success.

返回类型:

AsyncIterator[AgentEvent]

备注

Model errors and cancellation propagate as exceptions. Use contextlib.aclosing when stopping iteration early. Concurrent requests must use different session IDs; same-session overlap is rejected.

示例

Usage:

agent = await Agent.create(model, config=config, extensions=[history_extension])
async for event in agent.stream(
    "Continue the summary.",
    config=AgentRunConfig(session_id="session-42"),
    reasoning_effort=ReasoningEffort.HIGH,
    metadata={"source": "command-palette"},
    tags={"intent": "summary"},
):
    print(event.type, event.session_id)

async compact(*, config=None, model=None, metadata=None, tags=None)[源代码]#

Compact the session's history without a user message or a model turn.

The session is restored exactly as a request restores it — on_tool, then on_state — and every extension is asked to compact through its on_compact hook. Nothing else happens: no input message is appended and no primary model call is made, so a caller can summarize on demand instead of waiting for the context to grow past a budget.

参数:

config (AgentRunConfig | None) -- Session identity; defaults to the initialized configuration.
model (AgentModel | None) -- Model an extension compacts with when it has none of its own.
metadata (Mapping[str, JsonValue] | None) -- Request data visible to extensions and persistence.
tags (Mapping[str, JsonValue] | None) -- Request classifications visible to extensions and persistence.

生成器:

AgentEvent objects in execution order -- whatever the extensions emit while compacting, and nothing at all when they find nothing to compact.

返回类型:

AsyncIterator[AgentEvent]

备注

A pass that changes nothing ends without a terminal event, which is why AgentClient.compact() exists for callers that want one result instead of an event stream. Concurrent requests must use different session IDs; a maintenance pass claims its session like a request does.