Ctx Rag¶

ctx_rag — mvp.ctx_rag

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

Overview¶

ctx_rag implements a full RAG pipeline: chunk → index → retrieve → optional rerank → optional generate. Supports three retrieval backends: TF-IDF (sparse, default), BM25 (sparse), and embedding (dense via litellm). Optional SQLite persistence via db_path ensures indexed chunks survive process restart. The 25-operation MCP sub-package adds document management, hybrid search, reranking, embedding, and retrieval evaluation.

Public outputs carry the G6 envelope fields completion_state, warning_card, evidence, request_id, task_id, and run_id. Degraded retrieval, rerank suppression/fallback, generation failure, validation errors, and resource limits surface stable G6_E_* codes rather than requiring callers to parse [RAG_ERROR] strings. MCP info includes machine-readable capabilities for retrieval strategies, persistence, embeddings, LLM paths, constraints, source operations, and production gate status.

Retrieved and indexed document content is treated as untrusted external input. Content-surfacing operations (tier-1 infer retrieval and MCP retrieve, search, hybrid_search, rerank, get_document) stamp evidence.content_provenance="untrusted_indexed_documents" plus an injection_scan over chunk/candidate text, and surface as qualified-draft (never verified) so callers do not treat retrieved passages as trusted. On a prompt-injection signal the output carries a G6_E_RAG_UNTRUSTED_CONTENT_INJECTION_DETECTED warning card and the content is preserved (annotate, not block). Deterministic metric and introspection operations (info, list_patterns, chunk_stats, index_info, embed_*, similarity, score_relevance, evaluate_retrieval) remain verified.

When to use:

• Retrieving relevant context from document collections for RAG pipelines
• Indexing and searching large corpora with TF-IDF, BM25, or dense embedding strategies
• Knowledge-augmented reasoning with scored, chunked retrieval
• Persistent document indexing that survives restart (set db_path)

Example:

from mvp.ctx_rag import CtxRAGBlock, RAGInput

block = CtxRAGBlock(name="rag")
result = block.infer(RAGInput(
    query="neural architecture search",
    documents=["NAS automates model design...", "Transformers use attention..."],
))
# result.ok → True; result.value → RAGOutput with retrieved_chunks and augmented_context

Works well with: grounding, ctx_colbert, ctx_search, ctx_recursive

Public API¶

CtxRAGDecisionError(ValueError)¶

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

RAGRerankDecision¶

Validated advisory rerank verdict over an eligible candidate set.

LLMRAGRerankRuntime¶

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

Constructor:

Methods:

rerank(query: str, candidates: list[str]) -> RAGRerankDecision¶

CtxRAGRerankPatternRuntime¶

Stateless, load-bearing eligibility-ceiling enforcement.

Methods:

enforce_eligibility(decision: RAGRerankDecision, candidates: list[str], scores: list[float]) -> tuple[list[str], list[float], bool, bool]¶

CtxRAGPlanner¶

Runtime-first advisory rerank facade with deterministic fallback.

Constructor:

Methods:

rerank(query: str, chunks: list[str], scores: list[float]) -> tuple[list[str], list[float]]¶

CtxRAGBlock(AIBlock[RAGInput, RAGOutput, dict])¶

RAG pipeline: chunk -> index -> retrieve -> optional rerank -> optional generate.

Methods:

close() -> None¶

Close the SQLite connection if one exists.

infer(data: RAGInput) -> Result[RAGOutput]¶

bias() -> dict¶

RAGInput(BaseModel)¶

Input to CtxRAGBlock.

RAGOutput(BaseModel)¶

Output from CtxRAGBlock.

CtxRAGMCPBlock(AIBlock[MCPRAGInput, MCPRAGOutput, dict])¶

Full-featured RAG block with SQLite persistence.

Methods:

close() -> None¶

infer(data: MCPRAGInput) -> Result[MCPRAGOutput]¶

MCPRAGRecord(BaseModel)¶

MCPRAGInput(BaseModel)¶

MCPRAGOutput(BaseModel)¶

RAGStore¶

Sync SQLite RAG store with 5 tables.

Constructor:

Methods:

close() -> None¶

add_document(content: str, source: str = '', file_type: str = '', hash_val: str = '', metadata: dict | None = None, tags: list[str] | None = None, chunk_count: int = 0) -> str¶

get_document(doc_id: str) -> dict[str, Any] | None¶

list_documents(source: str = '', file_type: str = '', limit: int = 50) -> list[dict[str, Any]]¶

delete_document(doc_id: str) -> bool¶

add_chunk(doc_id: str, content: str, chunk_index: int = 0, embedding_json: str = '', metadata: dict | None = None) -> str¶

get_chunk(chunk_id: str) -> dict[str, Any] | None¶

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

get_all_chunk_contents() -> list[str]¶

upsert_index(name: str, config_json: str = '{}', doc_count: int = 0, chunk_count: int = 0) -> str¶

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

add_search_entry(query: str, results_json: str = '[]', scores_json: str = '[]', strategy: str = 'tfidf', top_k: int = 5, result_count: int = 0) -> str¶

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

add_evaluation(search_id: str = '', metric: str = '', score: float = 0.0, details_json: str = '{}') -> str¶

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

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

Functions¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

Decide whether the agentic rerank planner should be used.

eligible_fingerprint(candidates: list[str]) -> str¶

sha256 over the eligible candidate texts (order-sensitive).

clamp_to_returned_set(reordered: list[Any], candidates: list[Any]) -> list[Any]¶

Block-boundary eligible-set ceiling (defense in depth, injection seam).

validate_rerank_decision(decision: RAGRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

Eligible-set / anti-injection guard for a 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_rag-applied vendored patterns.

get_skill_catalog() -> CtxRAGSkillCatalog¶

MCP Tools¶