用 AgentOps 给多智能体做会话回放与成本审计：生产环境可观测实操

先在项目里装上 SDK（建议顺带 python-dotenv 管密钥），然后在程序入口初始化一次即可，之后所有受支持的 LLM 调用都会被自动埋点：

# 安装
pip install agentops python-dotenv

# .env（切勿提交到公开仓库）
OPENAI_API_KEY=sk-你的官方模型Key
AGENTOPS_API_KEY=ao-你的AgentOpsKey   # 从 AgentOps Dashboard 获取

import os
import agentops
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
# 一行接入：不传 api_key 时会自动读取环境变量 AGENTOPS_API_KEY
agentops.init(api_key=os.environ["AGENTOPS_API_KEY"])

client = OpenAI()  # 之后所有受支持的 LLM 调用都被自动记录

照常写调用逻辑，结束时用 agentops.end_session(...) 标记会话终态（Success / Fail / Indeterminate）。这是结算成本与归类失败的关键一步：

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "用一句话解释什么是多智能体系统。"}],
)
print(resp.choices[0].message.content)

agentops.end_session("Success")  # 必调：标记会话结局，否则成本可能不结算

运行后回到 AgentOps Dashboard 的 Sessions 页面，应该能看到这次会话——模型、Token、耗时、状态都列出来了。看不到的话，先确认 AGENTOPS_API_KEY 已加载、且 agentops.init() 在调用之前执行。

自动埋点能录调用，但多个 Agent 混在一起时你需要「分身」。用 @agent 标子类、@operation 标方法，AgentOps 就会把类里的 LLM 调用归到对应 Agent 名下，仪表盘按 Agent 拆成本：

from agentops.sdk.decorators import agent, operation

@agent(name="researcher")          # 子智能体 A
class ResearcherAgent:
    @operation
    def search(self, topic):
        return client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"研究：{topic}"}],
        ).choices[0].message.content

@agent(name="writer")              # 子智能体 B
class WriterAgent:
    @operation
    def write(self, draft):
        return client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"据此写报告：{draft}"}],
        ).choices[0].message.content

r = ResearcherAgent()
w = WriterAgent()
report = w.write(r.search("AgentOps 的会话回放能力"))
agentops.end_session("Success")

在 Dashboard 的 Cost / Sessions 面板里，就能看到 researcher 与 writer 各自的 Token 与花费占比——哪个子 Agent 是「烧钱大户」一目了然。没用装饰器的调用会归入 Default Agent。

Agent 常调用外部工具（搜索、计算、查库）。用 @tool(cost=...) 给工具标单次成本，仪表盘会自动汇总；用 @workflow 把多步逻辑归到一个有名字的流程下，便于在回放里整体查看：

from agentops.sdk.decorators import tool, workflow

@tool(cost=0.05)                  # 标注单次工具成本，Dashboard 自动累加
def web_search(query):
    return f"搜索结果：{query}"

@workflow(name="report_pipeline", tags=["production"])
def report_pipeline(topic):
    data = web_search(topic)      # 工具调用被记录并按 cost 计费
    return w.write(r.search(topic))

print(report_pipeline("多智能体可观测"))

此外还有 @session（根跨度）、@trace（推荐用于自定义分组）等装饰器。官方建议：大多数应用只需 agentops.init() 即可；需要更细粒度时再用装饰器。

Step 5：在仪表盘做会话回放与失败定位

Sessions   # 会话列表；点开任一会话可逐步回放执行轨迹
Cost       # 按 Agent / 会话 / 工具聚合的花费
Errors     # 异常、解析失败、提示词注入攻击检测

Step 6：自托管（数据留自己机器）+ 生产环境告警

git clone https://github.com/AgentOps-AI/agentops
cd agentops/app
cp .env.example .env        # 填入 Supabase / ClickHouse 等变量
docker compose up -d       # 启动后访问 localhost:3000 即你的私有仪表盘

常见问题速查