Build a Hybrid Evidence Gate

Build a hybrid retrieval gate that traces evidence and abstains on weak results.

Introduction

30 Second Summary

Search results can look convincing even when they rest on one weak source or miss the exact medical term that matters. Without a clear trail back to the evidence, confidence can hide how little support an answer has.

In this project, you will build a CPU-friendly evidence layer that fuses existing dense retrieval results with a BM25 lexical lane through Reciprocal Rank Fusion. Its JSON report uses passage provenance to make a transparent answer-or-abstain decision before generation.

What You'll Build

Picture one command producing a traceable evidence report that reveals why a supported question can proceed while a weak question is stopped.

By the end of this project, you'll have:

A visible lexical search showing ranked passage IDs beside positive BM25 scores for exact-term matches.
A hybrid retrieval trace showing each candidate's dense rank beside its BM25 rank. Every row also keeps its RRF score plus paper ID.
An answer-or-abstain evidence gate that accepts the supported fixture. The same gate rejects the unsupported question when independent-paper coverage or cross-lane agreement is missing.
Secret Mission: Sweep the evidence thresholds to compare decisions across stricter paper-coverage settings. Repeat the comparison across stricter cross-lane-agreement settings.

Are there any prerequisites?

You'll need a working MedCPT retrieval path connected to Chroma whose rows expose passage ID, paper ID, text, plus dense score.

The walkthrough assumes Python 3.8 or later on Windows with Visual Studio Code plus Git already available.

Before We Start

A retriever can still return plausible-looking passages for an unsupported question. This checkpoint helps you define the evidence failure your gate must catch before generation begins.

Set Up the Evidence-Gate Module

Your existing MedCPT retriever already works. Its Chroma storage also holds a working baseline.

This step gives the evidence gate its own folder. A local virtual environment holds bm25s version 0.3.14 without changing that baseline.

In this step, get ready to:

Create an isolated hybrid_gate workspace inside the existing repository.
Pin the bm25s dependency in requirements.txt.
Activate a local environment with the verified package version.

Create the isolated workspace

The new folder keeps the evidence-gate files separate from the working retrieval baseline. Visual Studio Code shows that boundary clearly in its Explorer.

Press the Windows key to open search.
Type Visual Studio Code.
Press Enter to start Visual Studio Code.
Select File from the menu bar.
Select Open Folder....
Select the folder that contains your existing retrieval repository.
Select Select Folder.

Your working repository is now visible in the Explorer without any files being changed.

Select the repository folder at the top of the Explorer.
Select the New Folder icon.
Enter hybrid_gate as the folder name.
Select File from the menu bar.
Select Open Folder....
Select the hybrid_gate folder inside the existing repository.
Select Select Folder.

The Explorer now shows hybrid_gate as the workspace folder. The original retrieval files remain outside this workspace.

Select the New File icon in the Explorer.
Enter requirements.txt as the file name.
Add the pinned dependency by pasting this line into requirements.txt:

bm25s==0.3.14

What does this requirement do?

The == pin tells pip to install exactly version 0.3.14. A fixed package version keeps the retrieval setup reproducible.

Press Ctrl+S to save requirements.txt.
Confirm that the Explorer lists requirements.txt inside hybrid_gate.

Cannot find the requirements file?

Check that the Explorer shows hybrid_gate as the top-level folder. Reopen that folder if requirements.txt was created elsewhere.

Ask for help with finding the misplaced requirements file or visit the NextWork community.

✔️ Awesome, I've got everything!

Great. Your hybrid_gate workspace now contains the saved dependency file.

ⓧ I'd like to double check the full code

Compare your complete requirements.txt file with this reference.

bm25s==0.3.14

What should match?

The file contains one dependency pin. The package name and version must match the reference exactly.

Create and activate the local environment

The local environment stores this module's packages inside hybrid_gate\.venv. Your existing MedCPT environment keeps its current package versions.

Right-click the hybrid_gate folder in the Explorer.
Select Open in Integrated Terminal.

A terminal panel opens with hybrid_gate as its current folder.

Select the Launch Profile dropdown in the terminal panel.
Select PowerShell.

You now have a PowerShell terminal rooted in the isolated workspace.

Create the local virtual environment by running this command:

python -m venv .venv

What does this command do?

Python creates a virtual environment in the .venv folder. Packages installed after activation stay inside this workspace.

Confirm that the Explorer now lists a .venv folder inside hybrid_gate.

Missing the virtual environment folder?

Confirm that the terminal is rooted in hybrid_gate. A different terminal location places .venv in the wrong folder.

Ask for help with creating the virtual environment in the correct folder or visit the NextWork community.

Activate the local environment by running this command:

.venv\Scripts\Activate.ps1

Why activate the environment?

Activation points Python package commands at the local .venv environment. This keeps the new dependency away from the working baseline.

You should see the environment name at the start of your PowerShell prompt. That prompt confirms the isolated environment is active.

PowerShell blocking activation?

PowerShell may enforce an execution policy that blocks activation scripts. Follow your organization's approved PowerShell policy before changing any security setting.

Ask for help with activating the local environment safely or visit the NextWork community.

Install the pinned dependency from requirements.txt by running this command:

python -m pip install -r requirements.txt

What does this install command do?

The command reads the dependency pin from requirements.txt. The active environment receives bm25s version 0.3.14.

The terminal prints installation progress before returning to the activated prompt. Your evidence-gate dependency is now isolated from the existing retriever.

Dependency installation failed?

Confirm that the environment name still appears at the start of the prompt. Also confirm that requirements.txt contains the exact pinned line from the reference tab.

Ask for help with diagnosing the bm25s installation or visit the NextWork community.

Verify the pinned package version

Package inspection confirms which version the active environment can use. This final check proves the installation followed the pin in requirements.txt.

Before you check, which version do you expect the active environment to report?

Inspect the installed package by running this command:

python -m pip show bm25s

What does this check prove?

The command asks the active Python environment for its installed bm25s package details. The reported version comes from the isolated environment you just activated.

You should see package details containing Version: 0.3.14. That result confirms the pin and the installation match.

Why the baseline stays unchanged

Every package command in this step runs through the activated hybrid_gate\.venv environment. No MedCPT packages were upgraded.

No Chroma migration command touched the existing storage. The working retrieval path remains available for the later adapter step.

Your isolated evidence-gate workspace is ready. Next up, you will add a visible lexical retrieval lane without changing the dense retriever.

Build the BM25 Retrieval Lane

That isolation is now in place. Your working MedCPT and Chroma path remains untouched.

Dense retrieval can miss exact biomedical terms. A separate BM25 lexical retrieval lane gives those terms an observable ranking before fusion.

In this step, get ready to:

Model each passage with stable provenance.
Build a BM25 index that removes zero-score results.
Run the supported fixture through the lexical lane.

Model passages for retrieval

The lexical index needs a stable connection between each text passage and its source paper. The bm25s library can return your passage IDs alongside its scores.

Use the file sidebar in Visual Studio Code to create hybrid_gate.py inside the open hybrid_gate folder.
Add the imports plus the two retrieval records by pasting this code into hybrid_gate.py:

from dataclasses import dataclass
from typing import Dict, List, Sequence

import bm25s


@dataclass(frozen=True)
class Passage:
    passage_id: str
    paper_id: str
    text: str


@dataclass(frozen=True)
class DenseHit:
    passage_id: str
    score: float

What Do These Records Hold?

Passage keeps the stable passage ID with the source paper ID.
Passage.text stores the text that BM25 indexes.
DenseHit defines the normalized dense-result shape used by the later fusion boundary.
frozen=True prevents these records from changing after creation.

Save hybrid_gate.py.
Check the file sidebar. You should see hybrid_gate.py beside requirements.txt.

Is the New File Missing?

Check that the file name is exactly hybrid_gate.py.
Check that the file sits directly inside the open hybrid_gate folder.

Ask for help with the file location: Help me confirm that hybrid_gate.py is inside my hybrid_gate folder.

HybridEvidenceGate owns the corpus plus the retrieval settings. Its constructor rejects an empty corpus or invalid ranking limits before building an index.

Place this class below DenseHit in hybrid_gate.py:

class HybridEvidenceGate:
    def __init__(
        self,
        passages: Sequence[Passage],
        candidate_k: int = 10,
        output_k: int = 5,
        rrf_k: int = 60,
        min_distinct_papers: int = 2,
        min_cross_lane_hits: int = 1,
    ) -> None:
        if not passages:
            raise ValueError("At least one passage is required.")
        if candidate_k < 1 or output_k < 1:
            raise ValueError("candidate_k and output_k must be positive.")
        if rrf_k < 0:
            raise ValueError("rrf_k must be non-negative.")

        self.passages = list(passages)
        self.candidate_k = candidate_k
        self.output_k = output_k
        self.rrf_k = rrf_k
        self.min_distinct_papers = min_distinct_papers
        self.min_cross_lane_hits = min_cross_lane_hits

