mirror of
https://github.com/BagelHole/DevOps-Security-Agent-Skills.git
synced 2026-08-22 12:49:53 +02:00
Add 12 AI infrastructure and LLM operations skills
New skills covering hot-topic AI engineering subjects: Local AI Infrastructure: - vllm-server: High-throughput LLM serving with PagedAttention, tensor parallelism, quantization - llm-inference-scaling: KEDA-based GPU autoscaling for LLM inference on Kubernetes - rag-infrastructure: Production RAG with hybrid search, reranking, and embedding pipelines - llm-fine-tuning: QLoRA/LoRA fine-tuning with Axolotl, DeepSpeed ZeRO-3, and DPO alignment Infrastructure: - gpu-server-management: NVIDIA driver setup, MIG partitioning, DCGM monitoring - vector-database-ops: Qdrant, Weaviate, pgvector for production AI search - llm-gateway: LiteLLM-based API gateway with rate limiting, virtual keys, fallback routing DevOps/AI: - llm-cost-optimization: Model right-sizing, prompt/semantic caching, batch API, break-even analysis - llm-caching: Multi-layer exact + semantic + provider caching to cut costs 30-70% - ai-pipeline-orchestration: Prefect/Airflow/Dagster for RAG ingestion and training workflows Orchestration: - model-serving-kubernetes: KServe + Triton with canary deployments and GPU autoscaling Security: - ai-security-hardening: Prompt injection defense, PII scrubbing, model weight verification https://claude.ai/code/session_011MN1C4PrkCeg2Qmi7q1ZUe
This commit is contained in:
Claude
@@ -0,0 +1,262 @@
|
||||
---
|
||||
name: ai-pipeline-orchestration
|
||||
description: Orchestrate AI/ML pipelines for data ingestion, model training, batch inference, and RAG indexing using Prefect, Airflow, or Dagster. Build reliable, observable, and retriable workflows for production AI systems.
|
||||
license: MIT
|
||||
metadata:
|
||||
author: devops-skills
|
||||
version: "1.0"
|
||||
---
|
||||
|
||||
# AI Pipeline Orchestration
|
||||
|
||||
Build reliable, observable AI workflows — from document ingestion to batch inference to model training pipelines.
|
||||
|
||||
## When to Use This Skill
|
||||
|
||||
Use this skill when:
|
||||
- Scheduling recurring RAG document ingestion and re-indexing
|
||||
- Orchestrating multi-step batch LLM processing workflows
|
||||
- Running nightly model evaluation and fine-tuning jobs
|
||||
- Building ETL pipelines that feed into AI models
|
||||
- Managing dependencies between data preparation and model serving
|
||||
|
||||
## Tool Selection
|
||||
|
||||
| Tool | Best For | Complexity | GPU Jobs |
|
||||
|------|----------|------------|----------|
|
||||
| **Prefect** | Modern Python-first; easy to adopt | Low | Good |
|
||||
| **Airflow** | Complex DAGs; large teams; existing usage | High | Good |
|
||||
| **Dagster** | Asset-centric; strong data lineage | Medium | Excellent |
|
||||
| **Temporal** | Long-running workflows; reliability-first | Medium | Good |
|
||||
|
||||
## Prefect — Quick Start
|
||||
|
||||
```bash
|
||||
pip install prefect prefect-kubernetes
|
||||
|
||||
# Start Prefect server (or use Prefect Cloud)
|
||||
prefect server start
|
||||
|
||||
# In another terminal
|
||||
prefect worker start --pool default-agent-pool
|
||||
```
|
||||
|
||||
## Prefect: RAG Ingestion Pipeline
|
||||
|
||||
```python
|
||||
from prefect import flow, task, get_run_logger
|
||||
from prefect.tasks import task_input_hash
|
||||
from datetime import timedelta
|
||||
import hashlib
|
||||
|
||||
@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=24))
|
||||
def fetch_documents(source_url: str) -> list[dict]:
|
||||
"""Fetch documents from source; cached to avoid re-fetching."""
|
||||
logger = get_run_logger()
|
||||
logger.info(f"Fetching from {source_url}")
|
||||
# ... fetch logic
|
||||
return documents
|
||||
|
||||
@task(retries=3, retry_delay_seconds=30)
|
||||
def chunk_and_embed(documents: list[dict]) -> list[dict]:
|
||||
"""Chunk documents and generate embeddings with retry on failure."""
|
||||
from sentence_transformers import SentenceTransformer
|
||||
model = SentenceTransformer("BAAI/bge-large-en-v1.5")
|
||||
chunks = []
|
||||
for doc in documents:
|
||||
doc_chunks = chunk_text(doc["content"])
|
||||
embeddings = model.encode(doc_chunks, batch_size=64)
|
||||
for chunk, emb in zip(doc_chunks, embeddings):
|
||||
chunks.append({"text": chunk, "embedding": emb.tolist(),
|
||||
"source": doc["url"], "doc_hash": doc["hash"]})
|
||||
return chunks
|
||||
|
||||
@task(retries=2)
|
||||
def upsert_to_vector_store(chunks: list[dict]) -> int:
|
||||
"""Upsert embeddings to Qdrant, skip unchanged documents."""
|
||||
from qdrant_client import QdrantClient
|
||||
client = QdrantClient("http://qdrant:6333")
|
||||
client.upsert(collection_name="knowledge-base", points=[...])
|
||||
return len(chunks)
|
||||
|
||||
@flow(name="rag-ingestion", log_prints=True)
|
||||
def rag_ingestion_pipeline(sources: list[str]):
|
||||
"""Full RAG ingestion flow — runs daily."""
|
||||
logger = get_run_logger()
|
||||
total = 0
|
||||
for source in sources:
|
||||
docs = fetch_documents(source)
|
||||
chunks = chunk_and_embed(docs)
|
||||
count = upsert_to_vector_store(chunks)
|
||||
total += count
|
||||
logger.info(f"Ingested {count} chunks from {source}")
|
||||
logger.info(f"Pipeline complete: {total} total chunks indexed")
|
||||
|
||||
if __name__ == "__main__":
|
||||
rag_ingestion_pipeline.serve(
|
||||
name="daily-rag-ingestion",
|
||||
cron="0 2 * * *", # 2 AM daily
|
||||
parameters={"sources": ["https://docs.myapp.com", "https://api.myapp.com/kb"]},
|
||||
)
|
||||
```
|
||||
|
||||
## Prefect: Batch LLM Inference Pipeline
|
||||
|
||||
```python
|
||||
from prefect import flow, task
|
||||
from prefect.concurrency.sync import concurrency
|
||||
import asyncio
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
@task(retries=3, retry_delay_seconds=60)
|
||||
async def process_batch(items: list[dict], model: str = "gpt-4o-mini") -> list[dict]:
|
||||
"""Process a batch of items through LLM with rate limit protection."""
|
||||
client = AsyncOpenAI()
|
||||
async with concurrency("openai-api", occupy=len(items)): # rate limit
|
||||
tasks = [
|
||||
client.chat.completions.create(
|
||||
model=model,
|
||||
messages=[{"role": "user", "content": item["prompt"]}],
|
||||
max_tokens=256,
|
||||
)
|
||||
for item in items
|
||||
]
|
||||
responses = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
results = []
|
||||
for item, response in zip(items, responses):
|
||||
if isinstance(response, Exception):
|
||||
results.append({**item, "error": str(response), "output": None})
|
||||
else:
|
||||
results.append({**item, "output": response.choices[0].message.content})
|
||||
return results
|
||||
|
||||
@flow(name="batch-llm-inference")
|
||||
async def batch_inference_flow(input_file: str, output_file: str, batch_size: int = 50):
|
||||
import json
|
||||
items = [json.loads(line) for line in open(input_file)]
|
||||
batches = [items[i:i+batch_size] for i in range(0, len(items), batch_size)]
|
||||
|
||||
all_results = []
|
||||
for batch in batches:
|
||||
results = await process_batch(batch)
|
||||
all_results.extend(results)
|
||||
|
||||
with open(output_file, "w") as f:
|
||||
for result in all_results:
|
||||
f.write(json.dumps(result) + "\n")
|
||||
return len(all_results)
|
||||
```
|
||||
|
||||
## Airflow: Model Training DAG
|
||||
|
||||
```python
|
||||
from airflow.decorators import dag, task
|
||||
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
|
||||
from datetime import datetime
|
||||
from kubernetes.client import models as k8s
|
||||
|
||||
@dag(
|
||||
dag_id="llm_fine_tuning",
|
||||
schedule="@weekly",
|
||||
start_date=datetime(2025, 1, 1),
|
||||
catchup=False,
|
||||
tags=["ai", "training"],
|
||||
)
|
||||
def llm_fine_tuning_dag():
|
||||
|
||||
@task
|
||||
def prepare_dataset() -> str:
|
||||
"""Download and preprocess training data."""
|
||||
# ... data prep logic
|
||||
return "s3://my-bucket/training-data/2025-03-01/"
|
||||
|
||||
train = KubernetesPodOperator(
|
||||
task_id="train_model",
|
||||
name="llm-training-job",
|
||||
namespace="ml",
|
||||
image="nvcr.io/nvidia/pytorch:24.05-py3",
|
||||
cmds=["accelerate", "launch", "-m", "axolotl.cli.train", "/config/config.yaml"],
|
||||
resources=k8s.V1ResourceRequirements(
|
||||
limits={"nvidia.com/gpu": "4", "memory": "320Gi"},
|
||||
requests={"nvidia.com/gpu": "4"},
|
||||
),
|
||||
node_selector={"nvidia.com/gpu.product": "A100-SXM4-80GB"},
|
||||
volumes=[...],
|
||||
volume_mounts=[...],
|
||||
get_logs=True,
|
||||
is_delete_operator_pod=True,
|
||||
)
|
||||
|
||||
@task
|
||||
def evaluate_model(dataset_path: str) -> dict:
|
||||
"""Run evals; fail pipeline if quality drops."""
|
||||
metrics = run_evals()
|
||||
if metrics["accuracy"] < 0.85:
|
||||
raise ValueError(f"Model quality too low: {metrics}")
|
||||
return metrics
|
||||
|
||||
@task
|
||||
def deploy_model(metrics: dict):
|
||||
"""Push merged model to HF Hub and update vLLM config."""
|
||||
update_serving_config(new_model="org/fine-tuned-v2")
|
||||
|
||||
dataset = prepare_dataset()
|
||||
train.set_upstream(dataset)
|
||||
eval_result = evaluate_model(dataset)
|
||||
eval_result.set_upstream(train)
|
||||
deploy_model(eval_result)
|
||||
|
||||
llm_fine_tuning_dag()
|
||||
```
|
||||
|
||||
## Dagster: Asset-Based AI Pipeline
|
||||
|
||||
```python
|
||||
from dagster import asset, AssetExecutionContext, define_asset_job, ScheduleDefinition
|
||||
|
||||
@asset(description="Raw documents fetched from knowledge sources")
|
||||
def raw_documents(context: AssetExecutionContext) -> list[dict]:
|
||||
context.log.info("Fetching documents...")
|
||||
return fetch_all_documents()
|
||||
|
||||
@asset(
|
||||
deps=[raw_documents],
|
||||
description="Chunked and embedded document vectors",
|
||||
)
|
||||
def document_embeddings(context: AssetExecutionContext, raw_documents) -> int:
|
||||
chunks = process_and_embed(raw_documents)
|
||||
context.log.info(f"Generated {len(chunks)} embeddings")
|
||||
upsert_to_qdrant(chunks)
|
||||
return len(chunks)
|
||||
|
||||
@asset(
|
||||
deps=[document_embeddings],
|
||||
description="RAG system quality metrics",
|
||||
)
|
||||
def rag_quality_metrics(context: AssetExecutionContext) -> dict:
|
||||
metrics = evaluate_rag_system()
|
||||
context.add_output_metadata({"ragas_score": metrics["ragas_score"]})
|
||||
return metrics
|
||||
|
||||
# Schedule: refresh embeddings nightly
|
||||
nightly_refresh = ScheduleDefinition(
|
||||
job=define_asset_job("rag_refresh_job", [raw_documents, document_embeddings]),
|
||||
cron_schedule="0 1 * * *",
|
||||
)
|
||||
```
|
||||
|
||||
## Best Practices
|
||||
|
||||
- Use task-level retries for API calls; use flow-level retries for transient infra failures.
|
||||
- Cache expensive steps (embedding generation, data fetching) to speed up reruns.
|
||||
- Emit custom metrics from pipelines (chunk count, error rate, cost) to your observability stack.
|
||||
- Use `concurrency` limits in Prefect or `pool` slots in Airflow to respect external rate limits.
|
||||
- Separate ingestion, training, and deployment pipelines — don't couple them in one giant DAG.
|
||||
|
||||
## Related Skills
|
||||
|
||||
- [rag-infrastructure](../../infrastructure/local-ai/rag-infrastructure/) - RAG system setup
|
||||
- [llm-fine-tuning](../../infrastructure/local-ai/llm-fine-tuning/) - Training jobs
|
||||
- [agent-observability](../agent-observability/) - Pipeline monitoring
|
||||
- [kubernetes-ops](../orchestration/kubernetes-ops/) - Running pipeline pods on K8s
|
||||
@@ -0,0 +1,309 @@
|
||||
---
|
||||
name: llm-caching
|
||||
description: Implement multi-layer LLM caching with exact match, semantic similarity, and provider-side prompt caching. Reduce API costs by 30–70%, cut latency, and improve throughput using Redis, GPTCache, and provider caching APIs.
|
||||
license: MIT
|
||||
metadata:
|
||||
author: devops-skills
|
||||
version: "1.0"
|
||||
---
|
||||
|
||||
# LLM Caching
|
||||
|
||||
Cut LLM costs and latency with exact match, semantic, and provider-side caching layers.
|
||||
|
||||
## When to Use This Skill
|
||||
|
||||
Use this skill when:
|
||||
- The same or similar queries are asked repeatedly (FAQ bots, support tools)
|
||||
- LLM API costs are growing and you need immediate savings
|
||||
- Serving high request volumes where repeated queries cause bottlenecks
|
||||
- Implementing prompt caching for long system prompts (Anthropic/OpenAI)
|
||||
- Building offline-capable AI features that need response persistence
|
||||
|
||||
## Caching Layers
|
||||
|
||||
```
|
||||
Request → Exact Cache → Semantic Cache → Provider Cache → LLM API
|
||||
↓ hit ↓ hit ↓ hit
|
||||
instant ~5ms 50-80% cheaper
|
||||
```
|
||||
|
||||
## Layer 1: Exact Match Cache (Redis)
|
||||
|
||||
```python
|
||||
import hashlib
|
||||
import json
|
||||
import redis
|
||||
from openai import OpenAI
|
||||
|
||||
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
|
||||
client = OpenAI()
|
||||
|
||||
def build_cache_key(model: str, messages: list, temperature: float) -> str:
|
||||
"""Deterministic key from request parameters."""
|
||||
payload = json.dumps({
|
||||
"model": model,
|
||||
"messages": messages,
|
||||
"temperature": temperature,
|
||||
}, sort_keys=True)
|
||||
return f"llm:exact:{hashlib.sha256(payload.encode()).hexdigest()}"
|
||||
|
||||
def cached_completion(model: str, messages: list, temperature: float = 0.0,
|
||||
ttl: int = 3600) -> dict:
|
||||
key = build_cache_key(model, messages, temperature)
|
||||
|
||||
# Check cache
|
||||
if cached := r.get(key):
|
||||
return json.loads(cached)
|
||||
|
||||
# Call API
|
||||
response = client.chat.completions.create(
|
||||
model=model, messages=messages, temperature=temperature
|
||||
)
|
||||
result = response.model_dump()
|
||||
|
||||
# Cache result (only cache deterministic responses)
|
||||
if temperature == 0.0:
|
||||
r.setex(key, ttl, json.dumps(result))
|
||||
|
||||
return result
|
||||
```
|
||||
|
||||
## Layer 2: Semantic Cache (GPTCache)
|
||||
|
||||
```python
|
||||
from gptcache import cache, Config
|
||||
from gptcache.adapter import openai
|
||||
from gptcache.embedding import Onnx
|
||||
from gptcache.manager import CacheBase, VectorBase, get_data_manager
|
||||
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation
|
||||
|
||||
# Configure GPTCache with Qdrant backend
|
||||
def init_gptcache(cache_obj, llm: str):
|
||||
onnx = Onnx() # local embedding model
|
||||
data_manager = get_data_manager(
|
||||
CacheBase("redis"), # metadata store
|
||||
VectorBase("qdrant",
|
||||
host="localhost",
|
||||
port=6333,
|
||||
collection_name=f"llm-cache-{llm}",
|
||||
dimension=onnx.dimension),
|
||||
)
|
||||
cache_obj.init(
|
||||
embedding_func=onnx.to_embeddings,
|
||||
data_manager=data_manager,
|
||||
similarity_evaluation=SearchDistanceEvaluation(),
|
||||
config=Config(similarity_threshold=0.80), # 80% similarity = cache hit
|
||||
)
|
||||
|
||||
cache.set_openai_key()
|
||||
init_gptcache(cache, "gpt-4o-mini")
|
||||
|
||||
# Now openai calls are automatically cached
|
||||
response = openai.ChatCompletion.create(
|
||||
model="gpt-4o-mini",
|
||||
messages=[{"role": "user", "content": "What is machine learning?"}],
|
||||
)
|
||||
# Second call with similar question ("Explain machine learning") → cache hit
|
||||
```
|
||||
|
||||
## Custom Semantic Cache (Production-Grade)
|
||||
|
||||
```python
|
||||
from sentence_transformers import SentenceTransformer
|
||||
from qdrant_client import QdrantClient
|
||||
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, Range
|
||||
import numpy as np
|
||||
import uuid
|
||||
import time
|
||||
|
||||
embed_model = SentenceTransformer("BAAI/bge-small-en-v1.5") # fast, 33M params
|
||||
qdrant = QdrantClient("http://localhost:6333")
|
||||
|
||||
CACHE_COLLECTION = "semantic-cache"
|
||||
SIMILARITY_THRESHOLD = 0.88
|
||||
CACHE_TTL_SECONDS = 86400 # 24h
|
||||
|
||||
# Create collection once
|
||||
qdrant.create_collection(
|
||||
collection_name=CACHE_COLLECTION,
|
||||
vectors_config=VectorParams(size=384, distance=Distance.COSINE),
|
||||
on_disk_payload=True,
|
||||
)
|
||||
|
||||
def semantic_cache_lookup(query: str, model: str) -> str | None:
|
||||
embedding = embed_model.encode(query).tolist()
|
||||
results = qdrant.query_points(
|
||||
collection_name=CACHE_COLLECTION,
|
||||
query=embedding,
|
||||
query_filter=Filter(must=[
|
||||
FieldCondition(key="model", match={"value": model}),
|
||||
FieldCondition(key="expires_at", range=Range(gte=time.time())),
|
||||
]),
|
||||
limit=1,
|
||||
score_threshold=SIMILARITY_THRESHOLD,
|
||||
)
|
||||
if results.points:
|
||||
return results.points[0].payload["response"]
|
||||
return None
|
||||
|
||||
def semantic_cache_store(query: str, response: str, model: str):
|
||||
embedding = embed_model.encode(query).tolist()
|
||||
qdrant.upsert(
|
||||
collection_name=CACHE_COLLECTION,
|
||||
points=[PointStruct(
|
||||
id=str(uuid.uuid4()),
|
||||
vector=embedding,
|
||||
payload={
|
||||
"query": query,
|
||||
"response": response,
|
||||
"model": model,
|
||||
"created_at": time.time(),
|
||||
"expires_at": time.time() + CACHE_TTL_SECONDS,
|
||||
},
|
||||
)],
|
||||
)
|
||||
|
||||
def smart_llm_call(query: str, model: str = "gpt-4o-mini") -> dict:
|
||||
# 1. Semantic lookup
|
||||
if cached_response := semantic_cache_lookup(query, model):
|
||||
return {"response": cached_response, "source": "semantic_cache", "cost": 0}
|
||||
|
||||
# 2. LLM call
|
||||
response = client.chat.completions.create(
|
||||
model=model,
|
||||
messages=[{"role": "user", "content": query}],
|
||||
)
|
||||
text = response.choices[0].message.content
|
||||
cost = litellm.completion_cost(response)
|
||||
|
||||
# 3. Store in cache
|
||||
semantic_cache_store(query, text, model)
|
||||
|
||||
return {"response": text, "source": "llm_api", "cost": cost}
|
||||
```
|
||||
|
||||
## Layer 3: Provider-Side Prompt Caching
|
||||
|
||||
```python
|
||||
# Anthropic — cache long system prompts (saves 90% on cached input tokens)
|
||||
import anthropic
|
||||
|
||||
client = anthropic.Anthropic()
|
||||
|
||||
# Long system prompt — mark for caching
|
||||
SYSTEM_PROMPT = open("knowledge-base.txt").read() # e.g., 50k tokens
|
||||
|
||||
def call_with_prompt_cache(user_question: str) -> str:
|
||||
response = client.messages.create(
|
||||
model="claude-sonnet-4-6",
|
||||
max_tokens=1024,
|
||||
system=[
|
||||
{"type": "text", "text": "You are a helpful assistant."},
|
||||
{
|
||||
"type": "text",
|
||||
"text": SYSTEM_PROMPT,
|
||||
"cache_control": {"type": "ephemeral"}, # cache this block
|
||||
}
|
||||
],
|
||||
messages=[{"role": "user", "content": user_question}],
|
||||
)
|
||||
# Log cache efficiency
|
||||
usage = response.usage
|
||||
cache_savings = usage.cache_read_input_tokens * 0.9 # 90% discount on cached
|
||||
print(f"Cache hits: {usage.cache_read_input_tokens} tokens "
|
||||
f"(saved ~${cache_savings * 3.0 / 1_000_000:.4f})")
|
||||
return response.content[0].text
|
||||
|
||||
# OpenAI — automatic for repeated prefixes (≥1,024 tokens)
|
||||
# No code change needed; cached tokens appear in usage.prompt_tokens_details
|
||||
response = client.chat.completions.create(
|
||||
model="gpt-4o-mini",
|
||||
messages=[
|
||||
{"role": "system", "content": LONG_SYSTEM_PROMPT}, # auto-cached
|
||||
{"role": "user", "content": user_question},
|
||||
]
|
||||
)
|
||||
cached = response.usage.prompt_tokens_details.cached_tokens
|
||||
print(f"OpenAI cached {cached} tokens")
|
||||
```
|
||||
|
||||
## Cache Warming
|
||||
|
||||
```python
|
||||
async def warm_cache(common_queries: list[str], model: str):
|
||||
"""Pre-populate cache with known frequent queries."""
|
||||
import asyncio
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
aclient = AsyncOpenAI()
|
||||
|
||||
async def warm_single(query: str):
|
||||
if not semantic_cache_lookup(query, model):
|
||||
response = await aclient.chat.completions.create(
|
||||
model=model,
|
||||
messages=[{"role": "user", "content": query}],
|
||||
)
|
||||
text = response.choices[0].message.content
|
||||
semantic_cache_store(query, text, model)
|
||||
print(f"Warmed: {query[:50]}...")
|
||||
|
||||
await asyncio.gather(*[warm_single(q) for q in common_queries])
|
||||
|
||||
# Warm on startup
|
||||
import asyncio
|
||||
asyncio.run(warm_cache(FREQUENT_QUERIES, "gpt-4o-mini"))
|
||||
```
|
||||
|
||||
## Cache Metrics
|
||||
|
||||
```python
|
||||
from prometheus_client import Counter, Histogram
|
||||
|
||||
cache_hits = Counter("llm_cache_hits_total", "Cache hits", ["cache_layer", "model"])
|
||||
cache_misses = Counter("llm_cache_misses_total", "Cache misses", ["model"])
|
||||
cache_savings_usd = Counter("llm_cache_savings_usd_total", "USD saved by cache", ["model"])
|
||||
|
||||
# Use in your smart_llm_call function
|
||||
if source == "semantic_cache":
|
||||
cache_hits.labels(cache_layer="semantic", model=model).inc()
|
||||
cache_savings_usd.labels(model=model).inc(estimated_cost)
|
||||
else:
|
||||
cache_misses.labels(model=model).inc()
|
||||
```
|
||||
|
||||
## Redis Configuration for LLM Caching
|
||||
|
||||
```bash
|
||||
# redis.conf tuning for LLM cache workload
|
||||
maxmemory 8gb
|
||||
maxmemory-policy allkeys-lru # evict least-recently-used when full
|
||||
save "" # disable persistence (cache is ephemeral)
|
||||
appendonly no
|
||||
tcp-keepalive 60
|
||||
```
|
||||
|
||||
## Common Issues
|
||||
|
||||
| Issue | Cause | Fix |
|
||||
|-------|-------|-----|
|
||||
| Low cache hit rate | Threshold too strict | Lower `SIMILARITY_THRESHOLD` to 0.82–0.85 |
|
||||
| Stale cached responses | Long TTL | Use topic-specific TTLs; invalidate on data updates |
|
||||
| Cache serving wrong answers | Threshold too loose | Raise threshold or add model-name filtering |
|
||||
| Redis OOM | No eviction policy | Set `maxmemory` + `allkeys-lru` |
|
||||
| Slow semantic lookup | Large cache collection | Add payload index on `model` + `expires_at` |
|
||||
|
||||
## Best Practices
|
||||
|
||||
- Start with exact cache — zero cost, instant wins for identical queries.
|
||||
- Semantic threshold of 0.88–0.92 balances hit rate vs. accuracy; tune with your data.
|
||||
- Set per-model TTLs: longer for stable knowledge (1 week), shorter for news/events (1 hour).
|
||||
- Always filter by model name in semantic cache — different models give different answers.
|
||||
- Log cache hit rate as a KPI; target 30%+ for FAQ-style applications.
|
||||
|
||||
## Related Skills
|
||||
|
||||
- [llm-cost-optimization](../llm-cost-optimization/) - Full cost strategy
|
||||
- [llm-gateway](../../infrastructure/networking/llm-gateway/) - Gateway-level caching
|
||||
- [vector-database-ops](../../infrastructure/databases/vector-database-ops/) - Qdrant setup
|
||||
- [agent-observability](../agent-observability/) - Cache metrics dashboards
|
||||
@@ -0,0 +1,286 @@
|
||||
---
|
||||
name: llm-cost-optimization
|
||||
description: Reduce LLM API and infrastructure costs through model selection, prompt caching, batching, caching, quantization, and self-hosting strategies. Track spend by team and model, set budgets, and implement cost-aware routing.
|
||||
license: MIT
|
||||
metadata:
|
||||
author: devops-skills
|
||||
version: "1.0"
|
||||
---
|
||||
|
||||
# LLM Cost Optimization
|
||||
|
||||
Cut LLM costs by 50–90% with the right combination of caching, model selection, prompt optimization, and self-hosting.
|
||||
|
||||
## When to Use This Skill
|
||||
|
||||
Use this skill when:
|
||||
- LLM API spend is growing faster than revenue
|
||||
- You need to attribute AI costs to teams, products, or customers
|
||||
- Implementing caching to avoid redundant LLM calls
|
||||
- Deciding when to switch from API providers to self-hosted models
|
||||
- Optimizing prompt length without sacrificing quality
|
||||
|
||||
## Cost Levers by Impact
|
||||
|
||||
| Strategy | Typical Savings | Effort |
|
||||
|----------|-----------------|--------|
|
||||
| Semantic caching | 20–50% | Low |
|
||||
| Model right-sizing | 30–70% | Low |
|
||||
| Prompt compression | 10–30% | Medium |
|
||||
| Provider caching (prompt cache) | 10–25% | Low |
|
||||
| Batching offline workloads | 50% (Batch API) | Medium |
|
||||
| Self-hosting 7–8B models | 80–95% at scale | High |
|
||||
| Quantization | 30–50% VRAM cost | Medium |
|
||||
|
||||
## Track Costs First
|
||||
|
||||
```python
|
||||
# Use LiteLLM's cost tracking (automatic per-model pricing)
|
||||
import litellm
|
||||
|
||||
response = litellm.completion(
|
||||
model="gpt-4o-mini",
|
||||
messages=[{"role": "user", "content": "Hello"}],
|
||||
)
|
||||
cost = litellm.completion_cost(response)
|
||||
print(f"Cost: ${cost:.6f}")
|
||||
|
||||
# Add custom cost callbacks
|
||||
def log_cost(kwargs, completion_response, start_time, end_time):
|
||||
cost = kwargs.get("response_cost", 0)
|
||||
model = kwargs.get("model")
|
||||
user = kwargs.get("user")
|
||||
# Send to your analytics DB
|
||||
db.record_cost(user=user, model=model, cost=cost)
|
||||
|
||||
litellm.success_callback = [log_cost]
|
||||
```
|
||||
|
||||
## Model Right-Sizing
|
||||
|
||||
```python
|
||||
# Route by task complexity — don't use GPT-4o for everything
|
||||
def get_model_for_task(task_type: str) -> str:
|
||||
routing = {
|
||||
"classification": "gpt-4o-mini", # ~30× cheaper than gpt-4o
|
||||
"summarization": "gpt-4o-mini",
|
||||
"extraction": "gpt-4o-mini",
|
||||
"simple_qa": "gpt-4o-mini",
|
||||
"complex_reasoning": "gpt-4o",
|
||||
"code_generation": "claude-sonnet-4-6",
|
||||
"creative_writing": "claude-opus-4-6",
|
||||
}
|
||||
return routing.get(task_type, "gpt-4o-mini")
|
||||
|
||||
# Cost comparison (per 1M tokens, 2025 approx.)
|
||||
# gpt-4o-mini: input $0.15 / output $0.60
|
||||
# gpt-4o: input $2.50 / output $10.00
|
||||
# claude-sonnet-4-6: input $3.00 / output $15.00
|
||||
# llama-3.1-8b (self): ~$0.05–0.10 all-in (GPU amortized)
|
||||
```
|
||||
|
||||
## Prompt Caching (Provider-Side)
|
||||
|
||||
```python
|
||||
# Anthropic — cache long system prompts (saves 90% on cached tokens)
|
||||
import anthropic
|
||||
|
||||
client = anthropic.Anthropic()
|
||||
|
||||
response = client.messages.create(
|
||||
model="claude-sonnet-4-6",
|
||||
max_tokens=1024,
|
||||
system=[
|
||||
{
|
||||
"type": "text",
|
||||
"text": "You are a helpful assistant.",
|
||||
},
|
||||
{
|
||||
"type": "text",
|
||||
"text": open("large-context.txt").read(), # large doc
|
||||
"cache_control": {"type": "ephemeral"}, # cache this!
|
||||
}
|
||||
],
|
||||
messages=[{"role": "user", "content": "Summarize the key points."}],
|
||||
)
|
||||
# First call: full price. Subsequent calls: 90% discount on cached part.
|
||||
print(f"Cache read tokens: {response.usage.cache_read_input_tokens}")
|
||||
|
||||
# OpenAI — prompt caching is automatic for repeated prefixes >1024 tokens
|
||||
# No code change needed; check usage.prompt_tokens_details.cached_tokens
|
||||
```
|
||||
|
||||
## Batching with OpenAI Batch API (50% Discount)
|
||||
|
||||
```python
|
||||
import json
|
||||
from openai import OpenAI
|
||||
|
||||
client = OpenAI()
|
||||
|
||||
# Prepare batch requests
|
||||
requests = [
|
||||
{
|
||||
"custom_id": f"task-{i}",
|
||||
"method": "POST",
|
||||
"url": "/v1/chat/completions",
|
||||
"body": {
|
||||
"model": "gpt-4o-mini",
|
||||
"messages": [{"role": "user", "content": f"Classify: {text}"}],
|
||||
"max_tokens": 50,
|
||||
}
|
||||
}
|
||||
for i, text in enumerate(texts)
|
||||
]
|
||||
|
||||
# Write JSONL file
|
||||
with open("batch.jsonl", "w") as f:
|
||||
for req in requests:
|
||||
f.write(json.dumps(req) + "\n")
|
||||
|
||||
# Upload and create batch
|
||||
batch_file = client.files.create(file=open("batch.jsonl", "rb"), purpose="batch")
|
||||
batch = client.batches.create(
|
||||
input_file_id=batch_file.id,
|
||||
endpoint="/v1/chat/completions",
|
||||
completion_window="24h",
|
||||
)
|
||||
print(f"Batch ID: {batch.id}") # poll status with client.batches.retrieve(batch.id)
|
||||
```
|
||||
|
||||
## Semantic Caching
|
||||
|
||||
```python
|
||||
import hashlib
|
||||
import json
|
||||
import redis
|
||||
import numpy as np
|
||||
from sentence_transformers import SentenceTransformer
|
||||
|
||||
r = redis.Redis(host="localhost", port=6379)
|
||||
embed_model = SentenceTransformer("BAAI/bge-small-en-v1.5")
|
||||
|
||||
SIMILARITY_THRESHOLD = 0.92
|
||||
CACHE_TTL = 3600 * 24 # 24 hours
|
||||
|
||||
def cached_llm_call(prompt: str, llm_fn) -> str:
|
||||
# 1. Exact match (free)
|
||||
exact_key = f"exact:{hashlib.sha256(prompt.encode()).hexdigest()}"
|
||||
if cached := r.get(exact_key):
|
||||
return cached.decode()
|
||||
|
||||
# 2. Semantic match
|
||||
query_vec = embed_model.encode(prompt)
|
||||
cached_keys = r.keys("sem:*")
|
||||
for key in cached_keys:
|
||||
data = json.loads(r.get(key))
|
||||
similarity = np.dot(query_vec, data["embedding"]) / (
|
||||
np.linalg.norm(query_vec) * np.linalg.norm(data["embedding"])
|
||||
)
|
||||
if similarity >= SIMILARITY_THRESHOLD:
|
||||
return data["response"]
|
||||
|
||||
# 3. Cache miss — call LLM
|
||||
response = llm_fn(prompt)
|
||||
|
||||
# Store exact match
|
||||
r.setex(exact_key, CACHE_TTL, response)
|
||||
|
||||
# Store semantic embedding
|
||||
sem_key = f"sem:{hashlib.sha256(prompt.encode()).hexdigest()}"
|
||||
r.setex(sem_key, CACHE_TTL, json.dumps({
|
||||
"embedding": query_vec.tolist(),
|
||||
"response": response,
|
||||
"prompt": prompt,
|
||||
}))
|
||||
return response
|
||||
```
|
||||
|
||||
## Prompt Compression
|
||||
|
||||
```python
|
||||
# LLMLingua — compress long prompts by 3–20× with minimal quality loss
|
||||
from llmlingua import PromptCompressor
|
||||
|
||||
compressor = PromptCompressor(
|
||||
model_name="microsoft/llmlingua-2-bert-base-multilingual-cased-meetingbank",
|
||||
device_map="cpu",
|
||||
)
|
||||
|
||||
compressed = compressor.compress_prompt(
|
||||
long_context,
|
||||
ratio=0.5, # keep 50% of tokens
|
||||
rank_method="longllmlingua",
|
||||
)
|
||||
print(f"Original: {len(long_context.split())} words")
|
||||
print(f"Compressed: {len(compressed['compressed_prompt'].split())} words")
|
||||
print(f"Savings: {compressed['saving']}")
|
||||
```
|
||||
|
||||
## Self-Hosting Break-Even Calculator
|
||||
|
||||
```python
|
||||
def break_even_analysis(
|
||||
monthly_api_spend_usd: float,
|
||||
gpu_cost_per_hour_usd: float = 2.50, # e.g., A10G on AWS
|
||||
utilization: float = 0.70, # 70% GPU utilization
|
||||
) -> dict:
|
||||
monthly_gpu_cost = gpu_cost_per_hour_usd * 24 * 30 * utilization
|
||||
break_even = monthly_gpu_cost / monthly_api_spend_usd
|
||||
recommendation = (
|
||||
"Self-host now — strong ROI" if break_even < 0.5 else
|
||||
"Self-host if traffic grows 2×" if break_even < 0.8 else
|
||||
"Stick with API — not enough scale yet"
|
||||
)
|
||||
return {
|
||||
"monthly_gpu_cost": f"${monthly_gpu_cost:.0f}",
|
||||
"monthly_api_spend": f"${monthly_api_spend_usd:.0f}",
|
||||
"gpu_as_pct_of_api": f"{break_even*100:.0f}%",
|
||||
"recommendation": recommendation,
|
||||
}
|
||||
|
||||
# Example: $5k/month on OpenAI, $2.50/hr A10G
|
||||
print(break_even_analysis(5000))
|
||||
# → gpu_cost ~$1,260/mo = 25% of API spend → self-host now
|
||||
```
|
||||
|
||||
## Cost Dashboard (Grafana)
|
||||
|
||||
```python
|
||||
# Emit cost metrics to Prometheus
|
||||
from prometheus_client import Counter, Histogram
|
||||
|
||||
llm_cost_total = Counter(
|
||||
"llm_cost_usd_total",
|
||||
"Total LLM spend in USD",
|
||||
["model", "team", "task_type"],
|
||||
)
|
||||
llm_tokens_total = Counter(
|
||||
"llm_tokens_total",
|
||||
"Total tokens used",
|
||||
["model", "token_type"], # token_type: prompt, completion, cached
|
||||
)
|
||||
|
||||
def track_call(model, team, task_type, response):
|
||||
cost = calculate_cost(model, response.usage)
|
||||
llm_cost_total.labels(model=model, team=team, task_type=task_type).inc(cost)
|
||||
llm_tokens_total.labels(model=model, token_type="prompt").inc(
|
||||
response.usage.prompt_tokens)
|
||||
llm_tokens_total.labels(model=model, token_type="completion").inc(
|
||||
response.usage.completion_tokens)
|
||||
```
|
||||
|
||||
## Best Practices
|
||||
|
||||
- Use `gpt-4o-mini` or `claude-haiku` for 80% of tasks — they're 10–30× cheaper.
|
||||
- Enable prompt caching for system prompts >1,024 tokens (Anthropic) or >1,024 tokens (OpenAI).
|
||||
- Audit your top 5 prompts by token count — compress or cache them.
|
||||
- Set hard budget limits with LiteLLM virtual keys before costs spiral.
|
||||
- Self-host 7B–8B models when monthly API spend exceeds $2k/month.
|
||||
|
||||
## Related Skills
|
||||
|
||||
- [llm-gateway](../../infrastructure/networking/llm-gateway/) - Centralized cost control
|
||||
- [llm-caching](../llm-caching/) - Semantic caching patterns
|
||||
- [vllm-server](../../infrastructure/local-ai/vllm-server/) - Self-hosted inference
|
||||
- [agent-observability](../agent-observability/) - Token and cost telemetry
|
||||
Reference in New Issue
Block a user