人机协同工具审批（HITL）：基于自定义数据块的高风险写操作管控实践

人机协同工具审批是智能体高风险写操作管控机制，通过自定义数据块将写工具调用拆分为跨轮次状态机，结合前置补丁预检、服务端裁决守卫与提示词过滤，在管理员显式确认后执行数据库变更。

方案架构与核心组件

审批生命周期与三类自定义数据块

人机协同工具审批将“模型发起工具调用”与“实际执行底层数据库写操作”拆分为两个独立的对话轮次，通过三类自定义数据块在服务端、数据库与客户端 UI 之间流转状态：

数据块类型	产生阶段	归属角色	职责与载荷内容
`data-approval-request`	第 $N$ 轮模型调用写工具时	`assistant`	承载审批请求（Approval Request），包含工具 ID、工具名、原始参数与预合成预览元数据（`previewMeta`）
`data-approval-decision`	第 $N+1$ 轮管理员点击按钮或提交反馈时	`user`	承载审批决策（Approval Decision），绑定目标 `toolId`，取值为 `approve`（批准）或带 `reason` 的 `reject`（拒绝）
`data-approval-result`	第 $N+1$ 轮服务端执行裁决后	`assistant` / `user`	承载审批执行结果（Approval Result），记录写操作落库后的实体 ID、链接或拒绝/失败详情

定义人机协同审批的核心数据结构与异常类型：

TypeScript

export const ADMIN_WRITE_TOOL_NAMES = [
  'createPost',
  'updatePost',
  'deletePost',
  'createCollection',
  'updateCollection',
  'deleteCollection',
] as const;

export type AdminWriteToolName = (typeof ADMIN_WRITE_TOOL_NAMES)[number];

export interface ContentPatchItem {
  oldText: string;
  newText: string;
}

export interface ToolRequiringApproval {
  id: string;
  name: AdminWriteToolName;
  input: Record<string, unknown>;
  previewMeta?: {
    targetTitle?: string;
    previousStatus?: number;
    patchedContentPreview?: string;
    patchCount?: number;
    appendedLength?: number;
    orderedPosts?: Array<{ id: number; title: string }>;
    removedPosts?: Array<{ id: number; title: string }>;
  };
}

export type ToolApprovalDecision =
  | { type: 'approve' }
  | { type: 'reject'; reason: string };

export interface ToolApprovalResult {
  toolId: string;
  toolName: AdminWriteToolName;
  status: 'approved' | 'rejected' | 'error';
  output?: {
    id: number;
    title: string;
    status?: number;
    url: string;
    previewUrl?: string;
    adminEditUrl?: string;
    message: string;
  };
  reason?: string;
  error?: string;
}

export class HITLPendingDecisionError extends Error {
  readonly statusCode = 400;
  constructor(message = '存在待确认的操作请求，请先完成批准或拒绝') {
    super(message);
    this.name = 'HITLPendingDecisionError';
  }
}

为什么采用自定义数据块构建 HITL

相较于仅依赖简单的布尔开关，使用 AI SDK Streaming Custom Data 构建审批流具备以下工程扩展性：

前置业务校验与预览合成：在发出审批卡片前，服务端可先读取目标文章或专栏，校验局部补丁（contentPatches）是否在原文中唯一匹配，或校验专栏更新是否遗漏了现有文章。若前置校验失败，工具直接返回 success: false 而不发出审批卡片，允许模型在同一轮次内自动读取原文并修正参数重试。
携带丰富的预览元数据：审批请求可在 previewMeta 中附带合成后的完整正文预览、目标文章当前标题及专栏文章排序映射，供前端 ApprovalCard 渲染变更对比，而无需前端额外发起详情查询。
精细控制多轮模型上下文：可根据后续轮次的需要，将历史写工具调用参数中的万字长文截断为首尾各 200 字符的预览，并将审批结果合并回工具输出，兼顾上下文体积控制与协议一致性。

核心模块实现

写工具定义与前置补丁校验

以修改文章工具 updatePost 为例，技术博客正文通常长达数千至上万字符。若每次微调段落或续写章节都要求模型输出完整正文，会导致高延迟与高 Token 消耗。因此，updatePost 设计了三种互斥的正文修改模式：

局部补丁（contentPatches）：传入 oldText 与 newText 数组，用于微调特定段落或修正字句。
文末追加（appendContent）：直接传入待追加的 Markdown 文本，用于长文分段续写。
全量替换（content）：仅在整体重写全文时使用。

