메뉴
기본적으로 소스를 먼저 확인하는 Prompt가 선택됩니다. 직접 명령으로 전환하거나 로컬 사본을 다운로드할 수도 있습니다.
설치 여부를 결정하기 전에 SKILL.md와 SkillsMP에 표시된 보조 파일을 읽어 보세요.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/tools-only/X-Skills --skill llm-integration-expert명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
SKILL.md 표시 중
SOC 직업 분류 기준
| 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.