What Does the Constructor Set Up?

passages becomes a reusable list of indexed records.
candidate_k limits how many candidates each retrieval lane can return.
output_k limits the final evidence list used in later stages.
rrf_k plus the two minimum settings establish configuration fields for later fusion and gating.

Save hybrid_gate.py.
Check the constructor indentation. You should see every assignment nested inside __init__().

Does the Class Structure Look Broken?

Align def __init__ one indentation level inside HybridEvidenceGate.
Align each validation plus each assignment one indentation level inside __init__().

Ask for help with the constructor: Help me check the indentation in HybridEvidenceGate.__init__().

A passage lookup preserves stable IDs after retrieval. The BM25 corpus stores those IDs so a ranked result can be traced back to its text and paper.

Find the self.min_cross_lane_hits = min_cross_lane_hits line in hybrid_gate.py.
Add the lookup plus the BM25 index directly below that line:

        self._passages_by_id: Dict[str, Passage] = {}
        for passage in self.passages:
            if passage.passage_id in self._passages_by_id:
                raise ValueError(
                    "Duplicate passage_id: {0}".format(passage.passage_id)
                )
            self._passages_by_id[passage.passage_id] = passage

        corpus_tokens = bm25s.tokenize(
            [passage.text for passage in self.passages],
            stopwords="en",
        )
        self._bm25 = bm25s.BM25(
            corpus=[passage.passage_id for passage in self.passages]
        )
        self._bm25.index(corpus_tokens)

How Does the Index Preserve Provenance?

_passages_by_id maps every stable ID back to its complete passage record.
bm25s.tokenize() prepares the passage text for lexical matching with English stopwords removed.
bm25s.BM25() stores the passage IDs as the return corpus.
self._bm25.index() builds the searchable lexical index from the tokenized text.

Save hybrid_gate.py.
Check the final indentation. You should see the lookup plus the BM25 setup inside __init__().

Is the Index Setup Misaligned?

Keep the for passage loop aligned with corpus_tokens inside the constructor.
Keep the duplicate-ID check nested inside the loop.
Confirm that your activated environment still contains the installed bm25s package.

Ask for help with the index setup: Help me check the BM25 initialization inside HybridEvidenceGate.

Build the lexical search

A lexical query can return candidates whose score is zero. Filtering those rows gives an unsupported query an empty lexical result instead of arbitrary evidence.

Add lexical_search() below the constructor in hybrid_gate.py:

    def lexical_search(self, question: str) -> List[Dict[str, object]]:
        query_tokens = bm25s.tokenize([question], stopwords="en")
        limit = min(self.candidate_k, len(self.passages))
        result_ids, scores = self._bm25.retrieve(query_tokens, k=limit)

        ranked: List[Dict[str, object]] = []
        for passage_id, raw_score in zip(result_ids[0], scores[0]):
            score = float(raw_score)
            if score <= 0.0:
                continue
            ranked.append(
                {
                    "passage_id": str(passage_id),
                    "bm25_score": score,
                    "rank": len(ranked) + 1,
                }
            )
        return ranked

How Does the Search Stay Auditable?

query_tokens applies the same tokenization settings used for the corpus.
limit prevents retrieval from requesting more rows than the corpus contains.
score <= 0.0 removes candidates without positive lexical evidence.
rank is assigned after filtering so the returned positions remain consecutive.

Save hybrid_gate.py.
Check the method boundary. You should see lexical_search() aligned with __init__() inside the class.

Is the Search Method Nested Incorrectly?

Align def lexical_search with def __init__.
Keep the ranking loop inside lexical_search().
Check that return ranked sits outside the loop.

Ask for help with the method structure: Help me check the indentation and filtering in lexical_search().

✔️ Awesome, I've got everything!

Your data records, passage lookup, BM25 index, and lexical search method are now saved in hybrid_gate.py.

ⓧ I'd like to double check the full code

Compare your saved file with the complete hybrid_gate.py code for this step.

from dataclasses import dataclass
from typing import Dict, List, Sequence

import bm25s


@dataclass(frozen=True)
class Passage:
    passage_id: str
    paper_id: str
    text: str


@dataclass(frozen=True)
class DenseHit:
    passage_id: str
    score: float


class HybridEvidenceGate:
    def __init__(
        self,
        passages: Sequence[Passage],
        candidate_k: int = 10,
        output_k: int = 5,
        rrf_k: int = 60,
        min_distinct_papers: int = 2,
        min_cross_lane_hits: int = 1,
    ) -> None:
        if not passages:
            raise ValueError("At least one passage is required.")
        if candidate_k < 1 or output_k < 1:
            raise ValueError("candidate_k and output_k must be positive.")
        if rrf_k < 0:
            raise ValueError("rrf_k must be non-negative.")

        self.passages = list(passages)
        self.candidate_k = candidate_k
        self.output_k = output_k
        self.rrf_k = rrf_k
        self.min_distinct_papers = min_distinct_papers
        self.min_cross_lane_hits = min_cross_lane_hits

        self._passages_by_id: Dict[str, Passage] = {}
        for passage in self.passages:
            if passage.passage_id in self._passages_by_id:
                raise ValueError(
                    "Duplicate passage_id: {0}".format(passage.passage_id)
                )
            self._passages_by_id[passage.passage_id] = passage

        corpus_tokens = bm25s.tokenize(
            [passage.text for passage in self.passages],
            stopwords="en",
        )
        self._bm25 = bm25s.BM25(
            corpus=[passage.passage_id for passage in self.passages]
        )
        self._bm25.index(corpus_tokens)

    def lexical_search(self, question: str) -> List[Dict[str, object]]:
        query_tokens = bm25s.tokenize([question], stopwords="en")
        limit = min(self.candidate_k, len(self.passages))
        result_ids, scores = self._bm25.retrieve(query_tokens, k=limit)

        ranked: List[Dict[str, object]] = []
        for passage_id, raw_score in zip(result_ids[0], scores[0]):
            score = float(raw_score)
            if score <= 0.0:
                continue
            ranked.append(
                {
                    "passage_id": str(passage_id),
                    "bm25_score": score,
                    "rank": len(ranked) + 1,
                }
            )
        return ranked

What Does This Full File Contain?

This reference combines the immutable retrieval records with the indexed passage corpus. It also contains the positive-score lexical search used by the demo.

Run the supported fixture

A fixed fixture makes the first retrieval result repeatable. The demo prints the ranked output as JSON so every score stays visible.

Use the file sidebar to create fixtures.py inside the open hybrid_gate folder.
Add the imports plus the first two passages to fixtures.py:

from typing import Dict, List

from hybrid_gate import DenseHit, HybridEvidenceGate, Passage


DEMO_PASSAGES = [
    Passage(
        passage_id="p1",
        paper_id="paper-metformin",
        text=(
            "Metformin is a common first-line pharmacologic treatment "
            "for adults with type 2 diabetes."
        ),
    ),
    Passage(
        passage_id="p2",
        paper_id="paper-lifestyle",
        text=(
            "Lifestyle treatment for type 2 diabetes includes nutrition "
            "changes, physical activity, and weight management."
        ),
    ),
]

What Do These Fixtures Represent?

p1 links a metformin passage to paper-metformin.
p2 links a lifestyle passage to a separate paper.
DEMO_PASSAGES keeps the corpus order stable across repeated runs.

Save fixtures.py.
Check the file sidebar. You should see fixtures.py beside hybrid_gate.py.

Are the First Passages Misaligned?

Keep both Passage() records inside DEMO_PASSAGES.
Keep each continued text string inside its surrounding parentheses.

Ask for help with the fixture list: Help me check the first two Passage records in fixtures.py.

Find the closing bracket for DEMO_PASSAGES in fixtures.py.
Insert the remaining passages immediately before that closing bracket:

    Passage(
        passage_id="p3",
        paper_id="paper-insulin",
        text=(
            "Insulin therapy can be used for type 2 diabetes when other "
            "treatments do not provide adequate glycemic control."
        ),
    ),
    Passage(
        passage_id="p4",
        paper_id="paper-aspirin",
        text=(
            "Aspirin has antiplatelet effects and is used in selected "
            "cardiovascular settings."
        ),
    ),

Why Add Two Contrasting Passages?

The insulin passage adds another relevant diabetes treatment from an independent paper. The aspirin passage supplies an unrelated candidate for the unsupported scenario.

Save fixtures.py.
Count the records inside DEMO_PASSAGES. You should see four stable passage IDs from p1 through p4.

Does the Fixture List End Too Early?

Place p3 plus p4 before the final DEMO_PASSAGES bracket.
Keep a comma after every Passage() record.

Ask for help with the list boundary: Help me place p3 and p4 inside DEMO_PASSAGES.

