Использование производственных инструментов агентами: происхождение данных, повторные попытки и возможности наблюдения.
Агент банка неправильно использовал платный API и придумал номер телефона. Изучите минимальные наборы инструментов, происхождение параметров, трехуровневую систему повторных попыток и метрики для каждого инструмента.
Источники: Agent Tool Use and Function Calling in Production — When Agents Reach Into the World
Команда по кредитным операциям в банке в Ченнаи обеспечила своих аналитиков вспомогательным инструментом с восемью функциями. Четыре из них предназначались для поиска данных (информация о кредитных счетах, история транзакций, профили клиентов и данные кредитных бюро), а остальные использовались для расчетов по ежемесячным платежам, поиска документов политики, отправки уведомлений и передачи дел на более высокий уровень обработки. В течение двух недель проверка качества выявила две проблемы, которые полностью прошли незамеченными при стандартных проверках качества ответов.
Первой проблемой было ненужное использование инструментов. Для простого запроса о балансе требовался лишь поиск информации о кредитном счете, однако агент также выполнял запрос в кредитное бюро. Это приводило к увеличению времени обработки на 800 мс, а каждый запрос в бюро стоил 8 рупий. При 2000 запросах в день ненужные операции добавляли еще 16 000 рупий в день к стоимости реально необходимых запросов.
Второй пример — это вымышленный параметр. Поиск кредита вернул засекреченный номер телефона, как того требует политика конфиденциальности банка, и агент придумал номер, выглядящий правдоподобно, чтобы заполнить поле phone_number в инструменте уведомлений. Сообщение попало не тому человеку.
Оба сбоя оставались незаметными с точки зрения качества ответа. Окончательный ответ для аналитика выглядел корректно. Ущерб проявился в побочных эффектах: расходах на API и неправильно направленном уведомлении. Такова повседневная реальность использования инструментов агентами в производственных условиях, а далее в этой статье рассматривается, как этого избежать.
Решение о том, когда вообще использовать инструмент
Самый распространенный сбой — это обращение к инструменту, который никому не нужен, или к неправильному инструменту. Это обычно происходит, когда выбор основывается на семантическом сходстве между запросом и описанием инструмента, а не на реальной необходимости в его использовании.
Возьмем инструмент бюро кредитных историй, предназначенный для получения информации о рейтинге клиента, истории платежей и уровне использования кредита. Сотрудник, рассматривающий сумму кредита, может выбрать его просто потому, что слово «кредит» встречается как в контексте запроса, так и в описании. Данные бюро могут быть релевантными, но для ответа на этот вопрос они не являются обязательными, и именно эту разницу сотрудник упустил.
Одних лишь лучших описаний недостаточно. Помогает структурированный шаг обоснования. Перед любым звонком сотрудник должен ответить на один вопрос: какую конкретную информацию предоставляет этот инструмент, которой ещё нет в существующих данных или в результате другого уже запланированного звонка? Если ответ — «ничего существенного», инструмент остаётся невостребованным.
Три идеи способствуют реализации этого шага:
- Граф зависимостей инструментов фиксирует, какие инструменты должны запуститься раньше других. Для расчёта EMI необходимы сумма кредита и процентная ставка, которые берутся из поиска информации о кредитном счёте, поэтому этот поиск выполняется в первую очередь. Чёткое управление потоком информации предотвращает дублирование и ненужные вызовы, а данные, уже полученные ранее в ходе сессии, не требуют повторного получения.
- Принцип минимального набора инструментов гласит, что оптимальным является самый маленький набор инструментов, способный совместно обеспечить всё, что необходимо для выполнения запроса. Любой дополнительный вызов является либо избыточным, либо ненужным.
- Применение этого принципа подразумевает, что агент перечисляет необходимую информацию, соотносит каждый элемент с определённым инструментом и проверяет, уже имеются ли у него эти данные, прежде чем планировать следующий вызов.
Проблема галлюцинированных параметров
Проблема с номером телефона относится к категории ошибок, более опасных, чем использование неподходящего инструмента. Здесь инструмент выбран верно, но его параметры формируются самой моделью, а не на основе полученных данных.
Это происходит потому, что большие языковые модели по своей природе генерируют контент. Когда необходимый параметр отсутствует в контексте, стандартным поведением является генерация правдоподобного значения, так же как и при формировании текста. При запросе на правдоподобный номер телефона получаются десять цифр, которые выглядят корректно, но не принадлежат никому конкретному.
Решение имеет две части.
Отслеживание происхождения параметров. Каждое значение параметра должно иметь источник: результат предыдущего вызова инструмента, текст, извлеченный из разговора, информация, предоставленная пользователем, или значение, явно помеченное как сгенерированное. Сгенерированные значения отмечаются для проверки до запуска инструмента, а не после него.
Проверка обязательных полей перед выполнением. Перед использованием любого инструмента, вызывающего косвенные эффекты, таких как уведомления, передача запросов на более высокий уровень обработки, транзакции или другие изменения состояния, агент проверяет, что все обязательные параметры получены из достоверных источников, а не сгенерированы автоматически. Если параметр невозможно проверить, вызов блокируется, и агент либо получает отсутствующее значение, либо запрашивает его у пользователя.
Инструменты с ограниченными правами доступа, такие как поиск информации о счетах, поиск политик и выполнение расчётов, несут меньший риск, поскольку худшим исходом является неудачный поиск, а не неправильное действие. Для инструментов, которые изменяют состояние или взаимодействуют с третьими сторонами, проверка параметров, сгенерированных автоматически, является обязательной.
Повторные попытки и альтернативные решения
В реальных условиях внешние вызовы сбоют из-за истечения времени ожидания, ограничений по частоте запросов, простоев и некорректных параметров. То, как будет вести себя агент в таких ситуациях, определяет, улучшится ли качество работы или оно существенно ухудшится. Эффективно работает трехуровневый подход, применяемый в зависимости от типа сбоя.
Первый уровень: повторная попытка при временных сбоях. Истечение времени ожидания в сети, сбросы соединений и ответы HTTP 503, скорее всего, являются временными проблемами, поэтому следует немедленно попробовать снова с использованием экспоненциального затухания интервалов и без изменений первоначальных параметров. Разумным лимитом будет три попытки с интервалами в 500 мс, 1 000 мс и 2 000 мс. Этот уровень позволяет справиться с большинством сбоев в системах с высокой доступностью.
Второй уровень: повторная попытка с исправленными параметрами. Ошибка HTTP 400 или ошибка валидации означают, что инструмент отклонил отправленные данные, а не то, что произошла проблема с сетью. Агент должен перечитать отправленные данные, проанализировать сообщение об ошибках валидации и сформировать исправленный набор параметров перед новой попыткой. Разрешается только одна попытка коррекции. Если даже с исправленными параметрами возникают проблемы, агенту следует прекратить попытки.
Третий уровень: использование альтернативного инструмента. Когда тот или иной инструмент не работает или постоянно дает сбои, может существовать связанный инструмент с похожей, возможно менее точной информацией. Если запрос в бюро не удаётся, сотрудник может обратиться к внутренней оценке кредитоспособности, основанной на последних известных данных бюро в его системе памяти. Ответ менее актуален, но всё ещё полезен, и в ответе прямо указывается на это, например, сообщается о том, что используются данные бюро из кэша с последней оценки, поскольку API бюро недоступно.
Если любой уровень проверки не сработает, агент обязан это громко сказать, вместо того чтобы продолжать работу без необходимой информации. Хорошее сообщение сообщает клиенту, что текущий балл в кредитном агентстве невозможно получить из-за временной проблемы с системой, предлагает ответ относительно возможности получения кредита на основе имеющихся данных и советует попробовать снова через несколько минут для более точной оценки. Это честно, практично и не содержит выдуманных фактов.
Возможность отслеживания для инструментов
Ненужные запросы банка в кредитные агентства никогда не отражались в показателях качества. Ответы были правильными, баллы RAGAS оставались высокими, а удовлетворенность пользователей была приемлемой. Проблема проявлялась только в журналах вызовов инструментов, и лишь тогда, когда кто-то рассчитал стоимость использования каждого инструмента.
Четыре показателя на каждый инструмент, отслеживаемые по типам запросов в течение времени, позволяют выявить эти проблемы:
- Частота вызова показывает, как часто используется каждый инструмент на единицу типа запроса. Инструмент бюро, применяемый в 80% запросов, почти наверняка вызывается чрезмерно, поскольку ни один реальный финансовый ассистент не нуждается в данных бюро в четырех из пяти диалогов. Сигнал об опасности при превышении порогового значения для конкретного типа запроса указывает на ненужные вызовы.
- Частота ошибок параметров отражает количество случаев отклонения вызовов на уровне проверки. Инструмент с 15% отклонений либо постоянно получает неверные типы данных, либо к нему передаются значения, не проходящие проверку происхождения. В любом случае интеграция с механизмом принятия решений агента имеет недостатки.
Реализация
Полная производственная система представляет собой один большой модуль из почти тысячи строк, включающий реестр инструментов, трекер происхождения данных, валидатор, логику повторных попыток и трекер метрик.
"""
#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())
Что делает код
Реестр инструментов хранит полную схему каждого инструмента вместе с его категорией. Функция fetch_bureau_score отмечена как только для чтения, но её использование стоит 8 рупий за вызов; в описании указано, что её нельзя применять для запроса остатка средств, расчёта ежемесячных платежей или получения информации о полисе. Функция send_notification отнесена к инструментам уведомлений, что активирует более строгую проверку перед выполнением, предусмотренную для операций записи. Граф зависимостей показывает, что send_notification зависит от get_loan_status, то есть перед отправкой любого уведомления необходимо сначала получить проверенные контактные данные.
Отслеживатель происхождения данных хранит на уровне сессии все значения, полученные из предыдущего запроса или введённые пользователем. Когда агент заполняет поле phone_number для функции send_notification, отслеживатель проверяет, поступило ли это число из ранее полученных данных. Инструмент для отслеживания статуса кредита вернул замаскированный номер «+91-98XXXXX123», поэтому если агент попытается использовать полный, незамаскированный номер, система отметит его как сгенерированный. Агент не мог получить полный номер из источника с замаскированными данными.
Валидатор перед выполнением затем блокирует вызов и возвращает чёткую ошибку. В ней указывается, что необходимый для инструмента уведомлений phone_number имеет сгенерированное происхождение и сначала должен быть получен из инструмента профиля клиента. Вызов прерывается до его выполнения, а не после отправки неверного SMS.
Инструмент отслеживания наблюдаемости собирает показатели работы каждого инструмента в ходе выполнения запросов. Уже после трех тестовых запросов он показывает частоту вызовов, стоимость и случаи чрезмерного использования инструментов. Расчет экономии помогает выявить именно те случаи чрезмерного использования инструментов, которые стоили 16 000 рупий в день; это становится видно уже в первый день внедрения.
Описания — это спецификации
Качество выбора инструмента напрямую зависит от качества описания. Инструмент с плохим описанием может быть выбран для выполнения задач, которые он не может решить, в то время как инструменты с более точным описанием могут быть проигнорированы при необходимости.
Чаще всего встречаются три проблемы:
- Неоднозначность. Фраза «получение информации о клиенте» может относиться к инструменту для проверки статуса кредита, к инструменту для анализа профиля или к инструменту бюро. Сталкиваясь с тремя одинаково неоднозначными вариантами, агент опирается на поверхностные ключевые слова, а не на объем задачи.
get_loan_status или customer_profile» делает эту зависимость видимой в описании ещё до того, как начнет действовать структурное контролирование графа.Хорошее описание — это спецификация, а не маркетинговый текст. В нем указывается, что предоставляет инструмент, чего он требует, когда его использовать и когда — нет.
Через три месяца
После трех месяцев использования функций отслеживания происхождения данных и наблюдаемости за инструментами результаты банка стали конкретными.
Проблема чрезмерного количества звонков проявилась на второй день внедрения системы отслеживания, ещё до того, как кто-либо начал анализ. Сигнал об уровне звонков был активирован, когда функция fetch_bureau_score показала уровень звонков в 78%, что превышало установленный порог в 50%, а расчёт затрат указал на возможную экономию в размере 8 800 рупий в день. После того как в описании были добавлены прямые указания на исключение определённых случаев, уровень звонков от этого источника снизился с 78% до 29% в течение трёх дней.
Проблема с номерами телефонов возникла в первую неделю ретроспективного анализа на основе записей о блокированных звонках. За предыдущие две недели инструмент уведомлений использовался 23 раза с генерируемыми номерами телефонов, и все 23 номера были заблокированы. Без инструмента проверки это привело бы к отправке 23 SMS-сообщений неправильным людям.
Уровни повторных попыток устранили проблему надежности, которую никто не замечал. У API бюро наблюдалась ставка временных сбоев в 4%, и первоначальный агент просто терпел неудачу при каждой такой попытке. Метод экспоненциального откладывания снизил долю этих сбоев на 81%, уменьшив количество видимых для пользователей сбоев с 4% до 0,7%. Оставшаяся часть отражала реальные периоды простоя бюро, и в таких случаях система автоматически переходила на использование кэшированной внутренней оценки.
Общие показатели:
- Расходы на API бюро снизились с ₹32 000 до ₹11 600 в день, что на 64% меньше. Частично это было связано с уменьшением количества вызовов, а частично — с устранением запросов, которые, согласно логам, вызывали обращения к бюро из-за последствий предыдущих запросов в сессии; теперь трекер происхождения данных и решатель зависимостей предотвращают это.