agno 如何读取 Agent 运行轨迹的 trace 并聚合按天的 metrics 指标【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno在 agno 的 AgentOS 上运行 Agent 后你会经常遇到两类运维需求一是回放某次运行到底经历了哪些步骤Agent 执行、模型调用、工具调用各自耗时多少二是把一段时间内的活动按天聚合得到运行次数、会话数、用户数和 token 用量。AgentOS 对这两类数据提供了成对的接口trace 通过 OpenTelemetry span 产生用tracingTrue开启metrics 从持久化会话数据聚合用POST /metrics/refresh刷新、GET /metrics读取。本文按 cookbook/05_agent_os/13_observability/README.md 的示例走通「生成数据 → 通过监控客户端相同的路线读回」这条路径。两类数据的产生方式不同先分清两者后面的接口用法才不会混trace把 Agent 运行、模型调用、工具调用变成 span。开关是AgentOS(tracingTrue)同一开关覆盖注册到该 OS 的 agents、teams 和 workflows。trace 与 span 写入你为 OS 配置的 db示例中是SqliteDb。metrics按天聚合持久化下来的活动。metrics.py 的示例没有设置tracingTrue它只依赖SqliteDb持久化会话再调用 refresh 接口重算聚合。也就是说读 trace 需要开 tracing读 metrics 需要会话已持久化两者互相独立。准备环境真实运行 Agent 需要设置OPENAI_API_KEY。按 README 的要求demo 环境需要 Agno tracing 所用的 OpenTelemetry 包命令来自 README其中--python指定 demo 虚拟环境的解释器换成你自己的 venv 路径即可uv pip install --python .venvs/demo/bin/python \ opentelemetry-api opentelemetry-sdk openinference-instrumentation-agno示例均从仓库根目录运行例如.venvs/demo/bin/python cookbook/05_agent_os/13_observability/read_traces.py。开启 tracing基本配置如下关键点只有一个AgentOS(...)传tracingTruefrom agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.os import AgentOS db SqliteDb( idobservability-basic-db, db_filetmp/observability_basic.db, ) traced_agent Agent( idtraced-assistant, nameTraced Assistant, modelOpenAIResponses(idgpt-5.5), instructionsAnswer operational questions clearly and concisely., ) agent_os AgentOS( idobservability-basic-os, descriptionAgentOS with OpenTelemetry tracing stored in SQLite., dbdb, agents[traced_agent], tracingTrue, ) app agent_os.get_app() if __name__ __main__: agent_os.serve(appapp)运行.venvs/demo/bin/python cookbook/05_agent_os/13_observability/basic.py并调用其服务的 agent 后可以打开http://localhost:7777/traces和http://localhost:7777/traces/filter-schema查看已存储的 trace。读取 trace 列表与 span 树read_traces.py 演示了完整的读回路径先产生一次同步运行run和一次异步运行arun再通过 AgentOS 的 HTTP 接口读回两条 trace。它用httpx.ASGITransport直连get_app()返回的 app进程内就能跑通不需要另起服务sync_run trace_agent.run( Reply with a short greeting for the synchronous trace., session_idtrace-readback-sync, user_idobservability-user, ) async_run await trace_agent.arun( Reply with a short greeting for the asynchronous trace., session_idtrace-readback-async, user_idobservability-user, ) transport httpx.ASGITransport(appapp) async with httpx.AsyncClient( transporttransport, base_urlhttp://agent-os, ) as client: response await client.get( /traces, params{agent_id: AGENT_ID, db_id: DB_ID, limit: 20}, ) response.raise_for_status() trace_page response.json() traces_by_run { trace[run_id]: trace for trace in trace_page[data] if trace[run_id] } run_ids [sync_run.run_id, async_run.run_id] missing_run_ids [run_id for run_id in run_ids if run_id not in traces_by_run] if missing_run_ids: raise RuntimeError(fGET /traces omitted run IDs: {missing_run_ids}) for label, run_id in ((sync, sync_run.run_id), (async, async_run.run_id)): trace_id traces_by_run[run_id][trace_id] detail_response await client.get( f/traces/{trace_id}, params{db_id: DB_ID}, ) detail_response.raise_for_status() detail detail_response.json() if not detail[tree]: raise RuntimeError(fTrace {trace_id} has no span tree) print(f\n{label} run {run_id}: trace{trace_id}, spans{detail[total_spans]}) print_span_tree(detail[tree])两个接口的要点实现见 traces 路由GET /traces返回分页结构data中每条 trace 含trace_id、run_id、status、duration、total_spans等字段支持run_id、session_id、user_id、agent_id、team_id、workflow_id、statusOK/ERROR、start_time/end_timeISO 8601 带时区等过滤参数以及page1-indexed、limit默认 20、db_id。GET /traces/{trace_id}返回该 trace 的嵌套 span 树tree与total_spans。脚本按run_id把列表中的记录与本次运行对应起来再取trace_id拉详情最终递归打印name [type] status (duration)格式的层级树。脚本自身的校验就是验证方式如果两次运行对应的run_id没有出现在GET /traces结果里或详情里tree为空都会直接抛RuntimeError。正常输出会先打印GET /traces returned N trace(s)N 为meta.total_count随后是每条运行的 span 树。可选用 FilterExpr 做结构化搜索如果需要比查询参数更强的条件组合逻辑、任意字段filtering.py 展示的是FilterExprDSL它有两条等价路径filter_expr AND( EQ(status, OK), OR( EQ(agent_id, NEWS_AGENT_ID), EQ(agent_id, RELEASE_AGENT_ID), ), ) filter_dict filter_expr.to_dict() # 路径一直接在 Python 侧查 tracing 数据库 python_traces, python_count db.get_traces(filter_exprfilter_dict) # 路径二把同一表达式发给 AgentOS 的搜索接口 search_response await client.post( /traces/search, params{db_id: DB_ID}, json{filter: filter_dict, group_by: run, page: 1, limit: 20}, )配套的GET /traces/filter-schema返回可过滤字段fields[].key、操作符和枚举值、以及逻辑运算符客户端可以据此动态构建过滤 UI。该示例的验证点是schema 的字段集合必须包含status和agent_id逻辑运算符为[AND, OR]且搜索结果中同时出现两个 agent 的agent_id否则抛RuntimeError。刷新并读取按天的 metricsmetrics.py 是「聚合按天指标」的完整路径注意它的AgentOS没有tracingTrue只配置了SqliteDb先让 Agent 产生一次持久化运行arun带session_id和user_id这是 metrics 的原始数据来源。调用刷新接口重算聚合refresh_response await client.post( /metrics/refresh, params{db_id: DB_ID}, ) refresh_response.raise_for_status() refreshed_metrics refresh_response.json() if not refreshed_metrics: raise RuntimeError(POST /metrics/refresh returned no daily aggregate)再读回已存储的聚合metrics_response await client.get( /metrics, params{db_id: DB_ID}, ) metrics_response.raise_for_status() metrics_page metrics_response.json() if not metrics_page[metrics]: raise RuntimeError(GET /metrics returned no stored aggregate) latest max(metrics_page[metrics], keylambda metric: metric[date]) if latest[agent_runs_count] 1: raise RuntimeError(The refreshed aggregate omitted the new agent run) print(fDate: {latest[date]}) print(fAgent runs: {latest[agent_runs_count]}) print(fAgent sessions: {latest[agent_sessions_count]}) print(fUsers: {latest[users_count]}) print(fTotal tokens: {latest[token_metrics].get(total_tokens, 0)})接口语义实现见 metrics 路由POST /metrics/refresh默认同步执行并直接返回刷新后的每日聚合列表传backgroundtrue则立即返回 202随后用GET /metrics/refresh/status?db_id...轮询running/completed/failed状态从未触发时为idle同一数据库已有刷新在进行时会返回status: already_running而不会重复启动。GET /metrics返回metrics列表每个元素是一天的聚合含date、agent_runs_count、agent_sessions_count、team_runs_count、workflow_runs_count、users_count、token_metricsinput_tokens、output_tokens、total_tokens、audio_tokens、cached_tokens、reasoning_tokens等和model_metrics可用starting_date/ending_dateYYYY-MM-DD限定日期范围也可用user_id只取某用户的指标对非 admin 调用方该参数会被忽略。接口文档给出的示例响应中一条日聚合形如agent_runs_count: 5、total_tokens: 596——这是文档示例数值实际读数随你的运行数据变化。脚本最后打印的Date:、Agent runs:等行就是验证输出只要最新一天的agent_runs_count至少为 1说明新产生的一次运行已经被计入了按天聚合。边界与限制OS 注册了多个数据库时例如 traces_to_clickhouse.py 那种「会话存 SQLite、trace 批量写 ClickHouse」的拆分存储对GET /traces的请求在客户端传db_id之前是歧义的必须带上对应的db_id才能选中目标库。该变体还需要额外安装clickhouse-connect并用./cookbook/scripts/run_clickhouse.sh启动本地 ClickHouse默认端口8123用户/密码均为ai可用CLICKHOUSE_HOST等环境变量覆盖。metrics 是从持久化会话聚合的没有配置 db、运行未持久化的 OSrefresh 不会有数据可算。GET /traces的时间过滤要求 ISO 8601 带时区的时间串服务端会统一转成 UTC 再比较。完成上面两条路径后你就具备了监控客户端使用的全部读回能力用GET /tracesGET /traces/{trace_id}或POST /traces/search查单次运行用POST /metrics/refreshGET /metrics查按天聚合验证方式就是各示例脚本内置的RuntimeError检查和它们打印的条数、日期与计数。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
agno 如何读取 Agent 运行轨迹的 trace 并聚合按天的 metrics 指标？
agno 如何读取 Agent 运行轨迹的 trace 并聚合按天的 metrics 指标【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno在 agno 的 AgentOS 上运行 Agent 后你会经常遇到两类运维需求一是回放某次运行到底经历了哪些步骤Agent 执行、模型调用、工具调用各自耗时多少二是把一段时间内的活动按天聚合得到运行次数、会话数、用户数和 token 用量。AgentOS 对这两类数据提供了成对的接口trace 通过 OpenTelemetry span 产生用tracingTrue开启metrics 从持久化会话数据聚合用POST /metrics/refresh刷新、GET /metrics读取。本文按 cookbook/05_agent_os/13_observability/README.md 的示例走通「生成数据 → 通过监控客户端相同的路线读回」这条路径。两类数据的产生方式不同先分清两者后面的接口用法才不会混trace把 Agent 运行、模型调用、工具调用变成 span。开关是AgentOS(tracingTrue)同一开关覆盖注册到该 OS 的 agents、teams 和 workflows。trace 与 span 写入你为 OS 配置的 db示例中是SqliteDb。metrics按天聚合持久化下来的活动。metrics.py 的示例没有设置tracingTrue它只依赖SqliteDb持久化会话再调用 refresh 接口重算聚合。也就是说读 trace 需要开 tracing读 metrics 需要会话已持久化两者互相独立。准备环境真实运行 Agent 需要设置OPENAI_API_KEY。按 README 的要求demo 环境需要 Agno tracing 所用的 OpenTelemetry 包命令来自 README其中--python指定 demo 虚拟环境的解释器换成你自己的 venv 路径即可uv pip install --python .venvs/demo/bin/python \ opentelemetry-api opentelemetry-sdk openinference-instrumentation-agno示例均从仓库根目录运行例如.venvs/demo/bin/python cookbook/05_agent_os/13_observability/read_traces.py。开启 tracing基本配置如下关键点只有一个AgentOS(...)传tracingTruefrom agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.os import AgentOS db SqliteDb( idobservability-basic-db, db_filetmp/observability_basic.db, ) traced_agent Agent( idtraced-assistant, nameTraced Assistant, modelOpenAIResponses(idgpt-5.5), instructionsAnswer operational questions clearly and concisely., ) agent_os AgentOS( idobservability-basic-os, descriptionAgentOS with OpenTelemetry tracing stored in SQLite., dbdb, agents[traced_agent], tracingTrue, ) app agent_os.get_app() if __name__ __main__: agent_os.serve(appapp)运行.venvs/demo/bin/python cookbook/05_agent_os/13_observability/basic.py并调用其服务的 agent 后可以打开http://localhost:7777/traces和http://localhost:7777/traces/filter-schema查看已存储的 trace。读取 trace 列表与 span 树read_traces.py 演示了完整的读回路径先产生一次同步运行run和一次异步运行arun再通过 AgentOS 的 HTTP 接口读回两条 trace。它用httpx.ASGITransport直连get_app()返回的 app进程内就能跑通不需要另起服务sync_run trace_agent.run( Reply with a short greeting for the synchronous trace., session_idtrace-readback-sync, user_idobservability-user, ) async_run await trace_agent.arun( Reply with a short greeting for the asynchronous trace., session_idtrace-readback-async, user_idobservability-user, ) transport httpx.ASGITransport(appapp) async with httpx.AsyncClient( transporttransport, base_urlhttp://agent-os, ) as client: response await client.get( /traces, params{agent_id: AGENT_ID, db_id: DB_ID, limit: 20}, ) response.raise_for_status() trace_page response.json() traces_by_run { trace[run_id]: trace for trace in trace_page[data] if trace[run_id] } run_ids [sync_run.run_id, async_run.run_id] missing_run_ids [run_id for run_id in run_ids if run_id not in traces_by_run] if missing_run_ids: raise RuntimeError(fGET /traces omitted run IDs: {missing_run_ids}) for label, run_id in ((sync, sync_run.run_id), (async, async_run.run_id)): trace_id traces_by_run[run_id][trace_id] detail_response await client.get( f/traces/{trace_id}, params{db_id: DB_ID}, ) detail_response.raise_for_status() detail detail_response.json() if not detail[tree]: raise RuntimeError(fTrace {trace_id} has no span tree) print(f\n{label} run {run_id}: trace{trace_id}, spans{detail[total_spans]}) print_span_tree(detail[tree])两个接口的要点实现见 traces 路由GET /traces返回分页结构data中每条 trace 含trace_id、run_id、status、duration、total_spans等字段支持run_id、session_id、user_id、agent_id、team_id、workflow_id、statusOK/ERROR、start_time/end_timeISO 8601 带时区等过滤参数以及page1-indexed、limit默认 20、db_id。GET /traces/{trace_id}返回该 trace 的嵌套 span 树tree与total_spans。脚本按run_id把列表中的记录与本次运行对应起来再取trace_id拉详情最终递归打印name [type] status (duration)格式的层级树。脚本自身的校验就是验证方式如果两次运行对应的run_id没有出现在GET /traces结果里或详情里tree为空都会直接抛RuntimeError。正常输出会先打印GET /traces returned N trace(s)N 为meta.total_count随后是每条运行的 span 树。可选用 FilterExpr 做结构化搜索如果需要比查询参数更强的条件组合逻辑、任意字段filtering.py 展示的是FilterExprDSL它有两条等价路径filter_expr AND( EQ(status, OK), OR( EQ(agent_id, NEWS_AGENT_ID), EQ(agent_id, RELEASE_AGENT_ID), ), ) filter_dict filter_expr.to_dict() # 路径一直接在 Python 侧查 tracing 数据库 python_traces, python_count db.get_traces(filter_exprfilter_dict) # 路径二把同一表达式发给 AgentOS 的搜索接口 search_response await client.post( /traces/search, params{db_id: DB_ID}, json{filter: filter_dict, group_by: run, page: 1, limit: 20}, )配套的GET /traces/filter-schema返回可过滤字段fields[].key、操作符和枚举值、以及逻辑运算符客户端可以据此动态构建过滤 UI。该示例的验证点是schema 的字段集合必须包含status和agent_id逻辑运算符为[AND, OR]且搜索结果中同时出现两个 agent 的agent_id否则抛RuntimeError。刷新并读取按天的 metricsmetrics.py 是「聚合按天指标」的完整路径注意它的AgentOS没有tracingTrue只配置了SqliteDb先让 Agent 产生一次持久化运行arun带session_id和user_id这是 metrics 的原始数据来源。调用刷新接口重算聚合refresh_response await client.post( /metrics/refresh, params{db_id: DB_ID}, ) refresh_response.raise_for_status() refreshed_metrics refresh_response.json() if not refreshed_metrics: raise RuntimeError(POST /metrics/refresh returned no daily aggregate)再读回已存储的聚合metrics_response await client.get( /metrics, params{db_id: DB_ID}, ) metrics_response.raise_for_status() metrics_page metrics_response.json() if not metrics_page[metrics]: raise RuntimeError(GET /metrics returned no stored aggregate) latest max(metrics_page[metrics], keylambda metric: metric[date]) if latest[agent_runs_count] 1: raise RuntimeError(The refreshed aggregate omitted the new agent run) print(fDate: {latest[date]}) print(fAgent runs: {latest[agent_runs_count]}) print(fAgent sessions: {latest[agent_sessions_count]}) print(fUsers: {latest[users_count]}) print(fTotal tokens: {latest[token_metrics].get(total_tokens, 0)})接口语义实现见 metrics 路由POST /metrics/refresh默认同步执行并直接返回刷新后的每日聚合列表传backgroundtrue则立即返回 202随后用GET /metrics/refresh/status?db_id...轮询running/completed/failed状态从未触发时为idle同一数据库已有刷新在进行时会返回status: already_running而不会重复启动。GET /metrics返回metrics列表每个元素是一天的聚合含date、agent_runs_count、agent_sessions_count、team_runs_count、workflow_runs_count、users_count、token_metricsinput_tokens、output_tokens、total_tokens、audio_tokens、cached_tokens、reasoning_tokens等和model_metrics可用starting_date/ending_dateYYYY-MM-DD限定日期范围也可用user_id只取某用户的指标对非 admin 调用方该参数会被忽略。接口文档给出的示例响应中一条日聚合形如agent_runs_count: 5、total_tokens: 596——这是文档示例数值实际读数随你的运行数据变化。脚本最后打印的Date:、Agent runs:等行就是验证输出只要最新一天的agent_runs_count至少为 1说明新产生的一次运行已经被计入了按天聚合。边界与限制OS 注册了多个数据库时例如 traces_to_clickhouse.py 那种「会话存 SQLite、trace 批量写 ClickHouse」的拆分存储对GET /traces的请求在客户端传db_id之前是歧义的必须带上对应的db_id才能选中目标库。该变体还需要额外安装clickhouse-connect并用./cookbook/scripts/run_clickhouse.sh启动本地 ClickHouse默认端口8123用户/密码均为ai可用CLICKHOUSE_HOST等环境变量覆盖。metrics 是从持久化会话聚合的没有配置 db、运行未持久化的 OSrefresh 不会有数据可算。GET /traces的时间过滤要求 ISO 8601 带时区的时间串服务端会统一转成 UTC 再比较。完成上面两条路径后你就具备了监控客户端使用的全部读回能力用GET /tracesGET /traces/{trace_id}或POST /traces/search查单次运行用POST /metrics/refreshGET /metrics查按天聚合验证方式就是各示例脚本内置的RuntimeError检查和它们打印的条数、日期与计数。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
