DEV Community: Shantanav Kapse

Agent Skills 101: Giving Your AI Hands, Eyes, and Safety Rails

Shantanav Kapse — Mon, 07 Sep 2026 14:49:50 +0000

When I first started building with Large Language Models, I remember feeling a strange mix of awe and frustration.
You could ask an LLM to write a Shakespearean sonnet about Kubernetes, and it would do it in four seconds. But the moment you asked it to check the current weather, read a local SQLite database, or summarize a newly uploaded PDF, it hit an invisible wall. It was like chatting with a genius trapped inside a soundproof glass cube: brilliant, articulate, but completely disconnected from the outside world.

That wall disappears the moment you introduce agent skills.
Giving an agent "skills" (often called tool use or function calling) is what turns a passive text generator into an active problem solver. But moving from a toy demo to something reliable requires more than just passing a list of functions to an API endpoint.
In this guide, we'll walk through what agent skills actually are under the hood, explore battle-tested orchestration patterns, and cover the practical guardrails you need to keep your system safe and stable.

1. Traditional LLMs vs. Agentic Systems: The Mental Model

In a traditional setup, interaction is purely linear:

User Input -> [ LLM ] -> Output Text

You supply a prompt, the model calculates probability distributions over tokens, and it returns a stream of text. If the answer requires real-time data or an external calculation, the model has to guess-which usually leads to confident hallucinations.
An agentic architecture transforms this linear request into a continuous reasoning loop.
Instead of rushing to produce a final answer, the model follows four distinct steps:
Observe: Assess the user's request and examine the current state or environment.
Think: Reason about the goal. Does it know the answer, or does it need external data?
Act: If it needs data or needs to make a change, it triggers a skill (an API call, a database query, or a script).
Reflect: It evaluates the output returned by that skill and decides whether the task is complete or if another step is required.

2. What Is an "Agent Skill" Under the Hood?

There is no mystical AI magic happening inside a tool call. An agent skill consists of two simple pieces:
A plain code function: A Python or JavaScript function you wrote that does something concrete (like fetching an API or parsing a file).
A structured schema: A JSON specification telling the LLM what the tool is named, what it does, and what arguments it expects.
Here is what that looks like in Python:

import json

# 1. The actual executable function
def get_user_subscription(user_id: str) -> dict:
    """Fetches subscription status for a given user ID from our system."""
    database = {
        "usr_101": {"tier": "Pro", "status": "active", "renewal_days": 12},
        "usr_102": {"tier": "Free", "status": "active", "renewal_days": 0},
    }
    return database.get(user_id, {"error": "User not found"})

# 2. The schema the LLM reads to know this skill exists
subscription_tool_schema = {
    "type": "function",
    "function": {
        "name": "get_user_subscription",
        "description": "Look up account subscription tier and status using a user ID.",
        "parameters": {
            "type": "object",
            "properties": {
                "user_id": {
                    "type": "string",
                    "description": "The unique user ID, formatted like 'usr_123'",
                }
            },
            "required": ["user_id"],
        },
    },
}

When you send this schema alongside your user prompt, the LLM doesn't execute the function itself. Instead, it outputs a structured JSON object saying:
"Hey, I don't know the answer directly, but please run get_user_subscription(user_id='usr_101') and show me what it returns."
Your application executes the function locally, feeds the result back into the LLM as an observation, and the agent delivers the final answer.

3. Orchestration Patterns: When One Agent Isn't Enough

When you are starting out, it is tempting to give a single agent twenty different skills and hope for the best.
In practice, this causes "context pollution." The model gets overwhelmed by too many tool definitions, forgets instructions, and hallucinates parameters. To build reliable workflows, we split responsibilities across multiple specialized agents.

Pattern A: Orchestrator-Worker
Think of this like a project manager working with senior engineers:

