1. Why Do We Need a Workflow Engine: Starting from the Script Dilemma

When I first used a Python script to call the DeepSeek API, the code was only a few dozen lines and ran smoothly. But as business complexity grew—such as needing multi-turn conversations, knowledge base retrieval, result validation, and error retries—the script began to spiral out of control. Every if-else branch and every try-except made the code difficult to maintain, not to mention parallel execution and visual monitoring. I believe many developers have had similar experiences: scripts run fine locally, but once deployed to production, they become fragile when facing diverse inputs and sudden API errors.

The core value of a workflow engine lies in decoupling the "process" from the "code." It allows you to declaratively define dependencies, branches, and merges between tasks, while the engine handles scheduling, state management, and fault tolerance. This is like moving from handwritten SQL to using an ORM, or from bare functions to microservice orchestration. For AI applications, workflow engines are especially important because calling large model APIs typically involves network latency, cost control, and result uncertainty, all of which require fine-grained process management.

This article will take a practical approach, gradually demonstrating how to evolve from a simple DeepSeek calling script to an event-driven and DAG-based workflow engine. I will share the pitfalls I encountered along the way and provide runnable code examples.

2. Starting Point: A Simple DeepSeek Calling Script

Let's start with the most basic script. Suppose we have a requirement: input a product description, let DeepSeek generate marketing copy, and extract keywords. The Python code to directly call the DeepSeek API is as follows:

import requests
import json

def call_deepseek(prompt, api_key="your-deepseek-api-key"):
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    payload = {
        "model": "deepseek-chat",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.7
    }
    response = requests.post("https://api.deepseek.com/chat/completions", headers=headers, json=payload)
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"]

# Business logic
description = "A portable smart speaker with voice assistant, built-in battery, 12-hour battery life."

# Generate marketing copy
prompt1 = f"Please write an attractive marketing copy for the following product: {description}"
copy = call_deepseek(prompt1)

# Extract keywords
prompt2 = f"Extract 3-5 keywords from the following text: {copy}"
keywords = call_deepseek(prompt2)

print("Marketing copy:", copy)
print("Keywords:", keywords)

This script has two obvious drawbacks: first, the two API calls are serial; if keyword extraction did not depend on the copy, it could be parallel, but here it depends, so it must be sequential. Second, there is no error handling; once network fluctuations or API rate limits occur, the entire script crashes. Of course, we could add try-except and retry, but we would have to write that for every task, and the code quickly becomes redundant.

3. Escalating Difficulties: Multi-Task Orchestration and State Management

In real business, there are often more than two tasks. For example, we might need to perform sentiment analysis on the product description, generate copy, extract keywords, translate into English, and even check content compliance. Do these tasks have dependencies? Sentiment analysis and keyword extraction are independent and can be parallel; copy generation depends on the description; translation depends on the copy. If you write scripts manually, you might use multithreading or async, but data passing between threads, result aggregation, and exception handling will make the code complex.

More critically, you cannot visually monitor the execution status and duration of each task. Once, in production, I encountered occasional API timeouts, but I had no idea which task caused it; I could only guess. This prompted me to seek a more elegant solution.

4. Moving Toward Workflow: Task Abstraction and Graph Model

The core idea of a workflow engine is to abstract each step as a "node," and nodes are connected by "edges" representing dependencies. The entire process is a directed acyclic graph (DAG). Each node can be a function, an API call, or even a sub-workflow. The engine is responsible for traversing the graph, executing in topological order, and passing data through a context object.

Initially, I implemented a simple DAG executor in Python that supports node definition, dependency declaration, and result output. Here is a simplified implementation:

from dataclasses import dataclass
from typing import Callable, Any, Dict
import asyncio

@dataclass
class WorkflowNode:
    name: str
    func: Callable
    depends_on: list

class Workflow:
    def __init__(self):
        self.nodes = {}
        self.results = {}

    def add_node(self, name, func, depends_on=None):
        self.nodes[name] = WorkflowNode(name, func, depends_on or [])

    async def execute(self):
        # Simplified topological sort, assuming no cycles and valid order
        for node in self.nodes.values():
            # Wait for dependencies to complete
            for dep in node.depends_on:
                while dep not in self.results:
                    await asyncio.sleep(0.1)  # Simple polling
            # Execute node
            inputs = {dep: self.results[dep] for dep in node.depends_on}
            self.results[node.name] = await node.func(**inputs)
        return self.results

This implementation is crude but functional. It uses asyncio for async and polls for dependencies. In real engineering, we would use more mature workflow frameworks like Airflow, Prefect, or Temporal, which provide ample scheduling, retry, and monitoring capabilities. But for AI workflows, we often need special support, such as dynamic branching (based on LLM output to decide subsequent flow), human-in-the-loop (requiring manual review), etc.

5. Engineering Practice: Event-Driven AI Workflow

In production, I ultimately chose Prefect as the engine because it is Python-based, easy to customize, and natively supports async and event triggers. My architecture is to encapsulate each AI call as a Prefect task, and tasks pass parameters between them. For example, I defined a task named generate_copy that calls the DeepSeek API, and the extract_keywords task receives the copy as input.

