跳到主要内容

AgentOps — 架构与原理

30 秒导读： AgentOps 是给 AI agent 用的“黑匣子记录仪”。你在程序开头写一行 agentops.init()，之后你的 agent 调了哪个 LLM、花了多少 token、调了哪个工具、哪一步报了错——都会被自动记录成一棵“调用树”，上传到 dashboard 重放。它的底层是业界标准 OpenTelemetry。

1. 这是什么（零基础也能懂）​

  • 一句话定义： AgentOps 是一个 AI agent 的可观测性（observability）SDK——把你的 agent 跑起来的每一步（LLM 调用、工具调用、函数执行）记录成可重放、可计费、可调试的轨迹。

  • 解决什么问题 / 给谁用： 假设你写了一个用 CrewAI 或 OpenAI 的 agent，它在生产环境里“时不时抽风”。你想知道：它刚刚那次调用里到底给模型发了什么 prompt？哪个工具报错了？这一整个 session 花了多少钱？AgentOps 就是来回答这些问题的。

  • 它能做什么：

    • 自动拦截几十种 LLM provider（OpenAI、Anthropic、Google…）和 agent 框架（CrewAI、AutoGen、LangGraph…）的调用。
    • 用装饰器（@agent、@tool、@task）标记你自己的函数，让它们也进入调用树。
    • 跟踪 token 用量与成本。
    • 提供一个 validate_trace_spans()，可以在测试里断言“这次跑确实产生了 LLM 活动”。

  • 用起来什么样： 最小例子就两行：

import agentops
agentops.init("<YOUR_API_KEY>")   # 之后你代码里的 openai.chat.completions.create(...) 会被自动记录

# 想把你自己的函数也画进调用树，加装饰器即可：
@agentops.agent
class Researcher:
    @agentops.tool(cost=0.01)
    def search(self, q): ...
  • 一句话直觉 / 类比： 把它想成程序界的行车记录仪（黑匣子）：平时静静在后台记，出事了你把带子拉出来逐帧回放。不同的是，它用的“磁带格式”是业界标准 OpenTelemetry trace。

本节不涉及底层代码。记住一句话就够：init 一下，你 agent 跑的每一步都变成一棵可重放的 span 树。

2. 顶层全景（它大概怎么转）​

AgentOps 的一切都围着一个词：span（OpenTelemetry 里的“一次被计时的操作”）。你的整个 session 是一棵 span 树：根是 session span，里面嵌着 agent / task / tool / LLM 调用的子 span。

从代码到 dashboard，数据走这条线（从上到下）：

     你的代码
   ┌───────────────────┐
   │  agentops.init()    │──┐
   └───────────────────┘  │
         │                 │ 启动两条“拦截管道”
         ▼                 ▼
  ┌A装饰器手动标的函数┐  ┌B import 钩子自动拦截的┐
  │ @trace @agent @tool│  │ openai / crewai … │
  └───────┬───────┘  └───────┬───────┘
          └───────┬───────┘
                  ▼ 都生产 span
        ┌───────────────────┐
        │   TracingCore      │  中枢：管 OTel TracerProvider、管活跃 trace
        └───────┬──────────┘
                ▼ span on_end
        ┌─────────────────┐
        │ BatchSpanProcessor │  攒一批、定期刷
        └───────┬─────────┘
                ▼
        ┌───────────────────────┐
        │ AuthenticatedOTLPExporter │  带 JWT 发 HTTP 到 otlp.agentops.ai
        └───────┬──────────────┘
                ▼
          AgentOps 后端 / dashboard

部件一句话职责：

部件干什么在哪个文件
init()公开入口：创建单例 client、启动 tracer、可选启动自动拦截agentops/__init__.py:77
Client单例：读配置、启动 tracer、后台拉 JWT、可选自动起一个 sessionagentops/client/client.py:38
TracingCore中枢：管 OTel provider、start/end trace、跟踪活跃 traceagentops/sdk/core.py:151
装饰器工厂用一个工厂函数生成 @trace/@agent/@tool…agentops/sdk/decorators/factory.py:25
import 钩子接管 builtins.__import__，看到目标包就插探针agentops/instrumentation/__init__.py:375
Provider 探针每个 LLM/框架一个 instrumentor，用 wrapt 包方法agentops/instrumentation/providers/openai/instrumentor.py:53
导出器带动态 JWT 的 OTLP exporter，401/过期会降级agentops/sdk/exporters.py:16
验证回查后端 API，断言 span 数/LLM 活动agentops/validation.py:209

