Icepick：用于构建可扩展AI代理的TypeScript库

Icepick 是一个简单的 TypeScript 库，用于构建具有容错性和可扩展性的 AI 代理。它处理了持久执行、队列和调度的复杂性，让你能够专注于编写核心业务逻辑。它不是一个框架。

Icepick 中的一切都是你所编写的功能，这使得它容易与现有代码库和业务逻辑集成。你可以构建调用工具、其他代理或你定义的任何其他函数的代理：

https://github.com/user-attachments/assets/b28fc406-f501-4427-9574-e4c756b29dd4

不确定 Icepick 是否适合你？预约办公室时间

代码示例

import { icepick } from "@hatchet-dev/icepick";
import z from "zod";
import { myTool1, myTool2 } from "@/tools";

const MyAgentInput = z.object({
  message: z.string(),
});

const MyAgentOutput = z.object({
  message: z.string(),
});

export const myToolbox = icepick.toolbox({
  tools: [myTool1, myTool2],
});

export const myAgent = icepick.agent({
  name: "my-agent",
  executionTimeout: "15m",
  inputSchema: MyAgentInput,
  outputSchema: MyAgentOutput,
  description: "描述这个代理的功能",
  fn: async (input, ctx) => {
    const result = await myToolbox.pickAndRun({
      prompt: input.message,
    });

    switch (result.name) {
      case "myTool1":
        return {
          message: `结果：${result.output}`,
        };
      case "myTool2":
        return {
          message: `另一个结果：${result.output}`,
        };
      default:
        return myToolbox.assertExhaustive(result);
    }
  },
});

开始使用

开始使用只需两个命令：

pnpm i -g @hatchet-dev/icepick-cli
icepick create first-agent

这将提示你从模板创建一个新的 Icepick 项目，以查看 Icepick 的完整使用示例。

如需完整的快速入门，请查看我们的文档。

优势

Icepick 围绕持久执行这一优势展开，这为代理创建自动检查点，使其能够轻松从失败中恢复或等待外部事件很长时间而不会消耗资源。这是通过使用名为Hatchet的持久化任务队列实现的。

此外，Icepick 代理具有以下特点：

• 💻 以代码为中心 - 代理作为代码定义，设计为与你的业务逻辑集成。
• 🌐 分布式 - 所有代理和工具在一组机器上运行，Icepick 优雅地处理调度。当底层机器失败时，Icepick 会负责重新调度和在另一台机器上恢复代理。
• ⚙️ 可配置 - 简单的配置用于重试、速率限制、并发控制等。
• ☁️ 可在任何地方运行 - Icepick 代理可以在任何容器化平台（Hatchet、Railway、Fly.io、Porter、Kubernetes、AWS ECS、GCP Cloud Run）上运行。

可扩展性

Icepick 专为大规模设计：具体而言，是巨大的吞吐量和并行性。Hatchet 已经运行了单次执行生成数万任务的代理工作负载，并且每月运行数十亿任务。

哲学

Icepick 不是一个框架。代理和工具只是你所编写的功能。这意味着你可以选择或构建最佳的记忆、知识、推理或集成。它不强制要求你如何设计工具、调用 LLM 或实现代理记忆等功能。Icepick 对代理的基础设施层持观点，但不对代理的实现细节持观点。

文档

概念

• 概览 - Icepick 执行模型的概述
• 代理 - 代理是调用其他工具和代理的功能
• 工具 - 工具是执行特定任务的功能，可以被代理调用
• 工具箱 - 工具箱是包含工具的集合，具有 AI 驱动的选型能力

API 参考

• icepick.start
• icepick.agent
• icepick.tool
• icepick.toolbox

使用场景和模式

• 提示链
• 路由
• 并行化
• 评估器-优化器
• 多代理
• 人机协作

与其他工具的比较

与框架（Mastra、Voltagent）对比

Icepick 不是框架。它不强制要求你如何结构 LLM 调用、业务逻辑、提示或上下文；我们期望你自行编写这些内容（尽管 Icepick 有一些工具用于工具选择，并捆绑了调用 LLM 的 AI SDK）。Icepick 设计为可扩展和可修改——例如，你可以在 Icepick 上构建自己的代理库。

