Building Code Review Agent with AI Agents 2026

Introduction

AI agents that can automatically review pull requests for bugs and quality are transforming how developers work. This guide shows you how to build a production-ready Code Review Agent using LangGraph + GitHub API.

What We're Building

A Code Review Agent that can:

Understand complex requests

Break them into sub-tasks

Execute tasks autonomously

Handle errors and retry

Produce consistent, high-quality output

Architecture

User Request
    ↓
[Code Review Agent Orchestrator]
    ↓
[Task Planning] → [Tool Selection] → [Execution]
    ↓                                      ↓
[Validation] ←──────────────────── [Results]
    ↓
Final Output

Implementation with LangGraph

python
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
import json
State definition
class CodeReviewAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    task: str
    sub_tasks: List[str]
    completed_tasks: List[str]
    final_output: str | None
    iterations: int
Define tools for automatically review pull requests for bugs and quality
@tool
def analyze_task(task: str) -> str:
    """Break down a complex task into sub-tasks."""
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(f"Break this into 3-5 specific, actionable sub-tasks: {task}")
    return response.content
tools = [analyze_task, execute_sub_task, validate_output]
Initialize LLM with tools
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
llm_with_tools = llm.bind_tools(tools)
Agent node
def agent_node(state: CodeReviewAgentState):
    if state.get("iterations", 0) > 8:
        return {"final_output": "Max iterations reached", "iterations": 9}
    
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "iterations": state.get("iterations", 0) + 1
    }
Tool execution node
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)
def should_continue(state: CodeReviewAgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
        return "tools"
    return "end"
Build graph
workflow = StateGraph(CodeReviewAgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
workflow.add_edge("tools", "agent")
agent = workflow.compile()

Using the Agent

python
from langchain_core.messages import HumanMessage
def run_code_review_agent(request: str) -> str:
    """Run the Code Review Agent on a user request."""
    
    initial_state = {
        "messages": [HumanMessage(content=request)],
        "task": request,
        "sub_tasks": [],
        "completed_tasks": [],
        "final_output": None,
        "iterations": 0
    }
    
    result = agent.invoke(initial_state)
    
    # Extract the final answer
    last_message = result["messages"][-1]
    return last_message.content
Usage
output = run_code_review_agent(
    "Create a comprehensive plan to automatically review pull requests for bugs and quality"
)
print(output)

Adding Memory with Persistence

python
from langgraph.checkpoint.sqlite import SqliteSaver
def run_with_memory(request: str, session_id: str) -> str:
    config = {"configurable": {"thread_id": session_id}}
    
    state = {
        "messages": [HumanMessage(content=request)],
        "task": request,
        "sub_tasks": [],
        "completed_tasks": [],
        "final_output": None,
        "iterations": 0
    }
    
    result = agent_with_memory.invoke(state, config=config)
    return result["messages"][-1].content
First interaction
response1 = run_with_memory("Start automatically review pull requests for bugs and quality", session_id="session-001")
Follow-up (agent remembers context)
response2 = run_with_memory("Continue from where we left off", session_id="session-001")

Production: FastAPI Service

python
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio
app = FastAPI(title="Code Review Agent Service")
@app.post("/agent/run")
async def run_agent(request: AgentRequest):
    if request.stream:
        async def stream_response():
            async for event in agent.astream_events(
                {"messages": [HumanMessage(content=request.task)], 
                 "task": request.task, "sub_tasks": [], 
                 "completed_tasks": [], "final_output": None, "iterations": 0},
                version="v2"
            ):
                if event["event"] == "on_chat_model_stream":
                    content = event["data"]["chunk"].content
                    if content:
                        yield content
        
        return StreamingResponse(stream_response(), media_type="text/plain")
    
    result = run_code_review_agent(request.task)
    return {"result": result, "session_id": request.session_id}
@app.get("/health")
async def health():
    return {"status": "healthy", "agent": "Code Review Agent"}

Monitoring Agent Performance

python
from dataclasses import dataclass
from datetime import datetime
import statistics
metrics_store: List[AgentMetrics] = []
def run_with_metrics(task: str) -> tuple[str, AgentMetrics]:
    import time
    start = time.time()
    
    try:
        result = run_code_review_agent(task)
        success = True
    except Exception as e:
        result = f"Error: {e}"
        success = False
    
    duration = (time.time() - start) * 1000
    
    # Note: iterations would come from actual state in production
    metrics = AgentMetrics(
        task=task[:50],
        iterations=3,  
        duration_ms=duration,
        success=success,
        output_length=len(result)
    )
    metrics_store.append(metrics)
    return result, metrics
def print_metrics_report():
    if not metrics_store:
        return
    
    successful = [m for m in metrics_store if m.success]
    durations = [m.duration_ms for m in metrics_store]
    
    print(f"Total runs: {len(metrics_store)}")
    print(f"Success rate: {len(successful)/len(metrics_store):.1%}")
    print(f"Avg duration: {statistics.mean(durations):.0f}ms")
    print(f"p95 duration: {sorted(durations)[int(len(durations)*0.95)]:.0f}ms")

Best Practices

Conclusion

Building Code Review Agent with AI agents enables autonomous automatically review pull requests for bugs and quality. The LangGraph implementation provides the right balance of control and flexibility for production use.

Start with a simple proof of concept, add persistence, then scale up as confidence grows.

---

*Code Review Agent implementation using LangGraph + GitHub API | May 2026*