Ctx Mnm¶

Ctx MnM — mvp.ctx_mnm

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

Overview¶

Mix-and-Match dynamic context assembler that combines scored source chunks from multiple retrievers into a single, budget-aware context string. Three assembly strategies are supported: concat (input order), ranked (by score descending), and weighted (per-source weight multipliers applied before ranking). An optional Jaccard-based deduplication pass removes near-duplicate chunks before assembly.

When to use:

Merging results from several retrieval sources (vector search, BM25, knowledge graph) into one context window
Enforcing a character budget when building prompts from heterogeneous retrieved chunks
Removing redundant passages before sending context to an LLM

Example:

from mvp.ctx_mnm import CtxMnMBlock, MnMInput, SourceChunk

block = CtxMnMBlock(name="mnm")
result = block.infer(MnMInput(
    sources=[SourceChunk(content="Fact A", score=0.9, source_id="wiki"),
             SourceChunk(content="Fact B", score=0.7, source_id="arxiv")],
    strategy="ranked", max_chars=500,
))
# result.value.assembled_context → ranked, deduplicated context string

Works well with: ctx_rag, ctx_elastic, ctx_colbert

MCP surface caveat

ctx_mnm exposes a 25-tool FastMCP surface through mnm_mcp/server.py. The skill/mcp/server.py file is only a compatibility entry point that delegates to that server; do not treat it as the source of the tool list or expect the old 5-tool wrapper API. For first-user launch flows, hide this low-level tool count behind one or two guided workflows so non-technical users can assemble, store, and search context without choosing among 25 operations manually.

Guided MCP workflows: Start with assemble-store-search, trajectory-replay, or search-rerank from skill/SKILL.md instead of reading the full 25-op list. mnm_info exposes machine-readable capabilities, backend health, index_freshness, relevance_floor, and store_signature; mnm_search returns relevance_floor and store_signature directly. Search remains advisory because no positive relevance floor or persistent embedding index is configured.

Public API¶

`CtxMnMDecisionError(ValueError)`¶

The LLM did not produce a usable, validated search-rerank decision.

`MnMRerankDecision`¶

Validated advisory rerank verdict over a returned search-record set.

Field	Type	Default
`ordered_indices`	`tuple[int, ...]`	required
`dropped_indices`	`tuple[int, ...]`	()
`rationale`	`str`	''
`eligible_fingerprint`	`str`	''
`confidence`	`float`	0.0
`degraded`	`bool`	False
`raw_response`	`str`	''

`LLMMnMRerankRuntime`¶

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

Constructor:

Parameter	Type	Default
`llm`	`LLMCaller \\| None`	None

Methods:

`rerank(query: str, records: list[Any]) -> MnMRerankDecision`¶

`CtxMnMRerankPatternRuntime`¶

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

Methods:

`enforce_eligibility(decision: MnMRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]`¶

`CtxMnMPlanner`¶

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

Constructor:

Parameter	Type	Default
`runtime`	`MnMRerankRuntime \\| None`	None
`pattern_runtime`	`CtxMnMRerankPatternRuntime \\| None`	None

Methods:

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

`CtxMnMBlock(AIBlock[MnMInput, MnMOutput, None])`¶

Mix-and-Match context assembler.

