Build a Local AI Code Impact Reviewer

Build a local AI reviewer that traces the impact of Python code changes.

Introduction

30 Second Summary

A small code change can look harmless when you only inspect the edited file. Its effects often surface later in parts of the project you never checked.

In this project, you will build a local AI reviewer that turns a risky Python diff into a structured change-impact report. The reviewer uses static dependency analysis with codebase RAG to find the wider blast radius.

What You'll Build

You will run the reviewer against a seeded discount patch to watch ranked repository evidence become a validated report with a visible benchmark result.

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

  • Trace the wider blast radius by inspecting ranked pricing code, dependent checkout code, relevant tests, plus architecture rules.
  • Generate an evidence-checked report as structured JSON with file references plus line ranges validated against retrieved chunks.
  • Measure retrieval quality through a repeatable benchmark that prints recall scores plus a visible pass or fail result.
  • Secret Mission: Add a second benchmark that catches a breaking rename in the checkout response contract.

Are there any prerequisites?

You should be comfortable reading Python functions plus running commands in Windows PowerShell. Visual Studio Code plus Ollama should already be available locally because Step 1 verifies them before downloading the required models.

Before We Start

This first checkpoint defines the local reviewer you are about to build. It will inspect a unified diff, source code, tests, import relationships, and architecture documentation while evidence validation prevents unsupported AI claims.

Prepare the Local Review Lab

Your reviewer runs entirely on your machine. A dependable Python runtime is required before it can inspect code.

Ollama supplies an embedding model for retrieval. It also supplies a coding model for review generation.

This step prepares an empty Visual Studio Code workspace. You will verify every tool from PowerShell before any reviewer files exist.

In this step, get ready to:
  • Prepare the impact-reviewer workspace in Visual Studio Code.
  • Verify Python from PowerShell.
  • Make the two required local models ready through Ollama.
Create the project workspace and verify Python

A workspace gives Visual Studio Code a single project boundary. Keeping it empty here makes the setup check easy to verify.

  • Press the Windows key to open Windows search.
  • Type Visual Studio Code and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder....
  • Select Desktop as the folder location.
  • Click New Folder.
  • Enter impact-reviewer and press Enter.
  • Click Select Folder.

Why start with an empty workspace?

The fixture repository arrives in the next step. An empty workspace confirms that this step created only the project boundary.

  • Approve the workspace by clicking Yes, I trust the authors if the workspace trust dialog appears.
  • Confirm that the Explorer sidebar shows impact-reviewer at the top.

You should see the empty impact-reviewer folder in the Explorer sidebar. No source files or fixture folders should be listed.

  • Click View in the top menu bar.
  • Click Terminal.
  • Confirm that the terminal tab identifies PowerShell.
  • Verify the Python runtime by running this command:
py -3 --version

What does this command check?

  • The py launcher finds the Python installation registered with Windows.
  • The -3 flag selects Python 3.
  • The version output proves that PowerShell can start the runtime used by the reviewer.
  • Check the printed Python version.

You should see Python 3.14.7 in the terminal. This confirms that the required runtime is available.

Python version missing?

  • Choose PowerShell from the terminal profile menu if another shell is active.
  • Return to the Python installation you confirmed before this project if the Windows launcher is unavailable.
  • Restart Visual Studio Code after correcting the Python installation.

Help me diagnose my Python verification result.

Download and test the embedding model

EmbeddingGemma converts repository text into numeric embeddings. Those vectors later help the reviewer find files whose meaning relates to the code change.

The first run downloads a 622MB model. Expect the command to take a few minutes while Ollama stores it locally.

  • Download the embedding model and test a local response by running this command:
ollama run embeddinggemma "Hello world"

What does this command prove?

  • The ollama run command asks the local Ollama runtime to execute a model.
  • The embeddinggemma identifier selects the model used by the retrieval pipeline.
  • The Hello world input gives the model a small piece of text to embed.
  • Let the model download finish.
  • Check the response after the download completes.

You should see a JSON array containing many decimal numbers. That array confirms that local embedding generation works.

Embedding response missing?

  • Confirm that the Ollama application is running in Windows.
  • Check that your internet connection remains active while the model downloads.
  • Run the embedding command again after an interrupted download.

Help me troubleshoot the local embedding model.

Test the coding model and confirm both downloads

Qwen2.5-Coder 3B turns retrieved evidence into a written review. A short local question confirms that the model can generate a code-focused answer.

The first response can take a few minutes while Ollama downloads the coding model. The terminal stays inside an interactive model session after the download.

  • Start the coding model by running this command:
ollama run qwen2.5-coder:3b

What happens when this runs?

Ollama downloads qwen2.5-coder:3b when the model is absent. It then opens an interactive local session for code questions.

  • Wait until Ollama shows its interactive input prompt.
  • Type a short question about the default return value of a Python function.
  • Press Enter.

You should receive an explanation of the value returned when a Python function has no explicit return statement. That response proves the coding model can answer locally.

  • Click Terminal in the top menu bar.
  • Click New Terminal to open a second PowerShell session.

Before you run the final check, which two model names do you expect the local inventory to contain?

  • Verify the runtime and local model inventory by running these commands:
py -3 --version
ollama ls

What does the final check prove?

  • The py -3 --version command confirms that the Python runtime remains available inside the project workspace.
  • The ollama ls command prints the models stored by your local Ollama installation.

PowerShell should print Python 3.14.7. The model table should include embeddinggemma and qwen2.5-coder:3b.

Your local review lab is ready. The empty workspace is open with both models available for requests.

Model missing from the list?

  • Return to the terminal that tested the missing model.
  • Confirm that its download reached completion.
  • Run the relevant model command again if the earlier download stopped.

Help me find the missing Ollama model.

The local engine room is ready. Next, you will build a naive reviewer and watch it miss the real blast radius of a code change.

Expose the Naive Reviewer's Blind Spot

A diff can name the file someone edited. That narrow view hides callers.

Tests can disappear from the review too. Architecture rules can remain invisible.

In this step, you will create a small Python repository with a seeded unified diff. You will run a changed-file-only reviewer in PowerShell so the missing blast radius becomes visible.

In this step, get ready to:
  • Create a fixture repository containing pricing code, checkout code, tests, documentation, and an unrelated module.
  • Seed a patch that removes the upper discount limit.
  • Run a naive reviewer that reports only the directly changed file.
Build the fixture repository

A fixture repository gives the reviewer a known change with a known blast radius. Its callers, tests, and written rules provide evidence that a narrow scan can miss.

  • In the Visual Studio Code file tree from earlier, create a sample_repo folder inside impact-reviewer.
  • Inside sample_repo, create a src folder.

You should see sample_repo/src in the file tree.

  • Inside sample_repo, create a tests folder.
  • Inside sample_repo, create a docs folder.

The fixture now has separate locations for application code, tests, and architecture rules.

  • Inside sample_repo/src, create pricing.py.
  • Add the pricing boundary by pasting this code into pricing.py:
MAX_DISCOUNT_PERCENT = 30


def calculate_discount(subtotal_cents, discount_percent):
    if subtotal_cents < 0:
        raise ValueError("Subtotal cannot be negative.")
    if discount_percent < 0 or discount_percent > MAX_DISCOUNT_PERCENT:
        raise ValueError("Discount percentage is outside the allowed range.")
    return subtotal_cents * discount_percent // 100

What Does This Fixture Establish?

  • The MAX_DISCOUNT_PERCENT value sets the business limit to 30 percent.
  • The calculate_discount() function rejects negative subtotals.
  • The discount check rejects percentages outside the permitted range.
  • The final calculation returns an integer-cent discount.
  • Save pricing.py.
  • Confirm that MAX_DISCOUNT_PERCENT = 30 appears at the top of the editor.

Pricing File in the Wrong Place?

  • Check that pricing.py sits inside sample_repo/src.
  • Check that the filename ends with .py.

Help me check my pricing fixture path.

  • Inside sample_repo/src, create checkout.py.
  • Add the pricing caller by pasting this code into checkout.py:
from src.pricing import calculate_discount


def build_checkout_summary(subtotal_cents, discount_percent):
    discount_cents = calculate_discount(subtotal_cents, discount_percent)
    return {
        "subtotal_cents": subtotal_cents,
        "discount_cents": discount_cents,
        "total_cents": subtotal_cents - discount_cents,
    }

Why Is Checkout Affected?

  • The import makes checkout.py a direct dependent of pricing.py.
  • The build_checkout_summary() function passes the discount percentage into the changed calculation.
  • The returned dictionary preserves the public checkout response contract.
  • Save checkout.py.
  • Confirm that the first line imports calculate_discount from src.pricing.

Checkout Import Missing?

  • Check that the import begins with from src.pricing.
  • Check that checkout.py sits beside pricing.py.

Help me compare the checkout fixture.

  • Inside sample_repo/src, create notifications.py.
  • Add the unrelated module by pasting this code into notifications.py:
def build_shipping_message(order_number):
    return f"Order {order_number} has shipped."

Why Include an Unrelated Module?

