Multi-agent code review pipeline with LangGraph

A langgraph multi-agent code review pipeline gives you deterministic, auditable review flows that single-prompt approaches cannot. You get explicit state transitions, checkpointing for human-in-the-loop intervention, and the ability to swap individual agents without rewriting the whole system. This tutorial walks through building one from scratch with runnable code at each step.

Step 1: Project setup and dependencies

Create a fresh project and install the minimal dependency set. LangGraph sits on top of LangChain Core, so you need both. For the LLM calls, use any OpenAI-compatible endpoint — this works with local models, hosted providers, or a gateway like n4n.ai that handles fallback and routing automatically.

mkdir code-review-pipeline && cd code-review-pipeline
python -m venv .venv && source .venv/bin/activate
pip install langgraph langchain-core langchain-openai pydantic python-dotenv

Create a .env file with your API configuration:

# .env
OPENAI_API_KEY=your-key-here
OPENAI_BASE_URL=https://api.openai.com/v1  # or your gateway endpoint
REVIEW_MODEL=gpt-4o-mini

Verify the environment loads correctly:

# test_env.py
import os
from dotenv import load_dotenv

load_dotenv()
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY not set"
assert os.getenv("REVIEW_MODEL"), "REVIEW_MODEL not set"
print("Environment OK")

Run python test_env.py — you should see “Environment OK”.

Step 2: Define the state schema

The state schema is the contract between agents. Keep it flat and serializable; nested objects complicate checkpointing and debugging. Each agent reads what it needs and writes its findings to dedicated keys.

# state.py
from typing import TypedDict, List, Optional
from pydantic import BaseModel, Field

class Finding(BaseModel):
    file: str
    line: int
    severity: str  # "error" | "warning" | "info"
    message: str
    rule_id: Optional[str] = None
    suggested_fix: Optional[str] = None

class ReviewState(TypedDict):
    # Input
    pr_number: int
    repo_path: str
    changed_files: List[str]
    diff_content: str
    
    # Agent outputs
    static_findings: List[Finding]
    security_findings: List[Finding]
    style_findings: List[Finding]
    
    # Aggregated
    all_findings: List[Finding]
    summary: str
    
    # Control flow
    requires_human_review: bool
    current_agent: str

The current_agent field lets you observe which node is executing in the LangGraph studio or your own logging. requires_human_review becomes the conditional edge trigger later.

Step 3: Build the static analysis agent

This agent runs deterministic tools (ruff, mypy, eslint) and normalizes their output into the Finding schema. It does not call an LLM — that keeps it fast and predictable.

# agents/static_analysis.py
import subprocess
import json
import os
from pathlib import Path
from typing import List
from state import ReviewState, Finding

RUFF_RULES = {
    "E": "error", "W": "warning", "F": "error", "I": "warning",
    "UP": "warning", "B": "warning", "C4": "warning", "T20": "warning"
}

def run_ruff(repo_path: str, files: List[str]) -> List[Finding]:
    findings = []
    for file in files:
        if not file.endswith(".py"):
            continue
        full_path = Path(repo_path) / file
        if not full_path.exists():
            continue
        result = subprocess.run(
            ["ruff", "check", "--output-format=json", str(full_path)],
            capture_output=True, text=True, cwd=repo_path
        )
        if result.stdout:
            for issue in json.loads(result.stdout):
                code = issue.get("code", "")
                severity = RUFF_RULES.get(code[0] if code else "", "info")
                findings.append(Finding(
                    file=file,
                    line=issue["location"]["row"],
                    severity=severity,
                    message=issue["message"],
                    rule_id=code
                ))
    return findings

def run_mypy(repo_path: str, files: List[str]) -> List[Finding]:
    findings = []
    py_files = [f for f in files if f.endswith(".py")]
    if not py_files:
        return findings
    result = subprocess.run(
        ["mypy", "--json-report", "/tmp/mypy-report"] + py_files,
        capture_output=True, text=True, cwd=repo_path
    )
    report_path = Path("/tmp/mypy-report/errors.json")
    if report_path.exists():
        data = json.loads(report_path.read_text())
        for file, errors in data.items():
            for err in errors:
                findings.append(Finding(
                    file=file,
                    line=err["line"],
                    severity="error" if err["severity"] == "error" else "warning",
                    message=err["message"],
                    rule_id=f"mypy-{err.get('code', 'unknown')}"
                ))
    return findings

