Metacognition API

metacognition

Metacognition primitives — confidence, uncertainty, capability boundary.

The cognitive substrate for agents that know what they don't know. See docs/concepts/metacognition.md for the full picture; this module hosts:

• :class:EnrichedOutput[OutputT] — the runtime type that pairs an output with the agent's self-assessment of it.
• :class:ConfidenceProtocol — pluggable strategy for producing the assessment (zero-cost :class:StructuredOutputProtocol, +1-call :class:LLMSelfRatingProtocol, +k-call :class:EnsembleProtocol).
• :class:MetacognitionHook — :class:ToolHook bridge to :class:EventBus.
• :class:MetacognitionConfig — frozen orchestration policy.
• :func:confidence_salience — salience scorer for :class:ContextManager integration.

MetacognitionConfig dataclass

MetacognitionConfig(
    redecompose_threshold: float = 0.5,
    max_redecompositions: int = 2,
    confidence_floor: float = 0.0,
)

Policy knobs for confidence-driven orchestration.

redecompose_threshold class-attribute instance-attribute

redecompose_threshold: float = 0.5

Below this confidence, :class:MetaOrchestrator re-decomposes the remaining subtasks. Must be in [0, 1].

max_redecompositions class-attribute instance-attribute

max_redecompositions: int = 2

Hard cap on re-decompositions per solve() call. Prevents runaway cost when a sub-task can't recover.

confidence_floor class-attribute instance-attribute

confidence_floor: float = 0.0

Below this floor, downstream consumers should treat the output as if capability_boundary=True.

EnrichedOutput

Agent output paired with the agent's own self-assessment.

MetacognitionHook

MetacognitionHook(
    bus: EventBus, *, agent_name: str = "unknown"
)

ToolHook that publishes enriched-output telemetry to an EventBus.

Implements only after_tool: when the tool result is an :class:EnrichedOutput, emits one metacognition.confidence AgentEvent. Other tool results are ignored (best-effort: safe to register on any HookRunner, even in agents that don't use enrichment).

The hook returns None to honor the existing observation-only contract — :class:HookRunner auto-wraps to :class:Continue.

Source code in orqest/metacognition/hook.py

def __init__(
    self,
    bus: EventBus,
    *,
    agent_name: str = "unknown",
) -> None:
    self._bus = bus
    self._agent_name = agent_name

ConfidenceProtocol

Strategy for extracting (or producing) an agent's self-assessment.

Called by :meth:BaseAgent.run_enriched after the underlying _run_implementation has produced the OutputT. The protocol returns the enriched payload. Failures are swallowed and surfaced as confidence=None — never raise.

EnsembleProtocol

EnsembleProtocol(
    k: int = 3,
    *,
    agreement_fn: Callable[[list[Any]], float]
    | None = None,
)

Sample the agent k times, compute confidence from agreement.

Confidence is the share of pairs of samples that match (under Pydantic model_dump). The output of the first sample is returned alongside the confidence score — confidence is a signal about the output, not a way to swap outputs.

Cost: +k–1 LLM calls. Samples run in parallel via :func:asyncio.gather.

Source code in orqest/metacognition/protocol.py

def __init__(
    self,
    k: int = 3,
    *,
    agreement_fn: Callable[[list[Any]], float] | None = None,
) -> None:
    if k < 2:
        raise ValueError("EnsembleProtocol requires k >= 2")
    self._k = k
    self._agreement_fn = agreement_fn or _default_agreement

LLMSelfRatingProtocol

LLMSelfRatingProtocol(
    *,
    rating_prompt: str | None = None,
    summariser: Callable[[Any], str] | None = None,
)

Ask the model to rate its own output post-turn.

Cost: +1 LLM call per agent turn. The protocol parses a JSON-shaped rating off the rater's reply; on parse failure or empty reply, falls back to confidence=None and stores the error in metadata.

Source code in orqest/metacognition/protocol.py

def __init__(
    self,
    *,
    rating_prompt: str | None = None,
    summariser: Callable[[Any], str] | None = None,
) -> None:
    self._rating_prompt = rating_prompt or _DEFAULT_RATING_PROMPT
    self._summariser = summariser or (lambda x: str(x)[:1000])

StructuredOutputProtocol

StructuredOutputProtocol(
    *,
    confidence_field: str = "self_confidence",
    uncertainty_field: str = "uncertain_about",
    boundary_field: str = "outside_my_capability",
)

Lift confidence fields off the agent's own output schema.

Zero-extra-cost protocol. Consumers extend their OutputT with fields the protocol knows how to read::

class MyOutput(BaseModel):
    answer: str
    self_confidence: float | None = None
    uncertain_about: list[str] = []
    outside_my_capability: bool = False

Output models that don't declare these fields produce confidence=None — best-effort.

Source code in orqest/metacognition/protocol.py

def __init__(
    self,
    *,
    confidence_field: str = "self_confidence",
    uncertainty_field: str = "uncertain_about",
    boundary_field: str = "outside_my_capability",
) -> None:
    self._cf = confidence_field
    self._uf = uncertainty_field
    self._bf = boundary_field

confidence_salience

confidence_salience(
    message: Any,
    *,
    floor: float = 0.3,
    metadata_key: str = "metacognition_confidence",
) -> float

Salience score driven by attached confidence metadata.

Returns 1.0 when no confidence is attached (backward-compat property: un-tagged messages are fully salient — drop on age only, never confidence). Otherwise returns the confidence value, clamped to [floor, 1.0] to keep low-confidence content from going to zero salience and being dropped first.

Source code in orqest/metacognition/salience.py

def confidence_salience(
    message: Any,
    *,
    floor: float = 0.3,
    metadata_key: str = "metacognition_confidence",
) -> float:
    """Salience score driven by attached confidence metadata.

    Returns ``1.0`` when no confidence is attached (backward-compat
    property: un-tagged messages are fully salient — drop on age only,
    never confidence). Otherwise returns the confidence value, clamped
    to ``[floor, 1.0]`` to keep low-confidence content from going to
    zero salience and being dropped *first*.
    """
    value = _read_metadata(message, metadata_key)
    if value is None:
        return 1.0
    try:
        f = float(value)
    except (TypeError, ValueError):
        return 1.0
    if f < floor:
        return floor
    if f > 1.0:
        return 1.0
    return f

recency_salience

recency_salience(
    message: Any,
    *,
    decay: float = 0.95,
    age_attr: str = "_age_turns",
) -> float

Exponential-decay salience by message age.

Reads an integer _age_turns attribute (or metadata key) — older messages decay. Default decay 0.95 → 5 turns ≈ 0.77, 20 turns ≈ 0.36.

Source code in orqest/metacognition/salience.py

def recency_salience(
    message: Any,
    *,
    decay: float = 0.95,
    age_attr: str = "_age_turns",
) -> float:
    """Exponential-decay salience by message age.

    Reads an integer ``_age_turns`` attribute (or metadata key) — older
    messages decay. Default decay 0.95 → 5 turns ≈ 0.77, 20 turns ≈ 0.36.
    """
    age = getattr(message, age_attr, None)
    if age is None:
        age = _read_metadata(message, age_attr)
    if age is None:
        return 1.0
    try:
        n = int(age)
    except (TypeError, ValueError):
        return 1.0
    return max(0.0, min(1.0, decay**n))