The fixture includes one supported question plus one unsupported question. Its dense-hit rows keep both scenarios available through the same data boundary.

Add the scenario questions plus dense-hit rows below DEMO_PASSAGES:

QUESTIONS: Dict[str, str] = {
    "supported": "Which treatments are used for type 2 diabetes?",
    "unsupported": "Which antibiotic treats bacterial pneumonia?",
}

DENSE_HITS: Dict[str, List[DenseHit]] = {
    "supported": [
        DenseHit("p1", 0.91),
        DenseHit("p2", 0.87),
        DenseHit("p3", 0.82),
    ],
    "unsupported": [
        DenseHit("p4", 0.31),
    ],
}

How Are the Scenarios Paired?

QUESTIONS gives each scenario a repeatable query.
DENSE_HITS records the normalized dense baseline for those same scenario names.
supported points to three diabetes passages.
unsupported points to the unrelated aspirin passage.

Save fixtures.py.
Check the two dictionaries. You should see matching supported plus unsupported keys in each one.

Do the Scenario Names Differ?

Use supported as the first key in both dictionaries.
Use unsupported as the second key in both dictionaries.
Keep every dictionary key inside quotes.

Ask for help with the scenario mapping: Help me compare the keys in QUESTIONS and DENSE_HITS.

A small factory creates the same indexed gate for each demo run. Its candidate limit matches the three relevant fixture passages.

Add build_demo_gate() below DENSE_HITS in fixtures.py:

def build_demo_gate(
    min_distinct_papers: int = 2,
    min_cross_lane_hits: int = 1,
) -> HybridEvidenceGate:
    return HybridEvidenceGate(
        DEMO_PASSAGES,
        candidate_k=3,
        output_k=3,
        min_distinct_papers=min_distinct_papers,
        min_cross_lane_hits=min_cross_lane_hits,
    )

Why Use a Gate Factory?

build_demo_gate() keeps fixture construction in one place.
candidate_k=3 limits the lexical lane to three candidates.
output_k=3 prepares the same output limit for later hybrid retrieval.
The two minimum settings stay explicit so later checks can change them without editing the class.

Save fixtures.py.
Check the bottom of the file. You should see build_demo_gate() returning one HybridEvidenceGate instance.

Is the Factory Outside the File Structure?

Align def build_demo_gate with the left edge of the file.
Keep return HybridEvidenceGate nested inside the function.

Ask for help with the factory: Help me check build_demo_gate() in fixtures.py.

✔️ Awesome, I've got everything!

Your fixture corpus, scenario questions, dense-hit rows, and gate factory are now saved in fixtures.py.

ⓧ I'd like to double check the full code

Compare your saved file with the complete fixtures.py code for this step.

from typing import Dict, List

from hybrid_gate import DenseHit, HybridEvidenceGate, Passage


DEMO_PASSAGES = [
    Passage(
        passage_id="p1",
        paper_id="paper-metformin",
        text=(
            "Metformin is a common first-line pharmacologic treatment "
            "for adults with type 2 diabetes."
        ),
    ),
    Passage(
        passage_id="p2",
        paper_id="paper-lifestyle",
        text=(
            "Lifestyle treatment for type 2 diabetes includes nutrition "
            "changes, physical activity, and weight management."
        ),
    ),
    Passage(
        passage_id="p3",
        paper_id="paper-insulin",
        text=(
            "Insulin therapy can be used for type 2 diabetes when other "
            "treatments do not provide adequate glycemic control."
        ),
    ),
    Passage(
        passage_id="p4",
        paper_id="paper-aspirin",
        text=(
            "Aspirin has antiplatelet effects and is used in selected "
            "cardiovascular settings."
        ),
    ),
]

QUESTIONS: Dict[str, str] = {
    "supported": "Which treatments are used for type 2 diabetes?",
    "unsupported": "Which antibiotic treats bacterial pneumonia?",
}

DENSE_HITS: Dict[str, List[DenseHit]] = {
    "supported": [
        DenseHit("p1", 0.91),
        DenseHit("p2", 0.87),
        DenseHit("p3", 0.82),
    ],
    "unsupported": [
        DenseHit("p4", 0.31),
    ],
}


def build_demo_gate(
    min_distinct_papers: int = 2,
    min_cross_lane_hits: int = 1,
) -> HybridEvidenceGate:
    return HybridEvidenceGate(
        DEMO_PASSAGES,
        candidate_k=3,
        output_k=3,
        min_distinct_papers=min_distinct_papers,
        min_cross_lane_hits=min_cross_lane_hits,
    )

What Does This Full File Contain?

This reference combines the four-passage corpus with two contrasting questions. It also creates the indexed gate used by the demo.

The command-line demo selects a scenario plus a retrieval stage. This first version exposes only the lexical stage.

Use the file sidebar to create demo.py inside the open hybrid_gate folder.
Add the argument parser plus the lexical output dictionary to demo.py:

import argparse
import json

from fixtures import QUESTIONS, build_demo_gate


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Inspect lexical retrieval results."
    )
    parser.add_argument(
        "--scenario",
        choices=["supported", "unsupported"],
        required=True,
    )
    parser.add_argument(
        "--stage",
        choices=["lexical"],
        required=True,
    )
    args = parser.parse_args()

    gate = build_demo_gate()
    question = QUESTIONS[args.scenario]
    output = {
        "question": question,
        "lexical_results": gate.lexical_search(question),
    }

How Does the Demo Select a Query?

--scenario selects either the supported fixture or the unsupported fixture.
--stage restricts this version of the demo to lexical retrieval.
build_demo_gate() creates the indexed fixture corpus.
lexical_search() produces the ranked rows stored under lexical_results.

Save demo.py.
Check the file sidebar. You should see demo.py beside fixtures.py.

Does the Output Dictionary Sit Outside main()?

Keep gate plus question inside main().
Keep both output keys inside the output dictionary.

Ask for help with the demo body: Help me check the structure of main() in demo.py.

Add the output print plus the script entry point at the bottom of demo.py:

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

How Does the Demo Print Its Result?

json.dumps() converts the output dictionary into readable JSON.
indent=2 places nested fields on separate indented lines.
The script entry point runs main() when you execute demo.py.

Save demo.py.

Before you run this, consider which passage IDs the diabetes question is likely to return.

Run the supported lexical scenario in the activated PowerShell terminal with this command:

python demo.py --scenario supported --stage lexical

What Should You See?

The JSON output contains the supported diabetes question plus a ranked lexical_results list. Every returned row has a stable passage_id plus a positive bm25_score value.

The consecutive rank values prove that zero-score candidates were removed before ranking.

That is the first retrieval lane working: exact diabetes terms now produce traceable passage rankings.

Did the Lexical Demo Fail?

Check that PowerShell still shows the activated hybrid_gate\.venv environment.
Check that demo.py plus fixtures.py sit beside hybrid_gate.py.
Compare the spelling of build_demo_gate plus lexical_search across the three files.

Ask for help with the run: Help me diagnose why the supported lexical demo is not printing ranked BM25 results.

✔️ Awesome, I've got everything!

Your terminal now prints stable passage IDs with positive BM25 scores. Double check that demo.py is saved.

ⓧ I'd like to double check the full code

Compare your saved file with the complete demo.py code for this step.

import argparse
import json

from fixtures import QUESTIONS, build_demo_gate


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Inspect lexical retrieval results."
    )
    parser.add_argument(
        "--scenario",
        choices=["supported", "unsupported"],
        required=True,
    )
    parser.add_argument(
        "--stage",
        choices=["lexical"],
        required=True,
    )
    args = parser.parse_args()

    gate = build_demo_gate()
    question = QUESTIONS[args.scenario]
    output = {
        "question": question,
        "lexical_results": gate.lexical_search(question),
    }
    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

What Does This Full File Contain?

This reference selects one fixture scenario and runs it through the BM25 lexical lane. It prints the question plus the complete ranked result as readable JSON.

Your BM25 lane now turns exact biomedical terms into a visible ranking. Next, you will combine this signal with the existing dense results while preserving every source field.

Fuse Dense and Lexical Rankings

Your BM25 lane now ranks exact-term matches with stable passage IDs. The fixture data also supplies a dense retrieval ranking for the same corpus.

The two lanes produce incompatible raw scores. Reciprocal Rank Fusion combines their rank positions while keeping provenance visible.

In this step, get ready to:

Normalize dense hits into a stable candidate list.
Fuse dense ranks with BM25 ranks using RRF.
Inspect the hybrid JSON trace for supported and unsupported questions.

Normalize dense inputs

Dense results need stable passage IDs before fusion can compare their positions. Normalization also prevents duplicate rows from receiving extra influence.

