DEV Community: ZNY

Redis Caching for AI Applications: Reducing Latency and Cost

ZNY — Fri, 15 May 2026 18:56:31 +0000

AI API calls are expensive and slow. Redis caching dramatically reduces both by storing AI responses for reuse. Here's a complete implementation guide.

Why Cache AI Responses?

Without Cache	With Cache
Every request → AI API	Cache hit → Return immediately
1-3s latency per request	< 10ms for cache hits
Full API cost per request	Pay only for cache misses
Rate limit pressure	Rate limit relief

Semantic Caching vs Exact Match

`python
Exact match caching (simple)
cache_key = hash(messages) # Only matches identical prompts

Semantic caching (smart)
cachekey = generateembeddinghash(userprompt) # Matches similar prompts
`

This guide covers exact match caching first, then semantic.

Redis Setup

`python
import redis
import json
from typing import Optional

class AICache:
def init(self, redis_url: str = "redis://localhost:6379/0"):
self.redis = redis.fromurl(redisurl, decode_responses=True)

def makekey(self, messages: list[dict]) -> str:
"""Create cache key from messages."""

Sort to ensure same messages always produce same key

content = json.dumps(messages, sort_keys=True)
import hashlib
return f"ai:response:{hashlib.sha256(content.encode()).hexdigest()}"

def get(self, messages: list[dict]) -> Optional[str]:
"""Get cached response if exists."""
key = self.makekey(messages)
cached = self.redis.get(key)
if cached:

Move to front (LRU-like behavior)

self.redis.lrem("ai:recent", 1, key)
self.redis.rpush("ai:recent", key)
return cached
return None

def set(self, messages: list[dict], response: str, ttl: int = 86400):
"""Cache a response."""
key = self.makekey(messages)
pipe = self.redis.pipeline()
pipe.set(key, response, ex=ttl)
pipe.lpush("ai:recent", key)
pipe.ltrim("ai:recent", 0, 999) # Keep last 1000 keys
pipe.execute()

def invalidate(self, messages: list[dict]):
key = self.makekey(messages)
self.redis.delete(key)
`

Using the Cache with AI Client

`python
class CachedAIClient:
def init(self, api_key: str, cache: AICache):
self.client = AsyncAIClient(api_key)
self.cache = cache
self.cache_ttl = 86400 # 24 hours

async def chat(self, messages: list[dict], use_cache: bool = True) -> str:

Try cache first

if use_cache:
cached = self.cache.get(messages)
if cached:
print("Cache hit!")
return cached

Cache miss - call API

response = await self.client.chat(messages)

Store in cache

if use_cache:
self.cache.set(messages, response, self.cache_ttl)

return response
`

Intelligent Cache Invalidation

`python
class IntelligentCache(AICache):
"""
Cache that handles:

TTL expiration
Manual invalidation
LRU eviction
Cache statistics """

def init(self, redisurl: str, maxsize: int = 10000):
super().init(redis_url)
self.maxsize = maxsize
self.checksize()

def checksize(self):
"""Ensure cache doesn't exceed max size."""
size = self.redis.scard("ai:cache:keys")
if size >= self.max_size:

Remove oldest 10%

toremove = self.maxsize // 10
oldest = self.redis.lrange("ai:cache:order", 0, to_remove - 1)
pipe = self.redis.pipeline()
for key in oldest:
pipe.delete(f"ai:response:{key}")
pipe.delete("ai:cache:order")
pipe.execute()

Re-add remaining in order

remaining = self.redis.lrange("ai:cache:order", to_remove, -1)
for key in reversed(remaining):
self.redis.lpush("ai:cache:order", key)

def get_stats(self) -> dict:
"""Return cache statistics."""
info = self.redis.info("stats")
return {
"total_keys": self.redis.scard("ai:cache:keys"),
"hits": info.get("keyspace_hits", 0),
"misses": info.get("keyspace_misses", 0),
"hit_rate": (
info.get("keyspace_hits", 0) /
max(info.get("keyspacehits", 0) + info.get("keyspacemisses", 1), 1)
) * 100
}
`

Cache Warming Strategy

`python
async def warmcache(client: CachedAIClient, commonprompts: list[str]):
"""
Pre-populate cache with common prompts.
Run at startup or during off-peak hours.
"""
print(f"Warming cache with {len(common_prompts)} prompts...")

for i, prompt in enumerate(common_prompts):
try:
await client.chat(
[{"role": "user", "content": prompt}],
use_cache=True
)
if (i + 1) % 10 == 0:
print(f" Warmed {i + 1}/{len(common_prompts)}")
except Exception as e:
print(f" Failed on prompt {i}: {e}")

print("Cache warming complete!")

Example common prompts
COMMON_PROMPTS = [
"Explain async/await in Python",
"How do I use list comprehensions?",
"What is a context manager?",

... your most common queries

]
`

Production Deployment

`yaml
docker-compose.yml
version: '3.8'
services:
api:
build: .
depends_on:

redis environment:
REDIS_URL=redis://redis:6379/0
OFOXAPIKEY=${OFOXAPIKEY}

redis:
image: redis:7-alpine
volumes:

redis-data:/data command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru

volumes:
redis-data:
`

Cache Hit Rate Monitoring

`python
from prometheus_client import Counter, Gauge

cachehits = Counter('aicache_hits', 'Number of cache hits')
cachemisses = Counter('aicache_misses', 'Number of cache misses')
cachelatency = Gauge('aicachelatencyseconds', 'Cache operation latency')

@app.middleware("http")
async def cachemetricsmiddleware(request: Request, call_next):
if "/chat" in str(request.url):
start = time.time()
response = await call_next(request)
cache_latency.set(time.time() - start)
return response
return await call_next(request)
`

Cost Savings Calculator

`python
def calculate_savings(
requestsperday: int,
cachehitrate: float,
apicostper1ktokens: float = 0.003,
avgtokensper_request: int = 500
):
dailyrequests = requestsper_day
cachehits = int(dailyrequests * cachehitrate)
cachemisses = dailyrequests - cache_hits

Full cost for misses, free for hits

totalcost = (cachemisses avgtokensperrequest / 1000) apicostper1k_tokens

nocachecost = (dailyrequests avgtokensperrequest / 1000) apicostper1ktokens

savings = nocachecost - total_cost

return {
"requests": daily_requests,
"cachehitrate": f"{cachehitrate * 100:.1f}%",
"dailycost": f"${totalcost:.2f}",
"daily_savings": f"${savings:.2f}",
"monthly_savings": f"${savings * 30:.2f}"
}

Example: 10000 requests/day, 70% cache hit rate
print(calculate_savings(10000, 0.70))
{'requests': 10000, 'cachehitrate': '70.0%',
'dailycost': '$4.50', 'dailysavings': '$10.50',
'monthly_savings': '$315.00'}
`

Getting Started

Implement Redis caching for your AI applications with ofox.ai — their reliable API makes caching easy since responses are consistent and deterministic.

👉 Get started with ofox.ai

This article contains affiliate links.

Tags: redis,caching,ai,programming,developer
Canonical URL: https://dev.to/zny10289

Async Python for AI: Building High-Concurrency AI Applications

ZNY — Fri, 15 May 2026 18:51:19 +0000

AI API calls are I/O-bound — you're waiting on network responses. Async Python lets you run many AI requests concurrently, dramatically improving throughput. Here's how to build high-concurrency AI applications.

Why Async for AI?

A single AI API call might take 1-3 seconds. If you process 100 requests sequentially, that's 100-300 seconds. With async concurrency, you can process all 100 in seconds.

`python
import asyncio
import httpx

Sequential (slow)
async def process_sequential(requests):
results = []
for req in requests:
result = await call_ai(req) # 2 seconds each
results.append(result)
return results
Total: len(requests) × 2 seconds

Concurrent (fast)
async def process_concurrent(requests):
tasks = [call_ai(req) for req in requests]
results = await asyncio.gather(*tasks)
return results
Total: ~2 seconds total (all run in parallel)
`

Basic Async AI Client

`python
import asyncio
import httpx
from typing import Optional

class AsyncAIClient:
def init(self, apikey: str, baseurl: str = "https://api.ofox.ai/v1"):
self.apikey = apikey
self.baseurl = baseurl
self._client: Optional[httpx.AsyncClient] = None

async def aenter(self):
self._client = httpx.AsyncClient(
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=120.0
)
return self

async def aexit(self, *args):
if self._client:
await self._client.aclose()

async def chat(self, messages: list[dict], kwargs) -> str:
response = await self._client.post(
f"{self.base_url}/chat/completions",
json={
"model": kwargs.get("model", "claude-3-5-sonnet-20241022"),
"messages": messages,
"maxtokens": kwargs.get("maxtokens", 1024),
"temperature": kwargs.get("temperature", 0.7)
}
)
response.raiseforstatus()
data = response.json()
return data["choices"][0]["message"]["content"]

async def chat_stream(self, messages: list[dict], kwargs):
"""Streaming version for real-time output."""
async with self._client.stream(
"POST",
f"{self.base_url}/chat/completions",
json={
"model": kwargs.get("model", "claude-3-5-sonnet-20241022"),
"messages": messages,
"stream": True,
"maxtokens": kwargs.get("maxtokens", 1024)
}
) as response:
async for line in response.aiter_lines():
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
break
yield data

Usage
async def main():
async with AsyncAIClient("your-api-key") as client:
response = await client.chat([
{"role": "user", "content": "Hello, Claude!"}
])
print(response)
`

