智能体何时该停下来问人：用 Node.js 构建风险分级与人工审批门

智能体不应把所有工具调用都当成普通函数执行，也不能遇到不确定性就一律停下。本文用三个可解释维度进行风险分级，并以 Node.js 实现人工审批、超时关闭和审计留痕。

先定义需要停下来的边界

一次工具调用是否需要人工介入，不能只看模型声称的“置信度”。更实用的判断来自三个维度：

维度	关注问题	高风险示例
工具副作用	是否会修改外部世界，能否撤销	删除文件、发送邮件、修改生产配置
结果不确定性	参数和结论是否经过验证	收件人来自非结构化文本、关键字段缺失
操作成本	失败后的资金、时间和恢复成本	付款、批量任务、长时间计算

三个维度都使用 0 到 5 的离散值，比假装精确的概率更容易解释。评分必须来自可检查的事实，例如工具元数据、参数校验结果和执行环境，而不是让模型随意给自己打分。

此外还需要硬规则。涉及付款、密钥或明确不可逆的操作，即使加权分数不高，也应直接进入高风险等级。风险评分负责覆盖常见情况，硬规则负责守住不能被平均值稀释的底线。

建立可解释的风险评分

本文采用如下权重：副作用 45%、不确定性 35%、操作成本 20%。每项先归一化到 0 至 100，再计算总分：

0—29：低风险，可自动执行；
30—59：中风险，可自动执行，但应增加参数校验、限额或 dry-run；
60—100：高风险，必须等待人工审批；
命中硬规则：无条件按高风险处理。

权重不是通用真理。内容生成系统可能更重视不确定性，运维智能体则通常更重视副作用。关键是把评分、贡献值和触发的硬规则一起保存，让审批人知道系统为何停下。

例如，副作用为 4、不确定性为 3、成本为 2 时，总分为 65。解释信息可以写成“副作用贡献 36 分、不确定性贡献 21 分、成本贡献 8 分”，而不是只展示一个缺乏上下文的红色数字。

用状态机约束审批流程

审批门不应只是一个 if。真实执行中会遇到重复确认、超时后迟到的批准、执行失败等情况，因此需要显式状态机：

PLANNED -> EXECUTING -> SUCCEEDED | FAILED
PLANNED -> WAITING_APPROVAL -> APPROVED -> EXECUTING
                            -> REJECTED
                            -> EXPIRED

状态转换白名单可以阻止两类常见错误：已拒绝任务被再次执行，以及审批超时后仍接受迟到确认。每次转换都应写入审计记录，至少包含运行 ID、步骤 ID、操作者、时间、前后状态和原因。

超时应采用 fail closed 策略：没有明确批准，就视为未获授权。不能因为审批系统不可用而自动放行，否则人工审批门会在最需要它的时候失效。

完整可运行的 Node.js 示例

下面的示例仅使用 Node.js 内置模块，可在 Node.js 18 及以上版本运行。将其保存为 agent-gate.mjs，执行 node agent-gate.mjs。示例步骤会写入本地文件，并把审计事件追加到 audit.jsonl。可通过环境变量 APPROVAL_TIMEOUT_MS 调整默认的 30 秒审批时间。

import { appendFile, writeFile } from 'node:fs/promises';
import { randomUUID } from 'node:crypto';
import readline from 'node:readline/promises';
import process from 'node:process';

const AUDIT_FILE = new URL('./audit.jsonl', import.meta.url);

const transitions = {
  PLANNED: ['WAITING_APPROVAL', 'EXECUTING'],
  WAITING_APPROVAL: ['APPROVED', 'REJECTED', 'EXPIRED'],
  APPROVED: ['EXECUTING'],
  EXECUTING: ['SUCCEEDED', 'FAILED'],
  REJECTED: [],
  EXPIRED: [],
  SUCCEEDED: [],
  FAILED: []
};

class AgentRun {
  constructor(step) {
    this.runId = randomUUID();
    this.step = step;
    this.status = 'PLANNED';
  }

  async audit(event, details = {}) {
    const record = {
      timestamp: new Date().toISOString(),
      runId: this.runId,
      stepId: this.step.id,
      event,
      status: this.status,
      ...details
    };
    await appendFile(AUDIT_FILE, `${JSON.stringify(record)}\n`, 'utf8');
  }

  async transition(next, actor, reason) {
    if (!transitions[this.status].includes(next)) {
      throw new Error(`非法状态转换：${this.status} -> ${next}`);
    }

    const previous = this.status;
    await this.audit('STATE_TRANSITION', {
      previous,
      next,
      actor,
      reason
    });
    this.status = next;
  }
}

function scoreRisk(input) {
  const factors = [
    ['工具副作用', input.sideEffect, 0.45],
    ['结果不确定性', input.uncertainty, 0.35],
    ['操作成本', input.operationCost, 0.20]
  ];

  for (const [, value] of factors) {
    if (!Number.isInteger(value) || value < 0 || value > 5) {
      throw new Error('风险维度必须是 0 到 5 的整数');
    }
  }

  const details = factors.map(([name, value, weight]) => ({
    name,
    value,
    contribution: Math.round((value / 5) * weight * 100)
  }));
  const score = details.reduce((sum, item) => sum + item.contribution, 0);

  const hardRules = [];
  if (input.irreversible) hardRules.push('操作不可逆');
  if (input.touchesSecrets) hardRules.push('接触敏感凭据');
  if (input.externalPayment) hardRules.push('涉及外部付款');

  const level = hardRules.length > 0 || score >= 60
    ? 'HIGH'
    : score >= 30 ? 'MEDIUM' : 'LOW';

  return { score, level, details, hardRules };
}