实现局部补丁唯一性校验与文末追加纯函数：

TypeScript

// 按顺序校验每个 patch.oldText 在原文中必须恰好出现 1 次，并返回预合成的新正文
export function validateContentPatches(
  originalContent: string,
  patches: ContentPatchItem[],
):
  | { success: true; patchedContent: string }
  | { success: false; error: string } {
  if (!Array.isArray(patches) || patches.length === 0) {
    return { success: true, patchedContent: originalContent };
  }

  let currentText = originalContent;

  for (let i = 0; i < patches.length; i++) {
    const patch = patches[i];
    if (!patch.oldText) {
      return { success: false, error: `第 ${i + 1} 个补丁旧片段不能为空。` };
    }

    const firstIndex = currentText.indexOf(patch.oldText);
    if (firstIndex === -1) {
      return {
        success: false,
        error: `第 ${i + 1} 个补丁旧片段未在文章正文中找到，请核实原文后再试：${patch.oldText.slice(0, 50)}`,
      };
    }

    const secondIndex = currentText.indexOf(patch.oldText, firstIndex + 1);
    if (secondIndex !== -1) {
      return {
        success: false,
        error: `第 ${i + 1} 个补丁旧片段在文章正文中出现多次，存在歧义，请提供更具体的前后上下文：${patch.oldText.slice(0, 50)}`,
      };
    }

    currentText =
      currentText.slice(0, firstIndex) +
      (patch.newText ?? '') +
      currentText.slice(firstIndex + patch.oldText.length);
  }

  return { success: true, patchedContent: currentText };
}

// 文末追加正文，自动补充缺失的换行符防止段落粘连
export function appendPostContent(
  originalContent: string,
  appendContent: string,
): string {
  if (!appendContent) return originalContent;
  if (!originalContent) return appendContent;
  if (!originalContent.endsWith('\n') && !appendContent.startsWith('\n')) {
    return `${originalContent}\n${appendContent}`;
  }
  return `${originalContent}${appendContent}`;
}

在 updatePost 工具的 execute 中执行前置校验并向流写入 data-approval-request：

TypeScript

const updatePost = tool({
  description:
    '管理员专属写工具：修改已有博客文章元数据、局部补丁段落（contentPatches）、文末追加正文（appendContent）或替换完整正文（content）。调用后向客户端发起操作审批请求，待管理员确认后执行。',
  inputSchema: updatePostInputSchema,
  execute: async (input, options?: { toolCallId?: string }) => {
    const toolCallId = options?.toolCallId || crypto.randomUUID();
    const existing = await getAdminPostDetailForAgent(input.postId);
    if (!existing) {
      return {
        success: false,
        error: `未找到 ID 为 ${input.postId} 的文章。`,
      };
    }

    const hasContent =
      typeof input.content === 'string' && input.content.length > 0;
    const hasPatches =
      Array.isArray(input.contentPatches) && input.contentPatches.length > 0;
    const hasAppendContent =
      typeof input.appendContent === 'string' && input.appendContent.length > 0;

    const activeContentModes = [
      hasContent,
      hasPatches,
      hasAppendContent,
    ].filter(Boolean).length;
    if (activeContentModes > 1) {
      return {
        success: false,
        error:
          'content、contentPatches 与 appendContent 只能三选一互斥，不能同时提供。',
      };
    }

    let patchedContent: string | undefined = undefined;
    if (hasPatches) {
      const patchResult = validateContentPatches(
        existing.content,
        input.contentPatches as ContentPatchItem[],
      );
      if (!patchResult.success) {
        // 补丁校验失败时返回 success: false，不发出审批请求，允许模型在同轮内自愈重试
        return {
          success: false,
          error: patchResult.error,
        };
      }
      patchedContent = patchResult.patchedContent;
    } else if (hasAppendContent) {
      patchedContent = appendPostContent(
        existing.content,
        input.appendContent!,
      );
    }

    // 前置校验通过，向 UI 消息流写入审批请求数据块
    writer?.write({
      type: 'data-approval-request',
      data: {
        tool: {
          id: toolCallId,
          name: 'updatePost',
          input,
          previewMeta: {
            targetTitle: existing.title,
            previousStatus: existing.status,
            patchedContentPreview: patchedContent,
            patchCount: input.contentPatches?.length,
            ...(hasAppendContent
              ? { appendedLength: input.appendContent!.length }
              : {}),
          },
        },
      },
    });

    return {
      success: true,
      message: `已提交更新文章 "${existing.title}" 的审批请求，请在下方审批卡片中核对并确认。`,
    };
  },
  toModelOutput: ({ output }) => ({
    type: 'text',
    value: output.message || output.error || JSON.stringify(output),
  }),
});