def static_analysis_node(state: ReviewState) -> ReviewState:
    repo_path = state["repo_path"]
    files = state["changed_files"]
    
    ruff_findings = run_ruff(repo_path, files)
    mypy_findings = run_mypy(repo_path, files)
    
    all_findings = ruff_findings + mypy_findings
    return {
        **state,
        "static_findings": all_findings,
        "all_findings": state.get("all_findings", []) + all_findings,
        "current_agent": "static_analysis"
    }

Verify it works in isolation:

# test_static.py
from agents.static_analysis import static_analysis_node
from state import ReviewState

test_state: ReviewState = {
    "pr_number": 1,
    "repo_path": ".",
    "changed_files": ["state.py"],
    "diff_content": "",
    "static_findings": [],
    "security_findings": [],
    "style_findings": [],
    "all_findings": [],
    "summary": "",
    "requires_human_review": False,
    "current_agent": ""
}

result = static_analysis_node(test_state)
print(f"Static findings: {len(result['static_findings'])}")
for f in result['static_findings'][:3]:
    print(f"  {f.file}:{f.line} [{f.severity}] {f.message}")

Run python test_static.py — you should see ruff/mypy findings for your own code.

Step 4: Build the security scanning agent

This agent uses an LLM to detect patterns that static analysis misses: hardcoded secrets, SQL injection vectors, insecure deserialization, and so on. Give it the diff context, not the whole file — token efficiency matters.

# agents/security.py
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
from typing import List
import json
import os
from state import ReviewState, Finding

SECURITY_PROMPT = """You are a security-focused code reviewer. Analyze the provided diff for security vulnerabilities.
Focus on: hardcoded secrets/keys, SQL injection, command injection, path traversal, insecure deserialization,
weak cryptography, missing authentication/authorization, XSS, SSRF, and insecure defaults.

Return ONLY a JSON array of findings. Each finding must have:
- file: string
- line: integer (best estimate from diff)
- severity: "error" | "warning" | "info"
- message: string
- rule_id: string (e.g., "SEC-001", "SEC-002")
- suggested_fix: string (optional)

If no issues found, return []. No markdown, no commentary."""

llm = ChatOpenAI(
    model=os.getenv("REVIEW_MODEL", "gpt-4o-mini"),
    temperature=0,
    max_tokens=2000
)

def security_node(state: ReviewState) -> ReviewState:
    diff = state["diff_content"]
    if not diff.strip():
        return {**state, "security_findings": [], "current_agent": "security"}
    
    messages = [
        SystemMessage(content=SECURITY_PROMPT),
        HumanMessage(content=f"Diff to analyze:\n```diff\n{diff}\n```")
    ]
    
    response = llm.invoke(messages)
    try:
        findings_data = json.loads(response.content)
    except json.JSONDecodeError:
        # Fallback: try to extract JSON from markdown code fence
        content = response.content.strip()
        if content.startswith("```"):
            content = content.split("\n", 1)[1].rsplit("\n", 1)[0]
        findings_data = json.loads(content)
    
    findings = [Finding(**f) for f in findings_data]
    return {
        **state,
        "security_findings": findings,
        "all_findings": state.get("all_findings", []) + findings,
        "current_agent": "security"
    }

Test with a known-bad diff:

# test_security.py
from agents.security import security_node
from state import ReviewState

bad_diff = """--- a/auth.py
+++ b/auth.py
@@ -1,5 +1,8 @@
 import os
+API_KEY = "sk-live-abcdef1234567890"  # hardcoded secret
+
 def verify_token(token: str) -> bool:
+    query = f"SELECT * FROM users WHERE token = '{token}'"  # SQL injection
+    return db.execute(query).fetchone()
     return True
"""

test_state: ReviewState = {
    "pr_number": 1,
    "repo_path": ".",
    "changed_files": ["auth.py"],
    "diff_content": bad_diff,
    "static_findings": [],
    "security_findings": [],
    "style_findings": [],
    "all_findings": [],
    "summary": "",
    "requires_human_review": False,
    "current_agent": ""
}

result = security_node(test_state)
print(f"Security findings: {len(result['security_findings'])}")
for f in result['security_findings']:
    print(f"  {f.file}:{f.line} [{f.severity}] {f.message} ({f.rule_id})")

Run python test_security.py — expect at least two findings (hardcoded secret, SQL injection).

