← Back to DevBytes

Building a Code Review Agent with LangChain: Complete Guide

Introduction: What Is a Code Review Agent?

A Code Review Agent is an AI-powered assistant that automatically analyzes source code changes, identifies potential issues, suggests improvements, and enforces coding standards — all without human intervention. Built with LangChain, this agent leverages large language models (LLMs) to understand code semantics, reason about best practices, and provide actionable feedback in natural language.

Unlike traditional static analysis tools that rely on pattern matching or AST parsing alone, a LangChain-based code review agent combines the precision of programmatic code parsing with the nuanced understanding of LLMs. It can detect not just syntax errors, but also logical flaws, security vulnerabilities, style violations, and even architectural concerns — effectively acting as an always-available senior developer on your team.

Why a Code Review Agent Matters

Manual code review is expensive, time-consuming, and inconsistent. Studies show that human reviewers miss up to 40% of defects due to fatigue, time pressure, or cognitive bias. An AI agent addresses these challenges in several critical ways:

Organizations adopting AI code review report 30-60% reduction in review cycle time and a measurable decrease in production defects. The agent becomes a force multiplier for engineering teams.

Prerequisites and Environment Setup

Before building the agent, ensure you have the following installed and configured:

# Create a new Python project directory
mkdir code-review-agent
cd code-review-agent

# Create and activate a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install required packages
pip install langchain langchain-openai langchain-community
pip install gitpython tree-sitter rich
pip install pydantic python-dotenv

You'll also need an OpenAI API key (or an alternative LLM provider). Store it securely in a .env file:

# .env file
OPENAI_API_KEY=sk-your-actual-key-here
OPENAI_MODEL=gpt-4o-mini

Load environment variables at the top of your main script:

import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

Core Architecture Overview

The agent follows a modular pipeline architecture powered by LangChain's chain composition capabilities. The high-level flow is:

  1. Diff Parser — Extracts changed files and hunks from a git diff
  2. Code Segmenter — Splits large diffs into manageable, context-aware chunks
  3. Context Enricher — Fetches surrounding code (file context) for each chunk
  4. Analysis Chain — LangChain chain that orchestrates multiple LLM calls per chunk
  5. Aggregator — Combines findings, deduplicates, and formats the final report

Each component is implemented as a focused, testable module. Let's build them one by one.

Step 1: Parsing Git Diffs

We start by extracting structured information from a git diff. This function uses gitpython to get the diff between two commits or branches:

import git
from typing import List, Dict, Optional
from dataclasses import dataclass

@dataclass
class DiffHunk:
    """Represents a single hunk within a file diff."""
    file_path: str
    old_start: int
    old_count: int
    new_start: int
    new_count: int
    lines: List[str]
    change_type: str  # 'added', 'modified', 'deleted'

@dataclass
class FileDiff:
    """Represents all changes to a single file."""
    file_path: str
    change_type: str
    hunks: List[DiffHunk]
    old_content: Optional[str] = None
    new_content: Optional[str] = None

def parse_git_diff(repo_path: str, base_branch: str = "main", 
                   target_branch: str = "HEAD") -> List[FileDiff]:
    """
    Parse the git diff between two branches into structured FileDiff objects.
    
    Args:
        repo_path: Path to the git repository
        base_branch: The base branch for comparison (e.g., 'main')
        target_branch: The branch being reviewed (e.g., 'HEAD' or 'feature-branch')
    
    Returns:
        List of FileDiff objects representing all changed files
    """
    repo = git.Repo(repo_path)
    
    # Get the diff between base and target
    base_commit = repo.rev_parse(base_branch)
    target_commit = repo.rev_parse(target_branch)
    
    # Get diff index
    diff_index = base_commit.diff(target_commit)
    
    file_diffs = []
    
    for change in diff_index:
        change_type = change.change_type  # 'A', 'M', 'D', 'R'
        
        if change_type == 'D':
            # File was deleted - we may still want to review the removal
            file_diffs.append(FileDiff(
                file_path=change.b_path or change.a_path,
                change_type='deleted',
                hunks=[],
                old_content=None,
                new_content=None
            ))
            continue
        
        # For added/modified files, get the diff text
        diff_text = repo.git.diff(base_branch, target_branch, "--", 
                                   change.b_path or change.a_path)
        
        hunks = _parse_unified_diff_hunks(diff_text, change.b_path or change.a_path)
        
        file_diffs.append(FileDiff(
            file_path=change.b_path or change.a_path,
            change_type='added' if change_type == 'A' else 'modified',
            hunks=hunks,
            old_content=None,  # Will be populated later if needed
            new_content=None
        ))
    
    return file_diffs

