Skip to content

basic-usage.md

Latest commit

 

History

History
460 lines (367 loc) · 10.4 KB

basic-usage.md

File metadata and controls

460 lines (367 loc) · 10.4 KB

Basic Usage Examples

Practical examples for using the Agent Advisor Platform.

Quick Start Example

Simple Function Analysis

# example.py
def calculate_sum(a, b):
    return a + b

result = calculate_sum(5, 10)
print(result)

Submit for analysis:

curl -X POST "http://localhost:8000/api/v1/analysis/sync" \
  -H "Content-Type: application/json" \
  -d '{
    "source_code": "def calculate_sum(a, b):\n    return a + b\n\nresult = calculate_sum(5, 10)\nprint(result)",
    "filename": "example.py",
    "analysis_type": "full"
  }'

Expected Results

{
  "suggestions": [
    {
      "type": "quality",
      "severity": "warning",
      "message": "Missing type hints for function 'calculate_sum'",
      "line": 1,
      "suggestion": "Add type hints: def calculate_sum(a: int, b: int) -> int:"
    },
    {
      "type": "documentation",
      "severity": "info",
      "message": "Missing docstring for function 'calculate_sum'",
      "line": 1
    }
  ],
  "metrics": {
    "cyclomatic_complexity": 1,
    "maintainability_index": 85.5,
    "lines_of_code": 4
  }
}

Analysis Types

Quick Analysis

For fast validation during development:

curl -X POST "http://localhost:8000/api/v1/analysis/sync" \
  -H "Content-Type: application/json" \
  -d '{
    "source_code": "def hello():\n    print(\"Hello World\")",
    "filename": "quick.py",
    "analysis_type": "quick"
  }'

Use Case: Pre-commit hooks, IDE integrations

Full Analysis

Comprehensive code review:

curl -X POST "http://localhost:8000/api/v1/analysis/sync" \
  -H "Content-Type: application/json" \
  -d '{
    "source_code": "class UserService:\n    def get_user(self, user_id):\n        return db.query(User).filter(User.id == user_id).first()",
    "filename": "service.py",
    "analysis_type": "full"
  }'

Use Case: Code reviews, CI/CD pipelines

Security Analysis

Focus on security vulnerabilities:

curl -X POST "http://localhost:8000/api/v1/analysis/sync" \
  -H "Content-Type: application/json" \
  -d '{
    "source_code": "import pickle\n\ndef load_data(file_path):\n    with open(file_path, \"rb\") as f:\n        return pickle.load(f)",
    "filename": "security.py",
    "analysis_type": "security"
  }'

Expected Finding: Pickle security vulnerability

Common Code Issues

1. Missing Type Hints

Before:

def process_data(data):
    return [x * 2 for x in data]

Analysis Result:

{
  "type": "quality",
  "severity": "warning",
  "message": "Missing type hints",
  "suggestion": "def process_data(data: List[int]) -> List[int]:"
}

After:

from typing import List

def process_data(data: List[int]) -> List[int]:
    return [x * 2 for x in data]

2. High Complexity

Before:

def calculate_discount(price, customer_type, is_member, has_coupon):
    if customer_type == "premium":
        if is_member:
            if has_coupon:
                return price * 0.7
            else:
                return price * 0.8
        else:
            if has_coupon:
                return price * 0.85
            else:
                return price * 0.9
    else:
        if is_member:
            return price * 0.95
        else:
            return price

Analysis Result:

{
  "type": "quality",
  "severity": "error",
  "message": "Cyclomatic complexity too high (8)",
  "suggestion": "Refactor into smaller functions or use strategy pattern"
}

After:

def calculate_discount(price: float, customer_type: str,
                      is_member: bool, has_coupon: bool) -> float:
    base_discount = get_base_discount(customer_type, is_member)
    coupon_discount = 0.1 if has_coupon else 0
    total_discount = min(base_discount + coupon_discount, 0.3)
    return price * (1 - total_discount)

def get_base_discount(customer_type: str, is_member: bool) -> float:
    discounts = {
        ("premium", True): 0.2,
        ("premium", False): 0.1,
        ("regular", True): 0.05,
        ("regular", False): 0.0
    }
    return discounts.get((customer_type, is_member), 0.0)

3. Security Vulnerabilities

Before:

def execute_query(table, condition):
    query = f"SELECT * FROM {table} WHERE {condition}"
    return db.execute(query)

Analysis Result:

{
  "type": "security",
  "severity": "critical",
  "message": "SQL Injection vulnerability detected",
  "suggestion": "Use parameterized queries or ORM"
}

After:

from sqlalchemy import text

def execute_query(table: str, condition: dict):
    query = text("SELECT * FROM :table WHERE column = :value")
    return db.execute(query, {"table": table, "value": condition["value"]})

4. Poor Performance

Before:

def find_duplicates(items):
    duplicates = []
    for i in range(len(items)):
        for j in range(i + 1, len(items)):
            if items[i] == items[j]:
                duplicates.append(items[i])
    return duplicates