Concurrent Batch Processing

`python
import asyncio
from dataclasses import dataclass
from typing import List

@dataclass
class BatchItem:
id: str
prompt: str
metadata: dict = None

@dataclass
class BatchResult:
id: str
success: bool
response: str = ""
error: str = ""

async def process_batch(
items: List[BatchItem],
client: AsyncAIClient,
max_concurrency: int = 10
) -> List[BatchResult]:
"""
Process items in batches with controlled concurrency.
"""
semaphore = asyncio.Semaphore(max_concurrency)

async def process_one(item: BatchItem) -> BatchResult:
async with semaphore:
try:
response = await client.chat([
{"role": "user", "content": item.prompt}
])
return BatchResult(id=item.id, success=True, response=response)
except Exception as e:
return BatchResult(id=item.id, success=False, error=str(e))

tasks = [process_one(item) for item in items]
results = await asyncio.gather(*tasks)
return list(results)

Usage
async def main():
items = [
BatchItem(id=f"req-{i}", prompt=f"Process item {i}")
for i in range(100)
]

async with AsyncAIClient("your-api-key") as client:
results = await processbatch(items, client, maxconcurrency=20)

successes = [r for r in results if r.success]
failures = [r for r in results if not r.success]

print(f"Completed: {len(successes)}/{len(items)}")
`

Rate-Limited Concurrency

`python
import asyncio
import time

class RateLimiter:
"""Token bucket rate limiter for API calls."""

def init(self, callsperminute: int = 60):
self.callsperminute = callsperminute
self.interval = 60.0 / callsperminute
self.last_call = 0.0
self._lock = asyncio.Lock()

async def acquire(self):
async with self._lock:
now = time.time()
waittime = self.lastcall + self.interval - now
if wait_time > 0:
await asyncio.sleep(wait_time)
self.last_call = time.time()

async def ratelimitedprocessing(items: List[BatchItem], client: AsyncAIClient):
limiter = RateLimiter(callsperminute=60) # 60 RPM

async def process(item: BatchItem) -> BatchResult:
await limiter.acquire() # Wait if needed
try:
response = await client.chat([{"role": "user", "content": item.prompt}])
return BatchResult(id=item.id, success=True, response=response)
except Exception as e:
return BatchResult(id=item.id, success=False, error=str(e))

tasks = [process(item) for item in items]
return await asyncio.gather(*tasks)
`

Retry with Exponential Backoff

`python
import asyncio
from typing import TypeVar

T = TypeVar('T')

async def retrywithbackoff(
fn,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0
) -> T:
for attempt in range(max_retries):
try:
return await fn()
except Exception as e:
if attempt == max_retries - 1:
raise

delay = min(basedelay (2 * attempt), maxdelay)

Check if error is retryable

if hasattr(e, 'response') and e.response:
status = e.response.status_code
if status not in (429, 500, 502, 503, 504):
raise # Don't retry client errors

print(f"Attempt {attempt + 1} failed, retrying in {delay}s: {e}")
await asyncio.sleep(delay)

raise RuntimeError("Unreachable")

Usage
async def robust_chat(client: AsyncAIClient, messages: list[dict]) -> str:
async def call():
return await client.chat(messages)

return await retrywithbackoff(call, max_retries=3)
`

Building a Production AI Queue

`python
import asyncio
from collections import deque
from dataclasses import dataclass
import uuid

@dataclass
class AIJob:
id: str
messages: list[dict]
future: asyncio.Future

class AsyncAIQueue:
def init(self, client: AsyncAIClient, workers: int = 5):
self.client = client
self.workers = workers
self.queue: asyncio.Queue[AIJob] = asyncio.Queue()
self.results: dict[str, str] = {}

async def worker(self):
while True:
job = await self.queue.get()
try:
response = await retrywithbackoff(
lambda: self.client.chat(job.messages)
)
self.results[job.id] = response
job.future.set_result(response)
except Exception as e:
job.future.set_exception(e)
finally:
self.queue.task_done()

async def start(self):
self.worker_tasks = [
asyncio.create_task(self.worker())
for _ in range(self.workers)
]

async def submit(self, messages: list[dict]) -> str:
future = asyncio.geteventloop().create_future()
job = AIJob(id=str(uuid.uuid4()), messages=messages, future=future)
await self.queue.put(job)
return job.id

async def getresult(self, jobid: str, timeout: float = 120.0) -> str:

Poll for result

start = time.time()
while time.time() - start < timeout:
if job_id in self.results:
return self.results.pop(job_id)
await asyncio.sleep(0.1)
raise TimeoutError(f"Job {job_id} timed out")
`

Getting Started

Build high-concurrency AI applications with ofox.ai — their reliable API supports async patterns with competitive pricing for production workloads.

👉 Get started with ofox.ai

This article contains affiliate links.

Tags: python,async,ai,programming,developer
Canonical URL: https://dev.to/zny10289

API Security Best Practices for AI Applications in 2026

ZNY — Fri, 15 May 2026 18:46:07 +0000

AI applications face unique security challenges. Beyond traditional API vulnerabilities, AI APIs expose new attack surfaces: prompt injection, data leakage, and model manipulation. Here's how to secure your AI-powered systems.

The AI Security Landscape

AI APIs introduce attack vectors traditional APIs don't have:

Prompt injection — Malicious input that manipulates AI behavior
Data exfiltration — AI accidentally leaking sensitive context
Token exhaustion — attackers exhausting your quota
Model extraction — Repeated queries to reverse-engineer the model
Context poisoning — Injecting malicious context into conversations

Input Validation and Sanitization

`python
import re
from typing import Optional

class InputSanitizer:

Block common prompt injection patterns

BLOCKED_PATTERNS = [
r'ignore\s+previous\s+instructions',
r'ignore\s+all\s+previous',
r'system\s:\s',
r'you\s+are\s+a\s+different',
r'forget\s+everything',
r'#\s*roleplay',
]

MAX_LENGTH = 10000 # Max 10k characters
MAXTOKENSESTIMATE = MAX_LENGTH // 4 # ~2500 tokens

@classmethod
def sanitize(cls, user_input: str) -> tuple[bool, Optional[str], str]:
"""
Returns: (issafe, reason, sanitizedinput)
"""

Check length

if len(userinput) > cls.MAXLENGTH:
return False, f"Input exceeds {cls.MAXLENGTH} chars", userinput[:cls.MAX_LENGTH]

Check for blocked patterns

for pattern in cls.BLOCKED_PATTERNS:
if re.search(pattern, user_input, re.IGNORECASE):
return False, "Blocked pattern detected", ""

Strip control characters

sanitized = re.sub(r'[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f]', '', user_input)

return True, None, sanitized
`

Rate Limiting

`python
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import time
from collections import defaultdict

app = FastAPI()

Simple in-memory rate limiter
class RateLimiter:
def init(self, requestsperminute: int = 60):
self.requestsperminute = requestsperminute
self.requests = defaultdict(list)

def isallowed(self, clientid: str) -> bool:
now = time.time()
minute_ago = now - 60

Clean old entries

self.requests[client_id] = [
t for t in self.requests[clientid] if t > minuteago
]

if len(self.requests[clientid]) >= self.requestsper_minute:
return False

self.requests[client_id].append(now)
return True

ratelimiter = RateLimiter(requestsper_minute=60)

@app.middleware("http")
async def ratelimitmiddleware(request: Request, call_next):
client_id = request.client.host # Or use API key
if not ratelimiter.isallowed(client_id):
return JSONResponse(
status_code=429,
content={"error": "Rate limit exceeded"}
)
response = await call_next(request)
return response
`

API Key Security

`python
import os
import hashlib
import hmac
from typing import Optional

class APIKeyManager:
"""
Never store raw API keys. Always hash them.
"""
def init(self):
self.keystore = {} # In production, use a proper database

def createkey(self, userid: str, scopes: list[str]) -> str:
import secrets
apikey = f"ofox{secrets.token_urlsafe(32)}"
keyhash = self.hashkey(apikey)

self.keystore[key_hash] = {
"userid": userid,
"scopes": scopes,
"created": time.time()
}

Return raw key ONLY ONCE to the user

return api_key

def validatekey(self, apikey: str) -> Optional[dict]:
keyhash = self.hashkey(apikey)
return self.keystore.get(key_hash)

def hashkey(self, key: str) -> str:
return hashlib.sha256(key.encode()).hexdigest()

Usage
key_manager = APIKeyManager()
rawkey = keymanager.create_key("user123", ["chat", "embeddings"])
print(f"Save this key securely: {raw_key}") # Show once
`

Prompt Injection Defense

