Cómo Reduje 73% los Costes de Inferencia LLM en Producción

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

# Para aprovechar prompt caching, estructura tu prompt así:
# 1. Context ESTÁTICO primero (>1024 tokens) - esto se cachea
# 2. User query DINÁMICO al final

# Example: RAG chatbot con docs cacheadas
SYSTEM_PROMPT = """You are a helpful customer support assistant for Acme SaaS.

KNOWLEDGE BASE (cached context - 5000+ tokens):
==============================================

## Cancellation Policy
To cancel your subscription:
1. Go to Settings > Billing
2. Click "Cancel Subscription"
3. Confirm cancellation

You have 24 hours to cancel after purchase...

## Shipping Policy
We ship within 24-48 hours...

[... 4500+ tokens más de docs estáticos ...]

## Return Policy
You can return products within 30 days...
"""


def query_with_cache(user_question: str) -> str:
    """
    Query OpenAI con prompt caching automático.
    
    El SYSTEM_PROMPT (5000 tokens) se cachea automáticamente porque es >1024 tokens
    y aparece al inicio.
    """
    response = client.chat.completions.create(
        model="gpt-4o",  # o gpt-4o-mini para más savings
        messages=[
            {
                "role": "system",
                "content": SYSTEM_PROMPT  # ← Esto se cachea (5-10min TTL)
            },
            {
                "role": "user",
                "content": user_question  # ← Esto NO se cachea (dinámico)
            }
        ]
    )
    return response.choices[0].message.content


# ============================================================
# PRICING EXAMPLE
# ============================================================

# Sin caching:
# - Input: 5050 tokens (5000 system + 50 user)
# - GPT-4o: $5 / 1M input tokens
# - Coste: (5050 / 1_000_000) * 5 = $0.02525 por query

# Con caching (después del primer request):
# - Cached input: 5000 tokens * 50% discount = $0.0125
# - Non-cached: 50 tokens full price = $0.00025
# - Total: $0.01275 por query
# - AHORRO: 49.5% en input tokens

print("Ejemplo queries:")
print(query_with_cache("¿Cómo cancelo mi suscripción?"))
print(query_with_cache("¿Cuál es su política de envíos?"))  # Cache hit en system prompt

import anthropic

client = anthropic.Anthropic(api_key="your-api-key")

# Anthropic requiere EXPLICIT cache breakpoints con "cache_control"
KNOWLEDGE_BASE = """
[... 8000 tokens de documentación estática ...]
"""

TOOL_DEFINITIONS = [
    {
        "name": "cancel_subscription",
        "description": "Cancel user subscription",
        "input_schema": {
            "type": "object",
            "properties": {
                "user_id": {"type": "string"},
                "reason": {"type": "string"}
            }
        }
    }
    # ... más tools ...
]


def query_with_cache_control(user_question: str) -> str:
    """
    Query Claude con explicit cache breakpoints.
    
    cache_control marca DÓNDE termina contenido cacheable.
    Puedes tener múltiples breakpoints.
    """
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": "You are a helpful customer support assistant.",
            },
            {
                "type": "text",
                "text": KNOWLEDGE_BASE,
                "cache_control": {"type": "ephemeral"}  # ← Cachea con 5min TTL
            }
        ],
        tools=TOOL_DEFINITIONS,  # Opcionalmente cachear tool definitions también
                                # (añadir cache_control al último tool)
        messages=[
            {
                "role": "user",
                "content": user_question
            }
        ]
    )

    # Verificar cache usage en metadata
    usage = response.usage
    print(f"Cache creation tokens: {getattr(usage, 'cache_creation_input_tokens', 0)}")
    print(f"Cache read tokens: {getattr(usage, 'cache_read_input_tokens', 0)}")
    print(f"Regular input tokens: {usage.input_tokens}")

    return response.content[0].text


# ============================================================
# PRICING CALCULATION (Claude 3.5 Sonnet)
# ============================================================

# Pricing:
# - Base input: $3 / MTok
# - Cache write: $3.75 / MTok (1.25x)
# - Cache read (5min TTL): $0.30 / MTok (0.1x = 90% discount)
# - Cache read (1h TTL): $7.50 / MTok (write 2x)

