能否在Python解释器重启后序列化并恢复原生async def函数？

我有如下异步代码：

async def start(self):
    name = await self.text_prompt("Hi there! What's your name?")
    await self.send_message(f"Hello, {name}! How are you?")
    # --- crash ---
    await asyncio.sleep(1)
    await self.send_message("Lol")

我希望让运行器在解释器崩溃后仍能稳健运行，理想情况下，若解释器在send_message执行后崩溃，重启后协程能从# --- crash ---标记处恢复执行。目前我通过改写为Step对象列表的方式实现了需求（序列化Step对象跟踪进度），想知道：仅使用原生async def风格能否实现该需求？

可以实现，但需要手动处理状态持久化和断点恢复逻辑，核心思路是把协程的执行进度、依赖变量序列化保存到持久化存储（文件、数据库等），重启后根据保存的状态跳过已执行步骤，加载变量后从断点处继续。

具体实现要点：

• 定义状态跟踪机制，记录已完成的步骤标识和需要保留的变量
• 每个关键步骤执行完成后，立即序列化保存当前状态
• 协程启动时先读取持久化状态，若存在未完成进度，直接跳过已执行步骤，加载变量后继续

以下是具体实现示例：

import asyncio
import pickle
import os

STATE_FILE = "coro_state.pkl"

class YourClass:
    async def text_prompt(self, msg):
        # 模拟获取用户输入
        return "Alice"
    
    async def send_message(self, msg):
        print(msg)

    async def start(self):
        # 加载之前的状态
        state = self._load_state()
        completed_steps = state.get("completed_steps", set())
        saved_vars = state.get("saved_vars", {})

        # 步骤1：获取用户名
        if "get_name" not in completed_steps:
            name = await self.text_prompt("Hi there! What's your name?")
            saved_vars["name"] = name
            completed_steps.add("get_name")
            self._save_state(completed_steps, saved_vars)
        else:
            name = saved_vars["name"]

        # 步骤2：发送问候
        if "send_greeting" not in completed_steps:
            await self.send_message(f"Hello, {name}! How are you?")
            completed_steps.add("send_greeting")
            self._save_state(completed_steps, saved_vars)

        # --- crash --- 断点位置
        # 步骤3：等待1秒
        if "sleep_1s" not in completed_steps:
            await asyncio.sleep(1)
            completed_steps.add("sleep_1s")
            self._save_state(completed_steps, saved_vars)

        # 步骤4：发送最后消息
        if "send_lol" not in completed_steps:
            await self.send_message("Lol")
            completed_steps.add("send_lol")
            self._save_state(completed_steps, saved_vars)

        # 执行完成后清理状态文件
        if os.path.exists(STATE_FILE):
            os.remove(STATE_FILE)

    def _load_state(self):
        if os.path.exists(STATE_FILE):
            with open(STATE_FILE, "rb") as f:
                return pickle.load(f)
        return {}

    def _save_state(self, completed_steps, saved_vars):
        state = {
            "completed_steps": completed_steps,
            "saved_vars": saved_vars
        }
        with open(STATE_FILE, "wb") as f:
            pickle.dump(state, f)

async def main():
    obj = YourClass()
    await obj.start()

if __name__ == "__main__":
    asyncio.run(main())

注意细节 #

• 序列化方案：pickle使用方便但有安全风险，生产环境可改用JSON（需处理复杂对象序列化）或其他安全方案
• 状态保存时机：必须在步骤成功执行完成后立即保存，避免未完成步骤被误标记为已完成
• 幂等性：确保重复执行步骤不会产生副作用（比如send_message需做幂等处理，或在状态中记录已发送消息ID）

这种方式完全基于原生async def风格，无需拆分Step对象，但需要手动编写状态跟踪和恢复逻辑，代码量比Step模式多，但能满足崩溃恢复需求。

内容的提问来源于stack exchange，提问作者Claudiu