In hybrid_gate.py, place this method directly below lexical_search:

    def _normalize_dense_hits(
        self, dense_hits: Sequence[DenseHit]
    ) -> List[DenseHit]:
        normalized: List[DenseHit] = []
        seen = set()
        for hit in dense_hits:
            if hit.passage_id not in self._passages_by_id:
                raise ValueError(
                    "Unknown dense passage_id: {0}".format(hit.passage_id)
                )
            if hit.passage_id in seen:
                continue
            seen.add(hit.passage_id)
            normalized.append(hit)
            if len(normalized) == self.candidate_k:
                break
        return normalized

What does this method do?

The seen set ensures that each passage contributes only once.
The lookup in _passages_by_id rejects dense results that cannot be connected to known provenance.
The candidate_k limit keeps the dense lane within the same candidate budget used by lexical retrieval.

Save hybrid_gate.py.
Confirm that lexical retrieval still works by rerunning the established command:

python demo.py --scenario supported --stage lexical

What should you see?

You should still see ranked passage IDs with positive bm25_score values. This confirms that the new helper did not disturb the existing lexical lane.

Did the lexical check stop working?

Check that _normalize_dense_hits has the same indentation as lexical_search. Make sure the new method begins after the final return ranked line.

Ask for help with the method placement:

Fuse the ranked lanes

RRF gives each passage a contribution based on its 1-indexed position in each lane. The configured rrf_k value of 60 controls how quickly that contribution changes between ranks.

Continue the HybridEvidenceGate class by adding the first part of hybrid_search below _normalize_dense_hits:

    def hybrid_search(
        self, question: str, dense_hits: Sequence[DenseHit]
    ) -> List[Dict[str, object]]:
        dense = self._normalize_dense_hits(dense_hits)
        lexical = self.lexical_search(question)

        dense_rank = {
            hit.passage_id: rank
            for rank, hit in enumerate(dense, start=1)
        }
        dense_score = {hit.passage_id: hit.score for hit in dense}
        lexical_rank = {
            str(row["passage_id"]): int(row["rank"])
            for row in lexical
        }
        lexical_score = {
            str(row["passage_id"]): float(row["bm25_score"])
            for row in lexical
        }

How are the lanes represented?

The method creates separate rank and score lookups for each lane. Rank numbering begins at 1 so each position can feed directly into the RRF calculation.

The raw scores remain available for auditing. They do not compete with each other during fusion.

Continue inside hybrid_search by adding the RRF calculation directly below lexical_score:

        rrf_scores: Dict[str, float] = {}
        for passage_id, rank in dense_rank.items():
            rrf_scores[passage_id] = rrf_scores.get(passage_id, 0.0) + (
                1.0 / (self.rrf_k + rank)
            )
        for passage_id, rank in lexical_rank.items():
            rrf_scores[passage_id] = rrf_scores.get(passage_id, 0.0) + (
                1.0 / (self.rrf_k + rank)
            )

        def sort_key(item: object) -> object:
            passage_id, score = item
            dense_position = dense_rank.get(
                passage_id, self.candidate_k + 1
            )
            lexical_position = lexical_rank.get(
                passage_id, self.candidate_k + 1
            )
            return (-score, min(dense_position, lexical_position), passage_id)

        ordered = sorted(rrf_scores.items(), key=sort_key)

How does the fusion work?

A passage receives one reciprocal contribution for each lane where it appears. Passages found by both lanes accumulate two contributions.

The sorting key puts the highest RRF score first. Earlier source ranks break score ties before the stable passage ID does.

Finish hybrid_search by adding the evidence builder directly below ordered:

        evidence: List[Dict[str, object]] = []
        for passage_id, rrf_score in ordered[: self.output_k]:
            passage = self._passages_by_id[passage_id]
            evidence.append(
                {
                    "passage_id": passage.passage_id,
                    "paper_id": passage.paper_id,
                    "text": passage.text,
                    "dense_rank": dense_rank.get(passage_id),
                    "dense_score": dense_score.get(passage_id),
                    "bm25_rank": lexical_rank.get(passage_id),
                    "bm25_score": lexical_score.get(passage_id),
                    "rrf_score": rrf_score,
                }
            )
        return evidence

What does each evidence row preserve?

Every row carries its passage text plus its passage and paper identifiers. The dense rank and BM25 rank show which lanes found that passage.

Missing lane values remain None in Python. The JSON output renders those missing values as null.

Save hybrid_gate.py.

Expose the hybrid stage

The demo needs the dense fixture rows before it can call the new fusion path. Its output keeps the lexical stage available for comparison.

Replace the contents of demo.py with the next two consecutive snippets.
Start with the imports and argument setup:

import argparse
import json

from fixtures import DENSE_HITS, QUESTIONS, build_demo_gate


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Inspect lexical and hybrid retrieval stages."
    )
    parser.add_argument(
        "--scenario",
        choices=["supported", "unsupported"],
        required=True,
    )
    parser.add_argument(
        "--stage",
        choices=["lexical", "hybrid"],
        required=True,
    )
    args = parser.parse_args()

    gate = build_demo_gate()
    question = QUESTIONS[args.scenario]
    dense_hits = DENSE_HITS[args.scenario]

What changes in the setup?

The import now includes DENSE_HITS. The stage choices now accept either lexical or hybrid.

The selected scenario supplies the same question and dense fixture rows to the gate.

Continue directly below dense_hits with the stage output logic:

    if args.stage == "lexical":
        output = {
            "question": question,
            "lexical_results": gate.lexical_search(question),
        }
    else:
        output = {
            "question": question,
            "warning": "Fusion ranks candidates but does not prove sufficiency.",
            "evidence": gate.hybrid_search(question, dense_hits),
        }

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

How does the demo choose a path?

The lexical branch preserves the earlier behavior. The hybrid branch sends the selected dense rows into hybrid_search.

The warning records the boundary of this stage. Fusion produces a ranking without making an evidence-sufficiency decision.

Save demo.py.
Run the supported hybrid scenario with this command:

python demo.py --scenario supported --stage hybrid

What should the supported trace show?

You should see evidence rows for p1, p2, and p3. Each row preserves its paper ID plus its dense rank and BM25 rank.

Your fusion path is working. You can now trace how both retrieval lanes contributed to the final ranking.

Missing a rank or evidence row?

Check that all three parts of hybrid_search remain inside the method. Confirm that DENSE_HITS appears in the import from fixtures.

Ask for help comparing the fused trace:

Before you run the unsupported scenario, do you think rank fusion will reject the weak question or still return a candidate?

Reveal the unsupported hybrid result by running:

python demo.py --scenario unsupported --stage hybrid

What does the result prove?

You should see p4 with a dense rank. Its bm25_rank and bm25_score values should be null.

This is the intended shortfall. RRF can rank the only available dense candidate even when the lexical lane found no supporting evidence.

Did the unsupported scenario fail to run?

Check that the --stage choices include hybrid. Confirm that the hybrid branch passes dense_hits into hybrid_search.

Ask for help tracing the unsupported path:

You have exposed the critical ranking gap. The trace now shows exactly why a fused candidate still needs an independent sufficiency decision.

✔️ Awesome, I've got everything!

Great. Double check that you saved hybrid_gate.py and demo.py.

ⓧ I'd like to double check the full code

Compare your cumulative hybrid_gate.py with this reference.

from dataclasses import dataclass
from typing import Dict, List, Sequence

import bm25s


@dataclass(frozen=True)
class Passage:
    passage_id: str
    paper_id: str
    text: str


@dataclass(frozen=True)
class DenseHit:
    passage_id: str
    score: float