# Example: 8000 tokens knowledge base + 100 user queries

# Request #1 (cache write):
# - 8000 tokens * $3.75 = $0.03 (cache creation)
# - 50 tokens user * $3 = $0.00015
# - Total: $0.03015

# Requests #2-100 (cache reads, dentro 5min TTL):
# - 8000 tokens * $0.30 = $0.0024 (cache read)
# - 50 tokens user * $3 = $0.00015
# - Total: $0.00255 * 99 = $0.25245

# TOTAL 100 queries: $0.03015 + $0.25245 = $0.2826
# Sin cache: 100 * (8050 * $3 / 1M) = $2.415
# AHORRO: 88.3%

print(query_with_cache_control("¿Cómo funciona la garantía?"))

import boto3
import json

# AWS Bedrock client
bedrock = boto3.client(
    service_name='bedrock-runtime',
    region_name='us-east-1'  # o tu región preferida
)


def query_bedrock_with_cache(user_question: str) -> str:
    """
    Query AWS Bedrock Claude con prompt caching.
    
    Bedrock usa automático cache breakpoints para bloques >1024 tokens.
    Más simple que Anthropic native (no necesitas cache_control manual).
    """
    # System prompt estático (se cachea automáticamente)
    system_prompt = """
You are a helpful assistant for Acme Corp.

COMPANY KNOWLEDGE BASE:
======================
[... 6000+ tokens de docs ...]
"""

    # Construct request body
    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1024,
        "system": system_prompt,  # ← Auto-cached si >1024 tokens
        "messages": [
            {
                "role": "user",
                "content": user_question
            }
        ],
        # Bedrock añade caching automático, no config extra necesaria
    })

    # Invoke model
    response = bedrock.invoke_model(
        modelId='anthropic.claude-3-5-sonnet-20241022-v2:0',  # Claude 3.5 Sonnet
        body=body
    )

    # Parse response
    response_body = json.loads(response['body'].read())

    # Check cache usage (en CloudWatch metrics típicamente)
    # Bedrock no devuelve cache stats en response directamente,
    # necesitas monitorear vía CloudWatch

    return response_body['content'][0]['text']


# ============================================================
# PRICING (AWS Bedrock Claude 3.5 Sonnet - us-east-1)
# ============================================================

# Pricing (similar Anthropic pero ligeramente diferente):
# - Input tokens: $3 / MTok
# - Cache write: Similar premium (~1.2-1.3x)
# - Cache read: ~10% base (90% savings)
# - Output tokens: $15 / MTok (igual)

# Example: Same 8000 tokens KB, 100 queries
# Resultado similar a Anthropic: ~88% savings

# VENTAJA AWS: Si ya estás en AWS ecosystem, mejor integración con:
# - IAM roles para security
# - VPC endpoints para private deployment
# - CloudWatch para monitoring unified
# - Cost Explorer para chargeback detallado

print(query_bedrock_with_cache("¿Política de devoluciones?")) 

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
import time

# Initialize Langfuse client
langfuse = Langfuse(
    public_key="your-public-key",
    secret_key="your-secret-key",
    host="http://localhost:3000"  # Self-hosted o cloud
)


