Retrieval pipeline

Every query runs through a RetrieverPipeline — an sklearn-shaped chain of named, addressable steps (Pipeline([(name, step), …])). A RetrieverPipeline is a RetrieverStep, so pipelines compose recursively: nest one as a step inside another for sub-routing, and address any step by name (pipeline["fetch"]) for introspection or testing.

The default chunk-search pipeline (dense + graph expansion)

The shipped default (python/pydocs_mcp/pipelines/chunk_search_graph.yaml) is a seven-step dense + reference-graph chain:

  1. pre_filter — parse + validate + scope-split; writes a typed result to state.scratch for the fetcher.

  2. dense_fetcher — embed the query, ANN-search the .tq sidecar; writes the candidate set (the ANN index score is used directly as relevance — no separate scorer step).

  3. metadata_post_filter — apply any remaining SearchQuery.post_filter in-memory.

  4. graph_expand — seed from the top dense hits, add their 1-hop caller/callee neighbours (the structurally-adjacent code an embedding alone misses).

  5. top_k_filter — sort by relevance, keep top K.

  6. limit — cap the final item count.

  7. token_budget_formatter — render the composite chunk for MCP output.

The former BM25-only chain (chunk_search.yaml) remains a shipped preset. On the RepoQA benchmark, this dense + graph default lifts recall@10 from 0.40 (BM25) to 0.77 on standard queries and to 1.00 on structurally-reachable answers.

Dense, hybrid, late-interaction, and tree-reasoning retrieval

Several more retrieval modes ship as opt-in pipeline presets:

  • Dense (chunk_search_dense.yaml, chunk_search_dense_ranked.yaml) — a DenseFetcherStep queries the TurboQuant vector store using embeddings from the configured Embedder (FastEmbed BAAI/bge-small-en-v1.5 by default; OpenAI and the on-device sentence_transformers provider — Qwen/Qwen3-Embedding-0.6B, Alibaba-NLP/gte-modernbert-base, or the code-strong codefuse-ai/F2LLM-v2-0.6B — optional). Its ANN index score is used directly as relevance, so the dense presets carry no separate scorer step.

  • Hybrid (chunk_search_hybrid.yaml, chunk_search_hybrid_ranked.yaml) — a ParallelStep runs the BM25 and dense branches concurrently, then an RRFFusionStep merges them with reciprocal-rank fusion into one ranking, followed by a post-fusion DenseScorerStep re-rank over the fused candidates.

  • Graph (dense + reference-graph expansion) (chunk_search_graph.yaml, chunk_search_graph_ranked.yaml) — a GraphExpandStep seeds from the top dense hits and pulls in their 1-hop reference-graph neighbours (callers / callees / overriding subclass methods), merging them into the dense ranking by max(dense_sim, seed_sim · decay) — embedding-centric (no RRF/BM25). Recovers the structurally-adjacent answer a dense embedder misses; degrades to dense-only when the index has no reference graph. See the structural-recall benchmark.

  • Late-interaction — opt-in multi-vector (ColBERT / PyLate via fast-plaid) MaxSim scoring; enable with the [late-interaction] extra and late_interaction.enabled: true.

  • LLM tree-reasoning (tree_only.yaml, chunk_search_with_tree_reasoning_parallel.yaml, chunk_search_with_tree_reasoning_after.yaml) — vectorless RAG: an LlmTreeReasoningStep walks the project’s DocumentNode tree (each node enriched with its real signature, decorators, and a docstring excerpt) with an LLM and fetches the nodes it selects — no embeddings required. Opt-in via the llm: config section; can run standalone, in a branch parallel to hybrid, or as a two-stage reranker over a BM25/dense candidate set (rerank_candidates).

Select a preset by pointing the chunk pipeline at it in your config overlay (see Configuration); the default is dense + graph expansion (chunk_search_graph.yaml).

Graph analytics (opt-in)

