Einsatz von Produktionswerkzeugen für Agenten: Herkunft, Wiederholungsversuche und Beobachtbarkeit
Ein Agent einer Bank rief eine bezahlte API zu oft an und erfand eine Telefonnummer. Erlernen Sie Mindestwerkzeugsets, Parameterherkunft, dreistufige Wiederholungsversuche sowie Metriken pro Werkzeug.
Quellen: Agent Tool Use and Function Calling in Production — When Agents Reach Into the World
Ein Team für Kreditgeschäfte bei einer Bank in Chennai stellte seinen Analysten einen Assistenten mit acht Tools zur Verfügung. Vier davon dienten zum Abrufen von Daten (Kreditkonten, Transaktionshistorie, Kundenprofile sowie das Kreditschufa-Büro), während die übrigen Tools für Berechnungen zu Ratenzahlungen, Suche nach Vertragsdokumenten, Versendung von Benachrichtigungen sowie die Weiterleitung von Fällen zuständig waren. Innerhalb von zwei Wochen zeigte eine Qualitätsprüfung zwei Probleme auf, die bei den üblichen Qualitätskontrollen völlig übersehen worden waren.
Das erste Problem bestand im unnötigen Einsatz der Tools. Für eine einfache Kontostandsanfrage reichte bereits die Abfrage des Kreditkontos aus, doch der Assistent führte zusätzlich auch eine Anfrage beim Kreditschufa-Büro durch. Dadurch entstand eine Verzögerung von 800 Millisekunden sowie ein Kostenfaktor von 8 Rupien pro Abfrage. Bei 2.000 Abfragen pro Tag führten diese unnötigen Anfragen zu zusätzlichen Kosten von 16.000 Rupien pro Tag obendrauf zu den tatsächlich notwendigen Anfragen.
Der zweite Fall betraf einen halluzinierten Parameter. Die Kreditabfrage hatte eine maskierte Telefonnummer zurückgegeben, wie es die Datenschutzrichtlinie der Bank vorschreibt, und der Agent erfand eine realistisch aussehende Nummer, um das phone_number-Feld des Benachrichtigungstools zu füllen. Die Nachricht gelangte somit an die falsche Person.
Sowohl Fehlerarten waren anhand der Qualität der Antwort nicht erkennbar. Die endgültige Antwort an den Analysten sah in Ordnung aus. Der Schaden zeigte sich in Nebeneffekten: höheren API-Kosten sowie einer fehlerhaft zugestellten Benachrichtigung. Das ist die alltägliche Realität beim Einsatz von Agentenwerkzeugen in der Produktion, und der Rest dieses Artikels erläutert, wie man dies vermeiden kann.
Entscheiden, wann überhaupt ein Tool aufgerufen werden soll
Der grundlegendste Fehler besteht darin, nach einem Tool zu greifen, das niemand benötigt oder das falsche. Dies tritt oft dann auf, wenn die Auswahl durch semantische Ähnlichkeit zwischen der Anfrage und der Beschreibung eines Tools bestimmt wird, anstatt durch tatsächliche Notwendigkeit.
Nehmen wir das Bureau-Tool, das dazu dient, die Bureau-Bewertung, die Zahlungshistorie sowie die Kreditnutzung eines Kunden abzurufen. Ein Mitarbeiter, der über den Kreditsaldo nachdenkt, könnte es einfach auswählen, weil das Wort „Kredit“ sowohl im Abfragekontext als auch in der Beschreibung vorkommt. Bureau-Daten könnten relevant sein, sind aber für diese Frage nicht notwendig – und genau diesen Unterschied hat der Mitarbeiter übersehen.
Bessere Beschreibungen allein lösen dieses Problem nicht. Hilfreich ist ein strukturierter Begründungsschritt. Vor jedem Anruf muss der Mitarbeiter eine Frage beantworten: Welche spezifischen Informationen liefert dieses Tool, die nicht bereits aus vorhandenen Daten oder einem bereits geplanten anderen Anruf verfügbar sind? Wenn die Antwort „nichts Wesentliches“ lautet, bleibt das Tool ungenutzt.
Drei Ansätze unterstützen diesen Schritt:
- Ein Abhängigkeitsgraph von Tools dokumentiert, welche Tools zuerst ausgeführt werden müssen. Für eine EMI-Berechnung sind der Kreditbetrag und die Zinssatz erforderlich, die aus der Abfrage des Kreditkontos stammen, weshalb diese Abfrage zuerst durchgeführt wird. Durch die explizite Darstellung des Informationsflusses werden doppelte und unnötige Aufrufe verhindert, und Daten, die bereits früher in der Sitzung abgerufen wurden, müssen nicht erneut geholt werden.
- Das Prinzip der minimalen Tool-Gruppe besagt, dass die richtige Auswahl die kleinste Gruppe von Tools ist, die gemeinsam alles liefern, was die Anfrage benötigt. Jeder zusätzliche Aufruf ist entweder überflüssig oder unnötig.
- Anwendung des Prinzips bedeutet, dass der Agent die benötigten Informationen auflistet, jeden Punkt auf ein Tool abbildet und prüft, ob er die Daten bereits besitzt, bevor er einen weiteren Aufruf plant.
Das Problem der halluzinierten Parameter
Der Vorfall mit der Telefonnummer gehört zu einer Fehlerklasse, die gefährlicher ist als das Verwenden des falschen Tools. Hier ist zwar das Tool richtig, doch seine Parameter stammen aus der eigenen Generierung des Modells statt aus abgerufenen Daten.
Dies tritt auf, weil LLMs von Natur aus generieren. Wenn ein erforderlicher Parameter nicht im Kontext vorhanden ist, ist das Standardverhalten die Erzeugung eines plausiblen Wertes – genauso wie bei der Textgenerierung. Wenn man eine plausible Telefonnummer anfordert, erhält man zehn Ziffern, die richtig aussehen, aber niemandem speziell gehören.
Die Lösung besteht aus zwei Teilen.
Verfolgung der Herkunft der Parameter. Jeder Parameterwert muss auf eine Quelle zurückverfolgbar sein: Ausgabe einer früheren Toolaufruf, aus dem Gespräch extrahierter Text, etwas, was der Benutzer bereitgestellt hat, oder ein Wert, der ausdrücklich als generiert gekennzeichnet ist. Generierte Werte werden vor dem Ausführen des Tools zur Überprüfung markiert, nicht danach.
Validierung der Pflichtfelder vor Ausführung. Bevor ein Tool mit Nebeneffekten wie Benachrichtigungen, Eskalationen, Transaktionen oder anderen Zustandsänderungen verwendet wird, prüft der Agent, ob jeder erforderliche Parameter aus einer überprüften Abfrage stammt und nicht aus generiertem Text. Wenn ein Parameter nicht überprüft werden kann, wird der Aufruf blockiert und der Agent entweder den fehlenden Wert abruft oder den Benutzer fragt.
Lese-only-Tools wie Kontenabfragen, Richtliniensuchen und Berechnungen bergen ein geringeres Risiko, da das schlimmste Ergebnis eine fehlgeschlagene Abfrage ist und nicht eine falsche Aktion. Bei Tools, die den Zustand ändern oder externe Parteien kontaktieren, ist die Validierung der generierten Parameter unverzichtbar.
Wiederholungsversuche und Alternativen
Außergewöhnliche Anrufe scheitern in der Produktion aufgrund von Zeitüberschreitungen, Rate Limits, Ausfallzeiten sowie abgelehnten Parametern. Wie sich der Agent in solchen Fällen verhält, bestimmt, ob die Benutzererfahrung sanft oder stark beeinträchtigt wird. Ein dreistufiger Ansatz, der je nach Fehlerart angewandt wird, funktioniert gut.
Ebene eins: Wiederholte Versuche bei vorübergehenden Fehlern. Netzwerkzeitüberschreitungen, Verbindungsabbrüche sowie HTTP-503-Antworten sind vermutlich vorübergehend, daher sollten sie unverzüglich mit exponentiellem Backoff und den ursprünglichen, unveränderten Parametern wiederholt werden. Drei Versuche in Abständen von 500 ms, 1.000 ms und 2.000 ms bilden eine sinnvolle Obergrenze. Diese Ebene bewältigt die meisten Fehler bei Endpunkten mit hoher Verfügbarkeit.
Schritt zwei: Wiederholen mit korrigierten Parametern. Ein HTTP 400 oder ein Validierungsfehler bedeutet, dass das Tool den gesendeten Inhalt abgelehnt hat – nicht, dass es zu einem Netzwerkproblem gekommen ist. Der Agent sollte das Gesendete erneut lesen, die Validierungsmitteilung interpretieren und vor dem erneuten Versuch eine korrigierte Version erstellen. Es darf nur ein Korrekturvorgang durchgeführt werden. Wenn auch die korrigierten Parameter fehlschlagen, muss der Agent aufhören, zu raten.
Ebene drei: Verwendung eines Ersatztools. Wenn ein Tool nicht verfügbar ist oder ständig versagt, könnte es ein verwandtes Tool mit ähnlichen, möglicherweise weniger präzisen Informationen geben. Fällt die Abfrage bei der Auskunftei fehl, kann der Mitarbeiter auf die interne Kreditbewertung zurückgreifen, die auf den letzten bekannten Daten der Auskunftei aus seinem Speichersystem basiert. Die Antwort ist weniger aktuell, aber dennoch nützlich – und das wird klar mitgeteilt, beispielsweise durch die Angabe, dass aufgrund der Unverfügbarkeit der API der Auskunftei Daten aus dem Cache der letzten Bewertung verwendet werden.
Falls alle Ebenen fehlschlagen, muss der Agent dies laut aussprechen, anstatt ohne die fehlenden Informationen weiterzumachen. Eine gute Nachricht teilt dem Kunden mit, dass aufgrund eines vorübergehenden Systemproblems keine aktuelle Kreditwürdigkeitsbewertung abrufbar war, bietet eine Einschätzung zur Kreditfähigkeit auf der Grundlage dessen, was verfügbar ist, und empfiehlt, in einigen Minuten erneut zu versuchen, um eine genauere Beurteilung zu erhalten. Das ist ehrlich, handlungsorientiert und frei von erfundenen Fakten.
Überwachbarkeit für Tools
Die unnötigen Abfragen bei den Kreditbüros durch die Bank tauchten nie in den Qualitätsmetriken auf. Die Antworten waren korrekt, die RAGAS-Werte gut und die Kundenzufriedenheit akzeptabel. Das Problem zeigte sich nur in den Protokollen der Tool-Aufrufe – und zwar erst, als jemand die Kosten pro Tool berechnete.
Vier Metriken pro Tool, die im Laufe der Zeit nach Anfrageart erfasst werden, machen diese Probleme sichtbar:
- Anrufrate zeigt an, wie oft jedes Tool pro Abfragenart verwendet wird. Ein Tool, das in 80 % der Abfragen aktiviert wird, wird fast sicher übermäßig aufgerufen, da kein echter Finanzassistent in vier von fünf Gesprächen tatsächlich Daten aus diesem Tool benötigt. Ein Alarm, wenn ein Tool einen Schwellenwert für seine Abfragenart überschreitet, weist auf unnötige Aufrufe hin.
- Fehlerrate der Parameter gibt an, wie oft Aufrufe in der Validierungsstufe abgelehnt werden. Ein Tool mit 15 % Abweisungen liefert entweder ständig falsche Datentypen oder wird mit Werten aufgerufen, die Prüfungen zur Herkunft nicht bestehen. In jedem Fall ist die Integration in das Denkvermögen des Assistenten fehlerhaft.
Die Implementierung
Das vollständige Produktionsystem besteht aus einem einzigen, langen Modul mit fast tausend Zeilen, das den Tool-Registry, den Provenienzverfolger, den Validierer, die Wiederholungslogik sowie den Metrikenverfolger umfasst.
"""
#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())
Was der Code tut
Das Tool-Register speichert das vollständige Schema jedes Tools zusammen mit einer Kategorie. fetch_bureau_score ist als nur zum Lesen markiert, kostet jedoch 8 ₹ pro Aufruf und weist eine Beschreibung auf, die den Einsatz für Abfrage des Kontostands, Berechnung von Ratenzahlungen oder Fragen zu Policen verbietet. send_notification ist als Benachrichtigungs-Tool gekennzeichnet, wodurch die strengeren Vorausvalidierungen vor der Ausführung aktiviert werden, die für Schreibvorgänge vorgesehen sind. Der Abhängigkeitsgraph zeigt an, dass send_notification von get_loan_status abhängt, was bedeutet, dass vor dem Versenden jeder Benachrichtigung verifizierte Kontaktdaten abgerufen werden müssen.
Der Provenienz-Tracker speichert auf Sitzungsebene jeden Wert, der aus einem früheren Aufruf abgerufen wurde oder vom Benutzer eingegeben wurde. Wenn der Agent den Wert für phone_number bei send_notification einträgt, überprüft der Tracker, ob diese Nummer aus einem Abruf stammt. Das Tool zur Überprüfung des Kreditstatus gab eine maskierte Nummer zurück, nämlich „+91-98XXXXX123“. Wenn der Agent daher versucht, eine vollständige, unmaskierte Nummer einzugeben, markiert der Tracker diese als generiert. Der Agent kann somit keine vollständige Nummer aus einer maskierten Quelle abgerufen haben.
Dann blockiert der Vor-Exekutions-Validator den Aufruf und gibt einen klaren Fehler an. Darin wird mitgeteilt, dass die für das Benachrichtigungs-Tool erforderliche phone_number eine generierte Quelle aufweist und zunächst aus dem Kundenprofil-Tool abgerufen werden muss. Der Aufruf wird bereits vor seiner Ausführung gestoppt, anstatt erst nach dem Versand einer fehlerhaften SMS entdeckt zu werden.
Der Observability-Tracker sammelt Metriken pro Tool über verschiedene Abfragen hinweg. Schon nach drei Testabfragen zeigt er die Aufrufraten, Kosten sowie das Problem des übermäßigen Aufrufs. Die Berechnung der Einsparungen macht den spezifischen Fall des übermäßigen Aufrufs durch das Büro sichtbar, der täglich 16.000 ₹ kostete – und das bereits am ersten Tag der Einführung.
Beschreibungen sind Spezifikationen
Die Qualität der Tool-Wahl hängt direkt von der Beschreibungqualität ab. Ein schlecht beschriebenes Tool wird für Aufgaben ausgewählt, die es nicht bewältigen kann, und übersehen wird stattdessen für Aufgaben, die es doch erledigen kann.
Drei Fehler treten am häufigsten auf:
- Vagheit. „Kundendaten abrufen“ könnte sich auf das Tool zum Status des Kredits, das Profiltool oder das Bürotool beziehen. Angesichts von drei ebenso vagen Optionen wählt der Agent aufgrund oberflächlicher Schlüsselwörter statt aufgrund des tatsächlichen Anwendungsbereichs aus.
get_loan_status oder customer_profile abgerufte verifizierte Telefonnummer“ wird die Abhängigkeit bereits in der Beschreibung sichtbar, noch bevor die strukturellen Vorgaben des Graphen wirksam werden.Eine gute Beschreibung ist eine Spezifikation, keine Werbetexte. Sie beschreibt, was das Tool bietet, was es erfordert, wann es verwendet werden sollte und wann nicht.
Drei Monate später
Nach drei Monaten mit Nachverfolgung der Herkunft und Überwachbarkeit der Tools waren die Ergebnisse der Bank konkret.
Das Problem mit den übermäßigen Anrufen trat bereits am zweiten Tag der Einführung der Überwachungsfunktionen auf, noch bevor jemand damit begann, genauer hinzuschauen. Der Alarm für die Anrufrate wurde ausgelöst, als fetch_bureau_score eine Anrufrate von 78 % gegenüber einem Schwellenwert von 50 % anzeigte, und die Kostenschätzung ergab potenzielle Einsparungen von 8.800 ₹ pro Tag. Nachdem in der Beschreibung ausdrückliche Ausschlussformulierungen hinzugefügt wurden, sank die Anrufrate des Dienstleisters innerhalb von drei Tagen von 78 % auf 29 %.
Das Problem mit der Telefonnummer tauchte in der ersten Woche der rückblickenden Analyse aus den Protokollen der blockierten Anrufe auf. In den vorangegangenen zwei Wochen wurde das Benachrichtigungstool 23 Mal mit generierten Telefonnummern verwendet, und alle 23 Nummern wurden blockiert. Ohne den Validierer wären damit 23 SMS-Nachrichten an die falschen Personen gesendet worden.
Die Wiederholungsstufen behoben ein Zuverlässigkeitsproblem, das niemand bemerkt hatte. Die Bureau-API wies eine Transient-Timout-Rate von 4 % auf, und der ursprüngliche Agent scheiterte bei jedem solchen Fall. Durch exponentielles Backoff konnten 81 % dieser Timouts abgefangen werden, wodurch die von den Nutzern sichtbaren Fehlerraten von 4 % auf 0,7 % sanken. Der verbleibende Anteil spiegelte tatsächliche Ausfallzeiten der Bureau-API wider, und es wurde korrekt auf die im Cache gespeicherte interne Bewertung zurückgegriffen.
Die Gesamtzahlen:
- Die Kosten für die Bureau-API sanken von ₹32.000 auf ₹11.600 pro Tag, was einer Reduzierung von 64 % entspricht. Ein Teil dieser Ersparnis resultierte aus der geringeren Anzahl der Aufrufe, ein weiterer Teil aus der Beseitigung von Abfragen, die laut Protokollen durch vorherige Abfragen in derselben Sitzung zu Aufrufen bei der Bureau-API führten – was nun durch den Ursprungsverfolger und den Abhängigkeitslöser verhindert wird.