AI Code Review at Scale: LinkedIn’s Multi-Agent Approach

Introduction

At enterprise scale, pull requests do not just accumulate; they fracture developer focus. When a single repository pushes thousands of commits daily, traditional static analysis tools and monolithic LLM reviewers hit hard limits. Context windows saturate, token costs scale linearly with diff size, and latency spikes degrade the developer experience. We observed that throwing a single large model at a 500-file diff produces noisy, overlapping comments and frequent hallucinations on edge cases.

The solution is not a larger model. It is a deliberate architectural shift: decomposing the review process into a multi-agent topology. By isolating concerns into specialized workers, routing execution through a deterministic DAG, and managing state through a shared artifact store, we can parallelize review logic, bound context windows, and maintain predictable P99 latency. This article breaks down the production architecture we used to scale AI-assisted code review across high-throughput engineering clusters.

Why This Matters

Review velocity is a direct multiplier of release throughput. When engineers wait minutes for AI feedback, or spend time filtering false positives, the automation becomes a tax rather than an asset. Single-agent pipelines fail at scale because they treat every diff as a monolithic text stream. They lack isolation, struggle with context drift, and cannot route low-risk files to faster, cheaper inference endpoints.

A multi-agent approach solves three production constraints:

1. Latency isolation: Parallel execution prevents a slow security scan from blocking style checks.
2. Context bounding: Agents only receive the code slices, AST nodes, and dependency metadata they need.
3. Cost governance: Tiered routing sends trivial changes to small models and reserves large models for architectural or security-critical diffs.

This pattern transforms AI review from a brittle, sequential bottleneck into a resilient, observable pipeline that scales with engineering output.

How It Works

The pipeline operates as a stateless ingestion layer feeding a DAG-based orchestrator. When a PR is opened, the system parses the diff, builds an abstract syntax tree (AST), and extracts dependency metadata. The router evaluates file types, change risk, and historical patterns to dispatch work to specialized agents. Each agent operates independently, writes structured results to a shared artifact store, and signals completion. An aggregator then deduplicates overlapping line ranges, resolves conflicting suggestions, and formats the final review payload for the PR interface.

flowchart TD
    A[PR Ingestion] --> B[Diff & AST Parser]
    B --> C{Agent Router}
    C -->|Security| D[SecAgent]
    C -->|Architecture| E[ArchAgent]
    C -->|Style/Lint| F[StyleAgent]
    C -->|Context/Dep| G[ContextAgent]
    D --> H[Shared Artifact Store]
    E --> H
    F --> H
    G --> H
    H --> I[Comment Aggregator]
    I --> J[PR Review Output]
    J --> K[Developer Interface]
    K -->|Accept/Dismiss/Feedback| L[Feedback Collector]
    L --> M[Agent Tuning & Cache Update]
    M --> C

The router does not chain LLM calls. It publishes structured work items and waits for completion events. The shared artifact store acts as the single source of truth for AST nodes, call graphs, and intermediate findings. The aggregator applies deterministic rules to merge results, ensuring the final output is concise and non-redundant. Feedback from developers flows back into the tuning loop, updating routing heuristics and prompt templates without interrupting the live pipeline.

Core Concepts

The architecture relies on four production-grade patterns:

1. DAG-Based Orchestration: Work flows through directed acyclic graphs. Parallel branches execute concurrently; merge nodes wait for dependencies. This prevents cascading delays and isolates agent failures.
2. Bounded Context Windows: Instead of feeding entire files, the system slices diffs into hunks, attaches relevant AST nodes, and injects only the signatures and imports required for accurate analysis. Context size remains O(K) where K is hunk complexity, not diff size.
3. Shared Artifact Store: A lightweight, schema-validated store (typically Redis or SQLite-backed) holds intermediate results. Agents read/write structured JSON with strict versioning. This eliminates conversational drift and enables idempotent retries.
4. Human-in-the-Loop Guardrails: Every AI comment includes a confidence score and a dismiss/accept toggle. Dismissed comments are logged for prompt refinement. High-impact changes (e.g., auth logic, database migrations) trigger mandatory human review regardless of AI output.

These concepts ensure the pipeline remains predictable, auditable, and safe for production CI/CD integration.

Examples & Code Walkthrough

Below is a production-ready implementation of the router, context bouncer, and aggregator. The code uses typed structures, defensive error handling, and realistic CI/CD constraints.

import asyncio
import hashlib
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional

logger = logging.getLogger(__name__)

class AgentType(Enum):
    SECURITY = "security"
    ARCHITECTURE = "architecture"
    STYLE = "style"
    CONTEXT = "context"

@dataclass
class DiffHunk:
    file_path: str
    start_line: int
    end_line: int
    content: str
    ast_nodes: List[str]
    risk_score: float  # 0.0 to 1.0

@dataclass
class AgentResult:
    agent_type: AgentType
    file_path: str
    line_range: tuple[int, int]
    comments: List[str]
    confidence: float
    error: Optional[str] = None

@dataclass
class ReviewOutput:
    file_path: str
    merged_comments: List[str]
    final_confidence: float

class ContextBouncer:
    """Enforces token limits by slicing diffs and anchoring AST context."""
    def __init__(self, max_tokens: int = 4096):
        self.max_tokens = max_tokens

    def prepare_payload(self, hunk: DiffHunk) -> dict:
        # Truncate content if it exceeds token budget
        token_estimate = len(hunk.content.split()) * 1.3
        payload_content = hunk.content
        if token_estimate > self.max_tokens:
            payload_content = hunk.content[:int(self.max_tokens / 1.3)]
            logger.warning("Context truncated for %s", hunk.file_path)
            
        return {
            "file": hunk.file_path,
            "lines": f"{hunk.start_line}-{hunk.end_line}",
            "content": payload_content,
            "ast_context": hunk.ast_nodes,
            "risk": hunk.risk_score
        }

class AgentRouter:
    """Dispatches hunks to specialized agents based on file type and risk."""
    def __init__(self):
        self.routes = {
            ".py": {AgentType.STYLE, AgentType.CONTEXT},
            ".java": {AgentType.ARCHITECTURE, AgentType.SECURITY},
            ".go": {AgentType.STYLE, AgentType.CONTEXT},
            "Dockerfile": {AgentType.SECURITY},
        }

    def resolve_agents(self, hunk: DiffHunk) -> set[AgentType]:
        ext = hunk.file_path.rsplit(".", 1)[-1] if "." in hunk.file_path else ""
        base = self.routes.get(ext, set())
        
        # High risk triggers security + architecture review
        if hunk.risk_score > 0.7:
            base.add(AgentType.SECURITY)
            base.add(AgentType.ARCHITECTURE)
        return base

class CommentAggregator:
    """Deduplicates overlapping line ranges and merges agent outputs."""
    def process_results(self, results: List[AgentResult]) -> Dict[str, ReviewOutput]:
        file_map: Dict[str, List[AgentResult]] = {}
        for r in results:
            if r.error:
                logger.error("Agent %s failed on %s: %s", r.agent_type.value, r.file_path, r.error)
                continue
            file_map.setdefault(r.file_path, []).append(r)

        output: Dict[str, ReviewOutput] = {}
        for fpath, agents in file_map.items():
            merged = self._deduplicate(agents)
            avg_conf = sum(a.confidence for a in merged) / len(merged) if merged else 0.0
            output[fpath] = ReviewOutput(
                file_path=fpath,
                merged_comments=sum([a.comments for a in merged], []),
                final_confidence=round(avg_conf, 2)
            )
        return output

    def _deduplicate(self, agents