DiwashBhandari

Software Engineer — Python, FastAPI, AI/ML & Distributed Systems

Architect production-grade backend systems, multi-tenant SaaS platforms, and AI-powered applications. 5+ years shipping scalable APIs, RAG pipelines, and secure licensing infrastructure.

5+ Years Exp.
15+ APIs Deployed
5+ Projects
About Me

Software Engineer & Exploring AI Systems

Building reliable backend systems and thoughtful AI applications that just work.

I build scalable backend systems, AI-powered applications, and multi-tenant SaaS platforms that ship fast and scale reliably.

Focused on high-performance APIs, AI-powered applications, and multi-tenant SaaS infrastructure — built with Python, FastAPI, PostgreSQL, Redis, and modern AI/ML tooling.

Backend Architecture & API DesignCloud Telephony & CRM IntegrationsMicroservices & Distributed SystemsAI/LLM & RAG ApplicationsEvent-Driven Architecture & Real-Time SystemsPerformance OptimizationDevOps & CI/CD

15+

APIs Deployed

5+

Projects

90%

Test Coverage

5+

Years Exp.

Experience

Where I've Worked

Building scalable systems and leading technical initiatives across diverse industries.

4 Companies

Career progression

May 2026 - Present

December 2020 — Present

2020202220232026Present

Senior Software Engineer

@ Freelance

May 2026 - Present
Remote
Freelance / Consulting
Key Impact

Architecting and delivering production-grade backend systems across warehouse management, large-scale member management, software licensing, and AI-powered customer support engagements.

Stack:
PythonFastAPIPostgreSQLRedisDockerAI/MLWebSockets
Notable Achievements
  • Delivered multi-tenant warehouse management backend with real-time slot locking and operational dashboards.
  • Built large-scale member management platform supporting ~30M records with RBAC, caching, and ML-powered recommendations.
  • Designed commercial licensing server and client-side enforcement SDK with Ed25519 signatures and offline grace periods.
  • Shipped LangGraph-based ISP support platform with hybrid RAG, OTP-gated plan changes, and QR-code payments.
Key Responsibilities
  • Design and develop scalable Python and FastAPI services using PostgreSQL, Redis, WebSockets, asynchronous processing, cloud infrastructure, and containerized deployment patterns.
  • Own end-to-end engineering work across system architecture, API design, database modeling, integrations, security controls, CI/CD, observability, deployment, and production support.
  • Build multi-tenant, high-volume, real-time, and security-sensitive systems with emphasis on reliability, maintainability, performance, and operational visibility.
  • Implement external-system integrations, background processing, caching, authentication, authorization, cryptographic verification, and automated workflows based on product and operational requirements.
  • Collaborate with stakeholders and engineering teams to translate business requirements into production-ready systems while maintaining code quality, security, and delivery standards.
Work

Featured Projects

Production-grade applications showcasing scalable architecture and AI integration.

FeaturedBackend Systems | Warehouse Operations

Warehouse Management System (WMS)

Multi-Tenant Warehouse Backend — Slot Locking, Picklists, ASN/LPN, SFTP Sync & Real-Time Dashboards

A multi-tenant warehouse backend that coordinates receiving, inventory, outbound picking, slot operations, and loading across tenants without conflicting updates. Real-time slot locking and reservation sync keep warehouse-floor state consistent. ASN/LPN and picklist workflows cover the full item lifecycle, while SFTP and CSV integrations sync with external systems. Operational dashboards provide visibility into traffic, pick logs, and audit history.

Impact

Reduced conflicting warehouse-floor operations through tenant-isolated PostgreSQL schemas and Redis Pub/Sub with WebSockets for real-time slot locking, reservations, and state synchronization.

Inbound + Outboundworkflows automated
SFTP, CSV, APIintegration types
Sentrymonitoring
FeaturedBackend Systems | Enterprise SaaS

Member Management System (MMS)

Large-Scale Member Platform — 30M Records, Lifecycle Workflows, RBAC, Caching & Store Recommendations

A large-scale member management platform for a loyalty and club-management program, handling onboarding, lifecycle changes, card transactions, and club summaries across very large datasets. Efficient pagination and caching keep high-volume queries responsive. Member events are tracked with full audit history, while role-based access control and secure document storage maintain compliance. External sync adapters and a location-recommendation component support operational integration and store planning.

Impact

Improved query scalability for data-intensive workloads through high-volume member CRUD, filtering, club summaries, card transaction history, and keyset pagination for large PostgreSQL tables.

~30Mrecords supported
Rediscaching
S3storage

More Projects

Backend Systems

Guard License Server

Commercial Software Licensing — Ed25519 Signatures, Node-Locked & Floating Licenses, Encrypted Storage

PythonFastAPIPostgreSQLSQLite
Backend Systems

Guard Trust — License Enforcement SDK

Client-Side License SDK — Offline Verification, Machine Binding, Background Monitoring & Nuitka Protection

PythonFastAPIEd25519AES-256-GCM
AI Support Platform

ISP Customer Support Platform

LLM-Powered ISP Support — LangGraph Agent, Hybrid RAG, QR Payments, Ticketing & Live Dashboard

PythonFastAPILangGraphUpstash Vector
Expertise

Technical Proficiency

PythonGoJavaScript
Blog & Writing

Tech Insights & Articles

In-depth thoughts on backend development, AI implementation, Python optimization, and the future of technology.

FeaturedfastapiAugust 19, 2026
11 min read

Transaction Outbox Pattern: Reliable Messaging in Distributed Systems with FastAPI, SQLAlchemy &…

