Saturday, November 22, 2025

The Rise of Small Language Models: A Practical Guide to Choosing SLMs Over Giants

 If we’ve been following the AI space, it feels like the narrative has been dominated by one thing: bigger is better. We've watched parameter counts soar into the hundreds of billions, with each new model claiming to be more powerful than the last.

But a quiet, revolutionary counter-trend is gaining massive momentum: Small Language Models (SLMs).

Models like Microsoft's Phi-3, Meta's Llama 3, and Mistral 7B are demonstrating that you don't always need a nuclear reactor to power a lightbulb. These smaller, more refined models are proving to be highly effective for a vast range of specific tasks, offering a compelling alternative to their gargantuan counterparts.

The question is no longer "What's the most powerful model?" but rather "What's the most appropriate model for my specific need?"

This guide will walk you through the key trade-offs—cost, latency, and data privacy—to help you decide when an SLM is the right tool for the job.

What Exactly is a Small Language Model (SLM)?

An SLM is a language model that is significantly smaller in parameter count (typically ranging from a few hundred million to around 10 billion parameters) and computational footprint than massive foundation models like GPT-4 or Claude 3 Opus. Their power doesn't come from brute-force scaling but from:

  • Better, Curated Training Data: Models like Phi-3 are trained on meticulously filtered, high-quality "textbook-quality" data, which leads to more efficient learning.

  • Innovative Architectures: Techniques like sliding window attention (from Mistral) and other optimizations make these models smarter with fewer resources.

  • Strategic Fine-Tuning: They are often designed and fine-tuned for specific domains or tasks from the outset.

When to Choose an SLM: The Three-Way Trade-Off

Choosing between an SLM and a large foundation model is a balancing act. Here’s your decision-making framework.

1. Cost: The Bottom Line

  • The Problem with Giants: Running inference on a model with hundreds of billions of parameters is incredibly expensive. Every API call adds up, and the costs for fine-tuning or training are astronomical. This can quickly become prohibitive for startups, SMEs, or projects with a tight budget.

  • The SLM Advantage: SLMs are dramatically cheaper to run. You can host a powerful 7B-parameter model on a single, affordable GPU instance (or even on CPU). This makes them perfect for:

    • High-Volume Tasks: Applications that require thousands or millions of API calls per day.

    • Prototyping and MVPs: Testing an AI feature without burning through your seed funding.

    • Cost-Sensitive Production Workloads: Any application where the cost per query is a primary concern.

Choose an SLM when: Your project is budget-conscious or requires a positive unit economics where the cost of the AI call is a small fraction of the value it provides.

2. Latency & Speed: The Need for Speed

  • The Problem with Giants: Large models are slow. Processing a single request can take several seconds, as it requires moving massive amounts of data through the model. This leads to high latency, which can ruin user experience in real-time applications.

  • The SLM Advantage: With their smaller size, SLMs offer lightning-fast inference. Response times can be in the low milliseconds. This is critical for:

    • Real-Time Applications: Live chatbots, customer service interfaces, or interactive assistants where a delay of even one second feels sluggish.

    • Edge Computing: Deploying AI directly on devices like phones, laptops, or IoT hardware where resources are limited and instant response is key.

    • User-Facing Features: Any application where a snappy, responsive feel is crucial for adoption.

Choose an SLM when: Your application demands low latency and a fast, seamless user experience.

3. Data Privacy & Control: Keeping It In-House

  • The Problem with Giants: When you use an API from a major provider, your data (including potentially sensitive prompts and outputs) is sent to a third-party server. For industries like healthcare, legal, and finance, this is a non-starter due to compliance regulations (HIPAA, GDPR) and intellectual property concerns.

  • The SLM Advantage: You can run SLMs entirely on your own infrastructure—be it your company's private cloud, a secure on-premise server, or even a fully air-gapped environment. This gives you full control and ownership over your data.

    • Use Cases: Processing confidential legal documents, analyzing private patient data, generating internal financial reports.

