Building AI Applications

import json
import os
from typing import List

from pydantic import (BaseModel, Field, ValidationError,
                      computed_field, field_validator)


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
    tags: List[str] = Field(..., min_length=3, max_length=10)
    category: str = Field(..., pattern="^(tech|business|science)$")

    @field_validator("content")
    @classmethod
    def must_be_500_words(cls, v: str) -> str:
        """The requirement is 500 WORDS.

        min_length on a str counts characters, so min_length=500 would accept a
        28-word paragraph. Check the thing you actually care about.
        """
        words = len(v.split())
        if words < 500:
            raise ValueError(f"content has {words} words, need at least 500")
        return v

    @computed_field
    @property
    def word_count(self) -> int:
        """Derived from the text, so it cannot disagree with it.

        Asking the model to report its own word_count invites a number that
        looks plausible and is wrong: in testing it claimed 612 for an article
        of 654 words, and nothing checked.
        """
        return len(self.content.split())


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, the full article, AT LEAST 500 WORDS",
    "tags": ["string", "string", "string"],  // 3-10 tags
    "category": "tech|business|science"
}

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
            # max_tokens covers thinking AND the response on current models.
            # A 500-word article as JSON lands near 2,000 output tokens on its
            # own, so max_tokens=2000 truncates intermittently and the JSON
            # arrives half-written. Leave real headroom.
            result = ai_client.generate(
                prompt=user_prompt,
                system_prompt=system_prompt,
                max_tokens=8000
            )

            # 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("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}  (derived from content)")

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

    # The article that used to slip through: 500+ characters, 28 words.
    try:
        ArticleOutput(
            title="A Perfectly Valid Title", author="Someone", summary="s" * 50,
            content="Introduction: " + ("word " * 26) + "Conclusion:",
            tags=["a", "b", "c"], category="tech",
        )
    except ValidationError as e:
        print(f"\nShort article now rejected: {e.errors()[0]['msg']}")

import hashlib
import json
import time
from datetime import datetime
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 (set(..., ex=) rather than the deprecated setex)
        self.redis.set(processing_key, "processing", ex=60)  # 60s timeout

        try:
            # Note there is no temperature here. Idempotency comes from the
            # cache, not from sampling settings: the model is asked once and
            # every later caller is served that stored answer verbatim.
            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.",
                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.set(
                cache_key,
                json.dumps(result),
                ex=self.result_ttl
            )

            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", "redis://localhost:6379"))

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

    document = (
        "Artificial intelligence is transforming software engineering. Teams now "
        "delegate boilerplate, tests, and refactors to assistants, while keeping "
        "architecture and review in human hands. The bottleneck is shifting from "
        "typing speed to judgment about what should be built at all."
    )

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

    # 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]}, "
          f"From cache: {result2['from_cache']}, Cost: {result2['cost']:.4f}")

    # Verify idempotency
    print(f"\nSame request id: {result1['request_id'] == result2['request_id']}")
    print(f"Same summary:    {result1['summary'] == result2['summary']}")

from enum import Enum

import anthropic
from pydantic import BaseModel

client = anthropic.Anthropic()


class Category(str, Enum):
    """The closed set. Anything outside it is not a possible answer."""
    ELECTRONICS = "Electronics"
    CLOTHING = "Clothing"
    HOME_GARDEN = "Home & Garden"
    SPORTS = "Sports & Outdoors"
    BOOKS_MEDIA = "Books & Media"


class Classification(BaseModel):
    category: Category
    confidence: float


def classify(text: str) -> Classification:
    """Return a category guaranteed to be one of the five above.

    The enum becomes an "enum" constraint in the JSON schema the API enforces,
    so there is no prompt asking nicely for one word, no .strip(), and no
    fuzzy-matching fallback to repair an off-list answer.
    """
    response = client.messages.parse(
        model="claude-haiku-4-5",
        max_tokens=200,
        messages=[{"role": "user", "content": f"Classify this product:\n\n{text}"}],
        output_format=Classification,
    )
    return response.parsed_output


PRODUCT = "Trail Runner GPS Watch\nHeart-rate strap, 40h battery, topo maps."

result = classify(PRODUCT)
print(f"category   : {result.category.value}")
print(f"is a member: {result.category in Category}")
print(f"confidence : {result.confidence}")

# When the stakes justify it, measure agreement instead of assuming it.
from collections import Counter

votes = Counter(classify(PRODUCT).category.value for _ in range(5))
print(f"5 runs   : {dict(votes)}")

top, n = votes.most_common(1)[0]
print(f"agreement: {n}/5 on {top!r}")

import json
import re
from datetime import datetime
from typing import List, Optional, Tuple

from pydantic import ValidationError


def extract_and_parse_json(text: str) -> dict:
    """Pull the JSON object out of a response that may wrap it in prose."""
    start, end = text.find("{"), text.rfind("}") + 1
    if start == -1 or end == 0:
        raise ValueError("No JSON object found in response")
    return json.loads(text[start:end])


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_words: int):
        self.required_sections = required_sections
        self.min_words = min_words

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

        # Count WORDS, not characters. len(content) would pass a 28-word
        # paragraph against a 500 "length" requirement.
        words = len(content.split())
        if words < self.min_words:
            return False, f"Content too short: {words} words < {self.min_words}"

        # 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()

        # Match on word boundaries. A plain substring test rejects "scampi"
        # for containing "scam", and this stage sits in front of an article
        # about AI safety, which has every reason to discuss scams.
        for word in self.banned_words:
            if re.search(rf"\b{re.escape(word)}\b", 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. max_tokens covers thinking plus the response, so the
        # 1000-token default truncates a 500-word article mid-JSON and every
        # attempt fails at the parse step.
        result = ai_client.generate(
            prompt=prompt,
            system_prompt=system_prompt,
            max_tokens=8000
        )

        # 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__":
    client = AIClient(api_key=os.getenv("ANTHROPIC_API_KEY"))

    # Create validation pipeline
    pipeline = ValidationPipeline([
        SchemaValidation(ArticleOutput),
        ContentValidation(
            required_sections=["Introduction:", "Conclusion:"],
            min_words=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. Return ONLY valid JSON with keys "
            "title, author, summary, content, tags, category. content must be at "
            "least 500 words, start with \"Introduction:\" and contain "
            "\"Conclusion:\". category is one of tech|business|science."
        ),
        validation_pipeline=pipeline,
        max_retries=3
    )

    print("Article passed all validation stages")
    print(f"  Title : {article_data['title']}")
    print(f"  Words : {len(article_data['content'].split())}")