荒谬的工作流：仅使用 Postgres 实现持久执行

Posted on 2025-11-03

2025-11-03-absurd-workflows-social.png

你可能并不惊讶我们正在培养代理商。毕竟，大家都在这么做。然而，培养一个优秀的代理商，却面临着一些由来已久的挑战，比如如何确保代理商能够长期稳定地运作。

不出所料，现在很多人都在构建持久执行系统。然而，其中许多系统极其复杂，需要你注册其他第三方服务。我通常尽量避免引入额外的复杂性，所以我想看看仅使用 Postgres 能做到什么程度。为此，我编写了Absurd 1 ，这是一个小巧的纯 SQL 库，带有一个非常精简的 SDK，可以在 Postgres 之上实现持久工作流——无需任何扩展。

持久执行 101

持久执行（或持久工作流）是一种运行长时间运行且可靠的函数的方法，这些函数即使在崩溃、重启和网络故障的情况下也能正常运行，而不会丢失状态或重复工作。持久执行可以被视为队列系统和状态存储的结合体，状态存储会记住最近一次的执行状态。

由于 Postgres 的SELECT ... FOR UPDATE SKIP LOCKED语句使其在队列方面表现出色，因此您可以将其用于队列（例如，与pgmq配合使用）。而且，由于它本身就是一个数据库，因此您也可以使用它来存储状态。

状态至关重要。持久执行并非将逻辑直接运行在内存中，而是将任务分解成更小的部分（阶跃函数），并记录每一步和每一个决策。当进程停止时（无论是失败、主动暂停还是机器崩溃），引擎可以重放这些事件，从而恢复到之前的状态，并从中断的地方继续执行，就好像什么都没发生过一样。

荒谬至极

Absurd 的核心是一个单独的.sql文件（ absurd.sql ），需要将其应用于您选择的数据库。该 SQL 文件的目标是将 SDK 的复杂性转移到数据库中。SDK 通过抽象底层操作，并充分利用您所使用编程语言的易用性，使系统更加便捷。

该系统非常简单：任务被派发到指定的队列中，然后由工作进程从队列中领取并执行。任务被细分为多个步骤，工作进程按顺序执行这些步骤。任务可能会被暂停或失败，当这种情况发生时，任务会重新执行（即运行）。每个步骤的结果会被存储在数据库中（即检查点）。为了避免重复工作，检查点会自动从Postgres的状态存储中重新加载。

此外，任务可以休眠或暂停以等待事件发生。事件会被缓存，这意味着它们不会发生竞争条件。

与经纪人

智能体与工作流之间有何关系？通常，工作流是由人预先定义的有向无环图（DAG）。而人工智能智能体则会在执行过程中自行定义其运行路径。这意味着它们本质上是一个工作流，其中大部分步骤都是重复执行，直到判断任务完成为止。Absurd 库通过自动计数重复步骤来实现这一点：

荒谬的。registerTask （ { name : "my-agent" }, 异步 （参数， ctx ） => {   
  让 消息 = [{角色： “用户” ， 内容： params.prompt }];   
  让 步 = 0 ；   
  尽管 （步骤++） < 20 ） {   
    常量 { 新消息， 完成原因 } = 等待 ctx.step ( "迭代" , 异步 () => {   
      返回 等待 singleStep （消息）；   
    });   
    messages.push ( ... newMessages ) ;   
    如果 （完成原因） !== “工具调用” ） {   
      休息;   
    }   
  }   
});

这定义了一个名为my-agent单个任务，它只有一个步骤。返回值是更改后的状态，但当前状态作为参数传入。每次执行步骤函数时，首先从检查点存储中查找数据。第一个检查点是iteration ，第二个iteration#2 ， iteration#3 ，依此类推。每个状态仅存储它生成的新消息，而不是完整的消息历史记录。

如果某个步骤失败，则任务失败，程序将重试。由于采用了检查点存储，如果在步骤 5 中崩溃，前 4 个步骤将自动从存储中加载。步骤本身不会重试，只有任务会重试。

如何启动它？只需将其加入队列即可：

等待 荒谬的。生成（ “我的代理”） ， {   
  迅速的： 波士顿的天气怎么样？   
}, {   
  最大尝试次数： 3 ，   
});

如果您感兴趣，以下是上面使用的singleStep函数的一个示例实现：

单步函数

异步 功能 singleStep （消息） {   
  常量 结果 = 等待 生成文本（{   
    模型： 人本主义的（ "claude-haiku-4-5" )，   
    系统： “你是一位很有帮助的代理人” 。   
    消息，   
    工具： {   
      getWeather ： { /* 工具定义 */ }   
    },   
  });   
   
  常量 新消息 = 等待​ 结果.响应).消息；   
  常量 完成原因 = 等待 结果.完成原因;   
   
  如果 （完成原因） === “工具调用” ） {   
    常量 工具结果 = [];   
    为了 （常量） 工具调用 的 结果.工具调用） {   
      /* 在这里处理工具调用 */   
      如果 （工具调用.工具名称 === "getWeather" ) {   
        常量 工具输出 = 等待 获取天气（工具调用.输入）；   
        toolResults.push ( {​   
          工具名称： toolCall.toolName ，   
          toolCallId ： toolCall.toolCallId ，   
          类型： “工具结果” ，   
          输出： {类型： “文本” ， 价值： toolOutput },   
        });   
      }   
    }   
    newMessages.push ( {​   
      角色： “工具” ，   
      内容： 工具结果   
    });   
  }   
   
  返回 { 新消息， 完成原因 };   
}

事件和睡眠

和其他解决方案一样，你也可以选择放弃。如果你想在 7 天后再回来解决这个问题，你可以这样做：

等待 ctx.sleep ( 60 )​ * 60 * 24 * 7 ）； // 睡7天

或者，如果您想等待某个事件发生：

常量 事件名称 = `email-confirmation- ${ userId } ` ；   
尝试 {   
  常量 有效载荷 = 等待 ctx.waitForEvent ( eventName ,​​ {暂停： 60 * 5 });   
  // 处理事件负载   
} 抓住 （ e ） {   
  如果 ( e 实例 超时错误） {   
    // 处理超时   
  } 别的 {   
    扔 e ；   
  }   
}

其他人可以发出：

常量 事件名称 = `email-confirmation- ${ userId } ` ；   
等待 荒唐的。发出事件(事件名称, { 已确认： 新的 Date (). toISOString () });

就是这样！

真的，就这么简单。它其实很简单，只需要一个队列和一个状态存储——仅此而已。不需要编译器插件，也不需要单独的服务或完整的运行时集成。只需要Postgres。这并不是要贬低其他解决方案；它们都很棒。但并非所有问题都需要扩展到如此复杂的程度，其实用更少的资源也能取得不错的成果。特别是如果你想构建一个其他人可以自行托管的软件，这可能就很有吸引力了。

  1. 之所以命名为“荒谬”，是因为持久化工作流程原本极其简单，但近年来却被过度复杂化了。

原文： https://lucumr.pocoo.org/2025/11/3/absurd-workflows/