运行一个隔离的子智能体

运行子智能体时使用只读工具，并显式共享临时存储。工具限制并不等同于操作系统沙箱。

运行

在仓库根目录下执行：

uv run python docs/_examples/subagent.py

输出包含：

Two isolated sessions; child links to parent

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

显式选择子智能体的能力

给子智能体配置自己的指令、模型和工具。父智能体可以把工作委派给该配置，而不必把父智能体拥有的全部能力都交给它：

async def main() -> None:
    with TemporaryDirectory(prefix="zett-subagent-") as directory:
        persistence = SQLiteSessionExtension(Path(directory) / "sessions.sqlite")
        try:
            definition = SubAgentDefinition(
                name="review",
                description="review a design with read-only tools",
                system_prompt="Review the supplied design and return a concise report.",
                model=ChildModel(),
                extensions=(persistence, FileSystemExtension(read_only=True), ToolGuidelinesExtension()),
            )
            client = await create_agent(
                ParentModel(),
                config=AgentRunConfig(session_id="parent"),
                extensions=[persistence, SubAgentExtension([definition]), ToolGuidelinesExtension()],
            )
            print((await client.run("Review the design")).content)
            sessions = await persistence.list_sessions()
            children = [session for session in sessions if session.parent_session_id == "parent"]
            assert len(sessions) == 2 and len(children) == 1
            assert children[0].session_id != "parent"
            print("Two isolated sessions; child links to parent")
        finally:
            await persistence.close()

该示例把父智能体和子智能体作为两个独立会话存入同一个临时数据库。父会话链接可以让应用展示它们之间的关系。只读文件系统工具限制的是对外暴露的操作；它们并不会隔离进程，也无法阻止任意代码写入。

完整源码

下载 subagent.py。

"""Delegate a task to a child profile and inspect its durable parent linkage."""

import asyncio
import json
from collections.abc import AsyncIterator
from pathlib import Path
from tempfile import TemporaryDirectory

from zett_agent.agent import AgentRunConfig
from zett_agent.client import create_agent
from zett_agent.extensions.file_system import FileSystemExtension
from zett_agent.extensions.sqlite import SQLiteSessionExtension
from zett_agent.extensions.subagent import (
    SubAgentDefinition,
    SubAgentExtension,
)
from zett_agent.extensions.tool_guidelines import ToolGuidelinesExtension
from zett_agent.messages import (
    AssistantMessage,
    ToolCall,
    ToolMessage,
)
from zett_agent.model import (
    ModelEvent,
    ModelRequest,
    ModelResponse,
    RetryOptions,
)


class ParentModel:
    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        last = request.messages[-1]
        if isinstance(last, ToolMessage):
            report = json.loads(last.text)
            message = AssistantMessage(content=f"Child report: {report['content']}")
        else:
            message = AssistantMessage(
                tool_calls=(
                    ToolCall(
                        "task-1",
                        "task",
                        {"description": "Review design", "prompt": "Review the design", "subagent_type": "review"},
                    ),
                )
            )
        yield ModelEvent.completed(ModelResponse(message))


class ChildModel:
    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        assert {tool.name for tool in request.tools} == {"read_file", "view_image", "glob", "grep"}
        yield ModelEvent.completed(ModelResponse(AssistantMessage(content="Keep the runtime small.")))


async def main() -> None:
    with TemporaryDirectory(prefix="zett-subagent-") as directory:
        persistence = SQLiteSessionExtension(Path(directory) / "sessions.sqlite")
        try:
            definition = SubAgentDefinition(
                name="review",
                description="review a design with read-only tools",
                system_prompt="Review the supplied design and return a concise report.",
                model=ChildModel(),
                extensions=(persistence, FileSystemExtension(read_only=True), ToolGuidelinesExtension()),
            )
            client = await create_agent(
                ParentModel(),
                config=AgentRunConfig(session_id="parent"),
                extensions=[persistence, SubAgentExtension([definition]), ToolGuidelinesExtension()],
            )
            print((await client.run("Review the design")).content)
            sessions = await persistence.list_sessions()
            children = [session for session in sessions if session.parent_session_id == "parent"]
            assert len(sessions) == 2 and len(children) == 1
            assert children[0].session_id != "parent"
            print("Two isolated sessions; child links to parent")
        finally:
            await persistence.close()


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

下一步：子智能体配置有哪些选项，请看组合内置扩展；由评估驱动的继续执行，请看迭代逼近目标。