Building AI Applications

┌──────────┐      ┌──────────────┐      ┌─────────────┐
│  Client  │─────>│ Your Backend │─────>│  AI API     │
│  (User)  │<─────│  (Flask/     │<─────│ (OpenAI/    │
└──────────┘      │   FastAPI)   │      │  Anthropic) │
                  └──────────────┘      └─────────────┘

Flow:
1. User sends request to your API
2. Your backend calls AI API
3. AI API returns response
4. Your backend processes and returns to user

Pros:
✓ Simple to implement
✓ No infrastructure to manage
✓ Always latest model versions
✓ Minimal code

Cons:
✗ Dependent on external API availability
✗ Latency (network + model inference)
✗ Cost per request
✗ Limited customization

┌──────────┐      ┌──────────────┐      ┌─────────┐      ┌─────────┐
│  Client  │─────>│ Your Backend │─────>│  Cache  │─────>│ AI API  │
└──────────┘      └──────────────┘      │ (Redis) │      └─────────┘
                                        └─────────┘

Flow:
1. Check cache for similar prompt (semantic similarity)
2. If cache hit: return cached response (fast, free)
3. If cache miss: call AI API, cache result
4. Return response to user

Example:
- "What is Python?" and "What's Python?" → same cache entry
- Saves ~50-80% of API costs for repeated queries

┌──────────┐      ┌──────────────┐      ┌──────────────────┐
│  Client  │─────>│ Your Backend │─────>│  Your GPU Server │
└──────────┘      └──────────────┘      │  (Llama 3, etc.) │
                                        └──────────────────┘

Pros:
✓ No per-request costs (only infrastructure)
✓ Full data privacy
✓ Low latency (no network to external API)
✓ Can fine-tune models for your domain
✓ No rate limits

Cons:
✗ Complex infrastructure (GPUs, orchestration)
✗ Model quality usually lower than commercial APIs
✗ You manage updates, scaling, monitoring
✗ High upfront and maintenance costs

Strategy:
- Simple tasks → Fast, cheap local model (classification, extraction)
- Complex tasks → Powerful API model (reasoning, generation)

Example routing:
┌────────────────────────┐
│ Is this spam?          │ → Local model (fast, cheap)
├────────────────────────┤
│ Classify sentiment     │ → Local model (good enough)
├────────────────────────┤
│ Write marketing copy   │ → API model (quality matters)
├────────────────────────┤
│ Complex code review    │ → API model (needs reasoning)
└────────────────────────┘

Result: Optimize cost, speed, and quality per use case

Components:
1. FastAPI REST API (Python web framework)
2. CachedAIClient (our AI client with Redis caching)
3. Background task queue (for async processing)
4. Rate limiting (protect from abuse)
5. Authentication (API keys)
6. Monitoring and logging

from pydantic import BaseModel, Field, ValidationError
from typing import List, Optional
import json


class ArticleOutput(BaseModel):
    """Strict schema for AI-generated article."""
    title: str = Field(..., min_length=10, max_length=200)
    author: str = Field(..., min_length=2)
    summary: str = Field(..., min_length=50)
    content: str = Field(..., min_length=500)
    tags: List[str] = Field(..., min_items=3, max_items=10)
    category: str = Field(..., pattern="^(tech|business|science)$")

    # Template requirements
    has_introduction: bool = Field(default=False)
    has_conclusion: bool = Field(default=False)
    word_count: int = Field(..., ge=500)


def generate_article_with_validation(
    ai_client,
    topic: str,
    max_retries: int = 3
) -> ArticleOutput:
    """
    Generate article with automatic validation and retry.

    Returns validated article or raises after max retries.
    """

    system_prompt = """You are a professional article writer.

CRITICAL: You must return ONLY valid JSON with this exact structure:
{
    "title": "string (10-200 chars)",
    "author": "string (2+ chars)",
    "summary": "string (50+ chars)",
    "content": "string (500+ chars)",
    "tags": ["string", "string", "string"],  // 3-10 tags
    "category": "tech|business|science",
    "has_introduction": true,
    "has_conclusion": true,
    "word_count": 500
}

The article content MUST have:
- An introduction section (start with "Introduction:")
- A conclusion section (end with "Conclusion:")
- At least 500 words"""

    user_prompt = f"""Write a complete article about: {topic}

Return only the JSON object, no additional text."""

    for attempt in range(max_retries):
        try:
            # Call AI
            result = ai_client.generate(
                prompt=user_prompt,
                system_prompt=system_prompt,
                temperature=0.7,
                max_tokens=2000
            )

            # Extract JSON (AI might wrap it in markdown or text)
            response_text = result['response']

            # Try to find JSON in response
            start_idx = response_text.find('{')
            end_idx = response_text.rfind('}') + 1

            if start_idx == -1 or end_idx == 0:
                raise ValueError("No JSON found in response")

            json_text = response_text[start_idx:end_idx]
            data = json.loads(json_text)

            # Validate structure with Pydantic
            article = ArticleOutput(**data)

            # Additional content validation
            if "Introduction:" not in article.content:
                raise ValueError("Missing introduction section in content")

            if "Conclusion:" not in article.content:
                raise ValueError("Missing conclusion section in content")

            # Success!
            logger.info(f"Valid article generated on attempt {attempt + 1}")
            return article

        except (json.JSONDecodeError, ValidationError, ValueError) as e:
            logger.warning(
                f"Attempt {attempt + 1}/{max_retries} failed: {type(e).__name__}: {str(e)}"
            )

            if attempt == max_retries - 1:
                # Final attempt failed
                raise ValueError(
                    f"Failed to generate valid article after {max_retries} attempts. "
                    f"Last error: {str(e)}"
                )

            # Add feedback to next prompt for better results
            user_prompt += f"\n\nPREVIOUS ATTEMPT FAILED: {str(e)}. Please fix and try again."

    # Should never reach here due to raise above
    raise RuntimeError("Unexpected end of retry loop")


