Agno 可观测性实战指南为 Agent、团队与工作流接入 OpenTelemetry 追踪体系【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南以 Agno 仓库中cookbook/observability/目录为蓝本系统讲解如何为 Agno Agent、多智能体 Team 与 Workflow 接入可观测性Observability能力既包含 Agno 内置的两表追踪落库方案也覆盖基于 OpenInference 与 OpenTelemetry 的通用接入路径以及 AgentOps、Arize Phoenix、Langfuse、LangSmith、Logfire、MLflow、Opik、Weave 等 20 个主流观测平台的集成示例。读完本文你将掌握在真实项目中一键开启追踪、把 trace 导出到任意 OTLP 兼容后端以及在数据库层直接查询 trace 与 span 的完整实战方案。一、Agno 可观测性整体架构Agno 的可观测性示例集中在cookbook/observability/目录按覆盖对象分为三层Agent 层单 Agent 的模型调用、工具执行、结构化输出追踪Team 层多智能体协作的调用链追踪见 teamsWorkflow 层多步骤工作流的执行追踪见 workflows。从源码结构看libs/agno/agno/tracing/init.pyAgno 的追踪体系建立在OpenTelemetry之上它使用openinference-instrumentation-agno完成自动插桩automatic instrumentation并提供自定义的DatabaseSpanExporter把 trace 落库。这意味着追踪能力存在两条清晰的落地路径内置数据库追踪通过agno.tracing.setup_tracing(db...)一键开启trace 写入本地/远程数据库无需外部平台第三方平台追踪通过 OpenInference OTLP Exporter 把 span 推送到 Phoenix、Langfuse、LangSmith、Logfire 等观测平台或用各平台 SDK 直接插桩。两条路径共用同一套 OpenTelemetry 基础设施这也是本文后面所有示例能够一套代码、多端导出的根本原因。二、内置追踪两表设计与数据库落库如果不希望依赖任何外部 SaaSAgno 提供了开箱即用的数据库追踪方案参考示例 trace_to_database.py。2.1 一行代码开启全局追踪from agno.db.sqlite import SqliteDb from agno.tracing import setup_tracing db SqliteDb(db_filetmp/traces.db) setup_tracing(dbdb) # 自动为所有 Agent/Team/Workflow 开启追踪调用setup_tracing()之后后续创建的Agent.run()、模型调用、工具执行、团队协作与工作流步骤都会被自动捕获无需逐个 Agent 配置。从 setup.py 源码可以看到setup_tracing的完整行为它会创建TracerProvider将DatabaseSpanExporter挂接到 span processor 上并通过AgnoInstrumentor().instrument(tracer_provider...)完成对 Agno 的自动插桩。其关键参数如下参数默认值说明db必填存储 trace 的数据库实例同步BaseDb、异步AsyncBaseDb或RemoteDb均可batch_processingFalse为True时使用BatchSpanProcessor批量导出、性能更好为False时使用SimpleSpanProcessor即时导出、便于调试max_queue_size2048批量处理器队列最大长度max_export_batch_size512单次批量导出的最大 span 数schedule_delay_millis5000批量导出的调度间隔毫秒需要注意若缺少opentelemetry-api、opentelemetry-sdk、openinference-instrumentation-agno等依赖setup_tracing会抛出ImportError并提示安装命令。另外该函数检测到已存在真实的TracerProvider时会直接返回幂等避免重复初始化。2.2 两表设计Trace 与 SpanAgno 的追踪采用每个 trace_id 一条 Trace 记录 每条 span 一条 Span 记录的两表设计数据模型定义在 schemas.pyTrace聚合一条完整调用的元信息字段包括trace_id、name取自根 span、statusOK/ERROR/UNSET、duration_ms、total_spans、error_count以及从根 span 提取的run_id、session_id、user_id、agent_id、team_id、workflow_id等上下文字段Span记录单次操作的明细字段包括span_id、trace_id、parent_span_id、span_kind、status_code、duration_ms与完整attributes字典。落库时exporter.py 中的DatabaseSpanExporter.export()会执行三步先把 OpenTelemetry 的ReadableSpan通过Span.from_otel_span()转换为 Agno 的Span模型再按trace_id分组最后对每组调用create_trace_from_spans()聚合成 Trace 记录通过db.upsert_trace(trace)与db.create_spans(spans)写入数据库。同步与异步数据库分别走_export_sync与_export_async分支RemoteDb远程数据库则由其自身处理追踪。值得一提的是create_trace_from_spans会刻意选取没有父 span 的根 span来提取上下文字段——若某批导出的只有子 span则上下文字段留空交由数据库的 upsert 合并逻辑COALESCE保留父级 trace 的身份信息避免子调用覆盖父 trace 的agent_id/session_id等标识。2.3 从数据库查询 Trace 与 Span示例程序在运行 Agent 之后直接通过数据库接口读取并打印了完整的追踪结构trace db.get_trace(run_idresponse.run_id) # 按 run_id 反查 trace spans db.get_spans(trace_idtrace.trace_id) # 获取该 trace 下所有 span这些接口定义在 libs/agno/agno/db/base.py 中是BaseDb的抽象方法所有具体数据库实现SQLite、Postgres、MySQL 等都会提供upsert_trace(trace)/create_spans(spans)写入接口get_trace(trace_idNone, run_idNone)按 trace_id 或 run_id两者都唯一精确查询单条 traceget_traces(...)支持按run_id、session_id、user_id、agent_id、team_id、workflow_id、status、时间范围过滤带limit/page分页并支持filter_expr高级过滤表达式EQ/NEQ/GT/LT/IN/CONTAINS 等组合逻辑get_spans按 trace 获取全部 span。示例中对 span 的解析展示了 OpenInference 语义属性的实际形态span 的openinference.span.kind属性区分AGENT、TOOL、LLM等类型TOOLspan 带tool.name、tool.parametersLLMspan 带llm.model_name、llm.token_count.prompt、llm.token_count.completion、llm.input_messages等通用属性则包括session.id、user.id、agno.agent.id、agno.run.id。同时兼容gen_ai.request.model、gen_ai.usage.prompt_tokens等更通用的语义属性名。这些正是你在任何观测平台 UI 上看到的结构化信息。提示使用BatchSpanProcessor时span 是异步批量落库的查询前可能需要time.sleep(5)等待刷新示例中以注释形式给出。三、OpenInference OTLP通往任意平台的通用路径目录中数量最多的示例*_via_openinference.py展示了同一条通用模式用 OpenInference 的AgnoInstrumentor对 Agno 插桩用 OpenTelemetry 的 OTLP HTTP Exporter 把 span 推送到目标平台。掌握这条路径就等于掌握了接入任何兼容 OTLP 后端的能力。以 langfuse_via_openinference.py 为例完整的插桩代码为from openinference.instrumentation.agno import AgnoInstrumentor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor tracer_provider TracerProvider() tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter())) # 开始插桩 agno AgnoInstrumentor().instrument(tracer_providertracer_provider)这段代码完成了三件事创建TracerProvider追踪提供者挂载SimpleSpanProcessor与OTLPSpanExporter把 span 序列化为 OTLP 协议并通过 HTTP 上报调用AgnoInstrumentor().instrument(...)对 Agno 的 Agent/Team/Workflow 运行过程自动插桩。各平台的差异集中在导出目标地址与鉴权头上下面按平台逐一拆解。3.1 LangfuseLangfuse 的鉴权采用 OTLP 标准 Basic Auth把LANGFUSE_PUBLIC_KEY与LANGFUSE_SECRET_KEY拼接后做 Base64 编码写入OTEL_EXPORTER_OTLP_HEADERSLANGFUSE_AUTH base64.b64encode( f{os.getenv(LANGFUSE_PUBLIC_KEY)}:{os.getenv(LANGFUSE_SECRET_KEY)}.encode() ).decode() os.environ[OTEL_EXPORTER_OTLP_ENDPOINT] https://cloud.langfuse.com/api/public/otel # EU 区域 os.environ[OTEL_EXPORTER_OTLP_HEADERS] fAuthorizationBasic {LANGFUSE_AUTH}示例中注释给出了三种可选端点按部署区域切换即可部署形态端点US 数据区域https://us.cloud.langfuse.com/api/public/otelEU 数据区域https://cloud.langfuse.com/api/public/otel本地部署 v3.22.0http://localhost:3000/api/public/otelAgent 侧可以是带工具与指令的常规 Agent示例中通过aprint_response(..., streamTrue)展示了异步 流式调用同样被完整追踪。3.2 Arize PhoenixPhoenix 提供了官方 helper 函数phoenix.otel.register()来简化配置见 arize_phoenix_via_openinference.pyos.environ[PHOENIX_API_KEY] os.getenv(PHOENIX_API_KEY) os.environ[PHOENIX_COLLECTOR_ENDPOINT] https://app.phoenix.arize.com/ # 追加你的组织后缀 tracer_provider register( project_namedefault, # 默认项目名为 default auto_instrumentTrue, # 自动使用已安装的 OpenInference 插桩 )其中project_name指定 trace 归属的 Phoenix 项目auto_instrumentTrue表示自动启用已安装的 OpenInference 插桩等价于手动执行AgnoInstrumentor().instrument()。本地部署变体arize_phoenix_via_openinference_local.py则把PHOENIX_COLLECTOR_ENDPOINT指向本地服务os.environ[PHOENIX_COLLECTOR_ENDPOINT] http://localhost:6006云端与本地版本的 Agent 都演示了 YFinance 工具 流式响应云端版还额外展示了dbInMemoryDb()、固定session_id、output_schema结构化输出等组合场景——这些都能在 trace 中体现。项目路由高级arize_phoenix_moving_traces_to_different_projects.py 演示了把不同 Agent 的 trace 路由到不同 Phoenix 项目全局只注册一个 tracer providerproject_namedefault运行时用openinference.instrumentation.dangerously_using_project()上下文管理器切换项目with dangerously_using_project(default): await stock_agent.aprint_response(What is the current price of Tesla?, streamTrue) with dangerously_using_project(Testing-agno): await search_agent.aprint_response(What is the latest news about AI?, streamTrue)stock_agent的 trace 进入default项目search_agent的 trace 进入Testing-agno项目。注意该函数名自带dangerously前缀——在并发场景中切换项目可能影响同进程内其他插桩代码的 trace 归属应谨慎用于串行流程。3.3 LangSmithLangSmith 的 OTLP 端点不需要 Basic Auth而是通过自定义 header 携带 API Key 与项目名见 langsmith_via_openinference.pyendpoint https://eu.api.smith.langchain.com/otel/v1/traces headers { x-api-key: os.getenv(LANGSMITH_API_KEY), Langsmith-Project: os.getenv(LANGSMITH_PROJECT), } tracer_provider TracerProvider() tracer_provider.add_span_processor( SimpleSpanProcessor(OTLPSpanExporter(endpointendpoint, headersheaders)) ) AgnoInstrumentor().instrument(tracer_providertracer_provider)这里展示了一种变体不依赖OTEL_EXPORTER_OTLP_ENDPOINT环境变量而是直接把endpoint与headers传给OTLPSpanExporter构造器——适合需要为不同 exporter 配置不同目标的场景。3.4 LogfireLogfire 使用 Write Token 作为鉴权头见 logfire_via_openinference.pyos.environ[OTEL_EXPORTER_OTLP_ENDPOINT] https://logfire-eu.pydantic.dev # EU 区域 os.environ[OTEL_EXPORTER_OTLP_HEADERS] fAuthorization{os.getenv(LOGFIRE_WRITE_TOKEN)}US 区域端点为https://logfire-us.pydantic.dev本地部署为http://localhost:4318。与 Langfuse 示例一样它使用SimpleSpanProcessor即时导出并通过异步流式调用触发追踪。3.5 LatitudeLatitude 的 OTLP 摄取端点会由 exporter 自动追加/v1/traces后缀见 latitude_via_openinference.pyos.environ[OTEL_EXPORTER_OTLP_ENDPOINT] https://ingest.latitude.so os.environ[OTEL_EXPORTER_OTLP_HEADERS] ( fAuthorizationBearer {os.getenv(LATITUDE_API_KEY)}, fX-Latitude-Project{os.getenv(LATITUDE_PROJECT)} )鉴权头同时携带 Bearer Token 与项目标识是组合 header 的又一示例。3.6 MLflowMLflow 提供了两条接入路径。路径一OpenInference OTLP 导出mlflow_via_openinference.py需要pip install -U mlflow opentelemetry-exporter-otlp-proto-http openinference-instrumentation-agno并先启动服务mlflow server --host 127.0.0.1 --port 5000然后配置MLFLOW_TRACKING_URI os.getenv(MLFLOW_TRACKING_URI, http://127.0.0.1:5000) endpoint f{MLFLOW_TRACKING_URI}/api/2.0/mlflow/traces tracer_provider TracerProvider() tracer_provider.add_span_processor( SimpleSpanProcessor( OTLPSpanExporter(endpointendpoint, headers{x-mlflow-experiment-id: 0}) ) ) AgnoInstrumentor().instrument(tracer_providertracer_provider)路径二MLflow 原生 autologmlflow_via_autolog.py无需 OpenInference 插桩只需在创建 Agent 前调用mlflow.set_tracking_uri(http://127.0.0.1:5000) mlflow.set_experiment(Agno Agent) mlflow.agno.autolog() # 开启 Agno 追踪也可以在环境变量中配置MLFLOW_TRACKING_URI与MLFLOW_EXPERIMENT_NAME代替 Python API。注意 autolog 必须在 Agent 创建之前调用。3.7 OpikOpik 通过环境变量OTEL_EXPORTER_OTLP_ENDPOINT/OTEL_EXPORTER_OTLP_HEADERS或 SDK 配置端点示例opik_via_openinference.py还演示了 Agno 原生的trace_attributes参数为 trace 注入自定义上下文agent Agent( nameStock Price Agent, modelOpenAIChat(idgpt-5.2), tools[YFinanceTools()], instructionsYou are a stock price analyst. Answer with concise, well-sourced updates., trace_attributes{ session.id: demo-session-001, environment: development, }, )示例注释明确指出 span 层级结构为agent → model → tool每次请求都会以该层级出现在 Opik 中——这也与前面trace_to_database.py中按 span kind 解析出的AGENT/LLM/TOOL结构相互印证。3.8 通用对照表汇总各平台在 OTLP 路径下的端点与鉴权方式平台端点示例鉴权头关键代码Arize Phoenix云端https://app.phoenix.arize.com/PHOENIX_API_KEYphoenix.otel.register()Arize Phoenix本地http://localhost:6006无phoenix.otel.register()Langfusehttps://us.cloud.langfuse.com/api/public/otelAuthorizationBasic base64(pub:sec)AgnoInstrumentor().instrument()LangSmithhttps://eu.api.smith.langchain.com/otel/v1/tracesx-api-keyLangsmith-Project自定义 header 传 exporterLogfirehttps://logfire-eu.pydantic.devAuthorizationWRITE_TOKENAgnoInstrumentor().instrument()Latitudehttps://ingest.latitude.soAuthorizationBearerX-Latitude-ProjectAgnoInstrumentor().instrument()MLflowhttp://127.0.0.1:5000/api/2.0/mlflow/tracesx-mlflow-experiment-idOTLP 或mlflow.agno.autolog()Opik环境变量配置OTEL_EXPORTER_OTLP_HEADERSAgnoInstrumentor().instrument()trace_attributes四、SDK 一键式集成AgentOps、Langtrace、LangWatch、Weave、Traceloop、Atla、Maxim除了统一的 OpenInference 路径目录还收录了各平台 SDK 的零配置式集成——通常一两行即可完成插桩适合快速上手。4.1 AgentOpsagent_ops.pyimport agentops agentops.init() # 初始化 AgentOps agent Agent(modelOpenAIChat(idgpt-5.6-luna)) response agent.run(Share a 2 sentence horror story)agentops.init()之后模型调用会被自动记录无需任何额外配置。4.2 Langtracelangtrace_op.pyfrom langtrace_python_sdk import langtrace langtrace.init() agent Agent( nameStock Price Agent, modelOpenAIChat(idgpt-5.2), tools[YFinanceTools()], instructionsYou are a stock price agent. Answer questions in the style of a stock analyst., ) agent.print_response(What is the current price of Tesla?)Langtrace 的插桩同样在初始化后自动生效Agent 带 YFinance 工具的调用链会被完整捕获。4.3 LangWatchlangwatch_op.pyimport langwatch from openinference.instrumentation.agno import AgnoInstrumentor langwatch.setup(instrumentors[AgnoInstrumentor()])LangWatch 的setup()直接接受 OpenInference instrumentor 列表把 AgnoInstrumentor 传进去即可完成插桩。4.4 Weaveweave_op.pyWeave 走追踪函数路线先weave.init(agno)初始化项目再用weave.op()装饰器包装 Agent 调用函数被装饰的调用就会产生可观测记录import weave weave.init(agno) agent Agent(modelOpenAIChat(idgpt-5.6-luna), markdownTrue) weave.op() def run(content: str): return agent.run(content) run(Share a 2 sentence horror story)4.5 Tracelooptraceloop_op.pyTraceloop 通过workflow装饰器创建父级 workflow span把 Agent 执行包进自定义业务流程中适合需要为多步骤流程建立统一边界的场景from traceloop.sdk import Traceloop from traceloop.sdk.decorators import workflow Traceloop.init(app_nameagno_workflows) agent Agent(nameAnalysisAgent, modelOpenAIChat(idgpt-5.2)) workflow(namedata_analysis_pipeline) def analyze_data(query: str) - str: response agent.run(query) return response.content result analyze_data(Analyze the benefits of observability in AI systems)4.6 Atlaatla_op.pyAtla 提供显式的插桩上下文管理器from atla_insights import configure, instrument_agno configure(tokengetenv(ATLA_API_KEY)) with instrument_agno(openai): agent.print_response(What are the latest news about the stock market?)configure(token...)设置 API Keyinstrument_agno(openai)作为上下文管理器包裹运行区间只在with块内生效。4.7 Maximmaxim_ops.pyMaxim 的集成更偏平台级——它不仅能追踪 Agent还能追踪 Team示例构建了一个 Web Search Agent Finance Agent 的金融分析 Team并在交互式对话循环中持续运行try: from maxim import Maxim from maxim.logger.agno import instrument_agno except ImportError: raise ImportError(maxim not installed. Please install using uv pip install maxim-py) instrument_agno(Maxim().logger())之后创建的两个 Agent 与一个 TeamTeam(members[web_search_agent, finance_agent], ...)都会被自动追踪示例以命令行对话方式循环接受用户输入并调用团队响应。4.8 通过 OpenLIT 中转langfuse_via_openlit.py除了直接对接 Langfuse还可以经由 OpenLIT 中转先用 OpenTelemetry 配置 exporter 指向 Langfuse再openlit.init(tracertracer, disable_batchTrue)完成 OpenLIT 插桩trace_provider TracerProvider() trace_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter())) trace.set_tracer_provider(trace_provider) tracer trace.get_tracer(__name__) import openlit openlit.init(tracertracer, disable_batchTrue) # disable_batchTrue 即时处理 tracedisable_batchTrue让 trace 立即处理而不是批量排队便于演示与调试。该示例的 Agent 使用WebSearchTools并开启markdownTrue输出。五、进阶场景结构化输出、Team 与 Workflow 追踪5.1 结构化输出追踪langfuse_via_openinference_response_model.py 演示了追踪 结构化输出的组合Agent 配置output_schemaStockPrice其中StockPrice是带Field(description...)的 Pydantic 模型还包含MarketArea枚举字段。Langfuse 会记录模型调用、工具调用以及最终的结构化响应方便在平台上直接审计 Agent 是否按 schema 返回。5.2 Team多智能体追踪teams/langfuse_via_openinference_team.py 展示了团队级追踪且同时覆盖同步与异步两条运行路径financial_team Team( nameFinancial Analysis Team, idstr(uuid4()), user_idstr(uuid4()), modelOpenAIChat(idgpt-4.1), members[market_data_agent, news_agent], instructions[...], show_members_responsesTrue, markdownTrue, )market_data_agent使用YFinanceTools()负责行情数据news_agent使用WebSearchTools()负责新闻检索团队通过run_mode变量在同步print_response与异步aprint_response之间切换。Langfuse 上可以看到团队编排 span 与成员 Agent 的完整调用层级uuid4()动态生成的id/user_id可用于区分每次运行的归属。5.3 Workflow工作流追踪workflows 子目录包含两个等价示例分别把多步骤工作流 trace 发送到 Arize Phoenixarize_phoenix_via_openinference_workflow.py与 Langfuselangfuse_via_openinference_workflows.py。两者构造了同一个研究→总结→条件分支事实核查→成稿的线性工作流basic_workflow Workflow( nameBasic Linear Workflow, descriptionResearch - Summarize - Condition(Fact Check) - Write Article, steps[ research_step, summarize_step, Condition( namefact_check_condition, descriptionCheck if fact-checking is needed, evaluatorneeds_fact_checking, steps[fact_check_step], ), write_article, ], )工作流由 4 个 Agent 步骤Step(name..., agent...)与 1 个Condition条件分支组成其中needs_fact_checking(step_input: StepInput) - bool作为分支求值器。Phoenix 版本通过PHOENIX_CLIENT_HEADERSapi_key...环境变量携带密钥Langfuse 版本则沿用 Basic Auth 方案。运行后观测平台上会呈现出完整的步骤级 span 链每一步的 Agent 调用都嵌套其中。六、运行示例所有示例遵循统一的运行方式沿用仓库 README 中的虚拟环境约定.venvs/demo/bin/python cookbook/observability/file.py运行单 Agent 示例.venvs/demo/bin/python cookbook/observability/langfuse_via_openinference.py运行 Team 示例.venvs/demo/bin/python cookbook/observability/teams/langfuse_via_openinference_team.py运行 Workflow 示例.venvs/demo/bin/python cookbook/observability/workflows/langfuse_via_openinference_workflows.py大多数第三方平台示例依赖环境变量如LANGFUSE_PUBLIC_KEY、LANGFUSE_SECRET_KEY、PHOENIX_API_KEY、LANGSMITH_API_KEY、LOGFIRE_WRITE_TOKEN、LATITUDE_API_KEY、ATLA_API_KEY等运行前需按所选平台配置对应密钥trace_to_database.py与本地 Phoenix 示例则无需外部凭据。目录结构校验可通过仓库脚本完成.venvs/demo/bin/python cookbook/scripts/check_cookbook_pattern.py --base-dir cookbook/observability --recursive该脚本见 check_cookbook_pattern.py用于校验示例的模块 docstring、分区横幅、create/run 顺序、__main__入口守卫等编码规范。七、总结与选型建议围绕 cookbook/observability/README.md 列出的全部示例Agno 的可观测性体系可以归纳为三个层次数据库层setup_tracing(db...)Trace/Span两表模型trace 与业务数据同库存储适合需要自建观测存储或离线审计的场景OTLP 平台层AgnoInstrumentorOTLPSpanExporter的标准组合一次插桩可对接 Arize Phoenix、Langfuse、LangSmith、Logfire、Latitude、MLflow、Opik 等任一 OTLP 兼容后端切换平台只需修改端点与鉴权头SDK 快速层AgentOps、Langtrace、LangWatch、Weave、Traceloop、Atla、Maxim 与 OpenLIT 等提供一行式插桩或装饰器封装适合快速验证部分平台如 Maxim原生支持 Team 级追踪。选型建议若追求平台能力trace 检索、评估、成本分析优先走 OTLP 平台层并配合trace_attributes注入业务上下文若强调数据自主可控使用内置数据库追踪多智能体与多步骤工作流场景下可参照 Team/Workflow 示例确认目标平台对嵌套 span 层级的展示效果。无论选择哪条路径底层的 OpenTelemetry 基础设施都是一致的这也意味着生产环境可以根据需要随时在多个后端之间切换或并行导出。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考