Transaction Outbox Pattern: Reliable Messaging in Distributed Systems with FastAPI, SQLAlchemy & RabbitMQ Master the pattern that guarantees “zero lost events” in your microservices without 2PC, with production-ready code and diagrams. In distributed systems, ensuring atomicity between database transactions and message publishing is a critical challenge. The Transaction Outbox Pattern solves this by persisting messages in the same database transaction, then publishing them through a separate relay. This blog post provides a complete guide with working FastAPI + SQLAlchemy + RabbitMQ code, architectural diagrams, real-world use cases, common pitfalls, and step-by-step implementation. Transaction Outbox Pattern DiagramThe Problem Imagine you’re building an e-commerce service. When a user places an order: You save the order to the database ✅ You publish an OrderCreated event to RabbitMQ ✅ But what happens if your service crashes between steps 1 and 2? # ❌ Naive approach — dangerousorder = Order(...)db.add(order)db.commit() # Step 1: order savedpublisher.send(OrderCreated(order_id)) # Step 2: CRASH happens here!# Result: Order exists, but nobody knows about it. Inventory won't be reserved.# Email won't be sent. Analytics won't track the sale. Total chaos. The database transaction and the message broker send are not atomic. A crash creates a consistency gap. Why Not Just Use Distributed Transactions? Distributed transactions (2PC/XA) are: Slow (network round-trips) Couples your service to the broker Not supported by most modern databases/brokers A scalability bottleneck We need a better solution. What Is the Transaction Outbox Pattern? The Transaction Outbox Pattern ensures atomicity between database writes and message publication — without distributed transactions. Core Idea Instead of sending the message directly to the broker, you write it to a local database table (the “outbox”) within the same transaction as your business data. Then, a separate background process (the relay) polls the outbox table and publishes messages to the broker. Why It Works If the transaction commits ✅ → the outbox message is guaranteed to exist in the database. If the transaction rolls back ✅ → no outbox message is persisted. Even if the relay crashes ✅ → messages survive in the database, waiting to be published. This achieves at-least-once delivery with zero message loss. Pro tip: Include a message_id in every outbox message. Consumers use this to deduplicate — "Have I already processed message_id abc-123?" If yes, skip. Architecture Overview Diagram ┌──────────────────────────────┐ │ FastAPI Service (Sender) │ │ │ │ async with session.begin(): │ │ INSERT INTO orders │ ◀── business data │ INSERT INTO outbox_messages│ ◬── message in same TX │ # atomic commit/rollback │ └──────────────┬───────────────┘ │ ┌────────────────────┴────────────────────────┐ │ PostgreSQL │ │ │ │ orders outbox_messages │ │ +------+ +------------------+ │ │ |order | | message_id (pk) | │ │ +------+ | aggregate_id | │ │ | aggregate_type | │ │ | event_type | │ │ | payload (jsonb) | │ │ | status = PENDING | │ │ | message_id | │ │ | created_at | │ │ | sent_at | │ │ +------------------+ │ └──────────────────────┬────────────────────┘ 1. polls unsent rows ▼ ┌─────────────────────────────┐ │ Relay Worker (background)│ │ │ │ SELECT * WHERE │ │ status='PENDING' │ │ ORDER BY created_at │ │ LIMIT 100 │ │ publish to RabbitMQ │ │ UPDATE status='SENT' │ └────────────┬──────────────┘ 2. dedup by message_id on retry ▼ ┌────────────────────────────┐ │ RabbitMQ Broker │ │ │ │ exchange: order.events │ │ routing key: order.created│ │ queue: inventory_update │ │ queue: email_worker │ │ queue: analytics_sink │ └────────────────────────────┘ ▼ ┌────────────────────────────┐ │ Consumer Services │ │ (idempotent) │ └────────────────────────────┘ Key Guarantee: At-Least-Once Delivery If the relay crashes after publishing but before marking the message as sent, the same message will be retried on the next poll. That’s at-least-once. Consumers must handle this with idempotent operations. Pro tip: Include a message_id in every outbox message. Consumers use this to deduplicate — "Have I already processed message_id abc-123?" If yes, skip. Implementation Guide Prerequisites pip install fastapi[all] sqlalchemy asyncpg aio-pika python-uuid psycopg2-binary 1. Database Schema CREATE TABLE orders ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), customer_id UUID NOT NULL, product_id UUID NOT NULL, quantity INT NOT NULL CHECK (quantity > 0), total_cents INT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now());CREATE TABLE outbox_messages ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), aggregate_id UUID, aggregate_type VARCHAR(100), event_type VARCHAR(200), payload JSONB, status VARCHAR(20) DEFAULT 'PENDING', message_id UUID DEFAULT gen_random_uuid(), created_at TIMESTAMPTZ NOT NULL DEFAULT now(), sent_at TIMESTAMPTZ);-- Index for fast polling of pending messagesCREATE INDEX idx_outbox_pending ON outbox_messages (status, created_at, id) WHERE status = 'PENDING'; 2. SQLAlchemy Models (models.py) import uuidfrom datetime import datetime, timezonefrom sqlalchemy import DateTime, funcfrom sqlalchemy.ext.asyncio import AsyncAttrsfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_columnfrom sqlalchemy.types import UUIDfrom sqlalchemy import Indexclass Base(AsyncAttrs, DeclarativeBase): passclass Order(Base): __tablename__ = "orders" __table_args__ = (Index("idx_order_customer", "customer_id"),) id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) customer_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True)) product_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True)) quantity: Mapped[int] total_cents: Mapped[int] created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() )class OutboxMessage(Base): __tablename__ = "outbox_messages" __table_args__ = ( Index("idx_outbox_pending", "status", "created_at", "id"), ) id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) aggregate_id: Mapped[uuid.UUID] aggregate_type: Mapped[str] event_type: Mapped[str] payload: Mapped[dict] status: Mapped[str] = "PENDING" message_id: Mapped[uuid.UUID] = mapped_column(default=uuid.uuid4) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() ) sent_at: Mapped[datetime | None] = None 3. FastAPI App (main.py) import uuidfrom datetime import datetime, timezonefrom fastapi import Depends, FastAPIfrom pydantic import BaseModelfrom sqlalchemy import insertfrom sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine,)from models import Base, Order, OutboxMessageDATABASE_URL = "postgresql+asyncpg://user:password@localhost/mydb"RABBITMQ_URL = "amqp://guest:guest@localhost/"engine = create_async_engine(DATABASE_URL, echo=False)async_session = async_sessionmaker(engine, expire_on_commit=False)app = FastAPI()# ── Dependency ───────────────────────────────────────────────async def get_session() -> AsyncSession: async with async_session() as session: yield session# ── Pydantic Schemas ─────────────────────────────────────────class OrderCreate(BaseModel): customer_id: uuid.UUID product_id: uuid.UUID quantity: int total_cents: int# ── Endpoints ────────────────────────────────────────────────@app.post("/orders")async def create_order(req: OrderCreate, session: AsyncSession = Depends(get_session)): order_id = uuid.uuid4() message_id = uuid.uuid4() # ⚡ THE KEY: business data + outbox message in ONE transaction async with session.begin(): await session.execute( insert(Order).values( id=order_id, customer_id=req.customer_id, product_id=req.product_id, quantity=req.quantity, total_cents=req.total_cents, ) ) await session.execute( insert(OutboxMessage).values( aggregate_id=order_id, aggregate_type="order", event_type="OrderCreated", payload={ "order_id": str(order_id), "customer_id": str(req.customer_id), "product_id": str(req.product_id), "quantity": req.quantity, "total_cents": req.total_cents, "created_at": datetime.now(tz=timezone.utc).isoformat(), }, message_id=message_id, ) ) # ✅ If we return 200, the message IS in the database - guaranteed. # The relay will publish it to RabbitMQ shortly. return {"order_id": str(order_id), "message_id": str(message_id)}@app.on_event("shutdown")async def shutdown_event(): await engine.dispose()# ── Startup ──────────────────────────────────────────────────@app.on_event("startup")async def startup(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) Critical: The entire operation is wrapped in async with session.begin(): — if either the INSERT INTO orders or INSERT INTO outbox_messages fails, both are rolled back. No partial state. 4. Relay Worker (workers/outbox_relay.py) """Polling Publisher Relay — runs as a separate background process.Polls the outbox table and publishes messages to RabbitMQ."""import asyncioimport jsonimport loggingimport osimport uuidimport aio_pikafrom sqlalchemy import select, update, funcfrom sqlalchemy.ext.asyncio import create_async_engine, async_sessionmakerfrom models import Base, OutboxMessageDATABASE_URL = os.getenv( "DATABASE_URL", "postgresql+asyncpg://user:password@localhost/mydb")RABBITMQ_URL = os.getenv("RABBITMQ_URL", "amqp://guest:guest@localhost/")BATCH_SIZE = 100POLL_INTERVAL = 1.0 # secondsengine = create_async_engine(DATABASE_URL)async_session = async_sessionmaker(engine, expire_on_commit=False)logging.basicConfig(level=logging.INFO)log = logging.getLogger("outbox-relay")async def fetch_pending(session, limit: int = BATCH_SIZE): """SELECT PENDING messages ordered by created_at (preserves order).""" result = await session.execute( select(OutboxMessage) .where(OutboxMessage.status == "PENDING") .order_by(OutboxMessage.created_at, OutboxMessage.id) .limit(limit) ) return result.scalars().all()async def mark_sent(session, message_ids: list[uuid.UUID]): """Mark a batch as SENT atomically.""" await session.execute( update(OutboxMessage) .where(OutboxMessage.id.in_(message_ids)) .values(status="SENT", sent_at=func.now()) ) await session.commit()async def run_relay(): connection = await aio_pika.connect_robust(RABBITMQ_URL) async with connection: channel = await connection.channel() exchange = await channel.declare_exchange( "order.events", aio_pika.ExchangeType.TOPIC, durable=True ) async with async_session() as session: while True: try: messages = await fetch_pending(session) if not messages: await asyncio.sleep(POLL_INTERVAL) continue sent_ids: list[uuid.UUID] = [] for msg in messages: try: body = json.dumps(msg.payload).encode() # Publish with message_id as a header for consumer idempotency await exchange.publish( aio_pika.Message( body=body, message_id=str(msg.message_id), content_type="application/json", delivery_mode=aio_pika.DeliveryMode.PERSISTENT, ), routing_key=f"order.{msg.event_type.lower()}", ) sent_ids.append(msg.id) log.info(f"Published {msg.event_type} to RabbitMQ") except Exception: log.exception( f"Failed to publish outbox msg {msg.id}, " f"leaving as PENDING for retry" ) await session.rollback() break # Retry same batch next poll if sent_ids: await mark_sent(session, sent_ids) except Exception: log.exception("Relay error, retrying in 5s...") await asyncio.sleep(5)if __name__ == "__main__": asyncio.run(run_relay()) 5. Consumer (Idempotent) Example # consumers/email_service.py"""Consumer service that sends order confirmation emails.MUST BE idempotent — handles duplicate messages safely."""import jsonimport uuidimport aio_pikafrom sqlalchemy import textasync def handle_message(message: aio_pika.IncomingMessage): async with message.process(): msg = message.body.decode() payload = json.loads(msg) # 1. Dedup: check if message_id already processed result = await db.fetch_one( text("SELECT 1 FROM processed_messages WHERE message_id = :mid"), {"mid": str(message.message_id)}, ) if result: # Already processed - safe to skip return # 2. Process the event order_id = payload["order_id"] customer_id = payload["customer_id"] await send_order_confirmation_email(customer_id, order_id) # 3. Record processing (idempotency store) await db.execute( text( "INSERT INTO processed_messages (message_id, processed_at) " "VALUES (:mid, now()) ON CONFLICT (message_id) DO NOTHING" ), {"mid": str(message.message_id)}, )async def main(): connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/") async with connection: channel = await connection.channel() exchange = await channel.declare_exchange( "order.events", aio_pika.ExchangeType.TOPIC, durable=True ) queue = await channel.declare_queue("order.email", durable=True) await queue.bind(exchange, routing_key="order.ordercreated") async with queue.iterator() as queue_iter: async for message in queue_iter: await handle_message(message) 6. Docker Compose version: "3.9"services: postgres: image: postgres:16 environment: POSTGRES_USER: user POSTGRES_PASSWORD: password POSTGRES_DB: mydb ports: - "5432:5432" rabbitmq: image: rabbitmq:3-management ports: - "5672:5672" - "15672:15672" app: build: . command: uvicorn main:app --host 0.0.0.0 --port 8000 depends_on: [postgres, rabbitmq] relay: build: . command: python workers/outbox_relay.py depends_on: [postgres, rabbitmq] Real-World Use Cases Here are the scenarios where the Transaction Outbox Pattern shines: Real- Word Use CasesCase Study: E-commerce Order Pipeline # Order service — single transaction, multiple consumers benefitasync with session.begin(): order = Order(...) session.add(order) # Outbox message for ALL downstream services session.add(OutboxMessage( aggregate_type="order", event_type="OrderCreated", payload={"order_id": order.id, "total": order.total}, message_id=uuid.uuid4() ))# Downstream: all get notified via RabbitMQ topic exchange# inventory.update.order.created# email.order.confirmation# analytics.order.event# shipping.schedule.order.created Common Mistakes and Pitfalls ❌ 1. Not Making Consumers Idempotent Problem: The relay publishes the same message twice (crash after publish, before marking sent). Consumer processes it twice — two emails, double-charged inventory. Solution: # Always dedup by message_idasync def handle_order_created(event): existing = await db.fetch_one( "SELECT id FROM processed_events WHERE message_id = :mid", {"mid": event.message_id} ) if existing: return # Skip duplicate # ... process ❌ 2. Forgetting to Clean Up the Outbox Table Problem: Outbox table grows indefinitely. Query performance degrades. Solution: -- Nightly cleanup jobDELETE FROM outbox_messages WHERE status = 'SENT' AND sent_at < now() - interval '7 days'; ❌ 3. Sending Messages After Commit (Breaking the Pattern) Problem: Some developers move the message send outside the transaction for “performance.” # ❌ WRONG — breaks atomicityasync with session.begin(): session.add(order)# Transaction committedawait publisher.send(OrderCreated(order.id)) # Can crash here! Solution: Keep everything in the same transaction — that’s the whole point. ❌ 4. No Error Handling in the Relay Problem: Relay crashes on a malformed message, stops processing everything. Solution: try: await exchange.publish(message)except Exception: await session.rollback() # Keep message as PENDING log.error(f"Failed: {msg.id}") continue # Keep processing other messages ❌ 5. No Dead Letter Queue for Poison Messages Problem: A message that can never be published (e.g., malformed payload) blocks the relay forever. Solution: DEAD_LETTER_THRESHOLD = 5# Track retry countif msg.retry_count >= DEAD_LETTER_THRESHOLD: await db.execute( "UPDATE outbox_messages SET status='DEAD_LETTER' WHERE id=:id", {"id": msg.id} ) ❌ 6. Not Indexing the Outbox Table Problem: SELECT WHERE status='PENDING' becomes a full table scan on a 10M-row table. Solution: CREATE INDEX idx_outbox_pending ON outbox_messages (status, created_at, id)WHERE status = 'PENDING'; Advanced Considerations Scaling the Relay For high-throughput systems, a single polling relay may not be enough: -- Use SKIP LOCKED for fair distribution across relay workersSELECT * FROM outbox_messagesWHERE status = 'PENDING'ORDER BY created_atLIMIT 100FOR UPDATE SKIP LOCKED; -- ← Multiple workers can claim different batches CDC-Based Relay (No Polling) At very high throughput, replace polling with Change Data Capture using Debezium: PostgreSQL WAL → Debezium CDC → Kafka → RabbitMQ Connector No polling, no latency, but added operational complexity. Exactly-Once Semantics The outbox gives at-least-once. For exactly-once, you need: Outbox + idempotent consumers (at-least-once) OR Kafka transactions (end-to-end exactly-once, but only applies to Kafka, not RabbitMQ) Choose based on your broker and requirements. Summary | Aspect | Description || ------------- | --------------------------------------------------------------------------------------------------------------------------------------- || **What** | Store the event message in a local **Outbox table** in the same database transaction as the business data. || **Why** | Prevents message loss between the database commit and message broker publishing. || **How** | A separate **relay worker** polls the Outbox table, publishes pending messages to the broker, and marks them as successfully published. || **Guarantee** | Provides **at-least-once delivery** without requiring distributed transactions such as 2PC. || **Trade-off** | Consumers must be **idempotent** because a message can be delivered more than once. The Outbox table also needs cleanup or archival. || **Best For** | Microservices that need to reliably publish events after changing local database state. |