But Prefect's default scheduling is flow polling, which is not fast enough. So I switched to event-driven: push new tasks to the engine via a message queue (like Redis Streams), and the engine triggers the corresponding flow. This way, each user request is an independent flow instance, unaffected by others. At the same time, I store the flow's checkpoints (state, results) in a database for UI display.

Here is a key engineering pitfall: API idempotency and retry strategy. The DeepSeek API occasionally returns 429 (rate limit) or 5xx, so we must implement exponential backoff retry in the workflow. But retries can cause duplicate execution; if a task has side effects (like sending emails), we need to implement idempotency. My approach is to assign a globally unique ID to each task and record the execution result, checking before retry whether it has already succeeded.

6. The Value of Visualization: Making Processes Transparent and Controllable

The biggest benefit of moving from scripts to visual orchestration is transparency. Through the Prefect UI or a custom frontend, I can see the running status, duration, input/output of each flow in real time, and even manually rerun failed nodes. This is especially important for debugging AI-generated quality issues.

For example, once a user reported that a certain copy was inappropriate. In the script era, I could only rerun the entire flow, but I couldn't determine whether it was a prompt issue or the model temperature. With the workflow, I could view the specific input and parameters of that node, reproduce the problem, and adjust the prompt or temperature accordingly. This capability is extremely valuable in AI application development because LLM outputs are non-deterministic, and we need observability to locate issues.

Furthermore, visual orchestration promotes team collaboration. My colleagues (without deep programming backgrounds) can also use the DAG editor to modify workflow logic, such as adjusting node order or adding new processing steps. This greatly lowers the barrier to AI applications.

7. Real-World Case Analysis: A Complete AI Workflow

Next, I will share a real case. We built a "smart customer service ticket analysis" system for a client, with the following flow:

  1. Event listening: receive new tickets.
  2. User intent classification (DeepSeek classifier).
  3. Sentiment analysis (DeepSeek sentiment model).
  4. Knowledge base matching: query vector database based on classification.
  5. Generate reply draft (DeepSeek generator).
  6. Manual review (event-driven suspension).
  7. Send reply.

In this flow, intent classification and sentiment analysis can be parallel; knowledge base matching depends on classification results; draft generation depends on matching and sentiment. We define each step as a Prefect task and use Redis Streams to trigger flow instances. The key code is as follows (simplified):

from prefect import flow, task, get_run_logger
from prefect.tasks import exponential_backoff

@task(retries=3, retry_delay_seconds=exponential_backoff(backoff_factor=2))
def sandbox_analysis(desc: str):
    # Call DeepSeek sentiment analysis
    ...

@task
async def kb_match(category: str):
    # Vector database query
    ...

@flow
async def process_ticket(ticket_id: str):
    logger = get_run_logger()
    ticket = fetch_ticket(ticket_id)
    cat_task = classify_async.submit(ticket.desc)
    senti_task = sentiment_async.submit(ticket.desc)
    cat, senti = await cat_task.result(), await senti_task.result()
    kb_results = await kb_match.submit(cat).result()
    draft = await generate_draft.submit(ticket.desc, cat, senti, kb_results).result()
    logger.info(f"Draft ready for {ticket_id}")
    # Suspend for manual review
    await wait_for_review(ticket_id, draft)
    send_reply(ticket_id, draft)

This flow has a clear DAG representation in the workflow engine, and each node has logs and observability.

8. Engineering Pitfalls and Solution Summary

During the transition from scripts to workflow, I encountered many challenges. Here are some points to help readers avoid them:

  • Dependency issues: Some third-party libraries (like prefect) are incompatible with certain Python versions. It is recommended to use a virtual environment and lock versions.
  • Async execution traps: Some Prefect tasks are synchronous by default. If you call an async function in a synchronous task, it will block the event loop. You need to use asyncio.run() or explicitly declare it as an async task.
  • Data serialization: Data passed between nodes must be serializable. The JSON returned by the DeepSeek API is safe, but if it is a custom object, you need to convert it to a dictionary or use cloud storage.
  • Monitoring and alerting: Although workflow engines provide a UI, it is best to integrate with Prometheus + Grafana to record task latency, success rate, and set up alerts. Otherwise, you might be the last to know when a flow gets stuck.
  • Cost control: In AI workflows, token consumption is a major cost. It is recommended to record token usage at the node level and dynamically adjust model or sampling parameters through workflow configuration to avoid unnecessary retries of high-cost tasks.

9. Summary and Outlook

Evolving from a simple script to visual orchestration is not just an upgrade of the technology stack, but also a shift in mindset. Workflow engines allow us to view AI applications as production lines, where each step is controllable, testable, and optimizable. In practice, I strongly recommend developers, especially those with some experience in AI calls, to adopt workflow thinking early on; the engineering benefits are enormous.

In the future, AI workflow engines will become more intelligent: for example, adaptively adjusting prompts, automatically caching similar results, and dynamically degrading on exceptions (like switching to cheaper models). The DeepSeek ecosystem also provides rich APIs and models, and we can combine these capabilities in workflows to build powerful systems. I hope this article inspires you, and feel free to share your practical experiences in the comments.