@observe()  # Decorator auto-tracks LLM calls
def query_with_monitoring(
    user_query: str,
    user_id: str,
    endpoint: str = "chatbot"
) -> dict:
    """
    Query LLM con full observability tracking.
    
    Langfuse captura automáticamente tokens, latency, costs.
    """
    start_time = time.time()

    # Check semantic cache first
    cached_response = semantic_cache.get(user_query)

    if cached_response:
        # Log cache HIT
        langfuse_context.update_current_trace(
            tags=["cache_hit", endpoint],
            metadata={
                "user_id": user_id,
                "cache_type": "semantic",
                "latency_ms": (time.time() - start_time) * 1000
            }
        )
        return {
            "response": cached_response,
            "source": "cache",
            "tokens": 0,
            "cost": 0
        }

    # Cache MISS → Call LLM
    llm_response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_query}
        ]
    )

    response_text = llm_response.choices[0].message.content
    usage = llm_response.usage

    # Calculate cost
    input_cost = (usage.prompt_tokens / 1_000_000) * 5  # $5/MTok
    output_cost = (usage.completion_tokens / 1_000_000) * 15  # $15/MTok
    total_cost = input_cost + output_cost

    # Cache la respuesta
    semantic_cache.set(user_query, response_text)

    # Log cache MISS + LLM call
    langfuse_context.update_current_trace(
        tags=["cache_miss", "llm_call", endpoint],
        metadata={
            "user_id": user_id,
            "model": "gpt-4o",
            "input_tokens": usage.prompt_tokens,
            "output_tokens": usage.completion_tokens,
            "cached_tokens": getattr(usage, 'prompt_tokens_details', {}).get('cached_tokens', 0),
            "total_cost": total_cost,
            "latency_ms": (time.time() - start_time) * 1000
        }
    )

    return {
        "response": response_text,
        "source": "llm",
        "tokens": usage.total_tokens,
        "cost": total_cost
    }


# ============================================================
# QUERYING LANGFUSE METRICS (API)
# ============================================================

def get_daily_metrics():
    """Pull aggregated metrics from Langfuse."""
    # Langfuse Python SDK tiene queries para analytics
    from datetime import datetime, timedelta

    yesterday = datetime.now() - timedelta(days=1)

    # Get traces for last 24h
    traces = langfuse.get_traces(
        from_timestamp=yesterday,
        tags=["chatbot"]
    )

    total_requests = len(traces)
    cache_hits = len([t for t in traces if "cache_hit" in t.tags])
    cache_miss = total_requests - cache_hits
    cache_hit_rate = (cache_hits / total_requests * 100) if total_requests > 0 else 0

    total_cost = sum([
        t.metadata.get('total_cost', 0)
        for t in traces
        if 'total_cost' in t.metadata
    ])

    avg_latency = sum([
        t.metadata.get('latency_ms', 0)
        for t in traces
    ]) / total_requests if total_requests > 0 else 0

    print(f"📊 Last 24h Metrics:")
    print(f"   Total requests: {total_requests:,}")
    print(f"   Cache hit rate: {cache_hit_rate:.1f}%")
    print(f"   Total cost: ${total_cost:.2f}")
    print(f"   Avg latency: {avg_latency:.0f}ms")

    return {
        "cache_hit_rate": cache_hit_rate,
        "total_cost": total_cost,
        "avg_latency_ms": avg_latency
    }

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# Define metrics
cache_requests_total = Counter(
    'semantic_cache_requests_total',
    'Total cache lookups',
    ['result']  # Labels: hit, miss
)

cache_hit_rate = Gauge(
    'semantic_cache_hit_rate',
    'Current cache hit rate percentage'
)

llm_latency_seconds = Histogram(
    'llm_request_latency_seconds',
    'LLM request latency',
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

llm_cost_dollars = Counter(
    'llm_cost_dollars_total',
    'Total LLM cost in USD',
    ['model', 'endpoint']
)

llm_tokens_total = Counter(
    'llm_tokens_total',
    'Total tokens processed',
    ['type', 'model']  # type: input, output, cached
)


def query_with_prometheus(user_query: str, endpoint: str = "chatbot") -> str:
    """Query LLM con Prometheus metrics export."""
    start = time.time()

    # Check cache
    cached = semantic_cache.get(user_query)

    if cached:
        # Metrics: cache hit
        cache_requests_total.labels(result='hit').inc()
        elapsed = time.time() - start
        llm_latency_seconds.observe(elapsed)

        # Update hit rate gauge (rolling calculation)
        update_cache_hit_rate()

        return cached

    # Cache miss → LLM call
    cache_requests_total.labels(result='miss').inc()

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_query}
        ]
    )

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

    # Metrics: tokens
    llm_tokens_total.labels(type='input', model='gpt-4o').inc(usage.prompt_tokens)
    llm_tokens_total.labels(type='output', model='gpt-4o').inc(usage.completion_tokens)

    # Metrics: cost
    cost = (usage.prompt_tokens / 1_000_000) * 5 + (usage.completion_tokens / 1_000_000) * 15
    llm_cost_dollars.labels(model='gpt-4o', endpoint=endpoint).inc(cost)

    # Metrics: latency
    elapsed = time.time() - start
    llm_latency_seconds.observe(elapsed)

    # Update hit rate
    update_cache_hit_rate()

    # Cache response
    semantic_cache.set(user_query, response_text)

    return response_text