The Orchestrator receives the high-level prompt, devises a multi-step plan, and delegates subtasks.
Each Worker has access only to 2–3 skills relevant to its specific domain.
Once the workers finish, the orchestrator aggregates the results into a cohesive final output.
Pattern B: Generator-Critic
Ever write an email when you were tired, only to cringe when reading it the next morning? That is why you need a critic.
The Generator creates an initial draft, writes a SQL query, or proposes a code patch using its skills.
The Critic inspects that output against a rubric (e.g., checking for SQL injection vulnerabilities, syntax errors, or missed requirements).
If the critic spots an issue, it provides structured feedback back to the generator to retry. This self-correction loop catches bugs before your code touches production.

4. State Management: The "Zero-Database" Philosophy

When building agent workflows, a common beginner trap is immediately reaching for complex infrastructure: vector databases, dedicated caching layers, and external message queues.
Before adding heavy databases, consider the zero-database approach: keep your state in-memory using clean state graphs.
A state graph is a mental and architectural model where every step in your agent's process is a node, and the transitions between them are edges based on conditional logic.

# A simple in-memory state dictionary
class AgentState:
    def __init__(self, task: str):
        self.task = task
        self.history = []
        self.current_worker = "planner"
        self.intermediate_results = {}
        self.iterations = 0

By passing a single structured state dictionary between functions:
You avoid race conditions and synchronization headaches.
Your workflow remains fully reproducible and testable in local development.
You can persist the entire state to a plain JSON file if you need persistence across app reboots.
Only reach for an external database when you have long-running tasks lasting days or when your history exceeds LLM context windows.

5. The "Infinite Loop" Problem & Circuit Breakers

Here is a rite of passage for every AI developer: you give an agent a code execution tool, run it against a tricky task, walk away to grab coffee, and return to find it has executed 78 consecutive tool calls trying to fix the same syntax error-burning through your monthly API budget in six minutes.
Agents are relentless optimizers. If a tool fails, they will often retry with minor variations indefinitely unless you put hard boundaries in place.
Enter the Circuit Breaker:

class CircuitBreaker:
    def __init__(self, max_steps: int = 8, max_consecutive_errors: int = 3):
        self.max_steps = max_steps
        self.max_consecutive_errors = max_consecutive_errors
        self.step_count = 0
        self.error_count = 0

    def record_step(self, is_error: bool = False):
        self.step_count += 1
        if is_error:
            self.error_count += 1
        else:
            self.error_count = 0  # Reset consecutive error counter

        if self.step_count >= self.max_steps:
            raise RuntimeError("Circuit breaker tripped: Maximum step limit reached.")

        if self.error_count >= self.max_consecutive_errors:
            raise RuntimeError("Circuit breaker tripped: Too many consecutive tool failures.")

Before invoking any tool or calling the LLM, tick your circuit breaker. If the agent gets stuck in a recursive failure spiral, your circuit breaker halts execution gracefully, logs the trace, and notifies you.

6. Sandboxing: Never Give an Agent Bare Metal

If one of your agent's skills is run_python_code or run_shell_command, never execute those commands directly on your primary workstation or host server.
Even a well-intentioned model might run os.remove() on an unintended folder, clone huge repos that fill up disk space, or spawn zombie processes.
Subprocess Isolation: At a bare minimum, run code through a restricted Python subprocess with strict timeouts and memory limits:

import subprocess

def run_isolated_code(script: str, timeout_seconds: int = 5) -> str:
    try:
        result = subprocess.run(
            ["python3", "-c", script],
            capture_output=True,
            text=True,
            timeout=timeout_seconds,
            check=True
        )
        return result.stdout
    except subprocess.TimeoutExpired:
        return "Execution timed out."
    except subprocess.CalledProcessError as e:
        return f"Error: {e.stderr}"

Container Sandboxes: For production environments, spin up lightweight, ephemeral Docker containers or micro-VMs that are destroyed the moment the agent finishes its run.

7. Headless Mock Testing: Don't Go Broke Testing