vs Temporal

Icepick 的执行模型与 Temporal 最相似，具有简化了的执行模型，并提供了更精细的流程调度控制：

特性	Icepick	Temporal
持久执行	✅	✅
流程内事件监听器	✅	✅
基于代码的流程定义	✅	✅
定时任务	✅	✅
单次调度	✅	✅
流控	✅	✅
持久休眠	✅	✅
全局速率限制	✅	❌
基于事件的触发	✅	❌
事件流	✅	❌
有向无环图支持	✅	❌
优先队列	✅	❌
粘性分配/复杂路由逻辑	✅	❌

Agent 最佳实践

在使用 Icepick 编写 Agent 时，建议遵循以下规则：

1. Agent 应该是 无状态的缩减器，且 无副作用。它们不应依赖外部 API 调用、数据库调用或本地磁盘调用；其全部状态应由工具调用的结果决定。详见 技术深度解析 以获取更多信息。

2. 所有工作量应作为任务或工具调用进行调用。

3. 将 LLM 调用视为库，并 负责数据查找：应用程序不应允许无约束的代理工具调用与数据查找结合使用。所有工具调用应验证用户权限，并将数据查找与 LLM 调用分离以确保安全。

贡献

欢迎贡献！在处理任何大于简单 bug 修复的问题前，请先在 Discord 开始讨论。

技术深度解析

Icepick 是建立在 Hatchet 之上的工具层。它基于 持久任务队列 的概念，这意味着 Hatchet 中调用的每个任务都会存储在数据库中。这很有用，因为任务可以轻松回放并从失败中恢复，即使底层硬件崩溃。另一种说法是：Hatchet 使 分布式系统极其容易部署和维护。

对于 Agent 来说，这特别有用，因为它们运行时间极长，因此需要具备对硬件故障的容错能力。Agent 还需要管理第三方速率限制，并通过并发控制防止系统过载。

Agent 的第一条规则是：它们应是 无状态的缩减器，无副作用。要理解原因，需要了解持久执行的一些概念。在核心层面，一个执行持久化的函数会存储其执行到目前为止的所有事件日志。假设一个 Agent 已调用工具 search_documents、get_document，并在处理 extract_from_document 时处于中间状态。其执行历史如下：

事件日志：
-> 开始 search_documents
-> 完成 search_documents
-> 开始 get_document
-> 完成 get_document
-> 开始 extract_from_document...

现在，假设运行 Agent 的机器在最后一步崩溃。为了从失败中恢复，Icepick 会自动回放执行历史中的所有步骤：

事件日志：
-> 开始 search_documents（回放）
-> 完成 search_documents（回放）
-> 开始 get_document（回放）
-> 完成 get_document（回放）
-> 开始 extract_from_document（回放）
-> （稍后）完成 extract_from_document

换句话说，Icepick 缓存了执行历史，这使得 Agent 可以从失败中优雅恢复，而不是重新执行大量工作。另一种说法是：Agent 会自动“检查点”其状态。

当需要等待外部系统（如人工审核或外部事件）时，这种执行模型会更加强大。构建容错系统变得更加困难，因为如果 Agent 从头开始，可能会丢失允许继续执行的事件。在这种模型中，事件会自动存储并回放。

import { icepick } from "@hatchet-dev/icepick";
import z from "zod";
import { myTool1, myTool2 } from "@/tools";

const MyAgentInput = z.object({
  message: z.string(),
});

const MyAgentOutput = z.object({
  message: z.string(),
});

export const myToolbox = icepick.toolbox({
  tools: [myTool1, myTool2],
});

export const myAgent = icepick.agent({
  name: "my-agent",
  executionTimeout: "15m",
  inputSchema: MyAgentInput,
  outputSchema: MyAgentOutput,
  description: "Description of what this agent does",
  fn: async (input, ctx) => {
    const result = await myToolbox.pickAndRun({
      prompt: input.message,
    });

    switch (result.name) {
      case "myTool1":
        return {
          message: `Result: ${result.output}`,
        };
      case "myTool2":
        return {
          message: `Another result: ${result.output}`,
        };
      default:
        return myToolbox.assertExhaustive(result);
    }
  },
});

icepick run my-agent