Choose an SLM when: Data privacy, security, and regulatory compliance are top priorities.

And When Should You Still Use a Giant Foundation Model?

SLMs are brilliant, but they aren't magical. There are still clear scenarios where a large foundation model is the undisputed champion:

  • For Complex, Creative, or Open-Ended Tasks: If you need highly creative writing, complex reasoning across multiple domains, or nuanced conversation that feels truly human, larger models still have the edge.

  • As a "Generalist" Brain: If you're building a product that needs to be a jack-of-all-trades—handling everything from code generation to poetry to complex analysis in a single interface—a larger model provides more consistent quality across this broad spectrum.

  • When You Have No Idea What Your Users Will Do: For public, exploratory platforms (like ChatGPT), the model needs to be capable of handling any conceivable query, which demands the vast knowledge and capability of a giant model.

The Bottom Line: It's About Fit, Not Just Power

The era of one-size-fits-all AI is over. The future is a diverse ecosystem of models, each optimized for a specific purpose.

Think of it this way:
You use a massive, power-hungry truck to move furniture (the large foundation model), but you use an efficient, nimble compact car for your daily commute (the SLM). Both are vehicles, but you choose the right one for the job.

For most practical business applications—specialized chatbots, content moderation, data extraction, text summarization, and internal automation—a well-chosen and finely-tuned Small Language Model isn't just a cheaper alternative. It's often a superior one—delivering the speed, affordability, and control that modern applications demand.

Ready to experiment? Start by exploring models like Llama 3 8BMistral 7B, or Microsoft's Phi-3 on platforms like Hugging Face. You might be surprised at how much power you can pack into such a small package.

Saturday, November 1, 2025

The AI-Powered DevOps Pipeline: Automating Code Reviews with Python and Node.js

 Excellent concept! AI-powered DevOps pipelines are revolutionizing how teams ship code. Let me show you how to build a comprehensive automated code review system using Python for AI analysis and Node.js for pipeline integration.

System Architecture Overview

text
[GitHub/GitLab] --> [Node.js Webhook Handler] --> [Python AI Analyzer]
       ^                                              |
       |                                              v
[Status Updates] <-- [Node.js Orchestrator] <-- [Review Results & Suggestions]

Core Components

1. Node.js Webhook Server & Orchestrator

javascript
// server.js - Main webhook handler
const express = require('express');
const axios = require('axios');
const crypto = require('crypto');
const { spawn } = require('child_process');

const app = express();
app.use(express.json());

// Configuration
const CONFIG = {
    GITHUB_WEBHOOK_SECRET: process.env.GITHUB_WEBHOOK_SECRET,
    PYTHON_AI_SERVICE: process.env.PYTHON_AI_SERVICE || 'http://localhost:5000',
    SUPPORTED_LANGUAGES: ['javascript', 'python', 'typescript', 'java', 'go']
};

// GitHub webhook verification
function verifyGitHubSignature(req, res, next) {
    const signature = req.headers['x-hub-signature-256'];
    if (!signature) {
        return res.status(401).json({ error: 'Missing signature' });
    }

    const hmac = crypto.createHmac('sha256', CONFIG.GITHUB_WEBHOOK_SECRET);
    const digest = `sha256=${hmac.update(JSON.stringify(req.body)).digest('hex')}`;
    
    if (!crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(digest))) {
        return res.status(401).json({ error: 'Invalid signature' });
    }
    
    next();
}

// AI Code Review Orchestrator
class CodeReviewOrchestrator {
    constructor() {
        this.reviewCache = new Map();
    }