class HybridEvidenceGate:
    def __init__(
        self,
        passages: Sequence[Passage],
        candidate_k: int = 10,
        output_k: int = 5,
        rrf_k: int = 60,
        min_distinct_papers: int = 2,
        min_cross_lane_hits: int = 1,
    ) -> None:
        if not passages:
            raise ValueError("At least one passage is required.")
        if candidate_k < 1 or output_k < 1:
            raise ValueError("candidate_k and output_k must be positive.")
        if rrf_k < 0:
            raise ValueError("rrf_k must be non-negative.")

        self.passages = list(passages)
        self.candidate_k = candidate_k
        self.output_k = output_k
        self.rrf_k = rrf_k
        self.min_distinct_papers = min_distinct_papers
        self.min_cross_lane_hits = min_cross_lane_hits

        self._passages_by_id: Dict[str, Passage] = {}
        for passage in self.passages:
            if passage.passage_id in self._passages_by_id:
                raise ValueError(
                    "Duplicate passage_id: {0}".format(passage.passage_id)
                )
            self._passages_by_id[passage.passage_id] = passage

        corpus_tokens = bm25s.tokenize(
            [passage.text for passage in self.passages],
            stopwords="en",
        )
        self._bm25 = bm25s.BM25(
            corpus=[passage.passage_id for passage in self.passages]
        )
        self._bm25.index(corpus_tokens)

    def lexical_search(self, question: str) -> List[Dict[str, object]]:
        query_tokens = bm25s.tokenize([question], stopwords="en")
        limit = min(self.candidate_k, len(self.passages))
        result_ids, scores = self._bm25.retrieve(query_tokens, k=limit)

        ranked: List[Dict[str, object]] = []
        for passage_id, raw_score in zip(result_ids[0], scores[0]):
            score = float(raw_score)
            if score <= 0.0:
                continue
            ranked.append(
                {
                    "passage_id": str(passage_id),
                    "bm25_score": score,
                    "rank": len(ranked) + 1,
                }
            )
        return ranked

    def _normalize_dense_hits(
        self, dense_hits: Sequence[DenseHit]
    ) -> List[DenseHit]:
        normalized: List[DenseHit] = []
        seen = set()
        for hit in dense_hits:
            if hit.passage_id not in self._passages_by_id:
                raise ValueError(
                    "Unknown dense passage_id: {0}".format(hit.passage_id)
                )
            if hit.passage_id in seen:
                continue
            seen.add(hit.passage_id)
            normalized.append(hit)
            if len(normalized) == self.candidate_k:
                break
        return normalized

    def hybrid_search(
        self, question: str, dense_hits: Sequence[DenseHit]
    ) -> List[Dict[str, object]]:
        dense = self._normalize_dense_hits(dense_hits)
        lexical = self.lexical_search(question)

        dense_rank = {
            hit.passage_id: rank
            for rank, hit in enumerate(dense, start=1)
        }
        dense_score = {hit.passage_id: hit.score for hit in dense}
        lexical_rank = {
            str(row["passage_id"]): int(row["rank"])
            for row in lexical
        }
        lexical_score = {
            str(row["passage_id"]): float(row["bm25_score"])
            for row in lexical
        }

        rrf_scores: Dict[str, float] = {}
        for passage_id, rank in dense_rank.items():
            rrf_scores[passage_id] = rrf_scores.get(passage_id, 0.0) + (
                1.0 / (self.rrf_k + rank)
            )
        for passage_id, rank in lexical_rank.items():
            rrf_scores[passage_id] = rrf_scores.get(passage_id, 0.0) + (
                1.0 / (self.rrf_k + rank)
            )

        def sort_key(item: object) -> object:
            passage_id, score = item
            dense_position = dense_rank.get(
                passage_id, self.candidate_k + 1
            )
            lexical_position = lexical_rank.get(
                passage_id, self.candidate_k + 1
            )
            return (-score, min(dense_position, lexical_position), passage_id)

        ordered = sorted(rrf_scores.items(), key=sort_key)
        evidence: List[Dict[str, object]] = []
        for passage_id, rrf_score in ordered[: self.output_k]:
            passage = self._passages_by_id[passage_id]
            evidence.append(
                {
                    "passage_id": passage.passage_id,
                    "paper_id": passage.paper_id,
                    "text": passage.text,
                    "dense_rank": dense_rank.get(passage_id),
                    "dense_score": dense_score.get(passage_id),
                    "bm25_rank": lexical_rank.get(passage_id),
                    "bm25_score": lexical_score.get(passage_id),
                    "rrf_score": rrf_score,
                }
            )
        return evidence

What should match?

This file includes the lexical lane plus the new dense normalization and hybrid fusion methods. The class ends after return evidence at this stage.

Compare your cumulative fixtures.py with this unchanged reference.

from typing import Dict, List

from hybrid_gate import DenseHit, HybridEvidenceGate, Passage


DEMO_PASSAGES = [
    Passage(
        passage_id="p1",
        paper_id="paper-metformin",
        text=(
            "Metformin is a common first-line pharmacologic treatment "
            "for adults with type 2 diabetes."
        ),
    ),
    Passage(
        passage_id="p2",
        paper_id="paper-lifestyle",
        text=(
            "Lifestyle treatment for type 2 diabetes includes nutrition "
            "changes, physical activity, and weight management."
        ),
    ),
    Passage(
        passage_id="p3",
        paper_id="paper-insulin",
        text=(
            "Insulin therapy can be used for type 2 diabetes when other "
            "treatments do not provide adequate glycemic control."
        ),
    ),
    Passage(
        passage_id="p4",
        paper_id="paper-aspirin",
        text=(
            "Aspirin has antiplatelet effects and is used in selected "
            "cardiovascular settings."
        ),
    ),
]

QUESTIONS: Dict[str, str] = {
    "supported": "Which treatments are used for type 2 diabetes?",
    "unsupported": "Which antibiotic treats bacterial pneumonia?",
}

DENSE_HITS: Dict[str, List[DenseHit]] = {
    "supported": [
        DenseHit("p1", 0.91),
        DenseHit("p2", 0.87),
        DenseHit("p3", 0.82),
    ],
    "unsupported": [
        DenseHit("p4", 0.31),
    ],
}


def build_demo_gate(
    min_distinct_papers: int = 2,
    min_cross_lane_hits: int = 1,
) -> HybridEvidenceGate:
    return HybridEvidenceGate(
        DEMO_PASSAGES,
        candidate_k=3,
        output_k=3,
        min_distinct_papers=min_distinct_papers,
        min_cross_lane_hits=min_cross_lane_hits,
    )

Why is this file unchanged?

The fixture already contains both supported and unsupported dense rankings. This step consumes those rows without changing the corpus or scenarios.

Compare your cumulative demo.py with this reference.

import argparse
import json

from fixtures import DENSE_HITS, QUESTIONS, build_demo_gate


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Inspect lexical and hybrid retrieval stages."
    )
    parser.add_argument(
        "--scenario",
        choices=["supported", "unsupported"],
        required=True,
    )
    parser.add_argument(
        "--stage",
        choices=["lexical", "hybrid"],
        required=True,
    )
    args = parser.parse_args()

    gate = build_demo_gate()
    question = QUESTIONS[args.scenario]
    dense_hits = DENSE_HITS[args.scenario]

    if args.stage == "lexical":
        output = {
            "question": question,
            "lexical_results": gate.lexical_search(question),
        }
    else:
        output = {
            "question": question,
            "warning": "Fusion ranks candidates but does not prove sufficiency.",
            "evidence": gate.hybrid_search(question, dense_hits),
        }

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

What should match?

This version imports DENSE_HITS and exposes both demo stages. The hybrid branch prints the warning plus the fused evidence trace.

Compare your cumulative requirements.txt with this reference.

bm25s==0.3.14

Why is the dependency file unchanged?

Fusion uses Python data structures around the existing pinned BM25 dependency. This step adds no new package.

Your retrieval lanes now merge into one auditable trace. Next, you will add explicit evidence requirements that can stop weak results before generation.

Add the Evidence Sufficiency Gate

Your fused ranking now preserves each passage's provenance across both retrieval lanes.

The unsupported question still receives a candidate, so ranking alone can send weak evidence toward generation. This step adds an explicit evidence gate for independent-paper coverage plus cross-lane agreement.

In this step, get ready to:

Measure independent-paper coverage in the fused evidence.
Measure agreement between the dense lane and lexical lane.
Return an answer-or-abstain decision in a JSON-ready report.

What makes evidence sufficient?

This project requires evidence from at least two distinct papers. It also requires at least one passage found by both retrieval lanes.

These settings form a transparent baseline. They are starting points for later evaluation.

Measure evidence support

The gate needs observable signals before it can make a decision. It counts unique papers plus passages shared by both retrieval lanes.

Return to hybrid_gate.py in Visual Studio Code.
Find the final return evidence line inside hybrid_search().
Add the following method directly below hybrid_search():

    def evaluate(
        self, question: str, dense_hits: Sequence[DenseHit]
    ) -> Dict[str, object]:
        evidence = self.hybrid_search(question, dense_hits)
        distinct_papers = len(
            {str(row["paper_id"]) for row in evidence}
        )
        cross_lane_hits = sum(
            1
            for row in evidence
            if row["dense_rank"] is not None
            and row["bm25_rank"] is not None
        )

        reasons: List[str] = []
        if distinct_papers < self.min_distinct_papers:
            reasons.append(
                "Fewer than {0} distinct papers support the result.".format(
                    self.min_distinct_papers
                )
            )
        if cross_lane_hits < self.min_cross_lane_hits:
            reasons.append(
                "Fewer than {0} passages appear in both retrieval lanes.".format(
                    self.min_cross_lane_hits
                )
            )

What does this code measure?

The evidence variable keeps the provenance-rich rows returned by hybrid_search().
The distinct_papers signal counts unique paper_id values in those rows.
The cross_lane_hits signal counts passages with both rank fields populated.
The reasons list records every baseline requirement that the evidence misses.

Save hybrid_gate.py.
Check that the existing hybrid stage still runs by using this command:

python demo.py --scenario unsupported --stage hybrid

What does this check prove?

