实战：把一个串行 Agent 循环改造成低延迟执行器：Node.js + Responses API + WebSocket

如果你最近在做 coding agent、自动化修复脚本，或者任何“模型 + 工具调用 + 多轮继续”的应用，大概率已经感受过一个问题：模型本身不算慢，慢的是整条执行回路。一次任务里要读文件、跑命令、回传结果、继续推理，只要你还是按“HTTP 请求一轮一轮串行发”的朴素写法做，用户很快就会开始怀疑人生。

这篇文章我不打算讲抽象概念，直接给一个最小可用的改造思路：把一个原本串行、每轮都重建上下文的 agent loop，改成基于 Responses API 和 WebSocket 的低延迟执行器。目标不是做出一个完整 IDE，而是做出一个你可以继续扩展的骨架。

先说结论：这类改造最值钱的，不是“换了个协议”，而是三件事一起发生——连接不再频繁重建、状态可以沿用、工具回路可以更平滑地继续。你会发现，很多体感上的卡顿，并不是智力问题，而是消息来回搬运得太笨。

一、先看最常见的低配写法

很多人最初的 agent loop 大概是这样的：

while (!done) {
  const response = await client.responses.create({
    model: "gpt-5.4",
    input: history,
    tools,
  });

  const toolCalls = extractToolCalls(response);

  for (const call of toolCalls) {
    const result = await runTool(call);
    history.push(call, result);
  }

  if (toolCalls.length === 0) {
    done = true;
    console.log(response.output_text);
  }
}

它能跑，但问题也很明显。

• 每一轮都把完整 history 重新发出去。
• 工具一多，串行等待很重。
• 服务端每次都要重新理解大量没变化的上下文。
• 你很难把“当前回合状态”和“历史状态”拆开管理。

如果你只是做玩具 demo，这样没问题。可一旦用户真的拿它干活，延迟和不稳定会立刻浮上来。

二、这次改造的目标是什么

我们要做的不是重写整个系统，而是把执行层换一种思路：

• 使用 WebSocket 持久连接，减少每轮请求的额外开销。
• 通过 previous_response_id 延续响应状态，而不是每次手工回放全部历史。
• 把工具调用封装成独立执行器，便于后续做权限控制、超时和回放。
• 保留日志，让每次工具调用都能被观察和复现。

你可以把这理解成：从“聊天式脚本”升级成“真正的任务执行循环”。

三、准备一个最小工具集

下面这个例子只做两个本地工具：读取文件和执行只读 shell 命令。注意，这里我故意保持保守，因为一上来就给 agent 无限 shell 权限，通常不是工程自信，而是工程鲁莽。

import fs from "node:fs/promises";
import path from "node:path";
import { execFile } from "node:child_process";
import { promisify } from "node:util";

const execFileAsync = promisify(execFile);
const WORKSPACE = process.cwd();

async function readFileSafe(relativePath) {
  const fullPath = path.resolve(WORKSPACE, relativePath);
  if (!fullPath.startsWith(WORKSPACE)) {
    throw new Error("path is outside workspace");
  }
  const content = await fs.readFile(fullPath, "utf8");
  return content.slice(0, 12000);
}

async function runCommandSafe(command, args = []) {
  const allowlist = new Set(["ls", "pwd", "cat", "rg", "git"]);
  if (!allowlist.has(command)) {
    throw new Error(`command not allowed: ${command}`);
  }

  const { stdout, stderr } = await execFileAsync(command, args, {
    cwd: WORKSPACE,
    timeout: 15000,
    maxBuffer: 1024 * 1024,
  });

  return { stdout, stderr };
}

这里先别追求“强大”，先追求“边界明确”。你后面可以继续加写文件、跑测试、提交 patch，但建议每加一个能力，都先把权限模型想清楚。

四、定义给模型看的工具描述

const tools = [
  {
    type: "function",
    name: "read_file",
    description: "Read a UTF-8 text file from the current workspace",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "Relative file path" }
      },
      required: ["path"],
      additionalProperties: false
    }
  },
  {
    type: "function",
    name: "run_command",
    description: "Run a read-only shell command in the workspace",
    parameters: {
      type: "object",
      properties: {
        command: { type: "string" },
        args: {
          type: "array",
          items: { type: "string" }
        }
      },
      required: ["command"],
      additionalProperties: false
    }
  }
];

别小看这一层。工具描述写得含糊，模型就会把你的执行器当许愿池。描述越清楚，越能减少无意义调用和来回修正。

五、用 WebSocket 建一个持续会话

官方工程文章已经透露了一个很关键的方向：WebSocket 模式下，服务端可以在连接范围内缓存前一轮 response 状态，从而减少 follow-up 请求里重复处理的开销。对我们这种工具型工作流，这正是最值钱的地方。

import WebSocket from "ws";

