10 Técnicas de Optimización LLM para Reducir Costes 73% en Producción

# Instalación: pip install anthropic from anthropic import Anthropic client = Anthropic(api_key="tu-api-key-aqui") # Prompt del sistema largo (ej: documentación, instrucciones) system_prompt = """ Eres un asistente experto en análisis de contratos legales. Tienes acceso a la siguiente documentación de referencia: [Aquí iría documentación de 10,000+ tokens sobre leyes, procedimientos, casos de ejemplo, etc.] Usa esta información para responder preguntas precisas sobre contratos. """ # Habilitar caching con cache_control response = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, system=[ { "type": "text", "text": system_prompt, "cache_control": {"type": "ephemeral"} # 🔑 Clave para caching } ], messages=[ { "role": "user", "content": "¿Cuáles son las cláusulas críticas en un NDA?" } ] ) # Verificar métricas de caché usage = response.usage print(f"Tokens de entrada: {usage.input_tokens}") print(f"Tokens leídos de caché: {usage.cache_read_input_tokens}") print(f"Tokens escritos en caché: {usage.cache_creation_input_tokens}") print(f"Tokens de salida: {usage.output_tokens}") # Calcular ahorros if usage.cache_read_input_tokens > 0: ahorro_porcentaje = (usage.cache_read_input_tokens / (usage.input_tokens + usage.cache_read_input_tokens)) * 100 print(f"\ 💰 Ahorro por caché: {ahorro_porcentaje:.1f}%") 

# Instalación: pip install openai from openai import OpenAI client = OpenAI(api_key="tu-api-key-aqui") # OpenAI cachea automáticamente prompts repetidos en GPT-4o/4o-mini # No requiere flags especiales, pero el ahorro se refleja en facturación response = client.chat.completions.create( model="gpt-4o", messages=[ { "role": "system", "content": "Largo prompt del sistema que se repite..." }, { "role": "user", "content": "Pregunta específica del usuario" } ] ) # OpenAI aplica descuento automático en tokens cacheados # Visible en tu dashboard de facturación como "Cached input tokens" print(f"Respuesta: {response.choices[0].message.content}") print(f"Uso total tokens: {response.usage.total_tokens}") 

# Instalación: pip install gptcache openai from gptcache import cache, Config from gptcache.adapter import openai from gptcache.manager import manager_factory from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation import os # Configurar caché semántico con similitud vectorial cache.init( config=Config( similarity_threshold=0.95 # 95% similitud para hit ), embedding_func=lambda x: openai.Embedding.create( input=x, model="text-embedding-3-small" )["data"][0]["embedding"], data_manager=manager_factory( "redis,faiss", # Redis para almacenamiento + Faiss para búsqueda vectorial scalar_params={"url": "redis://localhost:6379"}, vector_params={"dimension": 1536} # text-embedding-3-small dimension ), similarity_evaluation=SearchDistanceEvaluation() ) # Configurar clave API cache.set_openai_key(os.getenv("OPENAI_API_KEY")) # Uso normal de OpenAI - GPTCache intercepta automáticamente response = openai.ChatCompletion.create( model='gpt-3.5-turbo', messages=[ {'role': 'user', 'content': '¿Cómo puedo cancelar mi suscripción?'} ], ) print(f"Respuesta: {response['choices'][0]['message']['content']}") # Consulta semánticamente similar (debería dar cache hit) response2 = openai.ChatCompletion.create( model='gpt-3.5-turbo', messages=[ {'role': 'user', 'content': '¿Puedo cancelar mi plan de pago?'} ], ) print(f"Respuesta 2: {response2['choices'][0]['message']['content']}") print(f"Cache hit: {response2.get('gptcache', {}).get('hit', False)}") 

