Building a Code Review Agent with LangGraph: Complete Guide

What is a Code Review Agent with LangGraph?

A Code Review Agent is an automated system built on top of Large Language Models (LLMs) that analyzes source code, identifies potential issues, and provides structured feedback — much like a human reviewer would. Built with LangGraph, this agent operates as a stateful, multi-step workflow where the LLM can iterate over code, consult different "reviewer personas," aggregate findings, and produce a comprehensive review report.

LangGraph, a framework from the LangChain ecosystem, allows developers to model complex LLM workflows as directed graphs — where nodes represent discrete operations (like "analyze code style," "check for security vulnerabilities," or "suggest performance improvements") and edges define the flow of information between them. Unlike simple prompt-response chains, a LangGraph-powered agent maintains state across steps, can branch conditionally, and can even loop back for deeper analysis when needed.

Why a LangGraph-Based Code Review Agent Matters

Traditional static analysis tools (like ESLint, Pylint, or SonarQube) catch syntax errors and rule-based patterns, but they miss nuanced issues: readability concerns, architectural flaws, logical errors, or security vulnerabilities that require semantic understanding. LLMs bridge this gap by reasoning about code contextually. However, a single-pass LLM call often produces shallow or inconsistent reviews.

LangGraph addresses this by enabling:

• Multi-perspective reviews: Assign different "expert" personas (security reviewer, performance reviewer, style reviewer) as separate graph nodes.
• Iterative refinement: The agent can revisit code after an initial pass, diving deeper into flagged areas.
• Structured output aggregation: Combine findings from multiple nodes into a unified, actionable report.
• Stateful context tracking: Carry context (file contents, review history, severity scores) across the entire workflow.
• Conditional routing: Skip certain review dimensions based on the code's language or complexity.

The result is a review agent that feels thorough, consistent, and adaptable — suitable for CI/CD pipelines, pre-commit hooks, or interactive developer tooling.

Architecture Overview

A typical LangGraph code review agent follows this graph structure:

• Node 1: Code Parser / Preprocessor — Ingests raw code, extracts metadata (language, file type, line counts), and prepares it for analysis.
• Node 2: Multi-Dimensional Review — Branches into parallel or sequential sub-reviews: Security, Performance, Style & Readability, Logic & Correctness, Documentation.
• Node 3: Severity Scoring & Deduplication — Assigns severity levels (Critical, Warning, Suggestion) and merges overlapping findings.
• Node 4: Report Formatter — Produces the final structured output (JSON, Markdown, or inline comments).
• Node 5 (optional): Human-in-the-loop — Pauses for manual approval or override before posting results.

The state object flows through every node, accumulating findings and metadata at each step.

Complete Implementation: Step-by-Step

Step 1: Project Setup and Dependencies

Install the required packages. You'll need LangGraph, LangChain, and an LLM provider (OpenAI, Anthropic, or an open-source model via Ollama).

pip install langgraph langchain langchain-openai langchain-community
pip install openai  # or anthropic, or install ollama for local models

Create a new Python file code_review_agent.py and set up your imports:

import os
import json
from typing import TypedDict, List, Optional, Annotated
from operator import add

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate

# Set your API key
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

Step 2: Define the State Schema

The state object carries all data through the graph. Define it using TypedDict with annotations for fields that should be accumulated across nodes (like lists of findings).

class CodeReviewState(TypedDict):
    # Input fields
    source_code: str
    file_path: str
    language: str
    
    # Accumulated review findings (merged across nodes)
    findings: Annotated[List[dict], add]
    
    # Intermediate analysis
    parsed_structure: Optional[dict]
    
    # Final report
    final_report: Optional[str]
    
    # Control flow
    review_dimensions: List[str]
    current_dimension: Optional[str]

The Annotated[List[dict], add] syntax tells LangGraph to concatenate lists when multiple nodes return findings, rather than overwriting them. This is crucial for aggregating results from parallel review branches.

Step 3: Create the Preprocessor Node