# Example usage
if __name__ == "__main__":
    client = AIClient(api_key=os.getenv("ANTHROPIC_API_KEY"))

    try:
        article = generate_article_with_validation(
            ai_client=client,
            topic="The Impact of AI on Software Engineering",
            max_retries=3
        )

        print(f"✓ Valid article generated!")
        print(f"  Title: {article.title}")
        print(f"  Category: {article.category}")
        print(f"  Tags: {', '.join(article.tags)}")
        print(f"  Word count: {article.word_count}")
        print(f"  Has intro: {article.has_introduction}")
        print(f"  Has conclusion: {article.has_conclusion}")

    except ValueError as e:
        print(f"✗ Failed to generate valid article: {e}")

import hashlib
from datetime import datetime, timedelta
from typing import Optional


class IdempotentAIService:
    """
    AI service that guarantees idempotency using request IDs and result caching.

    Ensures:
    1. Same request_id always returns same result
    2. Duplicate requests don't call AI API again
    3. Results are stable for configured TTL
    """

    def __init__(self, ai_client, redis_client, result_ttl: int = 86400):
        self.ai_client = ai_client
        self.redis = redis_client
        self.result_ttl = result_ttl  # 24 hours default

    def _generate_request_id(
        self,
        operation: str,
        user_id: str,
        input_data: dict
    ) -> str:
        """
        Generate deterministic request ID from inputs.

        Same inputs → same request_id → same cached result
        """
        # Canonicalize input (sorted JSON for consistency)
        canonical_input = json.dumps(input_data, sort_keys=True)

        # Hash to create stable ID
        hash_input = f"{operation}:{user_id}:{canonical_input}"
        request_id = hashlib.sha256(hash_input.encode()).hexdigest()

        return request_id

    def generate_summary(
        self,
        user_id: str,
        document: str,
        max_length: int = 200,
        request_id: Optional[str] = None
    ) -> dict:
        """
        Generate document summary with idempotency guarantee.

        Args:
            user_id: User making request
            document: Document to summarize
            max_length: Max summary length
            request_id: Optional explicit request ID (for client retry)

        Returns:
            Dict with summary and metadata, guaranteed identical for same request_id
        """

        # Generate or use provided request_id
        if request_id is None:
            request_id = self._generate_request_id(
                operation="summarize",
                user_id=user_id,
                input_data={
                    "document_hash": hashlib.sha256(document.encode()).hexdigest(),
                    "max_length": max_length
                }
            )

        # Check if we already processed this exact request
        cache_key = f"idempotent_result:{request_id}"
        cached_result = self.redis.get(cache_key)

        if cached_result:
            result = json.loads(cached_result)
            result['from_cache'] = True
            result['request_id'] = request_id
            logger.info(f"Idempotent cache hit for request {request_id[:8]}")
            return result

        # Check if request is currently processing (prevent duplicate processing)
        processing_key = f"processing:{request_id}"
        if self.redis.exists(processing_key):
            # Wait for ongoing processing to complete
            for _ in range(30):  # Wait up to 30 seconds
                time.sleep(1)
                cached_result = self.redis.get(cache_key)
                if cached_result:
                    result = json.loads(cached_result)
                    result['from_cache'] = True
                    result['request_id'] = request_id
                    return result

            raise TimeoutError(
                f"Request {request_id[:8]} is still processing after 30s"
            )

        # Mark as processing
        self.redis.setex(processing_key, 60, "processing")  # 60s timeout

        try:
            # Call AI with temperature=0 for more deterministic output
            ai_result = self.ai_client.generate(
                prompt=f"Summarize this document in max {max_length} words:\n\n{document}",
                system_prompt="You are a precise summarizer. Be concise and accurate.",
                temperature=0.0,  # More deterministic
                max_tokens=max_length * 2  # Words to tokens rough estimate
            )

            # Build result
            result = {
                'request_id': request_id,
                'summary': ai_result['response'],
                'user_id': user_id,
                'timestamp': datetime.now().isoformat(),
                'cost': ai_result['cost'],
                'from_cache': False
            }

            # Store result with TTL
            self.redis.setex(
                cache_key,
                self.result_ttl,
                json.dumps(result)
            )

            logger.info(
                f"Processed new request {request_id[:8]}, "
                f"cached for {self.result_ttl}s"
            )

            return result

        finally:
            # Always clear processing lock
            self.redis.delete(processing_key)


