code agent

.env

@@ -0,0 +1,4 @@
+OLLAMA_API_URL=http://127.0.0.1:11434
+OLLAMA_MODEL=minimax-m2:cloud
+DEFAULT_TEMPERATURE=0.2
+MAX_TOKENS=1024

.env.example

@@ -0,0 +1,6 @@
+# Local Ollama server URL
+OLLAMA_API_URL=http://127.0.0.1:11434
+OLLAMA_MODEL=Minimax-M2
+# Optional controls
+DEFAULT_TEMPERATURE=0.2
+MAX_TOKENS=1024

README.md

@@ -1,3 +1,13 @@
 ---
 license: mit
 ---
+# Code Review Agent — v0.1 (MVP)
+
+
+Run locally (recommended to use a Python venv):
+
+
+```bash
+pip install -r requirements.txt
+# start ollama server separately and ensure Minimax-M2 is available
+uvicorn src.app:app --reload --port 8080

requirements.txt

@@ -0,0 +1,7 @@
+fastapi
+uvicorn[standard]
+python-multipart
+pydantic
+requests
+pytest
+python-dotenv

src/app.py

@@ -0,0 +1,42 @@
+import os
+from fastapi import FastAPI, UploadFile, File, Form, HTTPException
+from fastapi.responses import JSONResponse
+from dotenv import load_dotenv
+
+from .core.analyzer import quick_analyze
+from .core.reviewer import review_single_file
+from .utils.file_utils import save_upload_temp, read_file_text
+from .models.schemas import ReviewResponse
+
+load_dotenv()
+app = FastAPI(title="Code Review Agent v0.1", description="Single-file code review using MiniMax-M2 (Ollama).")
+
+@app.post("/review/file", response_model=ReviewResponse)
+async def review_file(file: UploadFile = File(...), persona: str = Form("general")):
+    if not file.filename:
+        raise HTTPException(status_code=400, detail="Missing filename")
+    temp_path = save_upload_temp(file)
+    code = read_file_text(temp_path)
+    ext = os.path.splitext(file.filename)[1].lower()
+    language = "python" if ext in (".py",) else "unknown"
+    analyzer_evidence = quick_analyze(temp_path, language)
+    review = review_single_file(temp_path, file.filename, code, analyzer_evidence, persona=persona)
+    try: os.remove(temp_path)
+    except OSError: pass
+    return JSONResponse(status_code=200, content=review.dict())
+
+@app.get("/health")
+async def health():
+    return {"status": "ok", "model": os.getenv("OLLAMA_MODEL", "Minimax-M2"), "engine": "Ollama"}
+
+@app.get("/")
+async def root():
+    return {
+        "message": "Welcome to Code Review Agent v0.1",
+        "usage": {
+            "endpoint": "/review/file",
+            "method": "POST",
+            "params": {"file": "upload your source file", "persona": "general|security|performance|style"},
+            "example": "curl -F 'file=@example.py' -F 'persona=security' http://localhost:8080/review/file"
+        }
+    }

src/core/analyzer.py

@@ -0,0 +1,33 @@
+"""
+Very simple static analyzer (Python only for v0.1)
+"""
+import subprocess
+import shutil
+from typing import List, Dict
+
+def run_python_flake(path: str) -> List[Dict]:
+    if shutil.which("flake8") is None:
+        return []
+    try:
+        out = subprocess.check_output(["flake8", path, "--format=%(row)d:%(col)d:%(code)s:%(text)s"], stderr=subprocess.STDOUT)
+        text = out.decode("utf-8", errors="ignore").strip()
+        findings = []
+        for line in text.splitlines():
+            parts = line.split(":", 3)
+            if len(parts) == 4:
+                row, col, code, msg = parts
+                findings.append({
+                    "tool": "flake8",
+                    "code": code,
+                    "line": int(row),
+                    "col": int(col),
+                    "message": msg.strip()
+                })
+        return findings
+    except subprocess.CalledProcessError:
+        return []
+
+def quick_analyze(path: str, language: str) -> List[Dict]:
+    if language == "python":
+        return run_python_flake(path)
+    return []

src/core/llm_client.py

@@ -0,0 +1,44 @@
+"""
+Ollama client wrapper for MiniMax-M2.
+Compatible with /api/generate streaming endpoint.
+"""
+import os
+import requests
+import json
+from typing import Dict
+
+OLLAMA_API_URL = os.getenv("OLLAMA_API_URL", "http://127.0.0.1:11434")
+OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "minimax-m2:cloud")
+DEFAULT_TEMPERATURE = float(os.getenv("DEFAULT_TEMPERATURE", 0.2))
+MAX_TOKENS = int(os.getenv("MAX_TOKENS", 1024))
+
+def ask_ollama(prompt: str, extra: Dict = None) -> str:
+    """Send a prompt to the local Ollama API and return the generated text."""
+    payload = {
+        "model": OLLAMA_MODEL,
+        "prompt": prompt,
+        "options": {
+            "temperature": DEFAULT_TEMPERATURE,
+            "num_predict": MAX_TOKENS
+        }
+    }
+
+    if extra:
+        payload.update(extra)
+
+    url = f"{OLLAMA_API_URL}/api/generate"
+    resp = requests.post(url, json=payload, stream=True, timeout=180)
+    resp.raise_for_status()
+
+    response_text = ""
+    for line in resp.iter_lines():
+        if not line:
+            continue
+        try:
+            data = json.loads(line.decode("utf-8"))
+            if "response" in data:
+                response_text += data["response"]
+        except Exception:
+            continue
+
+    return response_text.strip()

src/core/prompts.py

@@ -0,0 +1,23 @@
+BASE_SYSTEM = (
+    "You are a helpful code reviewer. For each finding provide: title, severity (blocker/high/medium/low/nit), "
+    "a short description, and a suggested fix if possible. Always include confidence percent (0-100). "
+    "If you are unsure, say 'confidence <60%'."
+)
+
+PERSONA_PROMPTS = {
+    "general": "Review for readability and correctness.",
+    "security": "Focus on security issues like injections, unsafe crypto, or secrets.",
+    "performance": "Focus on performance and complexity issues.",
+    "style": "Focus on code style, clarity, and idiomatic improvements."
+}
+
+def build_review_prompt(filename: str, code: str, analyzer_evidence: list, persona: str = "general") -> str:
+    prompt = BASE_SYSTEM + "\n\n"
+    prompt += f"Persona: {PERSONA_PROMPTS.get(persona, PERSONA_PROMPTS['general'])}\n\n"
+    prompt += f"Filename: {filename}\nCode:\n{code[:20000]}\n\n"
+    if analyzer_evidence:
+        prompt += "Static analyzer findings:\n"
+        for ev in analyzer_evidence[:10]:
+            prompt += str(ev) + "\n"
+    prompt += "\nReturn JSON array of findings."
+    return prompt

src/core/reviewer.py

@@ -0,0 +1,44 @@
+import uuid
+import json
+from typing import List
+from .llm_client import ask_ollama
+from .prompts import build_review_prompt
+from ..models.schemas import Finding, ReviewResponse
+
+def make_id() -> str:
+    return str(uuid.uuid4())[:8]
+
+def review_single_file(path: str, filename: str, code: str, analyzer_evidence: List[dict], persona: str = "general") -> ReviewResponse:
+    prompt = build_review_prompt(filename, code, analyzer_evidence, persona)
+    raw = ask_ollama(prompt)
+
+    findings = []
+    try:
+        arr = json.loads(raw)
+        for item in arr:
+            findings.append(Finding(
+                id=item.get("id", make_id()),
+                title=item.get("title", "Untitled"),
+                severity=item.get("severity", "low"),
+                file=filename,
+                line_range=tuple(item.get("line_range")) if item.get("line_range") else None,
+                description=item.get("description", ""),
+                suggested_fix=item.get("suggested_fix"),
+                confidence=int(item.get("confidence", 50)),
+                evidence=item.get("evidence")
+            ))
+    except Exception:
+        findings.append(Finding(
+            id=make_id(),
+            title="Raw Review",
+            severity="low",
+            file=filename,
+            line_range=None,
+            description=(raw or "No response from LLM")[:3000],
+            suggested_fix=None,
+            confidence=50,
+            evidence=None
+        ))
+
+    summary = f"Found {len(findings)} issues (persona={persona})."
+    return ReviewResponse(summary=summary, findings=findings)

src/models/schemas.py

@@ -0,0 +1,17 @@
+from pydantic import BaseModel
+from typing import List, Optional, Tuple
+
+class Finding(BaseModel):
+    id: str
+    title: str
+    severity: str  # blocker, high, medium, low, nit
+    file: str
+    line_range: Optional[Tuple[int, int]] = None
+    description: str
+    suggested_fix: Optional[str] = None
+    confidence: int = 0
+    evidence: Optional[List[dict]] = None
+
+class ReviewResponse(BaseModel):
+    summary: str
+    findings: List[Finding]

src/utils/file_utils.py

@@ -0,0 +1,14 @@
+import os
+from pathlib import Path
+
+def save_upload_temp(upload_file, dest_dir: str = ".tmp_uploads") -> str:
+    Path(dest_dir).mkdir(parents=True, exist_ok=True)
+    filename = upload_file.filename
+    dest_path = os.path.join(dest_dir, filename)
+    with open(dest_path, "wb") as f:
+        f.write(upload_file.file.read())
+    return dest_path
+
+def read_file_text(path: str) -> str:
+    with open(path, "r", encoding="utf-8", errors="ignore") as f:
+        return f.read()

tests/test_basic.py

@@ -0,0 +1,9 @@
+from fastapi.testclient import TestClient
+from src.app import app
+
+client = TestClient(app)
+
+def test_health():
+    r = client.get("/health")
+    assert r.status_code == 200
+    assert r.json()["status"] == "ok"