实用说明：生产环境中的代理工具使用与函数调用——何时使用
《实用笔记》操作指南：生产环境中的代理工具使用与函数调用——适用场景：采用该模式的团队在处理合同、校验以及插入代码时使用。
可将此内容视为《生产环境中的智能体工具使用与函数调用——当智能体介入现实世界》一文面向操作员的简化版本：清晰的阶段划分、有序的代码模块，以及便于交接时参考的恢复说明。 将“概览”阶段视为可量化的界面使用效果最为有效。在扩大范围之前，先记录一份最佳操作案例、一个故障实例以及回滚说明。 应将配置信息与应用程序代码分开。环境文件、密钥存储和功能开关应集中存放于一个位置，这样操作员无需查看整个系统结构即可进行审核。
智能体如何决定调用哪些工具及何时调用、防止生成虚假结果的结构化输出处理机制、针对工具故障的重试与回退策略，以及用于识别问题所在的工具监控功能
对于“代理如何决定执行哪个阶段”的问题，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 在网关处进行身份验证，在数据层进行重新授权。仅凭承载令牌并不能作为租户边界。
工具使用决策——代理何时应该以及何时不应该调用工具
在“工具使用决策”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一责任主体，而非复杂的流程链。应在网关处进行身份验证，在数据层再次授权——仅凭承载令牌并不足以界定租户边界。
结构化工具输出处理——幻觉参数问题
在结构化工具输出处理阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与已验证输出之间的契约。为相关产物命名，定义成功检测条件，并拒绝默许的半完成状态。 在网关处进行身份验证，在数据层再次授权。仅凭承载令牌并不足以界定租户边界。 在结构化工具输出处理阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于操作人员可审计的位置，无需查看整个系统结构。
工具调用重试与回退架构
在处理工具调用重试阶段时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续代码修改的规范性。 同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。若没有这些记录，调试过程将会浪费大量时间。
工具可观测性——识别出存在问题的工具
在处理“工具可观测性分析”阶段时，首先需明确相关约定：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向具体的责任模块，而非复杂的流程链。 需为每次调用记录工具名称、参数哈希值、延迟时间以及执行结果。没有这些记录，调试代理循环将耗费大量时间。
代码——生产环境代理工具使用系统
在处理“代码生成代理”阶段时，首先需写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改始终符合约定。 将此阶段视为输入与验证后输出之间的契约。为生成的成果命名，明确成功判定标准，杜绝无声的半完成状态。 需记录每次调用的工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试代理循环将会耗费大量时间。
"""
#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())
刚刚发生了什么——用通俗语言解释
在处理“What Just Happened Plain”阶段时，首先需写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在代码从演示环境转向共享环境时出现意外费用。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试过程将会浪费大量时间。
工具描述问题——描述质量如何影响选择质量
在处理“工具描述问题”阶段时，首先需写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 需为每次调用记录工具名称、参数哈希值、延迟时间以及执行结果。如果没有这些记录，调试代理将陷入无止境的循环，耗费大量时间。
真实企业案例
在处理“真实企业场景”阶段时，首先写下合同规范：所需的输入参数、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试代理将陷入无止境的循环，耗费大量时间。
深入探讨
在深入研究阶段，首先需写下接口规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。
操作检查清单
在完成操作检查清单阶段时，同样要首先列出接口规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单有助于确保后续代码修改的准确性。
在记录功能结果的同时，还需标注执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
记录每次调用的工具名称、参数哈希值、延迟时间以及执行结果。如果没有这些追踪信息，调试代理将陷入无限循环，耗费大量时间。
保持数据结构简洁且类型明确。嵌套的数据结构会掩盖具体是哪个节点设置了哪个字段，还会在中断后导致流程无法继续。
只要预算允许，就在持续集成过程中使用测试用例而非真实的付费 API 来对关键路径进行压力测试。
优先选择小型、可测试的单元，而非结构复杂的脚本。当某个步骤失败时，故障应能指向具体的责任模块，而非整个混乱的流程。
在升级技术栈之前，先冻结现有版本，为关键路径保存标准化的操作记录，并确认回滚步骤。共享环境需要设置速率限制、租户验证机制，以及明确的密钥轮换负责人。与其追求华丽的临时演示，不如注重扎实的可靠性。
e491f945eee7的批量处理说明：不要将提供者密钥放入代码仓库，为每个会话设置令牌使用上限，并将转录内容存储在评估测试文件旁，以便后续模型更换时保持可比性。
强化措施的第0阶段若作为可测量的指标来处理效果最佳。在扩大范围之前，先记录一份理想的转录样本、一个失败案例以及回滚说明。优先选择小型且可测试的单元，而非庞大的脚本；当某一步骤失败时，故障应能指向单一责任点，而非复杂的流程链。
强化措施细节0/881：针对此条说明需测量执行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非主观经验来决定是否保留该变更。
在强化措施的第一阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。除了功能测试结果外，还需记录执行时间以及代币或查询成本。提前了解这些成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。
强化措施细节 1/881：针对此项措施，需测量实际执行时间、错误类型以及代币消耗情况，然后依据固定的评估标准而非主观判断来决定是否保留该变更。
在处理强化措施的第2阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合要求。
同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续才添加的完善措施。
强化措施细节2/881：需测量该环节的耗时、错误类型以及令牌消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该变更。
将强化措施的第3阶段视为可量化的处理面最为有效。在扩大范围之前，先记录一份理想的操作流程、一个故障案例以及回滚说明。
要把这一阶段视为输入参数与验证后输出结果之间的契约。为相关文档命名，明确成功判定标准，杜绝无声的半完成状态。
强化措施细节 3/881：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
在强化措施的第4阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。配置应置于应用程序代码之外，环境文件、密钥存储以及功能标志应集中存放于一个操作人员可以审核的位置，无需查看整个系统结构。
强化措施细节 4/881：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
在处理强化措施的第5阶段时，首先写下相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合预期。 相比冗长的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障应能指向单一责任点，而非复杂的流程链。
强化措施细节5/881：需测量该步骤的运行时间、错误类型以及代币消耗情况，然后依据固定的评估标准而非主观判断来决定是否保留该修改。
将强化措施的第6阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份理想的运行示例、一个失败案例以及回滚说明。 在功能结果旁同时记录时间消耗及代币或查询成本。提前明确成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
强化措施细节6/881：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个别案例来决定是否保留该变更。
在强化措施的第7阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续的优化工作。
强化措施细节7/881：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个别案例来决定是否保留该变更。
在处理强化措施的第8阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合约定。 将这一阶段视为输入与验证后输出之间的契约。为相关产物命名，明确成功判定标准，杜绝默许的部分完成情况。
强化措施细节8/881：需测量该步骤的耗时、错误类型以及令牌消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该变更。
将强化措施的第9阶段视为可量化的目标面来处理效果最佳。在扩大范围之前，先记录一份标准操作示例、一个失败案例以及回滚说明。 应将配置信息与应用程序代码分开。环境文件、密钥存储和功能开关应集中存放于一处，以便操作人员无需查看全部代码结构即可进行审计。
强化措施细节 9/881：测量该任务的执行时间、错误类型以及代币消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
在强化措施的第10阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比复杂的脚本，更应采用小型且可测试的单元。当某一步骤失败时，故障原因应能指向单一责任方，而非混乱的整个流程。
强化措施细节 10/881：测量该任务的执行时间、错误类型以及代币消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。