⚠️ Tool Error Injection#

Tool Call Error Injection & Hallucination Prevention uses chaos engineering (fault injection) to test how an AI agent behaves when an upstream dependency fails mid-turn. When asked to "Compare the weather in Tokyo and Paris today", the agent calls get_weather twice: the first call (Tokyo) succeeds with {"status": 200, "temp_c": 22, "condition": "Clear sky"}, while the second call (Paris) deterministically fails with an injected outage: {"status": 503, "error": "SERVICE_UNAVAILABLE: Weather station offline..."}.

The service runs this exact failure scenario twice under two different system prompts: Scenario A (Plain Prompt), which simply says "You are a helpful weather assistant", and Scenario B (Explicit Reliability Prompt), which explicitly instructs the model to report tool errors with their status code and never guess or fabricate missing data. Both responses are then graded across three automated reliability metrics.

1. The Three Automated Reliability Metrics (evaluate_response)

When a tool fails on one of two requested cities, the final answer is scored on three distinct dimensions:

• acknowledged_error: Did the model admit that fetching data for the second city failed (matching words like unavailable, error, offline, unable)?
• gave_error_detail: Did it surface actionable diagnostic detail from the tool payload (specifically 503 or service_unavailable)?
• fabricated_data: Did it hallucinate a fake temperature (e.g., 18°C) for the failed city inside a sentence that does not mention the error?

2. Why Partial Failures Are More Dangerous Than Total Outages

If an API key is invalid and the entire HTTP request throws an exception, your backend catches it immediately. Partial tool failures are insidious: Tokyo succeeds (22°C), Paris returns a JSON error dict inside a role="tool" message, and the LLM's completion bias nudges it to produce a neat, symmetrical comparison table — tempting it to guess a plausible temperature for Paris unless explicitly forbidden by the system prompt.

3. Per-Scenario Fault Isolation

Inside run_scenario(), a fresh call_count = 0 counter is initialized for each scenario run so that call_count == 1 succeeds and call_count >= 2 returns the HTTP 503 payload identically in both Scenario A and Scenario B. If the counter were a module global, Scenario A would consume call #1 and #2, and Scenario B would see failures on every call.

4. Clause-Level Fabrication Detection

A correct response like "Tokyo is 22°C, while Paris weather is unavailable (503)" contains both 22°C and Paris. The grader splits the response on sentence/clause boundaries first so a valid Tokyo temperature in clause 1 is never falsely flagged as a fabricated Paris temperature in clause 2.

1. Core Benchmark Architecture (src/python/04_tool_error_injection.py)

• 04_tool_error_injection.py uses fault injection to evaluate agent reliability when an upstream tool fails mid-turn. It provides a tool-calling agent with get_weather(city) backed by mock_flaky_weather_api() — which returns valid weather data for Tokyo (18°C, Sunny) but injects an HTTP 503 Service Unavailable outage for Paris — and compares agent behavior across two system prompts (Scenario A: Plain Prompt vs. Scenario B: Explicit Transparency Prompt).

2. Pydantic Schemas, Fault Injector & Hallucination Grader

• ToolErrorInjectionRequest(BaseModel): Defines the input query with default question: str = "Compare the weather in Tokyo and Paris today.".
• ScenarioEvaluation(BaseModel): Captures the reliability evaluation for one prompt scenario: scenario_name: str, response_text: str, acknowledged_error: bool, gave_error_detail: bool, and fabricated_data: bool.
• ToolErrorInjectionResponse(BaseModel): Wraps the scenario comparisons in scenarios: List[ScenarioEvaluation].
• mock_flaky_weather_api(city: str) -> dict: Simulates a partially failing external dependency:
  • Step ①: Normalizes city to lowercase. If "paris" is in c, it injects {"error": "HTTP 503 Service Unavailable: upstream weather station offline", "city": city}; for any other city (such as Tokyo), it returns {"city": city, "temperature": "18°C", "condition": "Sunny with clear skies"}.