The notification function has no connection to discount calculation. Its presence tests whether later evidence ranking can avoid unrelated code.

  • Save notifications.py.
  • Confirm that build_shipping_message() is the only function in the file.

Notification File Not Listed?

  • Check that notifications.py sits inside sample_repo/src.
  • Check that the filename uses the plural word notifications.

Help me locate the notification fixture.

  • Inside sample_repo/tests, create test_pricing.py.
  • Add the pricing tests by pasting this code into test_pricing.py:
from src.pricing import calculate_discount


def test_discount_is_calculated_in_integer_cents():
    assert calculate_discount(10000, 15) == 1500


def test_discount_above_business_limit_is_rejected():
    try:
        calculate_discount(10000, 80)
    except ValueError:
        return
    raise AssertionError("Expected an excessive discount to be rejected.")

What Do the Pricing Tests Protect?

  • The first test protects integer-cent discount calculation.
  • The second test expects an 80 percent discount to raise ValueError.
  • Removing the upper discount check creates a direct conflict with that second expectation.
  • Save test_pricing.py.
  • Confirm that the file contains two test functions.

Pricing Test Looks Incomplete?

  • Check that the second test calls calculate_discount(10000, 80).
  • Check that the final line raises AssertionError.

Help me compare the pricing tests.

  • Inside sample_repo/tests, create test_checkout.py.
  • Add the checkout contract test by pasting this code into test_checkout.py:
from src.checkout import build_checkout_summary


def test_checkout_summary_keeps_public_money_fields():
    summary = build_checkout_summary(10000, 10)
    assert summary == {
        "subtotal_cents": 10000,
        "discount_cents": 1000,
        "total_cents": 9000,
    }

What Does the Checkout Test Protect?

This test protects the public money fields returned by build_checkout_summary(). A pricing change can affect the values delivered through that contract.

  • Save test_checkout.py.
  • Confirm that the expected summary contains total_cents.

Checkout Contract Does Not Match?

  • Check that the test imports build_checkout_summary.
  • Check that the expected total_cents value is 9000.

Help me compare the checkout test.

  • Inside sample_repo/docs, create architecture.md.
  • Add the repository rules by pasting this content into architecture.md:
# Checkout Architecture Rules

## Pricing boundary

Discount percentages must remain between 0 and 30. The pricing module owns this validation so every caller receives the same protection.

Any change to discount validation requires running `tests/test_pricing.py` and reviewing callers of `src.pricing`.

## Public checkout contract

Checkout summaries expose money as integer cents. The public response fields are `subtotal_cents`, `discount_cents`, and `total_cents`.

Renaming or removing a public response field is a compatibility change and requires the checkout contract test.

Why Does Documentation Count as Evidence?

The architecture file records rules that imports cannot express. It links discount validation to a required test plus a caller review.

It also identifies the checkout fields as a public contract. These written constraints belong in a complete impact review.

  • Save architecture.md.
  • Confirm that the file contains the Pricing boundary heading.

Architecture Rules Missing?

  • Check that architecture.md sits inside sample_repo/docs.
  • Check that the file retains the .md extension.

Help me verify the architecture fixture.

✔️ Awesome, I've got everything!

Your fixture repository now contains directly changed code, dependent code, tests, documentation, and an unrelated module.

  • Confirm that src contains three Python files.
  • Confirm that tests contains two Python files.
  • Confirm that docs contains architecture.md.

ⓧ I'd like to double check the full code

MAX_DISCOUNT_PERCENT = 30


def calculate_discount(subtotal_cents, discount_percent):
    if subtotal_cents < 0:
        raise ValueError("Subtotal cannot be negative.")
    if discount_percent < 0 or discount_percent > MAX_DISCOUNT_PERCENT:
        raise ValueError("Discount percentage is outside the allowed range.")
    return subtotal_cents * discount_percent // 100

This reference shows the complete pricing boundary fixture.

from src.pricing import calculate_discount


def build_checkout_summary(subtotal_cents, discount_percent):
    discount_cents = calculate_discount(subtotal_cents, discount_percent)
    return {
        "subtotal_cents": subtotal_cents,
        "discount_cents": discount_cents,
        "total_cents": subtotal_cents - discount_cents,
    }

This reference shows the complete checkout caller fixture.

def build_shipping_message(order_number):
    return f"Order {order_number} has shipped."

This reference shows the complete unrelated notification fixture.

from src.pricing import calculate_discount


def test_discount_is_calculated_in_integer_cents():
    assert calculate_discount(10000, 15) == 1500


def test_discount_above_business_limit_is_rejected():
    try:
        calculate_discount(10000, 80)
    except ValueError:
        return
    raise AssertionError("Expected an excessive discount to be rejected.")

This reference shows the complete pricing test fixture.

from src.checkout import build_checkout_summary


def test_checkout_summary_keeps_public_money_fields():
    summary = build_checkout_summary(10000, 10)
    assert summary == {
        "subtotal_cents": 10000,
        "discount_cents": 1000,
        "total_cents": 9000,
    }

This reference shows the complete checkout contract test.

# Checkout Architecture Rules

## Pricing boundary

Discount percentages must remain between 0 and 30. The pricing module owns this validation so every caller receives the same protection.

Any change to discount validation requires running `tests/test_pricing.py` and reviewing callers of `src.pricing`.

## Public checkout contract

Checkout summaries expose money as integer cents. The public response fields are `subtotal_cents`, `discount_cents`, and `total_cents`.

Renaming or removing a public response field is a compatibility change and requires the checkout contract test.

This reference shows the complete architecture rules.

Seed the risky discount patch

The repository holds the protected version of the pricing function. The patch represents a proposed change that removes its upper discount limit.

  • In the impact-reviewer file tree, create a fixtures folder.
  • Inside fixtures, create discount_cap.patch.

You should see fixtures/discount_cap.patch beside the sample_repo folder.

  • Add the seeded discount regression by pasting this diff into discount_cap.patch:
diff --git a/src/pricing.py b/src/pricing.py
index 1111111..2222222 100644
--- a/src/pricing.py
+++ b/src/pricing.py
@@ -4,7 +4,7 @@ MAX_DISCOUNT_PERCENT = 30
 def calculate_discount(subtotal_cents, discount_percent):
     if subtotal_cents < 0:
         raise ValueError("Subtotal cannot be negative.")
-    if discount_percent < 0 or discount_percent > MAX_DISCOUNT_PERCENT:
+    if discount_percent < 0:
         raise ValueError("Discount percentage is outside the allowed range.")
     return subtotal_cents * discount_percent // 100

How Does This Diff Describe the Change?

  • The +++ b/src/pricing.py header identifies the changed repository path.
  • The line beginning with - records the original validation.
  • The line beginning with + records the proposed validation.
  • The proposal still rejects negative percentages. Discounts above MAX_DISCOUNT_PERCENT now pass through.
  • Save discount_cap.patch.
  • Confirm that the added line contains only if discount_percent < 0:.

Patch Header Not Recognizable?

  • Check that the changed-file header begins with +++ b/.
  • Check that the path after the header is src/pricing.py.
  • Check that the filename is discount_cap.patch.

Help me inspect the seeded diff.

Create and run the naive reviewer

The first reviewer reads only changed-path headers. It prints a JSON result without inspecting repository relationships.

  • In the impact-reviewer file tree, create reviewer.py.

You should see an empty reviewer.py editor tab.

  • Add the imports plus changed-path parser by pasting this code into reviewer.py:
import argparse
import json
from pathlib import Path


def parse_changed_paths(diff_text):
    return sorted(
        {
            line[6:].strip()
            for line in diff_text.splitlines()
            if line.startswith("+++ b/")
        }
    )

What Does the Path Parser Do?

  • The imports provide command parsing, JSON rendering, and file-path handling.
  • The parse_changed_paths() function splits the diff into lines.
  • The prefix check keeps only new-file headers beginning with +++ b/.
  • The slice removes the six-character header prefix from each matching line.
  • Save reviewer.py.
  • Confirm that parse_changed_paths appears below the three imports.

Path Parser Indentation Looks Wrong?

  • Check that the set comprehension remains inside sorted().
  • Check that the prefix contains three plus signs followed by b/.

Help me debug the changed-path parser.

  • Below parse_changed_paths(), add the naive report function by pasting this code:
def naive_report(diff_text):
    return {
        "changed_files": parse_changed_paths(diff_text),
        "limitation": "Only directly changed files are visible.",
    }

What Does the Naive Report Contain?

  • The changed_files field contains paths extracted from the diff headers.
  • The limitation field states the boundary of this review method.
  • Save reviewer.py.
  • Confirm that naive_report returns both dictionary fields.

Naive Report Braces Do Not Match?

  • Check that both fields sit inside one dictionary.
  • Check that the closing brace aligns with return.

Help me debug the naive report.

  • Below naive_report(), add the command-line parser by pasting this code:
def build_parser():
    parser = argparse.ArgumentParser(
        description="Review a unified diff against local repository evidence."
    )
    parser.add_argument("--repo", type=Path, required=True)
    parser.add_argument("--diff", type=Path, required=True)
    parser.add_argument(
        "--mode",
        choices=["naive", "retrieve", "review"],
        default="review",
    )
    return parser

