Newer
Older
cortex-hub / ai-hub / app / core / pipelines / code_reviewer.py
import json
from typing import List, Dict, Any, Tuple, Optional, Callable

PROMPT_TEMPLATE = """
### ­ЪДа Core Directives

### **Code Review Directives**
Your role is a specialized code review AI. Your primary task is to review a set of code changes and confirm they **fully and accurately address the user's original request**.

---
### **Critical Constraints**
Your review is strictly limited to **code content completeness**. Do not suggest or perform any file splits, moves, or large-scale refactoring.
Identify and resolve any missing logic, placeholders (like "unchanged," "same as original," "to-do"), or incomplete code.

Return exactly one JSON object:
{{
  "reasoning": "A detailed explanation of why the decision was made.",
  "decision": "Either 'complete' or 'modify'.",
  "answer": "If 'complete', an empty string. If 'modify', the new execution plan instructions in JSON."
}}

Input:
- `original_question`: {original_question}
- `execution_plan`: {execution_plan}
- `final_code_changes`: {final_code_changes}
- `original_files`: {original_files}
"""

class CodeReviewer:
    def __init__(self):
        pass

    async def forward(
        self,
        original_question: str,
        execution_plan: str,
        final_code_changes: List[Dict[str, Any]],
        original_files: List[Dict[str, Any]],
        llm_provider = None
    ) -> Tuple[str, str, str]:
        if not llm_provider:
             raise ValueError("LLM Provider is required.")

        final_code_changes_json = json.dumps(final_code_changes)
        original_files_json = json.dumps(original_files)
        
        prompt = PROMPT_TEMPLATE.format(
            original_question=original_question,
            execution_plan=execution_plan,
            final_code_changes=final_code_changes_json,
            original_files=original_files_json
        )

        response = await llm_provider.acompletion(prompt=prompt, response_format={"type": "json_object"})
        content = response.choices[0].message.content
        
        try:
            data = json.loads(content)
            decision = data.get("decision", "modify")
            reasoning = data.get("reasoning", "")
            answer = data.get("answer", "")
            if isinstance(answer, list):
                answer = json.dumps(answer)
            return decision, reasoning, answer
        except json.JSONDecodeError:
            return "modify", f"Failed to parse JSON: {content}", ""