Infrastructure
Building Agentic AI Platforms: A Deep Dive into the Infrastructure
From orchestration to memory systems—the technical architecture behind production-grade AI agent platforms that actually work.
D
David Kim•Jan 15, 202618 minute read
The Agent Infrastructure Challenge
Building a demo agent that calls a few tools is easy. Building a production platform that orchestrates thousands of concurrent agents, maintains long-term memory, handles failures gracefully, and scales cost-effectively? That's an entirely different beast.
This post breaks down the infrastructure components that power real agentic platforms—the kind you'd find at companies like Anthropic, OpenAI, or AI-first startups processing millions of agent invocations daily.
Architecture Overview
A production agentic platform typically has these core layers:
┌─────────────────────────────────────────────────────────┐
│ API Gateway │
│ (Rate limiting, Auth, Routing) │
├─────────────────────────────────────────────────────────┤
│ Agent Orchestrator │
│ (Task planning, Tool routing, State) │
├─────────────────────────────────────────────────────────┤
│ Tool Registry │ Memory Systems │
│ (Functions, APIs) │ (Short-term, Long-term) │
├─────────────────────────────────────────────────────────┤
│ LLM Provider Integration │
│ (Anthropic, OpenAI, Self-hosted) │
├─────────────────────────────────────────────────────────┤
│ Execution Runtime │ Observability │
│ (Sandboxing, Timeouts)│ (Traces, Metrics, Logs) │
└─────────────────────────────────────────────────────────┘
Let's dive into each component.
1. Agent Orchestrator
The orchestrator is the brain of your agentic platform. It manages:
Task Planning & Decomposition
When an agent receives a complex task, it needs to break it down into executable steps. The orchestrator manages this planning loop:
class AgentOrchestrator:
def __init__(self, llm_client, tool_registry, memory):
self.llm = llm_client
self.tools = tool_registry
self.memory = memory
async def execute(self, task: str, context: dict) -> str:
# Retrieve relevant memories
memories = await self.memory.recall(task)
# Planning loop
plan = await self._plan(task, memories, context)
results = []
for step in plan.steps:
if step.requires_tool:
result = await self._execute_tool(step)
else:
result = await self._reason(step)
results.append(result)
# Check if we need to replan based on results
if self._should_replan(results):
plan = await self._replan(task, results)
# Store execution in memory
await self.memory.store(task, results)
return self._synthesize(results)
State Management
Agents need to track state across multi-step executions. For simple cases, in-memory state works. For production:
# Redis-based agent state
class AgentState:
def __init__(self, redis_client, agent_id: str):
self.redis = redis_client
self.agent_id = agent_id
self.key = f"agent:{agent_id}:state"
async def get(self) -> dict:
data = await self.redis.get(self.key)
return json.loads(data) if data else {}
async def update(self, updates: dict, ttl: int = 3600):
state = await self.get()
state.update(updates)
await self.redis.setex(self.key, ttl, json.dumps(state))
async def get_conversation(self) -> list:
return await self.redis.lrange(f"{self.key}:messages", 0, -1)
async def add_message(self, message: dict):
await self.redis.rpush(f"{self.key}:messages", json.dumps(message))
Concurrency & Resource Management
A single user might spawn multiple agents, and each agent might spawn sub-agents. You need:
- Semaphores to limit concurrent LLM calls per user
- Circuit breakers to prevent cascade failures
- Priority queues for agent task scheduling
class ResourceManager:
def __init__(self):
self.user_semaphores = defaultdict(lambda: asyncio.Semaphore(10))
self.global_semaphore = asyncio.Semaphore(1000)
self.circuit_breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=30
)
async def acquire(self, user_id: str):
await self.global_semaphore.acquire()
await self.user_semaphores[user_id].acquire()
if not self.circuit_breaker.is_closed():
raise ServiceUnavailable("System temporarily overloaded")
def release(self, user_id: str):
self.user_semaphores[user_id].release()
self.global_semaphore.release()
2. Tool Registry & Execution
Tools are what make agents useful. Your tool system needs to be:
Discoverable
Agents need to know what tools are available. A well-structured registry:
@dataclass
class ToolDefinition:
name: str
description: str
parameters: dict # JSON Schema
required_permissions: list[str]
rate_limit: RateLimit
timeout: int
executor: Callable
class ToolRegistry:
def __init__(self):
self._tools: dict[str, ToolDefinition] = {}
def register(self, tool: ToolDefinition):
self._tools[tool.name] = tool
def get_available_tools(self, permissions: list[str]) -> list[dict]:
"""Return tool definitions the agent is allowed to use."""
return [
{
"name": t.name,
"description": t.description,
"parameters": t.parameters
}
for t in self._tools.values()
if all(p in permissions for p in t.required_permissions)
]
Secure
Never trust agent-generated tool inputs. Sandbox everything:
class SecureToolExecutor:
def __init__(self, sandbox_config: SandboxConfig):
self.sandbox = Sandbox(sandbox_config)
async def execute(
self,
tool: ToolDefinition,
params: dict,
context: ExecutionContext
) -> ToolResult:
# Validate parameters against schema
validate(params, tool.parameters)
# Check permissions
if not context.has_permissions(tool.required_permissions):
raise PermissionDenied(f"Missing permissions for {tool.name}")
# Execute in sandbox with timeout
try:
async with asyncio.timeout(tool.timeout):
result = await self.sandbox.run(
tool.executor,
params,
network_access=tool.requires_network,
filesystem_access=tool.requires_filesystem
)
return ToolResult(success=True, data=result)
except asyncio.TimeoutError:
return ToolResult(success=False, error="Tool execution timed out")
Observable
Every tool call should be traced:
@trace_tool_execution
async def execute_tool(self, tool_name: str, params: dict):
span = tracer.start_span(f"tool:{tool_name}")
span.set_attribute("tool.params", json.dumps(params))
try:
result = await self._execute(tool_name, params)
span.set_attribute("tool.success", True)
return result
except Exception as e:
span.set_attribute("tool.error", str(e))
span.set_status(Status(StatusCode.ERROR))
raise
finally:
span.end()
3. Memory Systems
Agents need memory to be useful across sessions. There are two types:
Short-term Memory (Working Memory)
The conversation context within a single session. Usually managed through the LLM's context window, but for long sessions:
class WorkingMemory:
def __init__(self, max_tokens: int = 8000):
self.messages = []
self.max_tokens = max_tokens
def add(self, message: dict):
self.messages.append(message)
self._compress_if_needed()
def _compress_if_needed(self):
"""Summarize old messages if we're running out of space."""
current_tokens = self._count_tokens()
if current_tokens > self.max_tokens:
# Keep system message and recent messages
system = self.messages[0]
recent = self.messages[-10:]
old = self.messages[1:-10]
# Summarize old messages
summary = self._summarize(old)
self.messages = [system, {"role": "system", "content": f"Previous context summary: {summary}"}] + recent
Long-term Memory (Persistent Memory)
For agents that remember across sessions:
class LongTermMemory:
def __init__(self, vector_store, embedding_model):
self.vectors = vector_store
self.embedder = embedding_model
async def store(
self,
content: str,
metadata: dict,
user_id: str
):
embedding = await self.embedder.embed(content)
await self.vectors.upsert(
id=str(uuid4()),
vector=embedding,
metadata={
**metadata,
"user_id": user_id,
"content": content,
"timestamp": datetime.utcnow().isoformat()
}
)
async def recall(
self,
query: str,
user_id: str,
limit: int = 5
) -> list[Memory]:
embedding = await self.embedder.embed(query)
results = await self.vectors.query(
vector=embedding,
filter={"user_id": user_id},
top_k=limit
)
return [Memory.from_result(r) for r in results]
Entity Memory
Some agents need to track specific entities (people, projects, etc.):
class EntityMemory:
def __init__(self, graph_db):
self.graph = graph_db
async def update_entity(
self,
entity_type: str,
entity_id: str,
properties: dict,
relationships: list[tuple[str, str, str]] # (rel_type, target_type, target_id)
):
# Upsert entity node
await self.graph.upsert_node(
label=entity_type,
id=entity_id,
properties=properties
)
# Create relationships
for rel_type, target_type, target_id in relationships:
await self.graph.create_edge(
from_label=entity_type,
from_id=entity_id,
to_label=target_type,
to_id=target_id,
relationship=rel_type
)
async def get_entity_context(self, entity_id: str, depth: int = 2) -> dict:
"""Get entity and its neighborhood for context."""
return await self.graph.traverse(entity_id, max_depth=depth)
4. LLM Provider Integration
You'll likely use multiple LLM providers. Abstract them:
class LLMRouter:
def __init__(self, providers: dict[str, LLMProvider]):
self.providers = providers
self.fallback_order = ["anthropic", "openai", "local"]
async def complete(
self,
messages: list[dict],
model: str = None,
temperature: float = 0.7
) -> str:
provider_name, model_id = self._route(model)
provider = self.providers[provider_name]
try:
return await provider.complete(messages, model_id, temperature)
except RateLimitError:
# Try fallback
return await self._fallback_complete(messages, temperature)
def _route(self, model: str) -> tuple[str, str]:
"""Route to appropriate provider based on model name."""
if model.startswith("claude"):
return ("anthropic", model)
elif model.startswith("gpt"):
return ("openai", model)
else:
return ("local", model)
Streaming Support
Production agents need streaming for good UX:
async def stream_agent_response(
self,
task: str,
context: dict
) -> AsyncGenerator[str, None]:
messages = self._build_messages(task, context)
async for chunk in self.llm.stream(messages):
# Parse for tool calls
if tool_call := self._parse_tool_call(chunk):
result = await self._execute_tool(tool_call)
yield f"[Tool: {tool_call.name}] {result}\n"
else:
yield chunk.text
5. Observability
You can't debug what you can't see. Essential observability for agents:
Tracing
Every agent execution should produce a trace:
class AgentTracer:
def __init__(self, otlp_endpoint: str):
self.tracer = trace.get_tracer("agent")
@contextmanager
def trace_execution(self, task_id: str, user_id: str):
with self.tracer.start_as_current_span("agent_execution") as span:
span.set_attribute("task_id", task_id)
span.set_attribute("user_id", user_id)
try:
yield span
except Exception as e:
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR))
raise
Metrics
Track what matters:
# Key metrics for agent platforms
agent_latency = Histogram("agent_execution_seconds", ["task_type", "model"])
tool_calls = Counter("agent_tool_calls_total", ["tool_name", "success"])
llm_tokens = Counter("agent_llm_tokens_total", ["model", "direction"])
agent_errors = Counter("agent_errors_total", ["error_type"])
concurrent_agents = Gauge("agent_concurrent_executions", ["user_tier"])
Logging
Structured logs for every decision:
logger.info(
"agent_step_complete",
task_id=task_id,
step_number=step_num,
step_type=step.type,
tool_used=step.tool_name,
tokens_used=response.usage.total_tokens,
latency_ms=latency_ms
)
Deployment Patterns
Kubernetes-native Architecture
Most production agent platforms run on Kubernetes:
apiVersion: apps/v1
kind: Deployment
metadata:
name: agent-orchestrator
spec:
replicas: 3
template:
spec:
containers:
- name: orchestrator
resources:
requests:
memory: "2Gi"
cpu: "1"
limits:
memory: "4Gi"
cpu: "2"
env:
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: agent-secrets
key: redis-url
Queue-based Execution
For high-throughput systems, decouple request handling from execution:
User Request → API Gateway → Task Queue → Agent Workers → Result Store
↓
Multiple Worker Pods
(Autoscaling based on queue depth)
Key Takeaways
- Orchestration is the hard part: Planning, state management, and error recovery are where most complexity lives
- Tool execution needs sandboxing: Never trust agent-generated inputs
- Memory is a first-class concern: Both working and long-term memory require careful design
- Observability is non-negotiable: You will debug production issues at 3 AM
- Start simple, add complexity as needed: You don't need every component on day one
Building agentic platforms is one of the most exciting areas in AI infrastructure right now. The patterns are still emerging, but the fundamentals—reliability, observability, and scalability—remain the same as any distributed system.