This node normalizes the input code and extracts basic metadata. It also decides which review dimensions to activate based on the language.

def preprocess_code(state: CodeReviewState) -> CodeReviewState:
    """
    Parse the source code, detect language characteristics,
    and determine which review dimensions to apply.
    """
    code = state["source_code"]
    language = state.get("language", "unknown").lower()
    
    # Basic structure analysis (line count, imports detection, etc.)
    lines = code.split("\n")
    line_count = len(lines)
    
    # Detect if there are imports, classes, functions
    has_imports = any("import " in line or "require(" in line or "from " in line for line in lines[:50])
    has_classes = any("class " in line for line in lines)
    has_functions = any("def " in line or "function " in line for line in lines)
    
    parsed = {
        "line_count": line_count,
        "has_imports": has_imports,
        "has_classes": has_classes,
        "has_functions": has_functions,
        "language": language,
    }
    
    # Determine review dimensions based on language and structure
    dimensions = ["style", "logic"]
    
    if language in ("python", "javascript", "typescript", "java", "go", "rust"):
        dimensions.append("security")
    if language in ("python", "javascript", "typescript", "java", "cpp"):
        dimensions.append("performance")
    if has_functions or has_classes:
        dimensions.append("documentation")
    
    return {
        **state,
        "parsed_structure": parsed,
        "review_dimensions": dimensions,
        "findings": [],  # Initialize empty findings list
    }

Step 4: Build the Review Dimension Nodes

Each review dimension gets its own node with a specialized prompt. These nodes can be called in sequence or in parallel depending on your graph topology. Here's a reusable factory function that creates dimension-specific reviewer nodes:

def create_review_node(dimension: str, llm: ChatOpenAI):
    """
    Factory function that creates a review node for a specific dimension.
    Each node uses a specialized system prompt tailored to its review focus.
    """
    
    # Dimension-specific prompts
    prompts = {
        "security": """You are an expert security reviewer. Analyze the code for:
1. SQL injection, XSS, command injection vulnerabilities
2. Hardcoded secrets or API keys
3. Insecure cryptographic practices (weak algorithms, improper key management)
4. Path traversal risks, unsafe file operations
5. Authentication/authorization bypass risks
6. Unsafe deserialization

For each issue found, output a JSON object:
{
  "dimension": "security",
  "severity": "critical|warning|suggestion",
  "line_range": [start_line, end_line],
  "title": "Brief issue title",
  "description": "Detailed explanation",
  "suggestion": "How to fix it"
}
Output ONLY a valid JSON array. If no issues, output [].""",
        
        "performance": """You are an expert performance reviewer. Analyze the code for:
1. Inefficient algorithms (O(n²) where O(n log n) is possible)
2. Unnecessary memory allocations
3. Blocking I/O in hot paths
4. Missing caching opportunities
5. Redundant computations in loops
6. Large synchronous operations that could be async

Output your findings as a JSON array using the same schema with dimension="performance".
Output ONLY a valid JSON array. If no issues, output [].""",
        
        "style": """You are an expert code style reviewer. Analyze for:
1. Naming convention violations (snake_case vs camelCase based on language)
2. Overly long functions (>50 lines) that should be split
3. Deep nesting (more than 3 levels)
4. Magic numbers without named constants
5. Inconsistent spacing or formatting
6. Missing or excessive comments
7. Unclear variable names

Output your findings as a JSON array with dimension="style".
Output ONLY a valid JSON array. If no issues, output [].""",
        
        "logic": """You are an expert logic and correctness reviewer. Analyze for:
1. Off-by-one errors in loops
2. Potential null/undefined reference errors
3. Incorrect boolean logic (wrong AND/OR combinations)
4. Unhandled edge cases (empty inputs, boundary values)
5. Race conditions in sequential code
6. Incorrect error handling patterns
7. Type coercion issues

Output your findings as a JSON array with dimension="logic".
Output ONLY a valid JSON array. If no issues, output [].""",
        
        "documentation": """You are an expert documentation reviewer. Analyze for:
1. Missing docstrings on public functions/classes
2. Parameter descriptions that don't match actual usage
3. Outdated or misleading comments
4. Missing README or module-level documentation
5. Complex logic without explanatory comments
6. API documentation gaps

Output your findings as a JSON array with dimension="documentation".
Output ONLY a valid JSON array. If no issues, output []."""
    }
    
    prompt = prompts.get(dimension, prompts["style"])
    
    def review_node(state: CodeReviewState) -> CodeReviewState:
        code = state["source_code"]
        lang = state.get("language", "unknown")
        
        messages = [
            HumanMessage(content=f"""Review the following {lang} code for {dimension} issues.

Source code from file `{state.get('file_path', 'unknown')}`:

{lang}
{code}
{prompt}""")
        ]
        
        response = llm.invoke(messages)
        response_text = response.content
        
        # Extract JSON array from response (handle markdown code blocks)
        if "json" in response_text:
            start = response_text.find("json") + 7
            end = response_text.find("", start)
            json_str = response_text[start:end].strip()
        elif "" in response_text:
            start = response_text.find("") + 3
            end = response_text.find("", start)
            json_str = response_text[start:end].strip()
        else:
            # Try to find JSON array directly
            start = response_text.find("[")
            end = response_text.rfind("]") + 1
            json_str = response_text[start:end] if start >= 0 else "[]"
        
        try:
            findings = json.loads(json_str)
            if not isinstance(findings, list):
                findings = []
        except json.JSONDecodeError:
            findings = []
        
        # Add metadata to each finding
        for finding in findings:
            finding["source_dimension"] = dimension
            finding["file_path"] = state.get("file_path", "unknown")
        
        return {"findings": findings}
    
    return review_node

