Cómo Monitorear Costos de Agentes IA sin Configuración

#python #ai #tutorial #opensource

Strands Agents proporciona telemetría nativa y seguimiento de costos desde el primer momento. Deja de escribir contadores de tokens personalizados.

Construir agentes de IA es fácil. Desplegarlos a producción es donde la mayoría de los equipos se encuentran con un muro.

Una de las primeras preguntas de finanzas: "¿Cuánto costará esto por solicitud?"

La mayoría de los frameworks de agentes te obligan a construir tu propio contador de tokens. Strands Agents te proporciona uno.

El Problema con el Conteo Personalizado de Tokens

Cada aplicación de IA necesita monitoreo de costos. Pero rastrear tokens a través de:

Múltiples llamadas al modelo
Invocaciones de herramientas
Caché de prompts
Flujos de trabajo multi-agente

...requiere infraestructura personalizada que la mayoría de los equipos reconstruyen desde cero.

Telemetría Nativa en Strands Agents

Strands Agents incluye telemetría de grado de producción por defecto:

from strands import Agent
from strands_tools import calculator

# Crear un agente con herramientas
agent = Agent(tools=[calculator])

# Invocar el agente con un prompt y obtener un AgentResult
result = agent("¿Cuál es la raíz cuadrada de 144?")

# Acceder a métricas a través del AgentResult
print(f"Tokens totales: {result.metrics.accumulated_usage['totalTokens']}")
print(f"Tiempo de ejecución: {sum(result.metrics.cycle_durations):.2f} segundos")
print(f"Herramientas usadas: {list(result.metrics.tool_metrics.keys())}")

# Métricas de caché (cuando estén disponibles)
if 'cacheReadInputTokens' in result.metrics.accumulated_usage:
    print(f"Tokens leídos de caché: {result.metrics.accumulated_usage['cacheReadInputTokens']}")
if 'cacheWriteInputTokens' in result.metrics.accumulated_usage:
    print(f"Tokens escritos en caché: {result.metrics.accumulated_usage['cacheWriteInputTokens']}")

Sin configuración. Sin código personalizado. Simplemente funciona.

Lo Que Obtienes

Cada AgentResult incluye:

Métrica	Descripción
`inputTokens`	Tokens enviados al modelo
`outputTokens`	Tokens generados por el modelo
`totalTokens`	Costo total (entrada + salida)
`cacheReadInputTokens`	Tokens leídos desde caché (caché de prompts de Bedrock)
`cacheWriteInputTokens`	Tokens escritos en caché

Seguimiento de Tokens Multi-Agente

Para sistemas multi-agente (ejecutor → validador → crítico), agrega métricas a través de todos los agentes:

from strands.multiagent import Swarm

swarm = Swarm([executor, validator, critic])
result = swarm("Consulta")

total_tokens = 0
for node_result in result.results.values():
    usage = node_result.result.metrics.accumulated_usage
    total_tokens += usage['totalTokens']

print(f"Costo total a través de todos los agentes: {total_tokens} tokens")

Seguimiento por Ciclo

Para agentes que ejecutan múltiples ciclos de razonamiento, rastrea tokens por ciclo:

from strands import Agent
from strands_tools import calculator

agent = Agent(tools=[calculator])

# Primera invocación
result1 = agent("¿Cuánto es 5 + 3?")

# Segunda invocación
result2 = agent("¿Cuál es la raíz cuadrada de 144?")

# Acceder a métricas de la última invocación
latest_invocation = result2.metrics.latest_agent_invocation
cycles = latest_invocation.cycles
usage = latest_invocation.usage

# O acceder a todas las invocaciones
for invocation in response.metrics.agent_invocations:
    print(f"Uso de invocación: {invocation.usage}")
    for cycle in invocation.cycles:
        print(f"  Ciclo {cycle.event_loop_cycle_id}: {cycle.usage}")

# O imprimir el resumen (incluye todas las invocaciones)
print(result2.metrics.get_summary())