Self-Host AI Code Review on GPU Cloud: Deploy Open-Source PR Review Agents (2026 Guide)

GitHub/GitLab PR Event
       |
       v
FastAPI Webhook Handler   <-- validate HMAC-SHA256 signature
       |
       v
Diff Parser + Context Window Builder
  (chunk diffs to <=8K tokens, attach file context)
       |
       v
vLLM Inference Endpoint   (Qwen2.5-Coder 32B or DeepSeek V3)
       |
       v
Comment Formatter + Deduplication Store (Redis)
       |
       v
GitHub/GitLab Review API  <-- post inline review comments

nvidia-smi
# Should show H100 80GB HBM3 with ~80GB VRAM

# Install Docker
curl -fsSL https://get.docker.com | sh

# Install NVIDIA Container Toolkit
curl -fsSL https://nvidia.github.io/libnvidia-container/gpgkey | sudo gpg --dearmor \
  -o /usr/share/keyrings/nvidia-container-toolkit-keyring.gpg
curl -s -L https://nvidia.github.io/libnvidia-container/stable/deb/nvidia-container-toolkit.list | \
  sed 's#deb https://#deb [signed-by=/usr/share/keyrings/nvidia-container-toolkit-keyring.gpg] https://#g' | \
  sudo tee /etc/apt/sources.list.d/nvidia-container-toolkit.list
sudo apt-get update && sudo apt-get install -y nvidia-container-toolkit
sudo nvidia-ctk runtime configure --runtime=docker
sudo systemctl restart docker

docker run --gpus all \
  -p 8000:8000 \
  -v ~/.cache/huggingface:/root/.cache/huggingface \
  -e HUGGING_FACE_HUB_TOKEN=$HF_TOKEN \
  --ipc=host \
  vllm/vllm-openai:latest \
  --model Qwen/Qwen2.5-Coder-32B-Instruct \
  --dtype fp8 \
  --tensor-parallel-size 1 \
  --max-model-len 32768 \
  --enable-prefix-caching \
  --gpu-memory-utilization 0.90

curl http://localhost:8000/health
# {"status":"healthy"}

curl http://localhost:8000/v1/models
# Should list Qwen2.5-Coder-32B-Instruct

import hashlib
import hmac
import os
from fastapi import FastAPI, Request, HTTPException
from contextlib import asynccontextmanager
import asyncio
import httpx

app = FastAPI()

WEBHOOK_SECRET = os.environ["GITHUB_WEBHOOK_SECRET"]
GITHUB_APP_PRIVATE_KEY = os.environ["GITHUB_APP_PRIVATE_KEY"]
GITHUB_APP_ID = os.environ["GITHUB_APP_ID"]
VLLM_BASE_URL = os.environ.get("VLLM_BASE_URL", "http://localhost:8000/v1")

_bg_tasks: set = set()


def verify_signature(payload: bytes, signature: str) -> bool:
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(), payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)


@app.post("/webhook")
async def handle_webhook(request: Request):
    payload = await request.body()
    sig = request.headers.get("X-Hub-Signature-256", "")
    if not verify_signature(payload, sig):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = request.headers.get("X-GitHub-Event")
    if event != "pull_request":
        return {"status": "ignored"}

    body = await request.json()
    action = body.get("action")
    if action not in ("opened", "synchronize", "reopened"):
        return {"status": "ignored"}

    # Enqueue async processing and return 200 immediately
    t = asyncio.create_task(process_pr(body))
    _bg_tasks.add(t)
    t.add_done_callback(_bg_tasks.discard)
    return {"status": "queued"}

async def get_pr_files(
    owner: str,
    repo: str,
    pull_number: int,
    token: str
) -> list[dict]:
    url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pull_number}/files"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28"
    }
    files = []
    params: dict = {"per_page": 100, "page": 1}
    async with httpx.AsyncClient() as client:
        while url:
            resp = await client.get(url, headers=headers, params=params)
            resp.raise_for_status()
            files.extend(resp.json())
            next_url = resp.links.get("next", {}).get("url")
            url = next_url
            params = {}
    return files

from transformers import AutoTokenizer

CHUNK_TOKENS = 6000
OVERLAP_TOKENS = 512

_tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-Coder-32B-Instruct")

def chunk_diff(files: list[dict]) -> list[str]:
    enc = _tokenizer
    chunks = []
    current_chunk = []
    current_count = 0

    for f in files:
        filename = f["filename"]
        patch = f.get("patch", "")
        if not patch:
            continue
        file_text = f"### File: {filename}\n```\n{patch}\n```\n"
        token_ids = enc.encode(file_text)

        if current_count + len(token_ids) > CHUNK_TOKENS and current_chunk:
            chunks.append("".join(current_chunk))
            # Keep last OVERLAP_TOKENS worth of content for context
            overlap_text = "".join(current_chunk)
            overlap_ids = enc.encode(overlap_text)[-OVERLAP_TOKENS:]
            current_chunk = [enc.decode(overlap_ids)]
            current_count = len(overlap_ids)

        # If a single file exceeds CHUNK_TOKENS on its own, split it into sub-chunks
        if len(token_ids) > CHUNK_TOKENS:
            file_ids = enc.encode(file_text)
            for start in range(0, len(file_ids), CHUNK_TOKENS - OVERLAP_TOKENS):
                sub_ids = file_ids[start:start + CHUNK_TOKENS]
                chunks.append(enc.decode(sub_ids))
            current_chunk = []
            current_count = 0
            continue

        current_chunk.append(file_text)
        current_count += len(token_ids)

    if current_chunk:
        chunks.append("".join(current_chunk))

    return chunks

from openai import AsyncOpenAI