Read Article
02
ai-workflowAug 03, 2026
32 min

Loop Engineering vs Graph Engineering: Why Every AI Engineer Should Learn Them in 2026

Two ways to structure AI workflows — and when to use each one. Learn loop engineering vs graph engineering in 2026. See how retry loops, state machines, LangGraph, and CrewAI shape modern AI agents and workflows. Loop Engineering vs Graph EngineeringTable of Contents Introduction What Is Loop Engineering? What Is Graph Engineering? Loop Engineering vs Graph Engineering Why These Skills Matter in 2026 Building Loop Engineering in Python Building Graph Engineering in Python Frameworks That Support Loop and Graph Engineering Real-World Use Cases Common Mistakes Learning Roadmap for 2026 Should You Learn These Skills in 2026? Conclusion 1. Introduction For the past few years, “AI engineering” mostly meant prompt engineering. You wrote a good prompt, the model produced a good answer, and you moved on. That worked well for chatbots and one-shot tasks. It does not work for agents. Agents act. They call tools, run code, check results, and try again when something fails. That is execution logic, not just text generation. The prompt still matters, but it is only the input to a process. The process itself — the steps, the decisions, the retries — is where the real engineering happens. This shift is why job descriptions for AI engineers now mention workflows, orchestration, and agent design. Companies do not hire someone just to write prompts anymore. They hire people who can build a system that takes a task, executes it, verifies the result, and recovers when things go wrong. Two ideas sit at the center of that work: loop engineering and graph engineering. Both are ways to structure how an LLM-based system executes. They answer different questions. Loop engineering answers “how do we repeat work until it is good enough?” Graph engineering answers “how do we lay out the steps, branches, and decisions of a workflow?” Most production AI systems use both. In this article you will learn what each one is, how to build them in Python, which frameworks support them, and when to reach for each. You will also see real code you can adapt. By the end, you should be able to look at any AI workflow and name the loops and the graphs inside it. Key takeaway: Prompt engineering tells the model what to do. Loop and graph engineering tell the system how to get it done. In 2026, the second skill is what separates apps from agents. 2. What Is Loop Engineering? Loop engineering is the practice of structuring AI work as repeated execution with feedback. You run a step, check the output, and if it is not good enough, run it again with the previous errors included. The core concept A single LLM call is one-shot. You send a prompt, you get a completion, and it is done. A loop turns that one call into a process: Ask the model to do the work. Verify the result (tests, a validator, a second model, a human). If it passes, stop. If it fails, ask again with the failure details added to the prompt. Repeat until it passes or the budget runs out. The feedback is the secret. When you retry without feedback, you get the same guess twice. When you retry with feedback — the test output, the error message, the reviewer’s notes — the model has new information and can improve. Why iterative execution matters Models are probabilistic. On any given call, the model can miss a step, invent a wrong API name, or produce output that does not parse. For a chatbot answer, a miss is a minor annoyance. For an agent that writes to a database or edits a file, a miss is a bug. Iteration converts unreliable single calls into reliable processes. The system stays dumb-simple on each step, but the loop makes the overall behavior dependable. This is the same reason compilers, linters, and CI pipelines exist in normal software: you verify work and feed failures back. Types of loops you will meet Not every loop retries. Here are the common shapes: Loop typeWhat it doesExampleRetry loopRe-run a step after a failureCall a flaky API again after a timeoutReflection loopAsk the model to critique its own output”Review your plan, then improve it”Validation loopCheck output against rules, retry if invalidParse JSON until it is validEvaluation loopScore output against a rubric, retry if lowRewrite copy until it scores above 0.8Self-correction loopFeed errors back and let the model fix themAgent edits code until tests pass These are not exclusive. A self-correction loop usually contains a validation loop. You will combine them in practice. Real-world analogy Think of a junior engineer working through a code review. The engineer writes a fix. A senior engineer reviews it and says “this leaks the connection” or “this test fails on Windows.” The junior engineer reads the feedback, changes the code, and submits again. This repeats until the review passes or the reviewer gives up. The review feedback is the loop’s error message. The number of review rounds is the iteration budget. This is exactly how an agent loop works, except the senior reviewer is a test suite, a validator, or a second model. Simple Python pseudocode def run_until_success(task, checker, max_attempts=3): errors = [] for attempt in range(max_attempts): result = task(errors) # ask the model, pass in past errors if checker(result): # verify with a rule or a test return result errors.append(result["error"]) # feed the failure back next time raise RuntimeError("Budget exhausted") The task call is the model invocation. The checker is anything deterministic or probabilistic that decides whether the output is good enough. The loop itself has no idea what the work is — it just repeats, feeds errors back, and stops. Diagram of a self-correction loop +------------+| Ask model |+------------+ | v+----------------+| Check result ? |+----------------+ | |pass fail | | v v+-----------+ +---------------+| Return | | Collect error || result | +---------------++-----------+ | v +------------------------+ | Append error to prompt | +------------------------+ | +--------------------+ | v Ask model How loops improve AI outputs Loops improve outputs in three concrete ways: Reliability. Validation turns random quality into guaranteed quality. If a checker enforces the contract, the loop cannot return a broken result — it keeps trying until it fits. Context. Each retry carries the previous failure. The model stops guessing and starts debugging. Control. The budget (max attempts) bounds cost and latency. You never get an unbounded bill from a stubborn model. The cost is time and tokens. Every retry is another model call. Loop engineering is the art of getting the minimum number of retries that meets your quality bar. Key takeaway: A loop is not “try harder.” It is “try again with new information.” Without feedback, a retry is just a re-roll of the dice. 3. What Is Graph Engineering? Graph engineering is the practice of designing AI workflows as explicit steps, connections, and decisions. You draw the workflow as a graph — nodes are steps, edges are connections — and the runtime executes that graph. The core concept A graph is a set of nodes and the edges between them. In an AI workflow: Nodes are units of work. A node can call a model, run a tool, query a database, or wait for a human. Edges say which node runs after which. An edge from A to B means “when A finishes, run B.” State is the shared data every node can read and write. It is the workflow’s memory. Conditional routing is an edge with a decision attached: “run B if X, otherwise run C.” Branching is one node feeding several edges, so multiple nodes run in parallel. Fan-in is several nodes feeding one node, so the next step waits for all of them. The power of a graph is that you can see and change the workflow. The code stops being a wall of control flow and becomes a map you can inspect, log, and modify one edge at a time. State machines and decision-based workflows A workflow graph is essentially a state machine. State flows through nodes, and edges decide which state transition happens next. The AI part is that the decisions can be made by a model. For example, a support workflow might route based on a model’s judgment: An incoming ticket enters the system. A model classifies it as “refund,” “technical,” or “account.” A conditional edge sends each type to a different handler. A refund goes to a payments tool. A technical issue goes to a debugging agent. An account problem goes to a human. The classification node is a model call. The routing is a conditional edge. Together they let the system make decisions instead of blindly running steps in order. Real-world analogy A graph is like a process flowchart for a bank loan application. An applicant submits forms. An automated step checks their credit score. A decision point routes to “approved,” “needs a human review,” or “rejected.” If approved, several tasks run in parallel: drafting the contract, updating the system, sending an email. A final step waits for all of them before closing the application. That is a graph. Every box is a node, every arrow is an edge, and the diamonds are conditional routes. An AI workflow graph is the same picture — some of the boxes are LLM calls instead of scripts. Mermaid diagram of a decision-based workflow ┌─────────────────┐ │ Ticket arrives │ └────────┬────────┘ │ ▼ ┌────────────────────┐ │ Classify intent │ └─────────┬──────────┘ │ ▼ ┌────────────────┐ │ Is it urgent? │ └──────┬───┬─────┘ Yes │ │ No ▼ ▼ ┌──────────────────┐ ┌────────────────┐ │ Escalate to │ │ Agent handles │ │ human │ └───────┬────────┘ └────────┬─────────┘ │ │ │ └──────────┬─────────┘ ▼ ┌────────────────┐ │ Close ticket │ └────────────────┘ Why graphs let systems make decisions Without a graph, a workflow is linear: do A, then B, then C. The model can still produce decisions as text, but the system has no structured way to act on them. With a graph, decisions are real control flow. A model’s output selects an edge, and the workflow branches, parallelizes, or stops accordingly. This is the difference between an app that prints “the ticket is urgent” and an app that actually escalates it, then runs the post-escalation steps. Python example (framework-free, LangGraph-inspired) A graph can be expressed in plain Python with dictionaries: def plan(state): state["plan"] = "analyze the bug and pick a target file" return statedef execute_fix(state): state["code"] = "fixed_version = patched_code()" return statedef evaluate(state): state["passed"] = "PASSED" in run_tests() return stateNODES = {"plan": plan, "execute_fix": execute_fix, "evaluate": evaluate}def run_graph(state, start="plan"): node = start while node != "END": state = NODES[node](state) node = route(state, node) # decide the next edge from the state return state Every node takes the state and returns a modified state. A route function reads the state and picks the next node. This is the essence of graph execution: nodes are pure-ish transforms, and a separate decision layer picks the edges. Real frameworks add parallel execution, checkpoints, and tool calls, but the model is the same. Key takeaway: Graph engineering separates “what the steps are” from “how they connect.” The graph is a map you can read; the runtime is the vehicle that drives it. 4. Loop Engineering vs Graph Engineering The two ideas overlap, and both are needed, but they optimize for different things. DimensionLoop EngineeringGraph EngineeringCore shapeOne step repeated with feedbackMany steps connected with edgesComplexityLow to startHigher, more moving partsFlexibilityHigh for retry policyHigh for layout, branching, routingDecision makingImplicit (stop / retry)Explicit (conditional edges)State managementMinimal, owned by the loopCentral, shared across nodesScalabilityVertical — more iterationsHorizontal — more nodes and branchesLearning curveEasySteeperDebuggingCheck the last attemptInspect each node’s stateTypical use casesRetries, self-correction, verificationPipelines, multi-step agents, approvalsPerformanceCost grows per retryCost set by node layout, can parallelize The short version: Use a loop when the work is fundamentally the same task repeated until a condition is met. Fixing code until tests pass is a loop. Use a graph when the work has distinct stages, branching, or parallel paths. A pipeline of analyze → plan → execute → review → deploy is a graph. Use both when you need retry on top of structure. The graph defines the passes; the loop decides how many passes and what to feed back. The combination that most production agents use A common production shape is a one-pass graph driven by an outer loop. In the LangGraph example below, the graph runs plan → fix → check, and a separate while loop re-invokes the graph when the check fails: graph = build_fix_graph() # plan -> fix -> {security scan, tests} -> finalizestate = initial_state(bug_report)for attempt in range(max_iterations): state = graph.invoke(state) # one full pass through the graph if state["status"] == "resolved": break state["error_history"].append(state["test_results"]) # feed back into next pass This is the pattern to internalize: the graph answers “what are the steps,” and the loop answers “when do we stop and what do we do differently next time.” You can change the graph without touching the loop, and vice versa. That separation is what makes the system maintainable. Key takeaway: Loop and graph are not rivals. The graph draws the path; the loop decides how many times to walk it. 5. Why These Skills Matter in 2026 AI engineering in 2026 is moving from single model calls to systems that act over long periods. Several trends push in that direction. AI agents. An agent is a model given tools and a loop: observe, decide, act, check, repeat. Every agent you build is loop engineering plus tool calling, whether or not you use a framework. Autonomous systems. Tasks like “watch this directory and file PRs” or “monitor these metrics and page on call” run unattended for hours. They need budgets, stopping conditions, and recovery paths — loop discipline. Multi-agent orchestration. Real products use several agents: a planner, an executor, a reviewer. Coordinating them requires graphs. Someone has to decide who runs when and how results flow between them. Reasoning models and tool calling. Reasoning models already “loop” internally. Your job is to wrap that with the right outer loop and the right tools, so the model’s reasoning produces verifiable actions. Model Context Protocol (MCP). MCP standardizes how agents talk to tools and data sources. Once every agent can reach the same tool catalog, the differentiator becomes the workflow — the loops and graphs on top of the tools. Enterprise AI applications. Companies are done with demos. Production workloads need retries, logging, human approval steps, and bounded costs. Those are loop and graph concerns, not prompt concerns. Long-running workflows. Anything that runs for minutes or hours — data pipelines, code generation, research agents — needs checkpoints and resumability. Graph frameworks built for this, and loops with budgets, are the tools for it. The demand signal is clear: job posts for AI engineers now list LangGraph, CrewAI, and agent orchestration next to Python and prompt skills. Employers want people who can design workflows that fail safely, not just people who can write a good system prompt. To stay balanced: you do not need this to build every AI feature. A translation endpoint or a summarizer is still a single call. But the moment you build a tool-calling agent, a RAG pipeline with routing, or an autonomous task runner, loop and graph engineering become the job. Key takeaway: Prompt quality is table stakes in 2026. Workflow design is the differentiator. The market is pricing engineers who can make AI systems reliable, not just clever. 6. Building Loop Engineering in Python Here are the loop patterns you will use most, written in plain Python. Each one is deliberately small so you can adapt it. Retry loop The simplest loop. Good for flaky APIs and transient failures. import timedef call_with_retry(fn, max_attempts=3, delay=1.0): for attempt in range(1, max_attempts + 1): try: return fn() except Exception as e: if attempt == max_attempts: raise print(f"Attempt {attempt} failed: {e}. Retrying in {delay}s.") time.sleep(delay) What this does: tries the function, and on failure waits and tries again. The two important decisions are the attempt cap and the backoff delay. Without the cap, a permanently broken service loops forever. max_attempts is your stopping condition. Reflection loop Ask the model to critique and improve its own work. Cheap and often surprisingly effective. def reflect_and_improve(chat, task, passes=2): answer = chat(f"Solve this: {task}") for _ in range(passes - 1): critique = chat( f"Here is a draft answer:\n{answer}\n" "List its weaknesses and return an improved version." ) answer = critique return answer Why this works: the second call gets the first output as input. That extra context is the feedback. The risk is that a confident but wrong model agrees with itself — so reflection pairs well with an external checker. Validation loop Check output against a hard rule, and retry with the error. Classic example: parsing JSON that the model must produce. import jsondef parse_with_retry(chat, prompt, max_attempts=3): for attempt in range(max_attempts): raw = chat(prompt) try: return json.loads(raw) except json.JSONDecodeError as e: prompt += f"\nYour last output did not parse: {e}. Fix it." raise ValueError("Model kept producing invalid JSON") The validator is json.loads. The error message is appended to the prompt, so the model sees exactly what went wrong. This pattern generalizes to any schema validator, from Pydantic to a custom function. Evaluation loop Score the output against a rubric and retry until the score passes. def improve_until_score(chat, generate, score, target=0.8, max_attempts=4): best = None for _ in range(max_attempts): candidate = generate() s = score(candidate) if best is None or s > best[0]: best = (s, candidate) if s >= target: return candidate return best[1] Two design points matter. First, always keep the best candidate — you do not want to return a worse result than one you already had. Second, the scorer can be a model, a test suite, or a human. The loop does not care. Self-correction loop (the agent pattern) This is the loop behind coding agents and autonomous fixers. def run_fix_loop(agent, bug_report, max_iterations=3): errors = [] for iteration in range(max_iterations): result = agent.run(bug_report, errors) # agent has the tools if result["tests"] == "PASSED" and result["review"] == "SECURE": return {**result, "status": "resolved"} errors.append(f"Attempt {iteration + 1}: {result['failure_reason']}") return {"status": "max_iterations_reached", "errors": errors} The pattern: run, check, collect the error, feed it back. Notice the loop only owns retry policy — whether a single run passes is decided inside the agent. This mirrors the loop/graph split used in production systems. Stopping conditions and preventing infinite loops Every loop needs an exit, and the exit must be reachable: Budget: always cap attempts. max_iterations is non-negotiable in production. Progress check: break early if the error message repeats. If the model makes the same mistake twice, more retries are wasted tokens. Log everything: write the attempt number and the error to a log. When a workflow fails, the log is your only record of why. Make failure a first-class result: return a status like max_iterations_reached instead of raising. Callers should be able to handle "we gave up" gracefully. Key takeaway: A loop without a stopping condition is a cost bug. Budgets, progress checks, and logging turn an infinite retry into a controlled process. 7. Building Graph Engineering in Python Now let’s build a graph-based workflow. The concepts — state, nodes, edges, conditional routing, branching — are the same across frameworks, so we start from first principles. The state object State is a dictionary that every node reads and writes. It is the workflow’s memory. from dataclasses import dataclass, field@dataclassclass FixState: bug_report: str = "" file_path: str = "" plan: str = "" patched_code: str = "" test_results: str = "" security_report: str = "" error_history: list[str] = field(default_factory=list) State should hold only what nodes need to share. If a node does not read it, it does not belong in state. Keeping state small makes the graph easier to debug. Nodes A node is a function that takes state and returns updated state. def plan_node(state: FixState) -> FixState: state.plan = analyze_bug(state.bug_report) # an LLM call state.file_path = detect_file_path(state.plan) return statedef execute_node(state: FixState) -> FixState: state.patched_code = apply_fix(state.file_path, state.plan) return statedef security_node(state: FixState) -> FixState: state.security_report = scan_for_vulnerabilities(state.patched_code) return statedef test_node(state: FixState) -> FixState: state.test_results = run_tests(state.file_path) return state Nodes are pure functions of state. The same input state produces the same output state, which is what makes graphs testable — you can call any node in isolation. Edges, branching, and fan-in In LangGraph, you declare the structure and the runtime handles execution: from langgraph.graph import StateGraph, ENDdef build_fix_graph(): graph = StateGraph(FixState) graph.add_node("plan", plan_node) graph.add_node("execute_fix", execute_node) graph.add_node("security_scan", security_node) graph.add_node("evaluate", test_node) graph.add_node("finalize", finalize_node) graph.set_entry_point("plan") graph.add_edge("plan", "execute_fix") # Branch: two checks run in parallel graph.add_edge("execute_fix", "security_scan") graph.add_edge("execute_fix", "evaluate") # Fan-in: finalize waits for both checks graph.add_edge("security_scan", "finalize") graph.add_edge("evaluate", "finalize") graph.add_edge("finalize", END) return graph What this declares: after execute_fix, the security_scan and evaluate nodes run at the same time. The finalize node runs only after both finish. Parallel checks are free in a graph — you just add two edges. Conditional routing Routing decides the next node from the current state. Here a human approval node and an error recovery node appear. def route_after_review(state: FixState) -> str: if state.security_report == "SECURE" and state.test_results.startswith("PASSED"): return "approve" return "repair"def build_review_graph(): graph = StateGraph(FixState) graph.add_node("review", review_node) graph.add_node("approve", approve_node) graph.add_node("repair", repair_node) graph.add_edge("review", "approve", path_map={"approve": "approve"}) graph.add_conditional_edges( "review", route_after_review, {"approve": "approve", "repair": "repair"}, ) graph.add_edge("approve", END) graph.add_edge("repair", "review") # a back-edge: this is where a loop lives in a graph return graph Two ideas here: Conditional edges let a model or a rule choose the next step. This is how a graph “makes decisions.” A back-edge (repair → review) is the graph-native way to express a loop. The graph itself can loop, which is different from the outer-loop pattern — here the retry is drawn as an arrow. Human approval and tool execution nodes Two special node types appear in most production graphs: A tool node wraps a function or an API call so the model can invoke it. In frameworks this is declarative: you register a tool, and the runtime gives the model a schema for calling it. A human approval node pauses the workflow and waits. The workflow checkpoints its state, resumes when a human approves or rejects, and picks the next edge from the decision. This is where graphs beat plain loops: an outer loop cannot easily stop for a day and resume. Error recovery Wrap node execution so a failure routes to a recovery path instead of killing the workflow: def route_on_error(state: FixState) -> str: return "fallback" if state.error_history else "retry" Frameworks like LangGraph add checkpointing so you can resume a failed workflow from the last successful node. In plain Python, you model recovery as just another node and edge: on error, route to a cleanup node, then back into the workflow. Key takeaway: A graph is a map of the workflow you can inspect and edit one edge at a time. Loops, branches, approvals, and recovery are all just nodes and arrows. 8. Frameworks That Support Loop and Graph Engineering You can build both patterns in plain Python. Frameworks make the work safer and faster: they add checkpoints, parallel execution, streaming, and observability. Here is a grounded comparison. FrameworkLoop supportGraph supportBest forLangGraphRetry via edges or control flowFirst-class graphs, checkpoints, parallel edgesComplex, resumable, inspectable workflowsCrewAIManual loops over crews/flowsCrewAI Flow: event-driven @start/@listen flowsRole-based multi-agent teamsOpenAI Agents SDKBuilt-in guardrails and retry loopsLightweight agent handoffs and routingAgents with tool calling and guardrailsAutoGenConversational agent loopsMulti-agent conversations and group chatsResearch and multi-agent conversationsPydanticAIExplicit @agent.tool retryMinimal graph helpers, model-agnosticType-safe agents with Pydantic validationLlamaIndex WorkflowsWorkflow steps, some retryDAG workflows with @step decoratorsRAG pipelines and data applicationsHaystack PipelinesComponent loopsPipeline DAGs, branching and joiningNLP pipelines, RAG, document search LangGraph is a graph-first framework built on LangChain. Nodes, edges, conditional routing, and a checkpointer are first-class. It excels when you need to pause, resume, or inspect a long workflow. Its loop story is weaker out of the box — you usually write retry logic as edges or as an outer control loop. Choose it for production workflow orchestration where reliability and inspectability matter. CrewAI organizes agents into crews that work on tasks, and CrewAI Flow provides event-driven graphs with @start and @listen decorators. It shines when your mental model is "a team of role-played agents." Its strength is ergonomics for teams; its weakness is fine-grained control over low-level execution. Choose it when a role-and-task structure matches your problem. OpenAI Agents SDK is a lightweight toolkit for tool-calling agents with guardrails and handoffs. Loops and validation are built in, and multi-agent work happens through handoffs rather than a full graph API. Choose it when you want a simple agent that calls tools safely without committing to a heavyweight orchestrator. AutoGen centers on multi-agent conversation. Agents talk to each other, which creates emergent loop and orchestration behavior. It is great for research and open-ended problem solving, but it can be harder to control deterministically. Choose it for conversational multi-agent systems. PydanticAI is a type-safe agent framework from the Pydantic team. Models, tools, and outputs are validated with Pydantic, and it is framework-agnostic across model providers. Its loop support is explicit and its graph support is minimal. Choose it when type safety and predictable structured output are your top priority. LlamaIndex Workflows gives you DAG-based workflows with @step decorators and event passing, designed for data-heavy applications. Choose it when you are already in the LlamaIndex ecosystem for RAG and data pipelines. Haystack Pipelines is a mature NLP framework with DAG pipelines, branching, and joining. It is production-tested for search and RAG. Choose it when your workflow is mostly retrieval and NLP components rather than open-ended agent loops. Do not treat the table as a verdict. Frameworks evolve quickly, and the correct choice depends on your team, your existing stack, and your workflow shape. When evaluating, write the same small workflow in two candidates and see which one stays readable after a month. Verify framework capabilities against the official documentation before committing. Key takeaway: All serious AI frameworks converge on the same two primitives: repeat-until-good (loops) and step-and-route (graphs). Pick the framework whose default mental model matches your workflow, not the one with the most features. 9. Real-World Use Cases Every modern AI application is a combination of loops and graphs. Here is how the ideas show up in practice. Customer support agents. The agent classifies the ticket (a graph edge), routes it, calls the CRM (a tool node), and drafts a reply. A validation loop re-checks the draft against company policy before it is sent. Loops give quality control; graphs give routing and escalation paths. Coding assistants. The assistant completes code, and a validation loop runs the linter and tests. If they fail, the loop feeds the errors back. A graph routes between “autocomplete,” “refactor,” and “explain” modes. The loop makes suggestions safe; the graph makes the tool navigable. AI software engineers. An autonomous fixer is loop engineering by definition: plan, patch, run checks, retry until tests pass or the budget ends. When multiple fixes are coordinated — or a reviewer agent checks an executor agent — a graph organizes the handoffs. This is loop and graph engineering at full strength. Research assistants. A research agent iterates: search, read, synthesize, cite, then a reflection loop improves the synthesis. A graph routes to different sources based on the query type. Loops drive depth; graphs drive breadth. Document processing pipelines. Ingest, OCR, chunk, embed, and index are a linear graph. A validation loop retries OCR on low-confidence pages. The graph gives you a pipeline you can resume after a failure; the loop handles the messy input. RAG systems. A query classifier (a graph edge) routes to the right retriever. After retrieval, an evaluation loop re-ranks or re-queries when the top results are weak. This is the standard shape of production RAG in 2026. Data analysis workflows. An analyst agent runs SQL, checks the results against expectations, and retries with corrected queries (a loop). Multiple analysis steps form a graph with branches for different reports. Loops correct; graphs organize. DevOps automation. An agent investigates an incident: check logs, check metrics, propose a fix. A graph routes by severity — auto-remediate or page a human. Loops bound the investigation; graphs handle the decision tree. Security analysis. A scanner agent examines code and produces findings. An evaluation loop re-scans after a fix to confirm the vulnerability is gone. A graph routes each finding to the owning team. Verification loops make security work trustworthy. Internal enterprise copilots. Copilots wrap internal tools with a graph: authenticate, find the relevant tool, run it, and get human approval for writes. A validation loop double-checks that a destructive action is safe before the approval step. Graphs enforce process; loops enforce safety. AI-powered IDEs. IDEs combine all of the above: autocomplete (single calls), refactor previews (loops that verify), and multi-file changes (graphs that track dependencies). The best ones make the loop invisible and the graph resumable. In every case, the pattern is the same as traditional software: structure the process, verify the output, and bound the cost. The only difference is that the decision-maker is a model. Key takeaway: If your AI feature can be built as a single call, keep it a single call. The moment it acts, verifies, or routes, it is a loop or a graph — and you should design it like one. 10. Common Mistakes These are the failures I see most in production AI workflows. Infinite loops. A retry loop without a budget runs forever, burning tokens and money. Fix: every loop needs a max-iteration bound and a status like max_iterations_reached returned, not raised. Missing stopping conditions. Sometimes the loop runs the right number of times but never checks progress. Fix: break when the error message repeats. Retrying with identical feedback is wasted spend. Poor state management. Shoving everything into one giant state object makes nodes hard to test and workflows impossible to debug. Fix: keep state minimal and typed. If a node does not read a field, remove it. Over-engineering. Building a fifteen-node graph for a task a single call handles. Fix: start linear, add branches only when a decision actually exists. The smallest structure that meets the requirement is the right one. Ignoring logging. An agent that fails without logs is un-debuggable. Fix: log attempt number, node name, and the state at each checkpoint. In production, treat the workflow log like an application log. Lack of observability. Frameworks hide execution inside compiled graphs. Fix: instrument node entry and exit, and expose traces. When a customer reports a bad answer, you need to know which node produced it. Weak prompt design. A vague system prompt makes retries loop in circles. Fix: prompts should say exactly what a node must produce and what success looks like, so the checker and the model agree. Improper node separation. Nodes that mix model calls, tool calls, and side effects are untestable. Fix: one node, one responsibility. A model node should not also write to a database. Tight coupling between components. A loop that hard-codes the graph’s node names breaks the moment the graph changes. Fix: the loop should talk to the graph through a small interface (input, output, status). Change one without rewriting the other. Key takeaway: Most workflow bugs are not model bugs. They are control-flow bugs: no budget, no logs, no boundaries. Fix the structure and the model problems get easier to find. 11. Learning Roadmap for 2026 You do not need all of this on day one. Here is an order that builds each skill on the last. 1. **Python fundamentals.** Functions, classes, typing, and the standard library. You will write the tools and loops yourself before you trust a framework. - *Project:* a CLI that parses input, calls an API, and formats output. 2. **APIs.** HTTP, JSON, error codes, and retries. Almost every workflow talks to a service. - *Project:* a script that fetches data, handles rate limits, and retries failures. 3. **Prompt engineering.** Clear instructions, few-shot examples, and structured output. - *Project:* a summarizer with a well-tested prompt and output contract. 4. **Function calling.** Teach a model to produce structured calls to your functions. - *Project:* an assistant that can query a mock database. 5. **Tool calling.** Give the model a catalog of real tools with schemas. - *Project:* an agent that searches the web, reads a file, and runs a command. 6. **Loop engineering.** Build validation, retry, and self-correction loops by hand. - *Project:* an agent that writes code until tests pass, with a budget and logs. 7. **Graph engineering.** Model workflows as nodes, edges, and routing. - *Project:* a three-stage pipeline with a parallel check and a decision edge. 8. **AI agents.** Combine tools, loops, and graphs into one autonomous agent. - *Project:* a research agent that plans, searches, and writes a cited report. 9. **Multi-agent systems.** Split work across specialized agents with handoffs. - *Project:* a planning agent and an executing agent that review each other. 10. **Model Context Protocol (MCP).** Connect agents to shared tools and data. - *Project:* an MCP server exposing your internal tools, consumed by an agent. 11. **Production AI systems.** Add observability, budgets, approvals, and rollback. - *Project:* deploy the research agent behind an API with tracing and cost limits. The first five are foundation. Step 6 is where you learn loop engineering by hand, step 7 is graph engineering by hand, and steps 8–11 are where you use frameworks and ship. Each project should be small, runnable, and boring — boring is what makes it reliable. Key takeaway: Learn the primitives by hand first. Frameworks are easier to trust once you have built the same thing with dictionaries and while loops. 12. Should You Learn These Skills in 2026? An honest look at the decision. Industry demand. Yes, and growing. Job posts for AI engineers increasingly list workflow orchestration, LangGraph, CrewAI, and agent design. The demand is strongest for engineers who can ship reliable, inspectable systems — which is exactly what loops and graphs provide. Future relevance. These skills are not tied to a single model vendor. The primitives — repeat, verify, route — will outlast any framework or model. That makes them durable in a fast-moving field. Career opportunities. Engineers who design workflows sit above prompt-only engineers in scope and impact. They build the systems that run business processes, and they are hard to replace with a model call. That translates into leverage and compensation. Learning effort. Moderate. The concepts are simple. The challenge is judgment: knowing when a loop suffices, when a graph is justified, and how to keep both maintainable. That judgment comes from building a few real systems. Who should learn these. Python developers, backend engineers, and full-stack engineers who build anything with agents or LLM workflows. If your job involves tool calling, multi-step tasks, or autonomous behavior, this is core material. When you do not need them. If you only build single-shot features — translation, classification, summarization — a prompt and a good model still cover you. Solo prototyping and internal tools with a human in the loop can also skip the heavy machinery. There is no shame in a single call; it is often the correct engineering decision. The balanced recommendation: learn the primitives by hand, use frameworks only when a workflow crosses a real complexity threshold, and keep the “smallest structure that works” principle in front of you at all times. Key takeaway: Loop and graph engineering are core skills for anyone shipping agents in 2026 — but only worth the complexity when your workflow actually needs them. 13. Conclusion Prompt engineering taught you what to say to a model. Loop engineering and graph engineering teach you what to do with the answer. A loop takes an unreliable single call and turns it into a dependable process. Verify, feed the failure back, retry, and stop on budget. A graph takes a messy multi-step task and turns it into a visible map: nodes, edges, branches, and decisions. Most production agents need both — a graph that draws the path and a loop that decides how many times to walk it. Start small. Build a validation loop around one model call, then add a retry, then a reflection step. Once that feels natural, draw the same workflow as a graph and watch the structure make itself obvious. The primitives are simple; the skill is knowing where to apply them. In 2026, the engineers who stand out are not the ones who write the cleverest prompts. They are the ones who build systems that fail safely, recover gracefully, and make the model’s output trustworthy. That is loop engineering. That is graph engineering. That is the job. Key Takeaways Prompt engineering says what to do; loop and graph engineering say how to get it done. A loop repeats work with feedback until a check passes or a budget runs out. Loops without feedback are just re-rolls; feedback is what makes retries improve. A graph is nodes, edges, and state — the workflow as a visible, inspectable map. Conditional edges let workflows make real decisions; back-edges let graphs loop natively. Production agents usually combine a one-pass graph with an outer retry loop. Every loop needs a stopping condition, a progress check, and logging. Learn the primitives in plain Python before trusting a framework. Choose a framework for its mental model, not its feature list. The smallest structure that meets your requirement is the correct one. References LangGraph official documentation — https://langchain-ai.github.io/langgraph/ CrewAI documentation (Agents, Flows) — https://docs.crewai.com/ OpenAI Agents SDK — https://openai.github.io/openai-agents-python/ PydanticAI documentation — https://ai.pydantic.dev/ LlamaIndex Workflows — https://docs.llamaindex.ai/ Haystack Pipelines — https://docs.haystack.deepset.ai/ Model Context Protocol (MCP) — https://modelcontextprotocol.io/ Verify framework APIs against the official documentation, as versions evolve quickly.