function connectRealtime() {
  const ws = new WebSocket(
    "wss://api.openai.com/v1/realtime?model=gpt-5.4",
    {
      headers: {
        Authorization: `Bearer ${process.env.OPENAI_API_KEY}`,
        "OpenAI-Beta": "realtime=v1"
      }
    }
  );

  return new Promise((resolve, reject) => {
    ws.once("open", () => resolve(ws));
    ws.once("error", reject);
  });
}

这里有两个现实提醒。第一，不同模型和接口能力要以你当前账号下的官方文档为准，别把示例代码当成一成不变的协议说明。第二，就算你用了 WebSocket，也不代表所有问题都消失了；它只是给你减少重复工作提供了一条更好的通道。

六、建立一个可继续的 agent loop

核心思路是：第一次创建 response，后面每一轮都只提交新输入和 previous_response_id，让服务端沿用状态。模型如果发起工具调用，我们执行后再把工具结果作为新的输入项继续送回去。

import OpenAI from "openai";

const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

async function executeTool(call) {
  const args = JSON.parse(call.arguments || "{}");

  if (call.name === "read_file") {
    const content = await readFileSafe(args.path);
    return { type: "function_call_output", call_id: call.call_id, output: content };
  }

  if (call.name === "run_command") {
    const result = await runCommandSafe(args.command, args.args || []);
    return {
      type: "function_call_output",
      call_id: call.call_id,
      output: JSON.stringify(result)
    };
  }

  throw new Error(`unknown tool: ${call.name}`);
}

async function runAgent(task) {
  let previousResponseId = null;

  while (true) {
    const response = await client.responses.create({
      model: "gpt-5.4",
      previous_response_id: previousResponseId || undefined,
      input: previousResponseId
        ? []
        : [{
            role: "user",
            content: [{ type: "input_text", text: task }]
          }],
      tools
    });

    previousResponseId = response.id;

    const calls = (response.output || []).filter(
      item => item.type === "function_call"
    );

    if (calls.length === 0) {
      return response.output_text;
    }

    const toolOutputs = [];
    for (const call of calls) {
      toolOutputs.push(await executeTool(call));
    }

    const followup = await client.responses.create({
      model: "gpt-5.4",
      previous_response_id: previousResponseId,
      input: toolOutputs,
      tools
    });

    previousResponseId = followup.id;

    const followupCalls = (followup.output || []).filter(
      item => item.type === "function_call"
    );

    if (followupCalls.length === 0) {
      return followup.output_text;
    }
  }
}

上面这段代码还不够“生产可用”，但已经把关键改造点搭起来了：不再手工回放整段历史，而是依赖 previous_response_id 串联每一轮；工具输出作为新的输入项继续推进；执行器和模型循环分层。

七、再往前一步：把日志和观测补上

很多 demo 在这里就停了，但真正能不能上线，取决于你有没有观测。至少要记这些东西：

• response id 链路
• 每次工具调用的名称、参数、耗时、结果摘要
• 最终 answer 的输出长度
• 失败位置：模型调用失败、工具失败、解析失败还是超时

一个很实用的做法，是给每次任务生成一个 traceId，然后把所有 response id 和 tool call 挂到这个 traceId 下面。这样用户说“刚才那个任务卡住了”，你至少能知道是卡在模型、卡在 shell，还是卡在你自己写的 JSON 解析。

八、这套写法真正适合哪些场景

适合多轮、重工具、强上下文延续的任务，比如代码库分析、自动修 bug、测试归因、运维排障、数据库巡检、文档生成流水线。因为这些任务不是“一问一答”，而是“模型做决策，工具拿事实，模型再继续判断”。

不太适合极简单的一次性问答。如果你的场景只是用户问一句、模型回一句，那把系统做得太重，只会增加维护成本。工程上的成熟，不是把所有场景都往复杂架构里塞，而是知道什么时候该停在简单方案。

九、别忽略安全边界

做这类执行器时，最常见的翻车点不是模型不聪明，而是权限太大。我的建议很保守，但实用：

• 默认只读工作区。
• shell 命令做 allowlist，不要给任意命令。
• 每个工具都设超时和输出长度上限。
• 把写文件、删文件、执行部署这类高风险动作放到人工确认后。
• 记录完整审计日志，别让 agent 在你的机器上“做过什么都没人知道”。

agent 工具权限这件事，宁可一开始抠一点，也别一上来就做成 root 用户的自动驾驶。

十、我的判断

如果你正在做 AI 编程或工具型 agent，这种改造值得立刻投入。它不是炫技，而是把一个“能演示”的系统往“能长期使用”的系统推进。真正让用户愿意留下来的，通常不是模型多会说，而是工具链够不够顺、够不够快、够不够可控。

对独立开发者来说，这也是一个很现实的机会点。大平台会继续卷模型，但你完全可以在执行层做出差异：更低延迟的工作流、更安全的工具权限、更可回放的操作日志、更适合特定场景的 agent loop。这个层面没有那么热闹，却更容易做出产品壁垒。

一句话收尾：别再把 agent 当成“会调工具的聊天机器人”了。把它当成一个需要连接、状态、缓存、权限和观测的执行系统，你的产品才会真正开始长骨架。