第一个智能体#

你将学到：如何创建客户端、运行多个轮次，以及是什么让对话在轮次之间保持下来。前置条件：安装 Zett Agent。不需要 API 凭据。

本教程使用预设脚本的模型，因此你可以在任何机器上检查结果。它不是语言模型，也不会学习；它让你在选择模型提供商之前先体验运行时。如果你已经有凭据，可以直接看完整的工具调用请求。

1. 为循环提供一个模型#

智能体并不知道如何访问某个特定厂商的 HTTP API。它期望一个模型适配器：一个接受ModelRequest并流式返回ModelEvent对象的对象。

在第一个示例中，适配器会统计用户轮次并回显最近的一次。它不是 LLM，但它接入的循环与真实模型提供商完全相同。

class EchoModel:
    """Deterministic teaching adapter: count user turns and echo the latest one."""

    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        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)))

每个模型步骤都以一个完整的ModelResponse结束。它之前的文本增量只用于显示；运行时真正使用的是最终响应。

2. 创建一个客户端，运行多个轮次#

async def main() -> None:
    client = await create_agent(EchoModel())
    first = await client.run("Hello")
    second = await client.run("Remember this conversation")
    assert first.content == "Turn 1: Hello"
    assert second.content == "Turn 2: Remember this conversation"
    print(first.content)
    print(second.content)

create_agent()会构建运行时并返回一个AgentClient。由于没有提供会话，它会为你生成一个。

第二个请求能记住第一个请求，因为默认扩展会为会话保留内存中的历史记录。run()会返回完整的AssistantMessage。

3. 运行整个程序#

下载 first_agent.py。

一个完整的离线程序#
 1"""A complete offline conversation through the real Agent runtime."""
 2
 3import asyncio
 4from collections.abc import AsyncIterator
 5
 6from zett_agent.client import create_agent
 7from zett_agent.messages import (
 8    AssistantMessage,
 9    UserMessage,
10)
11from zett_agent.model import (
12    ModelEvent,
13    ModelRequest,
14    ModelResponse,
15    RetryOptions,
16)
17
18
19class EchoModel:
20    """Deterministic teaching adapter: count user turns and echo the latest one."""
21
22    retry = RetryOptions(max_retries=0)
23
24    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
25        users = [message for message in request.messages if isinstance(message, UserMessage)]
26        answer = f"Turn {len(users)}: {users[-1].text}"
27        yield ModelEvent.text(answer)
28        yield ModelEvent.completed(ModelResponse(AssistantMessage(content=answer)))
29
30
31async def main() -> None:
32    client = await create_agent(EchoModel())
33    first = await client.run("Hello")
34    second = await client.run("Remember this conversation")
35    assert first.content == "Turn 1: Hello"
36    assert second.content == "Turn 2: Remember this conversation"
37    print(first.content)
38    print(second.content)
39
40
41if __name__ == "__main__":
42    asyncio.run(main())

把下载的文件在已安装 Zett Agent 的环境中保存为first_agent.py后，运行如下:

python first_agent.py

或者在源码检出目录中运行:

uv run python docs/_examples/first_agent.py

预期输出:

Turn 1: Hello
Turn 2: Remember this conversation

发生了什么？#

create_agent用默认历史记录初始化了一个对话。每次run都会追加一条用户消息、调用适配器，并返回最终助手消息。第二次调用时，适配器统计到两条用户消息，因为上一轮已被恢复。你不需要手动传入历史记录，也不需要自己编写对话循环。

真实的模型提供商可能会在回答之前请求工具。运行时会执行这些工具，并把结果再次传给模型；因此一次run就可能涉及多个模型步骤。在发出付费请求之前，可以先试试确定性的工具往返，了解这种行为。

下一步：把预设脚本的模型换成真实模型提供商#

只有适配器的构造方式会变。同一个客户端、消息、扩展、工具和事件回调，在所有内置提供商上都能工作。接下来请阅读连接模型提供商，然后是把流式输出接入终端或界面。

警告

传入extensions=[]会移除这些可选的默认扩展，包括内存中的历史记录。每个请求都会从全新的上下文开始；如果希望对话延续下去，请添加InMemoryMessageAccumulator或一个持久化扩展。

相关 API：Agent、AgentRunConfig、InMemoryMessageAccumulator。