Part 3: Performance Architecture

This is the third post in the engineering series over Aki, my AI companion bot. Part 1 covered the Telegram interface and the Mini App design. Part 2 broke down the conversation processing, memory triggers, and LLM schemas. This final post revolves around the operational side: how a single Python process manages state, prevents hallucinations without tanking latency, handles message concurrency without crashing, and keeps the API tokens from putting me in debt.

The source code is on GitHub, found here. The project landing page is at aki-landing.vercel.app. Part 1 is here and Part 2 is here.

Every optimization layer in the architecture shares a singular core principle: Never execute heavy logic in the direct path the user is waiting on. Offload, defer, skip, or cache everything else.

The complete request flow looks something like this:

User Message
     │
     ▼
┌─────────────────────────────────────────────────────────┐
│  TELEGRAM WEBHOOK (FastAPI / uvicorn, async)            │
│                                                         │
│  ► Debounce Buffer (asyncio tasks)                      │
│  ► Rate Limiter (in-memory sliding window)              │
└─────────────────────────┬───────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────┐
│  ORCHESTRATOR                                           │
│                                                         │
│  ► User fetch (TTL cache → PostgreSQL async pool)       │
│  ► Conversation history fetch (async pool)              │
└─────────────────────────┬───────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────┐
│  SOUL AGENT                                             │
│                                                         │
│  ► Prompt constructed with cache breakpoints            │
│  ► LLM call (cached prefix, async via LiteLLM)          │
│  ► Background: memory creation (asyncio.create_task)    │
└─────────────────────────┬───────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────┐
│  TELEGRAM DELIVERY                                      │
│                                                         │
│  ► "Typing..." indicator (proportional to message len)  │
│  ► Multi-message smart splitting                        │
│  ► Sticker / reaction (non-blocking)                    │
└─────────────────────────────────────────────────────────┘

The only processes the user actively waits for are the prompt cache retrieval/LLM API inference and the Telegram network delivery. Everything else is either fetched instantly from a lightweight cache, processed silently in the background, or executed in less than a millisecond.

The fastest way to slow down a conversational agent is sequentially requesting multiple different pieces of data from a remote database. The agent needs three separate things before it can even begin to generate a response: the user's profile, recent raw chat logs, and long-term diary entries.

Instead of waiting for one database trip to resolve before starting the next, I used asyncio.gather in my MemoryManager to execute all three asyncpg SQL queries against the Neon database simultaneously:

async def get_user_context(self, user_id: int) -> UserContextSchema:
    # Multi-fetch using gather for performance
    user_info, conversations, diary_entries = await asyncio.gather(
        self.db.get_user_by_id(user_id),
        self.db.get_recent_conversations(user_id, limit=settings.CONVERSATION_CONTEXT_LIMIT),
        self.db.get_diary_entries(user_id, limit=settings.DIARY_FETCH_LIMIT),
    )
    
    return UserContextSchema(
        user_info=user_info,
        recent_conversations=conversations,
        diary_entries=diary_entries,
    )

Because get_user_by_id usually hits the TTLCache in RAM (taking 0ms), the event loop immediately starts the network requests for get_recent_conversations and get_diary_entries in parallel. The entire context retrieval phase rarely takes longer than a single database roundtrip (roughly ~25ms), leaving the vast majority of the latency budget solely for the LLM inference.

The backbone of this system relies entirely on Python's asyncio and an asynchronous database driver.

When an API processes sequential database reads/writes synchronously, it forces the entire event loop to freeze while waiting for PostgreSQL to respond across the network. By utilizing asyncpg within a high-throughput SQLAlchemy connection pool, the FastAPI webhook gracefully accepts new messages exactly when the database is fetching records for a different user.

self.engine = create_async_engine(
    db_url,
    pool_size=10,       # Up to 10 concurrent DB connections
    max_overflow=20,    # Up to 20 additional during traffic spikes
    pool_pre_ping=True, # Test the connection before executing a query
    pool_recycle=3600,  # Prevent PostgreSQL from dropping idle connections
)

The database pool easily manages 30 concurrent user transactions before uvicorn ever begins queuing requests, allowing a single lightweight process server on Railway to manage the traffic seamlessly.

One of the most frequent database queries is fetching the user's settings profile (their exact timezone, configurations, and name). Hitting the PostgreSQL pool for this static data on every single API request is wasteful.

I implemented TTLCache from the cachetools library directly in the async memory manager to hold this data completely in RAM.

from cachetools import TTLCache
 
class AsyncMemoryManager:
    def __init__(self):
        self.db = db
        # In-memory caches (maxsize 100 users, 5-minute TTL)
        self._user_cache = TTLCache(maxsize=100, ttl=300)

Because conversations often occur in bursts (e.g., 20 texts back and forth over 3 minutes, then silence for hours), the 5-minute TTL (Time-To-Live) completely neutralizes database roundtrips during an active chat session. The database is queried exactly once when the conversation starts, and the cache absorbs all subsequent state lookups.

Because LLM inferences are infinitely more expensive than traditional database operations, allowing unbounded requests translates instantly to a skyrocketed bill. I needed rate limiting directly integrated into the routing logic—but without relying on Redis, which would simply add another network hop.

I built a sliding-window rate limiter utilizing a fast in-memory object cache per user:

