Critic System

Add runtime quality control to your agent loop with the Critic system — inspect, score, retry, or halt decisions before they commit.

Every step the Engine takes flows through a pipeline: decide, act, reduce, then critic. The critic sits between reduce and check_stop, giving you a structured hook to inspect the agent’s latest decision and its results, assign a quality score, and — when necessary — force a retry or halt execution.

This tutorial covers the full critic API: the CriticResult contract, built-in critics, the @critic decorator, instruction and state patches, and composing multiple critics together.

Step 1: The CriticResult Contract

Every critic returns a CriticResult — a dataclass that tells the Engine what to do next.

from qitos.engine.critic_result import CriticResult

The fields:

Field	Type	Default	Description
action	str	"continue"	One of "continue", "stop", "retry".
reason	str	""	Human-readable explanation.
score	float	1.0	Quality score between 0.0 and 1.0. Higher is better.
details	dict	{}	Extra structured data for logging or telemetry.
instruction_patch	str | None	None	Appended to the agent’s system prompt on the next iteration (only when action="retry").
state_patch	dict | None	None	Key-value pairs merged into the agent’s state before the next iteration (only when action="retry").
modified_prompt	str | None	None	Full replacement system prompt for the next iteration (only when action="retry").

The three actions control the loop:

• "continue" — proceed to the next step as normal.
• "stop" — halt the Engine immediately. The state’s stop reason is set to StopReason.CRITIC_STOP.
• "retry" — discard the current step and re-run decide. Any instruction_patch or state_patch is applied before the retry.

A minimal continue result:

result = CriticResult(action="continue", reason="all good", score=1.0)

A retry with an instruction patch:

result = CriticResult(
    action="retry",
    reason="output was incomplete",
    score=0.3,
    instruction_patch="Make sure to include all required fields in your answer.",
)

A hard stop:

result = CriticResult(
    action="stop",
    reason="safety policy violation",
    score=0.0,
)

Step 2: Using Built-in Critics

QitOS ships with ready-made critics in qitos.kit.critic.

PassThroughCritic

The simplest critic — always continues with a perfect score.

from qitos.kit.critic import PassThroughCritic

critic = PassThroughCritic()
# evaluate() always returns {"action": "continue", "reason": "pass", "score": 1.0}

This is the default when you create an Engine without specifying critics. It is useful as a no-op placeholder or as part of a composed chain where later critics do the real work.

SelfReflectionCritic

Inspects tool results for errors and retries automatically up to a configurable limit.

from qitos.kit.critic import SelfReflectionCritic

critic = SelfReflectionCritic(max_retries=2)

Behavior:

• If any tool result contains an {"error": ...} key and retries have not been exhausted, it returns action="retry" with score=0.2.
• If errors persist beyond max_retries, it returns action="stop" with score=0.0.
• If no errors are found, it returns action="continue" with score=1.0.

Pass it to the Engine:

from qitos.engine import Engine

engine = Engine(
    agent=my_agent,
    critics=[SelfReflectionCritic(max_retries=3)],
    # ... other engine config
)

ReActSelfReflectionCritic

A richer variant designed for ReAct agents. On error, it builds a structured reflection note describing the failed action and the observed error, then appends it to the state’s metadata so the LLM can learn from the failure on the next retry.

from qitos.kit.critic import ReActSelfReflectionCritic

critic = ReActSelfReflectionCritic(max_retries=2)

Functional Equivalents

Each built-in critic also ships as a decorated function:

from qitos.kit.critic import pass_through_critic, self_reflection_critic

# These are @critic-decorated functions -- see Step 3 for details

Step 3: Writing a Custom Critic with the @critic Decorator

The @critic decorator converts any plain function into a Critic instance. The function receives (state, decision, results) and can return a quick shorthand or a full CriticResult.

from qitos.engine.critic_decorator import critic

Quick-return shorthands:

Return value	Meaning
"continue"	Proceed normally
("stop", "reason")	Halt with a reason
("retry", "reason")	Retry with the same prompt
("retry", "reason", instruction_patch)	Retry with an appended instruction
CriticResult(...)	Full structured result

A bare decorator (no arguments):

@critic
def no_empty_answers(state, decision, results):
    """Stop if the final answer is empty."""
    if decision.mode == "final" and not decision.final_answer:
        return "stop", "empty final answer"
    return "continue"