Testing an agent by running live queries through an LLM API endpoint is slow, non-deterministic, and expensive.
Instead, practice headless mock testing:
Separate your tool logic from your LLM calling logic.
Test your tools with standard unit tests just like any normal software module.
Mock the LLM's tool-call response using static JSON fixtures to verify that your orchestration loop parses arguments, updates state, and handles exceptions correctly without spending a penny.

def test_user_subscription_tool():
    # Unit test the skill directly without invoking any LLM
    result = get_user_subscription("usr_101")
    assert result["tier"] == "Pro"
    assert result["status"] == "active"

If your tool functions aren't reliable in isolation, wrapping them in an AI prompt will only amplify the failures.

8. Human-in-the-Loop (HITL): The Final Checkpoint

A simple rule of thumb for designing agent skills:
Automate reads. Guard writes.
Read actions (searching docs, analyzing data, running calculations) are safe to run autonomously.
Write actions (sending an email, modifying a production database, executing a bank transfer) should always trigger a human confirmation prompt.

def execute_database_update(query: str, state: AgentState):
    if not state.is_human_approved:
        return "Action paused. Awaiting explicit human approval to run update query."
    # Run the database query...

By designing your state machine to pause execution when encountering high-risk actions, you get all the speed advantages of automation without the risk of accidental chaos.

Quick Reference Summary

Component	Responsibility	Failure Mode to Watch For
Tool Schema	Declares name, types, and utility to the model.	Ambiguous descriptions causing inappropriate tool selection.
Orchestrator	Deconstructs goals and delegates to workers.	Attempting to execute tasks directly rather than delegating.
Circuit Breaker	Tracks iterations and limits consecutive errors.	Silent infinite retry loops that drain API balances.
Execution Sandbox	Isolates runtime commands from host machine.	Accidental filesystem corruption or unbound memory leaks.
HITL Interceptors	Enforces human approval for destructive mutations.	Autonomous writes leading to silent data corruption.

Multi-Agent Design Patterns: Beyond Sequential Chains

Shantanav Kapse — Thu, 27 Aug 2026 09:01:28 +0000

When engineers build their first LLM application, the journey almost always starts with a linear chain: Prompt -> LLM -> Output. If things get slightly more complex, they reach for a standard ReAct loop where a single model reasons, picks a tool, observes the result, and loops until finished.

In simple demos, this works. In production, it breaks.

In my previous deep-dive on building a real-time ASR engine, the primary architectural lesson was decoupling-separating audio ingestion from model processing to eliminate WebSocket backpressure. Multi-agent systems face a remarkably similar challenge: forcing a single generalist LLM to handle parsing, reasoning, code generation, and quality control creates cognitive backpressure, context contamination, and compounding failure rates. If node 2 hallucinates a variable, node 5 crashes.

As we explored when engineering the Autonomous Business Discovery Engine, moving past naive linear pipelines requires structured multi-agent choreography. When an agentic system must handle non-deterministic real-world inputs, decomposition and specialized control loops are the only way to build resilience.

Here is an architectural breakdown of the key multi-agent design patterns that take systems beyond sequential chains, how to implement them in LangGraph, and the trade-offs that come with each.

1. The Orchestrator-Worker (Router & Fan-Out) Pattern

In a naive linear pipeline, each task runs serially. If an agent needs to extract requirements from meeting transcripts, parse UI screenshots, and review API documentation, running them sequentially creates massive latency and forces irrelevant tokens into the shared context.

The Orchestrator-Worker pattern uses a central supervisor to classify the objective, break it down into independent tasks, and dispatch them to specialized sub-agents in parallel before synthesizing the final output.

Key Engineering Benefits:

Context Isolation: Each worker operates inside a minimal, dedicated context window. The vision model only sees image payloads; the code parser only receives AST trees and schemas.
Latency Reduction: Independent ingestion jobs execute concurrently via asyncio or worker pools.
Specialized Routing: You can route specific nodes to targeted models (e.g., lightweight vision models for image processing, specialized coder models for code generation) rather than paying for a giant generalist model on every task.

LangGraph Implementation:

from typing import Annotated, List, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

