我要投稿

别再迷信 Agent 框架了，你缺的其实是一套“线束”

发布日期：2026-03-04 08:04:38 浏览次数： 1564

作者：知识发电机

微信搜一搜，关注“知识发电机”

你的 Agent 需要的是"线束"，而不是"框架"

在每一个工程学科中，线束（harness）的含义都是一样的：它是连接、保护和编排各组件的那一层——它本身不做具体的事，但少了它，一切都乱了套。汽车线束在发动机、传感器和仪表盘之间传递信号；测试线束提供脚手架，让代码具备可重复性和可观测性；安全带在你跌落时托住你。

Agent 运行时也需要同一样东西。LLM 是引擎，工具是外设，记忆是存储。但谁来连接它们？当 LLM 在第五次迭代时超时，谁来兜底？谁来防止两条消息相互冲突？谁来把一个 webhook 事件路由到正确的处理函数、再路由到正确的回复通道？

这就是线束。而现在每一个 Agent 框架，都在从零开始构建自己的线束——自己的重试逻辑、自己的状态持久化、自己的任务队列、自己的事件路由。

持久化的、事件驱动的基础设施早就解决了这个问题。 每一次 LLM 调用或工具调用都变成一个"步骤"——一个可独立重试的工作单元。如果进程在第五次迭代时崩溃，前四次迭代已经持久化了。事件在函数之间路由触发器。并发控制防止冲突。步骤级别的追踪让你对 Agent 循环的每一次迭代都有完整的可观测性。基础设施本身就是线束。

我们构建了 Utah——Universally Triggered Agent Harness（通用触发式 Agent 线束）——来验证这个设想。这是一个支持 Telegram 或 Slack 的对话式 Agent，具备工具调用、记忆、子 Agent 委托和完整持久化能力。极简 TypeScript，无框架依赖。只用 Inngest 的函数、步骤和事件，为标准的思考 → 行动 → 观察循环提供线束。可以把它理解成一个持久化的、云原生的 OpenClaw。

"通用触发"这个词很重要：Telegram 或 Slack 的 webhook、定时任务、子 Agent 调用、函数间事件——Agent 不知道也不在乎自己是如何被激活的。触发器与工作解耦。明天想加一个 Slack 机器人？Agent 循环一行不用改，线束负责路由。

下面是它的工作原理。

架构设计

Utah 与大多数线束的不同之处在于：它是事件驱动的，将编排层与 Agent 循环解耦。它还利用 Inngest Cloud 来桥接公共 webhook 与本地 worker 之间的鸿沟。

Inngest Utah 项目架构图

Telegram 或 Slack 的 webhook 打到 Inngest Cloud，一个 webhook transform 将原始 HTTP 负载转换为类型化的 Inngest 事件。本地运行的 worker 接收事件，执行 Agent 函数，并发出 reply 事件，触发另一个独立函数通过各自频道的 API 发回响应（详见下文）。任何支持 webhook 的通信渠道（或任何服务）都可以接入。

worker 使用 Inngest 的 connect() API，从你的本地机器（或 Mac Mini，或远程服务器）建立一个持久的 WebSocket 连接到 Inngest Cloud，无需暴露公共端点。

worker 中运行的 Agent 循环很简单：一个带"步骤"的 while 循环，步骤负责调用 LLM 和执行工具。我们使用 Pi 的 provider 接口和工具，两者都很好用，但你可以替换成任何东西——AI SDK、TanStack AI、自定义工具，或者接入 MCP。

为什么本地运行还要用 Inngest？为什么不直接用 OpenClaw？

OpenClaw 和 pi coding-agent 库是这个项目的灵感来源。它们都在进程内使用事件，事件和编排在内存中处理。Inngest 本身是一个事件驱动的编排层，这个项目将执行与编排解耦。

这为线束带来了以下能力：

• 编排层通过 trace 和步骤级别的检查提供可观测性
• 内置的持久化执行提供可靠性和重试能力
• 解耦为多人分布式 Agent 编排奠定基础
• 事件历史记录提供系统行为的审计轨迹
• 调度内置支持，通过 cron 或定时/延迟函数实现

这些问题都是基础设施问题，而不是 AI 问题。

将 Agent 循环变成步骤

Utah 的核心是一个思考 → 行动 → 观察循环。每次迭代调用 LLM，检查是否需要使用工具，执行工具，并将结果反馈回去。关键洞察：每次 LLM 调用和每次工具执行都是一个 Inngest step。