Field	Type	Default
`name`	`str`	'ctx_mnm'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MnMInput) -> Result[MnMOutput]`¶

`SourceChunk(BaseModel)`¶

Field	Type	Default
`source_id`	`str`	required
`content`	`str`	required
`score`	`float`	1.0
`metadata`	`dict[str, str]`	Field(default_factory=dict)

`MnMInput(BaseModel)`¶

Field	Type	Default
`sources`	`list[SourceChunk]`	required
`query`	`str`	''
`max_chars`	`int`	4000
`strategy`	`Literal['concat', 'ranked', 'weighted']`	'ranked'
`deduplicate`	`bool`	True
`weights`	`dict[str, float]`	Field(default_factory=dict)
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

Methods:

`sources_not_empty(v: list[SourceChunk]) -> list[SourceChunk]`¶

`max_chars_positive(v: int) -> int`¶

`MnMOutput(BaseModel)`¶

Field	Type	Default
`assembled_context`	`str`	required
`chunks_used`	`int`	required
`total_chars`	`int`	required
`n_sources`	`int`	required
`strategy`	`str`	required
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None

`CtxMnMMCPBlock(AIBlock[MCPMnMInput, MCPMnMOutput, dict])`¶

25-op MnM context assembler with SQLite persistence.

Field	Type	Default
`name`	`str`	'ctx_mnm_mcp'
`state`	`dict \\| None`	None
`db_path`	`str`	':memory:'
`resource_bounds`	`ResourceBounds \\| None`	None
`agentic_planner`	`CtxMnMPlanner \\| None`	None

Methods:

`infer(data: MCPMnMInput) -> Result[MCPMnMOutput]`¶

`MCPMnMRecord(BaseModel)`¶

Generic record returned from the MnM MCP store.

Field	Type	Default
`id`	`str`	''
`record_type`	`str`	''
`key`	`str`	''
`value`	`str`	''
`timestamp`	`str`	''
`tags`	`str`	''
`metadata`	`dict[str, Any]`	Field(default_factory=dict)

`MCPMnMInput(BaseModel)`¶

Flattened input for all 25 MnM MCP operations.

Field	Type	Default
`op`	`MnMOp`	required
`chunks_json`	`str`	''
`strategy`	`str`	'ranked'
`max_chars`	`int`	4000
`deduplicate`	`bool`	True
`weights_json`	`str`	''
`session_id`	`str`	''
`tags`	`str`	''
`notes`	`str`	''
`source_id`	`str`	''
`content`	`str`	''
`score`	`float`	1.0
`metadata_json`	`str`	''
`chunk_id`	`str`	''
`name`	`str`	''
`benchmark`	`str`	''
`filter_type`	`str`	'string'
`pattern`	`str`	''
`instructions`	`str`	''
`function_name`	`str`	''
`code_snippet`	`str`	''
`steps_json`	`str`	''
`project_name`	`str`	''
`group_name`	`str`	''
`base_model`	`str`	''
`lora_r`	`int`	128
`max_lr`	`float`	1e-05
`n_epochs`	`int`	2
`student_dropout_rate`	`float`	0.9
`train_patterns_json`	`str`	''
`val_patterns_json`	`str`	''
`hyperparams_json`	`str`	''
`config_name_a`	`str`	''
`config_name_b`	`str`	''
`task`	`str`	''
`return_cls_name`	`str`	''
`half_steps_json`	`str`	''
`template_text`	`str`	''
`variables_json`	`str`	''
`auto_generate`	`bool`	False
`description`	`str`	''
`query`	`str`	''
`top_k`	`int`	5
`limit`	`int`	50
`threshold`	`float`	0.7
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`MCPMnMOutput(BaseModel)`¶

Output for all 25 MnM MCP operations.

Field	Type	Default
`op`	`str`	''
`found`	`bool`	False
`count`	`int`	0
`records`	`list[MCPMnMRecord]`	Field(default_factory=list)
`message`	`str`	''
`summary`	`str`	''
`assembled_context`	`str`	''
`chunks_used`	`int`	0
`total_chars`	`int`	0
`n_sources`	`int`	0
`strategy_used`	`str`	''
`comparison`	`list[dict[str, Any]]`	Field(default_factory=list)
`reasoning`	`str`	''
`accepted`	`bool \\| None`	None
`accept_count`	`int`	0
`reject_count`	`int`	0
`config_diff`	`dict[str, Any]`	Field(default_factory=dict)
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`Literal['qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`relevance_floor`	`dict[str, Any]`	Field(default_factory=dict)
`store_signature`	`dict[str, Any]`	Field(default_factory=dict)
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)

`MnMStore`¶

SQLite-backed store with 7 tables for the MnM MCP sub-package.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`close() -> None`¶

`add_assembly(**kwargs: Any) -> dict`¶

