Forem: Amito Vrito

Why I Modelled My LLM Pipeline as a DAG Instead of a Chain — and What I Found Out

Amito Vrito — Thu, 16 Apr 2026 09:27:58 +0000

The problem with chains in production

Every major Python LLM framework gives you the same primitive: a chain.

LangChain's LCEL. LlamaIndex's pipeline. Haystack's components. They all model your pipeline as a linear sequence of steps — input flows through A, then B, then C, output comes out the end.

For a hello-world RAG demo, that's fine. For a production system, you hit the wall fast.

What chains can't express cleanly

Here's a real pipeline I needed to build:

Classify the incoming query
Based on classification: route to either semantic search, keyword search, or both in parallel
If both: merge and re-rank results
Generate response from ranked context
If any retrieval stage fails: surface a clear error, don't silently continue

Try expressing that as a chain. You end up with nested chains, manual asyncio.gather() calls outside the framework, try/except blocks swallowing exceptions to keep the chain going, and no clean way to express "step D depends on both B and C."

The abstraction is fighting you.

DAGs are the right model

A directed acyclic graph expresses all of this naturally.

Nodes are tasks. Edges are data dependencies. Execution is topologically ordered — a node fires when all its upstream dependencies have resolved.

from synapsekit import Pipeline

pipeline = Pipeline()
pipeline.add_node("classify", ClassifierNode())
pipeline.add_node("semantic_search", RAGNode(store=vector_store))
pipeline.add_node("keyword_search", BM25Node(index=bm25_index))
pipeline.add_node("rerank", RerankerNode())
pipeline.add_node("generate", LLMNode(model="gpt-4o"))

pipeline.add_edge("classify", "semantic_search")
pipeline.add_edge("classify", "keyword_search")
pipeline.add_edge("semantic_search", "rerank")
pipeline.add_edge("keyword_search", "rerank")
pipeline.add_edge("rerank", "generate")

result = await pipeline.run(query="explain async-native design")

semantic_search and keyword_search run concurrently. rerank waits for both. generate waits for rerank. The execution engine handles ordering. You describe the dependencies.

Failure propagation that actually works

In a chain, a failed step either kills the whole pipeline or gets swallowed silently.

In a DAG, failure has semantics. If semantic_search fails, rerank — which depends on it — is cancelled. generate — which depends on rerank — is also cancelled. You get a clear error naming the failed node and its dependents.

No silent degradation. Failure is explicit and traceable.

The async piece

Nodes with no dependency relationship between them run concurrently automatically. The execution engine handles asyncio.gather() at each topological level. You write individual async node functions. The graph handles orchestration.

class SemanticSearchNode(Node):
    async def execute(self, inputs):
        results = await self.vector_store.asearch(inputs["query"])
        return {"results": results}

No manual gather calls. The graph structure encodes the parallelism.

Is it worth the complexity?

For simple A → B → C pipelines with no branching: a chain is fine.

The moment you have parallel retrieval, conditional routing, or stages where failure isolation matters — the chain abstraction costs more than it saves.

SynapseKit is the framework I built around this model:
https://github.com/SynapseKit/SynapseKit
API Doc: https://synapsekit.github.io/synapsekit-docs/

10k PyPI downloads since launch. The engineers who need this know exactly why.

What does your production RAG architecture look like? Drop it in the comments.

Introducing SynapseKit: The Async-Native Python LLM Framework I Built Because LangChain's Async is Broken

Amito Vrito — Wed, 15 Apr 2026 11:44:29 +0000

Article:

Why I built another LLM framework

I know. Another one.

But hear me out — because the reason I built SynapseKit is specific, and it might be the same reason you've been frustrated too.

The async lie in Python LLM frameworks

LangChain has async support. LlamaIndex has async support. Haystack has async support.

Technically true. Practically — look at the source.

You'll find asyncio.get_event_loop().run_in_executor() wrapping sync functions. You'll find internal blocking IO disguised behind async def. You'll find ThreadPoolExecutor doing the actual work.

That's not async-native. That's sync code wearing an async costume.

For simple scripts and demos, it doesn't matter. For production services handling concurrent requests — FastAPI services, real-time RAG systems, high-throughput agent workflows — it matters enormously. You're paying the cost of threads AND the overhead of the async event loop, with none of the actual throughput benefits.

What SynapseKit does differently

I built the async layer first. Every IO operation — LLM calls, retrieval, embedding generation — is genuinely non-blocking from the ground up. There's no sync wrapper underneath.

import asyncio
from synapsekit import Pipeline, RAGNode, LLMNode

async def main():
    pipeline = Pipeline()
    pipeline.add_node("retrieve", RAGNode(vectorstore=my_store))
    pipeline.add_node("generate", LLMNode(model="gpt-4o"))
    pipeline.add_edge("retrieve", "generate")

    result = await pipeline.run(query="What is async-native design?")
    print(result)

asyncio.run(main())

Notice: no .run_in_executor(). No thread pool. Just async.

DAGs, not chains

The second architectural decision: pipelines are directed acyclic graphs, not linear chains.

Every major framework pushes you toward .pipe() or | operator chains. That works for the happy path. Production systems aren't the happy path.

In a real RAG system you might:

Retrieve from multiple sources in parallel
Route to different generation strategies based on query classification
Have fallback paths when a retrieval stage fails
Run a re-ranking step that depends on two upstream retrievers

A chain can't express that cleanly. A DAG can.

pipeline.add_node("classify", ClassifierNode())
pipeline.add_node("retrieve_docs", RAGNode(vectorstore=doc_store))
pipeline.add_node("retrieve_web", WebSearchNode())
pipeline.add_node("rerank", RerankerNode())
pipeline.add_node("generate", LLMNode())

pipeline.add_edge("classify", "retrieve_docs")
pipeline.add_edge("classify", "retrieve_web")
pipeline.add_edge("retrieve_docs", "rerank")
pipeline.add_edge("retrieve_web", "rerank")
pipeline.add_edge("rerank", "generate")

Both retrieval nodes run concurrently. The reranker waits for both. The LLM waits for the reranker. Topological ordering handles execution automatically.

The numbers

~10,000 PyPI downloads in ~20 days of active development. No Product Hunt. No HN. No launch post.

Developers found it through PyPI search. That told me the demand is real.

Try it

pip install synapsekit

GitHub: https://github.com/SynapseKit/SynapseKit

If you've been frustrated with async in your LLM stack — or you're building something where throughput actually matters — I'd genuinely love your feedback.

And if this resonates, a GitHub star helps surface it to other developers who are hitting the same walls.

SynapseKit is open source under Apache license. Built by, Senior AI Specialists and founder of EngineersOfAI.