Scaling Agent Systems Basics

python -m venv .venv && source .venv/bin/activate
# stdlib only for the examples below

import time
 
 
def fake_model(prompt: str) -> str:
    time.sleep(0.05)  # stand-in for network + generation
    return f"answer:{prompt[:20]}"
 
 
def run_agent_inline(user_goal: str) -> str:
    # In real agents this is many model/tool turns
    return fake_model(user_goal)
 
 
print(run_agent_inline("summarize ticket 42"))

from dataclasses import dataclass, field
from queue import Queue
from typing import Any
import uuid
 
 
@dataclass
class AgentJob:
    job_id: str
    goal: str
    meta: dict[str, Any] = field(default_factory=dict)
 
 
jobs: Queue[AgentJob | None] = Queue()
 
 
def enqueue(goal: str) -> str:
    job = AgentJob(job_id=str(uuid.uuid4()), goal=goal)
    jobs.put(job)
    return job.job_id
 
 
job_id = enqueue("summarize ticket 42")
print("enqueued", job_id, "depth", jobs.qsize())

def process_job(job: AgentJob) -> dict:
    result = fake_model(job.goal)
    return {"job_id": job.job_id, "result": result, "status": "done"}
 
 
def worker_loop(q: Queue[AgentJob | None]) -> None:
    while True:
        job = q.get()
        if job is None:
            q.task_done()
            break
        print(process_job(job))
        q.task_done()
 
 
enqueue("goal-a")
enqueue("goal-b")
jobs.put(None)  # poison pill to stop demo
worker_loop(jobs)

from threading import Lock
 
results: dict[str, dict] = {}
_lock = Lock()
 
 
def process_and_store(job: AgentJob) -> None:
    out = process_job(job)
    with _lock:
        results[job.job_id] = out
 
 
def get_status(job_id: str) -> dict:
    with _lock:
        return results.get(job_id, {"job_id": job_id, "status": "pending"})
 
 
j1 = enqueue("status demo")
# In a real system the worker runs in another process
process_and_store(jobs.get())
print(get_status(j1))

from concurrent.futures import ThreadPoolExecutor
from threading import Semaphore
 
MAX_IN_FLIGHT = 2
sem = Semaphore(MAX_IN_FLIGHT)
 
 
def run_with_cap(job: AgentJob) -> dict:
    with sem:
        return process_job(job)
 
 
def drain_with_pool(pending: list[AgentJob]) -> list[dict]:
    with ThreadPoolExecutor(max_workers=4) as pool:
        # max_workers can be higher than MAX_IN_FLIGHT; sem is the real gate
        return list(pool.map(run_with_cap, pending))
 
 
batch = [AgentJob(str(i), f"goal-{i}") for i in range(5)]
print(drain_with_pool(batch))

MAX_DEPTH = 3
 
 
def enqueue_with_limit(goal: str, q: Queue[AgentJob], max_depth: int = MAX_DEPTH) -> str:
    if q.qsize() >= max_depth:
        raise RuntimeError("queue_full: try later or shed load")
    job = AgentJob(job_id=str(uuid.uuid4()), goal=goal)
    q.put(job)
    return job.job_id
 
 
q2: Queue[AgentJob] = Queue()
for g in ["a", "b", "c", "d"]:
    try:
        print("ok", enqueue_with_limit(g, q2))
    except RuntimeError as e:
        print("rejected", e)

import hashlib
 
cache: dict[str, str] = {}
 
 
def cache_key(goal: str) -> str:
    return hashlib.sha256(goal.strip().lower().encode()).hexdigest()
 
 
def run_cached(goal: str) -> tuple[str, bool]:
    key = cache_key(goal)
    if key in cache:
        return cache[key], True
    value = fake_model(goal)
    cache[key] = value
    return value, False
 
 
print(run_cached("FAQ: reset password"))
print(run_cached("FAQ: reset password"))  # hit

from dataclasses import dataclass
 
 
@dataclass
class Budget:
    spent: float = 0.0
    soft_limit: float = 1.0
 
 
def pick_model(budget: Budget) -> str:
    if budget.spent >= budget.soft_limit * 0.8:
        return "cheap-small"
    return "premium-large"
 
 
def run_with_routing(goal: str, budget: Budget) -> dict:
    model = pick_model(budget)
    # pretend costs
    cost = 0.02 if model == "cheap-small" else 0.15
    budget.spent += cost
    return {"model": model, "result": fake_model(f"{model}:{goal}"), "spent": budget.spent}
 
 
b = Budget(spent=0.7, soft_limit=1.0)
print(run_with_routing("hard task", b))
print(run_with_routing("easy task", b))