Ctx Recursive¶

ctx_recursive — mvp.ctx_recursive

Cluster: Context & Retrieval | Type: component | MCP Tools: 28

Overview¶

Hierarchical extractive summariser that compresses large texts to a target length without an LLM. At each depth level it splits the text into sentence-boundary chunks, scores sentences by TF-IDF word frequency, and selects the highest-scoring sentences up to a per-chunk budget. The process repeats recursively until the output fits in target_length or max_depth is exhausted. A simpler truncate strategy is also available.

When to use:

• Summarising long documents to fit within an LLM context window with no API cost
• Reducing agent working memory size before storing in a token-limited context block
• Building a quick extractive summary layer in a RAG or grounding pipeline

Example:

from mvp.ctx_recursive import CtxRecursiveBlock, RecursiveInput

block = CtxRecursiveBlock(name="recursive")
result = block.infer(RecursiveInput(
    text="<long article text>",
    target_length=500,
    max_depth=3,
    strategy="extractive",
))
# result.value.summary → compressed text ≤ 500 chars; result.value.depth → levels used

Works well with: ctx_ace, ctx_claude_context, ctx_markitdown

Public API¶

SummaryRerankDecision¶

Validated advisory rerank verdict over a returned summary set.

LLMRecursiveRerankRuntime¶

Provider-neutral summary-rerank runtime backed by G6's LLM caller.

Constructor:

Methods:

rerank(query: str, summaries: list[Any]) -> SummaryRerankDecision¶

CtxRecursiveRerankPatternRuntime¶

Stateless, load-bearing returned-set-ceiling enforcement.

Methods:

enforce_eligibility(decision: SummaryRerankDecision, summaries: list[Any]) -> tuple[list[Any], bool, bool]¶

CtxRecursivePlanner¶

Runtime-first advisory summary-rerank facade with returned-order fallback.

Constructor:

Methods:

rerank(query: str, summaries: list[Any]) -> list[Any]¶

CtxRecursiveBlock(AIBlock[RecursiveInput, RecursiveOutput, None])¶

Hierarchical extractive summariser.

Methods:

infer(data: RecursiveInput) -> Result[RecursiveOutput]¶

RecursiveInput(BaseModel)¶

Input to CtxRecursiveBlock.

RecursiveOutput(BaseModel)¶

Output from CtxRecursiveBlock.

CtxRecursiveSkillCatalog¶

Maps each applied pattern slug to a ctx_recursive rerank skill record.

Methods:

list_skills() -> list[CtxRecursiveSkill]¶

executable_skills() -> list[CtxRecursiveSkill]¶

get(slug: str) -> CtxRecursiveSkill | None¶

CtxRecursiveMCPBlock(AIBlock[MCPRecursiveInput, MCPRecursiveOutput, dict])¶

Full-featured recursive summarisation block with SQLite persistence.

Methods:

infer(data: MCPRecursiveInput) -> Result[MCPRecursiveOutput]¶

MCPRecursiveRecord(BaseModel)¶

MCPRecursiveInput(BaseModel)¶

MCPRecursiveOutput(BaseModel)¶

RecursiveStore¶

Sync SQLite recursive store with 6 tables.

Constructor:

Methods:

save_session(name: str, description: str, input_text: str, summary_text: str, strategy: str, target_length: int, chunk_size: int, max_depth: int, depth_used: int, chunks_processed: int, compression_ratio: float, tags: list[str]) -> str¶

load_session(name: str) -> dict[str, Any] | None¶

list_sessions(limit: int = 50) -> list[dict[str, Any]]¶

delete_session(name: str) -> bool¶

store_summary(key: str, original_text: str, summary_text: str, strategy: str, depth: int, chunks_processed: int, compression_ratio: float, original_length: int, summary_length: int, tags: list[str]) -> str¶

get_summary(key: str) -> dict[str, Any] | None¶

search_summaries(query: str, top_k: int = 5) -> list[dict[str, Any]]¶

delete_summary(key: str) -> bool¶

store_chunk_analysis(key: str, original_text: str, chunk_count: int, chunk_lengths_json: str, avg_chunk_length: float, total_chars: int, chunk_size_param: int, tags: list[str]) -> str¶

get_chunk_analysis(key: str) -> dict[str, Any] | None¶

list_chunk_analyses(limit: int = 50) -> list[dict[str, Any]]¶

create_template(name: str, description: str, strategy: str, target_length: int, chunk_size: int, max_depth: int, tags: list[str]) -> str¶

get_template(name: str) -> dict[str, Any] | None¶

list_templates(limit: int = 50) -> list[dict[str, Any]]¶

increment_template_use(name: str) -> None¶

store_context(name: str, text: str, description: str, tags: list[str]) -> str¶

get_context(name: str) -> dict[str, Any] | None¶

list_contexts(limit: int = 50) -> list[dict[str, Any]]¶

add_job(session_id: str, strategy: str, status: str, llm_calls: int, iterations: int, depth_used: int, error_message: str) -> str¶

get_recent_jobs(limit: int = 10) -> list[dict[str, Any]]¶

count_all() -> dict[str, int]¶

Functions¶

summarize_ctx_recursive_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]¶

Summarise runtime-vs-fallback ctx_recursive rerank decisions with redaction.

agentic_planner_enabled(default_enabled: bool) -> bool¶

Decide whether the agentic summary-rerank planner should be used.

validate_summary_rerank_decision(decision: SummaryRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

Returned-set / anti-injection guard for a summary-rerank decision.

planner_is_llm_trusted(planner: Any) -> bool¶

Whether the BLOCK may report llm_used=True for planner.

applied_agentic_patterns() -> list[dict[str, Any]]¶

Return compact metadata for ctx_recursive-applied vendored patterns.

get_skill_catalog() -> CtxRecursiveSkillCatalog¶

MCP Tools¶