def _parse_unified_diff_hunks(diff_text: str, file_path: str) -> List[DiffHunk]:
    """
    Parse unified diff format into structured hunk objects.
    Handles the standard @@ -a,b +c,d @@ header format.
    """
    hunks = []
    current_hunk_lines = []
    old_start = old_count = new_start = new_count = 0
    
    for line in diff_text.split('\n'):
        if line.startswith('@@ '):
            # Save previous hunk if any
            if current_hunk_lines:
                hunks.append(DiffHunk(
                    file_path=file_path,
                    old_start=old_start,
                    old_count=old_count,
                    new_start=new_start,
                    new_count=new_count,
                    lines=current_hunk_lines.copy(),
                    change_type='modified'
                ))
                current_hunk_lines = []
            
            # Parse hunk header: @@ -old_start,old_count +new_start,new_count @@
            parts = line.split()
            old_part = parts[1].lstrip('-')
            new_part = parts[2].lstrip('+')
            
            if ',' in old_part:
                old_start, old_count = map(int, old_part.split(','))
            else:
                old_start = int(old_part)
                old_count = 1
            
            if ',' in new_part:
                new_start, new_count = map(int, new_part.split(','))
            else:
                new_start = int(new_part)
                new_count = 1
        else:
            current_hunk_lines.append(line)
    
    # Don't forget the last hunk
    if current_hunk_lines:
        hunks.append(DiffHunk(
            file_path=file_path,
            old_start=old_start,
            old_count=old_count,
            new_start=new_start,
            new_count=new_count,
            lines=current_hunk_lines.copy(),
            change_type='modified'
        ))
    
    return hunks

This parser gives us structured access to every changed line, grouped by file and hunk — the foundation for targeted analysis.

Step 2: Building the LangChain Analysis Chain

Now we construct the core intelligence using LangChain. The analysis chain takes a code diff chunk and surrounding context, then produces structured review feedback. We'll use ChatOpenAI with a carefully engineered prompt:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser, PydanticOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from pydantic import BaseModel, Field
from typing import List, Optional

class ReviewFinding(BaseModel):
    """Structured output for a single review finding."""
    severity: str = Field(description="One of: 'critical', 'high', 'medium', 'low', 'suggestion'")
    category: str = Field(description="e.g., 'security', 'performance', 'style', 'logic', 'documentation'")
    line_number: Optional[int] = Field(description="The line number in the new file where the issue occurs")
    file_path: str = Field(description="Path to the file being reviewed")
    title: str = Field(description="Short, concise title of the finding")
    description: str = Field(description="Detailed explanation of the issue")
    suggestion: str = Field(description="Specific code or action recommendation to fix the issue")
    code_snippet: Optional[str] = Field(description="Relevant code excerpt showing the issue")

class ReviewResult(BaseModel):
    """Container for all findings from a review."""
    file_path: str
    findings: List[ReviewFinding]
    summary: str = Field(description="Overall assessment of the changes in this file")

# Create the output parser
review_output_parser = PydanticOutputParser(ReviewResult)