Step 5: Severity Scoring and Deduplication Node

After all review dimensions have reported their findings, we need to score them, resolve duplicates (where two reviewers flagged the same issue), and sort by severity.

def score_and_deduplicate(state: CodeReviewState) -> CodeReviewState:
    """
    Process all accumulated findings: assign severity scores,
    deduplicate overlapping issues, and sort by priority.
    """
    findings = state.get("findings", [])
    
    if not findings:
        return {**state, "findings": [], "final_report": "No issues found."}
    
    # Severity weight mapping
    severity_weights = {
        "critical": 100,
        "warning": 50,
        "suggestion": 10,
    }
    
    # Deduplicate: if two findings have overlapping line ranges and similar titles, keep the one with higher severity
    deduplicated = []
    seen_signatures = set()
    
    for finding in sorted(findings, key=lambda f: severity_weights.get(f.get("severity", "suggestion"), 0), reverse=True):
        line_range = finding.get("line_range", [0, 0])
        title = finding.get("title", "").lower().strip()
        
        # Create a signature: dimension + line overlap + title keywords
        sig_key = f"{finding.get('dimension')}:{line_range[0]}-{line_range[1]}:{title[:30]}"
        
        if sig_key not in seen_signatures:
            seen_signatures.add(sig_key)
            deduplicated.append(finding)
    
    # Sort: critical first, then warnings, then suggestions
    deduplicated.sort(key=lambda f: severity_weights.get(f.get("severity", "suggestion"), 0), reverse=True)
    
    return {
        **state,
        "findings": deduplicated,
    }

Step 6: Report Generation Node

Format the final review into a clean, actionable report — either Markdown for human readability or JSON for CI/CD integration.

