Building an Interruptible AI Workflow Engine with Python Generators

-

Building an Interruptible AI Workflow Engine with Python Generators

A resilient AI workflow engine can be constructed using Python generators to pause execution at any point where external input or LLM processing is required. By storing the history of inputs and replaying them through the generator, developers can persist state across ephemeral compute instances without complex database schemas.

It was 3:15 AM on a Tuesday when my PagerDuty alert finally snapped me awake. Usually, a 3 AM alert means a service is down or a database is hitting connection limits. This time, it was my GCP billing threshold. In less than two hours, a runaway loop in my AI agent orchestration layer had burned through $432 in Gemini API credits. My "smart" agent, designed to handle multi-step research tasks, had gotten stuck in a recursive logic trap, calling the LLM repeatedly with the same 15,000-token context window without making any progress.

The culprit was a traditional state machine I’d built using Python's async/await and a database-backed status column. The logic was brittle. If a step failed or required a "human-in-the-loop" (HITL) intervention, the state management became a nightmare of if/else blocks and manual flag updates. I realized that my architectural approach to long-running, non-linear workflows was fundamentally flawed. I needed a way to pause execution, persist state, and resume exactly where I left off without re-running the entire logic tree or managing a massive, 50-column "state" table.

I spent the next week gutting the orchestration layer and rebuilding it using a feature of Python that most developers overlook for high-level architecture: Generators. By leveraging yield, I transformed my complex, bug-prone state machine into a readable, linear AI workflow engine that is natively interruptible. Here is how I did it, the performance trade-offs I encountered, and why I’ll never go back to standard state machines for AI agents.

Why Traditional State Machines Fail in an AI Workflow Engine

Traditional state machines create brittle code because they require manual state management for every non-linear transition or failure point. When building AI agents, we rarely deal with a single request-response cycle. Most production-grade agents require multiple steps: searching a vector database, calling an external API, synthesizing results, and often asking a human for clarification. In my initial implementation, I tried to model this as a series of discrete functions triggered by a task queue. The flow looked something like this:


async def run_workflow(task_id: str):
    state = await db.get_state(task_id)
    if state.step == "initial":
        result = await call_llm(state.prompt)
        await db.update_state(task_id, "research", result)
    
    if state.step == "research":
        # What if the LLM needs more info? 
        # I have to manually save progress and exit here.
        if "need_info" in result:
            await db.update_state(task_id, "awaiting_human", result)
            return
        ...

The "state" becomes a sprawling mess. You end up writing more code to manage the state transitions than you do for the actual business logic. Furthermore, if you need to add a new step in the middle of the workflow, you have to migrate your database schema and update every single if/else block. It’s the definition of technical debt. This is specifically problematic when you are trying to implement structured AI agent output with Gemini and Pydantic, as the schema validation often needs to happen at multiple stages of the workflow.

How Python Generators Simplify AI Workflow Engine Architecture

Python generators allow developers to model complex, multi-step processes as a single linear stream of execution that can be paused and resumed using the yield and send() methods. These methods allow a function to pause its execution and return a value to the caller, while maintaining its local variable state and instruction pointer. When you call generator.send(value), the function resumes exactly where it left off.

I realized I could model an AI workflow as a generator. Each yield represents a point where the workflow needs to wait—either for an LLM response, a tool execution, or human input. This allows the code to look like a linear script, even though it might take hours or days to complete.

Designing the Core Workflow Engine

I started by defining a simple WorkflowStep object to pass information between the generator and the runner. The runner is responsible for handling the side effects (like calling APIs or saving to the DB), while the generator handles the logic.


from typing import Generator, Any
from pydantic import BaseModel

class WorkflowEvent(BaseModel):
    event_type: str
    payload: Any

def research_workflow(initial_query: str) -> Generator[WorkflowEvent, Any, str]:
    # Step 1: Initial Analysis
    analysis = yield WorkflowEvent(event_type="llm_call", payload={"prompt": initial_query})
    
    # Step 2: Decide if we need human input
    if "ambiguous" in analysis:
        clarification = yield WorkflowEvent(event_type="human_input", payload={"msg": "Please clarify..."})
        analysis = yield WorkflowEvent(event_type="llm_call", payload={"prompt": clarification})
    
    # Step 3: Final Synthesis
    final_report = yield WorkflowEvent(event_type="llm_call", payload={"prompt": f"Synthesize: {analysis}"})
    return final_report

