Startseite / Artikel / Praktische Hinweise: Verwendung von Agent-Tools und Aufruf von Funktionen in der Produktion – Wann

Praktische Hinweise: Verwendung von Agent-Tools und Aufruf von Funktionen in der Produktion – Wann

Schritt-für-Schritt-Anleitung zu den Praktischen Hinweisen: Verwendung von Agent-Tools und Aufruf von Funktionen in der Produktion – Wann: Verträge, Überprüfungen sowie Code-Blöcke für Teams, die dieses Muster einsetzen.

6300 Wörter

Nutzen Sie dies als für Operator zugängliche Neuformulierung der Ideen aus „Agent Tool Use and Function Calling in Production — When Agents Reach Into the World“: klare Phasen, geordnete Codeabschnitte sowie Wiederherstellungshinweise, die auch bei Übergaben erhalten bleiben. Die Überblicksphase funktioniert am besten, wenn sie als messbarer Überblick betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie die Notizen zur Rücksetzung. Bewahren Sie Konfigurationen außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Operator ohne das Lesen des gesamten Graphen überprüfen können.

Wie Agenten entscheiden, welche Tools sie wann aufrufen, strukturierte Ausgabeverarbeitung zur Vermeidung von Halluzinationsergebnissen, Wiederholungs- und Fallback-Muster bei Tool-Fehlern sowie Überwachungsmöglichkeiten, die zeigen, welche Tools Probleme verursachen

Für den Abschnitt „Wie entscheiden die Agenten welche Phase“ sollten Eingabedaten, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Authentifizieren Sie sich am Gateway und erteilen Sie erneut Berechtigungen auf der Datenebene. Ein alleiniges Trägertoken stellt keine Trennlinie zwischen verschiedenen Nutzern dar.

Die Entscheidung zur Tool-Nutzung – Wann sollten Agenten ein Tool aufrufen und wann nicht

Zur Phase der Entscheidung über die Verwendung des Tools sollten vor dem Ändern des Codes die Eingabedaten, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Vorzuziehen sind kleine, testbare Einheiten statt umfangreicher Skripte. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Authentifizieren Sie sich am Gateway und erteilen Sie erneut Berechtigungen auf der Datenebene. Ein alleiniges Trägertoken stellt keine Trennlinie zwischen verschiedenen Nutzern dar.

Gestützte Handhabung der Tool-Ausgaben – Das Problem der halluzinierten Parameter

Für die Phase der strukturierten Verarbeitung von Tool-Ausgaben sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Betreiber sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Authentifizieren Sie sich am Gateway und erteilen Sie erneut Berechtigungen auf der Datenebene. Ein alleinigesBearer-Token stellt keine Trennlinie zwischen verschiedenen Bereichen dar. Für die Phase der strukturierten Verarbeitung von Tool-Ausgaben sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Betreiber sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher sowie Feature-Flags sollten an einem Ort gespeichert werden, den die Betreiber ohne das Durchlesen des gesamten Systems überprüfen können.

Die Architektur für Wiederholungsversuche und Rückfallstrategien bei Toolaufrufen

Während der Phase der Wiederholungsversuche bei Toolaufrufen sollten Sie zunächst den Vertrag festhalten: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholungsversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Protokollieren Sie für jeden Aufruf den Toolnamen, den Hash der Argumente, die Latenzzeit sowie das Ergebnis. Ohne diese Aufzeichnungen verschwenden Debugging-Agenten wertvolle Stunden mit sinnlosem Herumprobieren.

Überwachbarkeit von Tools – Erkennen der Problemlöser

Wenn Sie die Phase „Tool Observability Finding“ durchlaufen, notieren Sie zunächst den Vertrag: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Protokollieren Sie für jeden Aufruf den Toolnamen, den Hash der Argumente, die Latenz sowie das Ergebnis. Ohne diese Aufzeichnungen verschwenden Sie beim Debuggen wertvolle Stunden.

Der Code – System zur Verwendung von Produktionsagenten-Tools

Während der Phase „The Code Production Agent“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Nennen Sie die Erzeugnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Protokollieren Sie für jeden Aufruf den Namen der Tool, den Hash der Argumente, die Latenzzeit sowie das Ergebnis. Ohne diese Aufzeichnungen verschwendet man Stunden beim Debuggen von Agentenschleifen.