# Instalación: pip install redis openai numpy import redis from openai import OpenAI import numpy as np import hashlib import json # Conectar a Redis redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True) openai_client = OpenAI() def get_embedding(text: str) -> list: """Genera embedding vectorial de un texto""" response = openai_client.embeddings.create( model="text-embedding-3-small", input=text ) return response.data[0].embedding def cosine_similarity(vec1: list, vec2: list) -> float: """Calcula similitud coseno entre dos vectores""" return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2)) def semantic_cache_lookup(query: str, threshold: float = 0.95) -> dict: """ Busca en caché semántica si existe respuesta similar Returns: dict con 'hit' (bool), 'response' (str), 'similarity' (float) """ query_embedding = get_embedding(query) # Iterar sobre todas las consultas cacheadas for key in redis_client.scan_iter("cache:*"): cached_data = json.loads(redis_client.get(key)) cached_embedding = cached_data["embedding"] similarity = cosine_similarity(query_embedding, cached_embedding) if similarity > threshold: print(f"✅ Cache HIT! Similitud: {similarity:.3f}") return { "hit": True, "response": cached_data["response"], "similarity": similarity } print(f"❌ Cache MISS. Llamando a LLM...") return {"hit": False, "response": None, "similarity": 0} def semantic_cache_store(query: str, response: str, ttl: int = 3600): """Almacena consulta y respuesta en caché semántica""" query_embedding = get_embedding(query) cache_key = f"cache:{hashlib.md5(query.encode()).hexdigest()}" redis_client.setex( cache_key, ttl, # TTL de 1 hora json.dumps({ "query": query, "embedding": query_embedding, "response": response }) ) # Ejemplo de uso def chatbot_with_semantic_cache(user_query: str) -> str: # Intentar obtener de caché cache_result = semantic_cache_lookup(user_query) if cache_result["hit"]: return cache_result["response"] # Si no hay hit, llamar a LLM response = openai_client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": user_query}] ) answer = response.choices[0].message.content # Guardar en caché semantic_cache_store(user_query, answer) return answer # Pruebas print(chatbot_with_semantic_cache("¿Cómo cancelo mi suscripción?")) print("\ ---\ ") print(chatbot_with_semantic_cache("¿Puedo cancelar mi plan?")) # Debería dar HIT 

# Instalación: pip install llmlingua transformers torch from llmlingua import PromptCompressor from openai import OpenAI # Inicializar compresor llm_lingua = PromptCompressor( model_name="microsoft/llmlingua-2-bert-base-multilingual-cased-meetingbank", use_llmlingua2=True # Usar versión más reciente ) # Prompt largo original (ej: instrucciones de servicio al cliente) original_prompt = """ Eres un asistente de servicio al cliente para una empresa de e-commerce de tecnología. Debes seguir estas reglas: 1. Siempre ser amable y profesional, incluso si el cliente está frustrado 2. Si el cliente pregunta por devoluciones, explicar que tienen 30 días desde la compra 3. Para problemas técnicos, ofrecer primero soluciones de troubleshooting básico 4. Si no puedes resolver el problema, escalar a un agente humano 5. Nunca hacer promesas sobre plazos de entrega sin verificar el sistema 6. Para consultas de facturación, siempre verificar la identidad del cliente primero 7. Mantener un tono conversacional pero profesional 8. No usar jerga técnica a menos que el cliente la use primero Ejemplos de interacciones correctas: - Cliente: "Mi producto llegó dañado" Agente: "Lamento mucho escuchar eso. Vamos a solucionarlo de inmediato..." - Cliente: "¿Cuándo llegará mi pedido?" Agente: "Déjame verificar el estado de tu pedido..." Ahora responde a la siguiente consulta del cliente manteniendo estas directrices. """ print(f"Tokens originales: {len(original_prompt.split())}") # Comprimir prompt compressed_result = llm_lingua.compress_prompt( original_prompt, instruction="", # Opcional: instrucción adicional sobre qué preservar question="", # La pregunta real del cliente se añade después target_token=200, # Objetivo: reducir a 200 tokens (80% compresión) condition_compare=True, condition_in_question='after', rank_method='longllmlingua', # Método optimizado para contextos largos use_sentence_level_filter=False, context_budget="+100", dynamic_context_compression_ratio=0.4, reorder_context="sort" # Reordenar por importancia ) compressed_prompt = compressed_result['compressed_prompt'] print(f"Tokens comprimidos: {len(compressed_prompt.split())}") print(f"Ratio de compresión: {compressed_result['ratio']:.2f}x") print(f"\ Prompt comprimido:\ {compressed_prompt}") # Usar con OpenAI openai_client = OpenAI() response = openai_client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": compressed_prompt}, {"role": "user", "content": "Mi producto llegó dañado, ¿qué hago?"} ] ) print(f"\ Respuesta: {response.choices[0].message.content}") print(f"Tokens usados: {response.usage.total_tokens}") 

