Groq Code Review & Best Practices Agent

groq_code_review_agent/
├── main.py                    # Entry point with async main() function
├── agent.py                   # Agent creation with Groq configuration
├── schemas.py                 # Pydantic output schemas
├── task_builder.py            # Task description builder
├── upsonic_configs.json       # Upsonic CLI configuration
└── README.md                  # Quick start guide

# Required: Set Groq API key
export GROQ_API_KEY="your-groq-api-key"

# Install dependencies from upsonic_configs.json
upsonic install

# Add a package
upsonic add <package> <section>
upsonic add pandas api

# Remove a package
upsonic remove <package> <section>
upsonic remove streamlit api

uv run main.py

upsonic run

curl -X POST http://localhost:8000/call \
  -H "Content-Type: application/json" \
  -d '{
    "code": "def get_user(id):\n    return db.query(\"SELECT * FROM users WHERE id=\" + id)",
    "language": "python",
    "focus_areas": ["security"]
  }'

================================================================================
🔍 CODE REVIEW COMPLETED SUCCESSFULLY
================================================================================

📋 Language: python
🎯 Focus Areas: security, performance, best_practices
⭐ Overall Rating: NEEDS_IMPROVEMENT

--------------------------------------------------------------------------------
📝 SUMMARY
--------------------------------------------------------------------------------
The code has several issues, including a critical SQL injection vulnerability,
high-severity input validation issue, and medium-severity performance issue.

--------------------------------------------------------------------------------
🚨 ISSUES FOUND (4)
--------------------------------------------------------------------------------

1. 🔴 [CRITICAL] SQL Injection Vulnerability
   Category: security
   Location: query = "SELECT * FROM users WHERE name = '" + user_input + "'"
   Suggestion: Use parameterized queries
   Example: query = "SELECT * FROM users WHERE name = ?"

2. 🟠 [HIGH] Input Validation
   Category: security
   Suggestion: Validate user input

3. 🟡 [MEDIUM] Inefficient User Lookup
   Category: performance
   Suggestion: Use a dictionary for user lookup

--------------------------------------------------------------------------------
🔒 SECURITY ANALYSIS
--------------------------------------------------------------------------------
   Risk Level: HIGH
   Vulnerabilities Found: 1
   OWASP Categories: A03:2021-Injection
   Recommendations:
     • Use parameterized queries
     • Validate user input

--------------------------------------------------------------------------------
⚡ PERFORMANCE ANALYSIS
--------------------------------------------------------------------------------
   Complexity Issues:
     • Inefficient user lookup
   Optimization Opportunities:
     • Use a dictionary for user lookup

--------------------------------------------------------------------------------
📊 CODE QUALITY METRICS
--------------------------------------------------------------------------------
   Readability: good
   Maintainability: fair
   Documentation: fair
   Test Coverage: Write unit tests for each function

--------------------------------------------------------------------------------
✅ POSITIVE ASPECTS
--------------------------------------------------------------------------------
   • Good naming conventions
   • Clear code structure

--------------------------------------------------------------------------------
🎯 PRIORITY FIXES
--------------------------------------------------------------------------------
   1. SQL injection vulnerability
   2. Input validation
   3. Inefficient user lookup

================================================================================

report: CodeReviewOutput = result["review_report"]
print(report.overall_rating)  # "needs_improvement"
print(report.security_analysis.risk_level)  # "high"
for issue in report.issues:
    print(f"{issue.severity}: {issue.title}")

from __future__ import annotations

from typing import Dict, Any

from upsonic import Task

try:
    from .agent import create_code_review_agent
    from .task_builder import build_review_task
    from .schemas import CodeReviewOutput
except ImportError:
    from agent import create_code_review_agent
    from task_builder import build_review_task
    from schemas import CodeReviewOutput


async def main(inputs: Dict[str, Any]) -> Dict[str, Any]:
    """
    Main function for code review and best practices analysis.
    
    Args:
        inputs: Dictionary containing:
            - code: The code snippet to review (required)
            - language: Programming language of the code (required)
            - focus_areas: Optional list of areas to focus on (security, performance, etc.)
            - context: Optional context about the codebase or project
            - model: Optional model identifier (default: "groq/llama-3.3-70b-versatile")
    
    Returns:
        Dictionary containing comprehensive code review
    """
    code = inputs.get("code")
    if not code:
        raise ValueError("code is required in inputs")
    
    language = inputs.get("language")
    if not language:
        raise ValueError("language is required in inputs")
    
    focus_areas = inputs.get("focus_areas", [])
    context = inputs.get("context")
    model = inputs.get("model", "groq/llama-3.3-70b-versatile")
    
    agent = create_code_review_agent(model=model)
    
    task_description = build_review_task(
        code=code,
        language=language,
        focus_areas=focus_areas,
        context=context,
    )
    
    task = Task(task_description, response_format=CodeReviewOutput)
    
    result = await agent.do_async(task)
    
    return {
        "language": language,
        "focus_areas": focus_areas,
        "review_report": result,
        "review_completed": True,
    }