`python
class PromptInjectionDetector:
"""
Detect attempts to override system behavior through user input.
"""
INJECTION_SIGNALS = [
"ignore previous",
"disregard your",
"new instructions:",
"[INST]",
"<>",
"<>",
"you are now",
"pretend you are",
"forget your",
"system prompt:",
]

@classmethod
def detect(cls, user_input: str) -> bool:
lowerinput = userinput.lower()
for signal in cls.INJECTION_SIGNALS:
if signal.lower() in lower_input:
return True
return False

Usage in your endpoint
@app.post("/chat")
async def chat(request: ChatRequest):
if PromptInjectionDetector.detect(request.messages[-1].content):

Log and block

logger.warning(f"Prompt injection attempt: {request.messages[-1].content[:100]}")
raise HTTPException(status_code=400, detail="Invalid input")

Continue with normal processing

Data Isolation in Multi-Tenant Systems

`python
class ConversationContext:
"""
Ensure user data doesn't leak between conversations.
"""
def init(self, userid: str, apikey: str):
self.userid = userid
self.apikey = apikey
self.conversationhistory = []

def add_message(self, role: str, content: str):
self.conversationhistory.append({
"role": role,
"content": content,
"userid": self.userid # Tag with user
})

def get_messages(self) -> list[dict]:

Always filter by user_id to prevent leakage

return [
m for m in self.conversationhistory
if m["userid"] == self.userid
]

def clear_history(self):

Only clear THIS user's history

self.conversationhistory = [
m for m in self.conversationhistory
if m["userid"] != self.userid
]
`

Secure Error Handling

`python
@app.exception_handler(Exception)
async def globalexceptionhandler(request: Request, exc: Exception):

Never expose internal error details in production

logger.error(f"Error: {exc}", exc_info=True)

return JSONResponse(
status_code=500,
content={
"error": "Internal server error",

Don't include: exc.message, stack trace, API keys

}
)

For API provider errors (like ofox.ai errors)
@app.post("/chat")
async def chat(request: ChatRequest):
try:
result = await callofoxapi(request)
return result
except httpx.HTTPStatusError as e:

Log full error internally

logger.error(f"ofox API error: {e.response.status_code} {e.response.text}")

Return sanitized error to client

raise HTTPException(
status_code=502,
detail="AI service temporarily unavailable"
)
`

Environment Variables (Never Hardcode)

`bash
.env (never commit this file)
OFOXAPIKEY=your-key-here
DATABASE_URL=postgresql://...
JWT_SECRET=your-secret-here

docker-compose.yml (use secrets in production)
environment:

OFOXAPIKEY=${OFOXAPIKEY} `

`python
Load from environment
from dotenv import load_dotenv
load_dotenv() # In development only

apikey = os.environ.get("OFOXAPI_KEY")
if not api_key:
raise ValueError("OFOXAPIKEY not set")
`

Security Checklist

[ ] Input validation on all user-provided text
[ ] Rate limiting on all endpoints
[ ] API keys hashed (never stored raw)
[ ] Prompt injection detection
[ ] Error messages don't expose internals
[ ] Environment variables for secrets (not hardcoded)
[ ] HTTPS only in production
[ ] Logging without sensitive data
[ ] Regular dependency audits (pip audit, npm audit)

Getting Started

Build secure AI applications with ofox.ai — their API includes built-in security features and 99.9% uptime guarantee.

👉 Get started with ofox.ai

This article contains affiliate links.

Tags: security,api,ai,programming,developer
Canonical URL: https://dev.to/zny10289

Docker for AI Development: Containerizing LLM Applications

ZNY — Fri, 15 May 2026 18:40:55 +0000

Docker simplifies AI application deployment by providing consistent environments from development to production. Here's how to containerize your AI applications powered by Claude and ofox.ai.

Why Docker for AI Apps?

Reproducible environments — Same behavior locally and in production
Dependency isolation — Python packages, system libraries, CUDA versions
Easy deployment — Ship to any cloud with Docker
Resource control — Limit CPU/memory per container

Basic Dockerfile for AI App

`dockerfile
Dockerfile
FROM python:3.11-slim

WORKDIR /app

Install system dependencies
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*

Copy requirements first (for caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

Copy application
COPY . .

Set environment variables
ENV PYTHONUNBUFFERED=1

Run the application
CMD ["python", "main.py"]
`

dockerfile requirements.txt fastapi==0.109.0 uvicorn==0.27.0 httpx==0.26.0 python-dotenv==1.0.0

Docker Compose for AI Services

`yaml
docker-compose.yml
version: '3.8'

services:
api:
build: .
ports:

"8000:8000" environment:
OFOXAPIKEY=${OFOXAPIKEY}
MODEL=claude-3-5-sonnet-20241022 restart: unless-stopped healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8000/health"] interval: 30s timeout: 10s retries: 3

Optional: Add a Redis cache

redis:
image: redis:7-alpine
ports:

"6379:6379" volumes:
redis-data:/data

volumes:
redis-data:
`

Production-Ready FastAPI + ofox.ai

`python
main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import httpx
import os

app = FastAPI(title="Claude API Service", version="1.0.0")

class Message(BaseModel):
role: str
content: str

class ChatRequest(BaseModel):
messages: List[Message]
model: str = "claude-3-5-sonnet-20241022"
max_tokens: Optional[int] = 1024
temperature: Optional[float] = 0.7

@app.get("/health")
async def health():
return {"status": "healthy"}

@app.post("/chat")
async def chat(request: ChatRequest):
async with httpx.AsyncClient(timeout=120.0) as client:
try:
response = await client.post(
"https://api.ofox.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {os.environ['OFOXAPIKEY']}",
"Content-Type": "application/json"
},
json={
"model": request.model,
"messages": [m.model_dump() for m in request.messages],
"maxtokens": request.maxtokens,
"temperature": request.temperature
}
)
response.raiseforstatus()
data = response.json()
return {
"content": data["choices"][0]["message"]["content"],
"model": data["model"],
"tokens": data["usage"]["total_tokens"]
}
except httpx.HTTPStatusError as e:
raise HTTPException(statuscode=e.response.statuscode, detail=str(e))
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
`

GPU Support for Local Models

`dockerfile
Dockerfile with GPU support
FROM nvidia/cuda:12.1.0-base-ubuntu22.04