03
temporalApr 01, 2026
12 min

Building Reliable Background Workflows with Temporal and FastAPI: A Step-by-Step Guide

How to handle long-running tasks without losing state when servers restart Workflows with Temporal and FastAPIThe Problem Every Backend Developer Knows You build an API endpoint. A user clicks a button. Your code starts a process that takes several minutes — maybe it processes a payment, sends emails, generates a report, or provisions a server. Then one of these happens: • The server restarts during a deploy • A database connection drops • An external API times out • The process runs out of memory Your task dies. The user sees nothing. You have no record of what happened. Someone has to manually figure out what went wrong and try again. This is the problem Temporal solves. And when you pair it with FastAPI, you get a system where your API responds instantly while complex work runs safely in the background — with full visibility into every step. What Is Temporal? Temporal is a workflow engine. It runs your code and remembers its state. If your server crashes, Temporal picks up where it left off. If an API call fails, Temporal retries it. If a task needs to wait for a human approval that takes three days, Temporal waits — without using any server resources. Think of it this way: instead of writing code that hopes nothing goes wrong, you write code that assumes things will go wrong — and Temporal handles the recovery. How FastAPI and Temporal Work Together FastAPI handles incoming HTTP requests. Temporal handles everything that happens after. Here is the flow: ┌─────────────┐ HTTP Request ┌──────────────┐│ │ ─────────────────────► │ ││ Client │ │ FastAPI ││ (Browser) │ │ Server ││ │ ◄───────────────────── │ │└─────────────┘ Quick Response └──────┬───────┘ │ Starts workflow via Temporal Client │ ▼ ┌────────────────┐ │ │ │ Temporal │ │ Server │ │ (Orchestrator)│ │ │ └───────┬────────┘ │ Assigns tasks to workers │ ┌─────────┴──────────┐ │ │ ▼ ▼ ┌──────────────┐ ┌──────────────┐ │ Workflow │ │ Activities │ │ (Orchestr. │───►│ (Real Work) │ │ Logic) │ │ │ └──────────────┘ └──────────────┘ │ │ │ Calls external │ services, DBs, │ APIs, etc. ▼ ▼ ┌─────────────────────────────────┐ │ Temporal Server │ │ Records every step taken │ │ Retries failures automatically│ │ Preserves state on crash │ └─────────────────────────────────┘ The key idea: FastAPI starts a workflow and returns immediately. Temporal takes over, runs the work across however many steps are needed, and stores the result. Your API stays fast. Your work gets done reliably. A Real-World Example: E-Commerce Order Processing Let’s build something real. An online store needs to process orders. When a customer places an order, the system must: 1. Validate the order details 2. Charge the customer’s payment method 3. Reserve items in the warehouse 4. Send a confirmation email 5. Update the order status If step 2 succeeds but step 3 fails, the customer has been charged but has no order. That’s a real problem. With Temporal, if any step fails, the system can retry it — or undo the steps that already completed (this is called the Saga pattern). Let’s build it. Step 1: Install the Dependencies pip install temporalio fastapi uvicorn httpx You also need a running Temporal server. For local development: # Install the Temporal CLI first, then:temporal server start-dev This starts a local Temporal server on localhost:7233 with a web UI at http://localhost:8233. Step 2: Define the Activities Activities are where the actual work happens. Each activity does one thing and can be retried independently. Create a file called activities.py: from temporalio import activityfrom dataclasses import dataclassfrom typing import Optionalimport logginglogger = logging.getLogger(__name__)@dataclassclass OrderInput: order_id: str customer_email: str items: list[str] total_amount: float payment_method: str@dataclassclass OrderResult: order_id: str status: str confirmation_number: Optional[str] = None@activity.defnasync def validate_order(order: OrderInput) -> bool: """Check that the order has valid items and amounts.""" logger.info(f"Validating order {order.order_id}") if not order.items: raise ValueError("Order has no items") if order.total_amount <= 0: raise ValueError("Order total must be positive") logger.info(f"Order {order.order_id} is valid") return True@activity.defnasync def process_payment(order: OrderInput) -> str: """Charge the customer's payment method.""" logger.info(f"Processing payment of ${order.total_amount} for order {order.order_id}") # In production, this calls Stripe, PayPal, etc. # For now, simulate the call confirmation = f"PAY-{order.order_id}-001" logger.info(f"Payment confirmed: {confirmation}") return confirmation@activity.defnasync def reserve_inventory(order: OrderInput) -> bool: """Reserve items in the warehouse.""" logger.info(f"Reserving inventory for order {order.order_id}") # In production, this calls your inventory service # Simulate success logger.info(f"Items reserved for order {order.order_id}") return True@activity.defnasync def send_confirmation_email(order: OrderInput, confirmation: str) -> bool: """Send an order confirmation email to the customer.""" logger.info(f"Sending confirmation email to {order.customer_email}") # In production, this calls SendGrid, SES, etc. logger.info(f"Email sent to {order.customer_email} with confirmation {confirmation}") return True@activity.defnasync def update_order_status(order_id: str, status: str) -> bool: """Update the order status in the database.""" logger.info(f"Updating order {order_id} status to: {status}") # In production, this writes to your database logger.info(f"Order {order_id} status updated to {status}") return True# --- Compensation activities (undo steps if something fails) ---@activity.defnasync def refund_payment(confirmation: str) -> bool: """Refund a payment if the order fails after charging.""" logger.info(f"Refunding payment {confirmation}") # In production, this calls your payment provider's refund API return True@activity.defnasync def release_inventory(order_id: str) -> bool: """Release reserved inventory if the order fails.""" logger.info(f"Releasing inventory reservation for order {order_id}") return True Step 3: Define the Workflow The workflow ties the activities together in order. It handles retries and compensations. Create a file called workflows.py: from datetime import timedeltafrom temporalio import workflow# Pass activities through the sandbox so they are not reloaded on each runwith workflow.unsafe.imports_passed_through(): from activities import ( OrderInput, OrderResult, validate_order, process_payment, reserve_inventory, send_confirmation_email, update_order_status, refund_payment, release_inventory, )@workflow.defnclass OrderProcessingWorkflow: @workflow.run async def run(self, order: OrderInput) -> OrderResult: """Process an order through all required steps.""" payment_confirmation = None inventory_reserved = False try: # Step 1: Validate the order is_valid = await workflow.execute_activity( validate_order, order, start_to_close_timeout=timedelta(seconds=10), ) if not is_valid: await workflow.execute_activity( update_order_status, order.order_id, "INVALID", start_to_close_timeout=timedelta(seconds=5), ) return OrderResult(order_id=order.order_id, status="INVALID") # Step 2: Process payment payment_confirmation = await workflow.execute_activity( process_payment, order, start_to_close_timeout=timedelta(seconds=30), retry_policy=workflow.RetryPolicy( maximum_attempts=3, initial_interval=timedelta(seconds=1), backoff_coefficient=2.0, ), ) # Step 3: Reserve inventory inventory_reserved = await workflow.execute_activity( reserve_inventory, order, start_to_close_timeout=timedelta(seconds=15), retry_policy=workflow.RetryPolicy( maximum_attempts=3, initial_interval=timedelta(seconds=1), ), ) if not inventory_reserved: raise RuntimeError("Could not reserve inventory") # Step 4: Send confirmation email await workflow.execute_activity( send_confirmation_email, order, payment_confirmation, start_to_close_timeout=timedelta(seconds=10), retry_policy=workflow.RetryPolicy( maximum_attempts=5, initial_interval=timedelta(seconds=2), ), ) # Step 5: Update order status to complete await workflow.execute_activity( update_order_status, order.order_id, "COMPLETED", start_to_close_timeout=timedelta(seconds=5), ) return OrderResult( order_id=order.order_id, status="COMPLETED", confirmation_number=payment_confirmation, ) except Exception as e: # Something failed. Undo what we can. workflow.logger.error(f"Order {order.order_id} failed: {e}") # Refund payment if it was charged if payment_confirmation: await workflow.execute_activity( refund_payment, payment_confirmation, start_to_close_timeout=timedelta(seconds=10), ) # Release inventory if it was reserved if inventory_reserved: await workflow.execute_activity( release_inventory, order.order_id, start_to_close_timeout=timedelta(seconds=10), ) # Mark order as failed await workflow.execute_activity( update_order_status, order.order_id, "FAILED", start_to_close_timeout=timedelta(seconds=5), ) return OrderResult( order_id=order.order_id, status="FAILED", ) Step 4: Create the FastAPI Server This is where the user’s request enters the system. The API starts the workflow and returns a workflow ID so the client can check on progress. Create a file called main.py: import asynciofrom contextlib import asynccontextmanagerfrom fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom temporalio.client import Clientfrom workflows import OrderProcessingWorkflowfrom activities import OrderInput, OrderResult# --- Request and response models ---class OrderRequest(BaseModel): order_id: str customer_email: str items: list[str] total_amount: float payment_method: strclass OrderResponse(BaseModel): workflow_id: str status: str message: strclass OrderStatusResponse(BaseModel): order_id: str workflow_status: str result: str | None = None# --- Global Temporal client ---temporal_client: Client | None = None@asynccontextmanagerasync def lifespan(app: FastAPI): """Start up and shut down the Temporal client with the app.""" global temporal_client temporal_client = await Client.connect("localhost:7233") yield # Client does not need explicit cleanupapp = FastAPI( title="Order Processing API", description="Accepts orders and processes them via Temporal workflows", version="1.0.0", lifespan=lifespan,)@app.post("/orders", response_model=OrderResponse)async def create_order(order: OrderRequest): """Submit a new order for processing.""" if temporal_client is None: raise HTTPException(status_code=503, detail="Temporal client not connected") # Convert the Pydantic model to our Temporal input order_input = OrderInput( order_id=order.order_id, customer_email=order.customer_email, items=order.items, total_amount=order.total_amount, payment_method=order.payment_method, ) # Start the workflow - this returns immediately handle = await temporal_client.start_workflow( OrderProcessingWorkflow.run, order_input, id=f"order-{order.order_id}", task_queue="order-processing", ) return OrderResponse( workflow_id=handle.id, status="PROCESSING", message=f"Order {order.order_id} submitted for processing", )@app.get("/orders/{order_id}/status", response_model=OrderStatusResponse)async def get_order_status(order_id: str): """Check the status of an order.""" if temporal_client is None: raise HTTPException(status_code=503, detail="Temporal client not connected") try: handle = temporal_client.get_workflow_handle(f"order-{order_id}") status = await handle.describe() result = None if status.status == 3: # WorkflowStatus.COMPLETED result = await handle.result() return OrderStatusResponse( order_id=order_id, workflow_status=status.status.name, result=str(result) if result else None, ) except Exception as e: raise HTTPException(status_code=404, detail=f"Order not found: {e}")@app.get("/health")async def health_check(): """Check that the API and Temporal connection are healthy.""" if temporal_client is None: return {"status": "unhealthy", "temporal": "not connected"} try: # Quick check that we can reach Temporal await temporal_client.list_workflows("WorkflowType='OrderProcessingWorkflow'") return {"status": "healthy", "temporal": "connected"} except Exception: return {"status": "degraded", "temporal": "connection error"} Step 5: Create the Worker The worker runs your workflow and activity code. It polls Temporal for tasks and executes them. Create a file called worker.py: import asyncioimport loggingfrom temporalio.client import Clientfrom temporalio.worker import Workerfrom workflows import OrderProcessingWorkflowfrom activities import ( validate_order, process_payment, reserve_inventory, send_confirmation_email, update_order_status, refund_payment, release_inventory,)async def main(): logging.basicConfig(level=logging.INFO) # Connect to Temporal client = await Client.connect("localhost:7233") # Create the worker worker = Worker( client, task_queue="order-processing", workflows=[OrderProcessingWorkflow], activities=[ validate_order, process_payment, reserve_inventory, send_confirmation_email, update_order_status, refund_payment, release_inventory, ], ) logging.info("Starting order processing worker...") await worker.run()if __name__ == "__main__": asyncio.run(main()) Step 6: Run Everything Open three terminal windows. Terminal 1 — Start Temporal: temporal server start-dev Terminal 2 — Start the Worker: python worker.py You should see: Starting order processing worker... Terminal 3 — Start the FastAPI Server: uvicorn main:app --reload Your API is now running at http://localhost:8000. Step 7: Test It Submit an order: curl -X POST http://localhost:8000/orders \ -H "Content-Type: application/json" \ -d '{ "order_id": "ORD-12345", "customer_email": "customer@example.com", "items": ["Widget A", "Widget B"], "total_amount": 49.99, "payment_method": "card_ending_4242" }' Response: { "workflow_id": "order-ORD-12345", "status": "PROCESSING", "message": "Order ORD-12345 submitted for processing"} Check the status: curl http://localhost:8000/orders/ORD-12345/status Response: { "order_id": "ORD-12345", "workflow_status": "COMPLETED", "result": "OrderResult(order_id='ORD-12345', status='COMPLETED', confirmation_number='PAY-ORD-12345-001')"} Open the Temporal Web UI at http://localhost:8233 to see the full execution history — every step, every retry, every decision. Why This Matters Here is what you get from this setup that you do not get from a simple background job: Automatic retries. If the payment service is down, Temporal retries with exponential backoff. You write the retry policy once. It applies to every run. State persistence. If your worker crashes mid-order, Temporal replays the workflow from the last completed step. No data is lost. No manual recovery needed. Full visibility. Every workflow execution has a complete history. You can see exactly what happened, when, and why. This is invaluable for debugging and compliance. Compensation handling. If payment succeeds but inventory fails, the workflow automatically refunds the payment. You do not need a separate reconciliation job. No infrastructure to manage for state. You do not need a job queue database, a state machine service, or a custom retry system. Temporal handles all of that. Other Common Use Cases Data pipelines. Ingest data from multiple sources, transform it, and load it into a warehouse. If any step fails, retry from that step — not from the beginning. User onboarding. Create accounts, send welcome emails, provision resources, and schedule follow-up messages over several days. Temporal handles the delays without keeping servers running. Approval workflows. Submit a request, wait for a manager to approve or reject it (which could take hours or days), then proceed based on the decision. Temporal waits using almost no resources. Scheduled reports. Generate and email reports on a schedule. If the report generation fails, retry automatically. Track every run in the UI. Multi-service transactions. Coordinate actions across several microservices. If one service fails, undo the actions of the others. This is the Saga pattern, and Temporal makes it straightforward. Key Concepts to Remember Concept What It Is Workflow The orchestration logic. It decides what steps run and in what order. Activity A single unit of work. Makes API calls, writes to databases, sends emails. Worker A process that runs your workflow and activity code. Polls Temporal for tasks. Task Queue A named queue that connects workflows to workers. Temporal Server The orchestrator. Stores state, schedules retries, and tracks history. Temporal Client Your code’s connection to the Temporal Server. Used to start and query workflows. Production Tips 1. Use meaningful workflow IDs. Include a business identifier like order-12345 or user-onboarding-abc. This makes debugging much easier. 2. Set timeouts on every activity. An activity without a timeout can hang forever. Always set start_to_close_timeout or schedule_to_close_timeout. 3. Use retry policies for flaky operations. External APIs fail. Set maximum_attempts and initial_interval so Temporal retries automatically. 4. Keep workflows deterministic. Workflows must produce the same result every time they run from the same starting point. Do not use random numbers, current time, or external calls inside workflow code. Put those in activities. 5. Monitor your workers. Track worker health, task queue depth, and workflow completion rates. Temporal exposes metrics that integrate with Prometheus and Grafana. 6. Separate environments with namespaces. Use different Temporal namespaces for development, staging, and production. This prevents test workflows from interfering with real ones. What Comes Next This example covers the basics. From here you can explore: • Signals — Send events to a running workflow (like a user approving or rejecting something) • Queries — Ask a running workflow about its current state without changing it • Child workflows — Break large workflows into smaller, reusable pieces • Scheduled workflows — Run workflows on a cron schedule • Temporal Cloud — Run Temporal as a managed service instead of self-hosting The Temporal Python SDK documentation and sample projects on GitHub are good starting points for each of these. Final Thoughts The combination of FastAPI and Temporal solves a real problem: how to run complex, multi-step processes reliably without making your API slow or fragile. FastAPI gives you a fast, clean API layer. Temporal gives you durability, retries, and visibility. Together, they let you write code that handles failure gracefully — without adding a lot of extra infrastructure. If you are building anything that involves more than one step, talks to external services, or needs to survive a server restart, this pattern is worth considering. All code in this article is available as a working project. The Temporal Python SDK is open source under the MIT license. Temporal server is available as a self-hosted option or through Temporal Cloud.

Background

Education

Academic foundation that shaped my technical approach.

2 Degrees

Academic background

Nepal Commerce Campus (NCC)

Bachelor in Information Management

2017 - 2021
Visit

Comprehensive program focused on producing IT professionals with strong technical and management skills

Ambition College

Higher Secondary Education (Management - Computer Science)

2014 - 2016
Visit

Focused on computer science fundamentals combined with business strategy and technical expertise

Open to opportunities

Let's build something great together.

I'm currently available for new projects and collaborations. Whether you need a scalable backend, AI integration, or full-stack development — let's talk.

Available now
5+ years experience
15+ APIs deployed
Fast turnaround
Contact

Let's Work Together

Tell me about your project and I'll get back to you within a few hours.

Send a Message