# The system prompt sets the agent's expertise and rules
SYSTEM_TEMPLATE = """\
You are an expert code reviewer with deep knowledge across multiple programming languages, 
frameworks, and software engineering best practices. Your task is to review code diffs 
and provide constructive, actionable feedback.

## Review Guidelines
Analyze the provided code diff for:
1. **Security vulnerabilities**: SQL injection, XSS, insecure deserialization, hardcoded secrets, 
   missing input validation, authentication bypass risks, unsafe dependencies
2. **Performance issues**: N+1 queries, unnecessary allocations, blocking operations in async code, 
   inefficient algorithms, missing caching opportunities
3. **Code quality**: Readability, naming conventions, function length, cyclomatic complexity, 
   duplicate code, dead code, proper error handling
4. **Logical correctness**: Off-by-one errors, null safety, edge case handling, race conditions, 
   incorrect assumptions
5. **Best practices**: Language-specific idioms, design patterns, SOLID principles, DRY violations
6. **Documentation**: Missing docstrings, unclear comments, outdated documentation

## Severity Levels
- **critical**: Security vulnerability or bug that will cause production failure
- **high**: Significant performance degradation or logical error likely to cause bugs
- **medium**: Code quality issue that impacts maintainability or may cause future problems
- **low**: Minor style or convention deviation
- **suggestion**: Optional improvement that would enhance the code

## Important Rules
- Only report findings that are actually present in the diff — do not invent issues
- Provide specific, concrete suggestions with code examples when applicable
- Reference exact line numbers from the NEW file (lines starting with '+' in the diff)
- If the code is clean and well-written, report an empty findings list with a positive summary
- Focus on the changed code, not pre-existing issues in unchanged lines

{format_instructions}
"""

HUMAN_TEMPLATE = """\
## File: {file_path}
## Change Type: {change_type}

### Surrounding Context (for reference — NOT part of the review)
{language}
{context_before}
### The Actual Diff to Review (lines with +/- are the changes)
diff
{diff_content}
### Context After the Diff
{language}
{context_after}
Please analyze the diff above and provide your review findings.
Focus ONLY on the changed lines (prefixed with + or -). The context before/after 
is provided solely to help you understand the code's surroundings.

Language: {language}
"""

def build_review_prompt():
    """Construct the full prompt template with format instructions injected."""
    return ChatPromptTemplate.from_messages([
        SystemMessagePromptTemplate.from_template(SYSTEM_TEMPLATE),
        HumanMessagePromptTemplate.from_template(HUMAN_TEMPLATE)
    ])

def create_analysis_chain(model_name: str = "gpt-4o-mini", temperature: float = 0.1):
    """
    Build the complete LangChain analysis chain.
    
    The chain: prompt -> LLM -> structured output parser
    """
    llm = ChatOpenAI(
        model=model_name,
        temperature=temperature,
        openai_api_key=OPENAI_API_KEY
    )
    
    prompt = build_review_prompt()
    
    # Partially fill the prompt's format_instructions
    prompt = prompt.partial(
        format_instructions=review_output_parser.get_format_instructions()
    )
    
    # Build the chain
    chain = prompt | llm | review_output_parser
    
    return chain

The prompt engineering here is critical. We clearly separate context (unchanged surrounding code) from the actual diff (lines to review), instruct the model to focus only on changes, and demand structured output via Pydantic parsing. This prevents the model from critiquing legacy code and keeps feedback relevant.

Step 3: Context Enrichment

For each diff hunk, we fetch surrounding lines from the actual file to give the LLM necessary context. Without context, the model sees isolated snippets and may misunderstand the code's purpose:

def get_file_context(file_path: str, hunk_new_start: int, 
                     context_lines: int = 50) -> tuple[str, str]:
    """
    Read context before and after a diff hunk from the actual file.
    
    Args:
        file_path: Absolute or relative path to the file
        hunk_new_start: The line number where the hunk starts in the new file
        context_lines: Number of lines of context to fetch before and after
        
    Returns:
        Tuple of (context_before, context_after) as strings
    """
    try:
        with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
            all_lines = f.readlines()
    except FileNotFoundError:
        return ("[File not found - unable to provide context]", 
                "[File not found]")
    except Exception as e:
        return (f"[Error reading file: {e}]", f"[Error: {e}]")
    
    # Context before: lines leading up to the hunk
    start_before = max(0, hunk_new_start - context_lines - 1)
    context_before_lines = all_lines[start_before:hunk_new_start - 1]
    context_before = ''.join(context_before_lines)
    
    # Context after: lines following the hunk
    # Estimate hunk end based on new_count
    hunk_end = hunk_new_start + hunk_new_count
    end_after = min(len(all_lines), hunk_end + context_lines)
    context_after_lines = all_lines[hunk_end:end_after]
    context_after = ''.join(context_after_lines)
    
    return (context_before.strip(), context_after.strip())