def generate_report(state: CodeReviewState) -> CodeReviewState:
    """
    Generate a formatted review report from all findings.
    Produces both a Markdown summary and a structured JSON report.
    """
    findings = state.get("findings", [])
    file_path = state.get("file_path", "unknown")
    language = state.get("language", "unknown")
    parsed = state.get("parsed_structure", {})
    
    if not findings:
        report = f"""## Code Review Report: `{file_path}`

**Language:** {language} | **Lines:** {parsed.get('line_count', 'N/A')}

✅ **No issues found.** The code appears clean across all reviewed dimensions.

*Review dimensions checked:* {', '.join(state.get('review_dimensions', []))}
"""
        return {**state, "final_report": report}
    
    # Group findings by severity
    critical = [f for f in findings if f.get("severity") == "critical"]
    warnings = [f for f in findings if f.get("severity") == "warning"]
    suggestions = [f for f in findings if f.get("severity") == "suggestion"]
    
    report = f"""## Code Review Report: `{file_path}`

**Language:** {language} | **Lines:** {parsed.get('line_count', 'N/A')}

### Summary
- 🔴 **{len(critical)} Critical** issues
- 🟡 **{len(warnings)} Warnings**
- 🔵 **{len(suggestions)} Suggestions**
- 📋 **{len(findings)} Total findings** across {len(state.get('review_dimensions', []))} dimensions

---
"""
    
    if critical:
        report += "\n### 🔴 Critical Issues\n\n"
        for i, finding in enumerate(critical, 1):
            lines = finding.get("line_range", [0, 0])
            report += f"""**{i}. [{finding.get('dimension', 'unknown').upper()}] {finding.get('title', 'Untitled')}**
- **Lines:** {lines[0]}-{lines[1]}
- **Description:** {finding.get('description', 'N/A')}
- **Suggested Fix:** {finding.get('suggestion', 'N/A')}

"""
    
    if warnings:
        report += "\n### 🟡 Warnings\n\n"
        for i, finding in enumerate(warnings, 1):
            lines = finding.get("line_range", [0, 0])
            report += f"""**{i}. [{finding.get('dimension', 'unknown').upper()}] {finding.get('title', 'Untitled')}**
- **Lines:** {lines[0]}-{lines[1]}
- **Description:** {finding.get('description', 'N/A')}
- **Suggested Fix:** {finding.get('suggestion', 'N/A')}

"""
    
    if suggestions:
        report += "\n### 🔵 Suggestions\n\n"
        for i, finding in enumerate(suggestions, 1):
            lines = finding.get("line_range", [0, 0])
            report += f"""**{i}. [{finding.get('dimension', 'unknown').upper()}] {finding.get('title', 'Untitled')}**
- **Lines:** {lines[0]}-{lines[1]}
- **Description:** {finding.get('description', 'N/A')}
- **Suggested Fix:** {finding.get('suggestion', 'N/A')}

"""
    
    report += "\n---\n*Report generated by LangGraph Code Review Agent*"
    
    return {**state, "final_report": report}

Step 7: Assemble the Graph

Now we wire everything together in a LangGraph StateGraph. This is where we define the topology — the order and conditions under which nodes execute.

def build_code_review_graph(llm: ChatOpenAI):
    """
    Construct the full LangGraph workflow for code review.
    """
    graph = StateGraph(CodeReviewState)
    
    # Add nodes
    graph.add_node("preprocess", preprocess_code)
    
    # Add review dimension nodes (created dynamically)
    dimensions = ["security", "performance", "style", "logic", "documentation"]
    for dim in dimensions:
        graph.add_node(f"review_{dim}", create_review_node(dim, llm))
    
    graph.add_node("score_deduplicate", score_and_deduplicate)
    graph.add_node("generate_report", generate_report)
    
    # Define edges: preprocess -> all review dimensions in parallel -> score -> report
    graph.set_entry_point("preprocess")
    
    # After preprocessing, route to all applicable review dimensions
    # We use a conditional edge that dynamically routes based on review_dimensions
    def route_to_reviews(state: CodeReviewState) -> List[str]:
        dims = state.get("review_dimensions", ["style", "logic"])
        return [f"review_{dim}" for dim in dims]
    
    graph.add_conditional_edges(
        "preprocess",
        route_to_reviews,
        {dim: f"review_{dim}" for dim in dimensions}
    )
    
    # All review nodes converge to score_deduplicate
    for dim in dimensions:
        graph.add_edge(f"review_{dim}", "score_deduplicate")
    
    # Score -> Report -> End
    graph.add_edge("score_deduplicate", "generate_report")
    graph.add_edge("generate_report", END)
    
    return graph.compile()

