生产 Agent 的权限、数据隔离与失败补偿

加固 Agent Runtime 的四项保障

生产级 Agent Runtime 不能把工具调用当作事后补上的功能。它必须落实以下四项不可妥协的保障：

• 确定性的基于角色的访问控制（RBAC）：根据调用者经过验证的会话限制可用工具，而不是由 LLM 凭空决定。
• 防范 IDOR 的访问守卫：在执行任何数据库读写之前，以确定性的逻辑检查数据的租户归属。
• 双通道上下文隔离：将 LLM 推理所需的信息与客户端 UI 渲染所需的信息分离。
• 持久化 Saga 编排：为多步数据变更提供补偿回滚操作，确保局部失败不会留下损坏的记录。

                  ┌──────────────────────────────────────────────┐
                  │             avantGate Agent Runtime          │
                  └──────────────────────┬───────────────────────┘
                                         │
                                (Invokes Tool)
                                         ▼
                     [ 1. dataAccessGuard(args, ctx) ] ──> Fail: Abort (Anti-IDOR)
                                         │ Pass
                                         ▼
                             [ 2. execute(args) ]
                                         │
                  ┌──────────────────────┴──────────────────────┐
                  │                                             │
                  ▼                                             ▼
       [ 🎭 LLM DTO Channel ]                        [ 🚀 Client Channel ]
  (dto.pick: minimal context)                    (clientDto: full payload)
                  │                                             │
                  ▼                                             ▼
      LLM Reasoning Context                           UI WebSocket / SSE Stream

定义隔离工具：createIsolatedTool

avantGate 不会直接把原始 JavaScript 函数交给 Agent，而是为每个工具包上一层隔离的执行边界：

import { createIsolatedTool, dto } from "avantgate/agent";
import { z } from "zod";

interface InvoiceRecord {
  invoiceId: string;
  tenantId: string;
  totalAmount: number;
  customerSecretTaxId: string;
  status: string;
}

export const getInvoiceTool = createIsolatedTool({
  name: "get_invoice",
  domain: "billing",
  roles: ["CUSTOMER_SUPPORT", "ADMIN"],
  parameters: z.object({
    invoiceId: z.string().uuid(),
    tenantId: z.string(),
  }),

  // 🛡️ 1. Anti-IDOR: Evaluated deterministically in memory BEFORE DB execution
  async dataAccessGuard(args, context) {
    return args.tenantId === (context?.tenantId as string);
  },

  // ⚙️ 2. The Actual Business Mutation
  async execute(args): Promise<InvoiceRecord> {
    return await db.invoices.findById(args.invoiceId);
  },

  // 🎭 3. Dual-Channel: The LLM ONLY sees what it strictly needs to reason
  llmDto: dto.pick(["invoiceId", "status", "totalAmount"]),

  // 🚀 4. Client Channel: Frontend receives unredacted record out-of-band
  clientDto(rawInvoice) {
    uiSocket.emit("invoice_rendered", rawInvoice);
  },

  // 🧼 5. Recursive Deep Scan for emergent PII in returned data
  sanitizePii: true,
});

应对多步操作失败：采用 Saga 模式的持久化工作流

当 Agent 不再只是查询数据，而是开始修改多个分布式服务中的数据时，例如同时操作计费、库存和通知服务，普通的 try-catch 代码块就不够用了。

如果第 1 步和第 2 步成功后，第 3 步失败：

• 钱已经扣了，座位却没有预订成功。
• 库存已经扣减，却没有创建可追踪的发货记录。

为了解决分布式一致性问题，同时避免引入 Temporal 这类重量级编排框架，avantGate 提供了一个进程内的持久化 Saga 工作流引擎：

[Agent Initiates Workflow]
           │
           ▼
     ┌───────────┐         Success
     │  Step 1   │ ──────────────────────┐
     └─────┬─────┘                       │
           │ Fails                       ▼
           │                       ┌───────────┐         Success
           │                       │  Step 2   │ ──────────────────────┐
           │                       └─────┬─────┘                       │
           │                             │ Fails                       ▼
           │                             │                       ┌───────────┐
           ▼                             ▼                       │  Step 3   │
     [Abort Run]             [Compensate Step 1]                 └─────┬─────┘
                             (e.g., Refund Charge)                     │ Fails
                                                                       ▼
                                                             [Compensate Step 2]
                                                             (Release Inventory)
                                                                       │
                                                             [Compensate Step 1]
                                                             (Refund Charge)

用 TypeScript 实现补偿式 Saga

使用 avantGate 的 createWorkflow 时，每个事务步骤都要定义一个 execute 处理函数，以及一个自动执行回滚的 compensate 钩子：

