迭代逼近目标

goal_mode 外部事件会选定接下来的某一次请求。随后一个私有评估器先报告未完成，最终报告已达成。这里演示的是控制流，而不是真实模型的评估质量。

运行

在仓库根目录下执行：

uv run python docs/_examples/goal.py

输出包含：

Implementation and tests are complete.

这个程序在需要时使用确定性的离线模型和临时存储。它不需要 API 密钥，并由文档测试套件执行。

把执行者与审查者分开

应用为某个即将发出的请求选择目标模式。审查者检查执行者的结果，并可以要求再试一次：

async def main() -> None:
    reviewer = ReviewModel()
    definition = SubAgentDefinition(
        name="review",
        description="verify goal completion",
        system_prompt="Inspect evidence and call report_goal_evaluation.",
        model=reviewer,
    )
    config = AgentRunConfig(session_id="goal-example", request_id="request-1")
    client = await create_agent(
        DraftModel(),
        config=config,
        extensions=[GoalExtension(definition, max_iterations=2)],
        event_dispatcher=GoalEvents(),
    )
    accepted = client.agent.emit_external_event(ExternalEvent(GOAL_MODE_EVENT_NAME, {}), config=config)
    assert accepted == ["GoalExtension"]
    result = await client.run("Implement the change with tests", config=config)
    assert reviewer.reviews == 2
    print(result.content)

这里审查者故意只报告一次缺少测试，然后接受第二次作答。在真实的工作流中，应让它检查具体的证据，而不是接受执行者的说法。请用迭代次数上限来限制继续执行，并为用户提供取消的途径；评估器并不能保证结果正确。

完整源码

下载 goal.py。

"""Run a bounded goal review using the normal Agent as a private evaluator."""

import asyncio
from collections.abc import AsyncIterator

from zett_agent.agent import AgentRunConfig
from zett_agent.client import create_agent
from zett_agent.dispatcher import AgentEventDispatcher
from zett_agent.events import AgentEvent
from zett_agent.extensions.external import ExternalEvent
from zett_agent.extensions.goal import (
    GOAL_MODE_EVENT_NAME,
    GoalExtension,
)
from zett_agent.extensions.subagent import SubAgentDefinition
from zett_agent.messages import (
    AssistantMessage,
    ToolCall,
    ToolMessage,
)
from zett_agent.model import (
    ModelEvent,
    ModelRequest,
    ModelResponse,
    RetryOptions,
)


class DraftModel:
    retry = RetryOptions(max_retries=0)

    def __init__(self) -> None:
        self.attempt = 0

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        self.attempt += 1
        answer = "A draft without tests." if self.attempt == 1 else "Implementation and tests are complete."
        yield ModelEvent.text(answer)
        yield ModelEvent.completed(ModelResponse(AssistantMessage(content=answer)))


class ReviewModel:
    """Script review decisions; real evaluators must inspect actual evidence."""

    retry = RetryOptions(max_retries=0)

    def __init__(self) -> None:
        self.reviews = 0

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        if isinstance(request.messages[-1], ToolMessage):
            message = AssistantMessage(content="Decision submitted.")
        else:
            self.reviews += 1
            achieved = self.reviews > 1
            message = AssistantMessage(
                tool_calls=(
                    ToolCall(
                        "review-1",
                        "report_goal_evaluation",
                        {
                            "achieved": achieved,
                            "summary": "Verified" if achieved else "Tests missing",
                            "remaining_work": [] if achieved else ["Add tests"],
                            "next_instruction": None if achieved else "Add and run the tests.",
                        },
                    ),
                )
            )
        yield ModelEvent.completed(ModelResponse(message))


class GoalEvents(AgentEventDispatcher):
    async def on_internal_message_started_event(self, event: AgentEvent) -> None:
        print("Continuing after private review")


async def main() -> None:
    reviewer = ReviewModel()
    definition = SubAgentDefinition(
        name="review",
        description="verify goal completion",
        system_prompt="Inspect evidence and call report_goal_evaluation.",
        model=reviewer,
    )
    config = AgentRunConfig(session_id="goal-example", request_id="request-1")
    client = await create_agent(
        DraftModel(),
        config=config,
        extensions=[GoalExtension(definition, max_iterations=2)],
        event_dispatcher=GoalEvents(),
    )
    accepted = client.agent.emit_external_event(ExternalEvent(GOAL_MODE_EVENT_NAME, {}), config=config)
    assert accepted == ["GoalExtension"]
    result = await client.run("Implement the change with tests", config=config)
    assert reviewer.reviews == 2
    print(result.content)


if __name__ == "__main__":
    asyncio.run(main())

下一步：普通的工作委派见运行一个隔离的子智能体，目标模式的选择和限制见GoalExtension。