GoalExtension#

Build extensions, choose hooks, and manage request-scoped state. See the guide.

class zett_agent.extensions.goal.GoalExtension(definition=None, *, max_iterations=36, max_decision_retries=3)[源代码]#

Continue externally selected Goal Mode requests until verification passes.

This extension does not implement another agent loop. It creates the normal Agent using a SubAgentDefinition, exactly like delegated subagents. The default definition installs CodingExtension and therefore provides read, write, replace, glob, grep, and shell tools. Its model, extensions, reasoning effort, and model-iteration limit all come from the definition and can be replaced as one unit.

An application first sends a goal_mode ExternalEvent with the AgentRunConfig of a future request. The extension arms that exact session/request identity. When the matching request reaches on_message(), its complete user input becomes the goal and is replaced with an explicit Goal Mode prompt before it reaches the model or Raw Log. Selection is consumed once; unrelated sessions and request IDs remain in normal mode.

The transformed UserMessage keeps a deep copy of its original content under attributes["goal_extension"]["raw_content"]. Evaluator AgentEvents remain private. An incomplete decision becomes an AgentMessage published through InternalMessageEvent; a successful decision injects nothing and lets the parent request complete normally.

max_iterations is the independent limit for goal-driven continuations; it defaults to eight. The parent Agent's max_internal_messages remains a global limit across all extensions and may stop processing earlier. When an evaluator returns normally without calling the decision tool, max_decision_retries continues the same evaluator with its history. Its default of three means one initial attempt plus three retries.

示例

Usage:

definition = SubAgentDefinition(
    name="goal",
    description="verify coding work",
    system_prompt="Inspect the result, then call report_goal_evaluation.",
    model=reviewer_model,
    extensions=(CodingExtension(), ToolGuidelinesExtension()),
    reasoning_effort=ReasoningEffort.HIGH,
    max_iterations=12,
)
agent = await Agent.create(
    primary_model,
    config=AgentRunConfig(session_id="session-42"),
    extensions=[GoalExtension(definition, max_iterations=8, max_decision_retries=3)],
    max_internal_messages=8,
)
config = AgentRunConfig(session_id="session-42", request_id="request-7")
agent.emit_external_event(ExternalEvent(GOAL_MODE_EVENT_NAME, {}), config=config)
result = await agent.run("Implement the feature and verify its tests.", config=config)
参数:
accept(config, event)[源代码]#

Arm Goal Mode once for the next request matching config.

The event payload is intentionally application-owned and ignored. Mode routing comes only from the separately supplied AgentRunConfig, avoiding duplicate or conflicting session identifiers inside the payload. Repeating the same event before consumption is idempotent.

参数:
返回类型:

bool

async on_message(context)[源代码]#

Consume a matching external selection and inject the Goal Mode prompt.

参数:

context (AgentRunContext)

返回类型:

None

async on_event(context, event)[源代码]#

Release request-local state when cancellation bypasses run callbacks.

参数:
返回类型:

None

async after_model(context, response)[源代码]#

Privately evaluate a candidate answer and enqueue one continuation.

A step that was refused or interrupted has no answer to judge, and sending the model back to try again would only meet the same refusal, so it is not evaluated: the run ends and the host is told why through MODEL_STOPPED.

参数:
返回类型:

None

async on_success(context, result)[源代码]#

Release request-local goal state after verified success.

参数:
返回类型:

None

async on_error(context, error)[源代码]#

Release request-local goal state after any failed request.

参数:
返回类型:

None

async after_run(context, result)#

Run after a successful final answer is appended.

参数:
返回类型:

None

async after_tool(context, call, result, error)#

Inspect or modify a tool result before it is appended.

error is the original execution exception for a failed result and None after successful execution. Changes made to result are included in context, persistence, and the next model request. Raising prevents the result from being appended.

参数:
返回类型:

None

async after_turn(context, result)#

Run once the turn's assistant message and tool results are settled.

result is the assistant message that opened the turn: either the final answer, or the tool-calling message whose ToolMessages were appended. Every requested call is settled first, including calls that failed, and calls that steering skipped, which arrive as unsuccessful or skipped ToolMessages.

Steering that interrupts a turn is accepted while the runtime settles that batch, so its UserMessage is already appended when this hook runs and the steering turn follows as the next turn.

A turn that never completed, because the request failed or was cancelled, has no after_turn call; use on_error and RunCancelledEvent for those paths.

参数:
返回类型:

None

async before_model(context, request)#

Inspect the current request before preprocessing a primary model call.

Request fields are frozen. Change messages through AgentRunContext.add_message() or AgentRunContext.replace_messages(), local tools through context.tools, and provider tools through context.server_tools; the runtime rebuilds these fields before the next hook and before calling the provider. This is a shallow request view, not an immutable copy of each message or tool definition.

示例

Inspect the tools offered in this iteration:

async def before_model(self, context, request):
    offered = {definition.name for definition in request.tools}
    if "write_file" in offered:
        context.add_message(
            SystemMessage(content="Read existing files before editing."),
        )
参数:
返回类型:

None

async before_run(context)#

Run after the transformed user input is appended and published.

参数:

context (AgentRunContext)

返回类型:

None

async before_tool(context, call)#

Run before one requested tool is invoked.

参数:
返回类型:

None

async before_turn(context)#

Run before this turn's model call.

Runs after before_run() on the first turn and after the previous turn's after_turn() on later ones. Model request assembly, compaction, and before_model() follow this hook, so context edited here reaches the provider.

参数:

context (AgentRunContext)

返回类型:

None

property name: str#

Unique registration identity, defaulting to the concrete class name.

Override with a class attribute or assign an instance name to register independently configured instances of the same extension type.

async on_compact(context)#

Summarize restored context when a caller asks for compaction.

Agent.compact restores the session through on_tool and on_state exactly as a request does, appends no input message, and makes no primary model call: the only work a compaction pass does is what this hook performs. The default does nothing, so an extension with nothing to compact pays nothing.

示例

Usage:

async def on_compact(self, context):
    await self.compact(context)
参数:

context (AgentRunContext)

返回类型:

None

async on_model_request(context, request, call_next)#

Wrap one provider request and its complete streamed response.

Pass request (or a dataclass replacement) to call_next. The default closes the inner stream so cancellation reaches the provider.

参数:
返回类型:

AsyncIterator[ModelEvent]

async on_state(context)#

Restore history and initialize state before any input conversion.

示例

Usage:

async def on_state(self, context):
    context.add_message(SystemMessage(content="Use concise answers."), index=0)
参数:

context (AgentRunContext)

返回类型:

None

async on_tool(context)#

Register local or provider-hosted tools before context restoration.

All extensions finish this hook before the first on_state() call, so prompt extensions can reliably inspect context.tools and context.server_tools as complete request-scoped registries.

示例

Usage:

async def on_tool(self, context):
    context.register_tool(read_file)
    context.register_server_tool(ServerToolDefinition(type="web_search"))
参数:

context (AgentRunContext)

返回类型:

None

async on_tool_call(context, call, call_next)#

Wrap one registered local tool handler and return its ToolResult.

An extension may adjust call.arguments, return without calling the next layer, transform its result, or raise into normal tool-failure handling.

参数:
返回类型:

ToolResult

property pending: bool#

Return whether this extension is participating in an active request.

The Agent updates this state for every registered extension, including extensions whose hooks are currently idle. One extension instance may be shared by multiple Agents and sessions; pending remains true until all requests using that instance have finished, failed, or been cancelled. Persistent data retained between requests does not count.