# Example usage
if __name__ == "__main__":
    import os
    import redis

    ai_client = AIClient(api_key=os.getenv("ANTHROPIC_API_KEY"))
    redis_client = redis.from_url(os.getenv("REDIS_URL"))

    service = IdempotentAIService(
        ai_client=ai_client,
        redis_client=redis_client,
        result_ttl=86400  # 24 hours
    )

    document = """
    Artificial intelligence is transforming software engineering...
    [long document text]
    """

    # First request
    result1 = service.generate_summary(
        user_id="user123",
        document=document,
        max_length=100
    )
    print(f"Request 1: {result1['request_id'][:8]}, From cache: {result1['from_cache']}")

    # Identical request - returns cached result, no AI call
    result2 = service.generate_summary(
        user_id="user123",
        document=document,
        max_length=100
    )
    print(f"Request 2: {result2['request_id'][:8]}, From cache: {result2['from_cache']}")

    # Verify idempotency
    assert result1['summary'] == result2['summary']
    assert result1['request_id'] == result2['request_id']
    print("✓ Idempotency guaranteed!")

def generate_consistent_classification(
    ai_client,
    text: str,
    categories: List[str]
) -> str:
    """
    Classify text into categories with maximum consistency.

    Same input should produce same category (nearly always).
    """

    system_prompt = f"""You are a classifier. Return ONLY the category name, nothing else.

Available categories: {', '.join(categories)}

Rules:
1. Return exactly one category from the list
2. No explanation, just the category name
3. If unclear, choose the closest match"""

    # Use temperature=0 for maximum determinism
    result = ai_client.generate(
        prompt=f"Classify this text:\n\n{text}",
        system_prompt=system_prompt,
        temperature=0.0,  # Most deterministic
        max_tokens=20
    )

    category = result['response'].strip()

    # Validate result
    if category not in categories:
        # Fallback: find closest match
        category = min(categories, key=lambda c: levenshtein_distance(c, category))

    return category


def generate_with_retry_until_stable(
    ai_client,
    prompt: str,
    system_prompt: str,
    stability_checks: int = 3
) -> str:
    """
    Generate response and retry until output stabilizes.

    Ensures we get consistent output by running multiple times
    and checking for agreement.
    """

    results = []

    for i in range(stability_checks):
        result = ai_client.generate(
            prompt=prompt,
            system_prompt=system_prompt,
            temperature=0.0  # Deterministic
        )
        results.append(result['response'])

    # Check if all results are identical or very similar
    if len(set(results)) == 1:
        # Perfect agreement
        logger.info("Output stable on first attempt")
        return results[0]

    # Calculate similarity between responses
    similarities = []
    for i in range(len(results)):
        for j in range(i + 1, len(results)):
            similarity = calculate_similarity(results[i], results[j])
            similarities.append(similarity)

    avg_similarity = sum(similarities) / len(similarities)

    if avg_similarity >= 0.95:  # 95% similar
        logger.info(f"Output stable with {avg_similarity:.2%} similarity")
        # Return the most common or first result
        return results[0]

    else:
        logger.warning(
            f"Output unstable: {avg_similarity:.2%} similarity. "
            "Consider stronger constraints or lower temperature."
        )
        # Return majority vote or first result
        return max(set(results), key=results.count)


# Example: Consistent product categorization
def categorize_product(ai_client, product_name: str, product_desc: str) -> str:
    """Categorize product with consistency."""

    categories = [
        "Electronics",
        "Clothing",
        "Home & Garden",
        "Sports & Outdoors",
        "Books & Media"
    ]

    text = f"{product_name}\n{product_desc}"

    # Use deterministic classification
    category = generate_consistent_classification(
        ai_client=ai_client,
        text=text,
        categories=categories
    )

    return category

from typing import List, Tuple


