Skip to content

Ctx Composer

ctx_composer — mvp.ctx_composer

Cluster: Uncategorised | Type: component | MCP Tools: None

Overview

The context-composition brain. Given a context need (goal + dataset shape + constraints), it composes the right STACK of context-management components — leveraging the ctx_* family's distinct strengths for large-dataset search, retrieval, and curation. A deterministic floor always produces a safe, offline, cost-bounded plan with honest fallbacks; an optional trusted-LLM advisor may refine it in-set (off by default, zero paid spend).

It composes a PLAN; it does not run the pipeline, and it never emits a verified label on its own.

When to use:

  • Deciding dense (ctx_colbert) vs sparse (ctx_rag/ctx_elastic) vs graph (ctx_cognee) retrieval for a dataset's size and shape
  • Planning a large-dataset search -> retrieve -> rerank -> compress -> curate pipeline
  • Choosing compression (ctx_recursive/ctx_claude_context/ctx_fenic), assembly (ctx_mnm), or memory (ctx_ace/ctx_claude_mem) components
  • Getting an offline-safe, cost-bounded recommendation with honest degradation

Example:

from mvp.ctx_composer import compose, ContextNeed

stack = compose(ContextNeed(goal="large-dataset-search", corpus_size="large"))
# stack.phases -> retrieve(ctx_colbert) -> rerank -> compress(ctx_recursive) -> assemble(ctx_mnm)

Works well with: every ctx_* component (it composes them), grounding, and context_engine.

Public API

CtxComposerDecisionError(ValueError)

The advisor did not produce a usable, in-set, constraint-honest decision.

AdviceDecision

An advisory refinement of the floor: a reorder + in-set primary overrides.

Field Type Default
ordered_phases tuple[str, ...] required
overrides dict[str, str] field(default_factory=dict)
rationale str ''
confidence float 0.0

LLMComposerAdvisorRuntime

Provider-neutral advice runtime backed by G6's LLM caller interface.

Constructor:

Parameter Type Default
llm LLMCaller \| None None

Methods:

advise(need: ContextNeed, floor_stack: ComposedStack) -> AdviceDecision

CtxComposerAdvisor

Advisory facade: refine the floor, degrading HONESTLY to it on any failure.

Constructor:

Parameter Type Default
runtime ComposerAdviceRuntime \| None None

Methods:

advise(need: ContextNeed, floor_stack: ComposedStack) -> ComposedStack

CatalogEntry(BaseModel)

One context component's curated capability profile.

Field Type Default
name str required
phases tuple[Phase, ...] required
modality Modality required
corpus_fit tuple[CorpusSize, ...] required
handles_structured bool False
content_kinds tuple[ContentKind, ...] ('text',)
offline_capable bool True
external_dependency str ''
cost_class CostClass 'free'
good_at tuple[str, ...] ()
bad_at tuple[str, ...] ()
mitigates_failure_modes tuple[str, ...] ()
degradation str ''

ComposerInput(BaseModel)

The invoke/MCP surface for the composer (carries the op + ContextNeed).

Field Type Default
op Literal['compose', 'catalog', 'explain', 'info'] 'compose'
task str ''
goal Goal 'retrieval'
corpus_size CorpusSize 'medium'
structured bool False
content_kinds tuple[ContentKind, ...] ('text',)
offline bool False
max_latency_ms int 0
max_cost_usd float 0.0
safety_tier SafetyTier 'R1'
component str ''
phase str ''
modality str ''
offline_only bool False

CtxComposerBlock(AIBlock[ComposerInput, dict, None])

The context-composition brain (deterministic floor + optional advisor).

Field Type Default
name str 'ctx_composer'
agentic_advisor Any None
advisor_default_enabled bool False

Methods:

infer(data: ComposerInput) -> Result[dict]

ContextNeed(BaseModel)

A context problem to solve: goal + dataset shape + constraints.

Field Type Default
task str ''
goal Goal 'retrieval'
corpus_size CorpusSize 'medium'
structured bool False
content_kinds tuple[ContentKind, ...] ('text',)
offline bool False
max_latency_ms int 0
max_cost_usd float 0.0
safety_tier SafetyTier 'R1'

PhaseChoice(BaseModel)

One phase of the composed pipeline with its chosen component.

Field Type Default
phase Phase required
primary str required
fallback_chain tuple[str, ...] ()
modality Modality 'sparse'
rationale str ''
degraded bool False
degrade_reason str ''

ComposedStack(BaseModel)

A recommended context-management pipeline (a PLAN, not a result).

Field Type Default
need ContextNeed required
phases tuple[PhaseChoice, ...] ()
notes tuple[str, ...] Field(default_factory=tuple)
advisory_used bool False
advisory_degraded bool False
label_hint LabelHint 'qualified-draft'

Functions

summarize_ctx_composer_agentic_evidence(stack: ComposedStack, advisor: Any = None) -> dict[str, Any]

Summarise the advisory provenance of a composed stack.

agentic_advisor_enabled(default_enabled: bool) -> bool

Whether the LLM advisor should run when none is explicitly injected.

validate_advice_decision(decision: AdviceDecision, floor_stack: ComposedStack, need: ContextNeed) -> None

In-set / constraint guard. Raises :class:CtxComposerDecisionError unless

advisor_is_llm_trusted(advisor: Any) -> bool

Whether the block may report a real LLM refinement for advisor.

get_entry(name: str) -> CatalogEntry | None

is_offline_safe(name: str) -> bool

True iff name runs with no external service/network/paid dependency.

is_cost_safe(name: str, max_cost_usd: float) -> bool

True iff name is allowed under a positive cost ceiling.

live_crosscheck() -> dict[str, object]

Cross-check the curated catalog against the live component registry.

compose(need: ContextNeed) -> ComposedStack

Compose a safe, deterministic context-management pipeline for need.