Building a Code Review Agent with LangGraph: Step-by-Step Tutorial

Building a Code Review Agent with LangGraph: Step-by-Step Tutorial

What Is a Code Review Agent?

A code review agent is an automated system that analyzes source code submissions and provides structured feedback — much like a human reviewer would. Powered by large language models (LLMs), such agents can examine code for style violations, potential bugs, security vulnerabilities, performance issues, and logical errors. When built with LangGraph, these agents gain the ability to orchestrate complex, multi-step review workflows where different specialized "reviewers" collaborate, results are aggregated, and the process can branch or loop based on findings.

LangGraph is a framework from the LangChain ecosystem designed for building stateful, multi-actor applications with LLMs. It lets you define workflows as directed graphs — nodes represent discrete steps (like running a security scan or checking style), and edges represent transitions between steps. State flows through the graph, accumulating results as the agent progresses. This graph-based approach is ideal for code review because real code review is inherently multi-dimensional: you need to check multiple concern areas, sometimes revisit files based on earlier findings, and produce a consolidated report.

Why It Matters

Manual code review is time-consuming and inconsistent. Senior developers spend hours reviewing pull requests, and fatigue leads to missed issues. An LLM-powered code review agent offers several compelling benefits:

• Consistency — The agent applies the same rules and scrutiny to every review, eliminating human variance.
• Speed — Reviews complete in seconds rather than hours, dramatically shortening the feedback loop.
• Comprehensiveness — Multiple specialized checkers (security, style, logic, performance) can run in parallel or sequence, covering areas a single human might overlook.
• Actionable output — The agent produces structured reports with severity levels, file references, and fix suggestions that integrate directly into CI/CD pipelines.
• Scalability — As your team grows and commit frequency increases, the agent scales effortlessly where human reviewers would become a bottleneck.

With LangGraph specifically, you gain the ability to model complex review policies. For example, you might route code through different reviewers based on language or file type, escalate high-severity findings to a second-pass analysis, or loop back for re-review after automated fixes are applied. This flexibility goes far beyond a simple "ask the LLM for feedback" script.

Prerequisites

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

• Python 3.10+
• A LangChain-compatible LLM API key (OpenAI, Anthropic, or a local model via Ollama)
• The required packages installed via pip

pip install langgraph langchain langchain-openai

For this tutorial, we'll use OpenAI's GPT-4o, but you can substitute any LangChain-compatible model. Make sure your API key is set as an environment variable:

export OPENAI_API_KEY="your-api-key-here"

Core Concepts of LangGraph

Before diving into code, let's quickly review the three building blocks of LangGraph:

• State — A typed dictionary (or Pydantic model) that holds all data flowing through the graph. For our code review agent, the state will carry the source code, review findings, and final report.
• Nodes — Python functions (synchronous or asynchronous) that read the current state, perform work, and return an updated state. Each specialized reviewer is a node.
• Edges — Connections between nodes. These can be normal edges (always go from A to B), conditional edges (route based on state values), or entry/exit points that define where the graph starts and ends.

The graph compiles into a runnable object. You invoke it with an initial state, and it executes nodes in the defined order until it reaches the exit point, returning the final state.

Step 1: Define the State Schema

The state is the backbone of our agent. It must carry the input code, accumulate findings from each reviewer, and store the final aggregated report. We'll use a Python TypedDict for simplicity:

from typing import TypedDict, List, Optional, Annotated
from langgraph.graph.message import add_messages
import operator

class CodeReviewState(TypedDict):
    # The source code to review (as a string)
    source_code: str
    # Optional: programming language for context-aware review
    language: str
    # Findings from individual reviewers, accumulated as a list
    findings: Annotated[List[str], operator.add]
    # Structured review report (final output)
    final_report: Optional[str]
    # Flag to track if critical issues were found
    has_critical_issues: bool
    # Number of review iterations (for loop control)
    iteration_count: int