class OverallState(TypedDict):
    objective: str
    subtasks: List[str]
    # Reducer appends parallel worker outputs concurrently
    worker_results: Annotated[List[dict], operator.add]
    final_output: str

class WorkerState(TypedDict):
    subtask: str

def orchestrator(state: OverallState) -> dict:
    # Break down the user objective into distinct, isolated work units
    tasks = [
        "extract_functional_specs",
        "parse_ui_wireframes",
        "audit_api_contracts"
    ]
    return {"subtasks": tasks}

def worker_node(state: WorkerState) -> dict:
    # Each worker executes independently in its own scoped context
    task = state["subtask"]
    result = f"Processed {task} successfully"
    return {"worker_results": [{"task": task, "result": result}]}

def fan_out(state: OverallState):
    # Dispatch tasks dynamically to parallel instances of worker_node
    return [Send("worker_node", {"subtask": t}) for t in state["subtasks"]]

def synthesizer(state: OverallState) -> dict:
    # Combine results once all workers complete
    combined = "\n".join([f"- {r['task']}: {r['result']}" for r in state["worker_results"]])
    return {"final_output": f"Consolidated Summary:\n{combined}"}

# Build the Graph
builder = StateGraph(OverallState)
builder.add_node("orchestrator", orchestrator)
builder.add_node("worker_node", worker_node)
builder.add_node("synthesizer", synthesizer)

builder.add_edge(START, "orchestrator")
builder.add_conditional_edges("orchestrator", fan_out, ["worker_node"])
builder.add_edge("worker_node", "synthesizer")
builder.add_edge("synthesizer", END)

graph = builder.compile()

2. The Generator-Critic (Evaluator-Optimizer) Pattern

One of the most persistent failure points in automated generation is assuming LLM output will compile or meet business rules on the first pass.

In our Business Discovery POC build, when our code generation node attempted to create an in-memory dashboard, it repeatedly hallucinated unimported libraries and attempted to open persistent database connections on an ephemeral runtime.

The Generator-Critic pattern solves this by introducing a closed feedback loop:

Generator Node: Drafts the initial artifact (code, document, plan).
Evaluator / Critic Node: Runs multi-tier validation (e.g., deterministic AST static analysis, runtime mocking, semantic checks).
Conditional Routing: If validation passes, emit to downstream consumers. If validation fails, route structured diagnostics back to the Generator to self-correct.

Deterministic vs. Semantic Evaluation

A common mistake is using another LLM to grade every aspect of the generator's output. Semantic evaluation is slow, costly, and can hallucinate its own critique.

The most resilient agent systems combine deterministic validators (linters, AST checks, schema validators, sandbox test runs) with an LLM-based semantic evaluator:

import ast

def lint_generated_python(code_str: str) -> list[str]:
    """Deterministic AST pass to catch syntax and forbidden imports before LLM critique."""
    errors = []
    try:
        tree = ast.parse(code_str)
    except SyntaxError as e:
        return [f"SyntaxError on line {e.lineno}: {e.msg}"]

    banned_modules = {"sqlite3", "psycopg2", "mysql"}
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name.split(".")[0] in banned_modules:
                    errors.append(f"ForbiddenImport: `{alias.name}` is disallowed. Use in-memory state.")
    return errors

When deterministic linters catch an error, they format a raw traceback directly into the graph state. The Generator receives targeted diagnostic feedback rather than a vague prompt like "Please fix your mistakes."

3. Circuit Breakers & State Checkpointing

Cyclic graphs and self-healing loops introduce a new operational risk: infinite execution loops. If an agent cannot resolve a compilation error, it will burn through inference compute indefinitely.

Every production multi-agent system requires hard circuit breakers and durable state management:

1. Loop Counters & Graceful Degradation

Always attach an explicit iteration counter to the graph state. When the retry threshold is reached (e.g., iteration_count >= 3), route execution to a fallback node instead of crashing. This fallback can surface a partial artifact alongside the diagnostic log for human-in-the-loop review.