With keyword arguments — set a custom name and default score:

@critic(name="safety", score=0.8)
def safety_check(state, decision, results):
    """Retry if any result looks unsafe."""
    for r in results:
        if isinstance(r, dict) and r.get("flagged"):
            return "retry", "unsafe output detected"
    return "continue"

When the function does not set a score explicitly, the decorator’s score parameter is used as the default. In the example above, a "continue" return gets score=0.8 instead of the usual 1.0.

Wire the critic into the Engine:

engine = Engine(
    agent=my_agent,
    critics=[no_empty_answers, safety_check],
    # ... other engine config
)

Step 4: Critic with Instruction and State Patches

When a critic returns action="retry", it can guide the next iteration in two ways:

• instruction_patch — a string appended to the agent’s system prompt, giving the LLM additional guidance.
• state_patch — a dictionary whose keys and values are merged into the agent’s state object via setattr.

These are applied before the Engine calls decide again, so the LLM and the state are already updated when the retry begins.

from qitos.engine.critic_result import CriticResult
from qitos.engine.critic_decorator import critic

@critic(name="format_enforcer")
def enforce_json_output(state, decision, results):
    """Retry with guidance if the output is not valid JSON."""
    if decision.mode == "final":
        answer = decision.final_answer or ""
        if not answer.strip().startswith("{"):
            return CriticResult(
                action="retry",
                reason="output is not JSON",
                score=0.2,
                instruction_patch=(
                    "You MUST output valid JSON. "
                    "Do not include any text outside the JSON object."
                ),
            )
    return "continue"

Using state_patch to inject tracking data:

@critic(name="retry_tracker")
def track_retries(state, decision, results):
    """Count retries and halt after the limit."""
    metadata = getattr(state, "metadata", {}) or {}
    retries = int(metadata.get("custom_retries", 0))

    has_error = any(isinstance(r, dict) and r.get("error") for r in results)
    if has_error and retries < 5:
        metadata["custom_retries"] = retries + 1
        state.metadata = metadata
        return CriticResult(
            action="retry",
            reason="tool error, retrying",
            score=0.3,
            state_patch={"last_error_type": "tool_failure"},
        )
    if has_error:
        return "stop", "exceeded retry limit"

    return "continue"

You can also use the tuple shorthand with an instruction patch:

@critic
def be_concise(state, decision, results):
    if decision.mode == "final" and len(decision.final_answer or "") > 2000:
        return "retry", "answer too long", "Keep your final answer under 2000 characters."
    return "continue"

Step 5: Combining Critics

The Engine accepts a list of critics. They are evaluated in order, and the first non-continue result wins. This lets you stack critics from strictest to most permissive.

from qitos.kit.critic import SelfReflectionCritic
from qitos.engine.critic_decorator import critic

@critic(name="safety_gate")
def safety_gate(state, decision, results):
    """Hard stop on safety violations -- nothing overrides this."""
    for r in results:
        if isinstance(r, dict) and r.get("safety_violation"):
            return "stop", "safety violation detected"
    return "continue"

@critic(name="format_check")
def format_check(state, decision, results):
    """Retry on formatting issues."""
    if decision.mode == "final":
        answer = decision.final_answer or ""
        if not answer.strip().startswith("{"):
            return "retry", "not JSON", "Output valid JSON only."
    return "continue"

Wire them into the Engine — safety first, then format, then the built-in self-reflection as a fallback:

from qitos.engine import Engine

engine = Engine(
    agent=my_agent,
    critics=[
        safety_gate,        # hardest stop -- checked first
        format_check,       # softer -- retry with guidance
        SelfReflectionCritic(max_retries=2),  # catch tool errors
    ],
    # ... other engine config
)

With this ordering:

1. If safety_gate returns "stop", the Engine halts immediately. Later critics are never called.
2. If safety_gate returns "continue" but format_check returns "retry", the Engine retries with the instruction patch.
3. If both return "continue", the SelfReflectionCritic gets a chance to catch tool errors.

This layered approach keeps each critic small and focused, while the ordering gives you full control over priority.

Hooks Lifecycle

Understand the full Engine event lifecycle that critics participate in.

Agent Module

Deep dive into the AgentModule interface that critics inspect.

Critics and Stop Criteria

How critics interact with stop criteria and retry budgets.