Resumen
Wraps agent sesiones con un credit-tracking layer ese registros search API spend per ejecutar, agrega diario totals, y envia alertas cuando un sesion o diario budget umbral es breached.
Desencadenador
After cada agent ejecutar completes (post-run hook o wrapper function)
Programación
After cada agent ejecutar (event-driven, no programado)
Pasos del flujo de trabajo
Initialize sesion credit tracker
Crear un sesion objeto con iniciar marca de tiempo, session_id, y credit_count = 0 antes de el agent inicia.
Intercept search herramienta calls
Wrap el search llamada un API function to increment credit_count by 1 (o by el actual credits_used campo de el respuesta de API) on cada call.
Log sesion resumen to base de datos
After el agent ejecutar completes (success o failure), escribir session_id, credits_used, marca de tiempo, agent_type, y task_description to un SQLite o PostgreSQL sesiones tabla.
Verificar contra umbrales
Comparar sesion credits_used contra per-session limite (e.g., 50 credits) y diario total contra diario limite (e.g., 500 credits). Consulta diario agregar de el sesiones tabla.
Fire alerta if umbral breached
Enviar un webhook o correo electronico alerta con sesion detalles, credits usado, y remaining diario budget cuando cualquiera umbral es exceeded.
Implementacion en Python
import sqlite3
import time
import requests
from datetime import date
from functools import wraps
DB_PATH = "agent_costs.db"
SCRAPING_API_BASE = "https://api.scavio.dev/api/v1/search"
SCRAVIO_API_KEY = "YOUR_API_KEY"
SESSION_LIMIT = 50 # credits per session
DAILY_LIMIT = 500 # credits per day
ALERT_WEBHOOK = "https://hooks.slack.com/services/YOUR/WEBHOOK"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
credits_used INTEGER,
agent_type TEXT,
task TEXT,
ts TEXT
)
""")
conn.commit()
return conn
class SearchCostTracker:
def __init__(self, session_id: str, agent_type: str, task: str):
self.session_id = session_id
self.agent_type = agent_type
self.task = task
self.credits_used = 0
self.conn = init_db()
def search(self, query: str, platform: str = "google") -> dict:
if self.credits_used >= SESSION_LIMIT:
raise RuntimeError(f"Session credit limit {SESSION_LIMIT} reached")
resp = requests.post(
SCRAPING_API_BASE,
headers={"x-api-key": SCRAVIO_API_KEY},
json={"query": query, "platform": platform}
)
resp.raise_for_status()
self.credits_used += 1
return resp.json()
def finalize(self):
today = date.today().isoformat()
self.conn.execute(
"INSERT INTO sessions VALUES (?, ?, ?, ?, ?)",
(self.session_id, self.credits_used, self.agent_type, self.task, today)
)
self.conn.commit()
# Check daily total
row = self.conn.execute(
"SELECT SUM(credits_used) FROM sessions WHERE ts = ?", (today,)
).fetchone()
daily_total = row[0] or 0
if self.credits_used > SESSION_LIMIT * 0.8 or daily_total > DAILY_LIMIT:
requests.post(ALERT_WEBHOOK, json={
"text": f"Search cost alert: session={self.credits_used} credits, daily={daily_total}/{DAILY_LIMIT}"
})
# Usage
tracker = SearchCostTracker("sess_001", "research_agent", "competitor pricing")
try:
result = tracker.search("competitor product pricing 2026")
# ... agent uses result ...
finally:
tracker.finalize()
Implementacion en JavaScript
const Database = require('better-sqlite3');
const fetch = require('node-fetch');
const DB_PATH = 'agent_costs.db';
const API_BASE = 'https://api.scavio.dev/api/v1/search';
const API_KEY = 'YOUR_API_KEY';
const SESSION_LIMIT = 50;
const DAILY_LIMIT = 500;
const ALERT_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK';
const db = new Database(DB_PATH);
db.prepare(`CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY, credits_used INTEGER,
agent_type TEXT, task TEXT, ts TEXT
)`).run();
class SearchCostTracker {
constructor(sessionId, agentType, task) {
this.sessionId = sessionId;
this.agentType = agentType;
this.task = task;
this.creditsUsed = 0;
}
async search(query, platform = 'google') {
if (this.creditsUsed >= SESSION_LIMIT) throw new Error('Session credit limit reached');
const res = await fetch(API_BASE, {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, platform })
});
if (!res.ok) throw new Error(`Search failed: ${res.status}`);
this.creditsUsed++;
return res.json();
}
async finalize() {
const today = new Date().toISOString().slice(0, 10);
db.prepare('INSERT INTO sessions VALUES (?,?,?,?,?)').run(
this.sessionId, this.creditsUsed, this.agentType, this.task, today
);
const { dailyTotal } = db.prepare(
'SELECT SUM(credits_used) as dailyTotal FROM sessions WHERE ts = ?'
).get(today);
if (this.creditsUsed > SESSION_LIMIT * 0.8 || dailyTotal > DAILY_LIMIT) {
await fetch(ALERT_WEBHOOK, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ text: `Search cost alert: session=${this.creditsUsed}, daily=${dailyTotal}/${DAILY_LIMIT}` })
});
}
}
}
// Usage
const tracker = new SearchCostTracker('sess_001', 'research_agent', 'competitor pricing');
(async () => {
try {
const result = await tracker.search('competitor product pricing 2026');
console.log(result);
} finally {
await tracker.finalize();
}
})();
Plataformas utilizadas
Búsqueda web con grafo de conocimiento, PAA y resúmenes de IA