def should_continue(state: AgentState) -> str:
    if state["validation_status"] == "PASSED":
        return "deploy_node"
    if state["iteration_count"] >= 3:
        return "fallback_human_review_node"
    return "generator_node"

2. Upgrading Checkpointers for Concurrency

In local experimentation, LangGraph's ephemeral MemorySaver works fine. However, in enterprise microservices where requests are distributed across multiple worker nodes (e.g., Celery / FastAPI), in-memory checkpoints cannot be shared.

Upgrading to persistent state backends like PostgresSaver or RedisSaver ensures that:

Any available worker can resume execution on an active thread_id.
Human-in-the-loop approvals can pause state safely for hours or days without consuming memory.
If a worker node crashes mid-generation, the agent resumes execution from the last successful node checkpoint rather than restarting the entire workflow.

Pattern Comparison Matrix

Architecture Pattern	Best Used For	Latency Profile	Primary Failure Mode	Mitigation Strategy
Sequential Chain	Fixed, deterministic pipelines (ETL, step-by-step summaries)	Low / Predictable	Cascading error propagation	Strong schema validation between nodes
Orchestrator-Worker	Multimodal inputs, large document parsing, independent sub-queries	Medium (Parallelized)	Incomplete task decomposition	Structured schema outputs on Orchestrator
Generator-Critic	Code generation, complex reasoning, structured JSON extraction	Variable (Depends on retries)	Infinite retry loops & token burn	Strict retry counters + AST/Deterministic linters
Human-in-the-Loop Swarm	High-stakes workflows (Financial transactions, live DB migrations)	High (Awaits user approval)	Worker desynchronization & state drift	Distributed persistent checkpointers (`PostgresSaver`)

Key Takeaways

Stop overloading single prompts: Decompose complex tasks into specialized sub-agents. Give each worker minimal, isolated context to eliminate noise and reduce context window exhaustion.
Combine deterministic linters with LLM critics: Don't rely solely on LLMs to judge other LLMs. Use AST trees, unit tests, and schema validators to give generator nodes precise, actionable bug tracebacks.
Engineer for failure: Build loop counters, circuit breakers, and persistent checkpointer backends from day one so your agents degrade gracefully instead of looping infinitely.

As agent systems continue to evolve, the differentiator between fragile prototypes and production systems isn't model size-it is architecture, isolation, and deterministic control loops.

What multi-agent design patterns have you found most effective in production? Let me know in the comments or connect with me on GitHub/LinkedIn!

Building an Autonomous AI Agent: From Messy Client Artifacts to Live Prototypes

Shantanav Kapse — Tue, 18 Aug 2026 18:16:26 +0000

Just 5 days ago, a persistent bottleneck in custom software delivery and solutions engineering landed on my desk: the Business Discovery Phase. It is notoriously manual, fragmented, and time-consuming. Teams regularly spend days or weeks sifting through unstructured meeting recordings, chaotic chat exports, PDF transcripts, and legacy UI screenshots just to draft a basic requirements document and propose an initial Proof of Concept (POC).
I wanted to tackle this challenge directly: Could a local, multi-agent AI system compress this entire discovery-to-prototype lifecycle into just a few minutes?

I gave myself a strict sprint to build a working prototype from scratch. Here is what I engineered, the technical hurdles I hit along the way, and the architecture that emerged from the experiment.

The Naive Prototype

I started by designing a linear 5-node LangGraph pipeline running on local Ollama models, wrapped inside a Streamlit interface:

Ingest raw multimodal inputs (PDF notes, chat logs, screenshots).
Synthesize operational pain points and missing requirements.
Propose three architectural solutions.
Generate a single-file Streamlit web application.
Validate & Compile the code in a sandbox.

While the flow was clean conceptually, testing it against real-world inputs broke it immediately.

I was initially executing the generated Streamlit script using Python's exec() directly inside the parent dashboard's execution context. This triggered immediate React DOM tree collisions and duplicate widget ID errors (DuplicateKeyError), which crashed the parent app’s session state.

True Process Isolation