// 简化版——实际实现使用 pi-ai 的 provider 无关类型
while (!done && iterations < config.loop.maxIterations) {
  iterations++;

  // 裁剪旧的工具结果，保持上下文聚焦
  pruneOldToolResults(messages);

  // 迭代次数快用完时注入预算警告
  const messagesForLLM = addBudgetWarning(messages, iterations);

  // 思考：调用 LLM
  const llmResponse = await step.run("think", async () => {
    return await callLLM(systemPrompt, messagesForLLM, tools);
  });

  const toolCalls = llmResponse.toolCalls;

  if (toolCalls.length > 0) {
    messages.push(llmResponse.message);

    // 行动：将每个工具调用作为独立步骤执行
    for (const tc of toolCalls) {
      const result = await step.run(`tool-${tc.name}`, async () => {
        validateToolArguments(tool, tc);
        return await executeTool(tc.id, tc.name, tc.arguments);
      });
      // 观察：把结果反馈进消息列表
      messages.push(toolResultMessage(tc, result));
    }
  } else if (llmResponse.text) {
    // 没有工具调用——文本响应即最终回复
    finalResponse = llmResponse.text;
    done = true;
  }
}

几点值得注意：

Inngest 自动对重复步骤 ID 建立索引。 当 step.run("think") 在循环中被调用十次时，Inngest 内部将它们追踪为 think:0、think:1……你不需要自己管理唯一步骤 ID，SDK 会处理。

每个步骤可以独立重试。 如果 LLM API 在第 3 次迭代返回 500，Inngest 只重试那个特定步骤。第 1、2 次迭代的结果已经持久化，不会重新执行。这正是持久化执行的设计初衷，只不过这里应用在 Agent 循环上，而不是结账流程。

有文本响应就意味着完成。 当 LLM 只返回文本而不带工具调用时，本轮结束。无需显式的"完成"信号。

你不需要自己造工具

Utah 没有手写文件 I/O 和 shell 执行，而是引入了 pi-coding-agent——来自 OpenClaw/Pi 生态的经过实战检验的工具实现：

• read、write、edit——文件操作，支持图片、二进制检测、智能截断（edit 工具对于上下文窗口管理尤为出色）
• bash——shell 执行，可配置超时和输出截断
• grep、find、ls——尊重 .gitignore 的搜索与导航

在这些基础上，Utah 额外添加了几个自定义工具：remember（将笔记持久化到每日日志）、web_fetch 和 delegate_task（后面详述）。

重点：AI Agent 的工具故事和其他软件一样——用现有的库，用 Inngest step 包装，完事。

import { createReadTool, createWriteTool, createBashTool, /* ... */ } from "@mariozechner/pi-coding-agent";

const tools = [
  createReadTool(config.workspace.root),
  createWriteTool(config.workspace.root),
  createBashTool(config.workspace.root),
  // ...
];

复制，粘贴，即刻上手。

六个函数，而不是一个大单体

六个关键函数通过事件解耦

Utah 不是一个包揽一切的单一函数，而是六个通过事件通信的函数：

const functions = [
  handleMessage,      // 主 Agent 循环
  sendReply,          // 将响应发回频道
  acknowledgeMessage, // 打字指示器——立即触发
  failureHandler,     // 跨所有函数的全局错误处理器
  heartbeat,          // 周期性定时心跳检查
  subAgent,           // 通过 step.invoke() 隔离运行的子 Agent
];

这种分离很重要。打字指示器在消息到达时立即触发，不等待 Agent 循环。reply 函数处理 Telegram/Slack 特定的格式化和错误处理（比如当 LLM 生成格式错误的 HTML 时降级为纯文本）。failure handler 捕获所有函数的未处理异常并通知用户。

每个函数都有自己的重试策略、并发控制和触发条件。这在 Inngest 中很自然——你是在用小型、专注的函数组合行为，用事件连接它们。

而那个 sendReply 函数？它可以从任何地方触发——如果想让子 Agent 或扇出工作流在循环中途向用户发送进度更新，只需从一个新工具发送事件即可。

通过 step.invoke() 实现子 Agent

有时 Agent 需要完成一个大到会撑爆上下文窗口的任务——重构一个文件、研究一个主题、写一份文档。对于像 OpenClaw 这样运行在单线程对话（如 Telegram）中的通用 Agent，几天内跨多个长会话可能会遇到上下文窗口问题。解决方案：派生一个子 Agent。

Utah 有一个 delegate_task 工具。当主 Agent 调用它时，它使用 step.invoke() 启动一个完全独立的 Agent 函数运行。子 Agent 将当前会话的上下文 fork 到自己的子会话（有自己的 session key），聚焦于一个特定任务和预期结果：

// 在主 Agent 循环中，当 delegate_task 被调用时：
const subResult = await step.invoke("sub-agent", {
  function: subAgent,
  data: {
    task: tc.arguments.task,
    subSessionKey: `sub-${sessionKey}-${Date.now()}`,
  },
});

