当多个 AI 代理共享搜索 API 密钥时,成本可能会急剧上升,而无法了解哪个代理花费了什么。本教程构建了一个集中式成本跟踪器,用代理 ID 标记每个搜索呼叫,按代理和时间段聚合支出,并在任何代理超出其预算时发出警报。可与任何规模的任何搜索提供商合作。
前置条件
- 已安装 Python 3.9+
- 请求已安装库
- 来自 scavio.dev 的 Scavio API 密钥
操作指南
步骤 1: 构建集中式成本跟踪器
创建一个跟踪器,拦截所有搜索呼叫,用代理 ID 标记它们,并记录成本数据。
Python
import time, json
from datetime import datetime, timedelta
from collections import defaultdict
from typing import Dict, List
class AgentCostTracker:
def __init__(self):
self.records: List[dict] = []
self.budgets: Dict[str, float] = {}
self.alerts: List[str] = []
def set_budget(self, agent_id: str, monthly_budget: float):
self.budgets[agent_id] = monthly_budget
def record(self, agent_id: str, cost: float, query: str, provider: str = 'scavio'):
self.records.append({
'agent_id': agent_id,
'cost': cost,
'query': query[:100],
'provider': provider,
'timestamp': datetime.now().isoformat()
})
# Check budget
monthly_spend = self.get_monthly_spend(agent_id)
budget = self.budgets.get(agent_id, float('inf'))
if monthly_spend > budget * 0.8:
alert = f'Agent {agent_id}: {monthly_spend/budget*100:.0f}% of budget used'
if alert not in self.alerts:
self.alerts.append(alert)
print(f'ALERT: {alert}')
def get_monthly_spend(self, agent_id: str) -> float:
month_start = datetime.now().replace(day=1, hour=0, minute=0, second=0)
return sum(r['cost'] for r in self.records
if r['agent_id'] == agent_id and r['timestamp'] >= month_start.isoformat())
tracker = AgentCostTracker()
tracker.set_budget('research-bot', 5.00)
tracker.set_budget('lead-gen-bot', 10.00)
tracker.set_budget('monitor-bot', 2.00)
print('Cost tracker initialized with 3 agent budgets')步骤 2: 为每个代理创建跟踪搜索功能
使用自动成本跟踪包装搜索 API 调用。每个代理都有自己的搜索功能来记录使用情况。
Python
import requests, os
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
COST_PER_QUERY = 0.005
def make_tracked_search(agent_id: str, tracker: AgentCostTracker):
def search(query: str, count: int = 10) -> list:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': count})
results = resp.json().get('organic_results', [])
tracker.record(agent_id, COST_PER_QUERY, query)
return [{'title': r['title'], 'link': r['link']}
for r in results]
return search
# Create per-agent search functions
research_search = make_tracked_search('research-bot', tracker)
lead_gen_search = make_tracked_search('lead-gen-bot', tracker)
monitor_search = make_tracked_search('monitor-bot', tracker)
# Simulate agent activity
research_search('python best practices 2026')
research_search('react server components guide')
lead_gen_search('plumber Austin TX')
lead_gen_search('electrician Denver CO')
lead_gen_search('HVAC contractor Portland OR')
monitor_search('scavio brand mentions')
print(f'Total records: {len(tracker.records)}')步骤 3: 生成成本仪表板报告
构建一个仪表板,显示每个代理的支出、预算利用率和成本趋势。
Python
def dashboard(tracker: AgentCostTracker):
print('\n' + '=' * 60)
print(' Search API Cost Dashboard')
print('=' * 60)
agents = set(r['agent_id'] for r in tracker.records)
total_cost = sum(r['cost'] for r in tracker.records)
total_queries = len(tracker.records)
print(f'\n Total queries: {total_queries}')
print(f' Total cost: ${total_cost:.3f}')
print(f' Agents tracked: {len(agents)}')
print(f'\n {"Agent":<20} {"Queries":>8} {"Spent":>8} {"Budget":>8} {"Used":>6}')
print(' ' + '-' * 54)
for agent_id in sorted(agents):
agent_records = [r for r in tracker.records if r['agent_id'] == agent_id]
spent = sum(r['cost'] for r in agent_records)
budget = tracker.budgets.get(agent_id, 0)
pct = (spent / budget * 100) if budget > 0 else 0
queries = len(agent_records)
print(f' {agent_id:<20} {queries:>8} ${spent:>6.3f} ${budget:>6.2f} {pct:>5.1f}%')
if tracker.alerts:
print(f'\n Alerts:')
for alert in tracker.alerts:
print(f' {alert}')
print(f'\n Avg cost/query: ${total_cost/total_queries:.4f}' if total_queries > 0 else '')
print('=' * 60)
dashboard(tracker)Python 示例
Python
import requests, os
from collections import defaultdict
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
spending = defaultdict(float)
counts = defaultdict(int)
def tracked_search(agent_id, query, cost=0.005):
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': 10})
spending[agent_id] += cost
counts[agent_id] += 1
return resp.json().get('organic_results', [])
tracked_search('research-bot', 'python frameworks 2026')
tracked_search('lead-gen', 'plumber Austin TX')
tracked_search('lead-gen', 'electrician Denver CO')
for agent in spending:
print(f'{agent}: {counts[agent]} queries, ${spending[agent]:.3f}')JavaScript 示例
JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
const spending = {};
const counts = {};
async function trackedSearch(agentId, query, cost = 0.005) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us', num_results: 10 })
});
spending[agentId] = (spending[agentId] || 0) + cost;
counts[agentId] = (counts[agentId] || 0) + 1;
return (await resp.json()).organic_results || [];
}
await trackedSearch('research-bot', 'python frameworks 2026');
await trackedSearch('lead-gen', 'plumber Austin TX');
Object.keys(spending).forEach(a => console.log(`${a}: ${counts[a]} queries, $${spending[a].toFixed(3)}`));预期输出
JSON
Cost tracker initialized with 3 agent budgets
Total records: 6
============================================================
Search API Cost Dashboard
============================================================
Total queries: 6
Total cost: $0.030
Agents tracked: 3
Agent Queries Spent Budget Used
------------------------------------------------------
lead-gen-bot 3 $ 0.015 $ 10.00 0.2%
monitor-bot 1 $ 0.005 $ 2.00 0.3%
research-bot 2 $ 0.010 $ 5.00 0.2%
Avg cost/query: $0.0050
============================================================