In this model, the yield statement does two things: it sends a request out to the runner and pauses. When the runner gets the result, it "sends" it back into the generator, which assigns it to the variable on the left (e.g., analysis). This is incredibly powerful. The logic reads top-to-bottom, but the execution can be interrupted at any yield point.

Solving Persistence in a Generator-Based AI Workflow Engine

Persistence in generator-based systems is best achieved by storing a history of inputs and replaying them to reconstruct the execution state. The immediate problem I hit—and the one that usually stops developers from using this pattern—is persistence. Python generators are live objects in memory. You cannot natively serialize (pickle) a generator's execution state to a database and resume it on a different server or after a restart. For a production system running on Cloud Run, where instances are ephemeral, this is a dealbreaker.

I spent three days trying to find a "clever" way to serialize the stack frame before realizing I was over-engineering. The solution wasn't to save the generator object itself, but to build a Replay Engine. Since the generator is deterministic (given the same inputs, it follows the same path), I only needed to store the history of inputs sent to the generator.

Implementing the Replay Runner

When a workflow needs to resume, I instantiate a fresh generator and "fast-forward" it by feeding it all the previous results from the database. Here is the simplified logic I implemented in my FastAPI backend:


async def execute_workflow(workflow_id: str, new_input: Any = None):
    # 1. Fetch history from Postgres
    history = await db.get_history(workflow_id)
    if new_input:
        history.append(new_input)
        await db.save_history(workflow_id, history)

    # 2. Re-instantiate the generator
    gen = research_workflow(initial_query=history[0])
    
    # 3. Fast-forward to the current state
    current_event = None
    try:
        # Skip the first history item as it's the initial_query
        for i, result in enumerate(history[1:]):
            current_event = gen.send(result)
        
        # If we just added a new_input, we might need to trigger the next step
        if not history[1:]:
             current_event = next(gen)
             
    except StopIteration as e:
        return {"status": "completed", "result": e.value}

    # 4. Handle the next event (e.g., call Gemini API)
    return await handle_event(workflow_id, current_event)

This approach is surprisingly efficient. Replaying 50 steps in a generator takes mere milliseconds of CPU time. The "cost" of replaying is negligible compared to the network latency of a single LLM call. This also solved my debugging issues: I could now pull the history for any failed workflow and replay it locally on my machine to see exactly where the logic diverged.

How an AI Workflow Engine Reduces Latency and API Costs

Centralizing workflow logic in a generator allows for aggressive caching of LLM responses and more efficient token management across multiple steps. One of the biggest wins with this architecture was the ability to aggressively optimize API costs. Because I had a clear record of every input and output for every step (the history), I could implement a caching layer at the runner level. If the generator requested an LLM call with a prompt that I had already processed in a previous "replay" or a similar workflow, I could return the cached result immediately.

I also integrated this with my previous work on optimizing LLM API costs for multi-turn conversations. By tracking the token count within the workflow event payload, I could decide when to truncate history or summarize the state before the next yield. My GCP bill for the following month dropped by 65%, even as my traffic increased.

According to the official Python documentation, generators are memory-efficient because they only yield one item at a time. In my case, they are "architecturally efficient" because they decouple the business logic from the infrastructure required to run it.

Improving Error Handling and Branching in AI Workflows

Using native Python try/except blocks within generators allows the workflow to handle its own recovery logic without external state flags. What happens when an API call fails? In a traditional state machine, you’d have to update the state to FAILED_STEP_3 and hope your retry logic is robust. With generators, I used standard Python try/except blocks inside the workflow logic.


def robust_workflow(query: str):
    try:
        data = yield WorkflowEvent(event_type="fetch_data", payload=query)
    except ExternalAPIError:
        # Fallback logic right here in the flow!
        data = yield WorkflowEvent(event_type="fetch_backup_data", payload=query)
    
    # ... continue as normal

To make this work, the runner simply uses gen.throw(ExternalAPIError) instead of gen.send(result). This allows the workflow to handle its own errors using native language constructs. It’s much cleaner than checking status codes after every database fetch.

