500-ai-ml

AI/ML & LLM Integration Best Practices

Audience: engineers building AI/ML applications, LLM integrations, and AI agents Goal: Reliable, safe, cost-effective AI applications with proper observability and evaluation

AI/ML Philosophy (Core Principles)

Core Principles:

• "Cost-aware by default" - Monitor token usage, choose appropriate models, implement caching
• "Reliability over speed" - Retries, fallbacks, timeouts, graceful degradation
• "Safety first" - Content filtering, prompt injection prevention, guardrails, output validation
• "Observability is essential" - Log prompts, responses, latency, costs, errors
• "Evaluate continuously" - Test outputs, measure quality metrics, A/B test prompts
• "Explicit over implicit" - Clear prompts, explicit instructions, documented assumptions
• "Fail gracefully" - Fallback strategies, error handling, user-friendly messages
• "Version everything" - Version prompts, models, evaluation datasets

Applying AI/ML Principles:

# BAD: No error handling, no cost tracking, no safety checks
def generate_text(prompt: str) -> str:
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# GOOD: Error handling, cost tracking, safety checks
def generate_text(
    prompt: str,
    model: str = "gpt-3.5-turbo",
    max_tokens: int = 1000,
) -> tuple[str, dict]:
    """Generate text with error handling and cost tracking."""
    # Safety check
    if not is_safe_content(prompt):
        raise ValueError("Unsafe content detected")

    try:
        response = openai.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=max_tokens,
        )

        content = response.choices[0].message.content

        # Track costs
        metrics = {
            "input_tokens": response.usage.prompt_tokens,
            "output_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens,
            "model": model,
        }

        # Safety check output
        if not is_safe_content(content):
            raise ValueError("Unsafe output generated")

        return content, metrics
    except openai.RateLimitError:
        # Implement retry with backoff
        time.sleep(5)
        return generate_text(prompt, model, max_tokens)
    except Exception as e:
        logger.error(f"Generation failed: {e}")
        raise