📚 Documentation

Complete guide to using agentic-chaos for resilience testing.

Quick Navigation

Installation

PyPI (Recommended)

pip install agentic-chaos

From Source

git clone https://github.com/DeepAgentLabs/agentic-chaos.git
cd agentic-chaos
uv sync --frozen

With AgenticLens Integration

pip install agentic-chaos[agenticlens]

This enables optional two-way integration: merge chaos events into AgenticLens observability reports.

Quick Start

Step 1: Import

from agentic_chaos.chaos import chaos_call, chaos_session, TokenTimeoutError

Step 2: Wrap Your Calls

with chaos_session(["token_timeout"]):
    try:
        result = chaos_call(llm.complete, "What is AI?", faults=["token_timeout"])
    except TokenTimeoutError:
        print("LLM timed out—handle gracefully")

Step 3: Run from CLI

agentic-chaos chaos run my_app.py --inject token_timeout,rate_limit_storm --save report.json
💡 Tip: Outside of a chaos session, all calls are transparent pass-throughs. This means your instrumented code is safe to ship.

Python API Reference

chaos_call()

Inject chaos into a single function call.

result = chaos_call(
    fn,                    # The function to call
    *args,                 # Positional arguments
    faults=["fault_type"], # Which faults apply here
    step_name="step",      # For tracking/topology
    **kwargs               # Keyword arguments
)

chaos_session()

Create a chaos testing session. All chaos_call() inside use faults from this session.

with chaos_session([TokenTimeoutFault(hang_seconds=3.0), RateLimitStormFault()]):
    # chaos_call() here will use these faults
    result = chaos_call(llm_fn, query, faults=["token_timeout"])

Fault Classes

  • TokenTimeoutFault(hang_seconds=2.0, mode="raise")
  • RateLimitStormFault(burst_count=3, retry_after=1.0)
  • SilentDegradationFault(degrade_fn=None, seed=None)
  • ToolCallFailureFault(mode="error", timeout_seconds=5.0, tool_name=None)
  • MemoryCorruptionFault(mode="garble", seed=None)
  • InfiniteLoopFault(force_turns=5, continue_value="...")

CLI Reference

chaos run

Run a Python script under LLM-level chaos injection.

agentic-chaos chaos run my_app.py --inject FAULT1,FAULT2 --save report.json

agent run

Run an agent under agent-level chaos injection.

agentic-chaos agent run my_agent.py --inject tool_failure,memory_corruption --save report.json

chaos list-faults

List all available fault types.

agentic-chaos chaos list-faults

Fault Types

LLM-Level Faults (v0.1)

TokenTimeoutFault – Hung/slow completion
  • Modes: raise (error), delay (succeed late)
  • Use: Test timeout handling, slow response degradation
RateLimitStormFault – 429 burst then recovery
  • Stateful: First N calls fail, rest pass
  • Use: Test retry logic and backoff behavior
SilentDegradationFault – Same latency, garbage output
  • Outcome: Monitoring-blind failure
  • Use: Test app's own validation, catch output without monitoring

Agent-Level Faults (v0.2)

ToolCallFailureFault – Tool error/timeout/empty
  • Modes: error, timeout, empty
  • Use: Test tool unavailability, cascade failures
MemoryCorruptionFault – Truncate/inject/garble state
  • Modes: truncate (50%), inject (garbage), garble (letters)
  • Use: Test multi-agent memory resilience
InfiniteLoopFault – Force extra turns
  • Stateful: First N calls forced to continue
  • Use: Test turn-limit safeguards

Fidelity & Handoff Faults (v0.3)

HandoffCorruptionFault – Corrupt/drop/delay an edge, not a node
  • Modes: corrupt (garble in transit), drop (never arrives), delay (late arrival)
  • Use: Test resilience of the payload handed from one agent to another
MemoryCorruptionFault(mode="decay") – Progressive corruption across turns
  • Param: rate controls how quickly shared state degrades
  • Use: Model long-running-session state decay instead of a single corruption event
Fidelity Judges – fidelity_session() with HeuristicJudge, DeepEvalJudge, or PydanticEvalsJudge
  • Attaches a continuous fidelity_score (0.0–1.0) to a ChaosEvent
  • Use: Determine whether corrupted output is actually worse, not just different

Examples

Example 1: Test Recovery

with chaos_session([RateLimitStormFault(burst_count=2)]):
    for i in range(5):
        try:
            result = chaos_call(llm_fn, query, faults=["rate_limit_storm"])
            print(f"Call {i}: Success")
        except RateLimitStormError:
            print(f"Call {i}: Rate limited")

Example 2: Agent Cascade

from agentic_chaos.agents import wrap_tool, TopologyTracker

tracker = TopologyTracker()
search = wrap_tool(search_fn, "search", tracker=tracker, caller_node="Agent")

with chaos_session([ToolCallFailureFault(tool_name="search")]):
    agent_workflow()

# See cascade in tracker.topology.as_json()

Example 3: AgenticLens Integration

from agenticlens import profile, step
from agentic_chaos.integrations.agenticlens import attach_events, step_kwargs

with chaos_session(["token_timeout"]) as session:
    with profile("Agent") as workflow:
        with step("Search"):
            chunks = chaos_call(search, q, **step_kwargs(step), faults=["token_timeout"])

attach_events(session, workflow)
# agenticlens analyze workflow.json → shows cost + chaos impact
📖 More Examples: See examples/ folder in the repo.

Resources