Step 5: Build the style and consistency agent

This agent enforces team conventions that linters miss: naming patterns, architectural layering, error handling consistency, and documentation standards. It also runs fast because the prompt is narrow.

# agents/style.py
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
from typing import List
import json
import os
from state import ReviewState, Finding

STYLE_PROMPT = """You are a style and consistency reviewer. Check the diff for:
- Inconsistent naming (camelCase vs snake_case in same codebase)
- Missing docstrings on public functions/classes
- Inconsistent error handling (bare except, inconsistent exception types)
- Violation of project patterns (e.g., repository pattern, dependency injection)
- Overly complex functions (cyclomatic complexity > 10)
- Missing type hints on public APIs
- Inconsistent logging/observability patterns

Return ONLY a JSON array of findings with the same schema as the security agent.
If no issues found, return []. No markdown, no commentary."""

llm = ChatOpenAI(
    model=os.getenv("REVIEW_MODEL", "gpt-4o-mini"),
    temperature=0,
    max_tokens=1500
)

def style_node(state: ReviewState) -> ReviewState:
    diff = state["diff_content"]
    if not diff.strip():
        return {**state, "style_findings": [], "current_agent": "style"}
    
    messages = [
        SystemMessage(content=STYLE_PROMPT),
        HumanMessage(content=f"Diff to analyze:\n```diff\n{diff}\n```")
    ]
    
    response = llm.invoke(messages)
    try:
        findings_data = json.loads(response.content)
    except json.JSONDecodeError:
        content = response.content.strip()
        if content.startswith("```"):
            content = content.split("\n", 1)[1].rsplit("\n", 1)[0]
        findings_data = json.loads(content)
    
    findings = [Finding(**f) for f in findings_data]
    return {
        **state,
        "style_findings": findings,
        "all_findings": state.get("all_findings", []) + findings,
        "current_agent": "style"
    }

Step 6: Build the summary agent

The summary agent runs last. It sees all findings and produces a concise, actionable PR comment. It also decides whether human review is required — any “error” severity finding triggers it.

# agents/summary.py
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
import os
from state import ReviewState

SUMMARY_PROMPT = """You are a senior engineer summarizing code review findings for a PR comment.
Write a concise summary (max 300 words) covering:
1. Overall assessment (approve / request changes / comment)
2. Critical issues that block merge
3. Notable improvements or patterns
4. Action items for the author

Be direct. No fluff. Use bullet points for readability."""

llm = ChatOpenAI(
    model=os.getenv("REVIEW_MODEL", "gpt-4o-mini"),
    temperature=0.1,
    max_tokens=800
)

def summary_node(state: ReviewState) -> ReviewState:
    all_findings = state["all_findings"]
    error_count = sum(1 for f in all_findings if f.severity == "error")
    warning_count = sum(1 for f in all_findings if f.severity == "warning")
    info_count = sum(1 for f in all_findings if f.severity == "info")
    
    findings_summary = f"Errors: {error_count}, Warnings: {warning_count}, Info: {info_count}\n\n"
    for f in all_findings:
        findings_summary += f"- [{f.severity.upper()}] {f.file}:{f.line} - {f.message}\n"
        if f.suggested_fix:
            findings_summary += f"  Fix: {f.suggested_fix}\n"
    
    messages = [
        SystemMessage(content=SUMMARY_PROMPT),
        HumanMessage(content=f"All findings:\n{findings_summary}")
    ]
    
    response = llm.invoke(messages)
    requires_human = error_count > 0
    
    return {
        **state,
        "summary": response.content,
        "requires_human_review": requires_human,
        "current_agent": "summary"
    }

Step 7: Wire the graph with conditional edges

Now compose the nodes into a LangGraph StateGraph. The key design decision: run static analysis, security, and style in parallel (they’re independent), then fan in to the summary agent. Use add_edge for linear flow and add_conditional_edges for the human-review gate.

# graph.py
from langgraph.graph import StateGraph, END
from state import ReviewState
from agents.static_analysis import static_analysis_node
from agents.security import security_node
from agents.style import style_node
from agents.summary import summary_node

def should_continue_to_summary(state: ReviewState) -> str:
    # All three analysis agents write to all_findings; summary runs after all complete
    return "summary"

def requires_human_review(state: ReviewState) -> str:
    if state["requires_human_review"]:
        return "human_review"
    return "complete"

