尧图网站建设 
agno 团队级指标Team Metrics实战指南运行、流式、会话与工具调用时长的完整度量方案【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南以 cookbook/03_teams/22_metrics/README.md 为骨架系统讲解 agno 中团队Team级指标体系的四个核心维度单次运行指标run metrics、流式响应指标streaming metrics、跨多次运行的会话累计指标session metrics以及工具调用耗时指标tool call timing。读完本文你将掌握如何通过TeamRunOutput读取团队 Leader 与各成员的 Token 消耗、如何从流式输出中回收完整指标对象、如何在 PostgreSQL 持久化场景下获取会话级累计指标以及如何把 Eval 模型的消耗与工具执行耗时纳入统一的可观测视图并具备在嵌套团队中递归聚合指标的实战能力。一、团队指标总览六个示例覆盖的完整度量面在 agno 中一个 Team 由 Leader 模型与若干成员 Agent甚至是嵌套 Team组成。一次team.run()会产生多层级的执行行为Leader 自身的多次模型调用、每个成员的独立模型调用、成员执行工具的函数调用以及挂在post_hooks上的 Eval 模型调用。要精确回答这次团队运行到底花了多少 Token、多少钱、哪些环节最慢就需要一套分层级的指标结构。cookbook/03_teams/22_metrics/README.md 用六个示例文件完整覆盖了团队级指标的度量面示例文件度量维度核心看点01_team_metrics.py运行级Team、Session 与成员member三层的执行指标02_team_streaming_metrics.py流式级从流式响应中捕获指标并按模型逐项拆解明细03_team_session_metrics.py会话级借助 PostgreSQL 持久化指标跨多次 run 累计04_team_tool_metrics.py工具级工具执行耗时与成员级指标05_team_eval_metrics.pyEval 级通过 post_hook 收集 Eval 模型的指标eval_model键06_loop_team_and_member_metrics.py聚合级循环遍历 Leader 与每个成员的指标对嵌套团队递归并计算整次运行总消耗二、指标数据模型先读懂agno/metrics.py所有示例代码最终读取的都是 agno 指标框架的几类数据类它们集中定义在 libs/agno/agno/metrics.py 中。理解它们才能准确解读打印出来的指标对象。2.1 BaseMetrics所有指标的公共 Token 字段BaseMetrics是所有指标类型的基类定义了统一的 Token 消耗字段见 metrics.py字段含义input_tokens/output_tokens/total_tokens输入、输出与总 Token 数audio_input_tokens/audio_output_tokens/audio_total_tokens音频输入的 Token 消耗多模态场景cache_read_tokens/cache_write_tokens提示词缓存的读 / 写 Token如 OpenAI 的 prompt cachingreasoning_tokens推理reasoningToken 数cost本次消耗对应的成本浮点数由 Provider 计价2.2 ModelType指标明细的分类键RunMetrics.details是一个Dict[str, List[ModelMetrics]]其中的键就是ModelType枚举见 metrics.py。也就是说同一模型中扮演不同职能的调用会被拆分记账model团队/Agent 主模型的对话调用output_model、parser_model结构化输出与解析模型reasoning_model、memory_model、learning_model、session_summary_model推理、记忆、学习与会话摘要模型compression_model上下文压缩模型followup_model追问建议模型。这就是per-model detail breakdown按模型逐项拆解明细的含义details[model]下存放的是主模型的每次调用的ModelMetricsdetails[eval_model]下存放的是 Eval 模型调用。2.3 ModelMetrics按provider, id聚合的单模型指标ModelMetrics继承BaseMetrics额外携带id模型 ID、provider提供商与provider_metricsProvider 原始指标并提供accumulate()方法用于累加见 metrics.py。注意它的注释明确说明在 run 级别details[model_type]中每个唯一的provider, id对应一条记录在 session 级别同一结构下 Token 会在多次 run 之间求和。2.4 RunMetrics 与 SessionMetrics运行级与会话级RunMetrics见 metrics.py是RunOutput.metrics与TeamRunOutput.metrics的类型在BaseMetrics之上增加了timer/duration本次运行的计时器与总耗时time_to_first_token首 Token 延迟TTFTdetailsDict[str, List[ModelMetrics]]按模型职能分类的明细additional_metrics额外指标例如eval_duration。SessionMetrics见 metrics.py是会话级的聚合结构details结构与RunMetrics完全一致区别在于每一条ModelMetrics的 Token 都是跨多次 run 求和后的结果通过accumulate_from_run()方法从 run 级指标累加而来见 metrics.py。2.5 MessageMetrics 与 ToolCallMetrics消息级与工具级MessageMetrics见 metrics.py挂在Message.metrics上包含duration与time_to_first_token用于衡量单条消息层面的 Token 与延迟。ToolCallMetrics见 metrics.py只关心时间相关字段start_time、end_time与duration通过start_timer()/stop_timer()计时。它挂在ToolExecution.metrics上见 models/response.py是工具调用耗时的直接来源。三、环境准备与运行方式按 cookbook/03_teams/22_metrics/README.md 的说明运行这些示例需要三步加载环境变量在项目根目录执行direnv allow确保已导出OPENAI_API_KEY等凭据创建演示环境运行./scripts/demo_setup.sh随后所有示例统一用.venvs/demo/bin/python执行按需启动 PostgreSQL示例01与03使用了PostgresDb做会话持久化需要先运行./cookbook/scripts/run_pgvector.sh拉起数据库。执行单个示例的命令为.venvs/demo/bin/python cookbook/03_teams/22_metrics/file.py其中file.py替换为01_team_metrics.py至06_loop_team_and_member_metrics.py中的任意一个。四、示例一一次 run 中的三层指标01_team_metrics.py01_team_metrics.py 是理解团队指标结构的入门示例。它用YFinanceTools构造一个股票搜索成员并把团队接入 PostgreSQL 会话表from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIResponses from agno.team import Team from agno.tools.yfinance import YFinanceTools from agno.utils.pprint import pprint_run_response from rich.pretty import pprint db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url, session_tableteam_metrics_sessions) stock_searcher Agent( nameStock Searcher, modelOpenAIResponses(idgpt-5-mini), roleSearches the web for information on a stock., tools[YFinanceTools()], ) team Team( nameStock Research Team, modelOpenAIResponses(idgpt-5-mini), members[stock_searcher], dbdb, session_idteam_metrics_demo, markdownTrue, show_members_responsesTrue, store_member_responsesTrue, )示例的核心是三个数据访问点分别对应三层指标第一层Team Leader 的消息级指标。遍历run_output.messages对每个role assistant的消息打印message.metrics类型为MessageMetrics得到 Leader 每一条消息的 Token 与耗时if run_output.messages: for message in run_output.messages: if message.role assistant: if message.content: print(f Message: {message.content[:100]}...) elif message.tool_calls: print(fTool calls: {message.tool_calls}) print(- * 30, Metrics, - * 30) pprint(message.metrics)第二层Team Leader 的运行级指标。run_output.metrics的类型是RunMetrics聚合了 Leader 本次 run 的全部模型调用details[model]、details[output_model]等pprint(run_output.metrics)第三层会话级指标。调用team.get_session_metrics(session_idteam_metrics_demo)获取SessionMetrics。在 team/_session.py 中该 API 会回退到team.session_id真正读取逻辑在 utils/agent.py从会话的session_data[session_metrics]中还原SessionMetrics对象。成员级指标则在run_output.member_responses中——这是TeamRunOutput的关键结构见 run/team.pymember_responses是List[Union[TeamRunOutput, RunOutput]]每个成员响应自带metrics、tools、messages字段与 Leader 层完全同构if run_output.member_responses: for member_response in run_output.member_responses: if member_response.messages: for message in member_response.messages: if message.role assistant: ... pprint(message.metrics)五、示例二从流式响应中回收指标02_team_streaming_metrics.py流式场景下team.run(..., streamTrue)返回的是事件迭代器RunMetrics并不会出现在单个事件上。解决方法是 02_team_streaming_metrics.py 展示的yield_run_outputTrue流结束时框架会产出一个TeamRunOutput事件其metrics字段携带本次流的完整指标from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.run.team import TeamRunOutput from agno.team import Team from rich.pretty import pprint team Team( nameStreaming Team, modelOpenAIChat(idgpt-5.6-luna), members[assistant], markdownTrue, ) response None for event in team.run(Count from 1 to 5., streamTrue, yield_run_outputTrue): if isinstance(event, TeamRunOutput): response event if response and response.metrics: pprint(response.metrics)随后示例演示了按模型职能分类的明细遍历——这正是RunMetrics.detailsDict[str, List[ModelMetrics]]的典型消费方式if response.metrics.details: for model_type, model_metrics_list in response.metrics.details.items(): print(f\n{model_type}:) for model_metric in model_metrics_list: pprint(model_metric)这里每个model_metric都是ModelMetrics可直接读取provider、id、input_tokens、output_tokens、total_tokens与cost。六、示例三会话级指标的跨 run 累计03_team_session_metrics.py03_team_session_metrics.py 回答一个会话跑了很多次怎么统计总量。前提是团队配置了PostgresDb与固定session_idteam Team( nameResearch Team, modelOpenAIChat(idgpt-5.6-luna), members[assistant], dbdb, session_idteam_session_metrics_demo, markdownTrue, )脚本在同一会话上连续执行两次team.run()分别打印两次run_output.metrics最后调用team.get_session_metrics()获取累计值run_output_1 team.run(What is the capital of Japan?) pprint(run_output_1.metrics) run_output_2 team.run(What about South Korea?) pprint(run_output_2.metrics) session_metrics team.get_session_metrics() pprint(session_metrics)底层累计逻辑在 team/_session.py 的update_session_metrics中每次 run 结束后它先取get_session_metrics_internal得到已有会话指标再通过session_metrics.accumulate_from_run(run_response.metrics)累加 Leader 消耗并调用_accumulate_member_metrics递归累加所有成员含嵌套团队的消耗最后写回session.session_data[session_metrics]。持久化落点由 team/_storage.py 的get_session_metrics提供会话读取入口。七、示例四工具执行耗时与成员指标04_team_tool_metrics.py工具调用是团队里最值得做性能观测的环节。04_team_tool_metrics.py 在成员指标之外额外打印了每次工具调用的耗时if run_output.member_responses: for member_response in run_output.member_responses: print(f\nMember: {member_response.agent_name}) pprint(member_response.metrics) if member_response.tools: print(f\nTool calls ({len(member_response.tools)}):) for tool_call in member_response.tools: print(f Tool: {tool_call.tool_name}) if tool_call.metrics: pprint(tool_call.metrics)member_response.tools的元素类型是ToolExecution定义于 models/response.py它的metrics字段是ToolCallMetrics——即start_time、end_time、duration。在 metrics.py 中计时通过start_timer()/stop_timer()完成启动时记录开始时间停止时用Timer.elapsed写入duration。配合成员→工具的嵌套打印可以快速定位哪个成员调的哪个工具最慢。八、示例五把 Eval 模型的消耗纳入指标05_team_eval_metrics.py团队通过post_hooks挂上AgentAsJudgeEval做质量评估时评估器自身还会发起一次模型调用这次调用的 Token 消耗会被框架自动回写到本次 run 的指标中。05_team_eval_metrics.py 演示了这一点from agno.agent import Agent from agno.eval.agent_as_judge import AgentAsJudgeEval from agno.models.openai import OpenAIChat from agno.team import Team from rich.pretty import pprint eval_hook AgentAsJudgeEval( nameQuality Check, modelOpenAIChat(idgpt-5.6-luna), criteriaResponse should be accurate, well-structured, and concise, scoring_strategybinary, ) team Team( nameResearch Team, modelOpenAIChat(idgpt-5.6-luna), members[researcher], post_hooks[eval_hook], show_members_responsesTrue, store_member_responsesTrue, )运行后Eval 模型消耗会以eval_model为键出现在result.metrics.details中与团队主模型model键分开记账。示例代码分别汇总两者if result.metrics.details: if model in result.metrics.details: team_tokens sum(metric.total_tokens for metric in result.metrics.details[model]) print(Team model tokens:, team_tokens) if eval_model in result.metrics.details: eval_tokens sum(metric.total_tokens for metric in result.metrics.details[eval_model]) print(Eval model tokens:, eval_tokens) for metric in result.metrics.details[eval_model]: print(f Evaluator: {metric.id} ({metric.provider}))从实现上看AgentAsJudgeEval见 eval/agent_as_judge.py支持scoring_strategynumeric1-10 分制或binary二元判定与threshold数值策略下默认 7 分阈值。它的每次评估都是一次独立的模型调用因而在指标明细中自然形成一条eval_model记录。九、示例六递归遍历嵌套团队的指标并计算 run 总消耗06_loop_team_and_member_metrics.py最后一个示例 06_loop_team_and_member_metrics.py 是完整的指标走查工具其文件头注释给出了团队指标最关键的几条事实run_output.metrics只包含团队Leader的调用model、parser_model、output_model、followup_model以及后台的记忆/学习模型不包含成员的 Token 消耗每个成员的指标位于run_output.member_responses[i].metrics对嵌套团队要递归遍历member_responses需要跨 run 的会话总量时使用team.get_session_metrics()。示例定义了三个可复用的辅助函数。第一个统一打印任意RunMetrics含按模型拆分的detailsdef print_run_metrics(label: str, metrics: Optional[RunMetrics]) - None: if metrics is None: print(f{label}: no metrics) return print( f{label}: input{metrics.input_tokens}, foutput{metrics.output_tokens}, total{metrics.total_tokens} ) if metrics.details: for model_type, entries in metrics.details.items(): for entry in entries: print( f - [{model_type}] {entry.provider}/{entry.id}: finput{entry.input_tokens}, output{entry.output_tokens}, ftotal{entry.total_tokens} )第二个递归遍历成员兼容 Agent 与嵌套 Team 两种成员类型def walk_member_metrics( member_responses: Iterable[Union[RunOutput, TeamRunOutput]], depth: int 1, ) - None: for i, mr in enumerate(member_responses): kind team if isinstance(mr, TeamRunOutput) else agent name mr.team_name if isinstance(mr, TeamRunOutput) else mr.agent_name prefix * depth print_run_metrics(f{prefix}member[{i}] ({kind}: {name}), mr.metrics) if isinstance(mr, TeamRunOutput) and mr.member_responses: walk_member_metrics(mr.member_responses, depth 1)第三个递归求和得出Leader 所有成员含子团队成员的整次 run 总 Tokendef total_run_tokens(run_output: TeamRunOutput) - int: def walk(responses: Iterable[Union[RunOutput, TeamRunOutput]]) - int: s 0 for mr in responses: if mr.metrics is not None: s mr.metrics.total_tokens if isinstance(mr, TeamRunOutput) and mr.member_responses: s walk(mr.member_responses) return s leader_total run_output.metrics.total_tokens if run_output.metrics else 0 return leader_total walk(run_output.member_responses)主流程依次打印 Leader 指标、逐成员指标与总消耗print_run_metrics(leader, run_output.metrics) walk_member_metrics(run_output.member_responses) print(ftotal_tokens {total_run_tokens(run_output)})需要注意的是TeamRunOutput中的isinstance判断之所以成立是因为member_responses的元素类型是Union[TeamRunOutput, RunOutput]见 run/team.py嵌套团队在成员列表里同样以TeamRunOutput呈现。十、指标落点速查表你要查看的指标访问路径类型Leader 单条消息的 Token / 耗时 / TTFTrun_output.messages[i].metricsMessageMetricsLeader 本次 run 的聚合指标run_output.metricsRunMetrics按模型职能拆分的明细run_output.metrics.details[model / eval_model / ...]Dict[str, List[ModelMetrics]]成员Agent本次 run 的指标run_output.member_responses[i].metricsRunMetrics嵌套团队成员的指标对member_responses中TeamRunOutput递归RunMetrics成员执行的工具耗时member_response.tools[j].metricsToolCallMetricsstart_time/end_time/duration会话级累计指标跨 runteam.get_session_metrics(session_id)SessionMetricsEval 模型消耗run_output.metrics.details[eval_model]List[ModelMetrics]从源码结构看这套体系的核心设计原则是职责分离与同构递归TeamRunOutput与RunOutput共享metrics/tools/messages字段形态成员列表允许嵌套TeamRunOutput因此无论团队嵌套多少层都能用同一套RunMetrics结构递归聚合而ModelType分类键则保证同一模型、不同职能的调用在记账上互不混淆。对于需要精细化运营多智能体系统的团队这套指标可以作为成本核算、性能剖析与质量评估的统一数据底座。后续若需要将指标接入可观测平台或统计后台直接对RunMetrics.to_dict()/SessionMetrics.to_dict()的结果做序列化即可两者都提供了完善的to_dict/from_dict双向转换见 metrics.py 与 metrics.py。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 agno 团队级指标（Team Metrics）实战指南：运行、流式、会话与工具调用时长的完整度量方案
agno 团队级指标Team Metrics实战指南运行、流式、会话与工具调用时长的完整度量方案【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南以 cookbook/03_teams/22_metrics/README.md 为骨架系统讲解 agno 中团队Team级指标体系的四个核心维度单次运行指标run metrics、流式响应指标streaming metrics、跨多次运行的会话累计指标session metrics以及工具调用耗时指标tool call timing。读完本文你将掌握如何通过TeamRunOutput读取团队 Leader 与各成员的 Token 消耗、如何从流式输出中回收完整指标对象、如何在 PostgreSQL 持久化场景下获取会话级累计指标以及如何把 Eval 模型的消耗与工具执行耗时纳入统一的可观测视图并具备在嵌套团队中递归聚合指标的实战能力。一、团队指标总览六个示例覆盖的完整度量面在 agno 中一个 Team 由 Leader 模型与若干成员 Agent甚至是嵌套 Team组成。一次team.run()会产生多层级的执行行为Leader 自身的多次模型调用、每个成员的独立模型调用、成员执行工具的函数调用以及挂在post_hooks上的 Eval 模型调用。要精确回答这次团队运行到底花了多少 Token、多少钱、哪些环节最慢就需要一套分层级的指标结构。cookbook/03_teams/22_metrics/README.md 用六个示例文件完整覆盖了团队级指标的度量面示例文件度量维度核心看点01_team_metrics.py运行级Team、Session 与成员member三层的执行指标02_team_streaming_metrics.py流式级从流式响应中捕获指标并按模型逐项拆解明细03_team_session_metrics.py会话级借助 PostgreSQL 持久化指标跨多次 run 累计04_team_tool_metrics.py工具级工具执行耗时与成员级指标05_team_eval_metrics.pyEval 级通过 post_hook 收集 Eval 模型的指标eval_model键06_loop_team_and_member_metrics.py聚合级循环遍历 Leader 与每个成员的指标对嵌套团队递归并计算整次运行总消耗二、指标数据模型先读懂agno/metrics.py所有示例代码最终读取的都是 agno 指标框架的几类数据类它们集中定义在 libs/agno/agno/metrics.py 中。理解它们才能准确解读打印出来的指标对象。2.1 BaseMetrics所有指标的公共 Token 字段BaseMetrics是所有指标类型的基类定义了统一的 Token 消耗字段见 metrics.py字段含义input_tokens/output_tokens/total_tokens输入、输出与总 Token 数audio_input_tokens/audio_output_tokens/audio_total_tokens音频输入的 Token 消耗多模态场景cache_read_tokens/cache_write_tokens提示词缓存的读 / 写 Token如 OpenAI 的 prompt cachingreasoning_tokens推理reasoningToken 数cost本次消耗对应的成本浮点数由 Provider 计价2.2 ModelType指标明细的分类键RunMetrics.details是一个Dict[str, List[ModelMetrics]]其中的键就是ModelType枚举见 metrics.py。也就是说同一模型中扮演不同职能的调用会被拆分记账model团队/Agent 主模型的对话调用output_model、parser_model结构化输出与解析模型reasoning_model、memory_model、learning_model、session_summary_model推理、记忆、学习与会话摘要模型compression_model上下文压缩模型followup_model追问建议模型。这就是per-model detail breakdown按模型逐项拆解明细的含义details[model]下存放的是主模型的每次调用的ModelMetricsdetails[eval_model]下存放的是 Eval 模型调用。2.3 ModelMetrics按provider, id聚合的单模型指标ModelMetrics继承BaseMetrics额外携带id模型 ID、provider提供商与provider_metricsProvider 原始指标并提供accumulate()方法用于累加见 metrics.py。注意它的注释明确说明在 run 级别details[model_type]中每个唯一的provider, id对应一条记录在 session 级别同一结构下 Token 会在多次 run 之间求和。2.4 RunMetrics 与 SessionMetrics运行级与会话级RunMetrics见 metrics.py是RunOutput.metrics与TeamRunOutput.metrics的类型在BaseMetrics之上增加了timer/duration本次运行的计时器与总耗时time_to_first_token首 Token 延迟TTFTdetailsDict[str, List[ModelMetrics]]按模型职能分类的明细additional_metrics额外指标例如eval_duration。SessionMetrics见 metrics.py是会话级的聚合结构details结构与RunMetrics完全一致区别在于每一条ModelMetrics的 Token 都是跨多次 run 求和后的结果通过accumulate_from_run()方法从 run 级指标累加而来见 metrics.py。2.5 MessageMetrics 与 ToolCallMetrics消息级与工具级MessageMetrics见 metrics.py挂在Message.metrics上包含duration与time_to_first_token用于衡量单条消息层面的 Token 与延迟。ToolCallMetrics见 metrics.py只关心时间相关字段start_time、end_time与duration通过start_timer()/stop_timer()计时。它挂在ToolExecution.metrics上见 models/response.py是工具调用耗时的直接来源。三、环境准备与运行方式按 cookbook/03_teams/22_metrics/README.md 的说明运行这些示例需要三步加载环境变量在项目根目录执行direnv allow确保已导出OPENAI_API_KEY等凭据创建演示环境运行./scripts/demo_setup.sh随后所有示例统一用.venvs/demo/bin/python执行按需启动 PostgreSQL示例01与03使用了PostgresDb做会话持久化需要先运行./cookbook/scripts/run_pgvector.sh拉起数据库。执行单个示例的命令为.venvs/demo/bin/python cookbook/03_teams/22_metrics/file.py其中file.py替换为01_team_metrics.py至06_loop_team_and_member_metrics.py中的任意一个。四、示例一一次 run 中的三层指标01_team_metrics.py01_team_metrics.py 是理解团队指标结构的入门示例。它用YFinanceTools构造一个股票搜索成员并把团队接入 PostgreSQL 会话表from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIResponses from agno.team import Team from agno.tools.yfinance import YFinanceTools from agno.utils.pprint import pprint_run_response from rich.pretty import pprint db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url, session_tableteam_metrics_sessions) stock_searcher Agent( nameStock Searcher, modelOpenAIResponses(idgpt-5-mini), roleSearches the web for information on a stock., tools[YFinanceTools()], ) team Team( nameStock Research Team, modelOpenAIResponses(idgpt-5-mini), members[stock_searcher], dbdb, session_idteam_metrics_demo, markdownTrue, show_members_responsesTrue, store_member_responsesTrue, )示例的核心是三个数据访问点分别对应三层指标第一层Team Leader 的消息级指标。遍历run_output.messages对每个role assistant的消息打印message.metrics类型为MessageMetrics得到 Leader 每一条消息的 Token 与耗时if run_output.messages: for message in run_output.messages: if message.role assistant: if message.content: print(f Message: {message.content[:100]}...) elif message.tool_calls: print(fTool calls: {message.tool_calls}) print(- * 30, Metrics, - * 30) pprint(message.metrics)第二层Team Leader 的运行级指标。run_output.metrics的类型是RunMetrics聚合了 Leader 本次 run 的全部模型调用details[model]、details[output_model]等pprint(run_output.metrics)第三层会话级指标。调用team.get_session_metrics(session_idteam_metrics_demo)获取SessionMetrics。在 team/_session.py 中该 API 会回退到team.session_id真正读取逻辑在 utils/agent.py从会话的session_data[session_metrics]中还原SessionMetrics对象。成员级指标则在run_output.member_responses中——这是TeamRunOutput的关键结构见 run/team.pymember_responses是List[Union[TeamRunOutput, RunOutput]]每个成员响应自带metrics、tools、messages字段与 Leader 层完全同构if run_output.member_responses: for member_response in run_output.member_responses: if member_response.messages: for message in member_response.messages: if message.role assistant: ... pprint(message.metrics)五、示例二从流式响应中回收指标02_team_streaming_metrics.py流式场景下team.run(..., streamTrue)返回的是事件迭代器RunMetrics并不会出现在单个事件上。解决方法是 02_team_streaming_metrics.py 展示的yield_run_outputTrue流结束时框架会产出一个TeamRunOutput事件其metrics字段携带本次流的完整指标from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.run.team import TeamRunOutput from agno.team import Team from rich.pretty import pprint team Team( nameStreaming Team, modelOpenAIChat(idgpt-5.6-luna), members[assistant], markdownTrue, ) response None for event in team.run(Count from 1 to 5., streamTrue, yield_run_outputTrue): if isinstance(event, TeamRunOutput): response event if response and response.metrics: pprint(response.metrics)随后示例演示了按模型职能分类的明细遍历——这正是RunMetrics.detailsDict[str, List[ModelMetrics]]的典型消费方式if response.metrics.details: for model_type, model_metrics_list in response.metrics.details.items(): print(f\n{model_type}:) for model_metric in model_metrics_list: pprint(model_metric)这里每个model_metric都是ModelMetrics可直接读取provider、id、input_tokens、output_tokens、total_tokens与cost。六、示例三会话级指标的跨 run 累计03_team_session_metrics.py03_team_session_metrics.py 回答一个会话跑了很多次怎么统计总量。前提是团队配置了PostgresDb与固定session_idteam Team( nameResearch Team, modelOpenAIChat(idgpt-5.6-luna), members[assistant], dbdb, session_idteam_session_metrics_demo, markdownTrue, )脚本在同一会话上连续执行两次team.run()分别打印两次run_output.metrics最后调用team.get_session_metrics()获取累计值run_output_1 team.run(What is the capital of Japan?) pprint(run_output_1.metrics) run_output_2 team.run(What about South Korea?) pprint(run_output_2.metrics) session_metrics team.get_session_metrics() pprint(session_metrics)底层累计逻辑在 team/_session.py 的update_session_metrics中每次 run 结束后它先取get_session_metrics_internal得到已有会话指标再通过session_metrics.accumulate_from_run(run_response.metrics)累加 Leader 消耗并调用_accumulate_member_metrics递归累加所有成员含嵌套团队的消耗最后写回session.session_data[session_metrics]。持久化落点由 team/_storage.py 的get_session_metrics提供会话读取入口。七、示例四工具执行耗时与成员指标04_team_tool_metrics.py工具调用是团队里最值得做性能观测的环节。04_team_tool_metrics.py 在成员指标之外额外打印了每次工具调用的耗时if run_output.member_responses: for member_response in run_output.member_responses: print(f\nMember: {member_response.agent_name}) pprint(member_response.metrics) if member_response.tools: print(f\nTool calls ({len(member_response.tools)}):) for tool_call in member_response.tools: print(f Tool: {tool_call.tool_name}) if tool_call.metrics: pprint(tool_call.metrics)member_response.tools的元素类型是ToolExecution定义于 models/response.py它的metrics字段是ToolCallMetrics——即start_time、end_time、duration。在 metrics.py 中计时通过start_timer()/stop_timer()完成启动时记录开始时间停止时用Timer.elapsed写入duration。配合成员→工具的嵌套打印可以快速定位哪个成员调的哪个工具最慢。八、示例五把 Eval 模型的消耗纳入指标05_team_eval_metrics.py团队通过post_hooks挂上AgentAsJudgeEval做质量评估时评估器自身还会发起一次模型调用这次调用的 Token 消耗会被框架自动回写到本次 run 的指标中。05_team_eval_metrics.py 演示了这一点from agno.agent import Agent from agno.eval.agent_as_judge import AgentAsJudgeEval from agno.models.openai import OpenAIChat from agno.team import Team from rich.pretty import pprint eval_hook AgentAsJudgeEval( nameQuality Check, modelOpenAIChat(idgpt-5.6-luna), criteriaResponse should be accurate, well-structured, and concise, scoring_strategybinary, ) team Team( nameResearch Team, modelOpenAIChat(idgpt-5.6-luna), members[researcher], post_hooks[eval_hook], show_members_responsesTrue, store_member_responsesTrue, )运行后Eval 模型消耗会以eval_model为键出现在result.metrics.details中与团队主模型model键分开记账。示例代码分别汇总两者if result.metrics.details: if model in result.metrics.details: team_tokens sum(metric.total_tokens for metric in result.metrics.details[model]) print(Team model tokens:, team_tokens) if eval_model in result.metrics.details: eval_tokens sum(metric.total_tokens for metric in result.metrics.details[eval_model]) print(Eval model tokens:, eval_tokens) for metric in result.metrics.details[eval_model]: print(f Evaluator: {metric.id} ({metric.provider}))从实现上看AgentAsJudgeEval见 eval/agent_as_judge.py支持scoring_strategynumeric1-10 分制或binary二元判定与threshold数值策略下默认 7 分阈值。它的每次评估都是一次独立的模型调用因而在指标明细中自然形成一条eval_model记录。九、示例六递归遍历嵌套团队的指标并计算 run 总消耗06_loop_team_and_member_metrics.py最后一个示例 06_loop_team_and_member_metrics.py 是完整的指标走查工具其文件头注释给出了团队指标最关键的几条事实run_output.metrics只包含团队Leader的调用model、parser_model、output_model、followup_model以及后台的记忆/学习模型不包含成员的 Token 消耗每个成员的指标位于run_output.member_responses[i].metrics对嵌套团队要递归遍历member_responses需要跨 run 的会话总量时使用team.get_session_metrics()。示例定义了三个可复用的辅助函数。第一个统一打印任意RunMetrics含按模型拆分的detailsdef print_run_metrics(label: str, metrics: Optional[RunMetrics]) - None: if metrics is None: print(f{label}: no metrics) return print( f{label}: input{metrics.input_tokens}, foutput{metrics.output_tokens}, total{metrics.total_tokens} ) if metrics.details: for model_type, entries in metrics.details.items(): for entry in entries: print( f - [{model_type}] {entry.provider}/{entry.id}: finput{entry.input_tokens}, output{entry.output_tokens}, ftotal{entry.total_tokens} )第二个递归遍历成员兼容 Agent 与嵌套 Team 两种成员类型def walk_member_metrics( member_responses: Iterable[Union[RunOutput, TeamRunOutput]], depth: int 1, ) - None: for i, mr in enumerate(member_responses): kind team if isinstance(mr, TeamRunOutput) else agent name mr.team_name if isinstance(mr, TeamRunOutput) else mr.agent_name prefix * depth print_run_metrics(f{prefix}member[{i}] ({kind}: {name}), mr.metrics) if isinstance(mr, TeamRunOutput) and mr.member_responses: walk_member_metrics(mr.member_responses, depth 1)第三个递归求和得出Leader 所有成员含子团队成员的整次 run 总 Tokendef total_run_tokens(run_output: TeamRunOutput) - int: def walk(responses: Iterable[Union[RunOutput, TeamRunOutput]]) - int: s 0 for mr in responses: if mr.metrics is not None: s mr.metrics.total_tokens if isinstance(mr, TeamRunOutput) and mr.member_responses: s walk(mr.member_responses) return s leader_total run_output.metrics.total_tokens if run_output.metrics else 0 return leader_total walk(run_output.member_responses)主流程依次打印 Leader 指标、逐成员指标与总消耗print_run_metrics(leader, run_output.metrics) walk_member_metrics(run_output.member_responses) print(ftotal_tokens {total_run_tokens(run_output)})需要注意的是TeamRunOutput中的isinstance判断之所以成立是因为member_responses的元素类型是Union[TeamRunOutput, RunOutput]见 run/team.py嵌套团队在成员列表里同样以TeamRunOutput呈现。十、指标落点速查表你要查看的指标访问路径类型Leader 单条消息的 Token / 耗时 / TTFTrun_output.messages[i].metricsMessageMetricsLeader 本次 run 的聚合指标run_output.metricsRunMetrics按模型职能拆分的明细run_output.metrics.details[model / eval_model / ...]Dict[str, List[ModelMetrics]]成员Agent本次 run 的指标run_output.member_responses[i].metricsRunMetrics嵌套团队成员的指标对member_responses中TeamRunOutput递归RunMetrics成员执行的工具耗时member_response.tools[j].metricsToolCallMetricsstart_time/end_time/duration会话级累计指标跨 runteam.get_session_metrics(session_id)SessionMetricsEval 模型消耗run_output.metrics.details[eval_model]List[ModelMetrics]从源码结构看这套体系的核心设计原则是职责分离与同构递归TeamRunOutput与RunOutput共享metrics/tools/messages字段形态成员列表允许嵌套TeamRunOutput因此无论团队嵌套多少层都能用同一套RunMetrics结构递归聚合而ModelType分类键则保证同一模型、不同职能的调用在记账上互不混淆。对于需要精细化运营多智能体系统的团队这套指标可以作为成本核算、性能剖析与质量评估的统一数据底座。后续若需要将指标接入可观测平台或统计后台直接对RunMetrics.to_dict()/SessionMetrics.to_dict()的结果做序列化即可两者都提供了完善的to_dict/from_dict双向转换见 metrics.py 与 metrics.py。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考