Key Takeaways for Building a Resilient AI Workflow Engine

Decoupling business logic from side effects is the most effective way to build scalable and observable AI orchestration layers. Here are the core lessons learned from this transition:

  • Generators are for Logic, Runners are for Side Effects: Keep your generator functions pure. They should only yield descriptions of what they want to happen and receive the results. Don't put database calls inside the generator.
  • The Replay Pattern is a Superpower: Don't try to serialize the execution state. Serialize the inputs. This makes your system resilient to code changes (as long as you don't break the order of yields) and incredibly easy to debug.
  • Deterministic Execution is Mandatory: If your workflow uses datetime.now() or random numbers, you must pass those in as inputs or yield for them. If the generator isn't deterministic during replay, it will drift, and the history will no longer match the code path.
  • Human-in-the-Loop is Just Another Yield: Treating a human's response exactly like an API response simplifies the architecture. The runner just stops and waits for a webhook or a UI interaction to provide the next send() value.

Related Reading

Moving forward, I'm looking into how to handle "hot-swapping" code in the middle of a running workflow. Currently, if I change the order of yield statements in research_workflow, old history records become incompatible. I’m experimenting with a versioning decorator that maps specific history schemas to specific function versions. Building a custom AI workflow engine was a significant time investment, but the stability and observability I've gained in my AI agents have made it the most impactful architectural change I've made this year.

Comments

Post a Comment

Popular posts from this blog

Optimizing LLM API Latency: Async, Streaming, and Pydantic in Production

-
Optimizing LLM API Latency: Async, Streaming, and Pydantic in Production I remember the sinking feeling in my stomach. It was late afternoon, and our monitoring dashboards, usually a soothing sea of green, had started to flash angry reds and oranges. Our users were reporting painfully slow response times, some even encountering timeouts. The core of the problem, I quickly pinpointed, was our interaction with external Large Language Model (LLM) APIs. What should have been near-instantaneous content generation was taking upwards of 10-15 seconds, sometimes more. This wasn't just a poor user experience; it was a ticking time bomb for our infrastructure costs, as longer request durations meant more concurrent instances and higher compute bills. This wasn't an isolated incident. As a lead developer, I've seen my fair share of production fires, but this one felt particularly insidious because the underlying issue wasn't immediately obvious. I was dealing with a distribute...
Read more

How I Built a Semantic Cache to Reduce LLM API Costs

-
How I Built a Semantic Caching Layer for AutoBlogger's LLM Responses to Cut API Costs My heart sank as I reviewed our monthly cloud bill for AutoBlogger. While the user growth was fantastic, the cost line item for LLM API usage had skyrocketed past all projections. We were on track for an astronomical bill, easily doubling what we’d paid the previous month. This wasn't just a slight overshoot; it was a full-blown financial alarm bell ringing at 3 AM. Our core feature—generating unique, high-quality blog posts and content snippets—relies heavily on large language models, and with increased usage, direct API calls were burning through our budget at an unsustainable rate. Something had to give, and fast. My immediate thought wasn't about switching providers or optimizing prompts (though we do that too); it was about intelligently reducing the *number* of calls we made in the first place. The Problem: LLM API Costs Eating Our Budget Alive AutoBlogger’s magic lies in its...
Read more

How I Squeezed LLM Inference onto a Raspberry Pi for Local AI

-
I'm writing a DevLog post for the 'AutoBlogger' project, detailing my journey and struggles in getting Large Language Model (LLM) inference to run efficiently on a Raspberry Pi. This post chronicles the specific technical challenges I faced, my initial failures, and the iterative process of optimizing models through quantization and pruning. I'll share concrete examples of the software stack, code snippets, and the architectural choices I made to achieve acceptable performance on edge hardware, emphasizing the personal, expert, and transparent aspects of my development experience. I'll also link this experience to broader trends in small, domain-specific LLMs. From Cloud Bloat to Pi Power: How I Squeezed LLM Inference onto a Raspberry Pi for AutoBlogger When I was building the posting service for AutoBlogger, my vision was clear: a self-contained, low-cost blog automation bot that could generate content, optimize it, and publish, all from a tiny, energy...
Read more