Uso de herramientas en producción para agentes: origen, reintentos y observabilidad
El agente de un banco llamó indebidamente a una API de pago e inventó un número de teléfono. Aprenda sobre los conjuntos mínimos de herramientas, el origen de los parámetros, las reintentos en tres niveles y las métricas por herramienta.
Fuentes: Agent Tool Use and Function Calling in Production — When Agents Reach Into the World
Un equipo de operaciones crediticias en un banco de Chennai proporcionó a sus analistas una asistente virtual con ocho herramientas. Cuatro de ellas servían para buscar datos (cuentas de préstamo, historial de transacciones, perfiles de clientes y la oficina de crédito), mientras que las restantes se encargaban del cálculo de pagos mensuales, la búsqueda de documentos de pólizas, el envío de notificaciones y la escalada de casos. En el transcurso de dos semanas, la revisión de calidad detectó dos problemas que los habituales controles de calidad no habían identificado en absoluto.
El primero fue el uso innecesario de herramientas. Una simple consulta del saldo solo requería la búsqueda de la cuenta de préstamo, pero la asistente también realizó la consulta a la oficina de crédito. Eso añadió 800 ms de latencia y un costo de ₹8 por consulta. Con 2,000 consultas al día, estas búsquedas innecesarias generaron un gasto adicional de ₹16,000 diarios además de las consultas realmente necesarias.
El segundo caso fue un parámetro imaginario. La búsqueda de préstamos había devuelto un número de teléfono enmascarado, tal como exige la política de privacidad del banco, y el agente inventó un número que parecía real para completar el campo phone_number de la herramienta de notificaciones. El mensaje llegó a la persona equivocada.
Ambos fallos eran invisibles desde el punto de vista de la calidad de la respuesta. La respuesta final al analista parecía correcta. El daño se manifestó en efectos secundarios: gastos innecesarios con la API y una notificación dirigida a la persona incorrecta. Esa es la realidad cotidiana del uso de herramientas por parte de los agentes en entornos de producción, y el resto de este texto explica cómo prevenirlo.
Decidir cuándo utilizar una herramienta
El fallo más básico es recurrir a una herramienta que nadie necesitaba, o a la incorrecta. Suele ocurrir cuando la selección se basa en la similitud semántica entre la consulta y la descripción de una herramienta, en lugar de en una necesidad real.
Tomemos la herramienta del bureau, que sirve para obtener la puntuación del bureau, el historial de pagos y el nivel de utilización del crédito de un cliente. Un agente que analiza el saldo de un préstamo podría elegirla simplemente porque la palabra “crédito” aparece tanto en el contexto de la consulta como en su descripción. Los datos del bureau podrían ser relevantes, pero no son necesarios para esa pregunta, y ese es el detalle que el agente pasó por alto.
Solas las descripciones más claras no resuelven este problema. Lo que realmente ayuda es seguir un paso de justificación estructurado. Antes de realizar cualquier llamada, el agente debe responder a una pregunta: ¿qué información específica proporciona esta herramienta que aún no esté disponible en los datos existentes o en otra llamada ya planificada? Si la respuesta es “nada esencial”, entonces la herramienta no se utiliza.
Tres ideas respaldan ese paso:
- Un grafo de dependencias de herramientas registra qué herramientas deben ejecutarse antes que otras. El cálculo de EMI necesita el monto del préstamo y la tasa de interés, que provienen de la búsqueda en la cuenta del préstamo, por lo que esta búsqueda se realiza primero. Hacer explícito el flujo de información evita llamadas duplicadas e innecesarias, y cualquier dato ya obtenido anteriormente en la sesión no necesita ser recuperado nuevamente.
- El principio del conjunto mínimo de herramientas establece que la selección adecuada es el grupo más pequeño de herramientas que, juntas, proporcionan todo lo necesario para la consulta. Cualquier llamada adicional es redundante o innecesaria.
- Aplicar este principio implica que el agente enumera la información que necesita, asocia cada elemento a una herramienta y verifica si ya cuenta con los datos antes de planificar otra llamada.
El problema del parámetro imaginario
El incidente con el número de teléfono pertenece a una categoría de errores que es más peligrosa que elegir la herramienta incorrecta. En este caso, la herramienta es la adecuada, pero sus parámetros provienen de la generación propia del modelo en lugar de de datos recuperados.
Esto ocurre porque los LLM generan por naturaleza. Cuando un parámetro necesario no está en el contexto, el comportamiento predeterminado es producir un valor plausible, de la misma manera en que se genera el texto. Al solicitar un número de teléfono plausible, se obtienen diez dígitos que parecen correctos pero que no pertenecen a nadie en particular.
La solución consta de dos partes.
Rastreo del origen de los parámetros. Cada valor de parámetro debe poder rastrearse hasta una fuente: el resultado de una llamada a herramienta anterior, texto extraído de la conversación, algo proporcionado por el usuario o un valor etiquetado explícitamente como generado. Los valores generados se marcan para su revisión antes de que la herramienta se ejecute, y no después.
Validación de campos obligatorios antes de la ejecución. Antes de utilizar cualquier herramienta que genere efectos secundarios, como notificaciones, escaladas, transacciones u otros cambios de estado, el agente verifica que cada parámetro obligatorio provenga de una consulta verificada y no de texto generado. Si alguno no puede ser verificado, la llamada se bloquea y el agente obtiene el valor faltante o pregunta al usuario.
Las herramientas de solo lectura, como búsquedas de cuentas, búsquedas de políticas y cálculos, conllevan menos riesgo, ya que el peor resultado es una consulta fallida en lugar de una acción incorrecta. En el caso de herramientas que modifican el estado o se comunican con terceros, la validación de parámetros generados es indispensable.
Reintentos y soluciones alternativas
Las llamadas externas fallan en producción debido a tiempos de espera, límites de frecuencia, períodos de inactividad y parámetros rechazados. La forma en que se comporta el agente cuando esto ocurre determina si la experiencia empeora de manera gradual o grave. Un enfoque en tres niveles, aplicado según el tipo de fallo, funciona bien.
Nivel uno: intentar nuevamente los fallos transitorios. Los tiempos de espera de red, los reinicios de conexión y las respuestas HTTP 503 suelen ser temporales, por lo que se debe intentar nuevamente de inmediato con retroceso exponencial y los parámetros originales, sin cambios. Tres intentos a 500 ms, 1,000 ms y 2,000 ms constituyen un límite razonable. Este nivel absorbe la mayoría de los fallos en endpoints de alta disponibilidad.
Nivel dos: intentar de nuevo con parámetros corregidos. Un error HTTP 400 o un error de validación significa que la herramienta rechazó lo que se envió, no que haya habido un problema en la red. El agente debe volver a leer lo que envió, interpretar el mensaje de validación y generar un conjunto corregido antes de intentarlo nuevamente. Solo se permite un intento de corrección. Si los parámetros corregidos también fallan, el agente debe dejar de adivinar.
Nivel tres: utilizar una herramienta de respaldo. Cuando una herramienta deja de funcionar o sigue fallando, puede existir otra herramienta relacionada con información similar, aunque quizás menos precisa. Si la consulta a la agencia falla, el agente puede recurrir a la evaluación interna de crédito basada en los últimos datos conocidos de dicha agencia almacenados en su sistema de memoria. La respuesta no está tan actualizada, pero sigue siendo útil, y se indica claramente en el mensaje, por ejemplo, que se están utilizando los datos en caché de la última evaluación porque la API de la agencia no está disponible.
Si falla cada nivel de verificación, el agente debe indicarlo en voz alta en lugar de continuar sin la información faltante. Un mensaje adecuado informa al cliente que no fue posible obtener la puntuación actual del servicio de crédito debido a un problema temporal del sistema, ofrece una respuesta sobre la elegibilidad para el préstamo basada en lo disponible y sugiere intentarlo nuevamente en unos minutos para obtener una evaluación más precisa. Eso es honesto, práctico y libre de hechos inventados.
Observabilidad para las herramientas
Las solicitudes innecesarias del banco al servicio de crédito nunca se reflejaron en las métricas de calidad. Las respuestas eran correctas, las puntuaciones RAGAS eran buenas y la satisfacción del usuario era aceptable. El problema solo existía en los registros de llamadas a las herramientas, y solo una vez que alguien calculó el costo por herramienta.
Cuatro métricas por herramienta, registradas según el tipo de consulta a lo largo del tiempo, revelan estos problemas:
- Tasa de llamadas muestra con qué frecuencia se utiliza cada herramienta por tipo de consulta. Una herramienta del servicio externo que se activa en el 80% de las consultas casi con certeza recibe llamadas innecesarias, ya que ningún asistente financiero en entorno de producción necesita legítimamente datos del servicio externo en cuatro de cada cinco conversaciones. Una alerta cuando una herramienta supera un umbral para su tipo de consulta indica llamadas innecesarias.
- Tasa de errores en parámetros registra con qué frecuencia las llamadas son rechazadas en la capa de validación. Una herramienta con un 15% de rechazos o bien recibe constantemente tipos incorrectos o es llamada con valores que no pasan las verificaciones de origen. En cualquier caso, la integración con el razonamiento del agente es defectuosa.
La implementación
El sistema de producción completo es un único módulo extenso de casi mil líneas, que abarca el registro de herramientas, el rastreador de origen, el validador, la lógica de reintentos y el rastreador de métricas.
"""
#50DaysOfAgenticAI — Day 41
Topic: Agent Tool Use and Function Calling in Production
Author: Maneesh Kumar | Azure AI Architect
Series: #50DaysOfAgenticAI on Medium & LinkedIn
Book: "From Prompts to Agentic AI: Building Agentic AI & Enterprise RAG Systems on Azure."
Kindle: https://www.amazon.in/Prompts-Agentic-AI-Building-Enterprise-ebook/dp/B0GRD8XTHH/
Paperback: https://www.amazon.in/dp/B0GTLDQSSW
Production tool use system covering:
1. Tool registry with dependency graph and minimum tool set enforcement
2. Parameter provenance tracker — verified vs generated parameter values
3. Pre-execution validation blocking tools with unverified write parameters
4. Three-tier retry architecture: transient, parameter-correction, fallback
5. Tool dependency graph resolver preventing redundant and unnecessary calls
6. Tool call observability: call rate, error rate, latency, cost per query
7. ReAct reasoning loop with structured tool selection justification
8. Tool result validator preventing hallucinated outputs
"""
import asyncio
import json
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Callable, Any
from openai import AzureOpenAI
# ─── Logging ──────────────────────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s — %(message)s"
)
log = logging.getLogger("tool_use")
# ════════════════════════════════════════════════════════════════════════════════
# ENUMS
# ════════════════════════════════════════════════════════════════════════════════
class ParameterSource(str, Enum):
RETRIEVED = "retrieved" # From a prior tool call output
USER_PROVIDED = "user_provided" # Extracted from user message
CONVERSATION = "conversation" # From conversation history
GENERATED = "generated" # LLM generated — requires validation for write tools
SYSTEM = "system" # System-level constant
class ToolCategory(str, Enum):
READ_ONLY = "read_only" # Lookups, calculations, searches
WRITE = "write" # State-changing operations
NOTIFY = "notify" # Notifications, communications
ESCALATE = "escalate" # Routing to human systems
class ToolCallStatus(str, Enum):
SUCCESS = "success"
TRANSIENT_FAILURE = "transient_failure"
PARAMETER_ERROR = "parameter_error"
VALIDATION_BLOCKED = "validation_blocked"
FALLBACK_USED = "fallback_used"
ALL_RETRIES_FAILED = "all_retries_failed"
# ════════════════════════════════════════════════════════════════════════════════
# DATA STRUCTURES
# ════════════════════════════════════════════════════════════════════════════════
@dataclass
class ToolParameter:
"""Definition of one parameter in a tool's schema."""
name: str
type: str # "string", "number", "boolean", "object"
required: bool
description: str
example: Optional[Any] = None
@dataclass
class ToolDefinition:
"""
Complete definition of one tool available to the agent.
Includes the OpenAI function calling schema, metadata for dependency
resolution, and observability configuration.
"""
name: str
description: str
category: ToolCategory
parameters: list[ToolParameter]
depends_on: list[str] # Tool names that must be called first
fallback_tool: Optional[str] # Tool to use if this one fails
cost_per_call: float = 0.0 # External API cost in INR
timeout_ms: float = 2000.0
executor: Optional[Callable] = None
def to_openai_schema(self) -> dict:
"""Convert to OpenAI function calling schema."""
properties = {}
required = []
for param in self.parameters:
properties[param.name] = {
"type": param.type,
"description": param.description
}
if param.example is not None:
properties[param.name]["example"] = str(param.example)
if param.required:
required.append(param.name)
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
}
}
@dataclass
class ParameterValue:
"""A tool parameter value with its provenance."""
parameter_name: str
value: Any
source: ParameterSource
source_detail: str # e.g., "loan_lookup result field 'account_id'"
@dataclass
class ToolCallRecord:
"""Complete record of one tool call for observability."""
call_id: str
tool_name: str
parameters: dict
parameter_sources: list[ParameterValue]
status: ToolCallStatus
result: Optional[dict]
error: Optional[str]
latency_ms: float
retry_count: int
fallback_used: bool
cost_incurred: float
timestamp: float
@dataclass
class ToolExecutionPlan:
"""
The planned sequence of tool calls for one query.
Built by the dependency resolver before any calls are made.
"""
query: str
selected_tools: list[str]
execution_order: list[list[str]] # Groups that can run in parallel
justifications: dict[str, str] # tool_name → why it's needed
estimated_cost: float
estimated_latency: float
@dataclass
class AgentToolConfig:
azure_openai_endpoint: str
azure_openai_key: str
model: str = "gpt-5-mini"
api_version: str = "2025-01-01-preview"
max_tool_calls_per_turn: int = 5
max_retries_transient: int = 3
max_retries_parameter: int = 1
backoff_base_ms: float = 500.0
require_write_validation: bool = True
# ════════════════════════════════════════════════════════════════════════════════
# TOOL REGISTRY
# ════════════════════════════════════════════════════════════════════════════════
class ToolRegistry:
"""
Central registry of all tools available to the agent.
Provides tool lookup, dependency graph resolution, and schema generation.
"""
def __init__(self):
self._tools: dict[str, ToolDefinition] = {}
def register(self, tool: ToolDefinition) -> None:
"""Register a tool in the registry."""
self._tools[tool.name] = tool
log.info(f"[ToolRegistry] Registered: {tool.name} ({tool.category.value})")
def get(self, name: str) -> Optional[ToolDefinition]:
return self._tools.get(name)
def get_all(self) -> list[ToolDefinition]:
return list(self._tools.values())
def get_openai_tools_schema(self) -> list[dict]:
"""Return all tool schemas in OpenAI function calling format."""
return [tool.to_openai_schema() for tool in self._tools.values()]
def resolve_dependencies(self, selected_tools: list[str]) -> list[list[str]]:
"""
Given a list of selected tool names, return the execution order.
Tools with no dependencies can run in the first group.
Tools that depend on group 1 run in group 2, etc.
Returns a list of groups, where each group can run in parallel.
"""
# Build dependency-aware execution groups
resolved: list[list[str]] = []
remaining = list(selected_tools)
completed: set[str] = set()
max_rounds = len(selected_tools) + 1
rounds = 0
while remaining and rounds < max_rounds:
rounds += 1
ready = []
for tool_name in remaining:
tool = self._tools.get(tool_name)
if not tool:
continue
# Check if all dependencies are either in selected tools or already done
deps_satisfied = all(
dep in completed or dep not in selected_tools
for dep in tool.depends_on
)
if deps_satisfied:
ready.append(tool_name)
if ready:
resolved.append(ready)
completed.update(ready)
remaining = [t for t in remaining if t not in ready]
else:
# Circular dependency or missing dependency — add remaining as-is
resolved.append(remaining)
break
return resolved
def get_minimum_tool_set(
self,
required_information: list[str],
available_information: list[str]
) -> list[str]:
"""
Given a list of required information types and available information,
return the minimum set of tools needed to fill the gaps.
Each tool's description is matched against required information.
"""
information_gaps = [
info for info in required_information
if info not in available_information
]
if not information_gaps:
return []
needed_tools = []
for tool in self._tools.values():
tool_covers = any(
gap.lower() in tool.description.lower()
for gap in information_gaps
)
if tool_covers:
needed_tools.append(tool.name)
return needed_tools
# ════════════════════════════════════════════════════════════════════════════════
# PARAMETER PROVENANCE TRACKER
# ════════════════════════════════════════════════════════════════════════════════
class ParameterProvenanceTracker:
"""
Tracks the source of every parameter value used in tool calls.
Flags parameters sourced from LLM generation rather than retrieval.
The tracker maintains a session-level store of all values retrieved
from tool calls and user messages. When the agent fills a tool parameter,
the tracker checks whether the value came from the store or was generated.
"""
def __init__(self):
self._retrieved_values: dict[str, ParameterValue] = {} # key → ParameterValue
def record_retrieved(
self,
field_name: str,
value: Any,
source_tool: str,
source_detail: str = ""
) -> None:
"""Record a value retrieved from a tool call."""
key = f"{field_name}:{str(value)[:50]}"
self._retrieved_values[key] = ParameterValue(
parameter_name=field_name,
value=value,
source=ParameterSource.RETRIEVED,
source_detail=f"{source_tool}: {source_detail}"
)
def record_user_provided(
self,
field_name: str,
value: Any,
context: str = ""
) -> None:
"""Record a value provided directly by the user."""
key = f"{field_name}:{str(value)[:50]}"
self._retrieved_values[key] = ParameterValue(
parameter_name=field_name,
value=value,
source=ParameterSource.USER_PROVIDED,
source_detail=context
)
def assess_parameter(
self,
parameter_name: str,
value: Any,
tool_name: str
) -> ParameterValue:
"""
Assess whether a parameter value is verified or generated.
Returns the ParameterValue with source assessment.
"""
# Try exact match first
key = f"{parameter_name}:{str(value)[:50]}"
if key in self._retrieved_values:
return self._retrieved_values[key]
# Try matching by field name
for stored_key, stored_val in self._retrieved_values.items():
if stored_key.startswith(f"{parameter_name}:"):
return stored_val
# Try matching by value across any field
str_value = str(value)
for stored_key, stored_val in self._retrieved_values.items():
if str(stored_val.value)[:50] == str_value[:50]:
return ParameterValue(
parameter_name=parameter_name,
value=value,
source=stored_val.source,
source_detail=f"Matched by value to: {stored_val.source_detail}"
)
# Not found — mark as generated
log.warning(
f"[Provenance] Parameter '{parameter_name}'='{str(value)[:30]}' "
f"for tool '{tool_name}' not found in retrieved values — marking GENERATED"
)
return ParameterValue(
parameter_name=parameter_name,
value=value,
source=ParameterSource.GENERATED,
source_detail="Not found in retrieved or user-provided values"
)
def has_verified_value(self, field_name: str) -> bool:
"""Check whether a verified value exists for a field."""
return any(
key.startswith(f"{field_name}:")
for key in self._retrieved_values
)
# ════════════════════════════════════════════════════════════════════════════════
# PRE-EXECUTION VALIDATOR
# ════════════════════════════════════════════════════════════════════════════════
class PreExecutionValidator:
"""
Validates tool call parameters before execution.
For write and notify tools: blocks calls with any GENERATED required parameter.
For read-only tools: allows GENERATED parameters (worst case is a failed lookup).
"""
def __init__(self, config: AgentToolConfig):
self.config = config
def validate(
self,
tool: ToolDefinition,
parameters: dict,
provenance: list[ParameterValue]
) -> tuple[bool, str]:
"""
Validate that tool parameters are safe to execute.
Returns (is_valid, reason) tuple.
"""
if not self.config.require_write_validation:
return True, "Validation disabled"
# Read-only tools: allow generated parameters
if tool.category == ToolCategory.READ_ONLY:
return True, "Read-only tool — generated parameters permitted"
# Write/notify/escalate tools: block generated required parameters
required_param_names = {p.name for p in tool.parameters if p.required}
provenance_map = {pv.parameter_name: pv for pv in provenance}
for param_name in required_param_names:
pv = provenance_map.get(param_name)
if pv and pv.source == ParameterSource.GENERATED:
return False, (
f"Required parameter '{param_name}' for {tool.category.value} tool "
f"'{tool.name}' has GENERATED source. "
f"Retrieve '{param_name}' before calling this tool."
)
return True, "All required parameters verified"
# ════════════════════════════════════════════════════════════════════════════════
# RETRY MANAGER
# ════════════════════════════════════════════════════════════════════════════════
class ToolRetryManager:
"""
Implements the three-tier retry architecture for tool call failures.
Tier 1: Transient failure retry with exponential backoff
Tier 2: Parameter correction retry (one attempt after validation failure)
Tier 3: Fallback tool substitution
"""
def __init__(self, config: AgentToolConfig, registry: ToolRegistry):
self.config = config
self.registry = registry
async def execute_with_retry(
self,
tool: ToolDefinition,
parameters: dict,
executor: Callable
) -> tuple[Optional[dict], ToolCallStatus, int]:
"""
Execute a tool with the full retry architecture.
Returns (result, status, retry_count).
"""
retry_count = 0
# Tier 1: Transient failure retry
for attempt in range(self.config.max_retries_transient + 1):
try:
result = await asyncio.wait_for(
executor(parameters),
timeout=tool.timeout_ms / 1000
)
return result, ToolCallStatus.SUCCESS, retry_count
except asyncio.TimeoutError:
retry_count += 1
log.warning(
f"[Retry] {tool.name} timeout (attempt {attempt + 1}/"
f"{self.config.max_retries_transient + 1})"
)
if attempt < self.config.max_retries_transient:
backoff = self.config.backoff_base_ms * (2 ** attempt) / 1000
await asyncio.sleep(backoff)
except ValueError as e:
# Parameter error — don't retry with same params
log.warning(f"[Retry] {tool.name} parameter error: {e}")
return None, ToolCallStatus.PARAMETER_ERROR, retry_count
except Exception as e:
error_str = str(e).lower()
if "timeout" in error_str or "connection" in error_str or "503" in error_str:
retry_count += 1
if attempt < self.config.max_retries_transient:
backoff = self.config.backoff_base_ms * (2 ** attempt) / 1000
log.warning(
f"[Retry] {tool.name} transient error: {e} | "
f"Retrying in {backoff:.0f}ms"
)
await asyncio.sleep(backoff)
continue
else:
log.error(f"[Retry] {tool.name} non-retryable error: {e}")
return None, ToolCallStatus.TRANSIENT_FAILURE, retry_count
# Tier 3: Fallback tool
if tool.fallback_tool:
fallback_def = self.registry.get(tool.fallback_tool)
if fallback_def and fallback_def.executor:
log.info(
f"[Retry] Falling back to {tool.fallback_tool} "
f"for failed {tool.name}"
)
try:
fallback_result = await asyncio.wait_for(
fallback_def.executor(parameters),
timeout=fallback_def.timeout_ms / 1000
)
return fallback_result, ToolCallStatus.FALLBACK_USED, retry_count
except Exception as e:
log.error(f"[Retry] Fallback {tool.fallback_tool} also failed: {e}")
return None, ToolCallStatus.ALL_RETRIES_FAILED, retry_count
# ════════════════════════════════════════════════════════════════════════════════
# TOOL OBSERVABILITY TRACKER
# ════════════════════════════════════════════════════════════════════════════════
class ToolObservabilityTracker:
"""
Tracks per-tool metrics across all requests.
Provides call rate, error rate, latency, and cost analytics.
"""
def __init__(self):
self._records: list[ToolCallRecord] = []
def record(self, record: ToolCallRecord) -> None:
self._records.append(record)
def get_metrics_per_tool(self) -> dict:
"""Compute aggregate metrics per tool."""
if not self._records:
return {}
tool_data: dict[str, list[ToolCallRecord]] = {}
for r in self._records:
tool_data.setdefault(r.tool_name, []).append(r)
metrics = {}
total_queries = max(
len(set(r.call_id.split("_")[0] for r in self._records)), 1
)
for tool_name, records in tool_data.items():
total_calls = len(records)
success = sum(1 for r in records if r.status == ToolCallStatus.SUCCESS)
errors = sum(1 for r in records if r.status in (
ToolCallStatus.TRANSIENT_FAILURE,
ToolCallStatus.ALL_RETRIES_FAILED,
ToolCallStatus.PARAMETER_ERROR
))
blocked = sum(1 for r in records if r.status == ToolCallStatus.VALIDATION_BLOCKED)
total_cost = sum(r.cost_incurred for r in records)
avg_latency = sum(r.latency_ms for r in records) / total_calls
generated_params = sum(
1 for r in records
if any(pv.source == ParameterSource.GENERATED for pv in r.parameter_sources)
)
metrics[tool_name] = {
"total_calls": total_calls,
"call_rate": round(total_calls / total_queries, 2),
"success_rate": round(success / max(total_calls, 1), 3),
"error_rate": round(errors / max(total_calls, 1), 3),
"validation_block_rate": round(blocked / max(total_calls, 1), 3),
"generated_param_rate": round(generated_params / max(total_calls, 1), 3),
"avg_latency_ms": round(avg_latency, 1),
"total_cost_inr": round(total_cost, 2),
"cost_per_call_inr": round(total_cost / max(total_calls, 1), 2),
}
return metrics
def get_cost_savings_opportunities(self, call_rate_threshold: float = 0.50) -> list[dict]:
"""
Identify tools that appear to be over-called based on call rate.
Tools called more than threshold per query are candidates for optimisation.
"""
metrics = self.get_metrics_per_tool()
opportunities = []
for tool_name, m in metrics.items():
if m["call_rate"] > call_rate_threshold:
opportunities.append({
"tool": tool_name,
"current_call_rate": m["call_rate"],
"threshold": call_rate_threshold,
"total_cost_inr": m["total_cost_inr"],
"potential_saving_inr": round(
m["total_cost_inr"] *
(1 - call_rate_threshold / m["call_rate"]), 2
),
"recommendation": (
f"Call rate {m['call_rate']:.0%} exceeds threshold {call_rate_threshold:.0%}. "
f"Review tool selection justification for unnecessary calls."
)
})
return sorted(opportunities, key=lambda x: -x["potential_saving_inr"])
def get_alerts(self) -> list[str]:
"""Generate observability alerts for metrics exceeding thresholds."""
metrics = self.get_metrics_per_tool()
alerts = []
for tool_name, m in metrics.items():
if m["call_rate"] > 0.70:
alerts.append(
f"⚠️ {tool_name}: Call rate {m['call_rate']:.0%} — "
f"possible over-calling"
)
if m["error_rate"] > 0.10:
alerts.append(
f"🔴 {tool_name}: Error rate {m['error_rate']:.0%} — "
f"tool reliability issue"
)
if m["generated_param_rate"] > 0.05:
alerts.append(
f"🔴 {tool_name}: {m['generated_param_rate']:.0%} calls use "
f"generated parameters — provenance gap"
)
return alerts
# ════════════════════════════════════════════════════════════════════════════════
# REACT REASONING LOOP WITH TOOL USE
# ════════════════════════════════════════════════════════════════════════════════
class ReActAgentWithToolUse:
"""
ReAct (Reason + Act) agent with full tool use production infrastructure.
Each iteration:
1. Reason: LLM determines next action with justification
2. Plan: validate tool selection against minimum tool set principle
3. Validate: check parameter provenance for write/notify tools
4. Execute: call tool with retry architecture
5. Observe: record result and update provenance tracker
6. Repeat or respond
"""
SYSTEM_PROMPT = """You are a financial services AI assistant with access to banking tools.
When using tools:
1. Only call a tool when you genuinely NEED information it provides
2. Justify each tool call: what specific information does it provide that you don't already have?
3. Use information from prior tool results rather than calling the same tool again
4. Never call the bureau query tool unless credit eligibility assessment is specifically required
5. For notification tools: only call after you have retrieved the recipient's contact details
Think step by step before calling any tool."""
TOOL_SELECTION_PROMPT = """Given the user query and the tools available, determine:
1. Which tools are NECESSARY (not just potentially useful) to answer this query?
2. For each necessary tool, what specific information does it provide that cannot be obtained otherwise?
Query: {query}
Available tools: {tools}
Current available information: {available_info}
Return ONLY valid JSON:
{{"necessary_tools": ["tool_name_1", ...],
"justifications": {{"tool_name": "why this specific tool is necessary"}},
"can_answer_without_retrieval": true/false,
"direct_answer": "if can_answer_without_retrieval, provide the answer"}}"""
def __init__(
self,
config: AgentToolConfig,
registry: ToolRegistry
):
self.config = config
self.client = AzureOpenAI(
azure_endpoint=config.azure_openai_endpoint,
api_key=config.azure_openai_key,
api_version=config.api_version
)
self.registry = registry
self.validator = PreExecutionValidator(config)
self.retry_mgr = ToolRetryManager(config, registry)
self.obs_tracker = ToolObservabilityTracker()
self.provenance = ParameterProvenanceTracker()
async def _plan_tool_calls(
self,
query: str,
available_info: dict
) -> dict:
"""Determine necessary tools using the minimum tool set principle."""
tool_descriptions = "\n".join([
f"- {t.name}: {t.description} [category: {t.category.value}, cost: ₹{t.cost_per_call}]"
for t in self.registry.get_all()
])
available_str = json.dumps({k: v for k, v in available_info.items() if v is not None})
try:
response = self.client.chat.completions.create(
model=self.config.model,
messages=[
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": self.TOOL_SELECTION_PROMPT.format(
query=query,
tools=tool_descriptions,
available_info=available_str or "None"
)}
],
temperature=0,
max_tokens=400,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
except Exception as e:
log.error(f"[ReAct] Tool planning failed: {e}")
return {"necessary_tools": [], "justifications": {}, "can_answer_without_retrieval": True}
async def _execute_tool(
self,
tool_name: str,
parameters: dict,
query_id: str
) -> ToolCallRecord:
"""Execute one tool call with validation, retry, and observability."""
tool = self.registry.get(tool_name)
start = time.time()
if not tool:
return ToolCallRecord(
call_id=f"{query_id}_{tool_name}",
tool_name=tool_name,
parameters=parameters,
parameter_sources=[],
status=ToolCallStatus.ALL_RETRIES_FAILED,
result=None,
error=f"Tool '{tool_name}' not found in registry",
latency_ms=0,
retry_count=0,
fallback_used=False,
cost_incurred=0.0,
timestamp=time.time()
)
# Assess parameter provenance
param_sources = [
self.provenance.assess_parameter(param_name, value, tool_name)
for param_name, value in parameters.items()
]
# Pre-execution validation for write/notify tools
is_valid, validation_reason = self.validator.validate(
tool, parameters, param_sources
)
if not is_valid:
log.warning(
f"[ReAct] Tool call BLOCKED: {tool_name} — {validation_reason}"
)
record = ToolCallRecord(
call_id=f"{query_id}_{tool_name}",
tool_name=tool_name,
parameters=parameters,
parameter_sources=param_sources,
status=ToolCallStatus.VALIDATION_BLOCKED,
result=None,
error=validation_reason,
latency_ms=(time.time() - start) * 1000,
retry_count=0,
fallback_used=False,
cost_incurred=0.0,
timestamp=time.time()
)
self.obs_tracker.record(record)
return record
# Execute with retry architecture
if not tool.executor:
# Mock execution for demo
async def mock_executor(params):
await asyncio.sleep(0.05)
return self._mock_tool_result(tool_name, params)
executor = mock_executor
else:
executor = tool.executor
result, status, retry_count = await self.retry_mgr.execute_with_retry(
tool, parameters, executor
)
latency_ms = (time.time() - start) * 1000
# Update provenance tracker with results
if result:
for field_name, value in result.items():
self.provenance.record_retrieved(
field_name=field_name,
value=value,
source_tool=tool_name,
source_detail=f"result.{field_name}"
)
cost = tool.cost_per_call if status != ToolCallStatus.VALIDATION_BLOCKED else 0.0
record = ToolCallRecord(
call_id=f"{query_id}_{tool_name}",
tool_name=tool_name,
parameters=parameters,
parameter_sources=param_sources,
status=status,
result=result,
error=None if result else f"Tool failed after {retry_count} retries",
latency_ms=latency_ms,
retry_count=retry_count,
fallback_used=(status == ToolCallStatus.FALLBACK_USED),
cost_incurred=cost,
timestamp=time.time()
)
self.obs_tracker.record(record)
log.info(
f"[ReAct] {tool_name}: {status.value} | "
f"{latency_ms:.0f}ms | ₹{cost:.2f} | "
f"retries={retry_count}"
)
return record
def _mock_tool_result(self, tool_name: str, params: dict) -> dict:
"""Mock tool results for demo."""
mocks = {
"get_loan_status": {
"account_id": params.get("account_id", "LA4421"),
"account_holder": "Priya Sharma",
"outstanding": 4235000,
"next_emi_date": "2026-05-05",
"emi_amount": 45000,
"status": "Active",
"phone_masked": "+91-98XXXXX123" # Masked for privacy
},
"calculate_emi": {
"emi_amount": params.get("emi", 45000),
"principal": params.get("principal", 5000000),
"total_interest": 2100000,
"total_payment": 7100000
},
"fetch_bureau_score": {
"bureau_score": 756,
"credit_history": "Good",
"payment_history": "Consistent",
"utilisation": 0.32
},
"query_policy_docs": {
"content": "Prepayment of home loans carries no penalty for floating rate loans under RBI guidelines.",
"source": "RBI Master Circular 2024-HC-001",
"confidence": 0.94
}
}
return mocks.get(tool_name, {"status": "ok", "data": "mock result"})
async def run(
self,
user_query: str,
session_context: dict = None
) -> dict:
"""
Run the full ReAct tool use loop for one user query.
Returns the final answer with tool execution metadata.
"""
query_id = f"q_{int(time.time())}"
available_info = session_context or {}
all_records: list[ToolCallRecord] = []
total_cost = 0.0
log.info(f"\n[ReAct] Processing: '{user_query}'")
# Step 1: Plan tool calls using minimum tool set principle
plan = await self._plan_tool_calls(user_query, available_info)
if plan.get("can_answer_without_retrieval"):
log.info(f"[ReAct] Answering from context — no retrieval needed")
return {
"answer": plan.get("direct_answer", ""),
"tools_called": [],
"total_cost": 0.0,
"tool_records": []
}
necessary_tools = plan.get("necessary_tools", [])
justifications = plan.get("justifications", {})
log.info(
f"[ReAct] Planned tools: {necessary_tools} | "
f"Skipping: {[t.name for t in self.registry.get_all() if t.name not in necessary_tools]}"
)
# Step 2: Resolve execution order
execution_groups = self.registry.resolve_dependencies(necessary_tools)
log.info(f"[ReAct] Execution order: {execution_groups}")
# Step 3: Execute tools in dependency order (parallel within groups)
tool_results = {}
for group in execution_groups:
group_tasks = []
for tool_name in group:
tool = self.registry.get(tool_name)
if not tool:
continue
# Build parameters from available information
params = self._build_parameters(tool, available_info, tool_results)
group_tasks.append(self._execute_tool(tool_name, params, query_id))
if group_tasks:
group_records = await asyncio.gather(*group_tasks)
for record in group_records:
all_records.append(record)
total_cost += record.cost_incurred
if record.result:
tool_results[record.tool_name] = record.result
available_info.update(record.result)
# Step 4: Generate final answer using tool results
context = json.dumps(tool_results, indent=2, default=str)
try:
response = self.client.chat.completions.create(
model=self.config.model,
messages=[
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": (
f"Query: {user_query}\n\n"
f"Retrieved information:\n{context}\n\n"
f"Answer the query using only the retrieved information."
)}
],
temperature=0.1,
max_tokens=300
)
answer = response.choices[0].message.content.strip()
except Exception as e:
log.error(f"[ReAct] Final generation failed: {e}")
answer = f"Unable to generate response: {e}"
return {
"answer": answer,
"tools_called": [r.tool_name for r in all_records],
"tools_skipped": [t.name for t in self.registry.get_all()
if t.name not in [r.tool_name for r in all_records]],
"justifications": justifications,
"total_cost_inr": round(total_cost, 2),
"tool_records": all_records
}
def _build_parameters(
self,
tool: ToolDefinition,
available: dict,
prior_results: dict
) -> dict:
"""Build tool parameters from available context."""
params = {}
all_available = {**available, **{k: v for d in prior_results.values() for k, v in d.items()}}
for param in tool.parameters:
# Try to find the value in available context
if param.name in all_available:
params[param.name] = all_available[param.name]
elif param.example is not None:
params[param.name] = param.example
return params
# ════════════════════════════════════════════════════════════════════════════════
# DEMO — Building the Tool Registry and Running Queries
# ════════════════════════════════════════════════════════════════════════════════
async def demo():
"""
Demonstrates the production tool use system on four queries:
1. Simple balance query (should use loan_status only, NOT bureau)
2. Eligibility query (legitimately needs bureau score)
3. Notification with masked phone (should be blocked by validation)
4. Policy question (no retrieval needed — parametric knowledge)
"""
config = AgentToolConfig(
azure_openai_endpoint="https://your-resource.openai.azure.com/",
azure_openai_key="your-key",
require_write_validation=True
)
# Build the tool registry
registry = ToolRegistry()
registry.register(ToolDefinition(
name="get_loan_status",
description="Retrieves current loan account status, outstanding balance, next EMI date, and account holder details for a specific loan account",
category=ToolCategory.READ_ONLY,
parameters=[
ToolParameter("account_id", "string", True, "Loan account ID (e.g. LA4421)", "LA4421")
],
depends_on=[],
fallback_tool=None,
cost_per_call=0.50
))
registry.register(ToolDefinition(
name="calculate_emi",
description="Calculates EMI for given principal, interest rate and tenure. Does NOT require bureau data.",
category=ToolCategory.READ_ONLY,
parameters=[
ToolParameter("principal", "number", True, "Loan principal in INR", 5000000),
ToolParameter("rate", "number", True, "Annual interest rate as decimal", 0.085),
ToolParameter("months", "number", True, "Loan tenure in months", 240)
],
depends_on=[],
fallback_tool=None,
cost_per_call=0.0
))
registry.register(ToolDefinition(
name="fetch_bureau_score",
description="Retrieves credit bureau score, payment history, and credit utilisation for eligibility assessment. Use ONLY when credit eligibility needs to be assessed. Do NOT call for balance queries, EMI calculations, or policy questions.",
category=ToolCategory.READ_ONLY,
parameters=[
ToolParameter("customer_id", "string", True, "Customer ID", "CUST_001"),
ToolParameter("pan", "string", True, "PAN card number", "ABCDE1234F")
],
depends_on=[],
fallback_tool=None,
cost_per_call=8.0 # ₹8 per bureau pull
))
registry.register(ToolDefinition(
name="send_notification",
description="Sends SMS or WhatsApp notification to a customer. Requires verified phone number — do NOT use generated phone numbers.",
category=ToolCategory.NOTIFY,
parameters=[
ToolParameter("phone_number", "string", True, "Verified customer phone number", None),
ToolParameter("message", "string", True, "Notification message text", None),
ToolParameter("channel", "string", True, "sms or whatsapp", "sms")
],
depends_on=["get_loan_status"], # Must retrieve customer details first
fallback_tool=None,
cost_per_call=0.30
))
registry.register(ToolDefinition(
name="query_policy_docs",
description="Searches policy documents and regulatory guidelines for product terms, RBI circulars, and lending policies.",
category=ToolCategory.READ_ONLY,
parameters=[
ToolParameter("query", "string", True, "Search query for policy information", None)
],
depends_on=[],
fallback_tool=None,
cost_per_call=0.10
))
agent = ReActAgentWithToolUse(config, registry)
print("\n" + "="*65)
print("AGENT TOOL USE PRODUCTION DEMO")
print("="*65)
# Test queries
test_queries = [
{
"query": "What is the current outstanding balance on my home loan LA4421?",
"context": {"account_id": "LA4421", "customer_id": "CUST_001"},
"expect": "Should use get_loan_status ONLY — bureau call is unnecessary"
},
{
"query": "I want to apply for a new personal loan of ₹5 lakh. Am I eligible?",
"context": {"customer_id": "CUST_001", "pan": "ABCDE1234F"},
"expect": "Should use fetch_bureau_score — eligibility assessment genuinely needs it"
},
{
"query": "What does RBI say about prepayment charges on floating rate home loans?",
"context": {},
"expect": "Should use query_policy_docs or answer directly — no bureau needed"
},
]
total_sessions_cost = 0.0
total_bureau_calls = 0
for i, test in enumerate(test_queries, 1):
print(f"\n[Query {i}] {test['query'][:65]}")
print(f" Expected: {test['expect']}")
# Pre-seed provenance tracker with user-provided context
for key, value in test["context"].items():
agent.provenance.record_user_provided(key, value, "user_context")
result = await agent.run(test["query"], dict(test["context"]))
bureau_called = "fetch_bureau_score" in result["tools_called"]
total_bureau_calls += 1 if bureau_called else 0
total_sessions_cost += result["total_cost_inr"]
print(f"\n Tools called: {result['tools_called'] or ['none (parametric)']}")
print(f" Tools skipped: {result['tools_skipped'][:3]}")
print(f" Bureau called: {'YES ⚠️' if bureau_called else 'No ✅'}")
print(f" Total cost: ₹{result['total_cost_inr']:.2f}")
print(f" Answer preview: {result['answer'][:100]}...")
if result["tool_records"]:
print(f"\n Tool Records:")
for rec in result["tool_records"]:
status_icon = "✅" if rec.status == ToolCallStatus.SUCCESS else (
"🚫" if rec.status == ToolCallStatus.VALIDATION_BLOCKED else "❌"
)
print(
f" {status_icon} {rec.tool_name}: {rec.status.value} | "
f"{rec.latency_ms:.0f}ms | ₹{rec.cost_incurred:.2f}"
)
if rec.status == ToolCallStatus.VALIDATION_BLOCKED:
print(f" BLOCKED: {rec.error[:80]}")
# Reset provenance for next query
agent.provenance = ParameterProvenanceTracker()
# Observability summary
print(f"\n{'='*65}")
print("[Tool Observability Summary]")
metrics = agent.obs_tracker.get_metrics_per_tool()
for tool_name, m in metrics.items():
alert = " ⚠️ OVER-CALLED" if m["call_rate"] > 0.50 else ""
print(
f" {tool_name:<25} "
f"calls/query={m['call_rate']:.0%} "
f"cost=₹{m['total_cost_inr']:.2f} "
f"latency={m['avg_latency_ms']:.0f}ms"
f"{alert}"
)
opportunities = agent.obs_tracker.get_cost_savings_opportunities(0.50)
if opportunities:
print(f"\n[Cost Savings Opportunities]")
for opp in opportunities:
print(f" {opp['tool']}: save ₹{opp['potential_saving_inr']:.2f} | {opp['recommendation'][:70]}")
alerts = agent.obs_tracker.get_alerts()
if alerts:
print(f"\n[Observability Alerts]")
for alert in alerts:
print(f" {alert}")
print(f"\n Total session cost: ₹{total_sessions_cost:.2f}")
print(f" Bureau calls (should be 1 of 3): {total_bureau_calls}/3")
# Deep dive in "From Prompts to Agentic AI: Building Agentic AI & Enterprise RAG Systems on Azure."
# Kindle: https://www.amazon.in/Prompts-Agentic-AI-Building-Enterprise-ebook/dp/B0GRD8XTHH/
# Paperback: https://www.amazon.in/dp/B0GTLDQSSW
if __name__ == "__main__":
asyncio.run(demo())
Qué hace el código
El registro de herramientas almacena el esquema completo de cada herramienta junto con su categoría. fetch_bureau_score está marcado como de solo lectura, pero tiene un costo de ₹8 por llamada y una descripción que prohíbe su uso para consultas de saldo, cálculos de EMI o preguntas sobre pólizas. send_notification está clasificado como herramienta de notificación, lo que activa la validación más estricta previa a la ejecución reservada para las operaciones de escritura. El grafo de dependencias indica que send_notification depende de get_loan_status, lo que significa que se deben obtener los detalles de contacto verificados antes de enviar cualquier notificación.
El rastreador de procedencia mantiene un almacenamiento a nivel de sesión de todos los valores recuperados de una llamada anterior o proporcionados por el usuario. Cuando el agente completa el campo phone_number para send_notification, el rastreador verifica si el número proviene de una recuperación anterior. La herramienta de estado del préstamo devolvió un número enmascarado, “+91-98XXXXX123”, por lo que si el agente intenta ingresar uno completo sin enmascarar, la verificación indica que fue generado. El agente no pudo haber recuperado un número completo de una fuente enmascarada.
Luego, el validador de pre-ejecución bloquea la llamada y devuelve un error claro. Indica que el phone_number necesario para la herramienta de notificación proviene de una fuente generada y debe recuperarse primero desde la herramienta de perfil del cliente. La llamada se detiene antes de ejecutarse, y no se detecta después de que se haya enviado un SMS incorrecto.
El rastreador de observabilidad acumula métricas por herramienta a lo largo de las consultas. Después de solo tres consultas de prueba, muestra las tasas de llamadas, los costos y el exceso de llamadas. Su cálculo de ahorros revela ese mismo exceso de llamadas por parte de la oficina que costaba ₹16,000 al día, algo visible ya desde el primer día de implementación.
Las descripciones son especificaciones
La calidad de la selección de herramientas está directamente relacionada con la calidad de las descripciones. Una herramienta mal descrita es elegida para tareas que no puede realizar y se pasa por alto para aquellas que sí puede hacer.
Tres tipos de errores aparecen con mayor frecuencia:
- Vaguedad. “Obtener información del cliente” podría referirse a la herramienta de estado de préstamos, a la herramienta de perfil o a la herramienta de la oficina. Ante tres opciones igualmente vagas, el agente se basa en palabras clave superficiales en lugar del alcance real de la herramienta.
get_loan_status o customer_profile”, la dependencia queda visible en la descripción, antes de que entre en juego la verificación estructural del grafo.Una buena descripción es una especificación, no un texto publicitario. Indica qué proporciona la herramienta, qué requiere, cuándo usarla y cuándo no.
Tres meses después
Después de tres meses utilizando el seguimiento de origen y la observabilidad de herramientas, los resultados del banco fueron concretos.
El problema de llamadas excesivas surgió en el segundo día de la implementación de la funcionalidad de observabilidad, antes incluso de que alguien comenzara a revisarlo. La alerta por tasa de llamadas se activó cuando fetch_bureau_score mostró una tasa del 78% frente a un umbral del 50%, y el cálculo de costos indicó un ahorro potencial de ₹8,800 al día. Después de que se incluyera en la descripción un texto que excluía explícitamente esa situación, la tasa de llamadas del servicio bajó del 78% al 29% en tres días.
El problema con los números de teléfono apareció durante la primera semana del análisis retrospectivo, a partir de los registros de origen de las llamadas bloqueadas. En las dos semanas anteriores, la herramienta de notificaciones se había utilizado 23 veces con números de teléfono generados, y los 23 fueron bloqueados. Sin el validador, eso habría significado 23 mensajes SMS enviados a personas equivocadas.
Los niveles de reintentos solucionaron un problema de fiabilidad del que nadie se había dado cuenta. La API de la oficina tenía una tasa de tiempos de espera transitorios del 4%, y el agente original fallaba en cada uno de ellos. El retroceso exponencial absorbió el 81% de esos tiempos de espera y redujo las fallas visibles para los usuarios del 4% al 0,7%. El resto reflejaba verdaderos períodos de inactividad de la oficina, lo que hizo que se recurriera correctamente a la evaluación interna almacenada en caché.
Los números generales:
- Los costos de la API de la oficina disminuyeron de ₹32,000 a ₹11,600 por día, lo que representa una reducción del 64%. En parte se debió a una menor tasa de llamadas, y en parte a la eliminación de consultas que, según los registros, provocaban llamadas a la oficina debido a efectos de consultas anteriores en una misma sesión; ahora, el rastreador de origen y el resolvedor de dependencias impiden esto.
Lecturas relacionadas
- Memory de agentes AI en 2026: Mem0, Letta, Zep y comparación de plataformas en la nube — Un recorrido por la memoria de los agentes en 2026: los tipos de memoria CoALA, Mem0, Letta, Zep, las ofertas en la nube, los puntos de referencia, los riesgos de contaminación y qué diseño se adapta a cada tarea.
- Claude Agent SDK vs OpenAI Agents SDK vs Google ADK: Un agente, tres formas de desarrollarlo — El mismo agente para etiquetas nutricionales creado con Claude Agent SDK, OpenAI Agents SDK y Google ADK, mostrando claramente qué maneja cada kit y qué aún debes desarrollar tú mismo.
- Estrategias de particionamiento para RAG en producción: de tamaño fijo, semántico y padre-hijo — Una guía práctica sobre el particionamiento de RAG, desde la división por tamaño fijo y recursiva hasta los enfoques padre-hijo, proposiciones, particionamiento tardío, RAPTOR, tablas, código y pruebas de rendimiento.
- Instantáneas de memoria del agente, habilidades, decadencia y alcance por ramas — Cuatro mejoras a una capa de memoria tipo vector-DB integrada para agentes de programación: instantáneas portátiles con función de fusión, activación de habilidades, decadencia con verificaciones de conflictos y memoria con alcance por ramas.
- Explicación de las llamadas a herramientas: cómo los modelos piden al software que actúe y dónde fallan — Una guía práctica sobre las llamadas a funciones y el uso de herramientas: el ciclo de solicitudes, esquemas, herramientas de lectura frente a herramientas de acción, agentes, MCP, inyección de prompts, seguimiento y fallos comunes.
- Cinco arquitecturas RAG y cuándo es adecuado utilizar cada una en producción — Un recorrido práctico por las arquitecturas RAG ingenuas, híbridas, basadas en grafos, avanzadas y agenciales con LlamaIndex y LangGraph, que aborda los modos de fallo, los costos y cómo combinarlas.
- Construcción de un pipeline ETL de nivel producción para RAG con LangChain y FAISS — Diseño de procesos de fragmentación, incrustación e indexación vectorial para 40,000 PDFs relacionados con cumplimiento normativo, incluyendo mecanismos de reintentos, colas de mensajes no procesados, métricas de Prometheus y pruebas de rendimiento medibles.
- Arquitectura de IA para agentes en producción: modos de fallo y capas de control — Una guía de arquitectura capa por capa para sistemas de agentes en producción: modos de fallo, medidas de seguridad, evaluaciones, capacidad de observabilidad, control de costos, principio de confianza cero y gobernanza.
- AegisOps: Un agente SRE autónomo con cuatro capas de memoria — Cómo un agente LangGraph ReAct resuelve incidentes en Kubernetes, con memoria episódica basada en pgvector, niveles de autonomía controlados por políticas, dieciséis herramientas y un registro de auditoría completo.
- Dartboard RAG: Qué hacer cuando Top-K devuelve el mismo fragmento tres veces — La recuperación simple con top-k puede devolver fragmentos casi idénticos. Dartboard RAG selecciona tanto la relevancia como la diversidad; aquí se explica el algoritmo, el código y cuándo utilizarlo.
- GraphRAG o RAG simple? Donde los grafos son útiles y donde cuestan demasiado — GraphRAG puede superar al RAG estándar en razonamiento y resúmenes de múltiples pasos, pero pierde en búsquedas simples y cuesta más. Consulte los números de las pruebas de rendimiento y una guía rápida para saber cuándo usar cada uno.
- Por qué fallan las habilidades de agentes en cadena y la solución de tres capas que funciona — Las cadenas profundas de habilidades sobrecargan el contexto del agente. En su lugar, construya con átomos, moléculas y compuestos, con ejemplos para Claude Code, LangGraph, CrewAI y Agno.