Important: The conditional edge from preprocess dynamically branches only to the review dimensions that are relevant for the detected language. All active review nodes run in parallel (LangGraph handles this automatically when multiple edges originate from the same node) and their findings are merged via the add annotation on the findings field.

Step 8: Run the Agent

Here's how to invoke the compiled graph with actual source code:

def review_code(source_code: str, file_path: str = "unknown.py", language: str = "python") -> str:
    """
    Main entry point: run the full code review agent on a piece of code.
    Returns the formatted review report.
    """
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    app = build_code_review_graph(llm)
    
    initial_state: CodeReviewState = {
        "source_code": source_code,
        "file_path": file_path,
        "language": language,
        "findings": [],
        "parsed_structure": None,
        "final_report": None,
        "review_dimensions": [],
        "current_dimension": None,
    }
    
    result = app.invoke(initial_state)
    return result.get("final_report", "Error: Report generation failed.")


# Example usage
if __name__ == "__main__":
    sample_code = """
def calculate_total(items):
    total = 0
    for i in range(len(items)):
        total = total + items[i]
    return total

def process_data(data):
    password = "hardcoded_secret_123"
    result = []
    for i in range(len(data)):
        if data[i] > 0:
            result.append(data[i] * 2)
    return result
"""
    
    report = review_code(sample_code, "utils.py", "python")
    print(report)

Step 9: Adding a Human-in-the-Loop Node (Optional)

For critical workflows (like blocking a PR merge), you can insert an interrupt node that pauses execution and waits for human approval:

def human_approval(state: CodeReviewState) -> CodeReviewState:
    """
    Interrupt node: presents findings to a human reviewer for approval or override.
    In production, this would integrate with Slack, a web UI, or a CI bot.
    """
    critical_count = sum(1 for f in state.get("findings", []) if f.get("severity") == "critical")
    
    if critical_count > 0:
        # In a real system, this would pause and wait for input
        print(f"\n⚠️  {critical_count} critical issues found. Awaiting human approval...")
        # For demo purposes, auto-approve with a flag
        approved = True  # Replace with actual interrupt handling
        if not approved:
            # Override: downgrade severity or dismiss findings
            pass
    
    return state

# Add to graph:
# graph.add_node("human_approval", human_approval)
# graph.add_edge("score_deduplicate", "human_approval")
# graph.add_edge("human_approval", "generate_report")

For production interrupt handling, use LangGraph's interrupt API or integrate with checkpointer to persist state while waiting for external input.

Complete Graph Visualization

Here's the full node-edge topology in ASCII form:

                 ┌─────────────┐
                 │  preprocess │  (entry point)
                 └──────┬──────┘
                        │
            ┌───────────┼───────────┐
            │           │           │
            ▼           ▼           ▼
    ┌───────────┐ ┌───────────┐ ┌───────────┐
    │review_sec │ │review_perf│ │review_style│  ... (dynamic based on language)
    └─────┬─────┘ └─────┬─────┘ └─────┬─────┘
          │             │             │
          └─────────────┼─────────────┘
                        │
                        ▼
            ┌───────────────────────┐
            │  score_deduplicate    │  (merge & sort findings)
            └───────────┬───────────┘
                        │
                        ▼
            ┌───────────────────────┐
            │   generate_report     │  (final formatted output)
            └───────────┬───────────┘
                        │
                        ▼
                      END

Best Practices for Building Code Review Agents

1. Choose the Right LLM and Temperature

For code review, use a model with strong coding capabilities (GPT-4o, Claude 3.5 Sonnet, or fine-tuned open-source models like CodeLlama). Set temperature low (0.0–0.2) to get consistent, deterministic findings. High temperatures cause hallucinated issues and inconsistent severity ratings.

llm = ChatOpenAI(model="gpt-4o", temperature=0.1, max_tokens=2000)

2. Use Structured Output Formats