`query_assemblies(session_id: str = '', strategy: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

`add_chunk(source_id: str, content: str, score: float = 1.0, metadata_json: str = '{}', tags: str = '') -> dict`¶

`query_chunks(source_id: str = '', tag: str = '', text: str = '', limit: int = 50) -> list[dict]`¶

`delete_chunk(chunk_id: str) -> bool`¶

`list_chunks(source_id: str = '', limit: int = 100) -> list[dict]`¶

`upsert_filter(name: str, benchmark: str = '', filter_type: str = 'string', pattern: str = '', instructions: str = '', function_name: str = '', tags: str = '') -> dict`¶

`add_filter(name: str, benchmark: str = '', filter_type: str = 'string', pattern: str = '', instructions: str = '', function_name: str = '', tags: str = '') -> dict`¶

`query_filters(benchmark: str = '', filter_type: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

`get_filter_by_name(name: str) -> dict | None`¶

`update_filter_counts(name: str, run_delta: int = 0, accept_delta: int = 0) -> None`¶

`save_session(name: str, state_json: str, notes: str = '') -> dict`¶

`load_session(name: str) -> dict | None`¶

`add_trajectory(task: str, return_cls_name: str = '', half_steps_json: str = '[]', metadata_json: str = '{}', session_id: str = '', tags: str = '') -> dict`¶

`query_trajectories(task: str = '', session_id: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

`upsert_config(name: str, **kwargs: Any) -> dict`¶

`query_configs(project_name: str = '', group_name: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

`get_config_by_name(name: str) -> dict | None`¶

`add_hint(name: str, template_text: str, variables_json: str = '{}', tags: str = '') -> dict`¶

`get_latest_hint(name: str) -> dict | None`¶

`query_hints(name: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

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

Cross-table text search. Uses TF-IDF if sklearn available, else LIKE.

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

`store_signature() -> dict[str, Any]`¶

Return count/max-timestamp signature for callers detecting stale state.

`index_freshness() -> dict[str, Any]`¶

Describe freshness honestly: search is live SQL/TF-IDF, not a durable index.

Functions¶

`eligible_fingerprint(records: list[Any]) -> str`¶

sha256 over the returned record texts (order-sensitive).

`validate_mnm_rerank_decision(decision: MnMRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

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

`planner_is_llm_trusted(planner: Any) -> bool`¶

Whether the BLOCK may report llm_used=True for planner.

MCP Tools¶

Operation	Source
`assemble`	mnm_mcp
`compare_strategies`	mnm_mcp
`check_duplicates`	mnm_mcp
`record_assembly`	mnm_mcp
`query_assemblies`	mnm_mcp
`summarize_assemblies`	mnm_mcp
`store_chunk`	mnm_mcp
`query_chunks`	mnm_mcp
`delete_chunk`	mnm_mcp
`list_chunks`	mnm_mcp
`store_filter`	mnm_mcp
`query_filters`	mnm_mcp
`apply_filter`	mnm_mcp
`evaluate_filter`	mnm_mcp
`store_config`	mnm_mcp
`query_configs`	mnm_mcp
`training_status`	mnm_mcp
`compare_configs`	mnm_mcp
`store_trajectory`	mnm_mcp
`query_trajectories`	mnm_mcp
`save_session`	mnm_mcp
`load_session`	mnm_mcp
`store_hint`	mnm_mcp
`search`	mnm_mcp
`info`	mnm_mcp
`list_patterns`	mnm_mcp

Ctx Mnm¶

Overview¶

Public API¶

CtxMnMDecisionError(ValueError)¶

MnMRerankDecision¶

LLMMnMRerankRuntime¶

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

CtxMnMRerankPatternRuntime¶

enforce_eligibility(decision: MnMRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]¶

CtxMnMPlanner¶

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

CtxMnMBlock(AIBlock[MnMInput, MnMOutput, None])¶

infer(data: MnMInput) -> Result[MnMOutput]¶

SourceChunk(BaseModel)¶

MnMInput(BaseModel)¶

sources_not_empty(v: list[SourceChunk]) -> list[SourceChunk]¶

max_chars_positive(v: int) -> int¶

MnMOutput(BaseModel)¶

CtxMnMMCPBlock(AIBlock[MCPMnMInput, MCPMnMOutput, dict])¶

infer(data: MCPMnMInput) -> Result[MCPMnMOutput]¶

MCPMnMRecord(BaseModel)¶

MCPMnMInput(BaseModel)¶

MCPMnMOutput(BaseModel)¶

MnMStore¶

close() -> None¶

add_assembly(**kwargs: Any) -> dict¶

query_assemblies(session_id: str = '', strategy: str = '', tags: str = '', limit: int = 50) -> list[dict]¶

add_chunk(source_id: str, content: str, score: float = 1.0, metadata_json: str = '{}', tags: str = '') -> dict¶

query_chunks(source_id: str = '', tag: str = '', text: str = '', limit: int = 50) -> list[dict]¶

delete_chunk(chunk_id: str) -> bool¶

list_chunks(source_id: str = '', limit: int = 100) -> list[dict]¶

upsert_filter(name: str, benchmark: str = '', filter_type: str = 'string', pattern: str = '', instructions: str = '', function_name: str = '', tags: str = '') -> dict¶

add_filter(name: str, benchmark: str = '', filter_type: str = 'string', pattern: str = '', instructions: str = '', function_name: str = '', tags: str = '') -> dict¶

query_filters(benchmark: str = '', filter_type: str = '', tags: str = '', limit: int = 50) -> list[dict]¶

get_filter_by_name(name: str) -> dict | None¶

update_filter_counts(name: str, run_delta: int = 0, accept_delta: int = 0) -> None¶

save_session(name: str, state_json: str, notes: str = '') -> dict¶

load_session(name: str) -> dict | None¶

add_trajectory(task: str, return_cls_name: str = '', half_steps_json: str = '[]', metadata_json: str = '{}', session_id: str = '', tags: str = '') -> dict¶

query_trajectories(task: str = '', session_id: str = '', tags: str = '', limit: int = 50) -> list[dict]¶

upsert_config(name: str, **kwargs: Any) -> dict¶

query_configs(project_name: str = '', group_name: str = '', tags: str = '', limit: int = 50) -> list[dict]¶

get_config_by_name(name: str) -> dict | None¶

add_hint(name: str, template_text: str, variables_json: str = '{}', tags: str = '') -> dict¶

get_latest_hint(name: str) -> dict | None¶

query_hints(name: str = '', tags: str = '', limit: int = 50) -> list[dict]¶

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

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

store_signature() -> dict[str, Any]¶

index_freshness() -> dict[str, Any]¶

Functions¶

eligible_fingerprint(records: list[Any]) -> str¶

validate_mnm_rerank_decision(decision: MnMRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

planner_is_llm_trusted(planner: Any) -> bool¶

MCP Tools¶

`CtxMnMDecisionError(ValueError)`¶

`MnMRerankDecision`¶

`LLMMnMRerankRuntime`¶

`rerank(query: str, records: list[Any]) -> MnMRerankDecision`¶

`CtxMnMRerankPatternRuntime`¶

`enforce_eligibility(decision: MnMRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]`¶

`CtxMnMPlanner`¶

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

`CtxMnMBlock(AIBlock[MnMInput, MnMOutput, None])`¶

`infer(data: MnMInput) -> Result[MnMOutput]`¶

`SourceChunk(BaseModel)`¶

`MnMInput(BaseModel)`¶

`sources_not_empty(v: list[SourceChunk]) -> list[SourceChunk]`¶

`max_chars_positive(v: int) -> int`¶

`MnMOutput(BaseModel)`¶

`CtxMnMMCPBlock(AIBlock[MCPMnMInput, MCPMnMOutput, dict])`¶

`infer(data: MCPMnMInput) -> Result[MCPMnMOutput]`¶

`MCPMnMRecord(BaseModel)`¶

`MCPMnMInput(BaseModel)`¶

`MCPMnMOutput(BaseModel)`¶

`MnMStore`¶

`close() -> None`¶

`add_assembly(**kwargs: Any) -> dict`¶

`query_assemblies(session_id: str = '', strategy: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

`add_chunk(source_id: str, content: str, score: float = 1.0, metadata_json: str = '{}', tags: str = '') -> dict`¶

`query_chunks(source_id: str = '', tag: str = '', text: str = '', limit: int = 50) -> list[dict]`¶

`delete_chunk(chunk_id: str) -> bool`¶

`list_chunks(source_id: str = '', limit: int = 100) -> list[dict]`¶

`upsert_filter(name: str, benchmark: str = '', filter_type: str = 'string', pattern: str = '', instructions: str = '', function_name: str = '', tags: str = '') -> dict`¶

`add_filter(name: str, benchmark: str = '', filter_type: str = 'string', pattern: str = '', instructions: str = '', function_name: str = '', tags: str = '') -> dict`¶

`query_filters(benchmark: str = '', filter_type: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

`get_filter_by_name(name: str) -> dict | None`¶

`update_filter_counts(name: str, run_delta: int = 0, accept_delta: int = 0) -> None`¶

`save_session(name: str, state_json: str, notes: str = '') -> dict`¶

`load_session(name: str) -> dict | None`¶

`add_trajectory(task: str, return_cls_name: str = '', half_steps_json: str = '[]', metadata_json: str = '{}', session_id: str = '', tags: str = '') -> dict`¶

`query_trajectories(task: str = '', session_id: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

`upsert_config(name: str, **kwargs: Any) -> dict`¶

`query_configs(project_name: str = '', group_name: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

`get_config_by_name(name: str) -> dict | None`¶

`add_hint(name: str, template_text: str, variables_json: str = '{}', tags: str = '') -> dict`¶

`get_latest_hint(name: str) -> dict | None`¶

`query_hints(name: str = '', tags: str = '', limit: int = 50) -> list[dict]`¶

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

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

`store_signature() -> dict[str, Any]`¶

`index_freshness() -> dict[str, Any]`¶

`eligible_fingerprint(records: list[Any]) -> str`¶

`validate_mnm_rerank_decision(decision: MnMRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

`planner_is_llm_trusted(planner: Any) -> bool`¶