子 Agent 函数以自己全新的上下文窗口运行一个独立的 Agent 循环，拥有同样的工具（但不包括 delegate_task——不允许递归派生），并将摘要返回给父 Agent：

// 简化版子 Agent
export const subAgent = inngest.createFunction(
  { id: "agent-sub-agent", retries: 1 },
  { event: "agent.subagent.spawn" },
  async ({ event, step }) => {
    const { task, subSessionKey } = event.data;
    const agentLoop = createAgentLoop(task, subSessionKey, {
      tools: SUB_AGENT_TOOLS, // 不含 delegate_task
      isSubAgent: true,
    });
    return await agentLoop(step);
  }
);

这正是 step.invoke() 的设计用途——将另一个 Inngest 函数作为步骤调用，等待其结果，然后继续。子 Agent 有自己的重试、自己的步骤级可观测性、自己的持久化执行。父 Agent 看到的只是一个工具结果："我完成了，以下是结果。"

编排已处理妥当。无需任何 Agent 间协议，就是函数调用函数。

单例并发：同一时间只处理一个对话

每个"频道"（如 Slack）使用频道特定的 session key 来定义什么是一个"对话"。对于单线程频道（如 Telegram），是 chat id；对于有线程的平台（如 Slack），是频道加线程组合。

如果一个对话中先后发来多条消息，你不希望第一个 Agent 循环继续跑、然后第二个循环来响应——你希望 Agent 能拿到两条消息的完整上下文。要么取消第一个循环让第二个接手，要么在循环内处理"转向"。这个项目选择了"取消+重启"方案，因为重启后的循环携带了所有上下文，逻辑最清晰。

在消息处理函数上，我们用一行配置搞定：

singleton: { key: "event.data.sessionKey", mode: "cancel" },

这里发生了两件事：

1. 单例并发，以 sessionKey 为键——同一个聊天同一时间只有一个 Agent 在运行。没有竞争条件，没有响应交错。
2. 新消息时取消当前运行——如果用户在 Agent 仍在处理时发来新消息，当前运行被取消，新运行以最新消息重新开始。

在传统方案中，你需要为每个用户构建任务队列、管理锁、自己处理取消。用 Inngest，一行配置。

踩坑总结

上下文管理才是真正的挑战

最难的问题不是调用 LLM，而是管理 LLM 调用的输入内容。

Utah 的工具可能每次调用返回数千个字符。迭代几次后，对话上下文急剧膨胀，模型开始失去追踪。我们看到 Agent 无限循环调用工具，却始终无法产出响应。

我们用两级上下文裁剪解决了这个问题：

const PRUNING = {
  keepLastAssistantTurns: 3,
  softTrim: { maxChars: 4000, headChars: 1500, tailChars: 1500 },
  hardClear: { threshold: 50_000, placeholder: "[Tool result cleared]" },
};

旧的工具结果会被软裁剪（保留头部+尾部）或在总上下文过大时被硬清除。最近三次迭代始终保留完整。

在此之上，还有一套针对会话本身的压缩系统——当估计 token 数超过阈值时，在下一次运行开始前对对话历史进行摘要。裁剪处理单次运行内的上下文，压缩处理跨运行的积累。

我们还加入了预算警告——当 Agent 快达到迭代上限时注入系统消息，告诉它该收尾了。以及溢出恢复：如果 LLM 在运行中途返回"上下文太大"错误，强制压缩消息并重试，不浪费一次迭代。裁剪、压缩、预算压力、溢出恢复四管齐下，Agent 始终保持在轨道上。

多 Provider LLM 支持

Utah 不直接调用 Anthropic SDK，而是使用 pi-ai——一个支持 Anthropic、OpenAI 和 Google 的 provider 无关 LLM 抽象。切换 provider 只是改一个配置项：

llm: {
  provider: "anthropic", // 或 "openai" 或 "google"
  model: "claude-sonnet-4-20250514",
},

未来，如果子 Agent 进化到使用不同模型乃至不同 provider，这个设计也很有趣——编码子 Agent 用 Codex，研究子 Agent 用 Opus。更多内容即将到来。

"转向"仍是未解难题

当用户在 Agent 运行到一半时发来新消息，该怎么办？我们用单例模式——取消当前运行，启动新运行。这能用，但进行中的工作会丢失。新运行从持久化的会话状态中恢复，但并不无缝。这是我们正在积极探索的领域。

流式输出和循环中途实时更新的机会

每个 Inngest 步骤都是原子的——它运行、产出结果、结果被持久化。这个项目尚未加入流式支持，也没有利用 Inngest 的 Realtime 功能。Telegram 和 Slack 支持独立事件，但我们想为这个项目加一个 Web App 和 TUI，来探索如何向支持流式的客户端发送循环中途的进度更新。后续还有更多迭代。