AgentPulse (Py):用 Python 为你的 AI Agent 装上“算账雷达”

pip install agentpulse

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()

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()

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()

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()

@trace(    name="custom-trace",    metadata={"version": "1.2.0", "env": "staging"})def customized_agent():    ...@tool(metadata={"type": "database"})def query_db(sql: str):    ...

# 首次启动 Dashboard（需要安装并运行 AgentPulse Dashboard 镜像）docker run -p 3000:3000 agentpulse/dashboard:latest