def update_cache_hit_rate():
    """Calculate rolling cache hit rate (last 1000 requests)."""
    # Query Prometheus itself para calcular hit rate
    # Alternativamente, mantener counter local
    hits = cache_requests_total.labels(result='hit')._value.get()
    misses = cache_requests_total.labels(result='miss')._value.get()
    total = hits + misses

    if total > 0:
        hit_rate_pct = (hits / total) * 100
        cache_hit_rate.set(hit_rate_pct)


# Start Prometheus HTTP server (expone /metrics endpoint)
if __name__ == "__main__":
    start_http_server(8000)  # http://localhost:8000/metrics
    print("📊 Prometheus metrics available at http://localhost:8000/metrics")

    # Keep alive
    while True:
        time.sleep(1) 

import redis
import numpy as np
from sentence_transformers import SentenceTransformer
from typing import Optional, Dict, Any
import hashlib
import json
import logging

logger = logging.getLogger(__name__)


class SemanticCache:
    """
    Production-ready semantic caching con Redis + sentence-transformers.
    
    Features:
    - Vector similarity search (cosine similarity)
    - Configurable similarity threshold
    - TTL support para cache expiration
    - HNSW indexing para performance
    """

    def __init__(
        self,
        redis_host: str = "localhost",
        redis_port: int = 6379,
        redis_db: int = 0,
        embedding_model: str = "sentence-transformers/all-mpnet-base-v2",
        similarity_threshold: float = 0.79,
        default_ttl: int = 3600  # 1 hora default
    ):
        """
        Args:
            redis_host: Redis server host
            redis_port: Redis server port
            redis_db: Redis database number
            embedding_model: HuggingFace model name para embeddings
            similarity_threshold: Minimum cosine similarity para cache hit (0.0-1.0)
            default_ttl: Default TTL en segundos (3600 = 1h, 300 = 5min)
        """
        # Redis connection
        self.redis_client = redis.Redis(
            host=redis_host,
            port=redis_port,
            db=redis_db,
            decode_responses=False  # Binary mode para np arrays
        )

        # Sentence transformer model (self-hosted)
        logger.info(f"Loading embedding model: {embedding_model}")
        self.model = SentenceTransformer(embedding_model)
        self.embedding_dim = self.model.get_sentence_embedding_dimension()

        # Cache config
        self.similarity_threshold = similarity_threshold
        self.default_ttl = default_ttl

        # Redis index name
        self.index_name = "semantic_cache_idx"

        # Create HNSW index si no existe
        self._create_index_if_not_exists()

    def _create_index_if_not_exists(self):
        """Crea índice HNSW en Redis para vector search."""
        try:
            self.redis_client.ft(self.index_name).info()
            logger.info(f"Index {self.index_name} already exists")
        except:
            # Index no existe, crearlo
            from redis.commands.search.field import VectorField, TextField
            from redis.commands.search.indexDefinition import IndexDefinition, IndexType

            schema = (
                VectorField(
                    "embedding",
                    "HNSW",
                    {
                        "TYPE": "FLOAT32",
                        "DIM": self.embedding_dim,
                        "DISTANCE_METRIC": "COSINE"
                    }
                ),
                TextField("query"),
                TextField("response")
            )

            definition = IndexDefinition(
                prefix=["cache:"],
                index_type=IndexType.HASH
            )

            self.redis_client.ft(self.index_name).create_index(
                fields=schema,
                definition=definition
            )
            logger.info(f"Created HNSW index: {self.index_name}")

    def _generate_embedding(self, text: str) -> np.ndarray:
        """Genera embedding vector para texto."""
        return self.model.encode(text, convert_to_numpy=True)

    def _cosine_similarity(self, vec1: np.ndarray, vec2: np.ndarray) -> float:
        """Calcula cosine similarity entre dos vectores."""
        return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

    def get(self, query: str) -> Optional[str]:
        """
        Busca respuesta cacheada para query semánticamente similar.
        
        Args:
            query: User query text
            
        Returns:
            Cached response si similarity >= threshold, None si cache miss
        """
        # Generate query embedding
        query_embedding = self._generate_embedding(query)

        # Vector similarity search en Redis
        from redis.commands.search.query import Query

        # KNN search: top 1 resultado más similar
        q = Query(
            f"*=>[KNN 1 @embedding $vec AS score]"
        ).return_fields("query", "response", "score").dialect(2)

        try:
            results = self.redis_client.ft(self.index_name).search(
                q,
                query_params={"vec": query_embedding.astype(np.float32).tobytes()}
            )

            if results.total > 0:
                doc = results.docs[0]
                similarity = 1 - float(doc.score)  # COSINE distance → similarity

                if similarity >= self.similarity_threshold:
                    logger.info(f"✅ Cache HIT (similarity: {similarity:.3f})")
                    logger.debug(f"Original query: {doc.query}")
                    logger.debug(f"Current query: {query}")
                    return doc.response
                else:
                    logger.info(
                        f"❌ Cache MISS (similarity: {similarity:.3f} < {self.similarity_threshold})"
                    )
            else:
                logger.info("❌ Cache MISS (no results)")

        except Exception as e:
            logger.error(f"Redis search error: {e}")

        return None

    def set(
        self,
        query: str,
        response: str,
        ttl: Optional[int] = None
    ):
        """
        Guarda query + response en cache con embedding.
        
        Args:
            query: User query text
            response: LLM response to cache
            ttl: TTL en segundos (usa default_ttl si None)
        """
        ttl = ttl or self.default_ttl

        # Generate embedding
        embedding = self._generate_embedding(query)

        # Hash query como key
        query_hash = hashlib.sha256(query.encode()).hexdigest()[:16]
        key = f"cache:{query_hash}"

        # Store en Redis hash
        self.redis_client.hset(
            key,
            mapping={
                "query": query,
                "response": response,
                "embedding": embedding.astype(np.float32).tobytes()
            }
        )

        # Set TTL
        self.redis_client.expire(key, ttl)

        logger.info(f"💾 Cached query (TTL: {ttl}s)")
        logger.debug(f"Query: {query[:100]}...")

    def invalidate_all(self):
        """Borra TODO el cache (usar con cuidado)."""
        keys = self.redis_client.keys("cache:*")
        if keys:
            self.redis_client.delete(*keys)
            logger.warning(f"🗑️ Invalidated {len(keys)} cache entries")