What Does the Command Parser Capture?

  • The --repo argument captures the fixture repository path.
  • The --diff argument captures the patch path.
  • The --mode argument selects the reviewer path.
  • This step explicitly uses the naive choice.
  • Save reviewer.py.
  • Confirm that all three arguments begin with two hyphens.

Argument Parser Not Returning?

  • Check that return parser remains inside build_parser().
  • Check that both path arguments use type=Path.

Help me debug the command parser.

  • Below build_parser(), add the naive execution path by pasting this code:
def main():
    args = build_parser().parse_args()
    diff_text = args.diff.read_text(encoding="utf-8")

    if args.mode == "naive":
        result = naive_report(diff_text)

    rendered = json.dumps(result, indent=2)
    print(rendered)


if __name__ == "__main__":
    main()

How Does the Naive Execution Path Work?

  • The args value stores the parsed command options.
  • The diff_text value stores the patch contents.
  • The mode check sends the diff to naive_report().
  • The final lines format the result as indented JSON.
  • Save reviewer.py.
  • Confirm that the final line calls main().

Main Function Looks Incomplete?

  • Check that rendered is created after the mode check.
  • Check that the final condition uses __name__ plus __main__.

Help me compare the execution path.

✔️ Awesome, I've got everything!

Your naive reviewer can now parse changed-file headers and render its limited result.

  • Confirm that reviewer.py is saved before running it.

ⓧ I'd like to double check the full code

import argparse
import json
from pathlib import Path


def parse_changed_paths(diff_text):
    return sorted(
        {
            line[6:].strip()
            for line in diff_text.splitlines()
            if line.startswith("+++ b/")
        }
    )


def naive_report(diff_text):
    return {
        "changed_files": parse_changed_paths(diff_text),
        "limitation": "Only directly changed files are visible.",
    }


def build_parser():
    parser = argparse.ArgumentParser(
        description="Review a unified diff against local repository evidence."
    )
    parser.add_argument("--repo", type=Path, required=True)
    parser.add_argument("--diff", type=Path, required=True)
    parser.add_argument(
        "--mode",
        choices=["naive", "retrieve", "review"],
        default="review",
    )
    return parser


def main():
    args = build_parser().parse_args()
    diff_text = args.diff.read_text(encoding="utf-8")

    if args.mode == "naive":
        result = naive_report(diff_text)

    rendered = json.dumps(result, indent=2)
    print(rendered)


if __name__ == "__main__":
    main()

This reference matches the complete naive reviewer.py file for this step.

  • Switch back to the PowerShell terminal from earlier.

Before you run the reviewer, consider whether changed-file headers can reveal the checkout caller, pricing tests, or architecture rules.

  • Run the naive review from the impact-reviewer folder with this command:
py reviewer.py --repo sample_repo --diff fixtures/discount_cap.patch --mode naive

What Does This Command Check?

  • The --repo sample_repo argument points at the fixture repository.
  • The --diff fixtures/discount_cap.patch argument points at the seeded change.
  • The --mode naive argument limits analysis to changed-file headers.

You will see src/pricing.py as the only changed-file entry. You will also see Only directly changed files are visible. as the limitation.

That shortfall is intentional. The output misses src/checkout.py, both test files, and docs/architecture.md even though they contain relevant impact evidence.

Naive Review Did Not Run?

  • Check that PowerShell is currently inside the impact-reviewer folder.
  • Check that reviewer.py sits beside sample_repo and fixtures.
  • Check that the patch path ends with fixtures/discount_cap.patch.

Help me debug the naive reviewer command.

You have captured the reviewer's blind spot with a repeatable fixture. Next, you will make related code, tests, and architecture rules visible to the review process.

Build Hybrid Impact Retrieval

Your naive reviewer proved that a changed-file list can hide the true blast radius of a patch. It missed callers, tests, and architecture rules that could expose the removed discount limit.

This step builds a hybrid retrieval pipeline that follows Python imports. It also compares the diff with repository evidence through lexical signals and local semantic similarity.

In this step, get ready to:
  • Split repository files into line-aware evidence chunks.
  • Discover Python files that import the changed module.
  • Rank source code, tests, and documentation with hybrid retrieval.
Split repository files into evidence chunks

A retrieval system needs evidence units that are small enough to rank. Each chunk keeps its source path and line range so later reports can point back to concrete repository evidence.

  • In the open reviewer.py file, replace the import block with the code below:
import argparse
import ast
import json
from pathlib import Path
from urllib.request import Request, urlopen

What Do These Imports Add?

  • The ast module creates an abstract syntax tree from each Python file so the reviewer can inspect imports.
  • The Request and urlopen helpers let the reviewer call the local Ollama API without another package.
  • Add the local model settings and JSON request helper directly below the imports by copying this code:
OLLAMA_BASE_URL = "http://localhost:11434"
EMBED_MODEL = "embeddinggemma"
REVIEW_MODEL = "qwen2.5-coder:3b"
ALLOWED_SUFFIXES = {".py", ".md", ".txt"}
EXCLUDED_DIRS = {".git", ".venv", "__pycache__", "reports"}


def post_json(endpoint, payload):
    body = json.dumps(payload).encode("utf-8")
    request = Request(
        f"{OLLAMA_BASE_URL}{endpoint}",
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=180) as response:
        return json.loads(response.read().decode("utf-8"))

What Does This Helper Do?

  • The constants select the local API address, embedding model, review model, supported file types, and excluded folders.
  • The post_json() helper converts a Python payload into JSON before sending it to a local endpoint.
  • The response is decoded back into Python data so retrieval functions can read returned embeddings.
  • Save reviewer.py.
  • Confirm that the existing naive mode still runs by using this command:
py reviewer.py --repo sample_repo --diff fixtures/discount_cap.patch --mode naive

PowerShell still prints src/pricing.py with the changed-file-only limitation. This confirms that the new imports and HTTP helper did not break the working mode.

Does the Naive Mode Fail Now?

Check that the five constants sit below the complete import block. Confirm that every quote and parenthesis in post_json() matches the code above.

Use this prompt if PowerShell reports a syntax or import problem: Help me debug the new imports and post_json function in reviewer.py.

  • Add collect_chunks() directly above parse_changed_paths() by copying this code:
def collect_chunks(repo_root, lines_per_chunk=40, overlap=8):
    chunks = []
    stride = lines_per_chunk - overlap

    for path in sorted(repo_root.rglob("*")):
        if not path.is_file() or path.suffix.lower() not in ALLOWED_SUFFIXES:
            continue

        relative = path.relative_to(repo_root)
        if any(part in EXCLUDED_DIRS for part in relative.parts):
            continue

        lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
        if not lines:
            lines = [""]

        for start in range(0, len(lines), stride):
            end = min(start + lines_per_chunk, len(lines))
            chunks.append(
                {
                    "path": relative.as_posix(),
                    "start_line": start + 1,
                    "end_line": end,
                    "text": "\n".join(lines[start:end]),
                }
            )
            if end == len(lines):
                break

    return chunks

How Does Chunking Preserve Evidence?

  • The file scan keeps Python, Markdown, and text files that can contain implementation details or architecture rules.
  • Each chunk stores a relative path with inclusive start and end line numbers.
  • The overlap repeats eight lines between neighbouring chunks so relevant context near a boundary remains available.
  • Save reviewer.py.
  • Check the updated file for syntax errors by running the working naive command again:
py reviewer.py --repo sample_repo --diff fixtures/discount_cap.patch --mode naive

You should see the same naive JSON result. The successful run proves that collect_chunks() is now part of a valid Python file.

Seeing a Chunking Syntax Error?

Check the indentation beneath both for loops. Make sure return chunks lines up with the first statement inside the function.

Ask for targeted help here: Help me fix the indentation in collect_chunks without changing its chunking logic.

Trace imports and score text signals

Python imports provide a deterministic connection between changed code and direct dependents. This catches src/checkout.py and tests/test_pricing.py because both import the changed pricing module.

  • Add the path conversion helper and dependency-function foundation below collect_chunks() by copying this code:
def module_name_from_path(path):
    if not path.endswith(".py"):
        return ""
    return path[:-3].replace("/", ".")


def build_dependency_targets(repo_root, changed_paths):
    changed_modules = {
        module_name_from_path(path)
        for path in changed_paths
        if module_name_from_path(path)
    }
    dependents = set()

    return sorted(dependents)

Why Convert Paths to Modules?

The diff names src/pricing.py as a path. Python import statements refer to the same file as src.pricing.

The conversion gives the dependency scan a common value to compare. The set also removes duplicate dependents before sorting them.

  • Insert the following traversal inside build_dependency_targets() immediately before return sorted(dependents):
    for path in sorted(repo_root.rglob("*.py")):
        relative = path.relative_to(repo_root).as_posix()
        try:
            tree = ast.parse(
                path.read_text(encoding="utf-8", errors="replace"),
                filename=relative,
            )
        except SyntaxError:
            continue

        imported_modules = set()
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                imported_modules.update(alias.name for alias in node.names)
            elif isinstance(node, ast.ImportFrom) and node.module:
                imported_modules.add(node.module)

        if any(
            imported == changed or imported.startswith(f"{changed}.")
            for imported in imported_modules
            for changed in changed_modules
        ):
            dependents.add(relative)

