Control del bucle de llamadas y de los costes de tokens en agentes de IA personalizados tras la fase de demostración
makedream٢٨ يوليو ٢٠٢٦
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
El proceso de tomar un framework y crear un agente de IA con datos internos de la empresa suele avanzar sin problemas hasta la fase de prototipo. El verdadero problema comienza en el momento en que se despliega este código en un servicio de producción real. Un agente que funcionaba perfectamente en la pantalla puede quedar atrapado por una entrada inesperada, repitiendo llamadas en un bucle infinito y generando facturas de API de miles de dólares durante el fin de semana.
Gran parte de las fallas que ocurren en producción no se deben a las limitaciones del modelo LLM en sí, sino a la falla en la gestión del estado y a la falta de controles externos. Es necesario poner fin a la situación en la que, tras crear una herramienta personalizada por orden de la directiva, se termina perdiendo el tiempo resolviendo alucinaciones del agente que surgen sin previo aviso durante todo el día y gestionando abrumadoras facturas de tokens en lugar de realizar el trabajo principal.
Al salir del entorno de demostración, el agente se topa con tres barreras: operaciones de caja negra, bucles no deterministas y riesgos de fuga de datos.
Si las entradas y salidas de cada etapa —proceso de razonamiento (Chain-of-Thought), llamadas a herramientas externas (Tool Call) y búsquedas en bases de datos vectoriales— no quedan registradas en los logs, no será posible encontrar la causa de los problemas cuando estos ocurran. Solo rastrear en qué punto se enredó el prompt puede tomar un día entero.
Lo más grave es el bucle infinito. Los agentes con arquitectura ReAct (Reasoning + Acting) vuelven a enviar la misma solicitud repetidamente cuando el resultado de la ejecución de una herramienta es ambiguo.
┌─────────────────────────────────────────────────────────────────────────┐ │ ReAct Architecture Loop │ │ │ │ ┌────────────┐ User Query ┌────────────┐ Tool Call Request │ │ │ User │ ──────────────> │ Main LLM │ ────────────────────┐ │ │ └────────────┘ └────────────┘ │ │ │ ▲ ▼ │ │ │ Observation ┌──────┐ │ │ │ (Ambiguous/Failed) │ Tool │ │ │ └─────────────────────── │ A │ │ │ └──────┘ │ │ * Problem: When Observation fails, LLM retries Tool A endlessly. │ └─────────────────────────────────────────────────────────────────────────┘
En el momento en que el sistema no puede salir de la condición de finalización y continúa ejecutando la API con los mismos argumentos, el presupuesto de tokens de todo un mes se agota en cuestión de minutos. Las fallas en cadena, en las que el agente principal y los subagentes se llaman entre sí reintentando la operación, representan más del 30% de las causas de fallas generales del sistema.
También surge el problema de que el rendimiento del modelo disminuye a medida que el historial de conversación se acumula continuamente y llena la ventana de contexto. Además, los prompts codificados como cadenas de texto dentro del código Python obligan a reconstruir y desplegar todo el sistema incluso cuando se modifica una simple frase.
Es necesario separar la solución de depuración local de las herramientas de observabilidad en producción para definir adecuadamente los puntos de recolección de datos. En la etapa de desarrollo, se utiliza Arize Phoenix basado en OpenTelemetry para verificar la calidad de los embeddings de RAG y las llamadas a herramientas. En el entorno de servicio real, se conecta Langfuse utilizando ClickHouse como backend para monitorear en tiempo real los logs de traza y el consumo de tokens.
┌─────────────────────────────────────────────────────────────────────────┐ │ Semantic Caching & Routing Flow │ │ │ │ Client Query │ │ │ │ │ ▼ │ │ ┌───────────┐ Similarity >= 0.92? ┌─────────────────────────────┐ │ │ │ Redis Vector │ ─────────────────────────> │ Return Cached Response │ │ │ │ Cache │ (Cache Hit) │ (Latency -88%, Cost -86%) │ │ │ └───────────┘ └─────────────────────────────┘ │ │ │ │ │ │ (Cache Miss) │ │ ▼ │ │ ┌───────────┐ Execution & Save Cache ┌─────────────────────────────┐ │ │ │ External │ ────────────────────────> │ Store Result as Vector in │ │ │ │ LLM API │ │ Backend Database │ │ │ └───────────┘ └─────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────┘
Los costes de API generados por la repetición de consultas idénticas se bloquean mediante una capa de almacenamiento en caché semántico (Semantic Caching).
Según datos de análisis de 60.000 consultas publicados por AWS, al implementar un caché semántico aplicando criterios de similitud de coseno, el coste de inferencia de LLM se redujo hasta en un 86% y la latencia de respuesta mejoró en un 88%. El framework RouteLLM de LMSYS también dividió las solicitudes entre modelos de alto y bajo coste según la dificultad de la pregunta, logrando un ahorro de costes del 85%.
El almacenamiento en caché semántico se implementa en el siguiente orden:
Para bloquear fundamentalmente los bucles infinitos, se debe integrar un patrón de disyuntor (Circuit Breaker) que verifique el estado del agente directamente en el flujo de control del pipeline. Confiar en las opciones de reintento predeterminadas que ofrece el framework provocará que el servicio entero caiga al lanzarse una excepción. Es más seguro convertir el nombre de la herramienta y los valores de sus argumentos en un hash SHA-256, añadirlo a una lista y forzar el desvío a otro nodo si el mismo hash se acumula 3 veces consecutivas.
`python
import hashlib
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
steps: int
tool_hashes: List[str]
def hash_tool_call(tool_name: str, tool_args: str) -> str:
raw_str = f"{tool_name}:{tool_args}"
return hashlib.sha256(raw_str.encode('utf-8')).hexdigest()
def agent_circuit_breaker_router(state: AgentState) -> str:
# 1. Si el número total de ejecuciones supera 5, redirigir al nodo de salida controlado
if state["steps"] > 5:
return "fallback_graceful_node"
# 2. Si se llama a la misma herramienta con los mismos argumentos 3 veces consecutivas, bloquear inmediatamente
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
return "fallback_graceful_node"
# 3. Verificar la palabra clave de finalización normal
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
return "end"
return "continue_tools"
`
Independientemente del estado de salida no determinista del LLM, este enrutador evalúa las sentencias condicionales en el entorno de ejecución de Python, evitando físicamente la entrada en bucles causados por alucinaciones.
Dejar los prompts dentro del código impide detectar fallas de regresión (Regression Failure), donde la modificación del modelo provoca que funciones que antes operaban correctamente dejen de responder. Los prompts deben extraerse fuera del código Python y gestionarse como archivos YAML independientes.
`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:
Los prompts separados de esta manera se prueban automáticamente en el pipeline de CI/CD combinando Pytest con DeepEval, un framework de evaluación de código abierto. Se establece la métrica G-Eval como punto de referencia y se verifican los valores de respuesta del modelo antes del despliegue.
`python
import pytest
from deepeval import assert_test
from deepeval.metrics import GEval, TaskCompletionMetric
from deepeval.test_case import LLMTestCase, SingleTurnParams
correctness_metric = GEval(
name="정확성 및 스키마 준수",
criteria="LLM 응답이 질문에 정확히 답변하고, 요청된 JSON 형식을 완벽히 준수하는가?",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)
@pytest.mark.parametrize(
"user_input, expected_output",
[
("2024년 1분기 매출 데이터를 요약해줘.", "1분기 총 매출은 50억 원입니다."),
("퇴직금 계산 규정을 알려줘.", "퇴직금은 근속연수 1년에 대해 30일분 이상의 평균임금입니다.")
]
)
def test_agent_regression(user_input, expected_output):
actual_output = run_in_house_agent(user_input)
test_case = LLMTestCase(
input=user_input,
actual_output=actual_output,
expected_output=expected_output
)
# Detener la compilación si no se alcanza la puntuación mínima configurada
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])
`
El sistema de evaluación se integra en tres etapas:
deepeval test run se ejecute automáticamente al enviar un PR tras modificar el código.`yaml
name: AI Agent Evaluation Gate
on:
pull_request:
branches: [ main ]
jobs:
eval-gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Dependencies
run: |
pip install poetry
poetry install
- name: Run DeepEval Suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
poetry run deepeval test run tests/test_evals.py
`
Para evitar la fuga de datos internos de la empresa, se requieren barandillas de seguridad (guardrails) que desidentifiquen la información de identificación personal (PII) antes de que las solicitudes de la API salgan al exterior. Vincular el motor Microsoft Presidio al gateway permite enmascarar automáticamente números de registro de residente, correos electrónicos, números de teléfono y números de identificación de empleado.
`python
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
analyzer = AnalyzerEngine()
employee_id_recognizer = PatternRecognizer(
supported_entity="EMPLOYEE_ID",
regex="EMP-[0-9]{6}",
score=0.95
)
analyzer.registry.add_recognizer(employee_id_recognizer)
anonymizer = AnonymizerEngine()
def sanitize_user_prompt(raw_prompt: str) -> str:
results = analyzer.analyze(
text=raw_prompt,
entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "EMPLOYEE_ID"],
language="en"
)
anonymized_result = anonymizer.anonymize(
text=raw_prompt,
analyzer_results=results,
operators={
"DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
"EMPLOYEE_ID": OperatorConfig("mask", {"chars_to_mask": 6, "masking_char": "*", "from_end": True})
}
)
return anonymized_result.text
`
Incluso al integrar búsquedas RAG, se deben tomar medidas para extraer únicamente los documentos que correspondan al nivel de autorización del usuario. Las bases de datos vectoriales como Qdrant o Pinecone gestionan la separación de permisos mediante filtrado de metadatos (Metadata Filtering) al realizar búsquedas por similitud.
`python
from qdrant_client import QdrantClient
from qdrant_client.http import models
client = QdrantClient(host="localhost", port=6333)
def search_documents_with_rbac(query_vector: list, user_department: str, user_clearance_level: int):
search_result = client.search(
collection_name="enterprise_knowledge_base",
query_vector=query_vector,
query_filter=models.Filter(
must=[
models.FieldCondition(
key="department",
match=models.MatchValue(value=user_department)
),
models.FieldCondition(
key="security_level",
range=models.Range(lte=user_clearance_level)
)
]
),
limit=5
)
return search_result
`
El principio básico para defenderse contra las inyecciones de prompts es separar por completo el canal del prompt del sistema del canal de entrada del usuario. Las operaciones peligrosas que modifican bases de datos o llaman a API externas deben requerir un procedimiento de aprobación humana (Human-in-the-Loop) o restringir los permisos de ejecución al interior de un entorno aislado (sandbox) para garantizar la seguridad.
Para sustituir los módulos SaaS y operar continuamente con sistemas internos, se requieren rutinas de inspección periódicas.
| Frecuencia | Elemento de inspección operativa | Tarea detallada |
|---|---|---|
| Diario | Tasa de errores y uso de tokens | Verificación de la tasa de errores HTTP 5xx y el consumo de tokens por departamento en el panel de Langfuse |
| Diario | Historial de bloqueos del disyuntor | Recolección y corrección de patrones de nombres de herramientas y argumentos en sesiones cortadas por detección de bucles |
| Semanal | Extracción de consultas fallidas en RAG | Selección de solicitudes con puntuación de similitud inferior a 0,6 para añadirlas al conjunto de respuestas de prueba |
| Semanal | Verificación de falsos positivos en enmascaramiento de PII | Muestreo de logs de procesamiento de Presidio para comprobar omisiones en el enmascaramiento |
| Mensual | Actualización de criterios de evaluación CI/CD | Modificación de casos de prueba automáticos para reflejar cambios en el negocio |
Si se está evaluando la posibilidad de servir modelos propios (basados en vLLM) en lugar de suscribirse a API externas, se debe realizar el cálculo en función del tráfico diario.
Una instancia AWS EC2 g5.2xlarge con 1 GPU NVIDIA A10G tiene un coste de uso mensual de aproximadamente $880. Por otro lado, la tarifa de la API de GPT-4o es de $2,50 por cada millón de tokens de entrada y $10,00 por cada millón de tokens de salida. Si el tráfico diario es inferior a 50 millones de tokens, resulta más ventajoso el método de suscripción a API considerando la mano de obra de gestión de ingenieros DevOps y los costes fijos de GPU. Solo cuando el tráfico diario supera los 100 millones de tokens o el aislamiento absoluto dentro de la red interna es imprescindible, tiene sentido migrar al servicio propio basado en vLLM.
Para estar preparado ante fallas en el modelo principal, se debe construir una arquitectura de enrutamiento de respaldo de 4 niveles.
┌─────────────────────────────────────────────────────────────────────────┐ │ 4-Tier Graceful Degradation Architecture │ │ │ │ [Tier 1] Primary High-Performance Model (e.g., GPT-4o) │ │ │ │ │ ▼ (API Failure / Timeout / Circuit Breaker) │ │ [Tier 2] Lightweight Routing Model (e.g., GPT-4o-mini / On-Prem vLLM) │ │ │ │ │ ▼ (Continuous Outage) │ │ [Tier 3] Deterministic Regex & SQL Rule Engine │ │ │ │ │ ▼ (Unrecoverable Error) │ │ [Tier 4] Static Error Message & Async Admin Ticket Generation │ └─────────────────────────────────────────────────────────────────────────┘
El trabajo de perfeccionar el sistema de agentes personalizados se realiza en bloques de 90 días.
| Período | Objetivo de aplicación | Contenido específico de las tareas |
|---|---|---|
| Días 1 a 30 | Visibilidad y protección de datos | Instalación del motor de enmascaramiento Presidio e integración del SDK de registro de Langfuse |
| Días 31 a 60 | Bloqueo de bucles y reducción de costes | Aplicación de almacenamiento en caché semántico con Redis y conexión del disyuntor basado en Python |
| Días 61 a 90 | Automatización de validación y control de permisos | Gestión de prompts en Git, integración de CI/CD con DeepEval y aplicación de metadatos RBAC en Qdrant |
Durante el primer mes, se implementa una solución de observabilidad para evitar la fuga de información personal y garantizar que todas las solicitudes queden registradas en los logs. En el segundo mes, se conectan el almacenamiento en caché de prompts repetidos y el enrutador de bloqueo de bucles para cortar los gastos inesperados. Al completar la gestión de versiones de prompts y el sistema de evaluación mediante pruebas automáticas en el último mes, se logra una operación estable sin la preocupación por el mal funcionamiento del agente con cada nuevo despliegue.