# Build the graph
workflow = StateGraph(ReviewState)

# Add nodes
workflow.add_node("static_analysis", static_analysis_node)
workflow.add_node("security", security_node)
workflow.add_node("style", style_node)
workflow.add_node("summary", summary_node)

# Parallel fan-out from start
workflow.set_entry_point("static_analysis")
workflow.add_edge("static_analysis", "security")
workflow.add_edge("static_analysis", "style")

# Fan-in to summary (both security and style must complete)
workflow.add_edge("security", "summary")
workflow.add_edge("style", "summary")

# Conditional edge after summary
workflow.add_conditional_edges(
    "summary",
    requires_human_review,
    {
        "human_review": END,  # In practice, route to a human-review node or pause
        "complete": END
    }
)

# Compile with checkpointing for observability and resume
from langgraph.checkpoint.memory import MemorySaver
graph = workflow.compile(checkpointer=MemorySaver())

The MemorySaver checkpointer lets you inspect state at each step and resume from any node — critical for debugging and for human-in-the-loop workflows where you pause at human_review.

Step 8: Create the entry point and run the pipeline

Wire a CLI that loads a real PR diff (or a test diff) and invokes the graph. The config parameter with thread_id enables checkpointing.

# main.py
import sys
import os
from pathlib import Path
from graph import graph
from state import ReviewState

def load_diff(pr_number: int, repo_path: str) -> tuple[str, list[str]]:
    # In production, fetch from GitHub/GitLab API. Here we simulate.
    # Replace with actual git diff logic or API call.
    diff_file = Path(repo_path) / f".review_diffs/pr_{pr_number}.diff"
    if diff_file.exists():
        content = diff_file.read_text()
        # Parse changed files from diff header
        files = []
        for line in content.split("\n"):
            if line.startswith("+++ b/"):
                files.append(line[6:])
        return content, files
    return "", []

def main():
    if len(sys.argv) < 3:
        print("Usage: python main.py <pr_number> <repo_path>")
        sys.exit(1)
    
    pr_number = int(sys.argv[1])
    repo_path = sys.argv[2]
    
    diff_content, changed_files = load_diff(pr_number, repo_path)
    
    initial_state: ReviewState = {
        "pr_number": pr_number,
        "repo_path": repo_path,
        "changed_files": changed_files,
        "diff_content": diff_content,
        "static_findings": [],
        "security_findings": [],
        "style_findings": [],
        "all_findings": [],
        "summary": "",
        "requires_human_review": False,
        "current_agent": ""
    }
    
    config = {"configurable": {"thread_id": f"pr-{pr_number}"}}
    
    print(f"Running review for PR #{pr_number}...")
    final_state = graph.invoke(initial_state, config=config)
    
    print("\n" + "="*60)
    print("REVIEW SUMMARY")
    print("="*60)
    print(final_state["summary"])
    print(f"\nRequires human review: {final_state['requires_human_review']}")
    print(f"Total findings: {len(final_state['all_findings'])}")
    
    # Save findings as JSON for CI integration
    import json
    output = {
        "pr_number": pr_number,
        "findings": [f.model_dump() for f in final_state["all_findings"]],
        "summary": final_state["summary"],
        "requires_human_review": final_state["requires_human_review"]
    }
    Path(f"review_results_pr_{pr_number}.json").write_text(json.dumps(output, indent=2))
    print(f"\nResults saved to review_results_pr_{pr_number}.json")

if __name__ == "__main__":
    main()

Create a test diff file to verify end-to-end:

mkdir -p .review_diffs
cat > .review_diffs/pr_42.diff << 'EOF'
--- a/payment.py
+++ b/payment.py
@@ -1,10 +1,15 @@
 import os
+STRIPE_KEY = "sk_live_51H8x..."  # TODO: move to env
+
 def process_payment(amount: int, currency: str) -> dict:
+    # FIXME: add validation
+    query = f"INSERT INTO payments (amount, currency) VALUES ({amount}, '{currency}')"
+    db.execute(query)
     return {"status": "ok"}
 
 class PaymentProcessor:
     def __init__(self):
         self.api_key = os.getenv("STRIPE_KEY")
     
     def charge(self, amount: int) -> bool:
         return True
EOF

Run the pipeline:

python main.py 42 .

You should see a summary printed to stdout and a review_results_pr_42.json file with all findings. Verify success by checking:

1. Static analysis caught ruff/mypy issues (unused import, missing type hints)
2. Security agent flagged the hardcoded Stripe key and SQL injection
3. Style agent noted the missing docstring and FIXME comment
4. Summary includes “Request changes” due to error-severity findings
5. requires_human_review is true

Step 9: Add a human-review pause node (production)

In production, you don’t exit at human_review — you pause the graph and wait for a human decision. LangGraph’s interrupt mechanism handles this cleanly.

# agents/human_review.py
from langgraph.types import interrupt
from state import ReviewState

def human_review_node(state: ReviewState) -> ReviewState:
    # This pauses execution and returns control to the caller
    decision = interrupt({
        "action": "human_review_required",
        "pr_number": state["pr_number"],
        "summary": state["summary"],
        "findings": [f.model_dump() for f in state["all_findings"]]
    })
    
    # When resumed, decision contains the human's input
    # e.g., {"approve": true, "comment": "LGTM after fix"}
    return {
        **state,
        "summary": state["summary"] + f"\n\n**Human decision**: {decision.get('comment', 'Approved')}",
        "requires_human_review": False
    }

Update the graph to include this node:

# graph.py (additions)
from langgraph.types import interrupt
from agents.human_review import human_review_node

workflow.add_node("human_review", human_review_node)

workflow.add_conditional_edges(
    "summary",
    requires_human_review,
    {
        "human_review": "human_review",
        "complete": END
    }
)

workflow.add_edge("human_review", END)

To resume after interrupt:

# resume_review.py
from graph import graph

config = {"configurable": {"thread_id": "pr-42"}}
# Resume with human decision
graph.invoke(None, config=config, interrupt_value={"approve": True, "comment": "Security issues acknowledged, will fix in follow-up"})

Step 10: CI/CD integration

The pipeline produces a JSON artifact. Your CI job fails if requires_human_review is true or if error-count exceeds a threshold. Example GitHub Actions step:

# .github/workflows/code-review.yml
- name: Run multi-agent code review
  run: |
    python main.py ${{ github.event.pull_request.number }} .
    
- name: Check review results
  run: |
    RESULT=$(cat review_results_pr_${{ github.event.pull_request.number }}.json)
    ERRORS=$(echo $RESULT | jq '.findings | map(select(.severity == "error")) | length')
    REQUIRES_HUMAN=$(echo $RESULT | jq '.requires_human_review')
    
    if [ "$ERRORS" -gt 0 ] || [ "$REQUIRES_HUMAN" = "true" ]; then
      echo "::error::Code review found $ERRORS errors. Human review required: $REQUIRES_HUMAN"
      exit 1
    fi

Production considerations

Token costs: The three LLM agents (security, style, summary) each consume ~1-3k tokens per PR. At gpt-4o-mini pricing, that’s fractions of a cent. Cache the summary prompt’s system message if you process high volume.

Latency: Parallel fan-out keeps wall-clock time near the slowest single agent (~2-4s). Static analysis runs in-process and is negligible.

False positives: The security and style agents will hallucinate occasionally. Mitigate by:

• Adding few-shot examples to each prompt
• Running a “critic” agent that filters findings below a confidence threshold
• Letting the human-review step dismiss false positives, then feeding those dismissals back as few-shot examples

Model routing: If you route through a gateway that honors model and provider headers, you can send static analysis to a cheap model and security to a stronger one without changing the graph structure.

Observability: The current_agent field and LangGraph’s built-in tracing (LangSmith or OpenTelemetry) let you measure per-agent latency, token usage, and error rates. Add a metrics dict to state if you need custom counters.

Verification checklist

Before merging this pipeline to your own repo, confirm:

• python test_env.py passes
• python test_static.py outputs findings from your codebase
• python test_security.py catches the two injected vulnerabilities
• python main.py 42 . runs end-to-end and produces review_results_pr_42.json
• The JSON output contains findings from all three agents
• requires_human_review is true when error-severity findings exist
• GitHub Actions workflow fails on error findings
• Interrupt/resume cycle works with resume_review.py

You now have a langgraph multi-agent code review pipeline that runs deterministic tools in parallel with LLM-based agents, produces structured output for CI, and pauses for human judgment when warranted. Extend it by adding agents for performance, accessibility, or dependency review — each plugs into the same graph without touching the others.