def detect_language(file_path: str) -> str:
    """Detect programming language from file extension."""
    ext_map = {
        '.py': 'python',
        '.js': 'javascript',
        '.ts': 'typescript',
        '.jsx': 'jsx',
        '.tsx': 'tsx',
        '.java': 'java',
        '.go': 'go',
        '.rs': 'rust',
        '.cpp': 'cpp',
        '.c': 'c',
        '.h': 'c',
        '.rb': 'ruby',
        '.php': 'php',
        '.sql': 'sql',
        '.html': 'html',
        '.css': 'css',
        '.scss': 'scss',
        '.yaml': 'yaml',
        '.yml': 'yaml',
        '.json': 'json',
        '.md': 'markdown',
        '.sh': 'bash',
        '.bash': 'bash',
        '.tf': 'hcl',
        '.dockerfile': 'dockerfile',
    }
    
    ext = os.path.splitext(file_path)[1].lower()
    return ext_map.get(ext, 'plaintext')

Step 4: Assembling the Full Review Pipeline

Now we wire everything together into a complete agent that processes an entire git diff, enriches each hunk with context, runs the analysis chain, and aggregates results:

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict
import json
import hashlib

class CodeReviewAgent:
    """
    Complete AI code review agent powered by LangChain.
    
    The agent:
    1. Parses a git diff into structured hunks
    2. Enriches each hunk with file context
    3. Runs parallel LLM analysis chains
    4. Aggregates and deduplicates findings
    5. Produces a comprehensive review report
    """
    
    def __init__(self, model_name: str = "gpt-4o-mini", 
                 context_lines: int = 60,
                 max_workers: int = 5,
                 ignore_patterns: Optional[List[str]] = None):
        self.model_name = model_name
        self.context_lines = context_lines
        self.max_workers = max_workers
        self.ignore_patterns = ignore_patterns or []
        self.chain = create_analysis_chain(model_name)
    
    def _should_review_file(self, file_path: str) -> bool:
        """Check if file should be reviewed based on ignore patterns."""
        for pattern in self.ignore_patterns:
            if pattern in file_path:
                return False
        
        # Skip binary and generated files
        skip_extensions = {'.lock', '.min.js', '.min.css', '.map', '.svg', 
                          '.png', '.jpg', '.gif', '.ico', '.woff', '.ttf', '.eot'}
        ext = os.path.splitext(file_path)[1].lower()
        if ext in skip_extensions:
            return False
        
        return True
    
    def _prepare_hunk_for_review(self, file_diff: FileDiff, 
                                  hunk: DiffHunk) -> dict:
        """
        Prepare the input dictionary for a single hunk review.
        Extracts context and formats the diff for the LLM.
        """
        file_path = file_diff.file_path
        language = detect_language(file_path)
        
        # Get context
        context_before, context_after = get_file_context(
            file_path, hunk.new_start, self.context_lines
        )
        
        # Format diff content - only the changed lines
        diff_content = '\n'.join(hunk.lines)
        
        return {
            "file_path": file_path,
            "change_type": file_diff.change_type,
            "language": language,
            "context_before": context_before[:2000],  # Truncate very long context
            "context_after": context_after[:2000],
            "diff_content": diff_content[:3000],  # Truncate very large hunks
        }
    
    def _deduplicate_findings(self, all_results: List[ReviewResult]) -> List[ReviewResult]:
        """
        Remove duplicate findings across multiple hunks of the same file.
        Uses similarity hashing on finding descriptions.
        """
        seen_hashes = set()
        deduplicated = []
        
        for result in all_results:
            unique_findings = []
            for finding in result.findings:
                # Create a hash from file_path + title + description
                hash_input = f"{finding.file_path}|{finding.title}|{finding.description[:100]}"
                finding_hash = hashlib.md5(hash_input.encode()).hexdigest()
                
                if finding_hash not in seen_hashes:
                    seen_hashes.add(finding_hash)
                    unique_findings.append(finding)
            
            if unique_findings or result.findings:  # Keep results that had findings
                result.findings = unique_findings
                deduplicated.append(result)
        
        return deduplicated
    
    def review(self, repo_path: str, base_branch: str = "main", 
               target_branch: str = "HEAD") -> Dict:
        """
        Execute a complete code review on the diff between two branches.
        
        Args:
            repo_path: Path to the git repository
            base_branch: Base branch for comparison
            target_branch: Target branch with changes to review
            
        Returns:
            Dictionary containing all review results and metadata
        """
        print(f"šŸ” Parsing git diff: {base_branch}...{target_branch}")
        file_diffs = parse_git_diff(repo_path, base_branch, target_branch)
        
        # Filter files
        reviewable_diffs = [d for d in file_diffs if self._should_review_file(d.file_path)]
        print(f"šŸ“‚ Found {len(file_diffs)} changed files, {len(reviewable_diffs)} to review")
        
        # Flatten all hunks into review tasks
        review_tasks = []
        for file_diff in reviewable_diffs:
            for hunk in file_diff.hunks:
                task_input = self._prepare_hunk_for_review(file_diff, hunk)
                review_tasks.append(task_input)
        
        print(f"🧩 Prepared {len(review_tasks)} review chunks")
        
        # Run reviews in parallel
        all_results = []
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            future_to_task = {
                executor.submit(self.chain.invoke, task): task 
                for task in review_tasks
            }
            
            completed = 0
            for future in as_completed(future_to_task):
                completed += 1
                task = future_to_task[future]
                try:
                    result = future.result(timeout=120)
                    all_results.append(result)
                    print(f"  āœ… [{completed}/{len(review_tasks)}] Reviewed: {task['file_path']}")
                except Exception as e:
                    print(f"  āš ļø  [{completed}/{len(review_tasks)}] Error reviewing {task['file_path']}: {e}")
                    # Create an error result
                    all_results.append(ReviewResult(
                        file_path=task['file_path'],
                        findings=[],
                        summary=f"Review failed due to error: {str(e)}"
                    ))
        
        # Deduplicate
        print("šŸ”— Deduplicating findings...")
        final_results = self._deduplicate_findings(all_results)
        
        # Generate overall statistics
        total_findings = sum(len(r.findings) for r in final_results)
        critical_count = sum(1 for r in final_results for f in r.findings if f.severity == 'critical')
        high_count = sum(1 for r in final_results for f in r.findings if f.severity == 'high')
        
        return {
            "base_branch": base_branch,
            "target_branch": target_branch,
            "total_files_changed": len(file_diffs),
            "files_reviewed": len(reviewable_diffs),
            "total_findings": total_findings,
            "critical_findings": critical_count,
            "high_findings": high_count,
            "results": final_results,
        }
    
    def format_report(self, review_output: Dict) -> str:
        """
        Format the review output into a readable markdown report.
        """
        lines = []
        lines.append("# AI Code Review Report")
        lines.append(f"**Base:** `{review_output['base_branch']}` → **Target:** `{review_output['target_branch']}`")
        lines.append(f"**Files Changed:** {review_output['total_files_changed']} | **Reviewed:** {review_output['files_reviewed']}")
        lines.append(f"**Total Findings:** {review_output['total_findings']} "
                    f"(šŸ”“ Critical: {review_output['critical_findings']}, "
                    f"🟠 High: {review_output['high_findings']})")
        lines.append("---")
        
        # Group results by severity
        severity_order = ['critical', 'high', 'medium', 'low', 'suggestion']
        all_findings = []
        for result in review_output['results']:
            for finding in result.findings:
                all_findings.append(finding)
        
        all_findings.sort(key=lambda f: severity_order.index(f.severity) 
                          if f.severity in severity_order else 999)
        
        current_file = None
        for finding in all_findings:
            if finding.file_path != current_file:
                current_file = finding.file_path
                lines.append(f"\n## šŸ“„ `{current_file}`")
                # Add file summary if available
                matching_results = [r for r in review_output['results'] 
                                   if r.file_path == current_file]
                if matching_results and matching_results[0].summary:
                    lines.append(f"> {matching_results[0].summary}")
            
            emoji = {'critical': 'šŸ”“', 'high': '🟠', 'medium': '🟔', 
                    'low': 'šŸ”µ', 'suggestion': 'šŸ’”'}.get(finding.severity, '⚪')
            
            lines.append(f"\n### {emoji} [{finding.severity.upper()}] {finding.title}")
            lines.append(f"**Category:** {finding.category} | **Line:** {finding.line_number or 'N/A'}")
            lines.append(f"\n{finding.description}")
            
            if finding.code_snippet:
                lines.append(f"\n{detect_language(finding.file_path)}")
                lines.append(finding.code_snippet)
                lines.append("")
            
            lines.append(f"\n**Suggested Fix:** {finding.suggestion}")
            lines.append("---")
        
        return '\n'.join(lines)