# ============================================================
# USAGE EXAMPLE
# ============================================================

if __name__ == "__main__":
    # Setup logging
    logging.basicConfig(level=logging.INFO)

    # Initialize cache
    cache = SemanticCache(
        redis_host="localhost",
        redis_port=6379,
        similarity_threshold=0.79,  # 79% minimum similarity
        default_ttl=3600  # 1 hora
    )

    # Simulate LLM queries
    queries = [
        "¿Cómo cancelo mi pedido?",
        "Quiero cancelar mi orden",
        "Proceso para anular compra",
        "¿Dónde está mi paquete?"  # Diferente semánticamente
    ]

    # First query → cache MISS → call LLM
    query1 = queries[0]
    cached_response = cache.get(query1)

    if cached_response is None:
        # Simular LLM call
        llm_response = "Para cancelar tu pedido, ve a Mis Pedidos > Cancelar. Tienes 24h."
        cache.set(query1, llm_response)
        print(f"❌ MISS: {query1}")
        print(f"💬 LLM Response: {llm_response}\n")

    # Similar queries → cache HIT
    for query in queries[1:]:
        cached = cache.get(query)
        if cached:
            print(f"✅ HIT: {query}")
            print(f"💾 Cached: {cached}\n")
        else:
            print(f"❌ MISS: {query}\n")