The existing fusion path still loads after your edit. You should still see the fused p4 candidate with a dense rank plus no BM25 rank.

Does the hybrid stage fail to load?

Check that evaluate() lines up with hybrid_search() inside the class. Every line inside the new method needs one additional indentation level.

Ask for help with the method structure if the indentation still looks unclear: Help me fix the indentation in my evaluate method.

Expose the gate stage

The signals now reveal missing support. The rest of evaluate() converts those signals into a decision plus a traceable report.

In hybrid_gate.py, find the second reasons.append() block inside evaluate().
Add the following code directly below that block:

        decision = "abstain" if reasons else "answerable"
        if not reasons:
            reasons.append("The configured evidence baseline is satisfied.")

        return {
            "question": question,
            "decision": decision,
            "thresholds": {
                "min_distinct_papers": self.min_distinct_papers,
                "min_cross_lane_hits": self.min_cross_lane_hits,
            },
            "signals": {
                "distinct_papers": distinct_papers,
                "cross_lane_hits": cross_lane_hits,
                "evidence_count": len(evidence),
            },
            "reasons": reasons,
            "evidence": evidence,
        }

How does the decision work?

A non-empty reasons list produces the abstain decision.
A report with no failed requirements produces the answerable decision.
The returned dictionary keeps the configured thresholds beside the observed signals.
The evidence field preserves every provenance value from the fused rows.

Save hybrid_gate.py.
Confirm that the module still loads by running the existing hybrid scenario:

python demo.py --scenario unsupported --stage hybrid

What should still work?

You should see the same fusion report as before. This confirms that the completed method has not broken the existing retrieval stages.

Switch back to demo.py.
Select everything from import argparse through args = parser.parse_args().
Replace the selected section with this code:

import argparse
import json

from fixtures import DENSE_HITS, QUESTIONS, build_demo_gate


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Inspect lexical, hybrid, and gated retrieval stages."
    )
    parser.add_argument(
        "--scenario",
        choices=["supported", "unsupported"],
        required=True,
    )
    parser.add_argument(
        "--stage",
        choices=["lexical", "hybrid", "gate"],
        required=True,
    )
    args = parser.parse_args()

What changes in the parser?

The stage selector now accepts gate as a valid value. The description also names all three retrieval stages.

Save demo.py.
Confirm that the existing hybrid option remains available by running:

python demo.py --scenario unsupported --stage hybrid

What does this parser check show?

The command should still print the fusion warning plus the p4 evidence row. The new stage choice has preserved the earlier command path.

In demo.py, select everything from gate = build_demo_gate() through the final main() call.
Replace the selected section with this code:

    gate = build_demo_gate()
    question = QUESTIONS[args.scenario]
    dense_hits = DENSE_HITS[args.scenario]

    if args.stage == "lexical":
        output = {
            "question": question,
            "lexical_results": gate.lexical_search(question),
        }
    elif args.stage == "hybrid":
        output = {
            "question": question,
            "warning": "Fusion ranks candidates but does not prove sufficiency.",
            "evidence": gate.hybrid_search(question, dense_hits),
        }
    else:
        output = gate.evaluate(question, dense_hits)

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

How does the gate stage run?

The lexical branch still shows BM25 results. The hybrid branch still shows fused evidence.

The final branch calls evaluate() for the new gate stage. The resulting dictionary prints as formatted JSON.

Save demo.py.

Verify both decisions

The two fixtures now test opposite sides of the baseline. One has broad retrieval support while the other has one dense-only candidate.

Before you run the supported scenario, consider whether its evidence should permit generation.

Evaluate the supported fixture by running:

python demo.py --scenario supported --stage gate

What should the supported report show?

You should see "decision": "answerable" in the JSON report. The evidence satisfies both configured thresholds.

Before you run the unsupported scenario, consider whether one dense-only passage should permit generation.

Evaluate the unsupported fixture by running:

python demo.py --scenario unsupported --stage gate

What should the unsupported report show?

The report contains "decision": "abstain".
The distinct_papers signal is 1.
The cross_lane_hits signal is 0.
The reasons explain both failed baseline requirements.

That safeguard is now working. Weak retrieval reaches an explicit abstain decision before generation.

Seeing the wrong decision?

Check that the final branch in demo.py calls gate.evaluate(question, dense_hits). A call to hybrid_search() only returns ranked evidence.

Check that decision depends on whether reasons contains any failures. Get targeted help here: Help me debug my evidence-gate decision.

✔️ Awesome, I've got everything!

Your supported fixture is answerable. Your unsupported fixture now abstains with visible evidence signals.

ⓧ I'd like to double check the full code

Compare your completed hybrid_gate.py with this version:

from dataclasses import dataclass
from typing import Dict, List, Sequence

import bm25s


@dataclass(frozen=True)
class Passage:
    passage_id: str
    paper_id: str
    text: str


@dataclass(frozen=True)
class DenseHit:
    passage_id: str
    score: float


class HybridEvidenceGate:
    def __init__(
        self,
        passages: Sequence[Passage],
        candidate_k: int = 10,
        output_k: int = 5,
        rrf_k: int = 60,
        min_distinct_papers: int = 2,
        min_cross_lane_hits: int = 1,
    ) -> None:
        if not passages:
            raise ValueError("At least one passage is required.")
        if candidate_k < 1 or output_k < 1:
            raise ValueError("candidate_k and output_k must be positive.")
        if rrf_k < 0:
            raise ValueError("rrf_k must be non-negative.")

        self.passages = list(passages)
        self.candidate_k = candidate_k
        self.output_k = output_k
        self.rrf_k = rrf_k
        self.min_distinct_papers = min_distinct_papers
        self.min_cross_lane_hits = min_cross_lane_hits

        self._passages_by_id: Dict[str, Passage] = {}
        for passage in self.passages:
            if passage.passage_id in self._passages_by_id:
                raise ValueError(
                    "Duplicate passage_id: {0}".format(passage.passage_id)
                )
            self._passages_by_id[passage.passage_id] = passage

        corpus_tokens = bm25s.tokenize(
            [passage.text for passage in self.passages],
            stopwords="en",
        )
        self._bm25 = bm25s.BM25(
            corpus=[passage.passage_id for passage in self.passages]
        )
        self._bm25.index(corpus_tokens)

    def lexical_search(self, question: str) -> List[Dict[str, object]]:
        query_tokens = bm25s.tokenize([question], stopwords="en")
        limit = min(self.candidate_k, len(self.passages))
        result_ids, scores = self._bm25.retrieve(query_tokens, k=limit)

        ranked: List[Dict[str, object]] = []
        for passage_id, raw_score in zip(result_ids[0], scores[0]):
            score = float(raw_score)
            if score <= 0.0:
                continue
            ranked.append(
                {
                    "passage_id": str(passage_id),
                    "bm25_score": score,
                    "rank": len(ranked) + 1,
                }
            )
        return ranked

    def _normalize_dense_hits(
        self, dense_hits: Sequence[DenseHit]
    ) -> List[DenseHit]:
        normalized: List[DenseHit] = []
        seen = set()
        for hit in dense_hits:
            if hit.passage_id not in self._passages_by_id:
                raise ValueError(
                    "Unknown dense passage_id: {0}".format(hit.passage_id)
                )
            if hit.passage_id in seen:
                continue
            seen.add(hit.passage_id)
            normalized.append(hit)
            if len(normalized) == self.candidate_k:
                break
        return normalized

    def hybrid_search(
        self, question: str, dense_hits: Sequence[DenseHit]
    ) -> List[Dict[str, object]]:
        dense = self._normalize_dense_hits(dense_hits)
        lexical = self.lexical_search(question)

        dense_rank = {
            hit.passage_id: rank
            for rank, hit in enumerate(dense, start=1)
        }
        dense_score = {hit.passage_id: hit.score for hit in dense}
        lexical_rank = {
            str(row["passage_id"]): int(row["rank"])
            for row in lexical
        }
        lexical_score = {
            str(row["passage_id"]): float(row["bm25_score"])
            for row in lexical
        }

        rrf_scores: Dict[str, float] = {}
        for passage_id, rank in dense_rank.items():
            rrf_scores[passage_id] = rrf_scores.get(passage_id, 0.0) + (
                1.0 / (self.rrf_k + rank)
            )
        for passage_id, rank in lexical_rank.items():
            rrf_scores[passage_id] = rrf_scores.get(passage_id, 0.0) + (
                1.0 / (self.rrf_k + rank)
            )

        def sort_key(item: object) -> object:
            passage_id, score = item
            dense_position = dense_rank.get(
                passage_id, self.candidate_k + 1
            )
            lexical_position = lexical_rank.get(
                passage_id, self.candidate_k + 1
            )
            return (-score, min(dense_position, lexical_position), passage_id)

        ordered = sorted(rrf_scores.items(), key=sort_key)
        evidence: List[Dict[str, object]] = []
        for passage_id, rrf_score in ordered[: self.output_k]:
            passage = self._passages_by_id[passage_id]
            evidence.append(
                {
                    "passage_id": passage.passage_id,
                    "paper_id": passage.paper_id,
                    "text": passage.text,
                    "dense_rank": dense_rank.get(passage_id),
                    "dense_score": dense_score.get(passage_id),
                    "bm25_rank": lexical_rank.get(passage_id),
                    "bm25_score": lexical_score.get(passage_id),
                    "rrf_score": rrf_score,
                }
            )
        return evidence

    def evaluate(
        self, question: str, dense_hits: Sequence[DenseHit]
    ) -> Dict[str, object]:
        evidence = self.hybrid_search(question, dense_hits)
        distinct_papers = len(
            {str(row["paper_id"]) for row in evidence}
        )
        cross_lane_hits = sum(
            1
            for row in evidence
            if row["dense_rank"] is not None
            and row["bm25_rank"] is not None
        )

        reasons: List[str] = []
        if distinct_papers < self.min_distinct_papers:
            reasons.append(
                "Fewer than {0} distinct papers support the result.".format(
                    self.min_distinct_papers
                )
            )
        if cross_lane_hits < self.min_cross_lane_hits:
            reasons.append(
                "Fewer than {0} passages appear in both retrieval lanes.".format(
                    self.min_cross_lane_hits
                )
            )

        decision = "abstain" if reasons else "answerable"
        if not reasons:
            reasons.append("The configured evidence baseline is satisfied.")

        return {
            "question": question,
            "decision": decision,
            "thresholds": {
                "min_distinct_papers": self.min_distinct_papers,
                "min_cross_lane_hits": self.min_cross_lane_hits,
            },
            "signals": {
                "distinct_papers": distinct_papers,
                "cross_lane_hits": cross_lane_hits,
                "evidence_count": len(evidence),
            },
            "reasons": reasons,
            "evidence": evidence,
        }