How Does the Import Scan Work?

  • The scan parses every Python file into a tree of language constructs.
  • The ast.walk() traversal collects modules from regular imports and from-import statements.
  • A file becomes a dependency target when one of its imported modules matches a changed module.
  • Files with invalid Python syntax are skipped so one broken file does not stop repository retrieval.
  • Add tokenize() below build_dependency_targets() by copying this code:
def tokenize(text):
    normalized = "".join(
        character.lower()
        if character.isalnum() or character == "_"
        else " "
        for character in text
    )
    return {token for token in normalized.split() if len(token) >= 3}

What Does Tokenization Contribute?

Tokenization turns the diff and each chunk into sets of lowercase words. Their overlap rewards evidence that repeats concrete names from the patch.

The three-character minimum removes short fragments that add noise. Lexical matching gives exact identifiers a direct influence on ranking.

  • Add cosine_similarity() immediately below tokenize() by copying this code:
def cosine_similarity(left, right):
    dot_product = sum(a * b for a, b in zip(left, right))
    left_norm = sum(value * value for value in left) ** 0.5
    right_norm = sum(value * value for value in right) ** 0.5
    if left_norm == 0 or right_norm == 0:
        return 0.0
    return dot_product / (left_norm * right_norm)

Why Use Cosine Similarity?

Cosine similarity measures how closely two vectors point in the same direction. Here it compares the patch embedding with every evidence-chunk embedding.

The zero-length guard prevents division by zero. Higher similarity gives semantically related evidence a stronger score.

  • Save reviewer.py.
  • Confirm that the new analysis helpers leave the existing mode intact by running:
py reviewer.py --repo sample_repo --diff fixtures/discount_cap.patch --mode naive

The naive result should still print successfully. Your reviewer now contains valid chunking, dependency, tokenization, and vector-scoring helpers.

Did an Analysis Helper Break the File?

Check that the dependency traversal sits before the function's existing return line. Confirm that tokenize() and cosine_similarity() begin at the left edge of the file.

Get focused help with the failing helper: Help me diagnose a syntax error in my dependency, tokenization, or cosine similarity functions.

Rank hybrid evidence and run retrieval

Static imports give strong evidence for direct Python dependents. Semantic retrieval can surface architecture documentation that has no import edge, while lexical overlap protects exact terms from being diluted.

Why Hybrid Retrieval?

Import analysis alone cannot connect a Python patch to a Markdown architecture rule. Semantic search alone can overlook exact module relationships.

The hybrid score combines both signals with explicit boosts for changed files and direct dependents. Each method covers a gap left by the others.

  • Add the retrieval-function foundation below cosine_similarity() by copying this code:
def retrieve(repo_root, diff_text, chunks, top_k=4):
    changed_paths = set(parse_changed_paths(diff_text))
    dependency_paths = set(build_dependency_targets(repo_root, changed_paths))
    embedding_inputs = [diff_text] + [
        f"File: {chunk['path']}\n{chunk['text']}" for chunk in chunks
    ]

    response = post_json(
        "/api/embed",
        {
            "model": EMBED_MODEL,
            "input": embedding_inputs,
            "truncate": True,
        },
    )
    vectors = response["embeddings"]
    if len(vectors) != len(embedding_inputs):
        raise RuntimeError("Embedding count did not match input count.")

    query_vector = vectors[0]
    query_tokens = tokenize(diff_text)
    ranked = []

    return sorted(ranked, key=lambda item: item["score"], reverse=True)[:top_k]

How Are Embeddings Prepared?

  • The first embedding input is the diff that acts as the retrieval query.
  • Every remaining input combines a repository path with one line-aware chunk.
  • The local /api/embed request batches all inputs through embeddinggemma.
  • The count check catches any response that does not contain one vector per input.
  • Insert the scoring loop immediately before the return line in retrieve():
    for chunk, vector in zip(chunks, vectors[1:]):
        chunk_tokens = tokenize(chunk["text"])
        lexical = len(query_tokens & chunk_tokens) / max(1, len(query_tokens))
        semantic = cosine_similarity(query_vector, vector)
        changed_boost = 0.30 if chunk["path"] in changed_paths else 0.0
        dependency_boost = 0.20 if chunk["path"] in dependency_paths else 0.0
        score = 0.70 * semantic + 0.30 * lexical + changed_boost + dependency_boost
        ranked.append({**chunk, "score": round(score, 4)})

How Is the Hybrid Score Built?

  • Semantic similarity contributes 70 percent of the base score.
  • Lexical overlap contributes 30 percent of the base score.
  • A changed file receives a 0.30 boost.
  • A direct dependency receives a 0.20 boost.
  • Save reviewer.py.
  • Confirm that the new retrieval function parses by running the naive mode once more:
py reviewer.py --repo sample_repo --diff fixtures/discount_cap.patch --mode naive

PowerShell should print the familiar naive result. The retrieval function is now valid without changing the earlier demonstration.

  • Add format_context() directly below retrieve() by copying this code:
def format_context(ranked_chunks):
    sections = []
    for chunk in ranked_chunks:
        sections.append(
            "\n".join(
                [
                    f"FILE: {chunk['path']}",
                    f"LINES: {chunk['start_line']}-{chunk['end_line']}",
                    chunk["text"],
                ]
            )
        )
    return "\n\n---\n\n".join(sections)

Why Format Retrieved Context?

The formatter labels every evidence section with its file and line range. It separates chunks with a visible divider before a later step sends them to the review model.

This structure preserves the location information needed for evidence validation. Grounded reports depend on traceable source boundaries.

  • Replace the existing build_parser() function with this expanded version:
def build_parser():
    parser = argparse.ArgumentParser(
        description="Review a unified diff against local repository evidence."
    )
    parser.add_argument("--repo", type=Path, required=True)
    parser.add_argument("--diff", type=Path, required=True)
    parser.add_argument(
        "--mode",
        choices=["naive", "retrieve", "review"],
        default="review",
    )
    parser.add_argument("--top-k", type=int, default=4)
    parser.add_argument("--output", type=Path)
    return parser

What Changes in the CLI?

The mode selector now recognises the working naive and retrieve paths. It also reserves review for the evidence-checked generation added in the next step.

The top_k value controls how many ranked chunks survive. The output path prepares the CLI to save the later report.

  • Save reviewer.py.
  • Verify that the expanded parser still accepts the naive command by running:
py reviewer.py --repo sample_repo --diff fixtures/discount_cap.patch --mode naive

The naive JSON should still print in PowerShell. This confirms that the expanded command-line choices are valid.

  • Replace the current main() function and everything below it with this first part:
def main():
    args = build_parser().parse_args()
    diff_text = args.diff.read_text(encoding="utf-8")

    if args.mode == "naive":
        result = naive_report(diff_text)
    else:
        chunks = collect_chunks(args.repo)
        ranked = retrieve(args.repo, diff_text, chunks, args.top_k)
        dependencies = build_dependency_targets(
            args.repo, parse_changed_paths(diff_text)
        )

        if args.mode == "retrieve":
            result = {
                "changed_files": parse_changed_paths(diff_text),
                "dependency_targets": dependencies,
                "retrieved": [
                    {
                        "path": chunk["path"],
                        "lines": f"{chunk['start_line']}-{chunk['end_line']}",
                        "score": chunk["score"],
                    }
                    for chunk in ranked
                ],
            }
        else:
            result = generate_review(diff_text, ranked, dependencies)
            result["changed_files"] = parse_changed_paths(diff_text)
            result["dependency_targets"] = dependencies

How Does the Main Flow Branch?

Naive mode keeps its changed-file-only result. Every other mode collects chunks, ranks evidence, and calculates direct dependency targets.

Retrieve mode converts the ranked chunks into compact path, line, and score records. The final branch is ready for the review generation added next.

  • Continue the same main() function by placing this second part directly below the previous code:
            result["retrieval_context"] = [
                {
                    "path": chunk["path"],
                    "start_line": chunk["start_line"],
                    "end_line": chunk["end_line"],
                    "score": chunk["score"],
                }
                for chunk in ranked
            ]
            result["evidence_validation"] = validate_evidence(result, ranked)

    rendered = json.dumps(result, indent=2)
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(rendered + "\n", encoding="utf-8")
        print(f"Wrote {args.output}")
    else:
        print(rendered)


if __name__ == "__main__":
    main()

How Is the Result Rendered?

The result becomes indented JSON for readable terminal output. An optional output path creates its parent folders before saving the same JSON.

The final guard runs main() only when PowerShell executes reviewer.py directly. Imported reviewer functions remain available for the benchmark built later.

Before you run retrieval, which files do you expect import analysis and semantic matching to add beyond src/pricing.py?

  • Save reviewer.py.
  • Run the hybrid retrieval check from the existing PowerShell terminal with this command:
py reviewer.py --repo sample_repo --diff fixtures/discount_cap.patch --mode retrieve --top-k 4

You should see ranked evidence containing src/pricing.py, at least one importing component or test, and docs/architecture.md. The result now reaches beyond the single file exposed by naive mode.

Missing Expected Retrieval Evidence?

  • Confirm that Ollama remains available with the downloaded embeddinggemma model.
  • Check that the dependency traversal is inside build_dependency_targets() before its return line.
  • Check that the scoring loop is inside retrieve() before its return line.

Use this prompt to inspect a weak or failing result: Help me debug why my hybrid retrieval output is missing expected repository files.

✔️ Awesome, I've got everything!

Great work. Your reviewer now retrieves ranked evidence from code, tests, and architecture documentation.

ⓧ I'd like to double check the full code

import argparse
import ast
import json
from pathlib import Path
from urllib.request import Request, urlopen

OLLAMA_BASE_URL = "http://localhost:11434"
EMBED_MODEL = "embeddinggemma"
REVIEW_MODEL = "qwen2.5-coder:3b"
ALLOWED_SUFFIXES = {".py", ".md", ".txt"}
EXCLUDED_DIRS = {".git", ".venv", "__pycache__", "reports"}


def post_json(endpoint, payload):
    body = json.dumps(payload).encode("utf-8")
    request = Request(
        f"{OLLAMA_BASE_URL}{endpoint}",
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=180) as response:
        return json.loads(response.read().decode("utf-8"))


def parse_changed_paths(diff_text):
    return sorted(
        {
            line[6:].strip()
            for line in diff_text.splitlines()
            if line.startswith("+++ b/")
        }
    )


def naive_report(diff_text):
    return {
        "changed_files": parse_changed_paths(diff_text),
        "limitation": "Only directly changed files are visible.",
    }


def collect_chunks(repo_root, lines_per_chunk=40, overlap=8):
    chunks = []
    stride = lines_per_chunk - overlap

    for path in sorted(repo_root.rglob("*")):
        if not path.is_file() or path.suffix.lower() not in ALLOWED_SUFFIXES:
            continue

        relative = path.relative_to(repo_root)
        if any(part in EXCLUDED_DIRS for part in relative.parts):
            continue

        lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
        if not lines:
            lines = [""]

        for start in range(0, len(lines), stride):
            end = min(start + lines_per_chunk, len(lines))
            chunks.append(
                {
                    "path": relative.as_posix(),
                    "start_line": start + 1,
                    "end_line": end,
                    "text": "\n".join(lines[start:end]),
                }
            )
            if end == len(lines):
                break

    return chunks


def module_name_from_path(path):
    if not path.endswith(".py"):
        return ""
    return path[:-3].replace("/", ".")


def build_dependency_targets(repo_root, changed_paths):
    changed_modules = {
        module_name_from_path(path)
        for path in changed_paths
        if module_name_from_path(path)
    }
    dependents = set()

    for path in sorted(repo_root.rglob("*.py")):
        relative = path.relative_to(repo_root).as_posix()
        try:
            tree = ast.parse(
                path.read_text(encoding="utf-8", errors="replace"),
                filename=relative,
            )
        except SyntaxError:
            continue

        imported_modules = set()
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                imported_modules.update(alias.name for alias in node.names)
            elif isinstance(node, ast.ImportFrom) and node.module:
                imported_modules.add(node.module)

        if any(
            imported == changed or imported.startswith(f"{changed}.")
            for imported in imported_modules
            for changed in changed_modules
        ):
            dependents.add(relative)

    return sorted(dependents)


def tokenize(text):
    normalized = "".join(
        character.lower()
        if character.isalnum() or character == "_"
        else " "
        for character in text
    )
    return {token for token in normalized.split() if len(token) >= 3}


def cosine_similarity(left, right):
    dot_product = sum(a * b for a, b in zip(left, right))
    left_norm = sum(value * value for value in left) ** 0.5
    right_norm = sum(value * value for value in right) ** 0.5
    if left_norm == 0 or right_norm == 0:
        return 0.0
    return dot_product / (left_norm * right_norm)


def retrieve(repo_root, diff_text, chunks, top_k=4):
    changed_paths = set(parse_changed_paths(diff_text))
    dependency_paths = set(build_dependency_targets(repo_root, changed_paths))
    embedding_inputs = [diff_text] + [
        f"File: {chunk['path']}\n{chunk['text']}" for chunk in chunks
    ]

    response = post_json(
        "/api/embed",
        {
            "model": EMBED_MODEL,
            "input": embedding_inputs,
            "truncate": True,
        },
    )
    vectors = response["embeddings"]
    if len(vectors) != len(embedding_inputs):
        raise RuntimeError("Embedding count did not match input count.")

    query_vector = vectors[0]
    query_tokens = tokenize(diff_text)
    ranked = []

    for chunk, vector in zip(chunks, vectors[1:]):
        chunk_tokens = tokenize(chunk["text"])
        lexical = len(query_tokens & chunk_tokens) / max(1, len(query_tokens))
        semantic = cosine_similarity(query_vector, vector)
        changed_boost = 0.30 if chunk["path"] in changed_paths else 0.0
        dependency_boost = 0.20 if chunk["path"] in dependency_paths else 0.0
        score = 0.70 * semantic + 0.30 * lexical + changed_boost + dependency_boost
        ranked.append({**chunk, "score": round(score, 4)})

    return sorted(ranked, key=lambda item: item["score"], reverse=True)[:top_k]


def format_context(ranked_chunks):
    sections = []
    for chunk in ranked_chunks:
        sections.append(
            "\n".join(
                [
                    f"FILE: {chunk['path']}",
                    f"LINES: {chunk['start_line']}-{chunk['end_line']}",
                    chunk["text"],
                ]
            )
        )
    return "\n\n---\n\n".join(sections)


def build_parser():
    parser = argparse.ArgumentParser(
        description="Review a unified diff against local repository evidence."
    )
    parser.add_argument("--repo", type=Path, required=True)
    parser.add_argument("--diff", type=Path, required=True)
    parser.add_argument(
        "--mode",
        choices=["naive", "retrieve", "review"],
        default="review",
    )
    parser.add_argument("--top-k", type=int, default=4)
    parser.add_argument("--output", type=Path)
    return parser


def main():
    args = build_parser().parse_args()
    diff_text = args.diff.read_text(encoding="utf-8")

    if args.mode == "naive":
        result = naive_report(diff_text)
    else:
        chunks = collect_chunks(args.repo)
        ranked = retrieve(args.repo, diff_text, chunks, args.top_k)
        dependencies = build_dependency_targets(
            args.repo, parse_changed_paths(diff_text)
        )

        if args.mode == "retrieve":
            result = {
                "changed_files": parse_changed_paths(diff_text),
                "dependency_targets": dependencies,
                "retrieved": [
                    {
                        "path": chunk["path"],
                        "lines": f"{chunk['start_line']}-{chunk['end_line']}",
                        "score": chunk["score"],
                    }
                    for chunk in ranked
                ],
            }
        else:
            result = generate_review(diff_text, ranked, dependencies)
            result["changed_files"] = parse_changed_paths(diff_text)
            result["dependency_targets"] = dependencies
            result["retrieval_context"] = [
                {
                    "path": chunk["path"],
                    "start_line": chunk["start_line"],
                    "end_line": chunk["end_line"],
                    "score": chunk["score"],
                }
                for chunk in ranked
            ]
            result["evidence_validation"] = validate_evidence(result, ranked)

    rendered = json.dumps(result, indent=2)
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(rendered + "\n", encoding="utf-8")
        print(f"Wrote {args.output}")
    else:
        print(rendered)


if __name__ == "__main__":
    main()

Compare this reference with your saved reviewer.py. The retrieval command should run even though the reserved review branch is completed in the next step.

That is the blind spot closed. Next, you will turn this ranked evidence into a structured review and reject unsupported citations.

Generate an Evidence-Checked Review

Your hybrid retriever now surfaces evidence beyond the changed file. It finds direct Python dependents.

A useful evidence list still does not give a developer a stable decision. This step uses JSON Schema with Ollama to generate a fixed report.

Evidence validation checks every cited path against the retrieved chunks. It checks each line range too.

In this step, get ready to:
  • Define the required sections for every impact report.
  • Send the diff plus retrieved evidence to the local coding model.
  • Validate report citations before saving the JSON result.
Add the structured review pipeline

The report contract requires the model to cover impact, architecture, testing, risk, evidence, confidence, severity, plus unresolved questions. A consistent shape makes every review easier to inspect.

  • In the reviewer.py tab from earlier, compare your file with the full reference in the second tab below.
  • Add every missing section so your file matches the reference exactly.

✔️ Awesome, I've got everything!

Great. Double-check that your changes in reviewer.py are saved.

ⓧ I'd like to double check the full code

Compare your complete reviewer.py file with this reference.

import argparse
import ast
import json
from pathlib import Path
from urllib.request import Request, urlopen