if __name__ == "__main__":
    import asyncio
    import json
    import sys
    
    test_code = '''
def calculate_discount(price, discount_percent):
    if discount_percent > 100:
        discount_percent = 100
    final_price = price - (price * discount_percent / 100)
    return final_price

def process_user_data(user_input):
    query = "SELECT * FROM users WHERE name = '" + user_input + "'"
    return execute_query(query)

class UserManager:
    users = []
    
    def add_user(self, name, email):
        self.users.append({"name": name, "email": email})
    
    def get_user(self, name):
        for user in self.users:
            if user["name"] == name:
                return user
        return None
'''
    
    test_inputs = {
        "code": test_code,
        "language": "python",
        "focus_areas": ["security", "performance", "best_practices"],
        "context": "E-commerce application backend",
        "model": "groq/llama-3.3-70b-versatile",
    }
    
    if len(sys.argv) > 1:
        try:
            with open(sys.argv[1], "r") as f:
                test_inputs = json.load(f)
        except Exception as e:
            print(f"Error loading JSON file: {e}")
            print("Using default test inputs")
    
    async def run_main():
        try:
            result = await main(test_inputs)
            report: CodeReviewOutput = result.get('review_report')
            
            print("\n" + "=" * 80)
            print("🔍 CODE REVIEW COMPLETED SUCCESSFULLY")
            print("=" * 80)
            
            print(f"\n📋 Language: {result.get('language')}")
            print(f"🎯 Focus Areas: {', '.join(result.get('focus_areas', []))}")
            print(f"⭐ Overall Rating: {report.overall_rating.upper()}")
            
            print("\n" + "-" * 80)
            print("📝 SUMMARY")
            print("-" * 80)
            print(report.summary)
            
            print("\n" + "-" * 80)
            print(f"🚨 ISSUES FOUND ({len(report.issues)})")
            print("-" * 80)
            for i, issue in enumerate(report.issues, 1):
                severity_icons = {
                    "critical": "🔴",
                    "high": "🟠", 
                    "medium": "🟡",
                    "low": "🟢",
                    "info": "🔵"
                }
                icon = severity_icons.get(issue.severity, "⚪")
                print(f"\n{i}. {icon} [{issue.severity.upper()}] {issue.title}")
                print(f"   Category: {issue.category}")
                if issue.line_reference:
                    print(f"   Location: {issue.line_reference}")
                print(f"   Description: {issue.description}")
                print(f"   Suggestion: {issue.suggestion}")
                if issue.code_example:
                    print(f"   Example: {issue.code_example}")
            
            print("\n" + "-" * 80)
            print("🔒 SECURITY ANALYSIS")
            print("-" * 80)
            sec = report.security_analysis
            print(f"   Risk Level: {sec.risk_level.upper()}")
            print(f"   Vulnerabilities Found: {sec.vulnerabilities_found}")
            if sec.owasp_categories:
                print(f"   OWASP Categories: {', '.join(sec.owasp_categories)}")
            if sec.recommendations:
                print("   Recommendations:")
                for rec in sec.recommendations:
                    print(f"     • {rec}")
            
            print("\n" + "-" * 80)
            print("⚡ PERFORMANCE ANALYSIS")
            print("-" * 80)
            perf = report.performance_analysis
            if perf.complexity_issues:
                print("   Complexity Issues:")
                for issue in perf.complexity_issues:
                    print(f"     • {issue}")
            if perf.memory_concerns:
                print("   Memory Concerns:")
                for concern in perf.memory_concerns:
                    print(f"     • {concern}")
            if perf.optimization_opportunities:
                print("   Optimization Opportunities:")
                for opp in perf.optimization_opportunities:
                    print(f"     • {opp}")
            
            print("\n" + "-" * 80)
            print("📊 CODE QUALITY METRICS")
            print("-" * 80)
            quality = report.code_quality
            print(f"   Readability: {quality.readability_score}")
            print(f"   Maintainability: {quality.maintainability_score}")
            print(f"   Documentation: {quality.documentation_quality}")
            print(f"   Test Coverage: {quality.test_coverage_suggestion}")
            
            if report.positive_aspects:
                print("\n" + "-" * 80)
                print("✅ POSITIVE ASPECTS")
                print("-" * 80)
                for aspect in report.positive_aspects:
                    print(f"   • {aspect}")
            
            print("\n" + "-" * 80)
            print("🎯 PRIORITY FIXES")
            print("-" * 80)
            for i, fix in enumerate(report.priority_fixes, 1):
                print(f"   {i}. {fix}")
            
            if report.learning_resources:
                print("\n" + "-" * 80)
                print("📚 LEARNING RESOURCES")
                print("-" * 80)
                for resource in report.learning_resources:
                    print(f"   • {resource}")
            
            print("\n" + "=" * 80)
            
        except Exception as e:
            print(f"\n❌ Error during execution: {e}")
            import traceback
            traceback.print_exc()
            sys.exit(1)
    
    asyncio.run(run_main())