Compare your completed demo.py with this version:

import argparse
import json

from fixtures import DENSE_HITS, QUESTIONS, build_demo_gate


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Inspect lexical, hybrid, and gated retrieval stages."
    )
    parser.add_argument(
        "--scenario",
        choices=["supported", "unsupported"],
        required=True,
    )
    parser.add_argument(
        "--stage",
        choices=["lexical", "hybrid", "gate"],
        required=True,
    )
    args = parser.parse_args()

    gate = build_demo_gate()
    question = QUESTIONS[args.scenario]
    dense_hits = DENSE_HITS[args.scenario]

    if args.stage == "lexical":
        output = {
            "question": question,
            "lexical_results": gate.lexical_search(question),
        }
    elif args.stage == "hybrid":
        output = {
            "question": question,
            "warning": "Fusion ranks candidates but does not prove sufficiency.",
            "evidence": gate.hybrid_search(question, dense_hits),
        }
    else:
        output = gate.evaluate(question, dense_hits)

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

Your retrieval pipeline can now block weak evidence before generation. Next up, you will connect this gate to the existing MedCPT retrieval boundary and prove its behavior with automated tests.

Connect and Test the Retrieval Boundary

Your evidence gate now accepts strong fixture evidence. It also abstains when retrieval support falls below the baseline.

The remaining gap sits at the real retrieval boundary. This step connects rows from your existing MedCPT and Chroma path without taking ownership of model loading or persistent storage.

In this step, get ready to:

Create an adapter for caller-owned corpus rows and dense-result rows.
Place the evidence gate before the unfinished generation function.
Prove the retrieval boundary with four automated tests.

Build the row adapter

An adapter boundary translates dictionaries from the existing retriever into the objects expected by the evidence gate. Caller-supplied field names keep that translation independent from any one corpus schema.

Create integration_example.py beside hybrid_gate.py with the file-creation control in the Visual Studio Code Explorer sidebar.
Add the corpus-row adapter by pasting this code:

from typing import Dict, List, Mapping, Sequence

from hybrid_gate import DenseHit, HybridEvidenceGate, Passage


def build_gate_from_rows(
    corpus_rows: Sequence[Mapping[str, object]],
    passage_id_key: str,
    paper_id_key: str,
    text_key: str,
    min_distinct_papers: int = 2,
    min_cross_lane_hits: int = 1,
) -> HybridEvidenceGate:
    passages: List[Passage] = []
    for row in corpus_rows:
        passages.append(
            Passage(
                passage_id=str(row[passage_id_key]),
                paper_id=str(row[paper_id_key]),
                text=str(row[text_key]),
            )
        )
    return HybridEvidenceGate(
        passages,
        min_distinct_papers=min_distinct_papers,
        min_cross_lane_hits=min_cross_lane_hits,
    )

What does this adapter do?

The caller supplies the field names through passage_id_key. This preserves the stable passage identifier used by both retrieval lanes.
The paper_id_key argument preserves paper-level provenance for the independent-coverage signal.
The text_key argument supplies the passage text that the BM25 lane indexes.
The function builds one HybridEvidenceGate from normalized Passage objects.

Save integration_example.py.
Confirm integration_example.py appears beside hybrid_gate.py in the Explorer sidebar.

Does the adapter file show a problem?

Check that the imports match the names in hybrid_gate.py. Keep every line inside build_gate_from_rows() at the indentation shown above.

Ask for help with the adapter structure:

The dense-row adapter performs the second translation. It converts each dense result into the small input shape already accepted by evaluate().

Append the dense-row adapter below build_gate_from_rows() by pasting this code:

def evaluate_dense_rows(
    gate: HybridEvidenceGate,
    question: str,
    dense_rows: Sequence[Mapping[str, object]],
    passage_id_key: str,
    score_key: str,
) -> Dict[str, object]:
    dense_hits = [
        DenseHit(
            passage_id=str(row[passage_id_key]),
            score=float(row[score_key]),
        )
        for row in dense_rows
    ]
    return gate.evaluate(question, dense_hits)

How does the dense adapter work?

The passage_id_key argument identifies the dense result's stable passage field.
The score_key argument identifies the field containing the dense score.
Each dictionary becomes a DenseHit before the adapter calls gate.evaluate().
The returned dictionary retains the decision signals and provenance-rich evidence from the gate.

Save integration_example.py.
Confirm the file now contains both build_gate_from_rows() and evaluate_dense_rows().

Is the second function nested incorrectly?

