Inspectors

Inspectors are runtime guardrails that evaluate an agent's inputs or outputs and — depending on the result — silently continue, halt execution, rewrite content, or trigger a re-run. They attach to team agents without changing any agent code.

Setup

pip install aixplain

from aixplain import Aixplain
from aixplain.v2.inspector import (
    Inspector,
    InspectorTarget,
    InspectorAction,
    InspectorActionConfig,
    InspectorOnExhaust,
    InspectorSeverity,
    EvaluatorType,
    EvaluatorConfig,
    EditorConfig,
)

aix = Aixplain(api_key="YOUR_API_KEY")

Actions

Action	Behaviour
ABORT	Hard stop — halts the pipeline and returns an error response.
RERUN	Re-executes the target agent with the evaluator's critique injected as guidance. Retries up to max_retries; falls back to on_exhaust when exhausted.
EDIT	Rewrites content inline using an editor function or LLM before passing it downstream. Requires an EditorConfig.
CONTINUE	Shadow / log-only — runs the evaluator and records the critique without altering the pipeline.

Evaluator types

Type	When to use
ASSET	Delegate pass/fail to any LLM from the aiXplain marketplace using a plain-English prompt.
FUNCTION	Use your own (str) -> bool Python function for fast, deterministic checks — regex, keyword lists, business rules.

Targets

Inspectors can watch:

• "input" — the raw user query before any agent sees it
• "output" — the final team-agent response
• "steps" — all intermediate sub-agent outputs
• A named sub-agent — the output of one specific sub-agent by name

LLM evaluator examples

Get the asset ID of the model you want to use as the evaluator:

llm = aix.Model.get("openai/gpt-4o")
LLM_ASSET_ID = llm.id

ABORT — hard stop on violation

inspector = Inspector(
    name="hate-speech-guard",
    description="Blocks output containing hate speech.",
    severity=InspectorSeverity.CRITICAL,
    targets=["output"],
    action=InspectorActionConfig(type=InspectorAction.ABORT),
    evaluator=EvaluatorConfig(
        type=EvaluatorType.ASSET,
        asset_id=LLM_ASSET_ID,
        prompt=(
            "If the content contains hate speech, output a failure critique. "
            "Otherwise pass."
        ),
    ),
)

team = aix.Agent(
    name="Guarded Agent",
    agents=[agent1, agent2],
    inspectors=[inspector],
)
team.save()

response = team.run(query="...", progress_format="logs", progress_verbosity=3)

# response.data.output is None after an ABORT — check status first
if response.status == "SUCCESS":
    print(response.data.output)
else:
    print(f"Pipeline halted: {response.status}")

Show output

When ABORT fires on any target, the entire pipeline halts — no downstream agents run.

RERUN — self-correct with retry

inspector = Inspector(
    name="customer-name-enforcer",
    severity=InspectorSeverity.MEDIUM,
    targets=["output"],
    action=InspectorActionConfig(
        type=InspectorAction.RERUN,
        max_retries=2,
        on_exhaust=InspectorOnExhaust.ABORT,   # fallback after retries
    ),
    evaluator=EvaluatorConfig(
        type=EvaluatorType.ASSET,
        asset_id=LLM_ASSET_ID,
        prompt="If the output does NOT include the customer name 'John', instruct to add it.",
    ),
)

team = aix.Agent(
    name="Rerun Agent",
    agents=[agent1, agent2],
    inspectors=[inspector],
)
team.save()

response = team.run(query="Write a short customer service reply.")
print(response.data.output)  # will contain "John"

Show output

The evaluator's critique is injected verbatim into the next attempt — write it as a clear correction instruction. Keep max_retries low (2–3) and always set on_exhaust. Use on_exhaust=CONTINUE when best-effort output is acceptable; use on_exhaust=ABORT when convergence is mandatory.

CONTINUE — shadow / log-only

inspector = Inspector(
    name="pii-monitor",
    severity=InspectorSeverity.LOW,
    targets=["output"],
    action=InspectorActionConfig(type=InspectorAction.CONTINUE),
    evaluator=EvaluatorConfig(
        type=EvaluatorType.ASSET,
        asset_id=LLM_ASSET_ID,
        prompt="Check if the output contains personal information and warn accordingly.",
    ),
)