When an application's primary function is generating and hosting another interactive web application, in-process execution is a major failure point.
My first key breakthrough was completely isolating the generated application into a background subprocess on a dedicated port (8502).

Here is how I managed the lifecycle controls in main.py:

def launch_poc_subprocess(app_file: str, port: int = 8502):
    """Safely terminates old instances and spawns a new isolated background Streamlit process on port 8502."""
    stop_poc_subprocess()
    cmd = [
        sys.executable, "-m", "streamlit", "run", app_file,
        f"--server.port={port}",
        "--server.headless=true",
        "--browser.gatherUsageStats=false"
    ]
    new_proc = subprocess.Popen(
        cmd,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL
    )
    st.session_state["poc_subprocess"] = new_proc
    # Poll until child server responds with HTTP 200 OK
    for _ in range(15):
        time.sleep(0.3)
        try:
            with urllib.request.urlopen(f"http://localhost:{port}", timeout=1) as res:
                if res.status == 200:
                    break
        except Exception:
            pass

With the child process running independently, I embedded the live application into the parent UI using a clean st.iframe.

Ghost Widgets and Database Nightmares

As I pushed further, two deeper runtime issues emerged.
First, to verify that generated code would not crash prior to rendering, I needed a headless simulation harness. I wrote a MockStreamlit class and passed it into exec(). However, when the generated code executed import streamlit as st, Python's module loader resolved streamlit directly from sys.modules, overwrote the local mock, and rendered interactive "ghost widgets" across the parent UI during step execution!

Note: What is a "Ghost Widget"?
In this context, a ghost widget is an unintended, live UI component that accidentally renders across the main parent application's interface during background testing. This happens when the code execution bypasses local mocks and leaks into the global Python state, causing the background test to physically draw widgets on your screen instead of just simulating them in memory.

To fix this CPython import leak, I stripped out all Streamlit imports with regex before passing the string to exec():

class MockSessionState(dict):
    def __getattr__(self, key):
        return self.get(key)
    def __setattr__(self, key, value):
        self[key] = value

class MockStreamlit:
    def __init__(self):
        self.session_state = MockSessionState()
        self.sidebar = self
    def __getattr__(self, name):
        # Gracefully handle arbitrary Streamlit calls
        return lambda *a, **kw: None

# Strip imports prior to headless execution
stripped_code = re.sub(
    r'^(?:import\s+streamlit.*|from\s+streamlit\s+import.*)$',
    '# [stripped by testing agent]',
    poc_code,
    flags=re.MULTILINE
)

mock_globals = {"st": MockStreamlit(), "__name__": "__main__"}
exec(stripped_code, mock_globals)

The second issue was database lifecycles. The LLM repeatedly generated SQLite databases with top-level conn.close() calls, causing locked connections or wiping data on every Streamlit widget rerun. I established a strict Zero-Database Architecture, enforcing via prompt constraints that all transient data must live strictly in native st.session_state dictionaries and lists.

The Automated Self-Correction Loop

Once the generation was functioning, I noticed the model still occasionally hallucinated unimported dependencies like pd.DataFrame or timedelta.
Rather than relying on one-shot generation, I introduced a dedicated Testing & QA Agent (Node 6) into the LangGraph orchestration topology.
I implemented an AST (Abstract Syntax Tree) scanner to lint the generated code structure, flag forbidden database drivers, and detect missing imports. If the generated code fails AST verification, headless mock execution, or semantic outline auditing, the testing agent compiles a structured diagnostic report and routes the graph state back to the generator node for automatic self-healing.

# Flag any forbidden third-party DB / pandas imports
banned_db_modules = {"sqlite3", "sqlalchemy", "pymongo", "psycopg2", "mysql", "cx_Oracle", "pyodbc", "pandas"}

for node in ast.walk(tree):
    if isinstance(node, ast.Import):
        for alias in node.names:
            if alias.name.split(".")[0] in banned_db_modules:
                errors.append(f"ForbiddenImport: `import {alias.name}` -  store all data in st.session_state.")
    elif isinstance(node, ast.ImportFrom):
        if node.module and node.module.split(".")[0] in banned_db_modules:
            errors.append(f"ForbiddenImport: `from {node.module}` -  database libraries are forbidden.")