Note the Annotated type for findings. By using operator.add, we tell LangGraph that when multiple nodes return updates to findings, the lists should be concatenated rather than overwritten. This is crucial for accumulating results from parallel or sequential reviewers.

Step 2: Create the Reviewer Nodes

Each node is a function that receives the current state, performs an LLM-based analysis of a specific concern area, and returns a partial state update. We'll build four specialized reviewers: Style & Convention, Security, Logic & Bugs, and Performance.

We'll use LangChain's chat model interface for consistency. Each reviewer function constructs a system prompt tailored to its domain, sends the code to the LLM, and parses the response into a structured finding.

2a. Style & Convention Reviewer

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0.2)

def style_reviewer(state: CodeReviewState) -> dict:
    """Review code for style issues: naming, formatting, conventions."""
    system_prompt = SystemMessage(content="""You are an expert code style reviewer. 
    Analyze the provided code for:
    - Naming conventions (camelCase, snake_case, PascalCase as appropriate)
    - Indentation and formatting consistency
    - Line length and readability
    - Comment quality and placement
    - Adherence to language-specific style guides (PEP 8 for Python, etc.)
    
    For each issue found, provide:
    - File/line reference (if detectable from context)
    - Severity: LOW, MEDIUM, or HIGH
    - A clear description of the issue
    - A suggested fix
    
    Format each finding as a single structured line:
    [STYLE] {severity}: {description} | Suggestion: {fix}
    
    If no issues are found, respond with exactly: NO_STYLE_ISSUES_FOUND""")
    
    user_message = HumanMessage(content=f"""Review the following {state.get('language', 'code')} code for style issues:
    
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    if result == "NO_STYLE_ISSUES_FOUND":
        return {"findings": []}
    
    findings = [line.strip() for line in result.split("\n") if line.strip() and "[STYLE]" in line]
    return {"findings": findings}

2b. Security Reviewer

def security_reviewer(state: CodeReviewState) -> dict:
    """Review code for security vulnerabilities."""
    system_prompt = SystemMessage(content="""You are an expert application security reviewer.
    Analyze the provided code for:
    - Injection vulnerabilities (SQL, command, code injection)
    - Hardcoded secrets, API keys, or credentials
    - Insecure cryptographic practices
    - Missing input validation or sanitization
    - Authentication and authorization flaws
    - Path traversal risks
    - Unsafe deserialization
    
    For each vulnerability found, provide:
    - Severity: LOW, MEDIUM, HIGH, CRITICAL
    - A clear description with the vulnerable code pattern
    - A concrete fix recommendation
    
    Format each finding as:
    [SECURITY] {severity}: {description} | Fix: {recommendation}
    
    If no vulnerabilities are found, respond with exactly: NO_SECURITY_ISSUES_FOUND""")
    
    user_message = HumanMessage(content=f"""Review this code for security vulnerabilities:
    
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    if result == "NO_SECURITY_ISSUES_FOUND":
        return {"findings": [], "has_critical_issues": False}
    
    findings = [line.strip() for line in result.split("\n") if line.strip() and "[SECURITY]" in line]
    has_critical = any("CRITICAL" in f for f in findings)
    return {"findings": findings, "has_critical_issues": has_critical}

2c. Logic & Bug Reviewer

def logic_reviewer(state: CodeReviewState) -> dict:
    """Review code for logical errors and potential bugs."""
    system_prompt = SystemMessage(content="""You are an expert code logic reviewer.
    Analyze the provided code for:
    - Off-by-one errors and boundary condition bugs
    - Incorrect boolean logic or conditionals
    - Null/None reference risks
    - Type mismatches or unsafe type coercion
    - Resource leaks (unclosed files, connections, etc.)
    - Race conditions in concurrent code
    - Incorrect error handling or swallowed exceptions
    
    Format each finding as:
    [LOGIC] {severity}: {description} | Fix: {recommendation}
    
    Severity levels: LOW, MEDIUM, HIGH, CRITICAL
    If no issues found, respond with: NO_LOGIC_ISSUES_FOUND""")
    
    user_message = HumanMessage(content=f"""Review this code for logical errors and bugs:
    
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    if result == "NO_LOGIC_ISSUES_FOUND":
        return {"findings": [], "has_critical_issues": False}
    
    findings = [line.strip() for line in result.split("\n") if line.strip() and "[LOGIC]" in line]
    has_critical = any("CRITICAL" in f for f in findings)
    return {"findings": findings, "has_critical_issues": has_critical}

2d. Performance Reviewer

def performance_reviewer(state: CodeReviewState) -> dict:
    """Review code for performance issues."""
    system_prompt = SystemMessage(content="""You are an expert performance engineer.
    Analyze the provided code for:
    - Inefficient algorithms or data structures
    - Unnecessary loops or redundant computations
    - Memory allocation inefficiencies
    - Missing caching opportunities
    - N+1 query patterns (if database code)
    - Blocking operations that could be async
    - Excessive object creation
    
    Format each finding as:
    [PERFORMANCE] {severity}: {description} | Improvement: {suggestion}
    
    Severity: LOW, MEDIUM, HIGH
    If no issues found, respond with: NO_PERFORMANCE_ISSUES_FOUND""")
    
    user_message = HumanMessage(content=f"""Review this code for performance issues:
    
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    if result == "NO_PERFORMANCE_ISSUES_FOUND":
        return {"findings": []}
    
    findings = [line.strip() for line in result.split("\n") if line.strip() and "[PERFORMANCE]" in line]
    return {"findings": findings}

Step 3: Build the Report Aggregator Node

After all reviewers have contributed their findings, we need a node that synthesizes everything into a polished, actionable report. This node also handles cases where no issues were found anywhere:

def report_aggregator(state: CodeReviewState) -> dict:
    """Aggregate all findings into a structured final report."""
    findings = state.get("findings", [])
    
    if not findings:
        return {
            "final_report": "✅ **Code Review Complete** — No issues were found across all review dimensions (Style, Security, Logic, Performance). The code looks clean!",
            "iteration_count": state.get("iteration_count", 0) + 1
        }
    
    # Sort findings by severity priority
    severity_order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
    
    def severity_key(finding: str) -> int:
        for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW"]:
            if sev in finding:
                return severity_order[sev]
        return 99
    
    sorted_findings = sorted(findings, key=severity_key)
    
    # Build the report
    report_lines = ["# 📋 Automated Code Review Report", ""]
    report_lines.append(f"**Language:** {state.get('language', 'Unknown')}")
    report_lines.append(f"**Total Issues Found:** {len(sorted_findings)}")
    report_lines.append("")
    
    # Group by category
    categories = {}
    for finding in sorted_findings:
        for cat in ["STYLE", "SECURITY", "LOGIC", "PERFORMANCE"]:
            if f"[{cat}]" in finding:
                categories.setdefault(cat, []).append(finding)
                break
    
    category_labels = {
        "STYLE": "🎨 Style & Convention",
        "SECURITY": "🔒 Security",
        "LOGIC": "🐛 Logic & Potential Bugs",
        "PERFORMANCE": "⚡ Performance"
    }
    
    for cat_key in ["SECURITY", "LOGIC", "STYLE", "PERFORMANCE"]:
        if cat_key in categories:
            report_lines.append(f"## {category_labels[cat_key]}")
            report_lines.append("")
            for i, finding in enumerate(categories[cat_key], 1):
                # Extract just the description part after the tag
                clean = finding.replace(f"[{cat_key}] ", f"{i}. ")
                report_lines.append(clean)
            report_lines.append("")
    
    # Add summary
    critical_count = sum(1 for f in findings if "CRITICAL" in f)
    high_count = sum(1 for f in findings if "HIGH" in f)
    
    if critical_count > 0:
        report_lines.append(f"## ⚠️ Action Required")
        report_lines.append(f"**{critical_count} CRITICAL** issues need immediate attention before deployment.")
    
    report_lines.append("")
    report_lines.append("---")
    report_lines.append("*Report generated by LangGraph Code Review Agent*")
    
    final_report = "\n".join(report_lines)
    
    return {
        "final_report": final_report,
        "iteration_count": state.get("iteration_count", 0) + 1,
        "has_critical_issues": state.get("has_critical_issues", False) or critical_count > 0
    }

Step 4: Define the Graph Structure

Now we assemble the nodes and edges into a StateGraph. The key architectural decision is: should reviewers run in parallel or sequentially? For code review, parallel execution is ideal — each reviewer operates independently on the same source code, and their findings are merged automatically via the Annotated list. LangGraph handles this elegantly by fanning out from a single node to multiple nodes, then converging back.

Here's how to build the graph:

from langgraph.graph import StateGraph, START, END

# Initialize the graph with our state type
builder = StateGraph(CodeReviewState)

# Add all nodes
builder.add_node("style_reviewer", style_reviewer)
builder.add_node("security_reviewer", security_reviewer)
builder.add_node("logic_reviewer", logic_reviewer)
builder.add_node("performance_reviewer", performance_reviewer)
builder.add_node("report_aggregator", report_aggregator)

# Define the flow:
# START -> all four reviewers (parallel fan-out)
# Each reviewer -> report_aggregator (fan-in convergence)
# report_aggregator -> END

# Entry point: from START, go to all reviewers
builder.add_edge(START, "style_reviewer")
builder.add_edge(START, "security_reviewer")
builder.add_edge(START, "logic_reviewer")
builder.add_edge(START, "performance_reviewer")

# All reviewers converge to the aggregator
builder.add_edge("style_reviewer", "report_aggregator")
builder.add_edge("security_reviewer", "report_aggregator")
builder.add_edge("logic_reviewer", "report_aggregator")
builder.add_edge("performance_reviewer", "report_aggregator")

# Aggregator completes the workflow
builder.add_edge("report_aggregator", END)

# Compile the graph
graph = builder.compile()

This creates a simple but powerful fan-out/fan-in pattern. When invoked, LangGraph sends the initial state to all four reviewers simultaneously. Each returns its findings independently. LangGraph waits for all four to complete, merges the findings lists (thanks to the operator.add annotation), and then invokes the aggregator with the combined findings.

Step 5: Add Conditional Routing for Critical Issues

The basic graph works well, but real code review often requires escalation. Let's add a conditional edge: if critical security or logic issues are found, route to a deeper "second-pass" analysis node before finalizing the report. This demonstrates LangGraph's powerful conditional routing:

def deep_security_analysis(state: CodeReviewState) -> dict:
    """Perform a deeper security analysis when critical issues are flagged."""
    system_prompt = SystemMessage(content="""You are a senior application security specialist.
    A previous review flagged CRITICAL security issues. Perform a deeper analysis:
    - Trace the vulnerable code path end-to-end
    - Assess exploitability and potential impact
    - Provide a detailed remediation plan with code examples
    - Check if the vulnerability pattern appears elsewhere in the codebase
    
    Format findings as:
    [DEEP-SECURITY] {severity}: {detailed_analysis} | Remediation: {steps}""")
    
    user_message = HumanMessage(content=f"""Deep security analysis requested. Original findings:
    {chr(10).join(state.get('findings', []))}
    
    Source code:
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    deep_findings = [line.strip() for line in result.split("\n") if line.strip() and "[DEEP-SECURITY]" in line]
    return {"findings": deep_findings}

def should_escalate(state: CodeReviewState) -> str:
    """Conditional routing: escalate if critical issues exist and we haven't already."""
    if state.get("has_critical_issues", False) and state.get("iteration_count", 0) < 1:
        return "deep_security_analysis"
    return "report_aggregator"

# Build the enhanced graph
builder_v2 = StateGraph(CodeReviewState)

# Add all nodes (including the new deep analysis node)
builder_v2.add_node("style_reviewer", style_reviewer)
builder_v2.add_node("security_reviewer", security_reviewer)
builder_v2.add_node("logic_reviewer", logic_reviewer)
builder_v2.add_node("performance_reviewer", performance_reviewer)
builder_v2.add_node("deep_security_analysis", deep_security_analysis)
builder_v2.add_node("report_aggregator", report_aggregator)

# Fan-out from START to reviewers
builder_v2.add_edge(START, "style_reviewer")
builder_v2.add_edge(START, "security_reviewer")
builder_v2.add_edge(START, "logic_reviewer")
builder_v2.add_edge(START, "performance_reviewer")

# Instead of all going directly to aggregator, we route through a conditional
# after security and logic reviewers complete (they set has_critical_issues)
builder_v2.add_edge("style_reviewer", "report_aggregator")
builder_v2.add_edge("performance_reviewer", "report_aggregator")

# Security and logic reviewers go to a conditional check
builder_v2.add_conditional_edges(
    "security_reviewer",
    should_escalate,
    {
        "deep_security_analysis": "deep_security_analysis",
        "report_aggregator": "report_aggregator"
    }
)
builder_v2.add_conditional_edges(
    "logic_reviewer",
    should_escalate,
    {
        "deep_security_analysis": "deep_security_analysis",
        "report_aggregator": "report_aggregator"
    }
)

# Deep analysis always goes to aggregator afterward
builder_v2.add_edge("deep_security_analysis", "report_aggregator")
builder_v2.add_edge("report_aggregator", END)

graph_v2 = builder_v2.compile()

The conditional function should_escalate checks the state for critical issues and whether we've already performed a deep analysis (tracked via iteration_count). If conditions are met, the graph routes to deep_security_analysis; otherwise, it proceeds directly to the aggregator. This prevents infinite loops while enabling intelligent escalation.

Step 6: Run the Agent

With the graph compiled, running the agent is straightforward. Here's a complete example with sample code to review:

# Sample code with intentional issues to test the agent
sample_code = """
import os
import requests

def get_user_data(user_id):
    query = f"SELECT * FROM users WHERE id = {user_id}"
    db_password = "super_secret_123"
    conn = psycopg2.connect(
        host="localhost",
        password=db_password
    )
    cursor = conn.cursor()
    cursor.execute(query)
    result = cursor.fetchone()
    return result

def process_items(items):
    result = []
    for i in range(len(items)):
        item = items[i]
        if item != None:
            processed = heavy_computation(item)
            result.append(processed)
    return result

def heavy_computation(x):
    # Expensive operation without caching
    import time
    time.sleep(1)
    return x * x
"""

# Initialize state
initial_state: CodeReviewState = {
    "source_code": sample_code,
    "language": "python",
    "findings": [],
    "final_report": None,
    "has_critical_issues": False,
    "iteration_count": 0
}

# Run the enhanced graph
print("🔍 Running LangGraph Code Review Agent...")
print("=" * 60)

final_state = graph_v2.invoke(initial_state)

# Display the final report
print(final_state["final_report"])
print("\n" + "=" * 60)
print(f"Total findings: {len(final_state.get('findings', []))}")
print(f"Iterations: {final_state.get('iteration_count', 0)}")
print(f"Critical issues: {final_state.get('has_critical_issues', False)}")

Expected Output Structure

When you run this, the agent will produce a structured Markdown report similar to:

# 📋 Automated Code Review Report

**Language:** python
**Total Issues Found:** 7

## 🔒 Security
1. CRITICAL: SQL injection vulnerability in get_user_data() - query constructed with string formatting | Fix: Use parameterized queries
2. CRITICAL: Hardcoded database password 'super_secret_123' in source code | Fix: Use environment variables or secrets manager
3. ...

## 🐛 Logic & Potential Bugs
4. HIGH: item != None should use 'is not None' for proper None checking | Fix: Replace with 'if item is not None'
...

## 🎨 Style & Convention
...

## ⚡ Performance
...

## ⚠️ Action Required
**2 CRITICAL** issues need immediate attention before deployment.
---
*Report generated by LangGraph Code Review Agent*

Step 7: Visualizing the Graph (Optional)

LangGraph includes utilities to visualize your graph structure, which is invaluable for debugging complex workflows:

from langgraph.graph import Graph

# For the simple graph
graph_image = graph.get_graph().draw_png()
with open("code_review_graph.png", "wb") as f:
    f.write(graph_image)

# For the enhanced graph with conditional routing
graph_v2_image = graph_v2.get_graph().draw_png()
with open("code_review_graph_v2.png", "wb") as f:
    f.write(graph_v2_image)

print("Graph visualizations saved as PNG files.")

Complete Agent Code

Below is the full, self-contained code for the enhanced code review agent. Save it as code_review_agent.py:

"""
LangGraph Code Review Agent
A multi-specialist automated code reviewer with conditional escalation.
"""

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

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END


# ─── State Definition ───────────────────────────────────────────

class CodeReviewState(TypedDict):
    source_code: str
    language: str
    findings: Annotated[List[str], operator.add]
    final_report: Optional[str]
    has_critical_issues: bool
    iteration_count: int


# ─── LLM Configuration ──────────────────────────────────────────

llm = ChatOpenAI(model="gpt-4o", temperature=0.2)


# ─── Reviewer Nodes ─────────────────────────────────────────────

def style_reviewer(state: CodeReviewState) -> dict:
    system_prompt = SystemMessage(content="""You are an expert code style reviewer.
    Analyze the provided code for naming conventions, formatting, readability,
    comment quality, and language-specific style guide adherence.
    
    Format each finding as: [STYLE] {severity}: {description} | Suggestion: {fix}
    Severity: LOW, MEDIUM, HIGH
    If no issues found, respond with exactly: NO_STYLE_ISSUES_FOUND""")
    
    user_message = HumanMessage(content=f"""Review this {state.get('language', 'code')} code:
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    if result == "NO_STYLE_ISSUES_FOUND":
        return {"findings": []}
    findings = [line.strip() for line in result.split("\n") if line.strip() and "[STYLE]" in line]
    return {"findings": findings}


def security_reviewer(state: CodeReviewState) -> dict:
    system_prompt = SystemMessage(content="""You are an expert application security reviewer.
    Analyze for injection vulnerabilities, hardcoded secrets, insecure crypto,
    missing validation, auth flaws, path traversal, unsafe deserialization.
    
    Format: [SECURITY] {severity}: {description} | Fix: {recommendation}
    Severity: LOW, MEDIUM, HIGH, CRITICAL
    If no issues: NO_SECURITY_ISSUES_FOUND""")
    
    user_message = HumanMessage(content=f"""Security review:
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    if result == "NO_SECURITY_ISSUES_FOUND":
        return {"findings": [], "has_critical_issues": False}
    findings = [line.strip() for line in result.split("\n") if line.strip() and "[SECURITY]" in line]
    has_critical = any("CRITICAL" in f for f in findings)
    return {"findings": findings, "has_critical_issues": has_critical}


def logic_reviewer(state: CodeReviewState) -> dict:
    system_prompt = SystemMessage(content="""You are an expert logic reviewer.
    Analyze for off-by-one errors, boundary bugs, incorrect conditionals,
    null reference risks, type mismatches, resource leaks, race conditions,
    and error handling flaws.
    
    Format: [LOGIC] {severity}: {description} | Fix: {recommendation}
    Severity: LOW, MEDIUM, HIGH, CRITICAL
    If no issues: NO_LOGIC_ISSUES_FOUND""")
    
    user_message = HumanMessage(content=f"""Logic review:
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    if result == "NO_LOGIC_ISSUES_FOUND":
        return {"findings": [], "has_critical_issues": False}
    findings = [line.strip() for line in result.split("\n") if line.strip() and "[LOGIC]" in line]
    has_critical = any("CRITICAL" in f for f in findings)
    return {"findings": findings, "has_critical_issues": has_critical}


def performance_reviewer(state: CodeReviewState) -> dict:
    system_prompt = SystemMessage(content="""You are an expert performance engineer.
    Analyze for inefficient algorithms, redundant computations, memory waste,
    missing caching, N+1 patterns, blocking operations, excessive allocations.
    
    Format: [PERFORMANCE] {severity}: {description} | Improvement: {suggestion}
    Severity: LOW, MEDIUM, HIGH
    If no issues: NO_PERFORMANCE_ISSUES_FOUND""")
    
    user_message = HumanMessage(content=f"""Performance review:
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    if result == "NO_PERFORMANCE_ISSUES_FOUND":
        return {"findings": []}
    findings = [line.strip() for line in result.split("\n") if line.strip() and "[PERFORMANCE]" in line]
    return {"findings": findings}


def deep_security_analysis(state: CodeReviewState) -> dict:
    system_prompt = SystemMessage(content="""You are a senior application security specialist.
    Perform a deeper analysis of previously flagged CRITICAL issues:
    - Trace vulnerable code paths end-to-end
    - Assess exploitability and impact
    - Provide detailed remediation with code examples
    - Check for pattern repetition
    
    Format: [DEEP-SECURITY] {severity}: {detailed_analysis} | Remediation: {steps}""")
    
    user_message = HumanMessage(content=f"""Deep security analysis. Original findings:
    {chr(10).join(state.get('findings', []))}
    
    Source code:
    {state.get('language', '')}
    {state['source_code']}
    """)
    
    response = llm.invoke([system_prompt, user_message])
    result = response.content.strip()
    
    deep_findings = [line.strip() for line in result.split("\n") if line.strip() and "[DEEP-SECURITY]" in line]
    return {"findings": deep_findings}


# ─── Report Aggregator ──────────────────────────────────────────

def report_aggregator(state: CodeReviewState) -> dict:
    findings = state.get("findings", [])
    
    if not findings:
        return {
            "final_report": "✅ **Code Review Complete** — No issues found across all dimensions.",
            "iteration_count": state.get("iteration_count", 0) + 1
        }
    
    severity_order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
    sorted_findings = sorted(findings, key=lambda f: next(
        (severity_order[s] for s in ["CRITICAL", "HIGH", "MEDIUM", "LOW"] if s in f), 99))
    
    categories = {}
    for finding in sorted_findings:
        for cat in ["STYLE", "SECURITY", "LOGIC", "PERFORMANCE", "DEEP-SECURITY"]:
            if f"[{cat}]" in finding:
                categories.setdefault(cat, []).append(finding)
                break
    
    report_lines = ["# 📋 Automated Code Review Report", ""]
    report_lines.append(f"**Language:** {state.get('language', 'Unknown')}")
    report_lines.append(f"**Total Issues Found:** {len(sorted_findings)}")
    report_lines.append("")
    
    category_labels = {
        "DEEP-SECURITY": "🔴 Deep Security Analysis (Escalated)",
        "SECURITY": "🔒 Security",
        "LOGIC": "🐛 Logic & Potential Bugs",
        "STYLE": "🎨 Style & Convention",
        "PERFORMANCE": "⚡ Performance"
    }
    
    priority_order = ["DEEP-SECURITY", "SECURITY", "LOGIC", "STYLE", "PERFORMANCE"]
    for cat_key in priority_order:
        if cat_key in categories:
            report_lines.append(f"## {category_labels[cat_key]}")
            report_lines.append("")
            for i, finding in enumerate(categories[cat_key], 1):
                clean = finding.replace