result = main(inputs)
report: CodeReviewOutput = result["review_report"]

# Access structured fields
print(report.summary)
print(report.overall_rating)

# Iterate over issues
for issue in report.issues:
    if issue.severity == "critical":
        print(f"🔴 {issue.title}: {issue.suggestion}")

# Access nested analysis
print(report.security_analysis.owasp_categories)
print(report.performance_analysis.optimization_opportunities)

"""
Code Review Agent creation and configuration.

Creates the main Agent that performs comprehensive code reviews
using Groq's fast inference capabilities with web search for best practices.
"""

from __future__ import annotations

from typing import Optional, List

from upsonic import Agent
from upsonic.tools.common_tools.duckduckgo import duckduckgo_search_tool


def create_code_review_agent(
    model: str = "groq/llama-3.3-70b-versatile",
    tools: Optional[List] = None,
) -> Agent:
    """Create the code review agent with Groq model.
    
    Args:
        model: Groq model identifier for the agent
        tools: Optional list of additional tools
        
    Returns:
        Configured Agent instance for code review
    """
    ddg_search = duckduckgo_search_tool(duckduckgo_client=None, max_results=5)
    
    agent_tools = [ddg_search]
    if tools:
        agent_tools.extend(tools)
    
    agent = Agent(
        model=model,
        name="code-review-agent",
        role="Senior Software Engineer & Code Reviewer",
        goal="Provide comprehensive code reviews with actionable feedback on security, performance, best practices, and code quality",
        system_prompt="""You are an expert senior software engineer with 15+ years of experience 
        in code review and software architecture. Your expertise spans multiple programming languages 
        and you have deep knowledge of:
        
        - Security vulnerabilities and secure coding practices
        - Performance optimization and algorithmic efficiency
        - Design patterns and software architecture
        - Clean code principles and maintainability
        - Testing strategies and code coverage
        - Industry best practices and coding standards
        
        When reviewing code:
        1. Identify potential bugs and logic errors
        2. Detect security vulnerabilities (SQL injection, XSS, buffer overflows, etc.)
        3. Suggest performance improvements
        4. Recommend better design patterns or abstractions
        5. Point out code style and readability issues
        6. Suggest appropriate test cases
        
        Use web search to find current best practices and industry standards when needed.
        
        Always provide:
        - Clear explanation of issues found
        - Severity level (Critical, High, Medium, Low)
        - Specific code suggestions for fixes
        - References to relevant documentation or best practices
        
        **CRITICAL JSON FORMATTING REQUIREMENTS**:
        - NEVER include duplicate keys in your JSON response (each key must appear only once)
        - Use plain text strings for suggestion, code_example, description, and title fields - DO NOT wrap them in JSON objects
        - Escape quotes inside strings with backslash: \\" (e.g., "query = \\"SELECT * FROM users\\"")
        - Ensure all strings are properly closed with quotes
        - Validate your JSON is parseable before returning it
        - Example: "suggestion": "Use parameterized queries" (plain string, NOT "{ \\"fix\\": ... }")
        
        Be constructive and educational in your feedback. Help developers understand 
        not just what to fix, but why.""",
        tools=agent_tools,
        tool_call_limit=10,
    )
    
    return agent


