观察用量与耗时

记录 token 用量、提示词缓存命中和推理/内容的耗时，同时不必改变应用运行智能体的方式。模型给出的计数是固定值；该示例并不度量模型提供商的价格。

运行

在仓库根目录下执行：

uv run python docs/_examples/observability.py

输出包含：

Reasoning and content boundaries recorded

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

记录完成后的消息

如果你想同时拿到最终确定的消息、用量和耗时，而不是一个个显示增量，就使用扩展：

class UsageObserver(AgentExtension):
    """Retain immutable observations for a demo; use a bounded sink in production."""

    def __init__(self) -> None:
        self.records: list[tuple[str, MessageAppendedEvent]] = []

    async def on_event(self, context: AgentRunContext, event: ExtensionEvent) -> None:
        if isinstance(event, MessageAppendedEvent) and event.usage is not None:
            self.records.append((context.session_id, event))

该示例把记录保存在列表中，并检查 60% 的缓存命中率。对于长期运行的服务，应把记录发送到有界队列或遥测接收端，而不是在内存中保留一个无界列表。上报的缓存用量取决于模型提供商；缺少用量并不等于实测的零。

完整源码

下载 observability.py。

"""Observe persisted-message timing and provider usage without changing the loop."""

import asyncio
from collections.abc import AsyncIterator

from zett_agent.agent import AgentRunContext
from zett_agent.client import create_agent
from zett_agent.extensions.base import AgentExtension
from zett_agent.extensions.events import (
    ExtensionEvent,
    MessageAppendedEvent,
)
from zett_agent.messages import AssistantMessage
from zett_agent.model import (
    ModelEvent,
    ModelRequest,
    ModelResponse,
    ModelUsage,
    RetryOptions,
)


class UsageObserver(AgentExtension):
    """Retain immutable observations for a demo; use a bounded sink in production."""

    def __init__(self) -> None:
        self.records: list[tuple[str, MessageAppendedEvent]] = []

    async def on_event(self, context: AgentRunContext, event: ExtensionEvent) -> None:
        if isinstance(event, MessageAppendedEvent) and event.usage is not None:
            self.records.append((context.session_id, event))


class UsageModel:
    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        yield ModelEvent.reasoning("Inspect the request.")
        await asyncio.sleep(0)
        yield ModelEvent.text("Done")
        yield ModelEvent.completed(
            ModelResponse(
                AssistantMessage(content="Done", reasoning="Inspect the request."),
                usage=ModelUsage(input_tokens=100, output_tokens=20, cache_read_tokens=60, reasoning_tokens=5),
            )
        )


async def main() -> None:
    observer = UsageObserver()
    client = await create_agent(UsageModel(), extensions=[observer])
    await client.run("Inspect usage")
    _, record = observer.records[0]
    assert record.usage is not None
    assert record.timing.duration_ns >= 0
    assert record.timing.reasoning_duration_ns is not None
    assert record.timing.content_duration_ns is not None
    assert record.usage.total_tokens == 120
    assert record.usage.cache_hit_rate == 0.6
    print("Tokens: 120; cache hit: 60%")
    print("Reasoning and content boundaries recorded")


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

下一步：提示词缓存见连接模型提供商，在显示事件和生命周期事件之间做选择见三类事件。