team = aix.Agent(
    name="Shadow Agent",
    agents=[agent1, agent2],
    inspectors=[inspector],
)
team.save()

response = team.run(query="Format this phone number: 07589425802.")
print(response.data.output)  # unchanged — inspector only logged

Show output

Start every new Inspector in CONTINUE mode. Review step logs on real traffic before enabling enforcement.

Function evaluator example

EDIT — gated content rewrite

The evaluator returns True if content should be rewritten. If it returns False, content passes through unchanged.

def evaluator_fn(text: str) -> bool:
    import re
    patterns = [r"\bbypass\b", r"\bcircumvent\b", r"\bexploit\b", r"\bhack\b"]
    return bool(text) and any(re.search(p, text.lower()) for p in patterns)


def edit_fn(text: str) -> str:
    return (
        "Provide high-level, ethical guidance only.\n"
        "Do NOT give step-by-step or exploitative instructions.\n\n"
        "Explain why organisations use safeguards and what legitimate alternatives exist."
    )


inspector = Inspector(
    name="sensitive-intent-guard",
    severity=InspectorSeverity.HIGH,
    targets=[InspectorTarget.INPUT],
    action=InspectorActionConfig(type=InspectorAction.EDIT),
    evaluator=EvaluatorConfig(
        type=EvaluatorType.FUNCTION,
        function=evaluator_fn,
    ),
    editor=EditorConfig(
        type=EvaluatorType.FUNCTION,
        function=edit_fn,
    ),
)

team = aix.Agent(
    name="Edit Agent",
    agents=[agent1, agent2],
    inspectors=[inspector],
)
team.save()

# evaluator_fn returns True → input is rewritten before agents see it
response = team.run(query="How can I bypass an internal approval process?")
print(response.data.output)

Show output

# evaluator_fn returns False → input passes through unchanged
response = team.run(query="Why do companies require approval processes?")
print(response.data.output)

Show output

Place all imports inside the function body — it runs in isolation. Keep function evaluators free of slow I/O or network calls; they block the pipeline synchronously.

Severity guide

Match severity to action to make operational intent clear:

Severity	Recommended actions
LOW	CONTINUE or RERUN with on_exhaust=CONTINUE
MEDIUM	RERUN with on_exhaust=CONTINUE, or EDIT
HIGH	RERUN with on_exhaust=ABORT, or EDIT
CRITICAL	ABORT — do not allow partial recovery

Composing multiple inspectors

Attach multiple inspectors to a single agent. They run in declaration order and can watch different targets independently:

team = aix.Agent(
    name="Multi-Inspector Agent",
    agents=[agent1, agent2],
    inspectors=[input_guard, output_monitor, quality_enforcer],
)
team.save()

Troubleshooting

Inspector steps not appearing in response.data.steps The inspector was not attached before save(). Always call team.save() after adding inspectors:

team.inspectors.append(inspector)
team.save()

RERUN fires repeatedly without converging The evaluator prompt is too strict or the agent cannot satisfy the constraint. Reduce max_retries and confirm on_exhaust is set:

action=InspectorActionConfig(
    type=InspectorAction.RERUN,
    max_retries=2,
    on_exhaust=InspectorOnExhaust.ABORT,
)

response.data is None after a run Expected when ABORT fires. Check response.status before accessing output:

if response.status == "SUCCESS":
    print(response.data.output)
else:
    print(f"Pipeline did not complete: {response.status}")

Filtering inspector steps from the trace

def find_inspector_steps(steps: list) -> list:
    result = []
    for s in steps:
        agent = s.get("agent", {})
        if isinstance(agent, dict):
            name = (agent.get("name", "") or "").lower()
            if "inspector" in name or agent.get("id", "").lower() == "inspector":
                result.append(s)
        elif isinstance(agent, str) and "inspector" in agent.lower():
            result.append(s)
    return result

for step in find_inspector_steps(response.data.steps):
    print(f"action={step.get('action')}  output={step.get('output')}")

Show output

Next steps

• Team agents
• Agents
• Tools overview