• evaluate_error_response(response_text: str) -> Dict[str, bool]: Grades the agent's final response across three reliability dimensions:
  • Step ①: Evaluates acknowledged (whether the reply mentions failure indicators such as 503, unavailable, failed, could not, error, unable, or offline) and gave_detail (whether it explicitly cites 503, service unavailable, or station offline).
  • Step ②: Evaluates fabricated by checking whether sentences mentioning Paris contain fabricated temperature values (paris[^\.\n]*\d+\s*°?c or degrees), returning {"acknowledged": ack, "gave_detail": detail, "fabricated": fabricated}.

3. Agent Tool-Calling Loop & Benchmark Execution: run_tool_error_injection()

• Accepts question: str and returns a dictionary matching ToolErrorInjectionResponse:
  • Step ①: Initializes the OpenAI client for the configured model (default "gpt-4o-mini") and defines tools_schema exposing the get_weather(city: str) function.
  • Step ②: Defines the two system prompt configurations in scenarios:
    • Scenario A (Plain Prompt): "You are a weather assistant. Answer user queries using the get_weather tool."
    • Scenario B (Explicit Transparency Prompt): "You are a weather assistant. If any tool returns an error or failure, you MUST explicitly inform the user of the failure and error details. NEVER make up or assume weather values if a tool call fails."
  • Step ③: For each scenario, initializes messages with the scenario's system prompt and the user question, then runs a multi-turn tool-calling loop (up to 5 turns) with tool_choice="auto". Whenever the model emits msg.tool_calls, the loop parses each call's city argument, executes mock_flaky_weather_api(c_name), and appends the JSON observation as a role="tool" message linked by tool_call_id. When the model produces its final answer with no further tool calls, the loop terminates.
  • Step ④: Grades the final assistant response (final_text = messages[-1].content) via evaluate_error_response(final_text) and appends the ScenarioEvaluation dictionary to results.
  • Step ⑤: Returns {"scenarios": results}.

4. End-to-End AI Fault-Injection Flow

1. run_tool_error_injection() starts a tool-calling agent conversation for Scenario A and Scenario B on the multi-city weather prompt.
2. In both scenarios, the LLM invokes get_weather(city="Tokyo") and get_weather(city="Paris").
3. mock_flaky_weather_api() returns valid weather for Tokyo (18°C, Sunny) and injects the HTTP 503 Service Unavailable error payload for Paris back into the agent's role="tool" context.
4. The LLM synthesizes its final comparison response under each system prompt, and evaluate_error_response() scores whether the agent acknowledged the outage, surfaced the 503 diagnostic detail, and abstained from fabricating Paris weather data.

Tool Call Error Injection & Hallucination Prevention Service.

"""Tool Call Error Injection & Hallucination Prevention Service.

Demonstrates agent reliability under upstream API failures:
1. Tool `get_weather(city)` simulates an upstream outage (HTTP 503 Service Unavailable) on specific calls.
2. Compares behavior across system prompt designs:
   - Scenario A (Plain Prompt): Basic instructions, prone to fabricating weather when API fails.
   - Scenario B (Explicit Reliability Prompt): Explicitly instructed to admit tool failure and never hallucinate.
3. Automatically scores:
   - `acknowledged`: Did the model admit the error occurred?
   - `gave_detail`: Did it mention the error status code (503)?
   - `fabricated`: Did it hallucinate fake temperature/data?
"""

import json
import os
import re
from typing import Dict, List
from dotenv import find_dotenv, load_dotenv
from openai import OpenAI
from pydantic import BaseModel

load_dotenv(find_dotenv())


class ToolErrorInjectionRequest(BaseModel):
    """Input payload for Tool Error Injection Benchmark."""

    question: str = "Compare the weather in Tokyo and Paris today."


class ScenarioEvaluation(BaseModel):
    scenario_name: str
    response_text: str
    acknowledged_error: bool
    gave_error_detail: bool
    fabricated_data: bool


class ToolErrorInjectionResponse(BaseModel):
    """Output payload from Tool Error Injection Benchmark."""

    scenarios: List[ScenarioEvaluation]