Align def evaluate_dense_rows( with def build_gate_from_rows(. The two functions belong at the same indentation level.

Ask for help checking the function boundary:

✔️ Awesome, I've got everything!

Great. Your adapter now handles both corpus rows and dense-result rows.

ⓧ I'd like to double check the full code

Compare your complete integration_example.py file with this reference:

from typing import Dict, List, Mapping, Sequence

from hybrid_gate import DenseHit, HybridEvidenceGate, Passage


def build_gate_from_rows(
    corpus_rows: Sequence[Mapping[str, object]],
    passage_id_key: str,
    paper_id_key: str,
    text_key: str,
    min_distinct_papers: int = 2,
    min_cross_lane_hits: int = 1,
) -> HybridEvidenceGate:
    passages: List[Passage] = []
    for row in corpus_rows:
        passages.append(
            Passage(
                passage_id=str(row[passage_id_key]),
                paper_id=str(row[paper_id_key]),
                text=str(row[text_key]),
            )
        )
    return HybridEvidenceGate(
        passages,
        min_distinct_papers=min_distinct_papers,
        min_cross_lane_hits=min_cross_lane_hits,
    )


def evaluate_dense_rows(
    gate: HybridEvidenceGate,
    question: str,
    dense_rows: Sequence[Mapping[str, object]],
    passage_id_key: str,
    score_key: str,
) -> Dict[str, object]:
    dense_hits = [
        DenseHit(
            passage_id=str(row[passage_id_key]),
            score=float(row[score_key]),
        )
        for row in dense_rows
    ]
    return gate.evaluate(question, dense_hits)

Connect the existing retrieval call site

The adapter lets your existing retrieval path keep ownership of storage and model execution. The evidence gate receives normalized rows after dense retrieval finishes.

Why keep this boundary separate?

Your existing Chroma database remains under the control of the working baseline. The adapter does not open that database or trigger a migration.

MedCPT model loading also stays in the existing retrieval path. This module only evaluates the rows that path already returns.

Switch back to the existing retrieval call site from your working baseline.
Import the adapter helpers from integration_example.py.
Create the gate once after your corpus rows become available by calling build_gate_from_rows().
Match passage_id_key to the passage identifier field in your corpus dictionaries.
Match paper_id_key to the paper identifier field in your corpus dictionaries.
Match text_key to the passage text field in your corpus dictionaries.

Send each MedCPT result list to evaluate_dense_rows() before the unfinished generation function.
Match passage_id_key to the passage identifier field in each dense-result dictionary.
Match score_key to the dense score field in each result dictionary.
Keep generation disabled while you inspect the two evidence reports.

Before you try both questions, which report do you expect to satisfy the evidence baseline?

Run one known supported question through the existing application.
Inspect the returned report for the answerable decision.
Run one deliberately unsupported question through the existing application.
Inspect the returned report for the abstain decision.

The supported report proves that real retriever rows can cross the adapter boundary. The unsupported report proves that weak evidence stops before generation.

Add the automated smoke suite

A smoke test checks the most important behavior through a short automated run. This suite covers acceptance, abstention, cross-lane agreement, and provenance preservation.

Create test_hybrid_gate.py beside integration_example.py with the file-creation control in the Visual Studio Code Explorer sidebar.
Add the shared gate setup and supported-question test by pasting this code:

import unittest

from fixtures import DENSE_HITS, QUESTIONS, build_demo_gate


class HybridEvidenceGateTests(unittest.TestCase):
    def setUp(self) -> None:
        self.gate = build_demo_gate()

    def test_supported_question_is_answerable(self) -> None:
        report = self.gate.evaluate(
            QUESTIONS["supported"], DENSE_HITS["supported"]
        )
        self.assertEqual(report["decision"], "answerable")

What does the first test prove?

The setUp() method creates a fresh fixture gate before each test.
The supported fixture passes through evaluate().
The assertion requires its decision to equal answerable.

Save test_hybrid_gate.py.
Run the current smoke suite with this command:

python -m unittest -v test_hybrid_gate

You should see the supported-question test complete successfully. The run finishes with OK.

Does the first test fail?

Confirm your terminal is still using the local environment from earlier. Check that test_hybrid_gate.py sits beside fixtures.py.

Ask for help with the first test:

Append the unsupported-question decision test inside HybridEvidenceGateTests by pasting this code:

    def test_unsupported_question_abstains(self) -> None:
        report = self.gate.evaluate(
            QUESTIONS["unsupported"], DENSE_HITS["unsupported"]
        )
        self.assertEqual(report["decision"], "abstain")

What does the abstention test prove?

The unsupported fixture still passes through the complete evaluation path. The assertion requires the gate to return abstain.

Save test_hybrid_gate.py.
Run the expanded smoke suite with this command:

python -m unittest -v test_hybrid_gate

You should see both decision tests complete successfully. The run finishes with OK.

Does the abstention test fail?

Check that the test uses the unsupported entries from both fixture dictionaries. Confirm the expected decision is spelled abstain.

Ask for help with the decision assertion:

Append the cross-lane agreement test inside HybridEvidenceGateTests by pasting this code:

    def test_unsupported_question_has_no_cross_lane_agreement(self) -> None:
        report = self.gate.evaluate(
            QUESTIONS["unsupported"], DENSE_HITS["unsupported"]
        )
        self.assertEqual(report["signals"]["cross_lane_hits"], 0)

What does the signal test prove?

This test inspects the observed signal behind the abstention. It requires the unsupported fixture to have zero cross_lane_hits.

Save test_hybrid_gate.py.
Run the expanded smoke suite with this command:

python -m unittest -v test_hybrid_gate

You should see three tests complete successfully. The run finishes with OK.

Does the signal test fail?

Check the nested dictionary keys in the assertion. The path must pass through signals before reading cross_lane_hits.

Ask for help tracing the signal:

Append the provenance test and test runner below the existing tests by pasting this code:

    def test_fused_rows_preserve_provenance(self) -> None:
        evidence = self.gate.hybrid_search(
            QUESTIONS["supported"], DENSE_HITS["supported"]
        )
        self.assertTrue(evidence)
        self.assertIn("passage_id", evidence[0])
        self.assertIn("paper_id", evidence[0])
        self.assertIn("rrf_score", evidence[0])


if __name__ == "__main__":
    unittest.main()

What does the provenance test prove?

The test requires the fused evidence list to contain at least one row.
The passage_id assertion checks passage-level provenance.
The paper_id assertion checks paper-level provenance.
The rrf_score assertion checks that the fused ranking remains auditable.

Save test_hybrid_gate.py.

Before you run the completed suite, do you expect all four retrieval guarantees to hold?

Run the complete smoke suite with this command:

python -m unittest -v test_hybrid_gate

You should see all four tests finish with OK. That is the boundary proven: normalized retrieval rows now reach a tested evidence decision.

Does the complete suite fail?

Use the failing test name to narrow the issue to decision logic, signals, or provenance. Compare only that test with the matching fixture and gate method.

Ask for help diagnosing the suite:

✔️ Awesome, I've got everything!

Excellent. Your four automated checks now protect the retrieval boundary.

ⓧ I'd like to double check the full code

Compare your complete test_hybrid_gate.py file with this reference:

import unittest

from fixtures import DENSE_HITS, QUESTIONS, build_demo_gate


class HybridEvidenceGateTests(unittest.TestCase):
    def setUp(self) -> None:
        self.gate = build_demo_gate()

    def test_supported_question_is_answerable(self) -> None:
        report = self.gate.evaluate(
            QUESTIONS["supported"], DENSE_HITS["supported"]
        )
        self.assertEqual(report["decision"], "answerable")

    def test_unsupported_question_abstains(self) -> None:
        report = self.gate.evaluate(
            QUESTIONS["unsupported"], DENSE_HITS["unsupported"]
        )
        self.assertEqual(report["decision"], "abstain")

    def test_unsupported_question_has_no_cross_lane_agreement(self) -> None:
        report = self.gate.evaluate(
            QUESTIONS["unsupported"], DENSE_HITS["unsupported"]
        )
        self.assertEqual(report["signals"]["cross_lane_hits"], 0)

    def test_fused_rows_preserve_provenance(self) -> None:
        evidence = self.gate.hybrid_search(
            QUESTIONS["supported"], DENSE_HITS["supported"]
        )
        self.assertTrue(evidence)
        self.assertIn("passage_id", evidence[0])
        self.assertIn("paper_id", evidence[0])
        self.assertIn("rrf_score", evidence[0])


if __name__ == "__main__":
    unittest.main()

Secret mission

Sweep the Evidence Thresholds

One threshold pair can hide how sensitive an evidence gate is. This mission sweeps nine settings across supported and unsupported questions. You'll see exactly where each decision changes.

Clean Up Your Resources

Everything in this evidence-gate project runs on your computer, so these local resources have no ongoing cost. Decide whether to keep the module, pause your work, or delete it entirely.

Resources you used:

The hybrid_gate folder inside your existing repository. It contains requirements.txt, hybrid_gate.py, fixtures.py, demo.py, integration_example.py, test_hybrid_gate.py, and sweep.py.
The local virtual environment at hybrid_gate\.venv.

Keep everything running

No action is needed. Choose this option if you plan to keep testing the evidence gate or connect it to more retrieval questions.

Retain the hybrid_gate folder inside your existing repository.
Commit the module's source files with Git.
Leave .venv uncommitted.

Pause - I'll come back to this later

Pausing closes the active terminal session. Your files stay ready for later.

Close the PowerShell terminal when you finish.
Return to the hybrid_gate folder in PowerShell when you resume.
Reactivate the local environment by running this command:

.venv\Scripts\Activate.ps1

What does this command do?

The activation script restores the isolated environment for the new PowerShell session. Python can then use the packages installed in .venv.

Python now uses the packages from .venv for this PowerShell session.

Activation script not running?

Confirm that .venv still exists inside hybrid_gate.
Return to hybrid_gate in PowerShell before running the activation script.
Help me reactivate my local environment.

Delete - I don't want to use this again

Permanent deletion can feel drastic, but this option removes only the isolated hybrid_gate folder. Your original MedCPT and Chroma baseline stay untouched.

Use File Explorer to navigate to the existing repository that contains hybrid_gate.
Select the hybrid_gate folder.
Press Shift+Delete to remove the selected folder permanently.
Confirm the deletion when Windows asks.

You should no longer see the hybrid_gate folder in the repository. Its source files and .venv environment are gone with it.

Nice Work!

You did it! Your hybrid_gate module now blocks unsupported questions with a traceable evidence report.

What you learned:

Built a BM25 lexical lane that returns ranked passage IDs with positive scores.
Combined dense rankings with lexical rankings through Reciprocal Rank Fusion. Each fused result exposes its lane ranks plus scores. It also links each passage to its source paper.
Built a transparent evidence gate from independent-paper coverage plus cross-lane agreement. A version-independent adapter accepts normalized MedCPT rows. It also accepts normalized Chroma rows. Automated smoke tests prove answerability plus abstention behavior.
Secret Mission: Completed a threshold sensitivity sweep across every combination of min_distinct_papers values 1, 2, and 3 with min_cross_lane_hits values 0, 1, and 2. The output exposes how stricter settings change both decisions.

Ready to quiz yourself?