Spaces:
Sleeping
Sleeping
File size: 3,906 Bytes
b0636b2 8340baf d3bf5af 8340baf 3cf275a c8df408 3cf275a a3859cd 80b64e3 a79668c 4d69fe0 a3859cd d939b2e a3859cd c1d93b2 aefa24e 1f3e2cc 3cf275a 1f3e2cc 2a19d7a 1f3e2cc aefa24e a3859cd a79668c a3859cd d3bf5af a3859cd 227aada a3859cd a79668c a3859cd 5457bc2 51327a5 a79668c 51327a5 5457bc2 a79668c 5457bc2 a79668c 5457bc2 1251fb0 a79668c d939b2e a79668c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | import asyncio
from app.core.llm import call_llm_for_json
from app.core.logger import get_logger
from app.core.prompts import AUTO_FIX_PROMPT, CODE_REVIEW_PROMPT
from app.models.repository import RepositoryMetadata
from app.models.review import AutoFix, ReviewSuggestions
from app.tools.file_scanner import read_source_samples, get_python_files
from app.tools.github_tool import get_changed_files
import os
logger = get_logger(__name__)
async def run(local_path: str, metadata: RepositoryMetadata) -> ReviewSuggestions:
"""
Perform AI-powered code review on repository.
"""
logger.info("Starting Code Review Agent", extra={"path": local_path})
# Step 1 - Get Python files
base_sha = metadata.base_sha
head_sha = metadata.head_sha
changed_files = []
if base_sha and head_sha:
changed_files = get_changed_files(local_path, base_sha, head_sha)
#Step 2 - Read source code samples
all_files = changed_files or [
f for f in get_python_files(local_path)
if not os.path.basename(f).startswith("test_")
and not os.path.basename(f).endswith("_test.py")
]
source_samples = read_source_samples(
local_path,
max_files=4,
max_chars=2500,
file_list=all_files,
)
if not source_samples:
return ReviewSuggestions(
summary="No source files found to review.", overall_score=0.0
)
# Step 3 - Call LLM and parse JSON in one shot
source_code = "\n\n".join(source_samples)
prompt = CODE_REVIEW_PROMPT.format(source_code=source_code)
llm_data = await call_llm_for_json(prompt, task="code_review")
review = ReviewSuggestions(
solid_violations=llm_data.get("solid_violations", []),
duplicate_code=llm_data.get("duplicate_code", []),
refactor_suggestions=llm_data.get("refactor_suggestions", []),
overall_score=float(llm_data.get("overall_score", 5.0)),
summary=llm_data.get("summary", ""),
)
# Generate auto-fixes for top 3 violations (cap LLM calls)
auto_fixes: list[AutoFix] = []
top_findings = (review.solid_violations + review.refactor_suggestions)[:3]
if top_findings and source_samples:
first_file_content = source_samples[0] # already read
fix_tasks = [
call_llm_for_json(
AUTO_FIX_PROMPT.format(
source_code=first_file_content[:1500], finding=finding
)
)
for finding in top_findings
]
fix_results = await asyncio.gather(*fix_tasks, return_exceptions=True)
for res in fix_results:
if isinstance(res, dict) and res.get("fixed_snippet"):
auto_fixes.append(AutoFix(**res))
review = review.model_copy(update={"auto_fixes": auto_fixes})
# Second pass — if score is poor, ask LLM for a prioritised fix list
if review.overall_score < 5.0:
logger.info(
"Low score detected, running focused follow-up pass",
extra={"score": review.overall_score},
)
followup_prompt = (
f"The following code scored {review.overall_score}/10 in a review.\n\n"
f"Top violations: {review.solid_violations[:3]}\n\n"
f'Return ONLY a JSON object with one key: "priority_fixes": a list of up to 3 strings, '
f"each describing the single most impactful change to make first."
)
followup_data = await call_llm_for_json(followup_prompt)
priority_fixes = followup_data.get("priority_fixes", [])
if isinstance(priority_fixes, list) and priority_fixes:
review = review.model_copy(
update={
"refactor_suggestions": priority_fixes + review.refactor_suggestions
}
)
logger.info("Code review complete", extra={"score": review.overall_score})
return review
|