"""
#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 gerade passiert ist – in einfachen Worten

Wenn Sie die Phase „What Just Happened Plain“ durchlaufen, notieren Sie zunächst den Vertrag: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Tragen Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen ein. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von einer Demo in gemeinsame Umgebungen wechselt. Protokollieren Sie für jeden Aufruf den Namen der Tool, den Hash der Argumente, die Latenzzeit sowie das Ergebnis. Ohne diese Aufzeichnungen verschwenden Debugging-Agenten Stunden mit sinnlosem Herumprobieren.

Das Problem der Toolbeschreibung – Wie die Qualität der Beschreibung die Auswahlqualität bestimmt

Während der Phase „The Tool Description Problem“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreuer ohne das Durchlesen des gesamten Systems überprüfen können. Protokollieren Sie für jeden Aufruf den Toolnamen, den Hash der Argumente, die Latenzzeit sowie das Ergebnis. Ohne diese Aufzeichnungen verschwenden Debugging-Agenten wertvolle Stunden mit endlosen Schleifen.

Eine reale Unternehmensgeschichte

Während der Phase „Real-World Enterprise Story“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikatoren sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallplan gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Protokollieren Sie für jeden Aufruf den Namen der Tool, den Hash der Argumente, die Latenzzeit sowie das Ergebnis. Ohne diese Aufzeichnungen verschwenden Sie bei der Fehlerbehebung Stunden.

Weiter vertiefen

Beim Arbeiten in der Phase „Going Deeper“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Protokollieren Sie für jeden Aufruf den Namen des Tools, den Hash der Argumente, die Latenzzeit sowie das Ergebnis. Ohne diese Aufzeichnungen verschwenden Sie bei der Fehlersuche wertvolle Stunden.

Operative Checkliste

Während der Phase der operativen Checkliste sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben.

Notieren Sie neben den funktionalen Ergebnissen auch die Laufzeiten sowie die Kosten für Tokens oder Abfragen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo in gemeinsame Umgebungen wechselt.

Protokollieren Sie den Namen des Tools, den Hash der Argumente, die Latenz sowie das Ergebnis jeder Aufruf. Ohne diese Aufzeichnungen verschwenden Debugging-Agenten Stunden damit, im Kreis zu laufen.

Halten Sie den Zustand des Graphen strukturiert und typisiert. Verschachtelte Datenstrukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen dazu, dass die Fortsetzung nach Unterbrechungen nicht möglich ist.

Fügen Sie immer dann, wenn das Budget es zulässt, einen Smoke-Test hinzu, der den kritischen Pfad in CI mit Fixtures und nicht mit live genutzten, bezahlten APIs testet.

Ziehen Sie kleine, testbare Einheiten vor großen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf.

Vor der Einführung neuer Komponenten sollten Sie die Versionen einfrieren, ein „goldenes“ Protokoll des kritischen Pfades anfertigen und die Schritte zum Rollback überprüfen. Gemeinsam genutzte Umgebungen benötigen Rate-Limits, Überprüfungen der Zuordnung sowie einen klaren Verantwortlichen für die Rotation von Geheimnissen. Ziehen Sie langweilige Zuverlässigkeit vor cleveren, einmaligen Demonstrationen.

Batch-Hinweis für e491f945eee7: Halten Sie die Anbieter-Schlüssel außerhalb des Repositories, legen Sie eine Obergrenze für Tokens pro Sitzung fest und speichern Sie die Transkripte neben den Evaluierungs-Dateien, damit spätere Modellwechsel vergleichbar bleiben.

Die Sicherheitsmaßnahme der Stufe 0 funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie die Notizen zum Rollback. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor – wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren.

Sicherheitsdetail 0/881: Messen Sie für diesen Hinweis die Ausführungsdauer, die Fehlerklasse sowie den Tokenverbrauch und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragebogens statt von Einzelbeobachtungen, ob die Änderung beibehalten werden soll.

Für die erste Stufe der Verstärkungsmaßnahmen sollten vor dem Ändern des Codes die Eingabedaten, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Neben den funktionalen Ergebnissen sollten Zeiten sowie Kosten für Token oder Abfragen aufgezeichnet werden. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt.

Detail der Verstärkungsmaßnahmen 1/881: Messen Sie die Gesamtlaufzeit, die Fehlerklasse sowie den Tokenverbrauch für diese Maßnahme und entscheiden Sie anschließend anhand eines festgelegten Fragebogens statt aufgrund von Einzelfallbeobachtungen, ob die Änderung beibehalten werden soll.

Beim Bearbeiten der zweiten Phase der Verstärkungsmaßnahmen sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikatoren sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie gleichzeitig den erfolgreichen Ablauf sowie den Wiederherstellungsprozess. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

Detail 2/881 der Verstärkungsmaßnahmen: Messen Sie die Ausführungsdauer, die Fehlerklasse sowie den Tokenverbrauch für diese Maßnahme und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragekatalogs – und nicht aufgrund von Einzelfällen –, ob die Änderung beibehalten werden soll.

Die dritte Phase der Verstärkungsmaßnahmen funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlfall sowie eine Notiz zur Rücksetzung. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die relevanten Dokumente, definieren Sie Erfolgsprüfungen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab.

Verstärkungsmaßnahme Detail 3/881: Messen Sie die Laufzeit, die Fehlerklasse sowie den Tokenverbrauch für diese Notiz und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragebogens statt aufgrund von Einzelfällen, ob die Änderung beibehalten werden soll.

Für die vierte Stufe der Verstärkungsmaßnahme sollten vor dem Ändern des Codes die Eingabedaten, der Verantwortliche für den Schritt sowie die Abschlusskriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf – Umgebungsdateien, Geheimdatenspeicher sowie Feature-Flags sollten an einem Ort gesammelt sein, den die Operator überprüfen können, ohne den gesamten Codeverlauf durchlesen zu müssen.

Verstärkungsmaßnahme Detail 4/881: Messen Sie die Laufzeit, die Fehlerklasse sowie den Tokenverbrauch für diese Notiz und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragebogens statt aufgrund von Einzelfällen, ob die Änderung beibehalten werden soll.

Beim Bearbeiten der Stufe 5 der Sicherheitsverbesserungen sollten Sie zunächst den Ablaufplan aufschreiben: erforderliche Eingaben, Erfolgsindikatoren sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich verweisen und nicht auf ein verworrenes Ablaufschema.

Sicherheitsdetail 5/881: Messen Sie die Ausführungsdauer, die Fehlerklasse sowie den Tokenverbrauch für diese Anweisung und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragekatalogs – und nicht nur aufgrund von Einzelfällen –, ob die Änderung beibehalten werden soll.

Die Stufe 6 der Sicherheitsverbesserungen funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlfall sowie eine Notiz zur Rücksetzung. Notieren Sie die Zeiten sowie den Token- oder Abfragedurchsatz neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo in gemeinsame Umgebungen übergeht.

Verstärkungsmaßnahme Detail 6/881: Messen Sie die Ausführungsdauer, die Fehlerklasse sowie den Tokenverbrauch für diese Notiz und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragebogens statt aufgrund von Einzelfällen, ob die Änderung beibehalten werden soll.

Für die siebte Stufe der Verstärkungsmaßnahme sollten vor dem Ändern des Codes die Eingabedaten, der Verantwortliche für den Schritt sowie die Abschlusskriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallfall gemeinsam. Wiederholungsversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

Verstärkungsmaßnahme Detail 7/881: Messen Sie die Ausführungsdauer, die Fehlerklasse sowie den Tokenverbrauch für diese Notiz und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragebogens statt aufgrund von Einzelfällen, ob die Änderung beibehalten werden soll.

Beim Bearbeiten der Stufe 8 der Sicherheitsstärkung sollten Sie zunächst einen Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikatoren sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Stufe als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab.

Sicherheitsdetail 8/881: Messen Sie die Ausführungsdauer, die Fehlerklasse sowie den Tokenverbrauch für diese Anweisung und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragekatalogs – und nicht nur aufgrund von Einzelfällen –, ob die Änderung beibehalten werden soll.

Die Stufe 9 der Sicherheitsstärkung funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Bewahren Sie Konfigurationen außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt werden, den Betreiber ohne das Durchlesen des gesamten Systems prüfen können.

Verstärkungsmaßnahme Detail 9/881: Messen Sie die Ausführungsdauer, die Fehlerklasse sowie den Tokenverbrauch für diese Notiz und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragebogens statt aufgrund von Einzelfällen, ob die Änderung beibehalten werden soll.

Zur 10. Stufe der Verstärkungsmaßnahmen sollten die Eingabedaten, der Verantwortliche für den Schritt sowie die Abschlusskriterien vor der Codeänderung definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Bevorzugen Sie kleine, testbare Einheiten vor umfangreichen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzelne Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren.

Verstärkungsmaßnahme Detail 10/881: Messen Sie die Ausführungsdauer, die Fehlerklasse sowie den Tokenverbrauch für diese Notiz und entscheiden Sie anschließend auf der Grundlage eines festgelegten Fragebogens statt aufgrund von Einzelfällen, ob die Änderung beibehalten werden soll.