流暂停条件（StopCondition）

为了在发出审批请求后立即暂停当前对话轮次并等待管理员裁决，streamText 的 stopWhen 挂载了 hasAdminApprovalRequestEmitted()。该函数仅在管理写工具返回 output.success === true 时触发暂停；若因补丁失配返回 success: false，则允许模型继续执行下一步修正补丁。

实现审批暂停条件函数：

TypeScript

import type { StopCondition } from 'ai';
import { ADMIN_WRITE_TOOL_NAMES } from './hitl-types';

export function hasAdminApprovalRequestEmitted(): StopCondition<any, any> {
  return ({ steps }) => {
    const lastStep = steps[steps.length - 1];
    if (!lastStep?.toolResults) return false;
    return lastStep.toolResults.some((tr: any) => {
      if (!ADMIN_WRITE_TOOL_NAMES.includes(tr.toolName)) return false;
      const output = tr.output;
      return (
        output && typeof output === 'object' && (output as any).success === true
      );
    });
  };
}

未决审批守卫与决策匹配

当会话的最后一条已持久化 assistant 消息包含 data-approval-request 时，服务端在下一轮请求入口通过 findDecisionsToProcess 实施状态守卫：

若触发类型为 submit-message，则用户提交的消息 parts 中必须包含与所有待处理请求 ID 一一对应的 data-approval-decision，否则抛出 HITLPendingDecisionError（HTTP 400），防止绕过未决审批继续发起无关提问。
若触发类型为 regenerate-message，则直接放行，由后续数据库事务删除旧的 assistant 消息从而作废未决提案。

实现未决审批守卫与匹配函数：

TypeScript

export async function findDecisionsToProcess(params: {
  sessionId: string;
  trigger: 'submit-message' | 'regenerate-message';
  message: { role: 'user'; parts: Array<Record<string, unknown>> };
  dbInstance?: any;
}) {
  const { sessionId, trigger, message, dbInstance = db } = params;

  if (trigger === 'regenerate-message') {
    return [];
  }

  const lastMessages = await dbInstance
    .select()
    .from(agentMessageTable)
    .where(eq(agentMessageTable.session_id, sessionId))
    .orderBy(desc(agentMessageTable.seq))
    .limit(1);

  if (!lastMessages.length) return [];

  const lastMessage = lastMessages[0];
  if (lastMessage.role !== 'assistant' || !Array.isArray(lastMessage.parts)) {
    return [];
  }

  const pendingRequests: ToolRequiringApproval[] = [];
  for (const part of lastMessage.parts as any[]) {
    if (
      part &&
      (part.type === 'data-approval-request' || part['approval-request'])
    ) {
      const tool =
        part.data?.tool || part.tool || part['approval-request']?.tool;
      if (tool && tool.id) {
        pendingRequests.push(tool);
      }
    }
  }

  if (pendingRequests.length === 0) return [];

  const decisionsByToolId = new Map<string, ToolApprovalDecision>();
  if (Array.isArray(message.parts)) {
    for (const part of message.parts) {
      if (
        part &&
        (part.type === 'data-approval-decision' ||
          (part as any)['approval-decision'])
      ) {
        const data =
          (part.data as any) || (part as any)['approval-decision'] || part;
        if (data.toolId && data.decision) {
          decisionsByToolId.set(data.toolId, data.decision);
        }
      }
    }
  }

  for (const req of pendingRequests) {
    if (!decisionsByToolId.has(req.id)) {
      throw new HITLPendingDecisionError(
        '存在待确认的操作请求，请先完成批准或拒绝',
      );
    }
  }

  return pendingRequests.map((tool) => ({
    tool,
    decision: decisionsByToolId.get(tool.id)!,
  }));
}

完整流程演练与上下文协议治理

路由返回前执行写操作与 Next.js 缓存失效对齐

在 Next.js App Router 中，业务写服务（如 createPost、updatePost）在修改数据库后会调用 revalidateTag 失效公开页面的读取缓存。Next.js 要求 revalidateTag 必须在 Route Handler 返回 HTTP Response 之前调用才能生效；如果将 executeHITLDecisions 放在尚未被消费完的异步流回调（createUIMessageStream.execute）内部执行，当 Route Handler 已经返回流式响应头时，revalidateTag 入队的缓存标签将无法被 Next.js 刷写。

