1. 这是什么？一句话定位
AgentPulse (Py) 是一款轻量级、框架无关的 Python 可观测性库。想象一下，你的 AI Agent 在生产环境里狂奔，API 调用费像流水一样花出去，但你可能根本不知道：
每个请求到底消耗了多少 Token？
每次调用延迟多少毫秒？
有没有连续失败、哪里在悄悄烧钱？
AgentPulse 就是用来回答这些问题的——它零侵入地记录每一次 LLM 调用的成本、Token 用量、调用链与耗时，并且无需修改你的业务代码。
对比：AgentPulse (Py) 是专门为 Python 生态设计的库级工具，而非 Avalpoint、Rectify 等企业级的 Agent 治理平台——后者更多面向多代理统一管控，而 AgentPulse (Py) 专注于“开发者本机可观测”。
2. 安装
3. 整体工作流程
在深入代码之前，先看 AgentPulse 的监控闭环是怎么跑的：
4. 核心概念速览
AgentPulse 客户端 负责连接 Dashboard、发送监控数据的核心对象
@tool 装饰器 标记一个函数为 Agent 调用的“外部工具”，自动追溯
@trace 装饰器 标记一段业务逻辑为一条“可追踪的调用链”
ap.shutdown() 确保所有数据最终发送、连接安全关闭
5. 实战代码示例
5.1 最小可运行例子
from agentpulse import AgentPulse, trace, tool# 1️⃣ 初始化客户端ap = AgentPulse( endpoint="http://localhost:3000", # Dashboard 地址 api_key="ap_dev_default" # 开发环境默认密钥)# 2️⃣ 定义一个"工具"——Agent 用来干具体活的函数@tooldef search(query: str) -> str: """模拟一次网络搜索""" return f"Results for {query}"# 3️⃣ 定义一个"追踪"——Agent 的主流程@trace(name="research-agent")def run_agent(topic: str) -> str: """模拟一个会调用工具的研究型 Agent""" # 这行代码会被自动记录为一次工具调用 result = search(topic) return f"Research on '{topic}' completed: {result}"# 4️⃣ 执行并观察output = run_agent("quantum computing")print(output)# 5️⃣ 优雅关闭，确保数据完全发送ap.shutdown()
运行这段代码后，你会看到 Dashboard 中自动出现一条名为 research-agent 的 Trace，里面包含一次 search 工具调用，并且清晰地展示出 Token 用量和延迟。
5.2 多工具协作 + 嵌套调用
真实的 Agent 往往不会只调一个工具。下面的例子展示 AgentPulse 如何自动记录多个工具的执行情况：
import timeimport randomfrom agentpulse import AgentPulse, trace, toolap = AgentPulse( endpoint="http://localhost:3000", api_key="ap_dev_default")@tooldef fetch_weather(city: str) -> str: """获取天气（模拟网络请求）""" time.sleep(random.uniform(0.2, 0.5)) # 模拟网络延迟 return f"Weather in {city}: Sunny, 22°C"@tooldef fetch_news(topic: str) -> str: """获取新闻（模拟网络请求）""" time.sleep(random.uniform(0.3, 0.7)) return f"Latest news about {topic}: Some headlines..."@trace(name="advanced-assistant")def advanced_agent(user_query: str) -> str: """组合多个工具完成复杂任务""" # 工具1：查天气 weather_info = fetch_weather("London") # 工具2：查新闻 news_info = fetch_news("AI") # 可嵌套的简单处理（同样被追踪） summary = f"Summary: {weather_info} | {news_info}" return summaryprint(advanced_agent("What's happening today?"))ap.shutdown()
效果：在 Dashboard 中，advanced-assistant 这条 Trace 下会包含两条子 Span——分别标记为 fetch_weather 和 fetch_news，你可以很直观地看到每个步骤的耗时占比。
5.3 自定义追踪 + 异步工具
对于复杂的异步场景（例如 Agent 需要同时调用多个服务），你可以手动创建 Span，精细控制追踪粒度：
import asyncioimport randomfrom agentpulse import AgentPulse, traceap = AgentPulse( endpoint="http://localhost:3000", api_key="ap_dev_default")async def async_llm_call(model: str, prompt: str) -> str: """模拟异步 LLM 调用""" await asyncio.sleep(random.uniform(0.5, 1.0)) return f"Response from {model}: ..."@trace(name="multi-step-agent")async def execute_agent(query: str) -> str: """分阶段执行，每阶段独立追踪""" # 阶段1：使用 with 语法手动创建 Span with ap.span(name="phase-1-understanding", metadata={"query": query}): await asyncio.sleep(0.1) # 模拟分析 context = "Understood query context" # 阶段2：并行调用两个 LLM with ap.span(name="phase-2-llm-calls"): tasks = [ async_llm_call("gpt-4", query), async_llm_call("claude-3", query) ] results = await asyncio.gather(*tasks) # 阶段3：结果合并 with ap.span(name="phase-3-finalize"): final = f"Q: {query}\nContext: {context}\nResults: {results}" return final# 异步入口result = asyncio.run(execute_agent("How to learn Rust?"))print(result)ap.shutdown()
关键点：ap.span() 是手动创建 Span 的上下文管理器——你可以在任何地方插入它，AgentPulse 会自动将其纳入当前 Trace 的调用树中。
5.4 错误监控与成本追踪
生产环境里，API 失败在所难免。AgentPulse 会自动捕获这些异常，并在 Dashboard 中高亮展示：
import randomfrom agentpulse import AgentPulse, trace, toolap = AgentPulse( endpoint="http://localhost:3000", api_key="ap_dev_default")@tooldef risky_api_call(input_text: str) -> str: """模拟有一定概率失败的 API""" if random.random() < 0.3: # 30% 概率失败 raise ConnectionError("API timeout exceeded") return f"Processed: {input_text}"@trace(name="resilience-test")def test_agent(task: str) -> str: try: return risky_api_call(task) except Exception as e: # 🔴 即使你捕获了异常，AgentPulse 也已自动记录 return f"Fallback response (error: {e})"# 执行多次，观察失败率for i in range(5): print(f"Run {i+1}: {test_agent('test data')}")ap.shutdown()
效果：Dashboard 中会显示 5 次 resilience-test 的 Trace，其中一些 Trace 会标记为红色（异常），你可以立即发现失败率和失败原因，甚至看到每次失败对应的具体时间和输入参数
6. 真·流程图：Agent 内调用追踪
下面这张流程图更严格地描述了 AgentPulse 在一次函数调用内部的跟踪传播机制：
AgentPulse 的核心逻辑就是：在装饰器中拦截函数调用 → 自动生成 Span → 汇总成 Trace → 异步发送。你只需要关注业务逻辑本身。
7. 更多高级用法
7.1 自定义 Metadata
你可以给 Trace 和工具打上自定义标签，方便在 Dashboard 中按版本、环境等维度过滤：
@trace( name="custom-trace", metadata={"version": "1.2.0", "env": "staging"})def customized_agent(): ...@tool(metadata={"type": "database"})def query_db(sql: str): ...
7.2 支持 OpenAI / Anthropic 的“零侵入”监控
如果你的项目已经用 openai 或 anthropic 包，AgentPulse 可通过 monkey-patch 机制，在不修改你的调用代码的前提下，自动监控所有 LLM API 调用——仅需在初始化时开启相应插件即可。
7.3 本地 Dashboard
AgentPulse 的 Dashboard 体积小巧，可通过 Docker 一键启动：
# 首次启动 Dashboard（需要安装并运行 AgentPulse Dashboard 镜像）docker run -p 3000:3000 agentpulse/dashboard:latest
或者也可以使用官方提供的 SaaS Dashboard，只需将 endpoint 改为云端地址即可。
8. 总结：为什么选择 AgentPulse (Py)？
零侵入：无需修改现有业务代码。
轻量级：对主流程性能影响极小，数据通过缓冲区异步发送。
全面追踪：从 API 的 Token、成本、耗时到嵌套的工具调用，一次性覆盖。
框架无关：支持 OpenAI、Anthropic 等主流 LLM，也适配你自定义的任何工具函数。
如果你正为 AI Agent 的“烧钱”而烦恼，或者需要更清晰地诊断问题，AgentPulse 值得一试。
10个月宝宝每天需要喝多少奶粉？