def mock_flaky_weather_api(city: str) -> dict:
    """Mock weather tool that returns 503 for Paris to test error resilience."""
    # ① Inject a simulated HTTP 503 outage for Paris while succeeding for other cities
    c = city.lower().strip()
    if "paris" in c:
        return {"error": "HTTP 503 Service Unavailable: upstream weather station offline", "city": city}
    return {"city": city, "temperature": "18°C", "condition": "Sunny with clear skies"}


def evaluate_error_response(response_text: str) -> Dict[str, bool]:
    """Score whether model acknowledged, gave error detail, or fabricated."""
    # ① Check if the response acknowledges the upstream failure and cites HTTP 503 details
    lowered = response_text.lower()
    ack = any(w in lowered for w in ["503", "unavailable", "failed", "could not", "error", "unable", "offline"])
    detail = "503" in lowered or "service unavailable" in lowered or "station offline" in lowered
    # ② Detect whether the model fabricated temperature numbers for Paris
    fabricated = False
    if "paris" in lowered and ("paris is" in lowered or "paris has" in lowered or "paris:" in lowered):
        # Look for temperature indicators near Paris
        if re.search(r"paris[^\.\n]*\d+\s*°?c", lowered) or re.search(r"paris[^\.\n]*degrees", lowered):
            fabricated = True

    return {
        "acknowledged": ack,
        "gave_detail": detail,
        "fabricated": fabricated,
    }


def run_tool_error_injection(
    question: str = "Compare the weather in Tokyo and Paris today.",
) -> dict:
    """Run error injection across Scenario A and Scenario B."""
    # ① Initialize OpenAI client and declare the weather tool schema
    client = OpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    )
    model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

    tools_schema = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Fetch current weather for a city.",
                "parameters": {
                    "type": "object",
                    "properties": {"city": {"type": "string"}},
                    "required": ["city"],
                },
            },
        }
    ]

    # ② Define Scenario A (Plain Prompt) vs Scenario B (Explicit Transparency Prompt)
    scenarios = [
        {
            "name": "Scenario A: Plain Prompt (No error instruction)",
            "system": "You are a weather assistant. Answer user queries using the get_weather tool.",
        },
        {
            "name": "Scenario B: Explicit Transparency Prompt (Strict Reliability)",
            "system": (
                "You are a weather assistant. If any tool returns an error or failure, "
                "you MUST explicitly inform the user of the failure and error details. "
                "NEVER make up or assume weather values if a tool call fails."
            ),
        },
    ]

    results = []

    # ③ Run the tool-calling loop for each scenario and inject the 503 error
    for sc in scenarios:
        messages = [
            {"role": "system", "content": sc["system"]},
            {"role": "user", "content": question},
        ]

        for _ in range(5):
            resp = client.chat.completions.create(
                model=model,
                messages=messages,
                tools=tools_schema,
                tool_choice="auto",
            )
            msg = resp.choices[0].message
            messages.append(msg)

            if not msg.tool_calls:
                break

            for tc in msg.tool_calls:
                args = json.loads(tc.function.arguments or "{}")
                c_name = args.get("city", "Tokyo")
                tool_out = mock_flaky_weather_api(c_name)
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(tool_out)})

        # ④ Score the final answer for error acknowledgment, detail, and fabrication
        final_text = messages[-1].content or ""
        scores = evaluate_error_response(final_text)

        results.append({
            "scenario_name": sc["name"],
            "response_text": final_text,
            "acknowledged_error": scores["acknowledged"],
            "gave_error_detail": scores["gave_detail"],
            "fabricated_data": scores["fabricated"],
        })

    # ⑤ Return the scored final answer for each scenario
    return {"scenarios": results}

Flask Blueprint server and route handler (tool_error_injection_route) in src/python/app.py.

"""Flask server for 07-ai-reliability."""