async function requestApproval(run, assessment, timeoutMs) {
  await run.transition('WAITING_APPROVAL', 'system', '风险等级为 HIGH');

  console.log('\n需要人工审批：');
  console.log(JSON.stringify({
    runId: run.runId,
    step: run.step,
    assessment
  }, null, 2));

  const rl = readline.createInterface({
    input: process.stdin,
    output: process.stdout
  });

  let timer;
  const answerPromise = rl
    .question('输入 approve 批准，其他输入将拒绝：')
    .then(value => value.trim().toLowerCase())
    .catch(() => 'closed');
  const timeoutPromise = new Promise(resolve => {
    timer = setTimeout(() => resolve('timeout'), timeoutMs);
  });

  const answer = await Promise.race([answerPromise, timeoutPromise]);
  clearTimeout(timer);
  rl.close();

  if (answer === 'approve') {
    await run.transition('APPROVED', 'human', '审批人明确批准');
    return true;
  }
  if (answer === 'timeout' || answer === 'closed') {
    await run.transition('EXPIRED', 'system', '审批超时或输入通道关闭');
    return false;
  }

  await run.transition('REJECTED', 'human', `审批输入：${answer}`);
  return false;
}

const tools = {
  async writeFile(args) {
    if (typeof args.path !== 'string' || typeof args.content !== 'string') {
      throw new Error('writeFile 参数不合法');
    }
    await writeFile(args.path, args.content, 'utf8');
    return { path: args.path, bytes: Buffer.byteLength(args.content) };
  }
};

async function main() {
  const step = {
    id: 'step-write-report',
    tool: 'writeFile',
    args: {
      path: './demo-output.txt',
      content: '该文件由通过审批门的智能体步骤写入。\n'
    },
    risk: {
      sideEffect: 4,
      uncertainty: 3,
      operationCost: 2,
      irreversible: false,
      touchesSecrets: false,
      externalPayment: false
    }
  };

  const run = new AgentRun(step);
  await run.audit('RUN_CREATED');

  const assessment = scoreRisk(step.risk);
  await run.audit('RISK_ASSESSED', { assessment });
  console.log('风险评估：', assessment);

  if (assessment.level === 'HIGH') {
    const rawTimeout = Number(process.env.APPROVAL_TIMEOUT_MS ?? 30000);
    const timeoutMs = Number.isFinite(rawTimeout) && rawTimeout > 0
      ? rawTimeout
      : 30000;
    const approved = await requestApproval(run, assessment, timeoutMs);
    if (!approved) {
      console.log(`步骤未执行，最终状态：${run.status}`);
      return;
    }
  }

  await run.transition('EXECUTING', 'system', '已满足执行条件');

  try {
    const tool = tools[step.tool];
    if (!tool) throw new Error(`工具未在白名单中：${step.tool}`);
    const result = await tool(step.args);
    await run.transition('SUCCEEDED', 'system', JSON.stringify(result));
    console.log('执行成功：', result);
  } catch (error) {
    await run.transition('FAILED', 'system', error.message);
    console.error('执行失败：', error.message);
    process.exitCode = 1;
  }
}

main().catch(error => {
  console.error(error);
  process.exitCode = 1;
});

接入智能体循环时还要补什么

示例展示的是单进程最小实现。接入实际智能体时，风险评估应放在“模型提出工具调用”和“工具真正执行”之间，并遵守以下边界：

工具必须使用白名单，参数先通过 JSON Schema 或同等机制校验，再进行评分。
审批对象应绑定运行 ID、步骤 ID 和参数摘要。参数变化后，原批准立即失效，避免批准的是 A、执行的却是 B。
审批界面应展示结构化字段，不要直接信任模型生成的说明，以免提示注入内容误导审批人。
生产环境不宜依赖单机 JSONL。并发写入、查询和保留策略可交给事务数据库或集中式日志系统，同时限制审计记录中的敏感信息。
审批通过不等于可以无限执行。金额上限、路径范围、收件人域名和调用次数等约束仍应在工具层强制检查。

对于中风险步骤，可以先执行 dry-run、缩小批量、要求二次校验，或只允许在沙箱环境运行。人工审批是最后一道门，不应替代前面的最小权限和参数约束。

总结

从工具副作用、结果不确定性和操作成本三个维度建立统一评分，并保留每项贡献值。
对不可逆操作、敏感凭据和外部付款设置硬规则，避免风险被加权平均稀释。
使用显式状态机处理批准、拒绝、超时、成功和失败，非法状态不能继续流转。
审批超时采用 fail closed，所有关键事件写入可追溯的审计记录。
将审批绑定到具体步骤和参数，同时保留工具白名单、参数校验与最小权限，才能在自动执行和人工介入之间形成真正可落地的边界。

智能体何时该停下来问人：用 Node.js 构建风险分级与人工审批门

先定义需要停下来的边界

建立可解释的风险评分

用状态机约束审批流程

完整可运行的 Node.js 示例

接入智能体循环时还要补什么

总结

相关文章

别让 AI 值班变成盲目执行：用 Node.js 构建证据驱动的止损闭环

AI 应用也需要供应链清单：统一追踪模型、提示词与工具版本

智能体不该拿长期密钥：用 Node.js 构建短期凭证代理

评论