Step 5: Using the Agent

Here's how to invoke the agent from a script or CI/CD pipeline:

# main.py - Entry point for the code review agent

import os
import sys
from dotenv import load_dotenv

load_dotenv()

def main():
    """Run the code review agent on a repository."""
    
    # Configuration
    REPO_PATH = os.getenv("REPO_PATH", os.getcwd())
    BASE_BRANCH = os.getenv("BASE_BRANCH", "main")
    TARGET_BRANCH = os.getenv("TARGET_BRANCH", "HEAD")
    OUTPUT_FILE = os.getenv("OUTPUT_FILE", "review_report.md")
    
    print("šŸ¤– Initializing Code Review Agent...")
    
    agent = CodeReviewAgent(
        model_name="gpt-4o-mini",
        context_lines=60,
        max_workers=3,  # Adjust based on API rate limits
        ignore_patterns=[
            "node_modules/", ".venv/", "__pycache__/", 
            "dist/", "build/", ".git/", "package-lock.json",
            "yarn.lock", "pnpm-lock.yaml", "*.min.*"
        ]
    )
    
    print(f"šŸ“‹ Reviewing changes: {BASE_BRANCH}...{TARGET_BRANCH}")
    review_output = agent.review(REPO_PATH, BASE_BRANCH, TARGET_BRANCH)
    
    # Format and save the report
    report = agent.format_report(review_output)
    
    with open(OUTPUT_FILE, 'w', encoding='utf-8') as f:
        f.write(report)
    
    print(f"\nāœ… Review complete! Report saved to {OUTPUT_FILE}")
    print(f"   {review_output['total_findings']} findings "
          f"({review_output['critical_findings']} critical, "
          f"{review_output['high_findings']} high)")
    
    # Exit with non-zero code if critical issues found (for CI/CD integration)
    if review_output['critical_findings'] > 0:
        print("āš ļø  Critical issues detected — failing the check.")
        sys.exit(1)
    
    sys.exit(0)

