Build AI-Powered CI/CD Pipelines in 45 Minutes

# .github/workflows/ai-review.yml
name: AI Code Review

on:
  pull_request:
    types: [opened, synchronize]

env:
  ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}

jobs:
  ai-review:
    runs-on: ubuntu-latest
    permissions:
      pull-requests: write
      contents: read
    
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0  # Full history for context
      
      - name: Get changed files
        id: changes
        run: |
          # Only review changed files, not entire codebase
          git diff --name-only origin/${{ github.base_ref }}...HEAD > changed_files.txt
          echo "files=$(cat changed_files.txt | tr '\n' ',' | sed 's/,$//')" >> $GITHUB_OUTPUT

# scripts/ai_review.py
import anthropic
import os
import sys
from pathlib import Path

def analyze_changes(files: list[str], diff_content: str) -> dict:
    """Send code changes to Claude for context-aware review."""
    
    client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
    
    # Build context from repository
    context_files = []
    for f in files:
        if Path(f).suffix in ['.py', '.ts', '.go', '.rs']:
            try:
                content = Path(f).read_text()
                context_files.append(f"File: {f}\n{content}")
            except FileNotFoundError:
                continue  # File was deleted in PR
    
    prompt = f"""You are reviewing a pull request for production deployment.

CHANGED FILES:
{diff_content}

FULL FILE CONTEXT:
{chr(10).join(context_files[:5])}  # Limit to 5 files for token efficiency

Analyze for:
1. **Security**: SQL injection, XSS, exposed secrets, unsafe deserialization
2. **Performance**: N+1 queries, blocking I/O in hot paths, memory leaks
3. **Reliability**: Missing error handling, race conditions, improper retries
4. **Breaking changes**: API contract violations, database migration issues

Respond in JSON:
{{
  "severity": "block|warn|pass",
  "issues": [
    {{"type": "security", "line": 42, "description": "...", "suggestion": "..."}},
    ...
  ],
  "risk_score": 0-100,
  "safe_to_deploy": true|false
}}

Only flag real issues. Ignore style preferences."""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",  # Fast, accurate for code
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}]
    )
    
    # Extract JSON from response (Claude wraps in markdown sometimes)
    import json
    import re
    
    text = response.content[0].text
    json_match = re.search(r'\{.*\}', text, re.DOTALL)
    if json_match:
        return json.loads(json_match.group())
    else:
        return {"severity": "warn", "issues": [], "risk_score": 50, "safe_to_deploy": True}

if __name__ == "__main__":
    files = sys.argv[1].split(',')
    diff = sys.stdin.read()
    
    result = analyze_changes(files, diff)
    
    # Exit code determines pipeline behavior
    if result["severity"] == "block":
        print(f"🚫 BLOCKING ISSUES FOUND (risk: {result['risk_score']}/100)")
        for issue in result["issues"]:
            print(f"  {issue['type'].upper()}: Line {issue['line']} - {issue['description']}")
        sys.exit(1)
    elif result["severity"] == "warn":
        print(f"âš  WARNINGS (risk: {result['risk_score']}/100)")
        for issue in result["issues"]:
            print(f"  {issue['description']}")
        sys.exit(0)
    else:
        print(f"✅ Clean review (risk: {result['risk_score']}/100)")
        sys.exit(0)

# Continue in .github/workflows/ai-review.yml

      - name: Run AI Review
        id: review
        run: |
          pip install anthropic
          git diff origin/${{ github.base_ref }}...HEAD | \
            python scripts/ai_review.py "${{ steps.changes.outputs.files }}"
        continue-on-error: true  # Don't fail pipeline, just post comment
      
      - name: Post Review Comment
        if: always()
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            // Read review output from previous step
            const review = `${{ steps.review.outcome }}`;
            
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: `## 🤖 AI Code Review\n\n${review}\n\n*Powered by Claude Sonnet 4*`
            });

# scripts/smart_test_selector.py
import anthropic
import os
from pathlib import Path

def select_tests(changed_files: list[str]) -> list[str]:
    """Ask LLM which tests are impacted by changes."""
    
    client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
    
    # Get all test files
    test_files = list(Path("tests").rglob("test_*.py"))
    test_list = "\n".join([str(f) for f in test_files])
    
    prompt = f"""Given these changed files in a Python web application:
{chr(10).join(changed_files)}

And these available test files:
{test_list}

Which tests MUST run to verify these changes? Consider:
- Direct imports and dependencies
- Shared utilities or models
- API contracts if endpoints changed
- Database migrations if schema changed

Respond with ONLY a JSON array of test file paths, no explanation:
["tests/test_api.py", "tests/integration/test_auth.py"]

If changes are only documentation/comments, return empty array []."""

    response = client.messages.create(
        model="claude-haiku-4-20251001",  # Haiku is faster for simple tasks
        max_tokens=500,
        messages=[{"role": "user", "content": prompt}]
    )
    
    import json
    import re
    
    text = response.content[0].text
    json_match = re.search(r'\[.*\]', text, re.DOTALL)
    if json_match:
        return json.loads(json_match.group())
    else:
        # Fallback: run all tests if parsing fails
        return [str(f) for f in test_files]

if __name__ == "__main__":
    import sys
    files = sys.argv[1].split(',')
    tests = select_tests(files)
    
    if not tests:
        print("No tests needed for these changes")
    else:
        print(" ".join(tests))  # Space-separated for pytest

      - name: Select Tests
        id: tests
        run: |
          TESTS=$(python scripts/smart_test_selector.py "${{ steps.changes.outputs.files }}")
          echo "to_run=$TESTS" >> $GITHUB_OUTPUT
      
      - name: Run Selected Tests
        if: steps.tests.outputs.to_run != ''
        run: |
          pytest ${{ steps.tests.outputs.to_run }} -v

