DEV Community: Shantanav Kapse

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.