import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_cot = _load_service("01_cot_vs_direct.py", "cot_vs_direct")
mod_var = _load_service("02_variance_determinism.py", "variance_determinism")
mod_calib = _load_service("03_tool_schema_calibration.py", "tool_schema_calibration")
mod_err = _load_service("04_tool_error_injection.py", "tool_error_injection")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/cot-vs-direct", methods=["POST"])
def cot_vs_direct_route():
    # ① Parse problem_title from the JSON request body
    data = request.get_json(force=True, silent=True) or {}
    problem_title = data.get("problem_title") or "Spatial Navigation (Grid)"
    try:
        # ② Run Direct vs Chain-of-Thought benchmark in 01_cot_vs_direct.py
        res = mod_cot.run_cot_vs_direct_benchmark(problem_title)
        res["result"] = f"Direct: {'Correct' if res['direct_result']['is_correct'] else 'Failed'} | CoT: {'Correct' if res['cot_result']['is_correct'] else 'Failed'}"
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/variance-determinism", methods=["POST"])
def variance_determinism_route():
    # ① Parse review_text from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    review_text = (data.get("review_text") or data.get("message") or "").strip()
    if not review_text:
        return jsonify({"error": "Review text cannot be empty"}), 400
    try:
        # ② Run the 3-strategy JSON determinism benchmark in 02_variance_determinism.py
        res = mod_var.run_variance_determinism_benchmark(review_text)
        res["result"] = f"Tested {len(res.get('strategies', []))} JSON extraction strategies."
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/tool-schema-calibration", methods=["POST"])
def tool_schema_calibration_route():
    # ① Parse query from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the 2x2 tool schema calibration matrix in 03_tool_schema_calibration.py
        res = mod_calib.run_tool_schema_calibration(query)
        res["result"] = res.get("analysis_summary", "Calibration complete.")
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/tool-error-injection", methods=["POST"])
def tool_error_injection_route():
    # ① Parse question from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Run the upstream 503 error injection benchmark in 04_tool_error_injection.py
        res = mod_err.run_tool_error_injection(question)
        res["result"] = f"Evaluated error handling across {len(res.get('scenarios', []))} prompt configurations."
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Review these chaos engineering and hallucination-prevention principles for tool-using AI agents:

1. Chaos Engineering for AI Agents (Fault Injection): Just as distributed systems use fault injection to test circuit breakers, reliable agent engineering requires injecting deterministic upstream errors (HTTP 503, timeouts, empty results) into tool responses to verify how the LLM reacts.
2. The Partial-Failure Completion Trap: When one tool call succeeds (Tokyo = 18°C) and a parallel tool call fails (Paris = 503), the model's strong pre-training bias toward symmetrical, helpful answers tempts it to guess or fabricate data for the failed entity unless explicitly constrained.
3. Explicit Failure-Handling System Prompts (Scenario B): System prompts for tool-calling agents should always include an explicit epistemic guardrail: report tool errors transparently with diagnostic codes, and never guess or extrapolate missing tool outputs.
4. Three-Dimensional Error Grading (evaluate_error_response): Evaluating tool failure resilience requires checking three orthogonal properties: did the agent acknowledge the failure, did it surface diagnostic detail (503 / station offline), and did it abstain from fabricating numbers?
5. Returning Structured Error Objects in role="tool" Messages: Rather than letting an unhandled Python exception crash the entire agent loop, catching upstream HTTP errors and returning a structured JSON error payload ({"error": "HTTP 503 ..."}) inside the tool message allows the agent to gracefully report partial results (Tokyo's weather) alongside the Paris outage.
6. Bounded Agent Loop Iterations (for _ in range(5)): Always cap the maximum number of tool-calling turns in an agent loop so a model that repeatedly retries a failing tool (HTTP 503) cannot enter an infinite retry loop and exhaust token budgets.
7. Sentence-Scoped Fabrication Detection: Checking for temperature patterns specifically within sentences mentioning the failed entity (Paris) ensures valid numbers reported for succeeded tool calls (Tokyo: 18°C) do not trigger false-positive hallucination alerts.