工作流运行在 Python 的 asyncio 事件循环上。运行时使用协作式多任务：当某个步骤正在 await（例如等待 LLM 响应或网络调用）时，其他步骤与工作流可以自由推进。

如果你刚接触 Python 的异步编程，请先读异步 Python 入门，了解 asyncio、事件循环与 await 的总体图景。

步骤既可以定义成 async def，也可以定义成普通 def。本页说明两者各自的行为，以及如何在阻塞型或 CPU 密集型工作上不把事件循环拖住。

同步步骤（用 def 而不是 async def）

工作流步骤可以定义成普通的 def 函数而非 async def。运行时遇到同步步骤时，会自动用 asyncio.get_event_loop().run_in_executor() 把整个函数交给默认线程池，因此事件循环永不被阻塞：

from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
import requests

class SyncStepWorkflow(Workflow):
    @step
    def fetch_data(self, ev: StartEvent) -> StopEvent:
        # This runs in a thread automatically, so the event loop stays free
        response = requests.get("https://api.example.com/data")
        return StopEvent(result=response.json())

当步骤体完全是同步的，这是最简单的选择。框架替你处理了线程卸载，其中也包括跨线程边界保留 contextvars。

不过，有些场景下你在 async def 步骤内部仍需更细粒度的控制，例如只有一部分逻辑是阻塞的，或者你想为 CPU 密集型工作使用专属的执行器。下面几节会讲这些情形。

异步步骤中的阻塞 I/O

许多 Python 库只提供同步 API：数据库驱动、HTTP 客户端、文件系统操作、SDK 调用等等。当你需要在 async def 工作流步骤里使用它们时，请用 asyncio.to_thread 把调用卸载到线程池：

import asyncio
import requests
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent

class BlockingIOWorkflow(Workflow):
    @step
    async def fetch_data(self, ev: StartEvent) -> StopEvent:
        # Bad: this blocks the event loop until the request completes
        # response = requests.get("https://api.example.com/data")

        # Good: run the blocking call in a thread so the event loop stays free
        response = await asyncio.to_thread(
            requests.get, "https://api.example.com/data"
        )
        return StopEvent(result=response.json())

asyncio.to_thread 会把函数排到默认的 ThreadPoolExecutor 上并返回一个可 await 的对象。阻塞调用在另一个线程里跑的同时，事件循环可以继续处理其他步骤与工作流。

任何做 I/O 的同步库调用都适用：读文件、查数据库、调用外部 API 等等：

import asyncio
import json
from pathlib import Path
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent

def read_large_file(path: str) -> dict:
    """A synchronous function that reads and parses a large JSON file."""
    return json.loads(Path(path).read_text())

class FileReaderWorkflow(Workflow):
    @step
    async def process_file(self, ev: StartEvent) -> StopEvent:
        data = await asyncio.to_thread(read_large_file, ev.file_path)
        return StopEvent(result=data)

CPU 密集型操作

CPU 密集型工作（例如数据转换、图像处理、数值计算）带来的是另一类挑战。即使放到线程里跑，CPU 密集的 Python 代码仍会因为 GIL（全局解释器锁）而与事件循环争抢。

对于 CPU 密集的工作，请使用一个专属、更小的线程池（或进程池），让这类任务排队而不是把默认执行器占满：

import asyncio
from concurrent.futures import ThreadPoolExecutor
from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent

# A small, dedicated pool for CPU-bound work.
# Keeping this small ensures CPU tasks are queued rather than
# overwhelming the system with parallel CPU-bound threads.
cpu_pool = ThreadPoolExecutor(max_workers=2)

def expensive_computation(data: str) -> str:
    """A CPU-intensive operation, e.g. data parsing or transformation."""
    # Simulate heavy work
    result = data
    for _ in range(1_000_000):
        result = result.strip()
    return result

class ComputeEvent(Event):
    data: str

class CPUWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> ComputeEvent:
        return ComputeEvent(data=ev.input_data)

    @step
    async def compute(self, ev: ComputeEvent) -> StopEvent:
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(cpu_pool, expensive_computation, ev.data)
        return StopEvent(result=result)

与 I/O 场景的关键区别：

• 使用 loop.run_in_executor 并显式给出执行器，而不是 asyncio.to_thread，这样你能控制池的大小与类型。
• 保持池足够小。 1–2 个 worker 的池意味着 CPU 任务会排队，而不是相互争抢 CPU 时间。请根据你的负载与可用核心数调整。
• 考虑 ProcessPoolExecutor，把真正 CPU 密集、希望跑在 GIL 之外的工作放进去。API 完全相同，只需换掉执行器类型：

from concurrent.futures import ProcessPoolExecutor

cpu_pool = ProcessPoolExecutor(max_workers=2)

注意，提交给 ProcessPoolExecutor 的函数必须是可 pickle 的（顶层函数，而非 lambda 或闭包）。

总结

场景	解法	原因
整个步骤都是同步的	把步骤定义成 def 而非 async def	运行时会自动把它放到线程池里执行
async def 步骤里有阻塞调用	await asyncio.to_thread(fn, ...)	I/O 在线程中完成期间释放事件循环
CPU 密集型工作	await loop.run_in_executor(pool, fn, ...) 配一个小的专属池	让重计算排队，不至于让事件循环或其他任务挨饿

核心原则很简单：绝不要阻塞 asyncio 事件循环。 完全同步的步骤就用普通 def，把线程的事交给框架；async def 步骤里的阻塞调用则卸载到线程或进程，再 await 其结果。