RUN apt-get update && apt-get install -y \
python3.11 python3.11-venv python3-pip \
&& rm -rf /var/lib/apt/lists/*

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

For running local models like Ollama
RUN curl -fsSL https://ollama.ai/install.sh | sh

COPY . .
CMD ["python3", "main.py"]
`

`yaml
docker-compose.yml with GPU
services:
api:
build: .
deploy:
resources:
reservations:
devices:

driver: nvidia count: 1 capabilities: [gpu] `

Multi-Stage Build (Smaller Images)

`dockerfile
Build stage
FROM python:3.11-slim as builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

Production stage
FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY . .
ENV PATH=/root/.local/bin:$PATH
CMD ["python", "main.py"]
`

Environment-Based Configuration

`python
config.py
import os
from dataclasses import dataclass

@dataclass
class Config:
api_key: str
model: str
max_tokens: int
temperature: float

def get_config() -> Config:
return Config(
apikey=os.environ["OFOXAPI_KEY"],
model=os.environ.get("MODEL", "claude-3-5-sonnet-20241022"),
maxtokens=int(os.environ.get("MAXTOKENS", "1024")),
temperature=float(os.environ.get("TEMPERATURE", "0.7"))
)
`

Building and Running

`bash
Build
docker build -t claude-api-service .

Run
docker run -d -p 8000:8000 \
-e OFOXAPIKEY=your-key-here \
--name claude-api \
claude-api-service

With Docker Compose
docker-compose up -d

View logs
docker logs -f claude-api

Shell into container
docker exec -it claude-api /bin/bash
`

CI/CD with GitHub Actions

`yaml
name: Build and Deploy

on:
push:
branches: [main]

jobs:
build:
runs-on: ubuntu-latest
steps:

uses: actions/checkout@v4
name: Build image
run: docker build -t claude-api:${{ github.sha }} .
name: Run tests
run: |
docker run claude-api:${{ github.sha }} pytest
name: Push to registry
run: |
docker tag claude-api:${{ github.sha }} registry/app/claude-api:latest
docker push registry/app/claude-api:latest
`

Deploy Anywhere

With Docker, your AI application deploys to:
AWS ECS — Managed container service
Google Cloud Run — Serverless containers
Azure Container Instances — Simple deployment
DigitalOcean App Platform — Simple PaaS
Your own server — With docker-compose

Getting Started

Containerize your AI applications and deploy with confidence. Power them with ofox.ai — reliable Claude API with competitive pricing and 99.9% uptime.

👉 Get started with ofox.ai

This article contains affiliate links.

Tags: docker,devops,ai,programming,developer
Canonical URL: https://dev.to/zny10289

Testing AI-Powered Applications: Strategies for LLM Integration

ZNY — Fri, 15 May 2026 18:35:44 +0000

Testing AI applications is fundamentally different from testing traditional software. There's no deterministic output, prompts change behavior, and edge cases multiply. Here's how to build a robust testing strategy for AI-powered applications.

The AI Testing Challenge

Traditional testing:
Input → Function → Expected Output

AI testing:
Input → Prompt + Context → Probabilistic Output

You can't assert exact outputs. Instead, you test properties.

Property-Based Testing for AI

`typescript
// Instead of testing exact output, test properties

interface TestCase {
input: string;
constraints: Constraint[];
}

interface Constraint {
type: 'contains' | 'excludes' | 'length' | 'format' | 'json';
value: string | number | RegExp;
}

async function testAIOutput(testCase: TestCase, actualOutput: string): Promise {
for (const constraint of testCase.constraints) {
switch (constraint.type) {
case 'contains':
if (!actualOutput.includes(constraint.value as string)) return false;
break;
case 'excludes':
if (actualOutput.includes(constraint.value as string)) return false;
break;
case 'length':
if (actualOutput.length > (constraint.value as number)) return false;
break;
case 'json':
try {
JSON.parse(actualOutput);
} catch {
return false;
}
break;
}
}
return true;
}

// Example test
const testCase: TestCase = {
input: 'Extract the name and email from: John Doe, john@example.com',
constraints: [
{ type: 'contains', value: 'John' },
{ type: 'contains', value: 'john@example.com' },
{ type: 'excludes', value: 'undefined' },
{ type: 'length', value: 100 }
]
};
`

Prompt Versioning and Regression Testing

`python
import hashlib
from datetime import datetime

class PromptRegistry:
def init(self):
self.prompts = {}

def register(self, name: str, version: str, prompt: str, test_cases: list):
key = f"{name}:{version}"
self.prompts[key] = {
'prompt': prompt,
'testcases': testcases,
'hash': hashlib.md5(prompt.encode()).hexdigest(),
'registered': datetime.now()
}

def get_prompt(self, name: str, version: str) -> str:
return self.prompts[f"{name}:{version}"]['prompt']

def regressiontest(self, name: str, newversion: str,
llm_client, threshold: float = 0.8) -> bool:
"""Ensure new version passes existing test cases."""
old_prompt = self.prompts.get(f"{name}:{version}")
if not old_prompt:
return True

old_passes = 0
new_passes = 0

for tc in oldprompt['testcases']:
oldresult = await llmclient.complete(old_prompt['prompt'] + tc['input'])
newresult = await llmclient.complete(
self.getprompt(name, newversion) + tc['input']
)

oldok = await testAIOutput(tc, oldresult)
newok = await testAIOutput(tc, newresult)

if oldok: oldpasses += 1
if newok: newpasses += 1

New version should pass at least as many tests

return (newpasses / len(oldprompt['test_cases'])) >= threshold
`

Deterministic Output Testing

For structured outputs, test deterministically:

`typescript
import { z } from 'zod';

const CodeReviewSchema = z.object({
score: z.number().min(0).max(10),
issues: z.array(z.object({
severity: z.enum(['low', 'medium', 'high']),
line: z.number(),
description: z.string()
})),
summary: z.string()
});

async function testCodeReview(code: string, expectedScoreRange: [number, number]) {
const response = await llm.complete(
Review this code and return JSON: ${code}
);

// Parse and validate
const parsed = JSON.parse(response);
const validated = CodeReviewSchema.parse(parsed);

// Deterministic assertions
console.assert(
validated.score >= expectedScoreRange[0] &&
validated.score <= expectedScoreRange[1],
Score ${validated.score} outside expected range
);

console.assert(
validated.issues.length < 20,
'Too many issues reported'
);

return validated;
}
`

Mocking External AI Calls

`typescript
// For unit tests, mock the LLM client
class MockLLMClient {
constructor(private fixtures: Map) {}

async complete(prompt: string): Promise {
// Return fixture matching prompt pattern
for (const [pattern, response] of this.fixtures) {
if (prompt.includes(pattern)) {
return response;
}
}
return 'Mock response';
}

async *stream(prompt: string): AsyncGenerator {
const response = await this.complete(prompt);
for (const char of response) {
yield char;
}
}
}

// Usage in tests
const mockClient = new MockLLMClient(new Map([
['extract email', '{"email": "test@example.com"}'],
['summarize', 'This is a summary of the text.']
]));

// Now your business logic tests run fast and deterministically
`

Chaos Testing for AI Applications

`python
class AIChaosTests:
def testratelimits(self, client):
"""Does your app handle rate limits gracefully?"""
for _ in range(100):
try:
client.complete("test")
except RateLimitError:
assert client.retry_count > 0
break
else:
pytest.fail("Rate limit not encountered after 100 requests")

def testinvalidjson(self, client):
"""Does your app handle malformed JSON from LLM?"""

Inject bad response

client.mock_response('{"broken": }')
result = safeparsejson(client.complete("test"))
assert result is not None # Handled gracefully

def testemptycontext(self, client):
"""Does your app handle empty context?"""
result = client.complete("")
assert result is not None

def testmaxtokens_respected(self, client):
"""Does max_tokens actually limit output?"""
result = client.complete("test", max_tokens=10)
assert len(result) <= 50 # ~10 tokens
`

Integration Test Framework

`typescript
describe('AI Integration Tests', () => {
const client = new ClaudeClient(process.env.OFOXAPIKEY);

describe('Code Review Feature', () => {
it('identifies syntax errors', async () => {
const code = 'const x = ;';
const review = await reviewCode(client, code);
expect(review.issues.some(i => i.severity === 'high')).toBe(true);
});

it('handles valid code gracefully', async () => {
const code = 'const x = 42;';
const review = await reviewCode(client, code);
expect(review.issues.filter(i => i.severity === 'high')).toHaveLength(0);
});

it('respects max issues limit', async () => {
const code = '...'; // Large code
const review = await reviewCode(client, code, { maxIssues: 10 });
expect(review.issues.length).toBeLessThanOrEqual(10);
});
});
});
`

Building Testable AI Systems

Separate concerns — Keep prompts in config, not buried in code
Structured outputs — Use Zod/JSON Schema to constrain responses
Fallback handling — Plan for API failures at every call site
Snapshot testing — Store expected responses for regression

Getting Started

Build testable AI applications with ofox.ai — their API is reliable and consistent, making it easier to build deterministic test suites.

👉 Get started with ofox.ai

This article contains affiliate links.

Tags: testing,ai,programming,developer,quality
Canonical URL: https://dev.to/zny10289

LLM Context Window Management: Techniques for Handling Long Documents

ZNY — Fri, 15 May 2026 18:30:32 +0000

Every LLM has a context window limit — a maximum number of tokens you can pass in a single request. Claude 3.5 Sonnet offers 200K tokens, but that's still finite. Here's how to manage context efficiently for production AI applications.

Understanding Context Window Limits

Model	Context Window	Approximate Pages
Claude 3.5 Sonnet	200K tokens	~500 pages
GPT-4 Turbo	128K tokens	~300 pages
Claude 3 Opus	200K tokens	~500 pages

When you exceed the limit, you get an error. When you're close, you're wasting money on tokens that add no value.

Token Estimation

`python
import re

def estimate_tokens(text: str) -> int:
"""
Rough token estimation.
~4 characters per token for English text.
"""
return len(text) // 4

def estimatetokensprecise(text: str) -> int:
"""
More precise estimation using word count.
Average English word is ~1.3 tokens.
"""
words = len(re.findall(r'\w+', text))
return int(words * 1.3)
`

Technique 1: Semantic Chunking

Split documents by meaning, not by character count:

`python
import re

def semanticchunk(text: str, maxtokens: int = 4000, overlap: int = 200) -> list[str]:
"""
Split text into semantic chunks (paragraphs).
"""

Split by double newlines (paragraphs)

paragraphs = re.split(r'\n\n+', text)

chunks = []
current_chunk = []
current_tokens = 0

for para in paragraphs:
paratokens = estimatetokens(para)

if currenttokens + paratokens > max_tokens:

Save current chunk

if current_chunk:
chunks.append('\n\n'.join(current_chunk))

Start new chunk with overlap

overlap_paras = []
overlap_tokens = 0
for p in reversed(current_chunk):
t = estimate_tokens(p)
if overlap_tokens + t <= overlap:
overlap_paras.insert(0, p)
overlap_tokens += t
else:
break

currentchunk = overlapparas + [para]
currenttokens = overlaptokens + para_tokens
else:
current_chunk.append(para)
currenttokens += paratokens

if current_chunk:
chunks.append('\n\n'.join(current_chunk))

return chunks
`

Technique 2: RAG — Retrieval-Augmented Generation

Don't put everything in the prompt. Retrieve only what's relevant:

`python
class SimpleRAG:
def init(self, documents: list[str], chunk_size: int = 1000):
self.chunks = self.createchunks(documents, chunk_size)
self.embeddings = self.createembeddings(self.chunks)

def retrieve(self, query: str, top_k: int = 3) -> list[str]:
"""Find most relevant chunks for query."""
queryembedding = self.embed(query)

scores = [
self.cosinesimilarity(query_embedding, e)
for e in self.embeddings
]

top_indices = sorted(range(len(scores)),
key=lambda i: scores[i],
reverse=True)[:top_k]

return [self.chunks[i] for i in top_indices]

def createchunks(self, documents: list[str], chunk_size: int) -> list[str]:
chunks = []
for doc in documents:
chunks.extend(semanticchunk(doc, maxtokens=chunk_size))
return chunks

def _embed(self, text: str) -> list[float]:

In production, use OpenAI or ofox.ai embeddings

pass
`

Technique 3: Conversation Summary

Summarize older messages to preserve context:

`python
class SummarizingConversation:
def init(self, max_tokens: int = 16000):
self.maxtokens = maxtokens
self.messages = []
self.summary = ""

def add_message(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
self.maybesummarize()

def maybesummarize(self):
totaltokens = sum(estimatetokens(m["content"]) for m in self.messages)

if totaltokens > self.maxtokens:

Summarize older messages

older_messages = self.messages[:-5] # Keep last 5
recent = self.messages[-5:]

summary_prompt = f"""
Summarize this conversation concisely, preserving key information:

{chr(10).join(f'{m[\"role\"]}: {m[\"content\"]}' for m in older_messages)}
"""

Call LLM to summarize (pseudocode)

self.summary = callllmsummarize(summary_prompt)
self.messages = [{"role": "system", "content": f"Prior context: {self.summary}"}] + recent

def get_messages(self) -> list[dict]:
return self.messages
`

Technique 4: System Prompt Optimization

Keep system prompts lean:

`python
❌ Verbose system prompt (wastes tokens)
verbose_system = """
You are a helpful AI assistant. You are designed to be respectful,
professional, and helpful. You should provide accurate information
and be honest when you don't know something. You should ...
[200 more words]
"""

✅ Lean system prompt (effective)
lean_system = """
Role: helpful AI assistant
Goal: provide accurate, concise answers
When unsure: say "I don't know"
"""
`

Technique 5: Streaming with Token Tracking

`python
class StreamingTokenTracker:
def init(self, model: str = "claude-3-5-sonnet-20241022"):
self.model = model
self.totalinputtokens = 0
self.totaloutputtokens = 0

async def stream_chat(self, messages: list[dict]) -> str:
"""Stream response while tracking token usage."""
response = await fetch('https://api.ofox.ai/v1/chat/completions', {
method: 'POST',
headers: {
'Authorization': f'Bearer {API_KEY}',
'Content-Type': 'application/json'
},
body: json.dumps({
'model': self.model,
'messages': messages,
'stream': True
})
})

reader = response.body.getReader()
decoder = TextDecoder()
full_response = []

while True:
chunk = await reader.read()
if chunk.done: break

data = decoder.decode(chunk.value)
for line in data.split('\n'):
if line.startswith('data: '):
delta = json.loads(line[6:]).get('choices', [{}])[0].get('delta', {})
if content := delta.get('content'):
self.totaloutputtokens += estimate_tokens(content)
yield content

Track input tokens

self.totalinputtokens = sum(
estimate_tokens(m['content']) for m in messages
)

def getcost(self, inputcostper1k=0.003, outputcostper_1k=0.015):
inputcost = (self.totalinputtokens / 1000) * inputcostper1k
outputcost = (self.totaloutputtokens / 1000) * outputcostper1k
return inputcost + outputcost
`

Practical Rule of Thumb

Keep your prompt at < 50% of context window. This leaves room for: User input variations Model reasoning Unexpected response length

Getting Started

Build token-efficient AI applications with ofox.ai — their OpenAI-compatible API gives you access to Claude with generous context windows at competitive pricing.

👉 Get started with ofox.ai

This article contains affiliate links.

Tags: llm,artificial-intelligence,programming,developer,performance
Canonical URL: https://dev.to/zny10289

Building React AI Chat Components in 2026: Complete Guide

ZNY — Fri, 15 May 2026 18:25:20 +0000

Building a production-ready AI chat interface in React requires more than just displaying messages. You need streaming responses, markdown rendering, code highlighting, error handling, and a polished UX. Here's the complete implementation.

Core Chat Component

`typescript
// ChatWindow.tsx
import React, { useState, useRef, useEffect } from 'react';
import ReactMarkdown from 'react-markdown';
import { Prism as SyntaxHighlighter } from 'react-syntax-highlighter';

interface Message {
id: string;
role: 'user' | 'assistant';
content: string;
timestamp: Date;
}

interface ChatWindowProps {
apiKey: string;
model?: string;
}

export function ChatWindow({ apiKey, model = 'claude-3-5-sonnet-20241022' }: ChatWindowProps) {
const [messages, setMessages] = useState([]);
const [input, setInput] = useState('');
const [isLoading, setIsLoading] = useState(false);
const [error, setError] = useState(null);
const messagesEndRef = useRef(null);

const scrollToBottom = () => {
messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
};

useEffect(() => {
scrollToBottom();
}, [messages, isLoading]);

const sendMessage = async () => {
if (!input.trim() || isLoading) return;

const userMessage: Message = {
id: crypto.randomUUID(),
role: 'user',
content: input.trim(),
timestamp: new Date()
};

setMessages(prev => [...prev, userMessage]);
setInput('');
setIsLoading(true);
setError(null);

try {
const response = await fetch('https://api.ofox.ai/v1/chat/completions', {
method: 'POST',
headers: {
'Authorization': Bearer ${apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify({
model,
messages: [...messages, userMessage].map(m => ({
role: m.role,
content: m.content
})),
stream: true
})
});

if (!response.ok) {
throw new Error(API error: ${response.status});
}

// Handle streaming response
const reader = response.body?.getReader();
const decoder = new TextDecoder();
let assistantContent = '';

const assistantMessage: Message = {
id: crypto.randomUUID(),
role: 'assistant',
content: '',
timestamp: new Date()
};

setMessages(prev => [...prev, assistantMessage]);

while (reader) {
const { done, value } = await reader.read();
if (done) break;

const chunk = decoder.decode(value);
const lines = chunk.split('\n');

for (const line of lines) {
if (line.startsWith('data: ')) {
const data = JSON.parse(line.slice(6));
if (data.choices[0].delta.content) {
assistantContent += data.choices[0].delta.content;
setMessages(prev => prev.map(m =>
m.id === assistantMessage.id
? { ...m, content: assistantContent }
: m
));
}
}
}
}
} catch (err) {
setError(err instanceof Error ? err.message : 'Unknown error');
setMessages(prev => prev.filter(m => m.id !== userMessage.id));
} finally {
setIsLoading(false);
}
};

return (

{messages.map(msg => (

{msg.role === 'user' ? 'You' : 'Claude'}

{String(children).replace(/\n$/, '')}

) : (
{children}
);
}
}}
>
{msg.content}

))}
{isLoading && ( Claude ...

)}
{error &&

{error}} setInput(e.target.value)} onKeyDown={e => { if (e.key === 'Enter' && !e.shiftKey) { e.preventDefault(); sendMessage(); } }} placeholder="Ask Claude..." disabled={isLoading} /> Send
);
}
`

Streaming vs Non-Streaming

`typescript
// Non-streaming (simpler, good for short responses)
async function chat(apiKey: string, messages: Message[]) {
const response = await fetch('https://api.ofox.ai/v1/chat/completions', {
method: 'POST',
headers: {
'Authorization': Bearer ${apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify({
model: 'claude-3-5-sonnet-20241022',
messages: messages.map(m => ({ role: m.role, content: m.content }))
})
});

const data = await response.json();
return data.choices[0].message.content;
}

// Streaming (better UX for long responses)
async function* chatStream(apiKey: string, messages: Message[]) {
const response = await fetch('https://api.ofox.ai/v1/chat/completions', {
method: 'POST',
headers: {
'Authorization': Bearer ${apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify({
model: 'claude-3-5-sonnet-20241022',
messages: messages.map(m => ({ role: m.role, content: m.content })),
stream: true
})
});

const reader = response.body?.getReader();
const decoder = new TextDecoder();

while (reader) {
const { done, value } = await reader.read();
if (done) break;

Building with Modern React Patterns

`typescript
// useChat hook
function useChat(initialMessages: Message[] = []) {
const [messages, setMessages] = useState(initialMessages);
const [isLoading, setIsLoading] = useState(false);

const send = async (content: string) => {
const userMessage: Message = {
id: crypto.randomUUID(),
role: 'user',
content,
timestamp: new Date()
};

setMessages(prev => [...prev, userMessage]);
setIsLoading(true);

// ... streaming logic

setIsLoading(false);
};

const clear = () => setMessages([]);

return { messages, send, clear, isLoading };
}

// Usage
function App() {
const { messages, send, isLoading } = useChat();
return ;
}
`

Getting Started

Power your React chat app with ofox.ai — OpenAI-compatible API with Claude models. Sign up and get an API key to start building.

👉 Get started with ofox.ai

This article contains affiliate links.

Tags: react,javascript,ai,programming,webdev
Canonical URL: https://dev.to/zny10289

GitHub Actions + AI: Automating Code Quality with Claude

ZNY — Fri, 15 May 2026 18:20:09 +0000

Continuous integration with AI-powered code review catches bugs before they reach production. Here's how to build a GitHub Actions workflow that runs Claude-powered analysis on every pull request.

Why AI Code Review in CI?

Traditional CI catches syntax errors and test failures. AI code review catches:
Logic bugs
Security vulnerabilities
Performance issues
Code quality problems
Documentation gaps

The GitHub Actions Workflow

`yaml
.github/workflows/ai-review.yml
name: AI Code Review

on:
pull_request:
types: [opened, synchronize, reopened]

jobs:
ai-review:
runs-on: ubuntu-latest
permissions:
pull-requests: write
contents: read

steps:

name: Checkout code
uses: actions/checkout@v4
with:
fetch-depth: 0
name: Get PR diff
id: diff
run: |
git diff origin/${{ github.baseref }}...HEAD > prdiff.txt
echo "diffsize=$(wc -c < prdiff.txt)" >> $GITHUB_OUTPUT
echo "fileschanged=$(git diff --name-only origin/${{ github.baseref }}...HEAD | wc -l)" >> $GITHUB_OUTPUT
name: Run AI Code Review
if: steps.diff.outputs.diff_size < 50000 # Skip if too large
env:
OFOXAPIKEY: ${{ secrets.OFOXAPIKEY }}
run: |

Get PR context

PRNUMBER=$(echo ${{ github.event.pullrequest.number }})
REPO=${{ github.repository }}

Prepare review prompt

DIFF=$(cat pr_diff.txt)

Call Claude via ofox.ai

RESPONSE=$(curl -s -X POST https://api.ofox.ai/v1/chat/completions \
-H "Authorization: Bearer $OFOXAPIKEY" \
-H "Content-Type: application/json" \
-d '{
"model": "claude-3-5-sonnet-20241022",
"messages": [{
"role": "user",
"content": "You are an expert code reviewer. Review this PR diff and provide feedback on bugs, security issues, performance problems, and code quality. Be concise but thorough.\n\n'${DIFF}'"
}],
"max_tokens": 2000,
"temperature": 0.3
}')

echo "$RESPONSE" | jq -r '.choices[0].message.content' > review_comment.md
echo "REVIEWOUTPUT=$(cat reviewcomment.md)" >> $GITHUB_OUTPUT

name: Post review comment if: steps.diff.outputs.diff_size < 50000 uses: actions/github-script@v7 with: script: | github.rest.issues.createComment({ issuenumber: context.payload.pullrequest.number, owner: context.repo.owner, repo: context.repo.repo, body: process.env.REVIEW_OUTPUT }) `

Filtering Large Diffs

yaml name: Check diff size id: size run: | SIZE=$(wc -c < pr_diff.txt) echo "size=$SIZE" >> $GITHUB_OUTPUT if [ $SIZE -gt 50000 ]; then echo "::warning::PR diff too large ($SIZE bytes), skipping AI review" fi

Getting Your API Key

Set up your ofox.ai API key as a GitHub Secret:

Go to your repository → Settings → Secrets and variables → Actions
Click New repository secret
Name: OFOXAPIKEY
Value: your key from ofox.ai

👉 Get your ofox.ai API key

Expanding to Full Analysis

Beyond simple diff review, extend the workflow:

`yaml
Add after PR review
name: Run security scan
uses: github/codeql-action/analyze@v3
with:
category: "/language:javascript"

name: AI Summary
run: |

Generate PR summary with AI

SUMMARY=$(curl -s -X POST https://api.ofox.ai/v1/chat/completions \
-H "Authorization: Bearer $OFOXAPIKEY" \
-H "Content-Type: application/json" \
-d '{
"model": "claude-3-5-sonnet-20241022",
"messages": [{
"role": "user",
"content": "Summarize this PR in 3 bullet points. Focus on what changed and why.\n\n'$(cat pr_diff.txt | head -100)'"
}],
"max_tokens": 300
}')
echo "$SUMMARY" | jq -r '.choices[0].message.content'
`

Best Practices

Rate limit awareness — Don't run on every push; use types: [opened, synchronize]
Diff size limits — Skip reviews for massive changes (>50KB)
Token budget — Set max_tokens to control costs
Cache common prompts — Reuse system prompts across runs

Complete Example

`yaml
name: AI PR Review + Security Scan

on:
pull_request:
types: [opened, synchronize]

jobs:
review:
runs-on: ubuntu-latest
steps:

uses: actions/checkout@v4
with:
fetch-depth: 0
name: AI Review
env:
OFOXAPIKEY: ${{ secrets.OFOXAPIKEY }}
run: |
DIFF=$(git diff origin/${{ github.base_ref }}...HEAD)

curl -s -X POST https://api.ofox.ai/v1/chat/completions \
-H "Authorization: Bearer $OFOXAPIKEY" \
-H "Content-Type: application/json" \
-d '{
"model": "claude-3-5-sonnet-20241022",
"messages": [{
"role": "user",
"content": "Review this PR. Flag: bugs, security, performance, quality. Format: ## Bugs\n## Security\n## Performance\n## Quality\n\n' + "$DIFF" + '"
}],
"max_tokens": 1500
}' | jq -r '.choices[0].message.content' > review.md

github.rest.issues.createComment({
issuenumber: context.payload.pullrequest.number,
owner: context.repo.owner,
repo: context.repo.repo,
body: require('fs').readFileSync('review.md', 'utf8')
})
`

Get started with Claude-powered CI: ofox.ai

This article contains affiliate links.

Tags: github-actions,ci-cd,ai,programming,developer
Canonical URL: https://dev.to/zny10289

How to Price Your AI Development Services in 2026

ZNY — Fri, 15 May 2026 18:14:57 +0000

One of the most common questions developers ask when entering the AI services market: "What should I charge?" Pricing AI development is different from traditional software development. Here's a comprehensive guide based on real market data.

Why AI Development Commands Premium Rates

AI development isn't just coding — it involves:

Prompt engineering expertise — Crafting effective prompts is a specialized skill
Model selection and optimization — Choosing the right model for the right task
API integration complexity — Integrating multiple AI services reliably
Output validation — Ensuring AI outputs are correct and safe
Cost optimization — Managing token usage efficiently

These skills justify higher rates than traditional development.

Market Data: AI Developer Hourly Rates (2026)

Role	Entry	Mid	Senior
AI Integration Developer	$80/hr	$150/hr	$250+/hr
LLM Fine-tuning Specialist	$100/hr	$200/hr	$350+/hr
AI Product Engineer	$120/hr	$200/hr	$300+/hr
AI Security Auditor	$150/hr	$250/hr	$400+/hr

Pricing Models for AI Development

Model 1: Hourly Billing

Best for: Complex, undefined-scope projects

`python
Example hourly engagement
hourly_rate = 150 # USD
hours_estimated = 40
projectvalue = hourlyrate * hours_estimated # $6,000

Add AI API costs (pass-through)
apicosts = estimatedtokens * costpertoken
total = projectvalue + apicosts
`

Pros: Covers uncertainty
Cons: Client may resist open-ended billing

Model 2: Fixed Project Price

Best for: Well-defined, repeatable deliverables

`python
AI chatbot development (fixed price)
base_price = 5000 # Core chatbot
perfeaturemultiplier = 1.3 # Each major feature adds 30%
features = ["ollama", "vector-search", "web-scraping"]

price = base_price
for f in features:
price *= perfeaturemultiplier

print(f"Project price: ${price:.2f}") # ~$11,000
`

Pros: Client certainty, potentially higher value
Cons: Scope creep risk

Model 3: Value-Based Pricing

Best for: High-impact projects with measurable ROI

`python
Example: AI that saves 20 hours/week
hourssavedper_week = 20
hourly_value = 100
weeklyvalue = hourssavedperweek * hourly_value # $2,000/week

Price at 30% of annual value
annualvalue = weeklyvalue * 52 # $104,000
projectprice = annualvalue * 0.30 # $31,200
`

Pros: Captures true value
Cons: Harder to justify without data

Model 4: Subscription / Retainer

Best for: Ongoing AI development needs

`python
Monthly retainer model
base_hours = 20
hourly_rate = 125 # Discounted from $150
monthly = basehours * hourlyrate # $2,500/month

AI API costs passed through
estimatedapicost = 150
total = monthly + estimatedapicost # $2,650/month
`

Pros: Predictable revenue, deeper client relationship
Cons: Requires consistent value delivery

Real Project Examples

Example 1: RAG System for Law Firm

Project: Build a Retrieval-Augmented Generation system for case law research

Complexity: High (vector database, document parsing, citation verification)
Timeline: 3 weeks
Rate: $175/hr
Total: ~$21,000

Key pricing factors:
Specialized domain knowledge (legal)
High stakes = premium
Ongoing maintenance opportunity

Example 2: AI Code Review Bot

Project: GitHub integration that reviews pull requests using Claude

Complexity: Medium
Timeline: 1 week
Fixed price: $3,500

Key pricing factors:
Clear scope (GitHub PR → review comment)
Recurring usage (leads to subscription)
Developer audience (faster adoption)

Example 3: Customer Support AI Agent

Project: AI agent that handles tier-1 support tickets

Complexity: High (multi-turn conversation, tool use, escalation)
Timeline: 4 weeks
Value-based: $40,000

Key pricing factors:
Measurable ROI (80% ticket reduction)
Enterprise client
Integration complexity

AI API Cost Pass-Through

Always account for API costs in your pricing:

`python
def calculateaicost(monthlyusers, avgtokensperturn, turnsperconversation):

ofox.ai pricing example

costper1k_tokens = 0.003 # Claude 3.5 Sonnet

tokensperuser = avgtokensperturn turnsperconversation monthlyusers
monthlycost = (tokensperuser / 1000) * costper3ktokens

return monthly_cost

Example: 1000 users, 500 tokens/turn, 5 turns
users = 1000
tokensperturn = 500
turns = 5
monthlycost = calculateaicost(users, tokensper_turn, turns)
print(f"API cost: ${monthly_cost:.2f}/month") # ~$75/month
`

Negotiating AI Development Contracts

Separate AI costs from your labor — Makes pricing transparent
Build in revision limits — AI outputs may need more iteration
Define success metrics — "80% ticket resolution" not "good AI chatbot"
Include opt-out clauses — AI technology evolves fast
Price for uncertainty — Add 20-30% buffer for AI unpredictability

Getting Started with AI Development

Whether you're building AI integrations for clients or powering your own SaaS tools, the foundation is reliable API access. ofox.ai provides OpenAI-compatible Claude API with competitive pricing — perfect for production AI applications.

👉 Explore ofox.ai for your AI development needs

This article contains affiliate links.

Tags: freelancing,programming,ai,developer,career
Canonical URL: https://dev.to/zny10289

AI API Error Handling and Reliability: Production Best Practices

ZNY — Fri, 15 May 2026 18:09:46 +0000

Production AI applications fail in ways traditional software doesn't. Models go down, tokens run out, responses hallucinate, and rate limits hit at the worst moments. Here's how to build reliable AI-powered systems.

The AI Reliability Problem

Traditional APIs return consistent responses or clear errors. AI APIs introduce new failure modes:

Model outages — The provider's model goes down
Rate limits — You've exhausted your quota mid-request
Token limits — Your prompt exceeds context window
Hallucinations — Model returns plausible but wrong answers
Timeout — Request takes too long and hangs
Invalid JSON — Model returns malformed structured data

Retry Logic with Exponential Backoff

`typescript
async function withRetry(
fn: () => Promise,
options: {
maxRetries?: number;
baseDelay?: number;
maxDelay?: number;
onRetry?: (attempt: number, error: Error) => void;
} = {}
): Promise {
const {
maxRetries = 3,
baseDelay = 1000,
maxDelay = 30000,
onRetry
} = options;

for (let attempt = 1; attempt <= maxRetries + 1; attempt++) {
try {
return await fn();
} catch (error) {
const isRetryable = isRetryableError(error);
const isLastAttempt = attempt > maxRetries;

if (isLastAttempt || !isRetryable) {
throw error;
}

const delay = Math.min(baseDelay * Math.pow(2, attempt - 1), maxDelay);
onRetry?.(attempt, error as Error);
await sleep(delay);
}
}
throw new Error('Unreachable');
}

function isRetryableError(error: unknown): boolean {
if (error instanceof AIAPIError) {
// 429 = rate limit, 500 = server error, 503 = service unavailable
return [429, 500, 502, 503].includes(error.status);
}
// Network timeout
return (error as NodeJS.ErrnoException).code === 'ETIMEDOUT';
}
`

Handling Rate Limits Gracefully

`typescript
async function chatWithRateLimit(
client: ClaudeClient,
messages: ChatMessage[],
options: { maxWait?: number } = {}
): Promise {
const { maxWait = 60000 } = options;
const startTime = Date.now();

while (true) {
try {
return await client.chat(messages);
} catch (error) {
if (!(error instanceof AIAPIError) || error.status !== 429) throw error;

const retryAfter = error.retryAfter || 1000;
const elapsed = Date.now() - startTime;

if (elapsed + retryAfter > maxWait) {
throw new Error('Rate limit exceeded, max wait time reached');
}

console.log(Rate limited, waiting ${retryAfter}ms...);
await sleep(retryAfter);
}
}
}
`

Structured Output Validation

Models often return invalid JSON. Always validate:

`typescript
import { z } from 'zod';

const CodeReviewSchema = z.object({
bugs: z.array(z.object({
line: z.number(),
severity: z.enum(['low', 'medium', 'high']),
description: z.string()
})),
suggestions: z.array(z.string()),
score: z.number().min(0).max(10)
});

async function reviewCode(code: string): Promise {
const response = await client.chat([
{ role: 'user', content: Review this code:\n\n${code}\n\nReturn valid JSON. }
]);

try {
const parsed = JSON.parse(response.choices[0].message.content);
return CodeReviewSchema.parse(parsed);
} catch {
// Fallback: retry with stricter prompting
return reviewCodeWithFallback(code);
}
}
`

Circuit Breaker Pattern

Prevent cascading failures when AI API is degraded:

`typescript
class CircuitBreaker {
private failures = 0;
private lastFailure = 0;
private state: 'closed' | 'open' | 'half-open' = 'closed';

constructor(
private readonly threshold: number = 5,
private readonly timeout: number = 60000
) {}

async execute(fn: () => Promise): Promise {
if (this.state === 'open') {
if (Date.now() - this.lastFailure > this.timeout) {
this.state = 'half-open';
} else {
throw new Error('Circuit breaker is open');
}
}

try {
const result = await fn();
this.onSuccess();
return result;
} catch (error) {
this.onFailure();
throw error;
}
}

private onSuccess() {
this.failures = 0;
this.state = 'closed';
}

private onFailure() {
this.failures++;
this.lastFailure = Date.now();
if (this.failures >= this.threshold) {
this.state = 'open';
}
}
}

// Usage
const breaker = new CircuitBreaker(5, 60000);
const result = await breaker.execute(() => client.chat(messages));
`

Timeout Strategy

Set both connection and request timeouts:

typescript const response = await fetch(url, { method: 'POST', headers: { 'Authorization': Bearer ${apiKey}, 'Content-Type': 'application/json' }, body: JSON.stringify(payload), signal: AbortSignal.timeout(120000) // 2 minute timeout });

Cost Control

Prevent runaway costs with token budgets:

`typescript
class TokenBudget {
private spent = 0;

constructor(
private readonly maxBudget: number,
private readonly costPerToken: number
) {}

async executeWithBudget(fn: () => Promise): Promise {
const estimatedCost = this.estimateCost(fn);

if (this.spent + estimatedCost > this.maxBudget) {
throw new Error(Budget exceeded. Spent: ${this.spent}, Max: ${this.maxBudget});
}

const result = await fn();
this.spent += estimatedCost;
return result;
}

private estimateCost(fn: () => Promise): number {
// Rough estimate based on input size
return 0; // Would need implementation
}

getSpent(): number {
return this.spent;
}
}
`

Building Reliable AI Applications

The key insight: AI APIs require defensive programming at a level traditional APIs don't. Layer retry logic, circuit breakers, validation, and cost controls to build systems that degrade gracefully rather than fail catastrophically.

Get started with reliable AI API access: ofox.ai

This article contains affiliate links.

Tags: api,error-handling,programming,developer,reliability
Canonical URL: https://dev.to/zny10289

TypeScript AI SDK: Building Type-Safe AI-Powered Applications

ZNY — Fri, 15 May 2026 18:04:35 +0000

TypeScript's type system transforms AI API integration from guesswork into reliable, maintainable code. When you're building production AI applications, TypeScript catches errors at compile time, not runtime. Here's how to build type-safe AI integrations.

Why TypeScript for AI APIs?

Catch errors early — Invalid API responses caught at compile time
IDE autocompletion — Know exactly what the API returns
Refactoring confidence — Change types and the compiler finds all affected code
Documentation — Types serve as living documentation

Type-Safe API Client

`typescript
import { z } from 'zod';

// Define response schemas
const MessageSchema = z.object({
role: z.enum(['system', 'user', 'assistant']),
content: z.string()
});

const ChoiceSchema = z.object({
message: MessageSchema,
finish_reason: z.string(),
index: z.number()
});

const UsageSchema = z.object({
prompt_tokens: z.number(),
completion_tokens: z.number(),
total_tokens: z.number()
});

const ChatCompletionResponseSchema = z.object({
id: z.string(),
object: z.string(),
created: z.number(),
model: z.string(),
choices: z.array(ChoiceSchema),
usage: UsageSchema
});

type ChatMessage = z.infer;
type ChatResponse = z.infer;

// Type-safe client
class ClaudeClient {
constructor(
private readonly apiKey: string,
private readonly baseURL: string = 'https://api.ofox.ai/v1'
) {}

async chat(messages: ChatMessage[]): Promise {
const response = await fetch(${this.baseURL}/chat/completions, {
method: 'POST',
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify({
model: 'claude-3-5-sonnet-20241022',
messages
})
});

if (!response.ok) {
throw new Error(API error: ${response.status});
}

const data = await response.json();

// Zod validates at runtime AND gives you typed result
return ChatCompletionResponseSchema.parse(data);
}
}
`

Streaming with Types

`typescript
async function* streamChat(
client: ClaudeClient,
messages: ChatMessage[]
): AsyncGenerator {
const response = await fetch(${client.baseURL}/chat/completions, {
method: 'POST',
headers: {
'Authorization': Bearer ${client.apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify({
model: 'claude-3-5-sonnet-20241022',
messages,
stream: true
})
});

if (!response.body) throw new Error('No response body');

const decoder = new TextDecoder();

for await (const chunk of response.body) {
const lines = decoder.decode(chunk).split('\n');
for (const line of lines) {
if (line.startsWith('data: ')) {
const data = JSON.parse(line.slice(6));
if (data.choices[0].delta.content) {
yield data.choices[0].delta.content;
}
}
}
}
}

// Usage with full type safety
for await (const token of streamChat(client, messages)) {
process.stdout.write(token);
}
`

Building a Type-Safe Agent Framework

`typescript
// Tool definitions with full type safety
const ToolSchema = z.object({
name: z.string(),
description: z.string(),
parameters: z.record(z.string(), z.unknown())
});

type Tool = z.infer;

interface ToolResult {
success: boolean;
data?: T;
error?: string;
}

class TypedAgent {
constructor(
private client: ClaudeClient,
private tools: Tool[]
) {}

async execute(systemPrompt: string, task: string): Promise {
const messages: ChatMessage[] = [
{ role: 'system', content: systemPrompt },
{ role: 'user', content: task }
];

while (true) {
const response = await this.client.chat(messages);
const content = response.choices[0].message.content;

// Check if agent wants to use a tool
const toolCall = this.parseToolCall(content);
if (toolCall) {
const result = await this.executeTool(toolCall);
messages.push({ role: 'assistant', content });
messages.push({
role: 'user',
content: Tool result: ${JSON.stringify(result)}
});
} else {
return content;
}
}
}

private parseToolCall(content: string): { name: string; args: unknown } | null {
// Parse tool call from response
const match = content.match(/([\s\S]*?)<\/toolcall>/);
if (!match) return null;
return JSON.parse(match[1]);
}

private async executeTool(call: { name: string; args: unknown }): Promise> {
const tool = this.tools.find(t => t.name === call.name);
if (!tool) return { success: false, error: Unknown tool: ${call.name} };

try {
// Tool execution would be implemented here
return { success: true, data: null };
} catch (e) {
return { success: false, error: (e as Error).message };
}
}
}
`

Validation with Zod

Zod validates API responses at runtime:

`typescript
import { z } from 'zod';

// Define the exact shape you expect
const APIResponse = z.object({
model: z.string(),
id: z.string(),
choices: z.array(z.object({
message: z.object({
role: z.string(),
content: z.string()
}),
finish_reason: z.string()
}))
});

async function safeChat(client: ClaudeClient, messages: ChatMessage[]) {
try {
const raw = await client.chatRaw(messages); // Returns unknown
const validated = APIResponse.parse(raw); // Throws if shape doesn't match
return validated;
} catch (error) {
if (error instanceof z.ZodError) {
console.error('API response validation failed:', error.issues);
throw new Error('Unexpected API response format');
}
throw error;
}
}
`

Getting Started

Build type-safe AI applications with TypeScript and ofox.ai — their OpenAI-compatible API integrates seamlessly with TypeScript's type system. Sign up and start building.

👉 Get started with ofox.ai

This article contains affiliate links.

Tags: typescript,nodejs,ai,programming,developer
Canonical URL: https://dev.to/zny10289

Python AI API Development: Complete FastAPI + Claude Integration Guide

ZNY — Fri, 15 May 2026 17:59:23 +0000

FastAPI is the best Python framework for building AI-powered APIs. Combined with Claude via ofox.ai, you can create production-ready AI endpoints in minutes. Here's the complete guide.

Why FastAPI for AI APIs?

Async native — Handle concurrent AI requests efficiently
Automatic validation — Pydantic models validate inputs/outputs
OpenAPI docs — Built-in interactive API documentation
Type hints — Full IDE support and error checking
Production-ready — Used by major companies worldwide

Project Setup

bash mkdir claude-api-service cd claude-api-service python3 -m venv venv source venv/bin/activate pip install fastapi uvicorn httpx pydantic

Basic FastAPI + Claude Setup

`python
main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import httpx

app = FastAPI(title="Claude API Service")

class Message(BaseModel):
role: str
content: str

class ChatRequest(BaseModel):
model: str = "claude-3-5-sonnet-20241022"
messages: List[Message]
max_tokens: Optional[int] = 1024
temperature: Optional[float] = 0.7

class ChatResponse(BaseModel):
content: str
model: str
tokens_used: int

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
async with httpx.AsyncClient() as client:
response = await client.post(
"https://api.ofox.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {process.env.OFOXAPIKEY}",
"Content-Type": "application/json"
},
json={
"model": request.model,
"messages": [m.model_dump() for m in request.messages],
"maxtokens": request.maxtokens,
"temperature": request.temperature
},
timeout=60.0
)

if response.status_code != 200:
raise HTTPException(statuscode=response.statuscode, detail=response.text)

data = response.json()
return ChatResponse(
content=data["choices"][0]["message"]["content"],
model=data["model"],
tokensused=data["usage"]["totaltokens"]
)
`

Running the Server

bash export OFOXAPIKEY="your-key-here" uvicorn main:app --reload --port 8000

API docs available at http://localhost:8000/docs

Adding Streaming Responses

`python
from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
async def generate():
async with httpx.AsyncClient() as client:
async with client.stream(
"POST",
"https://api.ofox.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {process.env.OFOXAPIKEY}",
"Content-Type": "application/json"
},
json={
"model": request.model,
"messages": [m.model_dump() for m in request.messages],
"maxtokens": request.maxtokens,
"stream": True
},
timeout=60.0
) as response:
async for chunk in response.aiter_lines():
if chunk.startswith("data: "):
data = chunk[6:]
if data == "[DONE]":
break
yield f"{data}\n\n"

return StreamingResponse(generate(), media_type="text/event-stream")
`

Adding Authentication

`python
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

@app.post("/chat", response_model=ChatResponse)
async def chat(
request: ChatRequest,
credentials: HTTPAuthorizationCredentials = Depends(security)
):
if credentials.scheme != "Bearer" or not verifyapikey(credentials.credentials):
raise HTTPException(
statuscode=status.HTTP401_UNAUTHORIZED,
detail="Invalid API key"
)

... rest of chat logic

Production Deployment

For production, use:
Gunicorn + Uvicorn workers for concurrency
Redis for request caching
Rate limiting with middleware
Docker for containerization

dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Complete Example: Code Explanation API

`python
class ExplainRequest(BaseModel):
code: str
language: str = "python"

@app.post("/explain", response_model=ChatResponse)
async def explain_code(request: ExplainRequest):
prompt = f"""Explain this {request.language} code in simple terms:

{request.language} {request.code}

Provide a clear, concise explanation."""

chat_request = ChatRequest(
messages=[Message(role="user", content=prompt)]
)
return await chat(chat_request)
`

Getting Started

Build your AI API with FastAPI and Claude. ofox.ai provides the OpenAI-compatible Claude endpoint — sign up, get your API key, and deploy your first AI API in under 10 minutes.

👉 Get started with ofox.ai

This article contains affiliate links.

Tags: python,fastapi,claude-api,api,programming,developer
Canonical URL: https://dev.to/zny10289

DEV Community: ZNY

Redis Caching for AI Applications: Reducing Latency and Cost

Sort to ensure same messages always produce same key

Move to front (LRU-like behavior)

Try cache first

Cache miss - call API

Store in cache

Remove oldest 10%

Re-add remaining in order

... your most common queries

Full cost for misses, free for hits

Async Python for AI: Building High-Concurrency AI Applications

Check if error is retryable

Poll for result

API Security Best Practices for AI Applications in 2026

Block common prompt injection patterns

Check length

Check for blocked patterns

Strip control characters

Clean old entries

Return raw key ONLY ONCE to the user

Log and block

Continue with normal processing

Always filter by user_id to prevent leakage

Only clear THIS user's history

Never expose internal error details in production

Don't include: exc.message, stack trace, API keys

Log full error internally

Return sanitized error to client

Docker for AI Development: Containerizing LLM Applications

Optional: Add a Redis cache

Testing AI-Powered Applications: Strategies for LLM Integration

New version should pass at least as many tests

Inject bad response

LLM Context Window Management: Techniques for Handling Long Documents

Split by double newlines (paragraphs)

Save current chunk

Start new chunk with overlap

In production, use OpenAI or ofox.ai embeddings

Summarize older messages

Call LLM to summarize (pseudocode)

Track input tokens

Building React AI Chat Components in 2026: Complete Guide

GitHub Actions + AI: Automating Code Quality with Claude

Get PR context

Prepare review prompt

Call Claude via ofox.ai

Generate PR summary with AI

How to Price Your AI Development Services in 2026

ofox.ai pricing example

AI API Error Handling and Reliability: Production Best Practices

TypeScript AI SDK: Building Type-Safe AI-Powered Applications

Python AI API Development: Complete FastAPI + Claude Integration Guide

... rest of chat logic