import { createWorkflow } from "avantgate/workflow";
import { z } from "zod";

export const orderFulfillmentWorkflow = createWorkflow({
  name: "order_fulfillment",
  input: z.object({
    orderId: z.string(),
    customerId: z.string(),
    amount: z.number(),
    sku: z.string(),
  }),

  steps: [
    // Step 1: Capture Payment
    {
      name: "charge_payment",
      async execute({ input }) {
        const payment = await paymentGateway.charge({
          customerId: input.customerId,
          amount: input.amount,
        });
        return { paymentId: payment.id };
      },
      async compensate({ stepOutput }) {
        // Automatically invoked if subsequent steps throw
        await paymentGateway.refund({ paymentId: stepOutput.paymentId });
      },
    },

    // Step 2: Reserve Warehouse Inventory
    {
      name: "reserve_inventory",
      async execute({ input }) {
        const reservation = await warehouse.reserve({
          sku: input.sku,
          quantity: 1,
        });
        return { reservationId: reservation.id };
      },
      async compensate({ stepOutput }) {
        // Releases reserved item if Step 3 fails
        await warehouse.release({ reservationId: stepOutput.reservationId });
      },
    },

    // Step 3: Schedule Delivery Carrier
    {
      name: "dispatch_carrier",
      async execute({ input, steps }) {
        // If carrier API fails, Steps 2 and 1 roll back in reverse order!
        return await carrierService.bookPickup({
          orderId: input.orderId,
          reservationId: steps.reserve_inventory.reservationId,
        });
      },
    },
  ],
});

Runtime 如何保障故障恢复能力

• 正向执行：Runtime 按顺序执行各个步骤，并在运行时内存中逐步保存不可变的步骤输出。
• 反向补偿（回滚栈）：如果任意步骤抛出未处理的错误或触发超时，引擎就会按相反顺序展开回滚栈，依次执行第 2 步的 compensate，再执行第 1 步的 compensate。
• 幂等性保障：每次步骤执行都会标记一个执行 ID，确保重试不会造成重复扣款。

编排完整的安全 Runtime

你可以将隔离工具和 Saga 工作流直接接入 avantGate 引擎：

import { createAvantGate } from "avantgate";
import { getInvoiceTool } from "./tools/invoice";
import { orderFulfillmentWorkflow } from "./workflows/order";

// Initialize the secure runtime
const engine = createAvantGate({
  primary: {
    provider: "deepseek",
    model: "deepseek-chat",
    apiKey: process.env.DEEPSEEK_API_KEY!,
  },
  security: {
    detectPromptInjection: true, // Ingress AI-WAF
    maskPII: true,                // In-flight PII redaction
  },
  maxTokenBudget: 5000,          // Denial-of-Wallet circuit breaker
  maxTurns: 8,                   // Guard against recursive infinite loops
});

// Run an authenticated agent turn
async function handleUserMessage(userQuery: string, session: { tenantId: string; role: string }) {
  try {
    const result = await engine.executeAgent({
      userQuery,
      context: {
        tenantId: session.tenantId,
        role: session.role,
      },
      tools: [getInvoiceTool],
      workflows: [orderFulfillmentWorkflow],
    });

    return result.output;
  } catch (error: any) {
    console.error("Agent execution halted:", error.message);
    return "I encountered a security or budget policy constraint while processing your request.";
  }
}

为什么「进程内」方案优于代理方案

大多数现代 LLMOps 平台都建议部署独立的网关，包括代理、Redis 缓存和追踪容器。

集中式日志有它的用处，但工具执行和工作流补偿应该紧挨着你的数据：

• 额外网络延迟为 0ms：进程内校验、正则表达式编译和状态回滚都在原生 V8 Runtime 中执行，耗时不到一毫秒。
• 不增加基础设施负担：你不必仅仅为了保护事务，就启动并维护外部 sidecar 或工作流守护进程。
• 端到端类型安全：从 Zod schema 到数据库记录、客户端事件，再到工作流步骤之间的依赖，都有完整的 TypeScript 类型推断。

自主 Agent 不必成为行为难以预测的黑盒。通过为工具执行设置确定性的契约，使用 dataAccessGuard 校验租户边界，以双通道模式隔离上下文，再用持久化 Saga 工作流保护多步数据变更，你就能在实现零数据泄露的同时，保证自主系统的事务一致性。

你目前是如何为 Agent 设计工具调用、回滚机制和租户隔离的？欢迎在下方评论区交流！🛡️

👉 前往 GitHub 探索 avantGate 的代码库和文档：thienban/avantGate

如需采取进一步措施，你可以考虑屏蔽此人和／或举报滥用行为。