因此，在 streamAgentChat 中，已裁决的写操作必须在构造并返回 createUIMessageStreamResponse 之前同步执行完毕：

TypeScript

// 1. 返回流式响应之前先落库已批准的写操作，确保 Next.js 路由返回前收集到 revalidateTag
const approvalResults: ToolApprovalResult[] =
  matchedDecisions.length > 0
    ? await executeHITLDecisions({
        userId: user.id,
        userRole: user.role,
        decisions: matchedDecisions,
      })
    : [];

// 2. 启动 UI 消息流：先将 approvalResults 写入流，再注入模型上下文供助手总结或按拒绝原因重试
const uiStream = createUIMessageStream({
  execute: async ({ writer }) => {
    for (const res of approvalResults) {
      writer.write({
        type: 'data-approval-result',
        data: res,
      } as any);
    }

    const rawUiMessages: UIMessage[] = tailMessages.map((m: any) => ({
      id: m.id,
      role: m.role as 'user' | 'assistant',
      parts: m.parts as any,
    }));

    // 将 data-approval-result 插入到最后一条 user 消息的 decision 之后、用户文本指令之前
    const uiMessages = injectApprovalResultsIntoTurnMessages(
      rawUiMessages,
      approvalResults,
    );

    const streamResult = await runAgentTurn({
      role: user.role,
      userId: user.id,
      uiMessages,
      summary: compactionSummary,
      memories,
      writer,
      model: getChatModel(),
      abortSignal: combinedSignal,
    });

    writer.merge(
      withLengthTruncationNotice(
        toUIMessageStream({
          stream: streamResult.stream,
          sendReasoning: false,
          generateMessageId: () => crypto.randomUUID(),
        }),
      ),
    );
  },
  // onEnd 落库逻辑...
});

防止模型手写伪造审批协议的上下文治理

在将包含自定义数据块的历史 UIMessage 转换为 ModelMessage 时，AI SDK 的 convertToModelMessages 支持传入 convertDataPart 回调将自定义数据块投影为文本块。然而，如果在 assistant 角色的历史消息中把 data-approval-request 直接转换为 [操作审批请求] 工具: updatePost... 这样的纯文本，大语言模型在后续轮次的上下文学习（In-Context Learning）中，极易将该纯文本格式误认为触发审批的方式，从而在普通回复正文中直接手写 [操作审批请求] 工具: ... 而不发起原生 Function Calling，导致前端无法渲染审批卡片。

为规避该问题，系统在 prepareUiMessagesForModelContext 中实施了三项协议治理规则：

写工具保留原生 tool-call / tool-result 结构：不裁剪历史中的 6 个管理写工具调用，并将已裁决的 ToolApprovalResult 合并进写工具的 output 中；同时剔除 assistant 消息中与原生写工具重复的 data-approval-request。
执行结果归位至 user 消息：将历史 assistant 消息头部的 data-approval-result 移至前序 user 审批消息中（紧跟在 data-approval-decision 之后），使模型看到的文本投影全部来自 user 角色反馈，而 assistant 角色始终只包含原生 Function Calling 与正常自然语言回复。
长正文入参首尾截断：对历史写工具入参中超过 500 字符的 content、appendContent 或 contentPatches，在内存中截断为保留开头 200 字符与结尾 200 字符并标注总长度，既避免撑爆上下文，又让模型在分段续写时准确知晓上一轮停笔的末尾字句。

实现写工具长正文入参截断与数据块文本投影：

TypeScript

// 对管理写工具入参中的长正文（> 500 字符）执行首尾各 200 字符截断
export function sanitizeWriteToolInput(
  input: Record<string, unknown>,
): Record<string, unknown> {
  if (!input || typeof input !== 'object') return {};
  const sanitizedInput: Record<string, unknown> = { ...input };

  if (
    typeof sanitizedInput.content === 'string' &&
    sanitizedInput.content.length > 500
  ) {
    const fullLen = sanitizedInput.content.length;
    sanitizedInput.content = `${sanitizedInput.content.slice(0, 200)}... [正文共 ${fullLen} 字符，已截断预览] ...${sanitizedInput.content.slice(-200)}`;
  }

  if (
    typeof sanitizedInput.appendContent === 'string' &&
    sanitizedInput.appendContent.length > 500
  ) {
    const fullLen = sanitizedInput.appendContent.length;
    sanitizedInput.appendContent = `${sanitizedInput.appendContent.slice(0, 200)}... [追加内容共 ${fullLen} 字符，已截断预览] ...${sanitizedInput.appendContent.slice(-200)}`;
  }

  return sanitizedInput;
}