Always prompt the LLM to return structured JSON. This prevents parsing errors and makes downstream processing reliable. Consider using LangChain's with_structured_output method with Pydantic models for even stronger guarantees:

from pydantic import BaseModel
from typing import Literal

class Finding(BaseModel):
    dimension: str
    severity: Literal["critical", "warning", "suggestion"]
    line_range: tuple[int, int]
    title: str
    description: str
    suggestion: str

# llm_with_structure = llm.with_structured_output(Finding)

3. Implement Smart Deduplication

Multiple reviewer dimensions often flag the same code block. Use line-range overlap detection and title similarity (cosine similarity or fuzzy matching) to merge duplicates. Always keep the finding with the highest severity when merging.

4. Add Language-Aware Routing

Not all review dimensions apply to all languages. Security review for shell scripts differs vastly from Python. Build a language-to-dimensions mapping and use conditional edges to skip irrelevant checks. This saves tokens and reduces noise.

5. Cache and Incremental Reviews

For PR workflows, only review changed files (the diff). LangGraph's checkpointer can persist state between runs, enabling incremental reviews where unchanged files retain their previous review results. Use a thread ID per PR:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Run with thread_id for state persistence
result = app.invoke(state, config={"configurable": {"thread_id": "pr-42"}})

6. Set Token Budgets and Timeouts

Large files can exhaust context windows. Implement a preprocessor that splits files exceeding ~4000 tokens into chunks, reviews each chunk separately, then aggregates. Set per-node timeouts to prevent stalled reviews:

# In production, wrap LLM calls with timeout
import signal
from contextlib import contextmanager

@contextmanager
def timeout(seconds: int):
    signal.signal(signal.SIGALRM, lambda sig, frame: (_ for _ in ()).throw(TimeoutError))
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)

7. Ground Findings with Line Numbers

Always ask the LLM to cite specific line numbers or ranges. This makes findings actionable and allows automated annotation in PR interfaces (like GitHub's inline comments). Validate that cited line numbers exist in the actual file to catch hallucinations.

8. Build a Feedback Loop

Track which findings developers accept or dismiss. Use this data to fine-tune prompts, adjust severity thresholds, or train a classifier that filters low-value suggestions. LangGraph's persistence makes this straightforward — store feedback in the state and use it in subsequent runs.

9. Handle Rate Limits and Retries Gracefully

When using cloud LLM APIs, implement exponential backoff for rate limits. LangChain's built-in retry logic helps, but for LangGraph workflows, wrap individual node LLM calls with retry decorators:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=30))
def invoke_llm_with_retry(llm, messages):
    return llm.invoke(messages)

Advanced: Parallel Execution with Fan-Out/Fan-In

LangGraph automatically parallelizes independent branches. When preprocess routes to review_security, review_performance, and review_style simultaneously, all three LLM calls execute concurrently. The add annotation on findings handles the merge (fan-in) at score_deduplicate. This dramatically reduces end-to-end latency compared to sequential review chains.

To explicitly control parallelism, use the Send API in LangGraph for dynamic fan-out:

from langgraph.types import Send

def continue_to_reviews(state: CodeReviewState):
    dims = state.get("review_dimensions", [])
    # Send each review as a separate parallel branch
    return [Send(f"review_{dim}", {}) for dim in dims]

Conclusion

Building a Code Review Agent with LangGraph transforms what could be a simple LLM prompt into a robust, multi-dimensional, stateful review pipeline. By decomposing code review into specialized analysis nodes — security, performance, style, logic, documentation — and orchestrating them through a directed graph, you achieve reviews that are more thorough, consistent, and actionable than any single-pass LLM call could produce. LangGraph's state management, conditional routing, parallel execution, and persistence capabilities make it the ideal framework for this task. The architecture outlined here can be extended with custom review dimensions, integrated into CI/CD systems, enhanced with human-in-the-loop approvals, and continuously improved through feedback loops. Start with the core graph structure provided above, adapt the prompts to your team's coding standards, and deploy a review agent that truly elevates your code quality process.