Analysis Result:

{
  "type": "performance",
  "severity": "warning",
  "message": "O(n²) complexity detected",
  "suggestion": "Use set for O(n) complexity"
}

After:

from typing import List, Set

def find_duplicates(items: List[str]) -> List[str]:
    seen: Set[str] = set()
    duplicates: Set[str] = set()

    for item in items:
        if item in seen:
            duplicates.add(item)
        seen.add(item)

    return list(duplicates)

Python Client Usage

Synchronous Analysis

import httpx

def analyze_file(file_path: str) -> dict:
    with open(file_path, 'r') as f:
        source_code = f.read()

    response = httpx.post(
        "http://localhost:8000/api/v1/analysis/sync",
        json={
            "source_code": source_code,
            "filename": file_path,
            "analysis_type": "full"
        }
    )

    return response.json()

# Use it
result = analyze_file("my_script.py")
print(f"Found {len(result['suggestions'])} issues")

Asynchronous Analysis

import asyncio
import httpx

async def analyze_large_file(file_path: str) -> dict:
    async with httpx.AsyncClient() as client:
        # Submit analysis
        with open(file_path, 'r') as f:
            source_code = f.read()

        response = await client.post(
            "http://localhost:8000/api/v1/analysis/async",
            json={
                "source_code": source_code,
                "filename": file_path,
                "analysis_type": "full"
            }
        )

        task_data = response.json()
        task_id = task_data["task_id"]

        # Poll for completion
        while True:
            status_response = await client.get(
                f"http://localhost:8000/api/v1/analysis/task/{task_id}"
            )
            status = status_response.json()

            if status["status"] == "completed":
                return status["result"]
            elif status["status"] == "failed":
                raise Exception(f"Analysis failed: {status['error']}")

            await asyncio.sleep(2)

# Use it
result = asyncio.run(analyze_large_file("large_script.py"))

Batch Analysis

import httpx
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

def analyze_directory(directory: str) -> list:
    python_files = Path(directory).rglob("*.py")

    def analyze_one(file_path):
        with open(file_path, 'r') as f:
            source_code = f.read()

        response = httpx.post(
            "http://localhost:8000/api/v1/analysis/sync",
            json={
                "source_code": source_code,
                "filename": str(file_path),
                "analysis_type": "quick"
            },
            timeout=30.0
        )

        return {
            "file": str(file_path),
            "result": response.json()
        }

    with ThreadPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(analyze_one, python_files))

    return results

# Use it
results = analyze_directory("src/")
print(f"Analyzed {len(results)} files")

CI/CD Integration

GitHub Actions

name: Code Analysis

on: [push, pull_request]

jobs:
  analyze:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Start Agent Advisor
        run: docker-compose up -d

      - name: Wait for API
        run: |
          timeout 60 bash -c 'until curl -f http://localhost:8000/health; do sleep 2; done'

      - name: Analyze Python files
        run: |
          for file in $(find . -name "*.py"); do
            curl -X POST "http://localhost:8000/api/v1/analysis/sync" \
              -H "Content-Type: application/json" \
              -d "{\"source_code\":\"$(cat $file | jq -Rs .)\",\"filename\":\"$file\"}" \
              -o "analysis_$(basename $file).json"
          done

      - name: Upload results
        uses: actions/upload-artifact@v3
        with:
          name: analysis-results
          path: analysis_*.json

Pre-commit Hook

#!/bin/bash
# .git/hooks/pre-commit

STAGED_FILES=$(git diff --cached --name-only --diff-filter=ACM | grep "\.py$")

for FILE in $STAGED_FILES; do
    echo "Analyzing $FILE..."

    RESPONSE=$(curl -s -X POST "http://localhost:8000/api/v1/analysis/sync" \
        -H "Content-Type: application/json" \
        -d "{\"source_code\":\"$(cat $FILE | jq -Rs .)\",\"filename\":\"$FILE\",\"analysis_type\":\"quick\"}")

    CRITICAL_COUNT=$(echo $RESPONSE | jq '[.suggestions[] | select(.severity=="critical")] | length')

    if [ $CRITICAL_COUNT -gt 0 ]; then
        echo "❌ $FILE has $CRITICAL_COUNT critical issues. Commit blocked."
        echo $RESPONSE | jq '.suggestions[] | select(.severity=="critical")'
        exit 1
    fi
done

echo "✅ All files passed analysis"
exit 0

VS Code Integration

Create a custom VS Code task:

{
  "version": "2.0.0",
  "tasks": [
    {
      "label": "Analyze Current File",
      "type": "shell",
      "command": "curl",
      "args": [
        "-X", "POST",
        "http://localhost:8000/api/v1/analysis/sync",
        "-H", "Content-Type: application/json",
        "-d", "{\"source_code\":\"$(cat ${file} | jq -Rs .)\",\"filename\":\"${file}\"}"
      ],
      "problemMatcher": [],
      "presentation": {
        "reveal": "always",
        "panel": "new"
      }
    }
  ]
}

Next Steps