if __name__ == "__main__":
    main()

CI/CD Integration

The agent is designed to run as part of your CI pipeline. Here's a GitHub Actions workflow example:

# .github/workflows/ai-code-review.yml
name: AI Code Review

on:
  pull_request:
    branches: [main, develop]

jobs:
  ai-review:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout repository
        uses: actions/checkout@v4
        with:
          fetch-depth: 0  # Fetch full git history for diffing
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: |
          pip install langchain langchain-openai gitpython python-dotenv pydantic
      
      - name: Run AI Code Review
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          BASE_BRANCH: ${{ github.event.pull_request.base.sha }}
          TARGET_BRANCH: ${{ github.event.pull_request.head.sha }}
          OUTPUT_FILE: review_report.md
        run: python main.py
      
      - name: Post review as PR comment
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const report = fs.readFileSync('review_report.md', 'utf8');
            await github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: report.substring(0, 60000)  // GitHub comment limit
            });

Advanced: Adding Custom Review Rules

You can extend the agent with programmatic rules that run alongside the LLM analysis. These rules catch deterministic issues that don't require AI judgment:

class CustomRuleChecker:
    """
    Programmatic rule checker that runs before/during the LLM review.
    These rules are deterministic, fast, and don't consume LLM tokens.
    """
    
    def __init__(self):
        self.rules = [
            self._check_hardcoded_secrets,
            self._check_console_log_statements,
            self._check_todo_comments,
            self._check_overly_long_lines,
        ]
    
    def _check_hardcoded_secrets(self, file_path: str, diff_lines: List[str]) -> List[ReviewFinding]:
        """Detect potential hardcoded secrets in changed lines."""
        findings = []
        secret_patterns = [
            'password', 'secret', 'api_key', 'apikey', 'token', 'auth',
            'credentials', 'private_key', 'client_secret'
        ]
        
        for i, line in enumerate(diff_lines):
            if line.startswith('+') and not line.startswith('+++'):
                line_lower = line.lower()
                for pattern in secret_patterns:
                    if pattern in line_lower and ('=' in line or ':' in line):
                        # Check it's not a comment
                        stripped = line.lstrip('+').strip()
                        if not stripped.startswith('//') and not stripped.startswith('#'):
                            findings.append(ReviewFinding(
                                severity='critical',
                                category='security',
                                line_number=None,  # Will be resolved later
                                file_path=file_path,
                                title=f"Potential hardcoded secret: {pattern}",
                                description=f"Line appears to contain a hardcoded '{pattern}'. "
                                           "Secrets should never be committed to version control.",
                                suggestion="Move this value to an environment variable or secret manager.",
                                code_snippet=line
                            ))
        
        return findings
    
    def _check_console_log_statements(self, file_path: str, diff_lines: List[str]) -> List[ReviewFinding]:
        """Flag leftover console.log/debug statements."""
        findings = []
        debug_statements = ['console.log', 'console.debug', 'print(', 'console.error(',
                           'logger.debug', 'console.warn(']
        
        for line in diff_lines:
            if line.startswith('+') and not line.startswith('+++'):
                for stmt in debug_statements:
                    if stmt in line:
                        stripped = line.lstrip('+').strip()
                        if not stripped.startswith('//'):
                            findings.append(ReviewFinding(
                                severity='low',
                                category='style',
                                file_path=file_path,
                                title=f"Debug statement found: {stmt}",
                                description="Debug/log statements should typically be removed "
                                           "before merging to production.",
                                suggestion="Remove the debug statement or replace with proper logging.",
                                code_snippet=line
                            ))
                            break
        
        return findings
    
    def _check_todo_comments(self, file_path: str, diff_lines: List[str]) -> List[ReviewFinding]:
        """Flag TODO/FIXME comments that may indicate incomplete work."""
        findings = []
        for line in diff_lines:
            if line.startswith('+') and not line.startswith('+++'):
                if 'TODO' in line or 'FIXME' in line or 'HACK' in line:
                    findings.append(ReviewFinding(
                        severity='medium',
                        category='documentation',
                        file_path=file_path,
                        title="TODO/FIXME comment detected",
                        description="This comment suggests incomplete work or a known issue.",
                        suggestion="Address the TODO before merging, or create a tracking ticket.",
                        code_snippet=line
                    ))
        return findings
    
    def _check_overly_long_lines(self, file_path: str, diff_lines: List[str]) -> List[ReviewFinding]:
        """Flag lines exceeding a reasonable length."""
        findings = []
        max_length = 120
        for line in diff_lines:
            if line.startswith('+') and not line.startswith('+++'):
                content = line[1:]  #

— Ad —

Google AdSense will appear here after approval

← Back to all articles