Ctx Colbert¶
ctx_colbert -- mvp.ctx_colbert
Cluster: Context & Retrieval | Type: component | MCP Tools: 28
Overview¶
Stage 2 context retrieval inspired by ColBERT. It provides real ColBERT/RAGatouille retrieval when the optional dense backend is installed, embedding retrieval when EMBEDDING_MODEL is configured, and TF-IDF sparse fallback in default installs.
Full ColBERT backend is opt-in
This component is functional in the default install, but real ColBERT retrieval requires ragatouille==0.0.9.post2. Without it, outputs are marked degraded=True and use embedding retrieval when EMBEDDING_MODEL is configured, then TF-IDF fallback. If requires_dense_retrieval=True, fallback retrieval is blocked with completion_state="blocked" and warning_card.code="G6_E_COLBERT_DENSE_REQUIRED_UNAVAILABLE". Install with poetry install -E colbert, pip install "g6_mcp[colbert]", or build the MCP Docker image with --build-arg G6_FULL_RETRIEVAL=true. Run G6_RUN_RAGATOUILLE_SMOKE=1 python -m pytest tests/mvp/ctx_colbert/test_ragatouille_real_smoke.py -m heavy -q before claiming the target deployment has been verified with full ColBERT.
MCP index ops are metadata-only
create_index, rebuild_index, index_status, and list_indexes manage SQLite index records. They do not persist backend-native RAGatouille/ColBERT artifact directories. Outputs disclose metadata.index_artifact_mode="sqlite_metadata_only", warning_card.code="G6_E_COLBERT_INDEX_METADATA_ONLY", and supporting evidence.
When to use:
- High-quality passage retrieval over a document corpus
- Dense retrieval with late-interaction scoring when the optional backend is installed
- Document ranking and re-ranking pipelines
- Capability discovery before retrieval via
resource_check
Example:
from mvp.ctx_colbert import CtxColBERTBlock, ColBERTInput
block = CtxColBERTBlock(name="cb")
result = block.infer(ColBERTInput(query="transformer attention", corpus=["Attention is all you need.", "RNNs use recurrence."]))
# result.ok -> True; result.value -> ColBERTOutput with ranked results and scores
agentic_evidence is emitted only for advisory LLM rerank paths. It describes whether an LLM was used, whether the rerank degraded or diverged, fallback counts, and the bounded decision trail; agentic_rerank=False opts out and leaves this evidence empty.
Works well with: ctx_rag, grounding, ctx_search
Public API¶
CtxColBERTBlock(AIBlock[ColBERTInput, ColBERTOutput, dict])¶
Dense retrieval with graceful fallback.
| Field | Type | Default |
|---|---|---|
name | str | 'ctx_colbert' |
index_name | str | 'mvp_colbert' |
resource_bounds | ResourceBounds \| None | None |
usage | ResourceUsage | field(default_factory=ResourceUsage) |
agentic_planner | CtxColBERTPlanner \| None | None |
Methods:
infer(data: ColBERTInput) -> Result[ColBERTOutput]¶
bias() -> dict¶
ColBERTInput(BaseModel)¶
Input to CtxColBERTBlock.
| Field | Type | Default |
|---|---|---|
query | str | required |
corpus | list[str] | Field(default_factory=list) |
top_k | int | 5 |
requires_dense_retrieval | bool | False |
agentic_rerank | bool \| None | None |
run_mode | Literal['beta', 'production'] | 'beta' |
reviewer_signature | str | '' |
ColBERTOutput(BaseModel)¶
Output from CtxColBERTBlock.
| Field | Type | Default |
|---|---|---|
query | str | required |
results | list[str] | required |
scores | list[float] | required |
retriever | str | required |
degraded | bool | False |
degradation_reason | str | '' |
completion_state | str | 'complete' |
warning_card | dict | Field(default_factory=dict) |
evidence | dict | Field(default_factory=dict) |
request_id | str \| None | None |
task_id | str \| None | None |
run_id | str \| None | None |
agentic_evidence | dict | Field(default_factory=dict) |
CtxColBERTMCPBlock(AIBlock[MCPColBERTInput, MCPColBERTOutput, dict])¶
25-op ColBERT retrieval engine with SQLite persistence.
| Field | Type | Default |
|---|---|---|
name | str | 'ctx_colbert_mcp' |
state | dict \| None | None |
db_path | str | ':memory:' |
resource_bounds | ResourceBounds \| None | None |
agentic_planner | CtxColBERTPlanner \| None | None |
Methods:
infer(data: MCPColBERTInput) -> Result[MCPColBERTOutput]¶
MCPColBERTInput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | Literal['search', 'batch_search', 'rerank', 'batch_rerank', 'similarity', 'add_documents', 'remove_document', 'get_document', 'list_documents', 'clear_corpus', 'create_index', 'rebuild_index', 'delete_index', 'index_status', 'list_indexes', 'qa_retrieve', 'explain_results', 'summarize_results', 'search_history', 'corpus_stats', 'log_search', 'clear_history', 'get_config', 'update_config', 'resource_check', 'ops', 'help', 'list_patterns'] | required |
query | str | '' |
queries | list[str] | Field(default_factory=list) |
top_k | int | 5 |
namespace | str | 'default' |
documents | list[str] | Field(default_factory=list) |
document_id | str | '' |
metadata_json | str | '{}' |
candidates | list[str] | Field(default_factory=list) |
batch_candidates | list[list[str]] | Field(default_factory=list) |
text_a | str | '' |
text_b | str | '' |
requires_dense_retrieval | bool | False |
offset | int | 0 |
limit | int | 50 |
model | str | '' |
config_key | str | '' |
config_value | str | '' |
max_execution_seconds | int | 120 |
max_tokens_per_minute | int | 10000 |
max_tokens_per_hour | int | 100000 |
results_json | str | '[]' |
scores_json | str | '[]' |
retriever | str | '' |
agentic_rerank | bool \| None | None |
run_mode | Literal['beta', 'production'] | 'beta' |
reviewer_signature | str | '' |
MCPColBERTOutput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | str | '' |
status | str | 'ok' |
data_json | str | '{}' |
message | str | '' |
error | str | '' |
count | int | 0 |
found | bool | False |
results | list[dict] | Field(default_factory=list) |
metadata | dict[str, Any] | Field(default_factory=dict) |
degraded | bool | False |
degradation_reason | str | '' |
completion_state | str | 'complete' |
warning_card | dict | Field(default_factory=dict) |
evidence | dict | Field(default_factory=dict) |
request_id | str \| None | None |
task_id | str \| None | None |
run_id | str \| None | None |
agentic_evidence | dict | Field(default_factory=dict) |
ColBERTStore¶
Sync SQLite store with 5 tables.
Constructor:
| Parameter | Type | Default |
|---|---|---|
db_path | str | ':memory:' |
Methods:
add_document(content: str, namespace: str = 'default', metadata_json: str = '{}') -> str¶
add_documents(contents: list[str], namespace: str = 'default', metadata_json: str = '{}') -> list[str]¶
remove_document(doc_id: str, namespace: str = 'default') -> bool¶
get_document(doc_id: str, namespace: str = 'default') -> dict[str, Any] | None¶
list_documents(namespace: str = 'default', offset: int = 0, limit: int = 50) -> list[dict[str, Any]]¶
clear_corpus(namespace: str = 'default') -> int¶
get_documents_content(namespace: str = 'default') -> list[str]¶
Return all document contents for a namespace (for retrieval).
create_index(namespace: str = 'default', model: str = 'tfidf') -> str¶
delete_index(namespace: str = 'default') -> bool¶
get_index_status(namespace: str = 'default') -> dict[str, Any] | None¶
list_indexes() -> list[dict[str, Any]]¶
log_search(query: str, namespace: str = 'default', top_k: int = 5, results_json: str = '[]', retriever: str = '', scores_json: str = '[]') -> str¶
search_history(namespace: str = 'default', limit: int = 50) -> list[dict[str, Any]]¶
clear_history(namespace: str = 'default') -> int¶
save_qa_result(query: str, answer: str, sources_json: str = '[]', model: str = '', namespace: str = 'default') -> str¶
get_config(key: str) -> str | None¶
set_config(key: str, value_json: str) -> None¶
list_configs() -> dict[str, Any]¶
corpus_stats(namespace: str = 'default') -> dict[str, Any]¶
count_all() -> dict[str, int]¶
MCP Tools¶
| Operation | Source |
|---|---|
search | colbert_mcp |
batch_search | colbert_mcp |
rerank | colbert_mcp |
batch_rerank | colbert_mcp |
similarity | colbert_mcp |
add_documents | colbert_mcp |
remove_document | colbert_mcp |
get_document | colbert_mcp |
list_documents | colbert_mcp |
clear_corpus | colbert_mcp |
create_index | colbert_mcp |
rebuild_index | colbert_mcp |
delete_index | colbert_mcp |
index_status | colbert_mcp |
list_indexes | colbert_mcp |
qa_retrieve | colbert_mcp |
explain_results | colbert_mcp |
summarize_results | colbert_mcp |
search_history | colbert_mcp |
corpus_stats | colbert_mcp |
log_search | colbert_mcp |
clear_history | colbert_mcp |
get_config | colbert_mcp |
update_config | colbert_mcp |
resource_check | colbert_mcp |
ops | colbert_mcp |
help | colbert_mcp |
list_patterns | colbert_mcp |