def create_security_focused_agent(
    model: str = "groq/llama-3.1-8b-instant",
) -> Agent:
    """Create a security-focused code review agent.
    
    Args:
        model: Groq model identifier for the agent
        
    Returns:
        Configured Agent instance for security review
    """
    ddg_search = duckduckgo_search_tool(duckduckgo_client=None, max_results=5)
    
    return Agent(
        model=model,
        name="security-review-agent",
        role="Application Security Specialist",
        goal="Identify and report security vulnerabilities in code with remediation guidance",
        system_prompt="""You are a security expert specializing in application security and 
        secure coding practices. Your focus is on identifying:
        
        - SQL Injection vulnerabilities
        - Cross-Site Scripting (XSS)
        - Cross-Site Request Forgery (CSRF)
        - Insecure Direct Object References
        - Security Misconfiguration
        - Sensitive Data Exposure
        - Authentication/Authorization flaws
        - Input validation issues
        - Cryptographic weaknesses
        - Race conditions and timing attacks
        
        For each vulnerability found:
        1. Explain the attack vector
        2. Demonstrate potential exploit scenarios
        3. Provide secure code alternatives
        4. Reference OWASP guidelines when applicable
        
        Use web search to find current CVEs and security advisories related to the 
        libraries or patterns being used.""",
        tools=[ddg_search],
        tool_call_limit=8,
    )


def create_performance_focused_agent(
    model: str = "groq/llama-3.1-8b-instant",
) -> Agent:
    """Create a performance-focused code review agent.
    
    Args:
        model: Groq model identifier for the agent
        
    Returns:
        Configured Agent instance for performance review
    """
    ddg_search = duckduckgo_search_tool(duckduckgo_client=None, max_results=5)
    
    return Agent(
        model=model,
        name="performance-review-agent",
        role="Performance Engineering Specialist",
        goal="Analyze code for performance bottlenecks and optimization opportunities",
        system_prompt="""You are a performance engineering specialist with expertise in:
        
        - Algorithmic complexity analysis (Big O notation)
        - Memory management and optimization
        - Database query optimization
        - Caching strategies
        - Concurrent programming and parallelism
        - I/O optimization
        - Profiling and benchmarking
        
        When analyzing code:
        1. Identify inefficient algorithms or data structures
        2. Spot memory leaks or excessive allocations
        3. Find N+1 query problems and database issues
        4. Suggest caching opportunities
        5. Identify blocking operations that could be async
        6. Recommend profiling strategies
        
        Use web search to find current benchmarks and performance best practices 
        for the specific language and framework being used.""",
        tools=[ddg_search],
        tool_call_limit=8,
    )

"""
Output schemas for code review agent.

Defines structured Pydantic models for type-safe outputs from the
code review analysis.
"""

from __future__ import annotations

from typing import List, Optional, Literal
from pydantic import BaseModel, Field


class CodeIssue(BaseModel):
    """Represents a single code issue found during review."""
    
    severity: Literal["critical", "high", "medium", "low", "info"] = Field(
        description="Severity level of the issue"
    )
    category: str = Field(
        description="Category of the issue (e.g., security, performance, style, bug)"
    )
    line_reference: Optional[str] = Field(
        default=None,
        description="Line number or code section reference"
    )
    title: str = Field(
        description="Brief title describing the issue"
    )
    description: str = Field(
        description="Detailed description of the issue"
    )
    suggestion: str = Field(
        description="Suggested fix or improvement as a plain text string. Do NOT wrap in JSON objects. Example: 'Use parameterized queries or prepared statements'"
    )
    code_example: Optional[str] = Field(
        default=None,
        description="Example of correct/improved code as a plain text string. Do NOT wrap in JSON objects. Escape quotes with backslash. Example: 'query = \\\"SELECT * FROM users WHERE name = ?\\\"'"
    )


class SecurityAnalysis(BaseModel):
    """Security-focused analysis results."""
    
    vulnerabilities_found: int = Field(
        description="Number of security vulnerabilities found"
    )
    risk_level: Literal["critical", "high", "medium", "low", "none"] = Field(
        description="Overall security risk level"
    )
    owasp_categories: List[str] = Field(
        default_factory=list,
        description="Relevant OWASP categories for issues found"
    )
    recommendations: List[str] = Field(
        default_factory=list,
        description="Security recommendations"
    )


class PerformanceAnalysis(BaseModel):
    """Performance-focused analysis results."""
    
    complexity_issues: List[str] = Field(
        default_factory=list,
        description="Algorithmic complexity concerns"
    )
    memory_concerns: List[str] = Field(
        default_factory=list,
        description="Memory usage concerns"
    )
    optimization_opportunities: List[str] = Field(
        default_factory=list,
        description="Potential performance optimizations"
    )