主线走一遍（高层）：

  1. agentops.init(api_key) 创建单例 Client，用环境变量+参数读出 Config。
  2. Client.init() 调 tracer.initialize_from_config(...)：装上 OTel TracerProvider + 两个 span processor（一个批量导出、一个内部打印）+ 带 JWT 的 OTLP exporter。
  3. 若 instrument_llm_calls=True（默认），调 instrument_all() 接管 __import__，并扫一遍已 import 的模块补探针。
  4. 若 auto_start_session=True（默认），同时后台起一个异步线程拿 API key 换 JWT。
  5. 你的代码跑起来：每个装饰器/被拦截的调用都 start_span，结束时 on_end 进入 BatchSpanProcessor。
  6. processor 阶段性刷，exporter 拿最新 JWT 作 Bearer 头发 HTTP 到 otlp.agentops.ai。
  7. 程序退出时 atexit 把没结束的 trace 收尾、force flush。

3. 阅读地图（建议顺序）​

这是一个多子系统项目，拆成了四章，按“从中枢向外”的顺序读最顺：

顺序章节讲什么何时该读
101-tracing-coreTracingCore、trace=根 session span、start/end/finalize 生命周期想懂“一棵 span 树是怎么被创建和结束的”
202-decorators装饰器工厂：一个函数生成所有装饰器，同步/异步/生成器/类分流想用 @agent/@tool 标自己的代码
303-auto-instrumentationimport 钩子、wrapt 包方法、provider↔框架冲突解决想懂“为什么不改代码就能拦 OpenAI”
404-export-auth-validation导出、动态 JWT、降级、回查后端做 eval想懂“数据怎么安全发出去 / 怎么验证”

4. 巧妙之处（概览，详见各章）​

  • 把“会话”直接映射成 OTel 的根 span，而不是另造一套会话模型——整个 SDK 因此天生兼容任何 OTLP 后端。详见 §01。
  • 一个 create_entity_decorator(kind) 工厂生成全部装饰器，连同步/异步/生成器/类四种函数形态都在同一处分流。详见 §02。
  • 用 import 钩子做“零侵入”拦截：拦 builtins.__import__，你 import openai 的那一刻探针就装上了。详见 §03。
  • provider 与框架冲突的优先级规则：一旦检测到 agent 框架（如 CrewAI），就把底层 provider 探针卸掉，避免重复记录。详见 §03。
  • 动态 JWT exporter：token 后台异步拿到后不用重建 exporter，每次 export 现拿最新 token；401/过期会进入 60 秒冷却。详见 §04。

5. 边界与局限​

  • 不是评测框架：它重点是“记录+重放+计费”，不是跨样本打分。唯一带“eval 味”的是 validate_trace_spans()，也只是回查后端做存在性/LLM-活动断言（主要给测试用）。详见 §04。
  • 一个进程只能有一个“主框架”：多个 agent 框架同时在场时，只有第一个被拦截，之后的框架和 provider 都被跳过（_should_instrument_package）。
  • import 钩子是进程级副作用：它替换了 builtins.__import__，这是个全局动作。
  • 不拿 API key 也能跑：span 照样创建，只是 export 会优雅失败（适合本地调试）。

6. 横向对比（同 shelf 兄弟）​

AgentOps 在 evals-observability 区，它代表的是“生产级、标准协议（OTel）、多框架自动拦截”这一派。与它可对比的取舍：

  • 与纯 eval 框架（如跨样本打分类）相比：AgentOps 不管“答得对不对”，只管“发生了什么”。两者互补。
  • 与自建日志/回调方案相比：AgentOps 赌“不修改用户代码”，用 import 钩子 + wrapt 达成零侵入。

7. 代码地图（导航索引）​

主题文件符号名
公开入口agentops/__init__.pyinit / start_trace / end_trace / update_trace_metadata
单例 clientagentops/client/client.pyClient / Client.init / _fetch_auth_async
中枢 traceragentops/sdk/core.pyTracingCore / TraceContext / setup_telemetry
装饰器工厂agentops/sdk/decorators/factory.pycreate_entity_decorator
装饰器实例agentops/sdk/decorators/__init__.pytrace / agent / task / tool / guardrail
import 钩子agentops/instrumentation/__init__.pyinstrument_all / _import_monitor / _should_instrument_package
探针基类agentops/instrumentation/common/instrumentor.pyCommonInstrumentor
wrapt 包装agentops/instrumentation/common/wrappers.pyWrapConfig / wrap / _create_wrapper
导出器agentops/sdk/exporters.pyAuthenticatedOTLPExporter
验证agentops/validation.pyvalidate_trace_spans / check_llm_spans