    async performCodeReview(pullRequest) {
        const {
            repository,
            pull_request: pr
        } = pullRequest;

        const cacheKey = `${repository.full_name}#${pr.number}`;
        
        // Check cache for recent review
        if (this.reviewCache.has(cacheKey)) {
            return this.reviewCache.get(cacheKey);
        }

        try {
            // 1. Get PR details and changes
            const prDetails = await this.getPRDetails(repository, pr.number);
            const files = await this.getChangedFiles(repository, pr.number);
            
            // 2. Filter relevant files
            const reviewableFiles = files.filter(file => 
                this.isReviewableFile(file.filename) && 
                file.status !== 'removed'
            );

            // 3. Get file contents
            const fileContents = await this.getFileContents(repository, reviewableFiles);
            
            // 4. Send to Python AI service for analysis
            const aiAnalysis = await this.analyzeWithAI({
                files: fileContents,
                pr_title: pr.title,
                pr_description: pr.body,
                repository: repository.full_name,
                base_branch: pr.base.ref,
                head_branch: pr.head.ref
            });

            // 5. Post review comments to GitHub
            await this.postReviewComments(repository, pr.number, aiAnalysis.comments);

            // 6. Update PR status
            await this.updatePRStatus(repository, pr.head.sha, aiAnalysis.summary);

            // Cache the result
            this.reviewCache.set(cacheKey, aiAnalysis);
            setTimeout(() => this.reviewCache.delete(cacheKey), 300000); // 5 min cache

            return aiAnalysis;

        } catch (error) {
            console.error('Code review failed:', error);
            throw error;
        }
    }

    async analyzeWithAI(reviewContext) {
        try {
            const response = await axios.post(`${CONFIG.PYTHON_AI_SERVICE}/analyze`, reviewContext, {
                timeout: 60000 // 60 second timeout for AI analysis
            });
            return response.data;
        } catch (error) {
            console.error('AI service error:', error.message);
            throw new Error(`AI analysis failed: ${error.message}`);
        }
    }

    async getPRDetails(repo, prNumber) {
        // Implementation for getting PR details from GitHub API
        const response = await axios.get(
            `https://api.github.com/repos/${repo.full_name}/pulls/${prNumber}`,
            { headers: { Authorization: `token ${process.env.GITHUB_TOKEN}` } }
        );
        return response.data;
    }

    async getChangedFiles(repo, prNumber) {
        const response = await axios.get(
            `https://api.github.com/repos/${repo.full_name}/pulls/${prNumber}/files`,
            { headers: { Authorization: `token ${process.env.GITHUB_TOKEN}` } }
        );
        return response.data;
    }

    async getFileContents(repo, files) {
        const contents = [];
        
        for (const file of files) {
            try {
                const response = await axios.get(
                    `https://api.github.com/repos/${repo.full_name}/contents/${file.filename}?ref=${file.contents_url.split('?ref=')[1]}`,
                    { headers: { Authorization: `token ${process.env.GITHUB_TOKEN}` } }
                );
                
                const content = Buffer.from(response.data.content, 'base64').toString();
                contents.push({
                    filename: file.filename,
                    content: content,
                    changes: file.patch, // The diff patch
                    status: file.status
                });
            } catch (error) {
                console.warn(`Could not fetch content for ${file.filename}:`, error.message);
            }
        }
        
        return contents;
    }

    async postReviewComments(repo, prNumber, comments) {
        for (const comment of comments) {
            await axios.post(
                `https://api.github.com/repos/${repo.full_name}/pulls/${prNumber}/comments`,
                {
                    body: this.formatComment(comment),
                    commit_id: comment.commit_id,
                    path: comment.file_path,
                    line: comment.line_number,
                    side: comment.side || 'RIGHT'
                },
                { headers: { Authorization: `token ${process.env.GITHUB_TOKEN}` } }
            );
            
            // Rate limiting
            await new Promise(resolve => setTimeout(resolve, 1000));
        }
    }

    async updatePRStatus(repo, sha, summary) {
        const state = summary.issues_found > 0 ? 'failure' : 'success';
        
        await axios.post(
            `https://api.github.com/repos/${repo.full_name}/statuses/${sha}`,
            {
                state: state,
                target_url: process.env.CI_BUILD_URL,
                description: summary.issues_found > 0 ? 
                    `Found ${summary.issues_found} issues needing attention` :
                    'AI review passed - no critical issues found',
                context: 'ai-code-review/bot'
            },
            { headers: { Authorization: `token ${process.env.GITHUB_TOKEN}` } }
        );
    }