// 将 user 消息中的审批决策与执行结果数据块投影为模型可见文本
export function projectDataPartToText(
  part: Record<string, unknown>,
): string | null {
  if (!part || typeof part !== 'object') return null;

  if (
    part.type === 'data-approval-decision' ||
    (part as any)['approval-decision']
  ) {
    const data =
      (part.data as any) || (part as any)['approval-decision'] || part;
    const { toolId, decision } = data;
    if (!decision || typeof decision !== 'object') return null;

    if (decision.type === 'approve') {
      return `[管理员审批决策] 工具ID: ${toolId}, 结果: 已批准`;
    }
    if (decision.type === 'reject') {
      return `[管理员审批决策] 工具ID: ${toolId}, 结果: 已拒绝（原因: ${decision.reason}）`;
    }
  }

  if (
    part.type === 'data-approval-result' ||
    (part as any)['approval-result']
  ) {
    const result =
      (part.data as any) || (part as any)['approval-result'] || part;
    if (!result || !result.toolName) return null;

    const details =
      result.status === 'approved'
        ? (result.output ?? {})
        : result.status === 'rejected'
          ? { reason: result.reason ?? '' }
          : { error: result.error ?? '执行失败' };

    return `[操作执行结果] 工具: ${result.toolName}, 状态: ${result.status}, 详情: ${JSON.stringify(details)}`;
  }

  return null;
}

前端审批卡片与交互状态机

在客户端，聊天组件遍历当前会话的全部消息片段，建立 decisionsByToolId 与 resultsByToolId 映射表。当某条 data-approval-request 尚未出现在上述两个映射表中时，判定为待确认状态（isPending），渲染可交互的批准与拒绝按钮：

批准分支（Approve）：调用 sendMessage 直接发送携带 data-approval-decision（type: 'approve'）和确认文本的用户消息。
拒绝并附原因分支（Reject）：点击拒绝按钮后将当前 toolId 设为 rejectingToolId，锁定底部主聊天输入框并切换占位提示，引导管理员输入具体修改意见；提交时将拒绝原因封装进 data-approval-decision（type: 'reject', reason）发送给服务端，模型在收到拒绝原因后自动调整参数并发起新一轮提案。

实现客户端审批状态聚合与裁决提交逻辑：

TypeScript

// 聚合全部消息中的审批请求、审批决策与审批结果
const { decisionsByToolId, resultsByToolId, allApprovalRequests } =
  useMemo(() => {
    const decisions = new Map<string, ToolApprovalDecision>();
    const results = new Map<string, ToolApprovalResult>();
    const requests: ToolRequiringApproval[] = [];

    for (const msg of messages) {
      if (!Array.isArray(msg.parts)) continue;
      for (const part of msg.parts) {
        const req = extractApprovalRequest(part);
        if (req) requests.push(req);

        const dec = extractApprovalDecision(part);
        if (dec) decisions.set(dec.toolId, dec.decision);

        const res = extractApprovalResult(part);
        if (res) results.set(res.toolId, res);
      }
    }

    return {
      decisionsByToolId: decisions,
      resultsByToolId: results,
      allApprovalRequests: requests,
    };
  }, [messages]);

// 批准操作回调
const handleApproveTool = useCallback(
  (toolId: string) => {
    if (isGenerating) return;
    const targetReq = allApprovalRequests.find((r) => r.id === toolId);
    const actionLabel = targetReq
      ? getToolActionTitle(targetReq.name)
      : '管理操作';

    sendMessage({
      role: 'user',
      parts: [
        {
          type: 'data-approval-decision',
          data: {
            toolId,
            decision: { type: 'approve' },
          },
        } as any,
        {
          type: 'text',
          text: `已批准执行${actionLabel}`,
        },
      ],
    });
  },
  [isGenerating, allApprovalRequests, sendMessage],
);

人机协同工具审批（HITL）：基于自定义数据块的高风险写操作管控实践

方案架构与核心组件

审批生命周期与三类自定义数据块

为什么采用自定义数据块构建 HITL

核心模块实现

写工具定义与前置补丁校验

流暂停条件（StopCondition）

未决审批守卫与决策匹配

完整流程演练与上下文协议治理

路由返回前执行写操作与 Next.js 缓存失效对齐

防止模型手写伪造审批协议的上下文治理

前端审批卡片与交互状态机

本文收录于专栏

AI 应用开发笔记