# Instalación: pip install llmlingua from llmlingua import PromptCompressor # Para RAG: comprimir documentos recuperados antes de enviar al LLM llm_lingua = PromptCompressor(model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank") # Supongamos que recuperaste 5 documentos de 2,000 tokens cada uno = 10,000 tokens retrieved_documents = [ "Documento 1: Políticas de devolución detalladas...", "Documento 2: Procedimientos de garantía...", "Documento 3: Información de envío...", "Documento 4: FAQs técnicas...", "Documento 5: Términos y condiciones..." ] # Unir documentos context = "\ \ ".join(retrieved_documents) print(f"Contexto RAG original: {len(context.split())} tokens") # Comprimir manteniendo información relevante para la pregunta user_question = "¿Puedo devolver un producto después de 45 días?" compressed_context = llm_lingua.compress_prompt( context, instruction="Extract information about return policies", question=user_question, target_token=800, # De 10,000 a 800 tokens (92% compresión) condition_compare=True, condition_in_question='after', rank_method='longllmlingua' ) print(f"Contexto comprimido: {len(compressed_context['compressed_prompt'].split())} tokens") print(f"Ahorro: {(1 - compressed_context['ratio']) * 100:.1f}%") # Enviar a LLM final_prompt = f""" Context: {compressed_context['compressed_prompt']} Question: {user_question} Answer based on the context above: """ # Usar con tu LLM preferido... 

# Instalación: pip install litellm from litellm import Router import os # Configurar lista de modelos con prioridades y costes model_list = [ { "model_name": "gpt-4o", "litellm_params": { "model": "gpt-4o", "api_key": os.getenv("OPENAI_API_KEY") }, "model_info": {"cost": "high", "quality": "premium"} }, { "model_name": "gpt-3.5-turbo", "litellm_params": { "model": "gpt-3.5-turbo", "api_key": os.getenv("OPENAI_API_KEY") }, "model_info": {"cost": "medium", "quality": "good"} }, { "model_name": "deepseek-chat", "litellm_params": { "model": "deepseek/deepseek-chat", "api_key": os.getenv("DEEPSEEK_API_KEY") }, "model_info": {"cost": "low", "quality": "good"} }, { "model_name": "gemini-flash", "litellm_params": { "model": "gemini/gemini-1.5-flash", "api_key": os.getenv("GEMINI_API_KEY") }, "model_info": {"cost": "very-low", "quality": "medium"} } ] # Inicializar router router = Router( model_list=model_list, routing_strategy="cost-based-routing", # Prefiere modelos más baratos num_retries=2 ) def classify_query_complexity(query: str, conversation_history: list = None) -> str: """ Clasifica complejidad de una consulta Returns: 'low', 'medium', 'high' """ # Heurísticas simples (en producción, usar modelo ML) query_lower = query.lower() # Keywords que indican alta complejidad complex_keywords = [ "explain why", "analyze", "compare", "complex", "detailed", "step by step", "reasoning", "debug", "optimize" ] # Keywords simples simple_keywords = [ "what is", "define", "summarize", "translate", "list" ] # Longitud de la consulta token_count = len(query.split()) # Scoring complexity_score = 0 if any(kw in query_lower for kw in complex_keywords): complexity_score += 2 if any(kw in query_lower for kw in simple_keywords): complexity_score -= 1 if token_count > 100: complexity_score += 1 elif token_count < 20: complexity_score -= 1 if conversation_history and len(conversation_history) > 10: complexity_score += 1 # Conversaciones largas necesitan contexto # Clasificar if complexity_score >= 2: return 'high' elif complexity_score