Utilisation des outils de production par les agents : origine, tentatives et observabilité
Un agent bancaire a appelé par erreur une API payante et a inventé un numéro de téléphone. Apprenez les ensembles d’outils minimum, l’origine des paramètres, les tentatives en trois niveaux et les métriques par outil.
Sources: Agent Tool Use and Function Calling in Production — When Agents Reach Into the World
Une équipe des opérations de crédit d’une banque à Chennai a fourni à ses analystes un assistant virtuel doté de huit outils. Quatre d’entre eux servaient à rechercher des données (comptes de prêt, historique des transactions, profils clients et bureau de crédit), tandis que les autres géraient les calculs liés aux échéances, la recherche de documents d’assurance, l’envoi de notifications et le transfert des cas vers des niveaux hiérarchiques supérieurs. En l’espace de deux semaines, une revue de qualité a révélé deux problèmes que les contrôles habituels de qualité n’avaient pas du tout détectés.
Le premier problème concernait l’utilisation inutile d’outils. Une simple demande de solde ne nécessitait que la recherche du compte de prêt, mais l’assistant a également lancé une requête auprès du bureau de crédit. Cela a ajouté 800 ms de latence et entraîné des frais de 8 roupies par requête. Avec 2 000 requêtes par jour, ces recherches inutiles représentaient des coûts supplémentaires de 16 000 roupies par jour en plus des appels réellement nécessaires.
Le deuxième cas concernait un paramètre fantôme. La recherche de prêt avait renvoyé un numéro de téléphone masqué, comme l’exige la politique de confidentialité de la banque, et l’agent a inventé un numéro ayant l’air réaliste pour remplir le champ phone_number de l’outil de notification. Le message a été envoyé à la mauvaise personne.
Ces deux erreurs étaient invisibles au vu de la qualité de la réponse. La réponse finale adressée à l’analyste semblait correcte. Les conséquences se manifestaient sous forme d’effets secondaires : des coûts supplémentaires liés à l’API et une notification envoyée à la mauvaise adresse. Telle est la réalité quotidienne de l’utilisation des outils par les agents en environnement de production, et le reste de cet article explique comment y remédier.
Décider quand appeler un outil
L’erreur la plus fondamentale consiste à recourir à un outil dont personne n’a besoin, ou au mauvais outil. Cela arrive généralement lorsque le choix est dicté par une similitude sémantique entre la requête et la description de l’outil, plutôt que par une véritable nécessité.
Prenez l’outil Bureau, qui permet de récupérer le score Bureau, l’historique des paiements et le taux d’utilisation du crédit d’un client. Un agent qui réfléchit au solde d’un prêt peut choisir cet outil simplement parce que le mot « crédit » apparaît à la fois dans le contexte de la requête et dans sa description. Les données Bureau pourraient être pertinentes, mais elles ne sont pas nécessaires pour répondre à cette question, et c’est là la distinction que l’agent a manquée.
De meilleures descriptions seules ne suffisent pas. Ce qui aide, c’est une étape de justification structurée. Avant toute appel, l’agent doit répondre à une question : quelles informations spécifiques cet outil fournit-il qui ne sont pas déjà disponibles dans les données existantes ou lors d’un autre appel déjà planifié ? Si la réponse est « rien d’essentiel », l’outil reste inutilisé.
Trois idées soutiennent cette étape :
- Un graphe de dépendance des outils indique quels outils doivent être exécutés avant les autres. Le calcul d’un EMI nécessite le montant du prêt et le taux d’intérêt, informations provenant de la recherche dans le compte du prêt ; par conséquent, cette recherche est effectuée en premier. Rendre le flux d’informations explicite permet d’éviter des appels redondants ou inutiles, et les données déjà récupérées au cours de la session n’ont pas besoin d’être récupérées à nouveau.
- Le principe du ensemble d’outils minimal stipule que la sélection appropriée correspond au plus petit groupe d’outils capable de fournir conjointement tout ce dont la requête a besoin. Tout appel supplémentaire est soit redondant, soit inutile.
- Appliquer ce principe signifie que l’agent énumère les informations dont il a besoin, associe chaque élément à un outil et vérifie s’il dispose déjà des données avant de planifier un autre appel.
Le problème des paramètres hallucinés
L’incident lié au numéro de téléphone relève d’une catégorie d’erreur plus dangereuse que le choix du mauvais outil. Ici, l’outil est correct, mais ses paramètres proviennent de la génération propre du modèle plutôt que de données récupérées.
Cela se produit parce que les LLM génèrent naturellement. Lorsqu’un paramètre requis n’est pas présent dans le contexte, le comportement par défaut est de produire une valeur plausible, de la même manière que le texte est généré. En demandant un numéro de téléphone plausible, on obtient dix chiffres qui semblent corrects mais ne correspondent à personne en particulier.
La solution comporte deux parties.
Suivi de l’origine des paramètres. Chaque valeur de paramètre doit être traçable jusqu’à une source : une sortie d’une appel à outil précédent, un texte extrait de la conversation, quelque chose fourni par l’utilisateur ou une valeur explicitement étiquetée comme générée. Les valeurs générées sont marquées pour examen avant que l’outil ne s’exécute, et non après.
Vérification des champs obligatoires avant exécution. Avant toute utilisation d’outils ayant des effets secondaires, tels que les notifications, les escalades, les transactions ou d’autres modifications d’état, l’agent s’assure que chaque paramètre requis provient d’une récupération vérifiée et non de texte généré. Si un paramètre ne peut pas être vérifié, l’appel est bloqué et l’agent soit récupère la valeur manquante, soit demande à l’utilisateur.
Les outils en lecture seule, comme les recherches de comptes, les recherches de politiques et les calculs, présentent un risque moindre, car le pire scénario est une recherche infructueuse plutôt qu’une action incorrecte. Pour les outils qui modifient l’état ou prennent contact avec des tiers, la vérification des paramètres générés est indispensable.
Réessais et solutions de secours
Les appels externes échouent en production en raison de délais d’attente, de limites de fréquence, d’interruptions de service et de paramètres rejetés. La manière dont l’agent réagit dans de telles situations détermine si l’expérience se dégrade de manière souple ou grave. Une approche en trois niveaux, adaptée au type d’échec, s’avère efficace.
Niveau un : réessayer les échecs temporaires. Les délais de réseau, les réinitialisations de connexion et les réponses HTTP 503 sont probablement temporaires ; il convient donc de réessayer immédiatement en utilisant un mécanisme de backoff exponentiel et les mêmes paramètres, inchangés. Trois tentatives à 500 ms, 1 000 ms et 2 000 ms constituent une limite raisonnable. Ce niveau permet de gérer la plupart des échecs sur des points d’accès à haute disponibilité.
Niveau deux : essayer à nouveau avec des paramètres corrigés. Un code HTTP 400 ou une erreur de validation signifie que l’outil a rejeté ce qui lui a été envoyé, et non qu’il y a eu un problème réseau. L’agent doit relire ce qu’il a envoyé, interpréter le message de validation et générer une version corrigée avant de tenter à nouveau. Une seule tentative de correction est autorisée. Si les paramètres corrigés échouent également, l’agent doit cesser de tenter des hypothèses.
Niveau trois : utiliser un outil de secours. Lorsqu’un outil est indisponible ou continue de fonctionner incorrectement, un outil similaire fournissant des informations similaires, voire moins précises, peut exister. Si la requête auprès du bureau d’évaluation ne fonctionne pas, l’agent peut recourir à l’évaluation de crédit interne basée sur les dernières données connues du bureau stockées dans son système de mémoire. La réponse est moins à jour mais reste utile, et cela est clairement indiqué, par exemple en précisant que des données en cache provenant de la dernière évaluation sont utilisées car l’API du bureau n’est pas accessible.
Si tous les niveaux d’analyse échouent, l’agent doit le signaler à haute voix au lieu de continuer sans les informations manquantes. Un bon message informe le client que le score du bureau de crédit actuel ne pouvait pas être obtenu en raison d’un problème temporaire du système, propose une réponse concernant l’éligibilité au prêt en se basant sur les données disponibles, et suggère de réessayer dans quelques minutes pour obtenir une évaluation plus précise. C’est honnête, pratique et exempt de faits inventés.
Observabilité des outils
Les requêtes inutiles du bureau de crédit effectuées par la banque n’apparaissaient jamais dans les indicateurs de qualité. Les réponses étaient correctes, les scores RAGAS étaient bons et la satisfaction des utilisateurs était acceptable. Le problème ne se manifestait que dans les journaux d’appels des outils, et ce, uniquement après que quelqu’un ait calculé le coût par outil.
Quatre indicateurs par outil, suivis selon le type de requête au fil du temps, révèlent ces problèmes :
- Taux d’appel indique la fréquence à laquelle chaque outil est utilisé pour chaque type de requête. Un outil du bureau qui s’active dans 80 % des requêtes est presque certainement sollicité de manière excessive, car aucun assistant financier en production n’a réellement besoin des données du bureau lors de quatre sur cinq conversations. Une alerte lorsque l’outil dépasse un seuil pour son type de requête signale des appels inutiles.
- Taux d’erreur de paramètre compte le nombre de fois où les appels sont rejetés au niveau de la validation. Un outil présentant 15 % de rejets reçoit soit systématiquement des types incorrects, soit des valeurs qui échouent aux vérifications de provenance. Dans un cas comme dans l’autre, l’intégration avec le raisonnement de l’agent est défectueuse.
L’implémentation
Le système de production complet est un seul module long de près d’un millier de lignes, couvrant le registre des outils, le suivi de l’origine des données, le validateur, la logique de tentative répétée et le suivi des métriques.
"""
#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())
Que fait le code
Le registre des outils stocke le schéma complet de chaque outil ainsi que sa catégorie. fetch_bureau_score est marqué comme uniquement en lecture, mais son utilisation coûte 8 ₹ par appel ; sa description interdit son emploi pour des demandes de solde, des calculs d’EMI ou des questions relatives aux polices. send_notification est classé comme outil de notification, ce qui déclenche la validation stricte préalable réservée aux opérations d’écriture. Le graphe de dépendances indique que send_notification dépend de get_loan_status, ce qui signifie que des coordonnées de contact vérifiées doivent être obtenues avant d’envoyer toute notification.
Le suiveur de provenance conserve, au niveau de la session, toutes les valeurs récupérées lors d’une appel précédent ou fournies par l’utilisateur. Lorsque l’agent remplit le champ phone_number pour l’action send_notification, le suiveur vérifie si ce numéro provient d’une récupération antérieure. L’outil de suivi du statut du prêt a renvoyé un numéro masqué, « +91-98XXXXX123 », de sorte que si l’agent tente d’utiliser un numéro complet non masqué, la vérification indique qu’il s’agit d’un numéro généré. L’agent ne peut donc pas avoir obtenu un numéro complet à partir d’une source masquée.
Le vérificateur préalable à l’exécution bloque alors l’appel et renvoie une erreur explicite. Il indique que le phone_number requis pour l’outil de notification provient d’une source générée et doit d’abord être récupéré via l’outil de profil du client. L’appel est interrompu avant même d’être exécuté, et non après l’envoi d’un SMS incorrect.
Le suivi de l’observabilité accumule des métriques par outil au fil des requêtes. Après seulement trois requêtes de test, il affiche les taux d’appel, les coûts et le phénomène d’appels excessifs. Son calcul des économies met en évidence précisément ces appels excessifs de la part du bureau, qui coûtaient 16 000 roupies par jour, et cela dès le premier jour de déploiement.
Les descriptions sont des spécifications
La qualité du choix de l’outil est directement liée à la qualité de sa description. Un outil mal décrit est sélectionné pour des tâches qu’il ne peut pas effectuer, tandis que des tâches qu’il peut accomplir sont négligées.
Trois types d’échecs apparaissent le plus fréquemment :
- Vagueness. « Obtenir des informations sur le client » pourrait désigner l’outil relatif à l’état du prêt, l’outil de profilage ou l’outil du bureau. Face à trois options tout aussi vagues, l’agent se base sur des mots-clés superficiels plutôt que sur la portée réelle de l’outil.
get_loan_status ou customer_profile" rend cette dépendance visible dans la description, avant même que les règles structurelles du graphe ne s’appliquent.Une bonne description est une spécification, et non du texte marketing. Elle indique ce que l’outil fournit, ce dont il a besoin, quand l’utiliser et quand ne pas l’utiliser.
Trois mois plus tard
Après trois mois d’utilisation du suivi de l’origine des données et de l’observabilité des outils, les résultats de la banque étaient concrets.
Le problème de sur-appel est apparu dès le deuxième jour du déploiement des fonctionnalités d’observabilité, avant même que quiconque ne commence à examiner les données. L’alerte relative au taux d’appels a été déclenchée lorsque fetch_bureau_score a montré un taux d’appels de 78 %, contre un seuil de 50 %, et le calcul des coûts indiquait des économies potentielles de 8 800 roupies par jour. Après l’ajout de formulations excluant explicitement ce cas, le taux d’appels en provenance du bureau est passé de 78 % à 29 % en trois jours.
Le problème lié au numéro de téléphone est apparu durant la première semaine d’analyse rétrospective, à partir des enregistrements de provenance des appels bloqués. Au cours des deux semaines précédentes, l’outil de notification avait été utilisé 23 fois avec des numéros de téléphone générés, et tous ces numéros avaient été bloqués. Sans le validateur, cela aurait signifié 23 messages SMS envoyés aux mauvaises personnes.
Les niveaux de tentative ont résolu un problème de fiabilité que personne n’avait remarqué. L’API du bureau présentait un taux de temps d’attente temporaire de 4 %, et l’agent initial échouait systématiquement à chaque tentative. Le mécanisme d’ajustement exponentiel a permis de gérer 81 % de ces temps d’attente, réduisant ainsi les échecs visibles par les utilisateurs de 4 % à 0,7 %. Le reste reflétait des pannes réelles du bureau, et le système a correctement recouru à l’évaluation interne en mémoire tampon.
Les chiffres globaux :
- Le coût de l’API du bureau est passé de 32 000 ₹ à 11 600 ₹ par jour, soit une réduction de 64 %. Cette baisse est due en partie à un taux d’appels plus faible, et en partie à l’élimination des requêtes qui, selon les journaux, déclenchaient des appels au bureau en raison de requêtes antérieures dans une même session ; le suiveur d’origine et le résolveur de dépendances empêchent désormais cela.