REVIEW_SYSTEM_PROMPT = """You are a senior software engineer performing a code review.
Analyze the provided code diff and return a JSON array of review comments.

Each comment must have this structure:
{
  "path": "<file path>",
  "line": <line number in the diff, or null for general comments>,
  "severity": "error" | "warning" | "suggestion",
  "category": "security" | "logic" | "performance" | "style" | "documentation",
  "body": "<comment text>"
}

Focus on:
- Logic errors and off-by-one bugs
- Security issues (injection, auth bypass, unvalidated input)
- Performance problems (N+1 queries, unnecessary allocations)
- Missing error handling

Return only the JSON array, no other text."""


async def get_review_comments(diff_chunk: str) -> list[dict]:
    client = AsyncOpenAI(base_url=VLLM_BASE_URL, api_key="not-needed")
    response = await client.chat.completions.create(
        model="Qwen/Qwen2.5-Coder-32B-Instruct",
        messages=[
            {"role": "system", "content": REVIEW_SYSTEM_PROMPT},
            {"role": "user", "content": diff_chunk}
        ],
        temperature=0.1,
        max_tokens=2048
    )
    import json
    import re
    if not response.choices or response.choices[0].message.content is None:
        print("Empty response for chunk")
        return []
    content = response.choices[0].message.content.strip()
    content = re.sub(r'^```[a-zA-Z]*\n|```$', '', content, flags=re.MULTILINE).strip()
    try:
        return json.loads(content)
    except json.JSONDecodeError as e:
        print(f"Failed to parse review response as JSON: {e!r}")
        return []

import redis.asyncio as aioredis
import json

redis_client = aioredis.Redis(host="localhost", port=6379)

async def process_pr(body: dict):
    repo = body["repository"]
    pr = body["pull_request"]
    owner = repo["owner"]["login"]
    repo_name = repo["name"]
    pull_number = pr["number"]
    head_sha = pr["head"]["sha"]
    installation_id = body["installation"]["id"]

    # Idempotency: two-state key ("processing" | "done") so failures are retryable.
    # Setting "done" only after post_review succeeds means a spot-instance preemption
    # or GitHub API error leaves no key (or leaves "processing" which expires), so the
    # next webhook re-delivery or ARQ retry will run the review cleanly.
    idem_key = f"reviewed:{owner}/{repo_name}:{pull_number}:{head_sha}"
    existing = await redis_client.get(idem_key)
    if existing == b"done":
        return  # Already reviewed successfully
    if not await redis_client.set(idem_key, "processing", nx=True, ex=72 * 3600):
        return  # Another worker already claimed this job

    try:
        token = await get_installation_token(installation_id)
        files = await get_pr_files(owner, repo_name, pull_number, token)
        chunks = chunk_diff(files)

        # Process chunks in parallel
        all_comments = []
        results = await asyncio.gather(
            *[get_review_comments(chunk) for chunk in chunks],
            return_exceptions=True
        )
        for result in results:
            if isinstance(result, list):
                all_comments.extend(result)

        if not all_comments:
            await redis_client.delete(idem_key)
            return

        await post_review(owner, repo_name, pull_number, head_sha, all_comments, token)
        await redis_client.set(idem_key, "done", ex=72 * 3600)
    except Exception:
        await redis_client.delete(idem_key)
        raise


async def post_review(
    owner: str,
    repo: str,
    pull_number: int,
    commit_id: str,
    comments: list[dict],
    token: str
):
    # Fetch existing comments to deduplicate
    existing_url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pull_number}/comments"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28"
    }
    async with httpx.AsyncClient() as client:
        existing = []
        next_url: str | None = f"{existing_url}?per_page=100"
        while next_url:
            resp = await client.get(next_url, headers=headers)
            if resp.status_code == 200:
                existing.extend(resp.json())
                next_url = resp.links.get("next", {}).get("url")
            else:
                break

    existing_hashes = {
        hashlib.sha256(c["body"].encode()).hexdigest()
        for c in existing
    }

    formatted = []
    general_bodies = []
    for c in comments:
        label = f"**[{c['severity'].upper()}]** "
        body_hash = hashlib.sha256((label + c["body"]).encode()).hexdigest()
        if body_hash in existing_hashes:
            continue  # Skip duplicate
        if c.get("line") is not None:
            formatted.append({
                "path": c["path"],
                "line": c["line"],
                "side": "RIGHT",
                "body": label + c["body"]
            })
        else:
            # General file-level comment: post in the review body, not the comments array.
            general_bodies.append(f"{label}`{c['path']}`: {c['body']}")

    if not formatted and not general_bodies:
        return

    review_url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pull_number}/reviews"
    review_body = {
        "commit_id": commit_id,
        "event": "COMMENT",
        "body": "\n\n".join(general_bodies),
        "comments": formatted
    }
    async with httpx.AsyncClient() as client:
        resp = await client.post(review_url, headers=headers, json=review_body)
        resp.raise_for_status()

idem_key = f"reviewed:{owner}/{repo_name}:{pull_number}:{head_sha}"
existing = await redis_client.get(idem_key)
if existing == b"done":
    return  # Already reviewed successfully
if not await redis_client.set(idem_key, "processing", nx=True, ex=72 * 3600):
    return  # Another worker already claimed this job

try:
    # ... do all work ...
    await redis_client.set(idem_key, "done", ex=72 * 3600)
except Exception:
    await redis_client.delete(idem_key)  # Allow retries on failure
    raise

# worker.py
from arq import create_pool
from arq.connections import RedisSettings

async def process_pr_job(ctx, body: dict):
    await process_pr(body)

class WorkerSettings:
    functions = [process_pr_job]
    redis_settings = RedisSettings()
    retry_jobs = True
    max_tries = 3
    job_timeout = 300  # 5-minute max per job