class ValidationStage:
    """Base class for validation stages."""

    def validate(self, data: dict) -> Tuple[bool, Optional[str]]:
        """
        Validate data.

        Returns:
            (is_valid, error_message)
        """
        raise NotImplementedError


class SchemaValidation(ValidationStage):
    """Stage 1: Validate schema structure."""

    def __init__(self, model_class):
        self.model_class = model_class

    def validate(self, data: dict) -> Tuple[bool, Optional[str]]:
        try:
            self.model_class(**data)
            return True, None
        except ValidationError as e:
            return False, f"Schema validation failed: {str(e)}"


class ContentValidation(ValidationStage):
    """Stage 2: Validate content quality and completeness."""

    def __init__(self, required_sections: List[str], min_length: int):
        self.required_sections = required_sections
        self.min_length = min_length

    def validate(self, data: dict) -> Tuple[bool, Optional[str]]:
        content = data.get('content', '')

        # Check length
        if len(content) < self.min_length:
            return False, f"Content too short: {len(content)} < {self.min_length}"

        # Check required sections
        for section in self.required_sections:
            if section not in content:
                return False, f"Missing required section: {section}"

        return True, None


class SafetyValidation(ValidationStage):
    """Stage 3: Validate safety and appropriateness."""

    def __init__(self, banned_words: List[str]):
        self.banned_words = [word.lower() for word in banned_words]

    def validate(self, data: dict) -> Tuple[bool, Optional[str]]:
        content = data.get('content', '').lower()

        for word in self.banned_words:
            if word in content:
                return False, f"Content contains banned word: {word}"

        return True, None


class BusinessRuleValidation(ValidationStage):
    """Stage 4: Validate business rules."""

    def validate(self, data: dict) -> Tuple[bool, Optional[str]]:
        # Example: Check if product price is reasonable
        if 'price' in data:
            price = data['price']
            if price < 0:
                return False, "Price cannot be negative"
            if price > 1000000:
                return False, "Price suspiciously high"

        # Example: Check if date is in valid range
        if 'publish_date' in data:
            date = datetime.fromisoformat(data['publish_date'])
            if date > datetime.now():
                return False, "Publish date cannot be in future"

        return True, None


class ValidationPipeline:
    """Multi-stage validation pipeline for AI outputs."""

    def __init__(self, stages: List[ValidationStage]):
        self.stages = stages

    def validate(self, data: dict) -> Tuple[bool, List[str]]:
        """
        Run all validation stages.

        Returns:
            (all_valid, list_of_errors)
        """
        errors = []

        for i, stage in enumerate(self.stages):
            is_valid, error_msg = stage.validate(data)

            if not is_valid:
                errors.append(f"Stage {i+1} ({stage.__class__.__name__}): {error_msg}")

        all_valid = len(errors) == 0
        return all_valid, errors


def generate_with_pipeline_validation(
    ai_client,
    prompt: str,
    system_prompt: str,
    validation_pipeline: ValidationPipeline,
    max_retries: int = 3
) -> dict:
    """
    Generate AI output with multi-stage validation.

    Retries until output passes all validation stages.
    """

    for attempt in range(max_retries):
        # Generate
        result = ai_client.generate(
            prompt=prompt,
            system_prompt=system_prompt,
            temperature=0.3
        )

        # Parse
        try:
            data = extract_and_parse_json(result['response'])
        except Exception as e:
            logger.warning(f"Attempt {attempt + 1}: Failed to parse JSON: {e}")
            continue

        # Validate through pipeline
        is_valid, errors = validation_pipeline.validate(data)

        if is_valid:
            logger.info(f"Output passed all validation stages on attempt {attempt + 1}")
            return data

        else:
            logger.warning(
                f"Attempt {attempt + 1}: Validation failed:\n" +
                "\n".join(errors)
            )

            # Add feedback to prompt for next attempt
            prompt += f"\n\nPREVIOUS ATTEMPT HAD ERRORS:\n" + "\n".join(errors)

    raise ValueError(
        f"Failed to generate valid output after {max_retries} attempts"
    )


# Example usage
if __name__ == "__main__":
    # Create validation pipeline
    pipeline = ValidationPipeline([
        SchemaValidation(ArticleOutput),
        ContentValidation(
            required_sections=["Introduction:", "Conclusion:"],
            min_length=500
        ),
        SafetyValidation(banned_words=["spam", "scam"]),
        BusinessRuleValidation()
    ])

    # Generate with validation
    article_data = generate_with_pipeline_validation(
        ai_client=client,
        prompt="Write article about AI safety",
        system_prompt="You are a professional tech writer.",
        validation_pipeline=pipeline,
        max_retries=3
    )

    print("✓ Article passed all validation stages!")
    print(f"  Title: {article_data['title']}")
    print(f"  Length: {len(article_data['content'])} chars")