The Final Architecture & End-to-End Validation

The system ultimately evolved into a self-correcting 6-node multi-agent engine powered entirely by local open-weight models (qwen2.5:14b-instruct, qwen2.5-coder:14b-instruct, minicpm-v:latest) orchestrated via LangGraph and Qdrant.

To evaluate how well it generalized, I ran a benchmark session using an unstructured Fleet Dispatch & Maintenance Log scenario containing WhatsApp chats, scanned driver paper timesheets, and a meeting transcript PDF.
The Output:

Process Understanding: Identified that manual paper records were creating high vehicle downtime and delayed maintenance schedules.
Solution Proposal: Formulated a winning design for a Central Dispatcher Command Center & Automated Maintenance Scheduler.
Code Generation & Verification: Built a 180-line Streamlit application that passed static AST analysis, headless mock execution, and semantic QA checks on its first self-correction pass.
Deployment: Hosted the live application cleanly on port 8502 inside the studio's isolated iframe.

Key Takeaways

Subprocess Isolation is Essential for Meta-Apps: When building tools that generate and run other web applications, in-process execution will inevitably pollute global runtime state. Decoupling into isolated child processes and iframe embedding is the most resilient pattern.
Constraint-Driven Generation Reduces Failure Modes: Constraining code generation to an in-memory, zero-database state model (st.session_state) vastly improved reliability compared to allowing arbitrary third-party library imports.
Multi-Tier Testing is Non-Negotiable: Combining static AST checks, headless runtime mocking, and semantic LLM auditing creates a dependable self-healing loop for AI-generated code.

If you want to explore the implementation or run the pipeline locally, the repository is available here: https://github.com/shantanavKapse/poc_agent

Building Real-Time Dictation from Scratch: Escaping the Offline Trap

Shantanav Kapse — Sun, 09 Aug 2026 19:14:16 +0000

When the requirement first landed on my desk, it sounded simple enough: Build a real-time, cursor-based dictation feature. The user speaks into their browser, and the words appear live at the text cursor so they can draft long documents.

But there was a catch. The domain used highly specialized vocabulary, and the audio was strictly regulated. Sending the data to a slick cloud API was out of the question-audio could never leave our servers.

This is the story of my first deep dive into the trenches of speech-to-text (STT) engineering. Here is what I explored, the brutal obstacles I hit, and the architecture I ultimately shipped.

The Core Choice: Faking It vs. Native Streaming

My first architectural crossroad was choosing the model family. I initially looked at offline attention encoder-decoder models. They are famous for their high accuracy. However, they are trained to transcribe a complete audio file.

To use an offline model for "live" dictation, you have to fake it. You maintain a rolling window of audio (say, the last few seconds), and every 0.5 seconds, you re-transcribe the entire window. To decide what text to actually keep, you wait until two consecutive transcriptions agree on a word (local agreement) and then commit it.

I quickly realized this was a trap. It re-transcribes overlapping audio constantly, melting CPU cycles. Worse, when fed absolute silence, these offline models hallucinated wildly, inventing stock phrases or sign-offs. Because the hallucination was stable across passes, the system would permanently commit cascades of repeated ghost words.

The Pivot: I ditched the offline approach and moved to streaming neural transducers (the RNN-T family).

Transducers are built differently. They emit tokens as the audio arrives and have a built-in notion of "endpointing" (detecting a pause to mark the end of an utterance). They are natively real-time, mathematically stable, and most importantly, they don't hallucinate text out of thin air.

Key Takeaway: If you are building live dictation, force-fitting an offline model fights the use case. A streaming transducer is the right tool for the job.

Nailing the UX: The Two-State Contract

To make the live cursor feel magical rather than jarring, I needed a rock-solid UI contract. A jumping, flickering text cursor is incredibly frustrating for users drafting long documents.