class RateLimiter:
    def __init__(self, max_messages: int, window_seconds: int):
        self.max_messages = max_messages
        self.window_seconds = window_seconds
        self._user_windows: Dict[int, List[float]] = {}
 
    def check_rate_limit(self, user_id: int) -> tuple[bool, int]:
        now = time.time()
        cutoff = now - self.window_seconds
 
        timestamps = self._user_windows.setdefault(user_id, [])
        timestamps[:] = [t for t in timestamps if t > cutoff]
 
        if len(timestamps) >= self.max_messages:
            return False, 0
 
        timestamps.append(now)
        return True, self.max_messages - len(timestamps)

By default, users are capped at 10 requests per 60-second window. This computation relies on a lightweight array traversal, costing effectively nothing. When users hit the cap, the system traps the webhook and instantly responds with a locally-stored canned response without initiating an LLM API call at all.

The single largest performance bottleneck in conversational agents doesn't stem from the language model itself; it originates from user behavior. When texting naturally, humans rarely send monolithic paragraphs. We send fragments: "hey", followed by "so about yesterday", followed by the actual question.

If the FastAPI webhook handled every single incoming message sequentially, it would inevitably instantiate three disparate LLM API inferences concurrently. The resulting output would be chaotic, missing the combined contexts, and practically tripling system costs.

To solve this, I employed an asynchronous debounce buffer directly on the Webhook initialization layer:

# Check and cancel an existing timer if a new message arrives mid-thought
if chat_id in self._debounce_tasks:
    self._debounce_tasks[chat_id].cancel()
 
# Execute a new grouping timer (restarted on each text message received)
self._debounce_tasks[chat_id] = asyncio.create_task(
    self._process_buffered_messages(chat_id)
)

Under this model, the API holds the execution thread via asyncio.sleep(0.5) for precisely half a second. If the user stops typing, it merges the fragmented texts into a clean, unified payload and dispatches a singular API call. This effectively acts as cost-control middleware while giving the agent all the necessary context cleanly bundled.

Because LLM token costs can spiral out of control if a malicious user decides to spam the bot with novels all afternoon, I hardcoded a safety valve: the USER_DAILY_TOKEN_BUDGET.

Before the Orchestrator even invokes the heavy SoulAgent reasoning class, it runs a fast query aggregating the user's token consumption for the current day:

# agents/orchestrator.py
if settings.USER_DAILY_TOKEN_BUDGET > 0:
    usage_today = await self.memory.db.get_user_token_usage_today(user_id)
    if usage_today >= settings.USER_DAILY_TOKEN_BUDGET:
        return [
            "You've given me so much to think about today! 🧠✨ "
            "Let's pause here so I can process everything. "
            "I'll be refreshed and ready to talk more tomorrow!"
        ], "😴"

If the threshold is breached, the execution instantly short-circuits. It returns a gracefully formatted rejection message paired with a sleeping emoji (😴). No API calls to LiteLLM are made, completely shielding the system from an accidental denial-of-wallet.

Because tracking exact token consumption per model per user was vital for updating that exact usage_today calculation, I log every token spent directly into the database. However, executing this database write-logic sequentially would delay the bot from delivering its response to the Telegram user.

Similarly, every 30 messages, the system aggressively analyzes the chat log and generates two massive internal memories. Because these actions are critical internally but absolutely unessential for the user's perception of response speed, both are utterly offloaded into asyncio.create_task shells:

# Fired safely AFTER delivering the LLM output back to the Telegram client
asyncio.create_task(
    self.memory.record_token_usage(
        user_id=user_id,
        model=result.usage.model,
        # ...
    )
)
 
asyncio.create_task(
    self._maybe_create_compact_summary(
        user_id=user_id,
        conversation_history=conversation_history,
    )
)

The create_task() execution occurs simultaneously alongside the network I/O that renders the Telegram typing indicator animation on the native iOS application, keeping critical paths as short as absolutely possible.

Finally, the most expensive operation within an AI application is undoubtedly the LLM context window ingestion. I structured the prompt architecture identically for every user specifically to take advantage of Anthropic's Prompt Caching protocol.

The prompt comprises three fundamental layers defined explicitly across multiple dictionaries in the payload:

system_prompt_blocks = [
    {
        "type": "text",
        "text": static_text,
        "cache_control": {"type": "ephemeral"}  # ANTHROPIC BREAKPOINT 1
    },
    {
        "type": "text",
        "text": exchanges_block,
        "cache_control": {"type": "ephemeral"}  # ANTHROPIC BREAKPOINT 2
    },
    {
        "type": "text",
        "text": volatile_block
    }
]

By explicitly commanding Anthropic to cache these components ephemerally, the 3,000-token persona block (which remains functionally identical on every request) registers as a cache hit.

Anthropic's caching protocol is incredibly powerful for conversational agents: once a block is cached, subsequent "cache reads" are billed at a 90% discount compared to base input token costs. Furthermore, because the architecture doesn't have to ingest thousands of static tokens on every turn, the API inference latency drops by upwards of 85%. The cost plummets, and the bot responds significantly faster.

When combined tightly with the TTLCache in-memory module caching the user state, the asynchronous database fetch queues, and the debounce message grouping—this Python setup manages an exceptionally complex, highly-contextual AI application without burning a hole in the project budget.