class CodeQualityMetrics(BaseModel):
    """Code quality assessment metrics."""
    
    readability_score: Literal["excellent", "good", "fair", "poor"] = Field(
        description="Code readability assessment"
    )
    maintainability_score: Literal["excellent", "good", "fair", "poor"] = Field(
        description="Code maintainability assessment"
    )
    test_coverage_suggestion: str = Field(
        description="Suggestions for test coverage"
    )
    documentation_quality: Literal["excellent", "good", "fair", "poor", "missing"] = Field(
        description="Documentation quality assessment"
    )


class CodeReviewOutput(BaseModel):
    """Complete code review output."""
    
    summary: str = Field(
        description="Executive summary of the code review"
    )
    overall_rating: Literal["excellent", "good", "needs_improvement", "poor", "critical"] = Field(
        description="Overall code quality rating"
    )
    issues: List[CodeIssue] = Field(
        default_factory=list,
        description="List of all issues found"
    )
    security_analysis: SecurityAnalysis = Field(
        description="Security analysis results"
    )
    performance_analysis: PerformanceAnalysis = Field(
        description="Performance analysis results"
    )
    code_quality: CodeQualityMetrics = Field(
        description="Code quality metrics"
    )
    positive_aspects: List[str] = Field(
        default_factory=list,
        description="Positive aspects of the code"
    )
    priority_fixes: List[str] = Field(
        default_factory=list,
        description="Top priority items to fix, in order"
    )
    learning_resources: List[str] = Field(
        default_factory=list,
        description="Recommended resources for improvement"
    )

{
    "envinroment_variables": {
        "UPSONIC_WORKERS_AMOUNT": {
            "type": "number",
            "description": "The number of workers for the Upsonic API",
            "default": 1
        },
        "API_WORKERS": {
            "type": "number",
            "description": "The number of workers for the Upsonic API",
            "default": 1
        },
        "RUNNER_CONCURRENCY": {
            "type": "number",
            "description": "The number of runners for the Upsonic API",
            "default": 1
        },
        "GROQ_API_KEY": {
            "type": "string",
            "description": "Groq API key for authentication",
            "required": true
        }
    },
    "machine_spec": {
        "cpu": 1,
        "memory": 2048,
        "storage": 512
    },
    "agent_name": "Groq Code Review & Best Practices Agent",
    "description": "Fast and comprehensive code review agent powered by Groq's ultra-fast inference. Analyzes code for security vulnerabilities, performance issues, best practices, and provides actionable improvement suggestions.",
    "icon": "code",
    "language": "python",
    "streamlit": false,
    "proxy_agent": false,
    "dependencies": {
        "api": [
            "upsonic",
            "upsonic[tools]"
        ],
        "development": [
            "python-dotenv",
            "pytest"
        ]
    },
    "entrypoints": {
        "api_file": "main.py",
        "streamlit_file": "streamlit_app.py"
    },
    "input_schema": {
        "inputs": {
            "code": {
                "type": "string",
                "description": "The code snippet to review (required)",
                "required": true,
                "default": null
            },
            "language": {
                "type": "string",
                "description": "Programming language of the code (e.g., python, javascript, java)",
                "required": true,
                "default": null
            },
            "focus_areas": {
                "type": "array",
                "description": "Optional list of areas to focus on (security, performance, best_practices, style)",
                "required": false,
                "default": []
            },
            "context": {
                "type": "string",
                "description": "Optional context about the codebase or project",
                "required": false,
                "default": null
            },
            "model": {
                "type": "string",
                "description": "Groq model identifier (e.g., groq/llama-3.3-70b-versatile, groq/llama-3.1-8b-instant)",
                "required": false,
                "default": "groq/llama-3.3-70b-versatile"
            }
        }
    },
    "output_schema": {
        "language": {
            "type": "string",
            "description": "The programming language of the reviewed code"
        },
        "focus_areas": {
            "type": "array",
            "description": "The focus areas that were analyzed"
        },
        "review_report": {
            "type": "object",
            "description": "Comprehensive code review report with issues, suggestions, and metrics"
        },
        "review_completed": {
            "type": "boolean",
            "description": "Whether the review was successfully completed"
        }
    }
}

from upsonic import Task
from schemas import CodeReviewOutput

task = Task(task_description, response_format=CodeReviewOutput)
result = agent.do(task)  # Returns a CodeReviewOutput instance