I implemented a two-state output contract where every WebSocket update carries two fields: {committed, buffer}.

Committed: Stable text the client appends permanently (rendered in a normal font weight).
Buffer: The model's current provisional guess (rendered in grey or italics, replacing itself on every update). Because the streaming model handles endpointing naturally, it fires off the final text when the user pauses. The client promotes the buffer to committed, and the stream resets. The stable text never jumps backward, and the provisional tail visibly settles.

Obstacle 1: The Silent WebSocket Killer

Getting the browser to capture audio was straightforward. The browser records microphone audio as Opus-compressed chunks and streams them over a WebSocket every ~250 milliseconds. The server then decodes this into the raw 16 kHz mono PCM the model requires.

Then came the hardest engineering bug of the project.

The dictation would work perfectly for a sentence or two, and then the WebSocket would randomly disconnect. No errors, just a dropped connection.

I traced it to a concurrency and back-pressure failure. Initially, a single task was both reading the decoded audio and running the STT model inline. While the model was busy crunching numbers, nothing was draining the audio decoder's output pipe. The pipe filled up, the decoder stopped pulling input, the upstream socket blocked, and the WebSocket receive loop stalled. The keepalive ping/pong stopped, and the connection died.

The Fix: I decoupled the pipeline into two isolated tasks.

A Reader: Continuously drains the decoder into an in-memory buffer so the pipe never blocks.
A Processor: Consumes that buffer and runs the model at its own pace.

Once decoupled, back-pressure never reached the socket.

Obstacle 2: Formatting and Domain Jargon

Documents need capital letters and periods. Many streaming models emit normalized text (lowercase, no punctuation).

I had two choices: run a secondary punctuation-restoration model on the committed text, or find a model that handles casing natively. I opted for the latter. By checking the model's token vocabulary for mixed-case pieces and punctuation tokens, I bypassed the need for a secondary formatting layer entirely.

Handling specialized vocabulary was trickier. General models stumble on niche jargon. While contextual biasing (word boosting) is great, the lightweight runtime for my streaming model only supported greedy decoding, meaning acoustic biasing was unavailable.

I solved this pragmatically with a Post-ASR Correction Map-a simple dictionary of {misheard -> correct} terms applied to the committed text. It isn't as elegant as acoustic biasing, but it reliably catches repeatable mistakes based on actual user data.

The Deployment Reality: CPU over GPU

You might assume a real-time AI model requires massive GPU power. In reality, the streaming transducer is incredibly efficient.

The Real-Time Factor (RTF) is processing time divided by audio duration. For a ~0.6 billion parameter streaming model, I measured an RTF of roughly 0.09 on a standard laptop CPU. That is 11x real-time headroom.

This led to a counter-intuitive deployment strategy:

Keep it off the GPU: Co-locating this STT model with a large LLM on a single GPU is a bad idea (I already had an LLM deployed on the server, but the CPU was not utilized). The bursty generation of the LLM will stall the latency-sensitive audio stream, causing audible stuttering.
Run on CPU: The ASR runs comfortably on spare CPU cores. I wrapped it in a container with hard CPU and memory limits so it couldn't physically exceed its lane or trigger an Out-Of-Memory (OOM) crash on the server.

Final Thoughts

Building this from scratch taught me that the hardest parts of applied AI often aren't the neural networks themselves-they are the plumbing, the back-pressure, and the UX contracts.

Trade-off	Cloud API	Local Streaming Transducer (My Build)
Data Privacy	Audio leaves premises	Stays entirely on-prem
Cost	Per-minute billing	Free / Compute only
Accuracy / Drift	Vendor-managed, highly tuned	Requires manual correction maps
Latency	Network dependent	Near-instant (RTF < 0.1 on CPU)

Choosing to self-host a streaming model meant giving up the automatic updates and easy jargon-boosting of a cloud API. But for a privacy-regulated environment, it was the only honest trade. We achieved true real-time, zero-cost, private dictation-and the cursor feels just right.