在同步代码中使用

并非每个程序都是异步的。create_agent_sync在同一个后台事件循环上运行相同的运行时，因此脚本、CLI 和 notebook 都可以从普通的def代码中调用它。模型行为、工具、扩展、会话历史、重试和事件类型都与异步 API 完全一致。

运行与流式输出

from zett_agent.sync import create_agent_sync

with create_agent_sync(model) as agent:
    reply = agent.run("Hello")
    print(reply.content)

    with agent.stream("Continue") as events:
        for event in events:
            print(event.type, event.delta)

如果你可能会提前停止，请把流当作上下文管理器使用；关闭它会取消未完成的请求并等待清理完成。同一会话上的后续请求可以正常运行。

异步对象的阻塞视图

智能体、客户端、模型提供商、工具、存储和分发器都提供.sync()，它用阻塞调用映射它们各自的异步方法：

with existing_agent.sync() as agent:
    agent.initialize(config=config)
    reply = agent.run("Hello")

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

with storage.sync() as db:
    session = db.load("session-42")

对于其他情况——自由函数、第三方客户端——请直接使用SyncRuntime：

with SyncRuntime() as runtime:
    agent = runtime.call(Agent.create, model, config=config)
    result = runtime.call(agent.run, "Hello")
    with runtime.stream(agent.stream, "Continue") as events:
        for event in events:
            print(event)

对于应该配套使用的资源，请共享同一个运行时，并在离开之前关闭你自己拥有的部分。你显式传入的运行时在视图退出时仍保持打开。

回调与自定义模型

扩展钩子和分发器回调仍然是async def；它们在后台循环上运行，并带有真实的请求上下文。GUI 应用应该通过自己的 UI 线程安排界面更新，而不是在回调里直接操作控件。

如果你的模型是一个阻塞式的def stream(request)生成器，请用SyncModelAdapter包装它。现有的@tool函数已经同时支持这两种风格。

向运行中的请求发送输入

当同步流暂停在Ask User或 Plan Mode 事件上时，另一个线程可以用agent.emit_external_event(event, config=config)传递回复。不同会话可以并发运行；同一会话的重叠请求会被拒绝。

完整的离线示例

"""Run, stream, and persist a conversation without an asynchronous entry point."""

from pathlib import Path
from tempfile import TemporaryDirectory

from zett_agent.agent import AgentRunConfig
from zett_agent.dispatcher import AgentEventDispatcher
from zett_agent.events import AgentEventType
from zett_agent.extensions.sqlite import SQLiteSessionExtension
from zett_agent.messages import (
    AssistantMessage,
    UserMessage,
)
from zett_agent.model import (
    ModelEvent,
    ModelResponse,
)
from zett_agent.sync import (
    SyncModelAdapter,
    create_agent_sync,
)


class Echo:
    """A synchronous model used only to make this example runnable offline."""

    def stream(self, request):
        users = [message for message in request.messages if isinstance(message, UserMessage)]
        answer = f"Turn {len(users)}: {users[-1].text}"
        yield ModelEvent.text(answer)
        yield ModelEvent.completed(ModelResponse(AssistantMessage(content=answer)))


class Console(AgentEventDispatcher):
    async def on_text_delta_event(self, event):
        print(event.delta)


def main() -> None:
    with TemporaryDirectory() as directory:
        history = SQLiteSessionExtension(Path(directory) / "sessions.sqlite3")
        try:
            with create_agent_sync(
                SyncModelAdapter(Echo()),
                config=AgentRunConfig("sync-example"),
                extensions=[history],
                event_dispatcher=Console(),
            ) as agent:
                assert agent.run("Hello").content == "Turn 1: Hello"
                with agent.stream("Remember this") as events:
                    for event in events:
                        if event.type is AgentEventType.RUN_COMPLETED:
                            assert event.message is not None
                            assert event.message.content == "Turn 2: Remember this"
            # Storage is asynchronous; a plain function reaches it through the sync view.
            with history.storage.sync() as database:
                records = database.list_raw_messages("sync-example")
                assert [record.message.role for record in records] == [
                    "system",
                    "user",
                    "assistant",
                    "system",
                    "user",
                    "assistant",
                ]
            print("Synchronous history persisted")
        finally:
            with history.storage.sync() as database:
                database.close()


if __name__ == "__main__":
    main()

下载示例。

参见

把流式输出接入终端或界面介绍事件处理，保存并恢复对话介绍持久化，生命周期钩子参考列出全部钩子。