Understanding AI Agent Scaling
Scaling AI agents from a single user to a thousand concurrent users is one of the most challenging transitions in AI engineering. An AI agent that works beautifully in a developer's terminal often crumbles under production load due to state management conflicts, rate-limited LLM APIs, memory exhaustion, and unpredictable response times. Scaling is not merely about adding more servers—it requires rethinking your entire architecture around concurrency, fault tolerance, and resource optimization.
At its core, scaling AI agents means ensuring that every user receives a consistent, reliable, and responsive agent experience regardless of how many other users are simultaneously interacting with the system. This involves handling multiple concurrent conversation threads, managing shared and per-user state, orchestrating calls to external AI providers without exceeding their limits, and gracefully degrading when resources are strained.
What Defines an AI Agent in Production?
Before scaling, you must understand what constitutes an AI agent beyond a simple chat interface. A production AI agent typically includes:
- Tool execution — calling external APIs, querying databases, or running code in sandboxes
- Memory and context — maintaining conversation history, retrieving relevant documents from vector stores, and updating long-term memory
- Planning and reasoning — breaking complex tasks into subtasks, often requiring multiple LLM calls in sequence (chain-of-thought, ReAct loops)
- State machines — tracking where the agent is in a multi-step workflow
- Human-in-the-loop checkpoints — pausing execution to request user confirmation before sensitive actions
Each of these components introduces scaling challenges that compound as user count grows.
Why Scaling AI Agents Matters
The difference between a demo and a production system is measured in users. Here's why scaling matters concretely:
The Cost of Not Scaling
- Session collisions — without proper isolation, User A's conversation state can leak into User B's session, causing confusing or even dangerous agent behavior
- LLM provider rate limits — OpenAI, Anthropic, and other providers impose tokens-per-minute and requests-per-minute limits. A single user rarely hits these; 1,000 users will hit them constantly without proper queueing
- Exponential memory growth — each conversation accumulates history. Without eviction strategies, your vector store and conversation database grow linearly with users, but retrieval latency grows logarithmically, degrading everyone's experience
- Cold start amplification — when an agent process restarts, all active conversations may lose in-memory context. With 1,000 users, the probability of someone being mid-conversation during any restart approaches certainty
- Runaway cost loops — agents that enter reasoning loops (repeatedly calling tools without progress) waste tokens. One user's loop is annoying; 50 simultaneous loops can exhaust your monthly LLM budget in hours
The Business Impact
Scaling properly directly translates to user retention. If your agent takes 30 seconds to respond under load instead of the 3 seconds users expect, abandonment rates skyrocket. According to industry benchmarks, a 1-second increase in response time can reduce conversion rates by up to 7%. For AI agents that may already take several seconds to reason, adding scaling latency on top is fatal to user experience.
Architectural Patterns for Scaling
Let's build the scaling architecture step by step, starting from a simple single-user prototype and evolving it to handle 1,000 concurrent users.
Phase 1: The Prototype (Single User)
Your initial agent likely looks something like this—a straightforward loop with in-memory state:
# single_user_agent.py — DO NOT USE IN PRODUCTION
import openai
from datetime import datetime
# Global state — dangerous!
conversation_history = []
user_context = {}
async def run_agent(user_query: str) -> str:
global conversation_history
conversation_history.append({"role": "user", "content": user_query})
response = await openai.ChatCompletion.acreate(
model="gpt-4",
messages=conversation_history
)
reply = response.choices[0].message.content
conversation_history.append({"role": "assistant", "content": reply})
return replyThis code fails at scale because conversation_history is a single global list. Two concurrent users will interleave their messages, creating nonsensical merged conversations. The first step toward scaling is isolating state per user session.
Phase 2: Session Isolation
Introduce a session manager that maintains separate conversation state for each user:
# session_isolation.py — Production-ready session management
import asyncio
import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import openai
@dataclass
class AgentSession:
session_id: str
user_id: str
conversation_history: List[Dict] = field(default_factory=list)
working_memory: Dict = field(default_factory=dict)
created_at: datetime = field(default_factory=datetime.utcnow)
last_accessed: datetime = field(default_factory=datetime.utcnow)
active_tool_calls: int = 0
max_tool_calls: int = 15 # Prevent infinite loops
class SessionManager:
def __init__(self, max_sessions: int = 5000, ttl_minutes: int = 30):
self._sessions: Dict[str, AgentSession] = {}
self._lock = asyncio.Lock()
self.max_sessions = max_sessions
self.ttl = timedelta(minutes=ttl_minutes)
async def get_or_create_session(
self, user_id: str, session_id: Optional[str] = None
) -> AgentSession:
async with self._lock:
# Evict expired sessions before creating new ones
await self._evict_expired()
if session_id and session_id in self._sessions:
session = self._sessions[session_id]
session.last_accessed = datetime.utcnow()
return session
# Create new session
if len(self._sessions) >= self.max_sessions:
raise RuntimeError(
f"Session limit reached ({self.max_sessions}). "
"Scale your session store or increase max_sessions."
)
new_id = session_id or hashlib.sha256(
f"{user_id}:{datetime.utcnow().isoformat()}".encode()
).hexdigest()[:16]
session = AgentSession(
session_id=new_id,
user_id=user_id
)
self._sessions[new_id] = session
return session
async def _evict_expired(self):
now = datetime.utcnow()
expired = [
sid for sid, session in self._sessions.items()
if now - session.last_accessed > self.ttl
]
for sid in expired:
del self._sessions[sid]
if expired:
print(f"Evicted {len(expired)} expired sessions")
async def update_session(self, session: AgentSession):
async with self._lock:
session.last_accessed = datetime.utcnow()
self._sessions[session.session_id] = session
async def get_active_count(self) -> int:
return len(self._sessions)
# Usage
session_manager = SessionManager()
async def run_agent_for_user(user_id: str, user_query: str) -> str:
session = await session_manager.get_or_create_session(user_id)
# Per-session history — no cross-user contamination
session.conversation_history.append(
{"role": "user", "content": user_query}
)
# Guard against infinite loops
if session.active_tool_calls >= session.max_tool_calls:
return "I've reached my processing limit for this task. Let's simplify."
response = await openai.ChatCompletion.acreate(
model="gpt-4",
messages=session.conversation_history[-20:] # Window recent history
)
reply = response.choices[0].message.content
session.conversation_history.append(
{"role": "assistant", "content": reply}
)
await session_manager.update_session(session)
return replyThis solves session isolation, but in-memory storage still limits you to a single process. When you deploy multiple instances behind a load balancer, sessions created on one instance are invisible to others.
Phase 3: Externalizing State with Redis
Move session state to Redis so any application instance can serve any user:
# redis_session_manager.py — Horizontally scalable sessions
import json
import redis.asyncio as redis
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime
@dataclass
class AgentSession:
session_id: str
user_id: str
conversation_history: List[Dict]
working_memory: Dict
created_at: str # ISO format string for JSON serialization
last_accessed: str
active_tool_calls: int
class RedisSessionManager:
def __init__(
self,
redis_url: str = "redis://localhost:6379",
max_sessions_per_user: int = 10,
session_ttl_seconds: int = 1800,
history_window: int = 30
):
self.redis = redis.from_url(redis_url)
self.max_sessions_per_user = max_sessions_per_user
self.session_ttl = session_ttl_seconds
self.history_window = history_window
async def get_or_create_session(
self, user_id: str, session_id: Optional[str] = None
) -> AgentSession:
key = f"session:{user_id}:{session_id}" if session_id \
else f"session:{user_id}:{self._generate_id()}"
# Try to fetch existing session
raw = await self.redis.get(key)
if raw:
session = self._deserialize(raw.decode())
# Touch TTL
await self.redis.expire(key, self.session_ttl)
return session
# Check user session count
user_keys = await self.redis.keys(f"session:{user_id}:*")
if len(user_keys) >= self.max_sessions_per_user:
# Evict oldest session for this user
oldest_key = min(
user_keys,
key=lambda k: self._deserialize(
(await self.redis.get(k)).decode()
).last_accessed if await self.redis.get(k) else ""
)
await self.redis.delete(oldest_key)
# Create new session
new_session = AgentSession(
session_id=key.split(":")[-1],
user_id=user_id,
conversation_history=[],
working_memory={},
created_at=datetime.utcnow().isoformat(),
last_accessed=datetime.utcnow().isoformat(),
active_tool_calls=0
)
await self.redis.setex(
key,
self.session_ttl,
self._serialize(new_session)
)
return new_session
async def update_session(self, session: AgentSession):
key = f"session:{session.user_id}:{session.session_id}"
session.last_accessed = datetime.utcnow().isoformat()
# Window the conversation history to prevent unbounded growth
if len(session.conversation_history) > self.history_window * 2:
session.conversation_history = \
session.conversation_history[-self.history_window:]
await self.redis.setex(key, self.session_ttl, self._serialize(session))
def _serialize(self, session: AgentSession) -> str:
return json.dumps({
"session_id": session.session_id,
"user_id": session.user_id,
"conversation_history": session.conversation_history,
"working_memory": session.working_memory,
"created_at": session.created_at,
"last_accessed": session.last_accessed,
"active_tool_calls": session.active_tool_calls
})
def _deserialize(self, raw: str) -> AgentSession:
data = json.loads(raw)
return AgentSession(**data)
@staticmethod
def _generate_id() -> str:
import secrets
return secrets.token_hex(8)With Redis-backed sessions, you can now run dozens of application instances behind a load balancer. Every request finds the user's session regardless of which instance handles it.
Phase 4: LLM Rate Limiting and Queueing
At 1,000 users, your bottleneck shifts to the LLM provider. OpenAI's tiered rate limits mean you must implement client-side queueing with backpressure:
# llm_rate_limiter.py — Token bucket + priority queue
import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Callable, Coroutine, Optional
import time
class Priority(Enum):
CRITICAL = 0 # Human-in-the-loop responses
HIGH = 1 # Active user conversations
NORMAL = 2 # Background tasks
LOW = 3 # Batch processing, embeddings
@dataclass(order=True)
class LLMRequest:
priority: Priority
created_at: float = field(compare=False)
user_id: str = field(compare=False)
session_id: str = field(compare=False)
future: asyncio.Future = field(compare=False)
operation: Callable = field(compare=False)
args: tuple = field(compare=False)
kwargs: dict = field(compare=False)
def __post_init__(self):
if not hasattr(self, 'created_at'):
self.created_at = time.time()
class TokenBucketRateLimiter:
"""Token bucket algorithm for LLM API rate limiting"""
def __init__(
self,
tokens_per_minute: int = 90000, # GPT-4 tier 1 TPM
requests_per_minute: int = 3500, # GPT-4 tier 1 RPM
max_concurrent: int = 50,
max_queue_size: int = 500
):
self.tpm_limit = tokens_per_minute
self.rpm_limit = requests_per_minute
self.max_concurrent = max_concurrent
self.max_queue_size = max_queue_size
# Token buckets
self._token_bucket = tokens_per_minute
self._request_bucket = requests_per_minute
self._last_refill = time.monotonic()
# Concurrency control
self._in_flight = 0
self._semaphore = asyncio.Semaphore(max_concurrent)
# Priority queue
self._queue: asyncio.PriorityQueue = asyncio.PriorityQueue(
maxsize=max_queue_size
)
# Metrics
self.total_processed = 0
self.total_rejected = 0
self.total_queued = 0
self.avg_wait_ms = 0.0
self._refill_task = asyncio.create_task(self._periodic_refill())
async def _periodic_refill(self):
"""Refill tokens every second"""
while True:
await asyncio.sleep(1.0)
now = time.monotonic()
elapsed = now - self._last_refill
self._last_refill = now
# Refill proportional to elapsed time
self._token_bucket = min(
self.tpm_limit,
self._token_bucket + (self.tpm_limit / 60) * elapsed
)
self._request_bucket = min(
self.rpm_limit,
self._request_bucket + (self.rpm_limit / 60) * elapsed
)
async def submit(
self,
operation: Callable[..., Coroutine],
*args,
priority: Priority = Priority.NORMAL,
estimated_tokens: int = 1000,
user_id: str = "unknown",
session_id: str = "unknown",
**kwargs
) -> Any:
"""Submit an LLM request with priority queueing"""
# Create future for result
future: asyncio.Future = asyncio.Future()
request = LLMRequest(
priority=priority,
user_id=user_id,
session_id=session_id,
future=future,
operation=operation,
args=args,
kwargs=kwargs
)
# Try immediate processing for CRITICAL requests
if priority == Priority.CRITICAL:
try:
async with self._semaphore:
self._in_flight += 1
result = await operation(*args, **kwargs)
self._in_flight -= 1
future.set_result(result)
return await future
except Exception as e:
future.set_exception(e)
raise
# Queue the request
try:
self._queue.put_nowait(request)
self.total_queued += 1
except asyncio.QueueFull:
self.total_rejected += 1
raise RuntimeError(
f"LLM request queue full ({self.max_queue_size}). "
"Scale queue capacity or implement fallback responses."
)
# Start processing in background if not already running
asyncio.create_task(self._process_queue())
return await future
async def _process_queue(self):
"""Process queued requests while respecting rate limits"""
while not self._queue.empty():
# Check rate limits
if self._token_bucket < 1000 or self._request_bucket < 1:
await asyncio.sleep(0.5)
continue
async with self._semaphore:
request = await self._queue.get()
# Deduct tokens (simplified — production code would use
# tokenizers to estimate accurately)
estimated_tokens = request.kwargs.pop(
'estimated_tokens', 1000
)
self._token_bucket -= estimated_tokens
self._request_bucket -= 1
start_time = time.monotonic()
try:
self._in_flight += 1
result = await request.operation(
*request.args, **request.kwargs
)
elapsed_ms = (time.monotonic() - start_time) * 1000
# Update rolling average wait time
self.avg_wait_ms = (
0.95 * self.avg_wait_ms + 0.05 * elapsed_ms
)
self.total_processed += 1
request.future.set_result(result)
except Exception as e:
self.total_rejected += 1
request.future.set_exception(e)
finally:
self._in_flight -= 1
self._queue.task_done()
def get_metrics(self) -> dict:
return {
"token_bucket_remaining": int(self._token_bucket),
"request_bucket_remaining": int(self._request_bucket),
"in_flight": self._in_flight,
"queue_size": self._queue.qsize(),
"total_processed": self.total_processed,
"total_rejected": self.total_rejected,
"avg_wait_ms": round(self.avg_wait_ms, 1)
}
async def shutdown(self):
self._refill_task.cancel()
# Drain queue with timeout
while not self._queue.empty():
request = await self._queue.get()
request.future.set_exception(
RuntimeError("Rate limiter shutting down")
)
# Usage example
rate_limiter = TokenBucketRateLimiter(
tokens_per_minute=90000,
requests_per_minute=3500
)
async def call_llm_with_backpressure(messages: list, user_id: str, session_id: str):
return await rate_limiter.submit(
openai.ChatCompletion.acreate,
model="gpt-4",
messages=messages,
priority=Priority.HIGH,
estimated_tokens=len(str(messages)) // 4, # Rough estimate
user_id=user_id,
session_id=session_id
)This rate limiter uses a token bucket algorithm with priority queueing. Critical requests (like human-in-the-loop responses) can skip the queue, while normal requests wait for available capacity. The system also tracks metrics so you can monitor queue depth and adjust limits proactively.
Phase 5: Vector Store Scaling for RAG
Many agents use Retrieval-Augmented Generation (RAG) with vector databases. At 1,000 users, naive vector search becomes a bottleneck:
# vector_store_scaling.py — Partitioned vector search with caching
import asyncio
import hashlib
from typing import List, Dict, Optional
import numpy as np
from dataclasses import dataclass
@dataclass
class SearchResult:
document_id: str
content: str
score: float
metadata: Dict
class PartitionedVectorStore:
"""
Production vector store with semantic caching and sharding.
Assumes underlying Pinecone, Weaviate, or pgvector backend.
"""
def __init__(
self,
backend_client, # Your actual vector DB client
cache_ttl_seconds: int = 300,
max_cache_entries: int = 10000,
shard_count: int = 8,
default_top_k: int = 10
):
self.backend = backend_client
self.cache_ttl = cache_ttl_seconds
self.max_cache_entries = max_cache_entries
self.shard_count = shard_count
self.default_top_k = default_top_k
# LRU cache for query embeddings → results
self._cache: Dict[str, tuple] = {} # hash → (results, timestamp)
self._cache_lock = asyncio.Lock()
# Metrics
self.cache_hits = 0
self.cache_misses = 0
def _get_shard_key(self, user_id: str) -> int:
"""Determine which shard handles this user's documents"""
return int(hashlib.md5(user_id.encode()).hexdigest(), 16) % self.shard_count
def _query_cache_key(self, query_embedding: List[float], filters: Dict) -> str:
"""Create deterministic cache key from query"""
embedding_hash = hashlib.sha256(
np.array(query_embedding).tobytes()
).hexdigest()[:16]
filter_hash = hashlib.sha256(
str(sorted(filters.items())).encode()
).hexdigest()[:16]
return f"{embedding_hash}:{filter_hash}"
async def search(
self,
query_embedding: List[float],
user_id: str,
filters: Optional[Dict] = None,
top_k: int = None,
use_cache: bool = True
) -> List[SearchResult]:
top_k = top_k or self.default_top_k
filters = filters or {}
# Check semantic cache for identical queries
if use_cache:
cache_key = self._query_cache_key(query_embedding, filters)
async with self._cache_lock:
if cache_key in self._cache:
results, timestamp = self._cache[cache_key]
if (asyncio.get_event_loop().time() - timestamp) < self.cache_ttl:
self.cache_hits += 1
return results
# Cache miss — query backend with shard awareness
self.cache_misses += 1
shard_key = self._get_shard_key(user_id)
# Query only the relevant shard plus a fallback global shard
raw_results = await self.backend.query(
vector=query_embedding,
filter={"shard": {"$in": [shard_key, "global"]}, **filters},
top_k=top_k * 2 # Overfetch for reranking
)
# Rerank results based on user-specific relevance
results = self._rerank_for_user(raw_results, user_id)[:top_k]
# Update cache
async with self._cache_lock:
# Evict oldest entries if cache is full
if len(self._cache) >= self.max_cache_entries:
oldest_key = min(
self._cache.keys(),
key=lambda k: self._cache[k][1]
)
del self._cache[oldest_key]
self._cache[cache_key] = (
results,
asyncio.get_event_loop().time()
)
return results
def _rerank_for_user(self, results: List, user_id: str) -> List[SearchResult]:
"""Apply user-specific relevance scoring"""
# Production implementation would use cross-encoder reranking
# or learned user preferences
return sorted(
results,
key=lambda r: r.get('score', 0) * (
1.5 if r.get('metadata', {}).get('user_id') == user_id else 1.0
),
reverse=True
)
async def index_document(
self,
document_id: str,
content: str,
embedding: List[float],
user_id: str,
metadata: Dict
):
"""Index with shard assignment"""
shard_key = self._get_shard_key(user_id)
await self.backend.upsert(
id=document_id,
vector=embedding,
metadata={
**metadata,
"shard": shard_key,
"user_id": user_id
}
)
# Invalidate cache entries that might include this document
async with self._cache_lock:
self._cache.clear() # Simplified; production uses targeted invalidation
def get_cache_stats(self) -> Dict:
total = self.cache_hits + self.cache_misses
return {
"cache_hits": self.cache_hits,
"cache_misses": self.cache_misses,
"hit_rate": self.cache_hits / total if total > 0 else 0
}This partitioned vector store shards documents by user, caches frequent queries, and reranks results based on user relevance. At scale, this dramatically reduces query latency and backend load.
Phase 6: Full Production Agent Loop
Now let's integrate all these components into a production-grade agent loop that handles 1,000 concurrent users:
# production_agent.py — Full scalable agent implementation
import asyncio
import traceback
from datetime import datetime
from typing import Dict, List, Optional, Any
import json
from redis_session_manager import RedisSessionManager
from llm_rate_limiter import TokenBucketRateLimiter, Priority
from vector_store_scaling import PartitionedVectorStore
class ScalableAgent:
"""
Production AI agent designed for 1,000+ concurrent users.
Integrates session management, rate limiting, tool execution,
and vector search with full observability.
"""
def __init__(
self,
session_manager: RedisSessionManager,
rate_limiter: TokenBucketRateLimiter,
vector_store: PartitionedVectorStore,
max_tool_iterations: int = 10,
response_timeout_seconds: int = 30,
tool_registry: Optional[Dict[str, callable]] = None
):
self.sessions = session_manager
self.rate_limiter = rate_limiter
self.vector_store = vector_store
self.max_tool_iterations = max_tool_iterations
self.response_timeout = response_timeout_seconds
self.tools = tool_registry or {}
# Circuit breaker state
self._failure_count = 0
self._circuit_open = False
self._last_failure_time = None
async def handle_message(
self,
user_id: str,
message: str,
session_id: Optional[str] = None,
attachments: Optional[List[Dict]] = None
) -> Dict[str, Any]:
"""Main entry point — handles one user message end-to-end"""
# Circuit breaker check
if self._circuit_open:
if (datetime.utcnow() - self._last_failure_time).seconds < 60:
return {
"response": "I'm experiencing high load right now. "
"Please try again in a moment.",
"status": "degraded",
"session_id": session_id
}
else:
self._circuit_open = False # Half-open for testing
try:
# Get or create session
session = await self.sessions.get_or_create_session(
user_id, session_id
)
# Add user message to history
session.conversation_history.append({
"role": "user",
"content": message,
"timestamp": datetime.utcnow().isoformat()
})
# Check for tool loop guard
if session.active_tool_calls >= self.max_tool_iterations:
return await self._fallback_response(
session,
"Task complexity exceeded. Let's break this into smaller steps."
)
# Run the agent loop with timeout
try:
response = await asyncio.wait_for(
self._agent_loop(session, message, attachments),
timeout=self.response_timeout
)
except asyncio.TimeoutError:
return await self._fallback_response(
session,
"I'm taking longer than expected. "
"Let me summarize what I've found so far."
)
# Persist updated session
await self.sessions.update_session(session)
# Reset circuit breaker on success
self._failure_count = 0
return {
"response": response,
"session_id": session.session_id,
"status": "success",
"tool_calls_used": session.active_tool_calls
}
except Exception as e:
self._failure_count += 1
self._last_failure_time = datetime.utcnow()
if self._failure_count >= 5:
self._circuit_open = True
# Log the error with full context
print(f"Agent error for user {user_id}: {traceback.format_exc()}")
return {
"response": "I encountered an error processing your request. "
"Our team has been notified.",
"session_id": session_id,
"status": "error",
"error_code": "INTERNAL_ERROR"
}
async def _agent_loop(
self,
session,
user_message: str,
attachments: Optional[List[Dict]]
) -> str:
"""Core reasoning loop with tool execution"""
# Build system prompt with session context
system_prompt = self._build_system_prompt(session)
# Retrieve relevant documents if vector store available
if self.vector_store:
# Generate embedding (simplified — use actual embedding model)
query_embedding = await self._get_embedding(user_message)
relevant_docs = await self.vector_store.search(
query_embedding=query_embedding,
user_id=session.user_id,
top_k=5
)
context_chunks = "\n".join([
f"[Document {i+1}]: {doc.content[:500]}"
for i, doc in enumerate(relevant_docs)
])
else:
context_chunks = ""
# Prepare messages
messages = [
{"role": "system", "content": system_prompt},
{"role": "system", "content": f"Relevant context:\n{context_chunks}"}
if context_chunks else None,
*session.conversation_history[-20:] # Windowed history
]
messages = [m for m in messages if m is not None]
# Call LLM with rate limiting
response = await self.rate_limiter.submit(
self._call_llm,
messages=messages,
priority=Priority.HIGH,
estimated_tokens=len(str(messages)) // 4 + 500,
user_id=session.user_id,
session_id=session.session_id
)
# Parse tool calls from response
tool_calls = self._extract_tool_calls(response)
# Execute tools if any
for tool_call in tool_calls:
session.active_tool_calls += 1
if session.active_tool_calls >= self.max_tool_iterations:
break
tool_name = tool_call.get("name")
tool_args = tool_call.get("arguments", {})
if tool_name in self.tools:
try:
tool_result = await self.tools[tool_name](**tool_args)
# Feed tool result back into conversation
session.conversation_history.append({
"role": "tool",
"name": tool_name,
"content": json.dumps(tool_result)
})
except Exception as tool_error:
session.conversation_history.append({
"role": "tool",
"name": tool_name,
"content": f"Error: {str(tool_error)}"
})
# Extract final text response
final_response = self._extract_text_response(response)
session.conversation_history.append({
"role": "assistant",
"content": final_response,
"timestamp": datetime.utcnow().isoformat()
})
return final_response
async def _call_llm(self, messages: List[Dict]) -> str:
"""Actual LLM call — replace with your provider"""
# This is where you'd call OpenAI, Anthropic, etc.
# The rate limiter handles concurrency and quotas
import openai
response = await openai.ChatCompletion.acreate(
model="gpt-4",
messages=messages,
temperature=0.7,
max_tokens=1000
)
return response.choices[0].message.content
def _build_system_prompt(self, session) -> str:
return f"""You are a helpful AI assistant.
Current user: {session.user_id}
Session started: {session.created_at}
Tools available: {', '.join(self.tools.keys())}
Always be concise and accurate. If you need to use a tool,
specify it clearly in your response."""
async def _get_embedding(self,