# scripts/deployment_decision.py
import anthropic
import os
import sys
from datetime import datetime

def assess_deployment_risk(
    changed_files: list[str],
    diff: str,
    current_hour: int,
    recent_deploys: int
) -> dict:
    """Decide when to deploy based on change risk and timing."""
    
    client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
    
    prompt = f"""You are a deployment safety system for a production web app.

CURRENT TIME: {current_hour}:00 UTC (business hours: 9-17)
RECENT DEPLOYS: {recent_deploys} in last 24 hours
CHANGES:
{diff}

Assess deployment risk considering:
1. **Change scope**: Database migrations = high risk, config = low risk
2. **Timing**: Deploy breaking changes outside business hours
3. **Velocity**: >5 deploys/day = slow down, let previous deploy stabilize
4. **Rollback complexity**: Can this auto-rollback or needs manual intervention?

Respond in JSON:
{{
  "risk_level": "low|medium|high",
  "deploy_now": true|false,
  "wait_until_hour": 20,  # UTC hour to deploy, null if deploy now
  "reason": "Why this decision",
  "rollback_plan": "Automatic" | "Manual required"
}}"""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1000,
        messages=[{"role": "user", "content": prompt}]
    )
    
    import json
    import re
    text = response.content[0].text
    json_match = re.search(r'\{.*\}', text, re.DOTALL)
    return json.loads(json_match.group()) if json_match else {
        "risk_level": "medium",
        "deploy_now": True,
        "wait_until_hour": None,
        "reason": "Fallback: manual approval needed",
        "rollback_plan": "Manual required"
    }

if __name__ == "__main__":
    files = sys.argv[1].split(',')
    diff = sys.stdin.read()
    hour = datetime.utcnow().hour
    recent = int(os.environ.get("RECENT_DEPLOYS", "0"))
    
    decision = assess_deployment_risk(files, diff, hour, recent)
    
    print(f"Risk: {decision['risk_level']}")
    print(f"Deploy now: {decision['deploy_now']}")
    print(f"Reason: {decision['reason']}")
    
    # Set GitHub Actions output
    print(f"::set-output name=deploy_now::{decision['deploy_now']}")
    print(f"::set-output name=wait_until::{decision['wait_until_hour']}")

      - name: Deployment Decision
        id: deploy_decision
        run: |
          git diff origin/main...HEAD | \
            python scripts/deployment_decision.py "${{ steps.changes.outputs.files }}"
        env:
          RECENT_DEPLOYS: ${{ secrets.DEPLOY_COUNT_24H }}
      
      - name: Deploy to Production
        if: steps.deploy_decision.outputs.deploy_now == 'true'
        run: |
          kubectl apply -f k8s/
          echo "Deployed at $(date)"
      
      - name: Schedule Delayed Deploy
        if: steps.deploy_decision.outputs.deploy_now == 'false'
        run: |
          # Create scheduled GitHub Action or use cron
          echo "Deployment delayed until ${{ steps.deploy_decision.outputs.wait_until }}:00 UTC"

# Create test PR
git checkout -b test-ai-cicd
echo "# Test change" >> README.md
git add README.md
git commit -m "test: AI pipeline verification"
git push origin test-ai-cicd

# Open PR on GitHub and check:
# 1. AI review comment appears within 2 minutes
# 2. Only relevant tests run (check Actions log)
# 3. Deployment decision is logged

# Add to all AI scripts
import time
from functools import wraps

def rate_limit(calls_per_minute=10):
    """Prevent API rate limit errors."""
    min_interval = 60.0 / calls_per_minute
    last_call = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_call[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            last_call[0] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(calls_per_minute=10)
def analyze_changes(files, diff):
    # Your existing code
    pass

# Add retry logic with exponential backoff
import anthropic
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_llm_with_retry(prompt):
    """Retry API calls with exponential backoff."""
    try:
        client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
        return client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2000,
            messages=[{"role": "user", "content": prompt}]
        )
    except anthropic.RateLimitError:
        print("Rate limited, retrying...")
        raise  # tenacity will handle retry
    except Exception as e:
        print(f"API error: {e}, falling back to traditional CI")
        return None  # Fallback to non-AI review

# Add to workflow for observability
      - name: Log AI Metrics
        if: always()
        run: |
          # Send metrics to monitoring system
          curl -X POST https://your-metrics-endpoint.com/ai-cicd \
            -d "{
              \"pr_number\": \"${{ github.event.pull_request.number }}\",
              \"review_time\": \"${{ steps.review.duration }}\",
              \"risk_score\": \"${{ steps.review.outputs.risk_score }}\",
              \"tests_selected\": \"${{ steps.tests.outputs.count }}\",
              \"deployment_approved\": \"${{ steps.deploy_decision.outputs.deploy_now }}\"
            }"

# Sanitize before sending to LLM
import re

def sanitize_code(content: str) -> str:
    """Remove sensitive patterns before LLM analysis."""
    # Remove API keys, tokens
    content = re.sub(r'(api[_-]?key|token|password)\s*=\s*["\'].*?["\']', 
                     r'\1="REDACTED"', content, flags=re.IGNORECASE)
    # Remove connection strings
    content = re.sub(r'(postgres|mysql|mongodb)://.*?@', 
                     r'\1://REDACTED@', content)
    return content