Menu
Expert in LLM API integration, error handling, and production patterns
The review-first prompt is selected by default. You can switch to a direct command or download a local copy.
Read SKILL.md and any companion files shown by SkillsMP before deciding whether to install.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/tools-only/X-Skills --skill llm-integration-expertThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Showing SKILL.md
Based on SOC occupation classification
| name | llm-integration-expert |
| type | agent |
| description | Expert in LLM API integration, error handling, and production patterns |
| category | ai |
| version | 1.0.0 |
| author | Jeremy Longshore |
| activation_triggers | ["llm integration","api integration","openai api","anthropic api","llm error handling"] |
| capabilities | ["API integration patterns and best practices","Error handling and retry logic","Rate limiting and backoff strategies","Streaming responses implementation","Token counting and cost tracking","Multi-provider fallback systems"] |
You are an expert in integrating Large Language Model APIs into production applications. You understand API design patterns, error handling, rate limiting, streaming, and cost optimization for LLM services.
Major Providers:
API Characteristics:
import anthropic
from typing import Optional
class LLMClient:
"""Production-ready LLM client with error handling."""
def __init__(self, api_key: str):
self.client = anthropic.Anthropic(api_key=api_key)
def complete(
self,
prompt: str,
model: str = "claude-3-haiku-20240307",
max_tokens: int = 1024,
temperature: float = 1.0
) -> dict:
"""Generate completion with comprehensive error handling."""
try:
message = self.client.messages.create(
model=model,
max_tokens=max_tokens,
temperature=temperature,
messages=[{: , : prompt}]
)
{
: ,
: message.content[].text,
: {
: message.usage.input_tokens,
: message.usage.output_tokens
},
: model
}
anthropic.RateLimitError e:
{
: ,
: ,
: ,
: e.response.headers.get()
}
anthropic.APIConnectionError e:
{
: ,
: ,
: ,
: (e)
}
anthropic.APIError e:
{
: ,
: ,
: ,
: (e)
}
Exception e:
{
: ,
: ,
: ,
: (e)
}
client = LLMClient(api_key=)
result = client.complete()
result[]:
(result[])
:
()
import asyncio
from anthropic import AsyncAnthropic
class StreamingLLMClient:
"""Stream LLM responses for better user experience."""
def __init__(self, api_key: str):
self.client = AsyncAnthropic(api_key=api_key)
async def stream_complete(
self,
prompt: str,
on_token: callable,
model: str = "claude-3-haiku-20240307",
max_tokens: int = 1024
):
"""Stream tokens as they're generated."""
try:
async with self.client.messages.stream(
model=model,
max_tokens=max_tokens,
messages=[{"role": "user", "content": prompt}]
) as stream:
async for text in stream.text_stream:
await on_token(text) # Call handler for each token
# Get final message
message = await stream.get_final_message()
return {
"success": True,
"usage": {
"input_tokens": message.usage.input_tokens,
"output_tokens": message.usage.output_tokens
}
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
# Usage
async def handle_token(token: str):
"""Process each token as it arrives."""
print(token, end="", flush=True)
client = StreamingLLMClient(api_key="your-key")
result = await client.stream_complete(
"Write a short story about AI",
on_token=handle_token
)
import time
import random
from functools import wraps
def retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True
):
"""Decorator for retry logic with exponential backoff."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while retries < max_retries:
try:
return func(*args, **kwargs)
except Exception as e:
retries += 1
if retries >= max_retries:
raise # Re-raise after max retries
# Calculate delay with exponential backoff
delay = min(base_delay * (exponential_base ** retries), max_delay)
# Add jitter to avoid thundering herd
if jitter:
delay = delay * (0.5 + random.random())
print(f"Retry {retries}/{max_retries} after {delay:.2f}s: {e}")
time.sleep(delay)
return func(*args, **kwargs) # Final attempt
return wrapper
return decorator
class RobustLLMClient:
"""LLM client with automatic retries."""
def __init__(self, api_key: str):
self.client = anthropic.Anthropic(api_key=api_key)
@retry_with_backoff(max_retries=3, base_delay=1.0)
def complete(self, prompt: str, **kwargs):
"""Complete with automatic retries on transient failures."""
return self.client.messages.create(
messages=[{"role": "user", "content": prompt}],
**kwargs
)
# Usage
client = RobustLLMClient(api_key="your-key")
response = client.complete("Explain ML", model="claude-3-haiku-20240307", max_tokens=500)
import time
import threading
class TokenBucket:
"""Thread-safe token bucket for rate limiting."""
def __init__(self, capacity: int, refill_rate: float):
"""
Args:
capacity: Maximum tokens in bucket (e.g., 10 requests)
refill_rate: Tokens added per second (e.g., 2 requests/second)
"""
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate
self.last_refill = time.time()
self.lock = threading.Lock()
def consume(self, tokens: int = 1) -> bool:
"""Attempt to consume tokens. Returns True if successful."""
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def _refill(self):
"""Refill tokens based on time elapsed."""
now = time.time()
elapsed = now - self.last_refill
tokens_to_add = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + tokens_to_add)
self.last_refill = now
def wait_for_token(self, tokens: int = 1):
"""Block until tokens are available."""
while not self.consume(tokens):
time.sleep(0.1)
class RateLimitedLLMClient:
"""LLM client with rate limiting."""
def __init__(self, api_key: str, requests_per_minute: int = 50):
self.client = anthropic.Anthropic(api_key=api_key)
self.rate_limiter = TokenBucket(
capacity=requests_per_minute,
refill_rate=requests_per_minute / 60.0 # Per second
)
def complete(self, prompt: str, **kwargs):
"""Complete with rate limiting."""
self.rate_limiter.wait_for_token() # Block if rate limit exceeded
return self.client.messages.create(
messages=[{"role": "user", "content": prompt}],
**kwargs
)
# Usage
client = RateLimitedLLMClient(api_key="your-key", requests_per_minute=50)
# These requests will be automatically rate-limited
for i in range(100):
response = client.complete(f"Question {i}", model="claude-3-haiku-20240307", max_tokens=100)
from enum import Enum
from dataclasses import dataclass
from typing import Optional
class Provider(Enum):
ANTHROPIC = "anthropic"
OPENAI = "openai"
GOOGLE = "google"
@dataclass
class LLMConfig:
"""Configuration for LLM provider."""
provider: Provider
api_key: str
model: str
priority: int # Lower = higher priority
class MultiproviderLLMClient:
"""LLM client with automatic fallback across providers."""
def __init__(self, configs: list[LLMConfig]):
"""Initialize with multiple provider configurations."""
self.configs = sorted(configs, key=lambda c: c.priority)
self.clients = {}
for config in self.configs:
if config.provider == Provider.ANTHROPIC:
self.clients[config.provider] = anthropic.Anthropic(api_key=config.api_key)
elif config.provider == Provider.OPENAI:
import openai
self.clients[config.provider] = openai.OpenAI(api_key=config.api_key)
# Add more providers...
def complete(self, prompt: str, max_tokens: int = 1024) -> dict:
"""Try providers in priority order until success."""
last_error = None
for config in self.configs:
try:
if config.provider == Provider.ANTHROPIC:
response = self.clients[config.provider].messages.create(
model=config.model,
max_tokens=max_tokens,
messages=[{"role": "user", "content": prompt}]
)
return {
"success": True,
"content": response.content[0].text,
"provider": config.provider.value
}
elif config.provider == Provider.OPENAI:
response = self.clients[config.provider].chat.completions.create(
model=config.model,
max_tokens=max_tokens,
messages=[{"role": "user", "content": prompt}]
)
return {
"success": True,
"content": response.choices[0].message.content,
"provider": config.provider.value
}
except Exception as e:
last_error = e
print(f"Provider {config.provider.value} failed: {e}")
continue # Try next provider
return {
"success": False,
"error": "all_providers_failed",
"last_error": str(last_error)
}
# Usage
client = MultiproviderLLMClient([
LLMConfig(Provider.ANTHROPIC, "anthropic-key", "claude-3-haiku-20240307", priority=1),
LLMConfig(Provider.OPENAI, "openai-key", "gpt-3.5-turbo", priority=2),
])
result = client.complete("Explain AI")
if result["success"]:
print(f"Response from {result['provider']}: {result['content']}")
Common LLM API Errors:
| Error Type | Cause | Handling Strategy |
|---|---|---|
| Rate Limit (429) | Too many requests | Exponential backoff, retry |
| Context Length (400) | Prompt too long | Truncate, summarize, or split |
| Invalid API Key (401) | Bad authentication | Fail fast, alert ops team |
| Server Error (500/502/503) | Provider issue | Retry with backoff |
| Timeout | Slow response | Set reasonable timeout, retry |
| Connection Error | Network issue | Retry, check connectivity |
| Content Policy (400) | Blocked content | Log, return generic error |
Error Response Template:
{
"success": False,
"error_code": "rate_limit_exceeded",
"error_message": "User-friendly message",
"retry_after": 30, # Seconds (if applicable)
"details": { # For debugging
"provider": "anthropic",
"model": "claude-3-haiku",
"status_code": 429,
"raw_error": "..."
}
}
import tiktoken
class CostTracker:
"""Track LLM usage and costs."""
def __init__(self):
self.usage_history = []
def count_tokens(self, text: str, model: str = "gpt-4") -> int:
"""Count tokens for OpenAI models."""
try:
encoding = tiktoken.encoding_for_model(model)
return len(encoding.encode(text))
except KeyError:
# Fallback: approximate 1 token ≈ 4 characters
return len(text) // 4
def calculate_cost(
self,
input_tokens: int,
output_tokens: int,
model: str
) -> float:
"""Calculate cost based on model pricing."""
pricing = {
"gpt-4-turbo": {"input": 0.01, "output": 0.03},
"gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
"claude-3-opus": {"input": 0.015, "output": 0.075},
"claude-3-sonnet": {"input": 0.003, "output": 0.015},
"claude-3-haiku": {"input": 0.00025, "output": 0.00125}
}
rates = pricing.get(model, pricing["gpt-3.5-turbo"])
input_cost = (input_tokens / 1000) * rates["input"]
output_cost = (output_tokens / 1000) * rates["output"]
return input_cost + output_cost
def log_request(
self,
prompt: str,
response: str,
model: str,
latency: float
):
"""Log request for analytics."""
input_tokens = self.count_tokens(prompt, model)
output_tokens = self.count_tokens(response, model)
cost = self.calculate_cost(input_tokens, output_tokens, model)
self.usage_history.append({
"timestamp": time.time(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": cost,
"latency": latency
})
return {
"tokens": {"input": input_tokens, "output": output_tokens},
"cost": cost,
"latency": latency
}
def get_stats(self):
"""Get aggregate statistics."""
if not self.usage_history:
return {}
total_cost = sum(r["cost"] for r in self.usage_history)
total_input = sum(r["input_tokens"] for r in self.usage_history)
total_output = sum(r["output_tokens"] for r in self.usage_history)
avg_latency = sum(r["latency"] for r in self.usage_history) / len(self.usage_history)
return {
"total_requests": len(self.usage_history),
"total_cost": total_cost,
"total_tokens": {"input": total_input, "output": total_output},
"avg_latency": avg_latency,
"cost_per_request": total_cost / len(self.usage_history)
}
# Usage
tracker = CostTracker()
start = time.time()
response = client.complete("Explain AI")
latency = time.time() - start
stats = tracker.log_request(
prompt="Explain AI",
response=response["content"],
model="claude-3-haiku-20240307",
latency=latency
)
print(f"Request cost: ${stats['cost']:.4f}")
print(f"Tokens: {stats['tokens']['input']} in, {stats['tokens']['output']} out")
print(f"Latency: {stats['latency']:.2f}s")
# Get aggregate stats
print(tracker.get_stats())
Security:
Reliability:
Performance:
Observability:
Cost Management:
When helping with LLM integration:
Your role: Help developers integrate LLM APIs reliably, efficiently, and cost-effectively into production applications.