OLLAMA_BASE_URL = "http://localhost:11434"
EMBED_MODEL = "embeddinggemma"
REVIEW_MODEL = "qwen2.5-coder:3b"
ALLOWED_SUFFIXES = {".py", ".md", ".txt"}
EXCLUDED_DIRS = {".git", ".venv", "__pycache__", "reports"}

REPORT_SCHEMA = {
    "type": "object",
    "properties": {
        "summary": {"type": "string"},
        "affected_components": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "impact": {"type": "string"},
                    "confidence": {
                        "type": "string",
                        "enum": ["low", "medium", "high"],
                    },
                },
                "required": ["path", "impact", "confidence"],
            },
        },
        "architecture_rules": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "source": {"type": "string"},
                    "rule": {"type": "string"},
                },
                "required": ["source", "rule"],
            },
        },
        "test_recommendations": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "reason": {"type": "string"},
                },
                "required": ["path", "reason"],
            },
        },
        "risks": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "category": {"type": "string"},
                    "severity": {
                        "type": "string",
                        "enum": ["low", "medium", "high"],
                    },
                    "evidence": {"type": "string"},
                },
                "required": ["category", "severity", "evidence"],
            },
        },
        "evidence": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "start_line": {"type": "integer"},
                    "end_line": {"type": "integer"},
                    "claim": {"type": "string"},
                },
                "required": ["path", "start_line", "end_line", "claim"],
            },
        },
        "unresolved_questions": {
            "type": "array",
            "items": {"type": "string"},
        },
    },
    "required": [
        "summary",
        "affected_components",
        "architecture_rules",
        "test_recommendations",
        "risks",
        "evidence",
        "unresolved_questions",
    ],
}


def post_json(endpoint, payload):
    body = json.dumps(payload).encode("utf-8")
    request = Request(
        f"{OLLAMA_BASE_URL}{endpoint}",
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=180) as response:
        return json.loads(response.read().decode("utf-8"))


def parse_changed_paths(diff_text):
    return sorted(
        {
            line[6:].strip()
            for line in diff_text.splitlines()
            if line.startswith("+++ b/")
        }
    )


def naive_report(diff_text):
    return {
        "changed_files": parse_changed_paths(diff_text),
        "limitation": "Only directly changed files are visible.",
    }


def collect_chunks(repo_root, lines_per_chunk=40, overlap=8):
    chunks = []
    stride = lines_per_chunk - overlap

    for path in sorted(repo_root.rglob("*")):
        if not path.is_file() or path.suffix.lower() not in ALLOWED_SUFFIXES:
            continue

        relative = path.relative_to(repo_root)
        if any(part in EXCLUDED_DIRS for part in relative.parts):
            continue

        lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
        if not lines:
            lines = [""]

        for start in range(0, len(lines), stride):
            end = min(start + lines_per_chunk, len(lines))
            chunks.append(
                {
                    "path": relative.as_posix(),
                    "start_line": start + 1,
                    "end_line": end,
                    "text": "\n".join(lines[start:end]),
                }
            )
            if end == len(lines):
                break

    return chunks


def module_name_from_path(path):
    if not path.endswith(".py"):
        return ""
    return path[:-3].replace("/", ".")


def build_dependency_targets(repo_root, changed_paths):
    changed_modules = {
        module_name_from_path(path)
        for path in changed_paths
        if module_name_from_path(path)
    }
    dependents = set()

    for path in sorted(repo_root.rglob("*.py")):
        relative = path.relative_to(repo_root).as_posix()
        try:
            tree = ast.parse(
                path.read_text(encoding="utf-8", errors="replace"),
                filename=relative,
            )
        except SyntaxError:
            continue

        imported_modules = set()
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                imported_modules.update(alias.name for alias in node.names)
            elif isinstance(node, ast.ImportFrom) and node.module:
                imported_modules.add(node.module)

        if any(
            imported == changed or imported.startswith(f"{changed}.")
            for imported in imported_modules
            for changed in changed_modules
        ):
            dependents.add(relative)

    return sorted(dependents)


def tokenize(text):
    normalized = "".join(
        character.lower()
        if character.isalnum() or character == "_"
        else " "
        for character in text
    )
    return {token for token in normalized.split() if len(token) >= 3}


def cosine_similarity(left, right):
    dot_product = sum(a * b for a, b in zip(left, right))
    left_norm = sum(value * value for value in left) ** 0.5
    right_norm = sum(value * value for value in right) ** 0.5
    if left_norm == 0 or right_norm == 0:
        return 0.0
    return dot_product / (left_norm * right_norm)


def retrieve(repo_root, diff_text, chunks, top_k=4):
    changed_paths = set(parse_changed_paths(diff_text))
    dependency_paths = set(build_dependency_targets(repo_root, changed_paths))
    embedding_inputs = [diff_text] + [
        f"File: {chunk['path']}\n{chunk['text']}" for chunk in chunks
    ]

    response = post_json(
        "/api/embed",
        {
            "model": EMBED_MODEL,
            "input": embedding_inputs,
            "truncate": True,
        },
    )
    vectors = response["embeddings"]
    if len(vectors) != len(embedding_inputs):
        raise RuntimeError("Embedding count did not match input count.")

    query_vector = vectors[0]
    query_tokens = tokenize(diff_text)
    ranked = []

    for chunk, vector in zip(chunks, vectors[1:]):
        chunk_tokens = tokenize(chunk["text"])
        lexical = len(query_tokens & chunk_tokens) / max(1, len(query_tokens))
        semantic = cosine_similarity(query_vector, vector)
        changed_boost = 0.30 if chunk["path"] in changed_paths else 0.0
        dependency_boost = 0.20 if chunk["path"] in dependency_paths else 0.0
        score = 0.70 * semantic + 0.30 * lexical + changed_boost + dependency_boost
        ranked.append({**chunk, "score": round(score, 4)})

    return sorted(ranked, key=lambda item: item["score"], reverse=True)[:top_k]


def format_context(ranked_chunks):
    sections = []
    for chunk in ranked_chunks:
        sections.append(
            "\n".join(
                [
                    f"FILE: {chunk['path']}",
                    f"LINES: {chunk['start_line']}-{chunk['end_line']}",
                    chunk["text"],
                ]
            )
        )
    return "\n\n---\n\n".join(sections)


def generate_review(diff_text, ranked_chunks, dependency_targets):
    schema_text = json.dumps(REPORT_SCHEMA, indent=2)
    prompt = f"""Review this code change using only the supplied evidence.

DIFF:
{diff_text}

STATIC DEPENDENCY TARGETS:
{json.dumps(dependency_targets, indent=2)}

RETRIEVED EVIDENCE:
{format_context(ranked_chunks)}

RULES:
1. Do not invent files, behavior, tests, or architecture rules.
2. Every evidence item must use an exact FILE path and a line range shown above.
3. Put uncertainty or missing context in unresolved_questions.
4. Treat removed validation and public response-field changes as possible compatibility or business-rule risks when the evidence supports that conclusion.
5. Return only JSON matching this schema:
{schema_text}
"""

    response = post_json(
        "/api/chat",
        {
            "model": REVIEW_MODEL,
            "messages": [
                {
                    "role": "system",
                    "content": "You are a cautious code change impact reviewer. Prefer verified evidence over speculation.",
                },
                {"role": "user", "content": prompt},
            ],
            "stream": False,
            "format": REPORT_SCHEMA,
            "options": {"temperature": 0},
        },
    )
    return json.loads(response["message"]["content"])


def validate_evidence(report, ranked_chunks):
    valid_ranges = {}
    for chunk in ranked_chunks:
        valid_ranges.setdefault(chunk["path"], []).append(
            (chunk["start_line"], chunk["end_line"])
        )

    invalid = []
    for item in report.get("evidence", []):
        path = item.get("path")
        start = item.get("start_line")
        end = item.get("end_line")
        ranges = valid_ranges.get(path, [])
        is_valid = (
            isinstance(start, int)
            and isinstance(end, int)
            and start <= end
            and any(start >= low and end <= high for low, high in ranges)
        )
        if not is_valid:
            invalid.append(item)

    return {
        "valid_count": len(report.get("evidence", [])) - len(invalid),
        "invalid_count": len(invalid),
        "invalid_evidence": invalid,
    }


def build_parser():
    parser = argparse.ArgumentParser(
        description="Review a unified diff against local repository evidence."
    )
    parser.add_argument("--repo", type=Path, required=True)
    parser.add_argument("--diff", type=Path, required=True)
    parser.add_argument(
        "--mode",
        choices=["naive", "retrieve", "review"],
        default="review",
    )
    parser.add_argument("--top-k", type=int, default=4)
    parser.add_argument("--output", type=Path)
    return parser