    formatComment(comment) {
        return `
🤖 **AI Code Review**

**${comment.category.toUpperCase()}**: ${comment.title}

${comment.description}

**Suggestion**: ${comment.suggestion}

**Confidence**: ${(comment.confidence * 100).toFixed(0)}%

${comment.example ? `**Example**: \`\`\`${comment.language}\n${comment.example}\n\`\`\`` : ''}
        `.trim();
    }

    isReviewableFile(filename) {
        const extension = filename.split('.').pop();
        const supported = ['js', 'ts', 'py', 'java', 'go', 'cpp', 'c', 'rs', 'php'];
        return supported.includes(extension);
    }
}

// Initialize orchestrator
const orchestrator = new CodeReviewOrchestrator();

// Webhook endpoint
app.post('/webhook/github', verifyGitHubSignature, async (req, res) => {
    const event = req.headers['x-github-event'];
    const payload = req.body;

    // Only process pull request events
    if (event !== 'pull_request') {
        return res.status(200).json({ status: 'ignored', reason: 'Not a pull request event' });
    }

    // Only process opened, synchronize, or reopened PRs
    if (!['opened', 'synchronize', 'reopened'].includes(payload.action)) {
        return res.status(200).json({ status: 'ignored', reason: 'Action not relevant for review' });
    }

    try {
        // Process in background
        setImmediate(async () => {
            try {
                await orchestrator.performCodeReview(payload);
            } catch (error) {
                console.error('Background processing error:', error);
            }
        });

        res.status(202).json({ status: 'accepted', message: 'Code review started' });
    } catch (error) {
        console.error('Webhook processing error:', error);
        res.status(500).json({ error: 'Internal server error' });
    }
});

// Health check endpoint
app.get('/health', (req, res) => {
    res.json({ 
        status: 'healthy', 
        service: 'ai-code-review-orchestrator',
        timestamp: new Date().toISOString()
    });
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
    console.log(`AI Code Review Orchestrator running on port ${PORT}`);
});

2. Python AI Analysis Engine

python
# ai_analyzer.py - Core AI analysis engine
from flask import Flask, request, jsonify
import openai
from anthropic import Anthropic
import os
import re
import ast
import tempfile
import subprocess
from pathlib import Path
from typing import List, Dict, Any
import logging
from dataclasses import dataclass

app = Flask(__name__)

# Configuration
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
ANTHROPIC_API_KEY = os.getenv('ANTHROPIC_API_KEY')
GITHUB_TOKEN = os.getenv('GITHUB_TOKEN')

# Initialize clients
openai_client = openai.OpenAI(api_key=OPENAI_API_KEY) if OPENAI_API_KEY else None
anthropic_client = Anthropic(api_key=ANTHROPIC_API_KEY) if ANTHROPIC_API_KEY else None

@dataclass
class CodeIssue:
    file_path: str
    line_number: int
    category: str
    title: str
    description: str
    suggestion: str
    confidence: float
    severity: str  # low, medium, high, critical
    example: str = ""

class CodeAnalyzer:
    def __init__(self):
        self.supported_patterns = {
            'security': self._analyze_security,
            'performance': self._analyze_performance,
            'maintainability': self._analyze_maintainability,
            'bug_risk': self._analyze_bug_risk,
            'best_practices': self._analyze_best_practices
        }
    
    def analyze_code(self, files: List[Dict], pr_context: Dict) -> Dict[str, Any]:
        """Main analysis entry point"""
        all_issues = []
        
        for file in files:
            file_issues = self._analyze_file(file, pr_context)
            all_issues.extend(file_issues)
        
        # Categorize and summarize
        summary = self._generate_summary(all_issues)
        
        return {
            'issues': all_issues,
            'summary': summary,
            'comments': self._format_for_github(all_issues)
        }
    
    def _analyze_file(self, file: Dict, pr_context: Dict) -> List[CodeIssue]:
        """Analyze a single file"""
        issues = []
        filename = file['filename']
        content = file['content']
        changes = file.get('changes', '')
        
        # Language-specific analysis
        file_extension = Path(filename).suffix.lower()
        
        # Pattern-based analysis
        for category, analyzer in self.supported_patterns.items():
            category_issues = analyzer(filename, content, changes, file_extension)
            issues.extend(category_issues)
        
        # AI-powered analysis
        ai_issues = self._ai_analysis(filename, content, changes, pr_context)
        issues.extend(ai_issues)
        
        return issues
    
    def _analyze_security(self, filename: str, content: str, changes: str, extension: str) -> List[CodeIssue]:
        """Security vulnerability analysis"""
        issues = []
        security_patterns = {
            'javascript': [
                (r'eval\s*\(', 'Use of eval() function', 'high'),
                (r'localStorage\.setItem\s*\([^)]*password', 'Storing passwords in localStorage', 'critical'),
                (r'innerHTML\s*=', 'Potential XSS vulnerability', 'high'),
                (r'https?://[^\s"\']*', 'Hardcoded URLs', 'medium'),
            ],
            'python': [
                (r'subprocess\.call|os\.system', 'Shell command injection risk', 'high'),
                (r'pickle\.loads', 'Unsafe deserialization', 'critical'),
                (r'exec\(|eval\(', 'Code execution vulnerability', 'high'),
                (r'password\s*=\s*["\']', 'Hardcoded credentials', 'critical'),
            ]
        }
        
        lang = 'javascript' if extension in ['.js', '.ts', '.jsx', '.tsx'] else 'python'
        patterns = security_patterns.get(lang, [])
        
        for pattern, description, severity in patterns:
            matches = re.finditer(pattern, content, re.IGNORECASE)
            for match in matches:
                line_number = content[:match.start()].count('\n') + 1
                issues.append(CodeIssue(
                    file_path=filename,
                    line_number=line_number,
                    category='security',
                    title=description,
                    description=f"Potential security vulnerability found",
                    suggestion=f"Consider using safer alternatives and validate inputs",
                    confidence=0.8,
                    severity=severity,
                    example=self._get_example_fix(pattern, lang)
                ))
        
        return issues
    
    def _analyze_performance(self, filename: str, content: str, changes: str, extension: str) -> List[CodeIssue]:
        """Performance issue analysis"""
        issues = []
        
        # Common performance patterns
        performance_patterns = [
            (r'for\s*\([^)]*\)\s*{[\s\S]*?\bfor\s*\([^)]*\)\s*{', 'Nested loops - O(n²) complexity', 'medium'),
            (r'JSON\.parse\s*\([^)]*\)\s*in\s+loop', 'JSON parsing in loop', 'medium'),
            (r'document\.querySelectorAll\s*\([^)]*\)\s*\.forEach', 'DOM query in loop', 'medium'),
        ]
        
        for pattern, description, severity in performance_patterns:
            if re.search(pattern, content, re.MULTILINE):
                issues.append(CodeIssue(
                    file_path=filename,
                    line_number=1,  # Would need more sophisticated line detection
                    category='performance',
                    title=description,
                    description="Performance optimization opportunity",
                    suggestion="Consider optimizing the algorithm or caching results",
                    confidence=0.7,
                    severity=severity
                ))
        
        return issues
    
    def _analyze_maintainability(self, filename: str, content: str, changes: str, extension: str) -> List[CodeIssue]:
        """Code maintainability analysis"""
        issues = []
        
        # Large function detection
        functions = self._extract_functions(content, extension)
        for func_name, func_info in functions.items():
            if func_info['line_count'] > 50:
                issues.append(CodeIssue(
                    file_path=filename,
                    line_number=func_info['start_line'],
                    category='maintainability',
                    title=f"Large function: {func_name} ({func_info['line_count']} lines)",
                    description="Functions should generally be under 50 lines for maintainability",
                    suggestion="Consider breaking this function into smaller, focused functions",
                    confidence=0.9,
                    severity='medium'
                ))
        
        # Complex conditional detection
        complex_conditionals = re.findall(r'if\s*\([^)]{100,}\)', content)
        for conditional in complex_conditionals:
            issues.append(CodeIssue(
                file_path=filename,
                line_number=1,
                category='maintainability',
                title="Complex conditional expression",
                description="Conditional logic is complex and hard to maintain",
                suggestion="Extract complex conditions into well-named variables or functions",
                confidence=0.8,
                severity='low'
            ))
        
        return issues
    
    def _analyze_bug_risk(self, filename: str, content: str, changes: str, extension: str) -> List[CodeIssue]:
        """Potential bug detection"""
        issues = []
        
        bug_patterns = {
            'javascript': [
                (r'==\s*(null|undefined)', 'Use === for null/undefined checks', 'medium'),
                (r'console\.log\(', 'Leftover debug statement', 'low'),
                (r'try\s*{[\s\S]*?}\s*catch\s*\([^)]*\)\s*{}', 'Empty catch block', 'high'),
            ],
            'python': [
                (r'except:\s*pass', 'Bare except with pass', 'high'),
                (r'print\(', 'Leftover debug statement', 'low'),
                (r'assert\s+[^,)]*$', 'Assert without message', 'medium'),
            ]
        }
        
        lang = 'javascript' if extension in ['.js', '.ts'] else 'python'
        patterns = bug_patterns.get(lang, [])
        
        for pattern, description, severity in patterns:
            matches = re.finditer(pattern, content, re.MULTILINE)
            for match in matches:
                line_number = content[:match.start()].count('\n') + 1
                issues.append(CodeIssue(
                    file_path=filename,
                    line_number=line_number,
                    category='bug_risk',
                    title=description,
                    description="Potential bug or code smell detected",
                    suggestion="Fix the issue to prevent potential bugs",
                    confidence=0.85,
                    severity=severity
                ))
        
        return issues
    
    def _analyze_best_practices(self, filename: str, content: str, changes: str, extension: str) -> List[CodeIssue]:
        """Coding best practices analysis"""
        issues = []
        
        best_practice_patterns = [
            (r'//\s*TODO:', 'TODO comment left in code', 'low'),
            (r'//\s*FIXME:', 'FIXME comment left in code', 'medium'),
            (r'function\s+[a-z][a-zA-Z]*', 'Function name should be camelCase', 'low'),
        ]
        
        for pattern, description, severity in best_practice_patterns:
            if re.search(pattern, content):
                issues.append(CodeIssue(
                    file_path=filename,
                    line_number=1,
                    category='best_practices',
                    title=description,
                    description="Code style or best practice issue",
                    suggestion="Follow team coding conventions and best practices",
                    confidence=0.9,
                    severity=severity
                ))
        
        return issues
    
    def _ai_analysis(self, filename: str, content: str, changes: str, pr_context: Dict) -> List[CodeIssue]:
        """Use AI models for advanced code analysis"""
        if not anthropic_client:
            return []
        
        try:
            prompt = self._build_analysis_prompt(filename, content, changes, pr_context)
            
            response = anthropic_client.messages.create(
                model="claude-3-sonnet-20240229",
                max_tokens=1000,
                temperature=0.1,
                system="You are an expert code reviewer. Analyze code for issues and provide specific, actionable feedback.",
                messages=[{"role": "user", "content": prompt}]
            )
            
            return self._parse_ai_response(response.content[0].text, filename)
            
        except Exception as e:
            logging.error(f"AI analysis failed: {e}")
            return []
    
    def _build_analysis_prompt(self, filename: str, content: str, changes: str, pr_context: Dict) -> str:
        """Build prompt for AI analysis"""
        return f"""
        Analyze this code file for the pull request:
        
        PR Title: {pr_context.get('pr_title', 'N/A')}
        PR Description: {pr_context.get('pr_description', 'N/A')}
        Repository: {pr_context.get('repository', 'N/A')}
        
        File: {filename}
        
        Code Content:
        ```
        {content}
        ```
        
        Changes in this PR:
        ```
        {changes}
        ```
        
        Please analyze for:
        1. Security vulnerabilities
        2. Performance issues
        3. Code smells and maintainability
        4. Potential bugs
        5. Best practices violations
        6. Architecture concerns
        
        Provide specific, actionable feedback with line numbers if possible.
        """
    
    def _parse_ai_response(self, response: str, filename: str) -> List[CodeIssue]:
        """Parse AI response into structured issues"""
        # This would parse the AI response into CodeIssue objects
        # Implementation depends on the AI model's response format
        issues = []
        
        # Simple pattern matching for demonstration
        lines = response.split('\n')
        current_issue = None
        
        for line in lines:
            if 'CRITICAL:' in line or 'HIGH:' in line or 'MEDIUM:' in line or 'LOW:' in line:
                if current_issue:
                    issues.append(current_issue)
                
                severity = 'medium'
                if 'CRITICAL' in line: severity = 'critical'
                elif 'HIGH' in line: severity = 'high'
                elif 'LOW' in line: severity = 'low'
                
                current_issue = CodeIssue(
                    file_path=filename,
                    line_number=1,
                    category='ai_analysis',
                    title=line.strip(),
                    description="",
                    suggestion="",
                    confidence=0.8,
                    severity=severity
                )
            elif current_issue and line.strip():
                if not current_issue.description:
                    current_issue.description = line.strip()
                else:
                    current_issue.suggestion = line.strip()
        
        if current_issue:
            issues.append(current_issue)
        
        return issues
    
    def _extract_functions(self, content: str, extension: str) -> Dict[str, Any]:
        """Extract function information from code"""
        functions = {}
        
        if extension in ['.py']:
            # Python function extraction
            try:
                tree = ast.parse(content)
                for node in ast.walk(tree):
                    if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                        start_line = node.lineno
                        end_line = node.end_lineno or start_line
                        functions[node.name] = {
                            'start_line': start_line,
                            'line_count': end_line - start_line + 1
                        }
            except:
                pass
        
        return functions
    
    def _get_example_fix(self, pattern: str, language: str) -> str:
        """Provide example fixes for common issues"""
        fixes = {
            'eval': {
                'javascript': '// Instead of eval(expression)\n// Use: const result = safeEval(expression) or a parser',
                'python': '# Instead of eval(expression)\n# Use: ast.literal_eval(expression) or a safe parser'
            },
            'localStorage password': {
                'javascript': '// Instead of localStorage.setItem("password", pwd)\n// Use: secure authentication with backend'
            }
        }
        
        for key, examples in fixes.items():
            if key in pattern:
                return examples.get(language, '')
        
        return ''
    
    def _generate_summary(self, issues: List[CodeIssue]) -> Dict[str, Any]:
        """Generate summary of all issues found"""
        severity_counts = {'critical': 0, 'high': 0, 'medium': 0, 'low': 0}
        category_counts = {}
        
        for issue in issues:
            severity_counts[issue.severity] += 1
            category_counts[issue.category] = category_counts.get(issue.category, 0) + 1
        
        return {
            'total_issues': len(issues),
            'issues_by_severity': severity_counts,
            'issues_by_category': category_counts,
            'has_critical_issues': severity_counts['critical'] > 0,
            'has_high_issues': severity_counts['high'] > 0,
            'review_passed': severity_counts['critical'] == 0 and severity_counts['high'] == 0
        }
    
    def _format_for_github(self, issues: List[CodeIssue]) -> List[Dict]:
        """Format issues for GitHub API"""
        return [
            {
                'file_path': issue.file_path,
                'line_number': issue.line_number,
                'category': issue.category,
                'title': issue.title,
                'description': issue.description,
                'suggestion': issue.suggestion,
                'confidence': issue.confidence,
                'severity': issue.severity,
                'example': issue.example
            }
            for issue in issues
        ]

# Initialize analyzer
analyzer = CodeAnalyzer()

@app.route('/analyze', methods=['POST'])
def analyze_code():
    """Main analysis endpoint"""
    try:
        data = request.json
        files = data.get('files', [])
        pr_context = {
            'pr_title': data.get('pr_title', ''),
            'pr_description': data.get('pr_description', ''),
            'repository': data.get('repository', ''),
            'base_branch': data.get('base_branch', ''),
            'head_branch': data.get('head_branch', '')
        }
        
        result = analyzer.analyze_code(files, pr_context)
        
        return jsonify(result)
        
    except Exception as e:
        logging.error(f"Analysis error: {e}")
        return jsonify({
            'error': 'Analysis failed',
            'message': str(e)
        }), 500

@app.route('/health', methods=['GET'])
def health_check():
    return jsonify({
        'status': 'healthy',
        'service': 'ai-code-analyzer',
        'ai_available': bool(anthropic_client or openai_client)
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)

3. Package.json and Dependencies

json
{
  "name": "ai-code-review-devops",
  "version": "1.0.0",
  "description": "AI-powered automated code review system",
  "main": "server.js",
  "scripts": {
    "start": "node server.js",
    "dev": "nodemon server.js",
    "test": "jest",
    "lint": "eslint .",
    "build": "echo 'No build step required'"
  },
  "dependencies": {
    "express": "^4.18.2",
    "axios": "^1.6.0",
    "crypto": "^1.0.1",
    "dotenv": "^16.3.1",
    "winston": "^3.11.0",
    "express-rate-limit": "^7.1.5",
    "helmet": "^7.1.0",
    "cors": "^2.8.5"
  },
  "devDependencies": {
    "nodemon": "^3.0.2",
    "jest": "^29.7.0",
    "eslint": "^8.54.0",
    "supertest": "^6.3.3"
  },
  "engines": {
    "node": ">=18.0.0"
  }
}

4. Docker Configuration

dockerfile
# Node.js Dockerfile
FROM node:18-alpine

WORKDIR /app

COPY package*.json ./
RUN npm ci --only=production

COPY . .

USER node

EXPOSE 3000

CMD ["node", "server.js"]
dockerfile
# Python Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

USER nobody

EXPOSE 5000

CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "ai_analyzer:app"]
yaml
# docker-compose.yml
version: '3.8'
services:
  node-orchestrator:
    build: ./node-orchestrator
    ports:
      - "3000:3000"
    environment:
      - GITHUB_WEBHOOK_SECRET=${GITHUB_WEBHOOK_SECRET}
      - GITHUB_TOKEN=${GITHUB_TOKEN}
      - PYTHON_AI_SERVICE=http://python-analyzer:5000
      - NODE_ENV=production
    depends_on:
      - python-analyzer

  python-analyzer:
    build: ./python-analyzer
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - GITHUB_TOKEN=${GITHUB_TOKEN}

  redis:
    image: redis:alpine
    ports:
      - "6379:6379"

  # Monitoring
  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"

Key Features

  1. Multi-Layer Analysis:

    • Pattern-based security scanning

    • Performance optimization detection

    • Maintainability and complexity analysis

    • AI-powered contextual review

  2. Smart Integration:

    • GitHub webhook handling

    • PR status updates

    • Inline comment posting

    • Caching for performance

  3. Production Ready:

    • Error handling and retries

    • Rate limiting

    • Health checks

    • Comprehensive logging

Setup and Deployment

  1. Environment Variables:

bash
# GitHub Configuration
GITHUB_WEBHOOK_SECRET=your_webhook_secret
GITHUB_TOKEN=your_github_token

# AI Services
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key

# Deployment
NODE_ENV=production
PYTHON_AI_SERVICE=http://localhost:5000
  1. Webhook Configuration:

    • Set up GitHub webhook to point to your Node.js service

    • Configure for pull request events

    • Set content type to application/json

  2. Python Requirements:

txt
flask==2.3.3
openai==1.3.7
anthropic==0.7.4
gunicorn==21.2.0
python-dotenv==1.0.0