Two further reference-graph signals are computed at index time when enabled (both off by default; PageRank/community detection needs the [graph] extra — pip install 'pydocs-mcp[graph]'):

  • Node scores (reference_graph.node_scores.enabled: true) — a single post-index pass computes in-degree, PageRank, and Louvain community per symbol into a node_scores table. Two rerank steps consume it: centrality_prior (boost structurally central “god-node” APIs by a normalised PageRank/in-degree prior — rerank-only, can’t hurt recall) and community_diversity (greedy MMR-by-community so the top-k spans subsystems instead of near-duplicates from one module). Add either step to a chunk pipeline YAML.

  • Parent rollup (parent_rollup step in a chunk pipeline YAML) — when several results are children of one symbol or document section (e.g. three methods of the same class), the step replaces them with the parent’s own indexed chunk at the group’s best rank. Collapse requires at least two co-retrieved siblings AND a per-kind share of the parent’s children, tuned via min_coverage_by_kind (defaults: class: 0.3, module: 0.6, markdown_heading: 0.5; a supplied mapping replaces the defaults wholesale) with the min_coverage fallback (0.5) for other kinds. Place it after top_k_filter and before limit:

    - name: rollup
  type: parent_rollup
  params:
    min_coverage: 0.5
    min_coverage_by_kind:
      class: 0.3
      module: 0.6
      markdown_heading: 0.5

  • Similar edges (reference_graph.similar_edges.enabled: true, top_m: N) — the synthesize_similar_edges ingestion stage adds kind='similar' embedding-kNN edges between each symbol and its nearest neighbours, densifying the AST graph so graph_expand (with kinds: [calls, inherits, similar]) can reach semantically-related code that has no call/inherit edge. similar edges are excluded from node-score centrality (which stays structural). Note: the kNN runs over chunks embedded in the current index pass, so a complete similar-edge graph requires a full index --force; an incremental reindex of a touched package recomputes them from its re-embedded chunks only.

Embedder inference runs on CPU by default. Pass --gpu to serve / index / watch to move it onto CUDA — same vectors, same cache, lower latency (see GPU inference).

Routing

ConditionalStep and RouteStep route per query type — e.g. send long or structural queries down a different branch than short keyword lookups — without modifying the branches themselves.

Building pipelines in Python

For tests, benchmarks, or embedded usage, build an IngestionPipeline and a RetrieverPipeline programmatically, no YAML required:

import asyncio
import tempfile
from pathlib import Path

from pydocs_mcp.application import ProjectIndexer
from pydocs_mcp.db import open_index_database
from pydocs_mcp.extraction import (
    AstMemberExtractor,
    PipelineChunkExtractor,
    StaticDependencyResolver,
    build_ingestion_pipeline,
)
from pydocs_mcp.models import SearchQuery
from pydocs_mcp.retrieval.config import AppConfig
from pydocs_mcp.retrieval.formatters import ChunkFormatter
from pydocs_mcp.retrieval.pipeline import (
    PerCallConnectionProvider,
    RetrieverPipeline,
    RetrieverState,
)
from pydocs_mcp.retrieval.steps import (
    BM25ScorerStep,
    ChunkFetcherStep,
    LimitStep,
    MetadataPostFilterStep,
    TokenBudgetStep,
    TopKFilterStep,
)
from pydocs_mcp.storage.factories import (
    build_connection_provider,
    build_sqlite_indexing_service,
    build_sqlite_uow_factory,
)
from pydocs_mcp.storage.sqlite import SqliteChunkRepository, SqliteFilterAdapter


async def main() -> None:
    # 1. Fresh SQLite + ingestion pipeline from default AppConfig
    db_path = Path(tempfile.mkstemp(suffix=".sqlite")[1])
    open_index_database(db_path).close()
    config = AppConfig.load()

    indexer = ProjectIndexer(
        indexing_service=build_sqlite_indexing_service(db_path),
        dependency_resolver=StaticDependencyResolver(),
        chunk_extractor=PipelineChunkExtractor(pipeline=build_ingestion_pipeline(config)),
        member_extractor=AstMemberExtractor(),
        uow_factory=build_sqlite_uow_factory(db_path),
    )
    await indexer.index_project(Path("/path/to/your/project"))
    await SqliteChunkRepository(provider=build_connection_provider(db_path)).rebuild_index()

    # 2. RetrieverPipeline composed from named, addressable steps
    provider = PerCallConnectionProvider(cache_path=db_path)
    pipeline = RetrieverPipeline(
        name="chunk_search",
        steps=(
            ("fetch", ChunkFetcherStep(provider=provider, filter_adapter=SqliteFilterAdapter())),
            ("score", BM25ScorerStep(name="bm25_scorer")),
            ("post_filter", MetadataPostFilterStep(name="metadata_post_filter")),
            ("topk", TopKFilterStep(name="top_k_filter")),
            ("limit", LimitStep(name="limit")),
            ("budget", TokenBudgetStep(formatter=ChunkFormatter(), budget=2000, name="token_budget_formatter")),
        ),
    )

    # 3. Run a search
    state = await pipeline.run(RetrieverState(query=SearchQuery(terms="async retry")))
    if state.result is not None:
        print(state.result.items[0].text[:500])


asyncio.run(main())