def main():
    args = build_parser().parse_args()
    diff_text = args.diff.read_text(encoding="utf-8")

    if args.mode == "naive":
        result = naive_report(diff_text)
    else:
        chunks = collect_chunks(args.repo)
        ranked = retrieve(args.repo, diff_text, chunks, args.top_k)
        dependencies = build_dependency_targets(
            args.repo, parse_changed_paths(diff_text)
        )

        if args.mode == "retrieve":
            result = {
                "changed_files": parse_changed_paths(diff_text),
                "dependency_targets": dependencies,
                "retrieved": [
                    {
                        "path": chunk["path"],
                        "lines": f"{chunk['start_line']}-{chunk['end_line']}",
                        "score": chunk["score"],
                    }
                    for chunk in ranked
                ],
            }
        else:
            result = generate_review(diff_text, ranked, dependencies)
            result["changed_files"] = parse_changed_paths(diff_text)
            result["dependency_targets"] = dependencies
            result["retrieval_context"] = [
                {
                    "path": chunk["path"],
                    "start_line": chunk["start_line"],
                    "end_line": chunk["end_line"],
                    "score": chunk["score"],
                }
                for chunk in ranked
            ]
            result["evidence_validation"] = validate_evidence(result, ranked)

    rendered = json.dumps(result, indent=2)
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(rendered + "\n", encoding="utf-8")
        print(f"Wrote {args.output}")
    else:
        print(rendered)


if __name__ == "__main__":
    main()

What Changed in This File?

  • The REPORT_SCHEMA dictionary fixes the required report sections plus the allowed confidence and severity values.
  • The generate_review() function combines the diff with dependency targets plus retrieved evidence.
  • The validate_evidence() function checks model citations against retrieved file ranges.
  • The review branch adds retrieval metadata before writing the final report.

Why Use a Structured Review?

A free-form answer can skip tests or bury uncertainty. The schema makes each promised review section mandatory.

Citation checks provide a separate trust boundary. A plausible claim cannot pass unnoticed when its path or line range falls outside the retrieved evidence.

  • Save reviewer.py.
Inspect the local generation request

The generate_review() function turns retrieved context into the prompt for Qwen2.5-Coder. The exact local model identifier remains qwen2.5-coder:3b.

  • In reviewer.py, find def generate_review(diff_text, ranked_chunks, dependency_targets):.
  • Confirm the prompt includes the diff under DIFF:.
  • Confirm the prompt includes static dependents under STATIC DEPENDENCY TARGETS:.
  • Confirm the prompt includes ranked chunks under RETRIEVED EVIDENCE:.
  • Confirm the request sends data to /api/chat.
  • Confirm the request sets stream to False.
  • Confirm the request puts REPORT_SCHEMA in format.
  • Confirm the request sets temperature to 0.

How Does the Request Stay Grounded?

The prompt tells the model to use only supplied evidence. It also sends the static dependency targets beside the retrieved chunks.

The schema appears in the prompt plus the request's format field. A temperature of 0 makes the completion more deterministic.

Run and verify the report

The final layer checks whether each cited path exists in the retrieval context. It also checks whether every cited line range fits inside a retrieved chunk.

Before you run the review, which required report sections do you expect the schema to preserve?

  • Generate the evidence-checked report in PowerShell by running this command:
py reviewer.py --repo sample_repo --diff fixtures/discount_cap.patch --mode review --top-k 4 --output reports/discount_review.json

What Does This Command Do?

The review mode retrieves four evidence chunks before generating the structured result. It validates the model's citations against those chunks.

The --output option creates the reports directory when needed. It saves the rendered result as reports/discount_review.json.

PowerShell prints a confirmation naming reports/discount_review.json. That is the key handoff complete: your reviewer has turned ranked evidence into a saved decision artifact.

Report File Not Created?

  • Confirm the local Ollama application is still running.
  • Check that the two models from Step 1 still appear in your installed model list.
  • Compare the indentation in generate_review() plus validate_evidence() with the full reference.

Help me debug the evidence-checked review command.

  • Expand the reports folder in the Explorer sidebar.
  • Select discount_review.json.

You should see a structured summary plus affected components. Architecture rules, test recommendations, risks, evidence, plus unresolved questions should follow.

  • Confirm the report records changed_files.
  • Confirm the report records dependency_targets.
  • Confirm the report records retrieval_context.

Before you inspect the validation result, do you expect every model citation to fit inside one retrieved chunk?

  • Compare each entry under evidence with its matching entry under retrieval_context.
  • Locate evidence_validation near the end of the report.
  • Check the value of invalid_count.

With grounded citations, you should see invalid_count set to 0. Any nonzero value identifies a citation outside the retrieved file ranges.

Your reviewer now turns local retrieval into a structured report with inspectable evidence. Next, you will benchmark whether it consistently finds the files that matter.

Benchmark the Seeded Defect

Your reviewer now produces an evidence-checked report for the seeded discount defect. One convincing report still cannot prove that the retrieval pipeline consistently finds the known blast radius.

A benchmark compares the retrieved files with a known answer set. It turns retrieval recall into a repeatable quality signal. A separate dependency-graph recall score checks the deterministic import analysis.

In this step, get ready to:
  • Define the expected evidence for the seeded discount defect.
  • Build an evaluation command that calculates both recall scores.
  • Run the benchmark to produce a visible pass or fail result.
Define the seeded evaluation case

An evaluation case records the files a trustworthy reviewer should find. It also sets the minimum scores required for the case to pass.

  • Return to the file sidebar in Visual Studio Code from earlier.
  • Select the fixtures folder.
  • Use the new-file control at the top of the sidebar.
  • Name the file eval_cases.json.
  • Fill fixtures/eval_cases.json with the seeded case by pasting this specification:
{
  "cases": [
    {
      "name": "discount-cap-bypass",
      "repo": "sample_repo",
      "diff": "fixtures/discount_cap.patch",
      "top_k": 4,
      "expected_paths": [
        "src/pricing.py",
        "src/checkout.py",
        "tests/test_pricing.py",
        "docs/architecture.md"
      ],
      "expected_dependency_paths": [
        "src/checkout.py",
        "tests/test_pricing.py"
      ],
      "minimum_retrieval_recall": 0.75,
      "minimum_dependency_recall": 1.0
    }
  ]
}

What Does This Evaluation Case Measure?

  • The expected_paths list defines the known evidence set for the discount regression.
  • The expected_dependency_paths list isolates files that should be discovered through Python imports.
  • A top_k value of 4 limits the semantic retrieval result to four chunks.
  • The retrieval threshold of 0.75 requires at least three of the four expected paths.
  • The dependency threshold of 1.0 requires both expected dependents.
  • Save fixtures/eval_cases.json.
  • Confirm the file shows one case named discount-cap-bypass in the editor.

Seeing a JSON Syntax Problem?

  • Check that every property except the final property in an object has a trailing comma.
  • Check that the closing braces match the opening braces.
  • Make sure the file sits inside the existing fixtures folder.

Help me check the evaluation case syntax.

Build the evaluation command

The evaluator reuses the same retrieval functions as the reviewer. This keeps the benchmark focused on the production pipeline instead of a separate test implementation.

  • Select the top-level impact-reviewer folder in the file sidebar.
  • Use the new-file control at the top of the sidebar.
  • Name the file evaluate.py.
  • Add the imports and recall calculation by pasting this first section:
import argparse
import json
from pathlib import Path

from reviewer import (
    build_dependency_targets,
    collect_chunks,
    parse_changed_paths,
    retrieve,
)


def calculate_recall(expected, actual):
    if not expected:
        return 1.0
    return len(expected & actual) / len(expected)

How Does Recall Work?

  • The imports reuse chunking from reviewer.py.
  • They also reuse hybrid retrieval from reviewer.py.
  • The calculate_recall function divides the number of expected files found by the total expected file count.
  • An empty expected set returns 1.0 because the evaluator has no required files to miss.
  • Save evaluate.py.
  • Confirm Visual Studio Code shows no red syntax underline in the new section.

Seeing an Import Warning?

  • Make sure evaluate.py is beside reviewer.py in the impact-reviewer folder.
  • Check that every imported function name matches the spelling in reviewer.py.

Help me resolve the evaluator imports.

The command-line parser accepts an evaluation specification as a file path. Its default points to the case you just created.

  • Add build_parser directly below calculate_recall by pasting this code:
def build_parser():
    parser = argparse.ArgumentParser(
        description="Evaluate retrieval and dependency recall on seeded diffs."
    )
    parser.add_argument(
        "--cases",
        type=Path,
        default=Path("fixtures/eval_cases.json"),
    )
    return parser

What Does the Parser Control?

  • The --cases argument lets the learner select an evaluation specification.
  • The Path type gives the evaluator file-reading methods.
  • The default keeps the seeded benchmark available without changing the source code.
  • Save evaluate.py.
  • Confirm the editor shows the --cases argument below calculate_recall.

Parser Section Underlined?

  • Check that build_parser starts at the left edge of the file.
  • Check that each nested parser line uses consistent indentation.

Help me fix the parser indentation.

The first half of main loads each case. It then runs the real chunking pipeline before collecting both observed path sets.

  • Add the first half of main directly below build_parser by pasting this code:
def main():
    args = build_parser().parse_args()
    specification = json.loads(args.cases.read_text(encoding="utf-8"))
    results = []

    for case in specification["cases"]:
        repo_root = Path(case["repo"])
        diff_text = Path(case["diff"]).read_text(encoding="utf-8")
        chunks = collect_chunks(repo_root)
        ranked = retrieve(
            repo_root,
            diff_text,
            chunks,
            top_k=case.get("top_k", 4),
        )

        expected_paths = set(case["expected_paths"])
        retrieved_paths = {item["path"] for item in ranked}
        expected_dependencies = set(case["expected_dependency_paths"])
        dependency_paths = set(
            build_dependency_targets(
                repo_root,
                parse_changed_paths(diff_text),
            )
        )

How Does Each Case Run?

  • The evaluator reads the repository path from the case specification.
  • The evaluator passes the seeded diff through the same collect_chunks and retrieve functions used by the reviewer.
  • Python sets remove duplicate paths before the recall calculation.
  • The dependency path set comes from the same AST import analysis used in review mode.
  • Save evaluate.py.
  • Confirm the nested lines beneath for case share one indentation level.

Seeing Indentation Marks?

  • Check that every line belonging to the case loop is indented beneath for case.
  • Check that the lines inside build_dependency_targets are indented one additional level.

Help me check the main loop indentation.

The continuation compares each observed set with its expected set. It records the measurements before calculating one overall benchmark result.

  • Complete main by pasting this continuation directly below the dependency_paths calculation:
        retrieval_recall = calculate_recall(expected_paths, retrieved_paths)
        dependency_recall = calculate_recall(
            expected_dependencies, dependency_paths
        )
        passed = (
            retrieval_recall >= case["minimum_retrieval_recall"]
            and dependency_recall >= case["minimum_dependency_recall"]
        )

        results.append(
            {
                "name": case["name"],
                "retrieved_paths": sorted(retrieved_paths),
                "dependency_paths": sorted(dependency_paths),
                "retrieval_recall": round(retrieval_recall, 3),
                "dependency_recall": round(dependency_recall, 3),
                "passed": passed,
            }
        )

    summary = {
        "passed": all(result["passed"] for result in results),
        "cases": results,
    }
    print(json.dumps(summary, indent=2))
    return 0 if summary["passed"] else 1


if __name__ == "__main__":
    raise SystemExit(main())

How Is Pass or Fail Decided?

  • Each recall value is compared with the matching minimum from fixtures/eval_cases.json.
  • The case passes only when both minimums are met.
  • The top-level summary passes only when every case passes.
  • The process returns status 0 for an overall pass.
  • The process returns status 1 for an overall failure.
  • Save evaluate.py.
  • Confirm Visual Studio Code shows no red syntax underline anywhere in the file.

Final Section Misaligned?

  • Keep the recall calculations inside the for case loop.
  • Keep the summary block outside the loop.
  • Keep the final module check at the left edge of the file.

Help me verify the evaluator structure.

✔️ Awesome, I've got everything!

Your evaluator now loads benchmark cases. It measures both retrieval signals before producing one JSON summary.

ⓧ I'd like to double check the full code

import argparse
import json
from pathlib import Path

from reviewer import (
    build_dependency_targets,
    collect_chunks,
    parse_changed_paths,
    retrieve,
)


def calculate_recall(expected, actual):
    if not expected:
        return 1.0
    return len(expected & actual) / len(expected)


def build_parser():
    parser = argparse.ArgumentParser(
        description="Evaluate retrieval and dependency recall on seeded diffs."
    )
    parser.add_argument(
        "--cases",
        type=Path,
        default=Path("fixtures/eval_cases.json"),
    )
    return parser


def main():
    args = build_parser().parse_args()
    specification = json.loads(args.cases.read_text(encoding="utf-8"))
    results = []

    for case in specification["cases"]:
        repo_root = Path(case["repo"])
        diff_text = Path(case["diff"]).read_text(encoding="utf-8")
        chunks = collect_chunks(repo_root)
        ranked = retrieve(
            repo_root,
            diff_text,
            chunks,
            top_k=case.get("top_k", 4),
        )

        expected_paths = set(case["expected_paths"])
        retrieved_paths = {item["path"] for item in ranked}
        expected_dependencies = set(case["expected_dependency_paths"])
        dependency_paths = set(
            build_dependency_targets(
                repo_root,
                parse_changed_paths(diff_text),
            )
        )

        retrieval_recall = calculate_recall(expected_paths, retrieved_paths)
        dependency_recall = calculate_recall(
            expected_dependencies, dependency_paths
        )
        passed = (
            retrieval_recall >= case["minimum_retrieval_recall"]
            and dependency_recall >= case["minimum_dependency_recall"]
        )

        results.append(
            {
                "name": case["name"],
                "retrieved_paths": sorted(retrieved_paths),
                "dependency_paths": sorted(dependency_paths),
                "retrieval_recall": round(retrieval_recall, 3),
                "dependency_recall": round(dependency_recall, 3),
                "passed": passed,
            }
        )

    summary = {
        "passed": all(result["passed"] for result in results),
        "cases": results,
    }
    print(json.dumps(summary, indent=2))
    return 0 if summary["passed"] else 1


if __name__ == "__main__":
    raise SystemExit(main())
Run the benchmark and inspect recall

The benchmark now has a known answer set. Its output reveals whether the reviewer finds enough relevant evidence before you treat the generated report as trustworthy.

Before you run the benchmark, consider whether both recall thresholds will pass.

  • Return to PowerShell from earlier.
  • Run the seeded evaluation by entering this command:
py evaluate.py --cases fixtures/eval_cases.json

You should see a JSON result for discount-cap-bypass. The case should report passed: true.

The output should include retrieval recall of at least 0.75. Dependency recall should be 1.0.

What Does a Passing Result Prove?

  • The retrieved_paths list shows which files entered the review context.
  • The dependency_paths list shows which direct import relationships the static graph discovered.
  • The recall scores compare those observed paths with the known affected paths.
  • The passed value makes the quality threshold visible and repeatable.

Benchmark Did Not Pass?

  • Check that Ollama from earlier is still available to serve the local embedding model.
  • Check that fixtures/eval_cases.json contains the four expected retrieval paths.
  • Check that evaluate.py passes the configured top_k value into retrieve.

Help me diagnose the failed benchmark.

You now have measurable proof that the reviewer finds the seeded defect's important files. That benchmark gives the evidence-checked report a repeatable quality gate.

Secret mission

Catch a Public Contract Regression

One successful benchmark can hide a reviewer that only works for one familiar defect. Add a public checkout contract regression to prove the same retrieval pipeline can uncover a different compatibility risk.

Clean Up Your Resources

Clean Up Your Resources

All of your reviewer resources run on your Windows machine. There are no paid cloud services or ongoing financial costs.

Choose whether to keep the workspace ready, pause active local sessions, or delete the project files.

Your Ollama model files stay installed because this project omits unverified model-specific removal commands. The installed tags are embeddinggemma plus qwen2.5-coder:3b.

Resources you used:

  • The local impact-reviewer project folder.
  • The generated reports/discount_review.json review report inside the project folder.

Keep everything running

Keep the workspace intact if you want to test more diffs or improve the retrieval benchmark.

  • Continue working in the existing impact-reviewer workspace in Visual Studio Code.
  • Retain reports/discount_review.json as evidence of the generated review.
  • Leave both local models installed for future codebase reviews.

The project continues to use local disk space. It creates no financial cost.

Pause - I'll come back to this later

Pause active work to free editor memory while preserving your files. Both downloaded models remain available for your next session.

  • End any active Ollama model prompt by pressing Ctrl+C in the PowerShell panel.
  • Close the PowerShell terminal panel.
  • Save any open files in impact-reviewer.
  • Close Visual Studio Code.

Your benchmark files remain inside impact-reviewer. Your models remain downloaded on your machine.

Delete - I don't want to use this again

Permanent deletion can feel drastic. This removes only the local impact-reviewer folder.

The downloaded models stay installed because model-specific removal commands are intentionally omitted.

  • Right-click the impact-reviewer folder in the Explorer sidebar in Visual Studio Code.
  • Select Reveal in File Explorer.
  • Close Visual Studio Code.
  • Select the impact-reviewer folder in File Explorer.
  • Press Shift+Delete to permanently remove the folder.
  • Confirm the permanent deletion prompt.
  • Check that impact-reviewer no longer appears in its parent folder.

You should no longer see the project folder. The reports/discount_review.json report disappears with it.

Nice Work!

Nice Work!

You did it! Your local AI reviewer now turns risky patches into evidence-checked impact reports that pass two seeded regression benchmarks.

You've learned how to:

  • Build a hybrid retrieval pipeline that uncovers the blast radius hidden by a changed-file-only review. AST import analysis finds direct Python dependents. Lexical matching rewards shared terms. Local embeddings surface architecture documentation.
  • Generate a schema-constrained JSON report for the discount patch. Validate every evidence path against the retrieved chunks. Check each cited line range against the same evidence.
  • Benchmark retrieval recall for seeded defects. Measure dependency recall from the import graph. Receive an overall passing result only when every case meets its threshold.
  • Secret Mission: Extended the unchanged retrieval pipeline with a public contract regression benchmark. Caught the rename from total_cents to total without adding case-specific retrieval logic.

Ready to quiz yourself?