有界代理循环：用于LLM工具开发的可靠TypeScript模式

了解如何运用带有限制条件的智能体循环、Zod验证、确定性的工具执行机制以及错误恢复功能来构建TypeScript LLM智能体，从而实现生产级可靠性。

2150 词

大多数开发者仍将大型语言模型视为升级版的搜索框：输入提示语，解析返回的字符串，然后希望其中的算术运算、权限检查或格式规则没有在某个环节出错。

一旦将生成式 AI 应用于那些正确性至关重要的系统，比如计费、合规或库存管理，这种做法就会变得危险。将概率性令牌生成视为唯一真实来源在架构层面存在风险，绝非小问题。即便当前版本的模型通常能正确处理简单算术运算，也绝不能让概率性文本生成成为处理关键业务逻辑的权威依据。

真正具备扩展性的模式可被称为有限代理循环：无需直接要求模型生成答案，而是让它充当协调者。它调用确定性工具，检查这些工具返回的强类型结果，并在严格的操作范围内处理故障。

单轮提示法与有限代理循环的对比

1. 逻辑与数学运算处理

单轮提示法：依赖概率性的下一个标记预测，这会在处理法定公式、货币换算或交易规则等任务时带来真实的幻觉风险。
有限代理循环：将所有计算任务交由确定性后端服务处理，从而确保数值结果的准确性。

2. 数据验证与系统完整性

单轮提示机制：依赖脆弱的正则表达式或手动字符串解析，从自由文本中提取结构化数据。
有限代理循环：在任何内容到达业务逻辑层或数据库层之前，都会实时将每个工具参数与Zod模式进行比对。

3. 容错能力与模型恢复

单轮提示机制：一旦模型提供无效参数或遗漏必填字段，要么直接失败且不给出解释，要么完全停止运行。
有限代理循环：会将验证失败和业务规则错误以工具输出的形式反馈到对话中，使模型能够在下一轮中使用修正后的参数重新尝试。

4. 并发与并行执行

单轮提示机制：仍依赖顺序式的整体生成方式，导致除简单请求外的任何任务处理速度都会变慢。
受限智能体循环：使用Promise.all同时发起多个独立的工具调用。

职责划分：大语言模型与后端

要构建可靠的智能体，就需要将模型的决策与实际执行的任务分开：

大语言模型的职责：理解用户意图，从语义层面选择合适的工具，为该工具生成所需参数，并组合出最终的自然语言回复。

后端负责的内容：在运行时验证架构，对调用者进行身份认证和授权，执行业务规则，确保操作的可重试性，进行计算，应用任何副作用，并记录所有操作以供审计。

四阶段智能体循环架构

一个稳健的智能体循环会经历四个独立的阶段：

规划：模型读取用户的意图，查看可用的工具，然后发出结构化的工具调用请求。
验证：后端拦截该调用，在任何逻辑执行之前，根据Zod架构及调用者的权限对其进行检查。
执行：后端执行实际的确定性操作，无论是数据库读取、外部API调用还是计算任务，并将结果记录在对应的可重试性键下。

注意：该执行的结果会以结构化的tool消息形式返回给模型。随后模型会判断是否需要再次调用工具，或者可以直接生成最终答案。

TypeScript中的生产环境实现

下面的实现可以保存为agent.ts文件。它包含了类型化的工具契约、运行时的Zod验证、授权检查、幂等性保护以及执行次数的硬限制。

import OpenAI from "openai";
import { z } from "zod";
import { randomUUID } from "node:crypto";

const openai = new OpenAI({
  apiKey: process.env.OPENAI_API_KEY,
});
// ============================================================================
// 1. Types & Operational Contracts
// ============================================================================
export interface SecurityContext {
  userId: string;
  tenantId: string;
  roles: string[];
}
export type ToolResultStatus =
  | "success"
  | "validation_error"
  | "permission_denied"
  | "business_error"
  | "fatal_error";
export interface ToolResult<T = unknown> {
  status: ToolResultStatus;
  data?: T;
  error?: {
    code: string;
    message: string;
    details?: unknown;
  };
  metadata: {
    toolName: string;
    toolCallId: string;
    latencyMs: number;
    idempotencyKey?: string;
  };
}
export interface AgentTool<TInput = unknown, TOutput = unknown> {
  name: string;
  description: string;
  schema: z.ZodType<TInput>;
  openAiDefinition: OpenAI.Chat.Completions.ChatCompletionTool;
  isMutating: boolean;
  requiredPermission?: string;
  handler: (
    input: TInput,
    context: SecurityContext,
    idempotencyKey?: string
  ) => Promise<TOutput>;
}
// Custom Error Classes
class BusinessRuleError extends Error {
  constructor(public readonly code: string, message: string) {
    super(message);
    this.name = "BusinessRuleError";
  }
}
// ============================================================================
// 2. Demonstration Tax Tool (Explicit Jurisdictions, No Silent Fallbacks)
// ============================================================================
const CalculateTaxSchema = z.object({
  subtotal: z.number().positive("Subtotal must be greater than 0"),
  countryCode: z
    .string()
    .length(2, "Country code must be a 2-letter ISO code")
    .toUpperCase(),
});
type CalculateTaxInput = z.infer<typeof CalculateTaxSchema>;
interface TaxResult {
  jurisdiction: string;
  rateApplied: number;
  taxAmount: number;
  grossTotal: number;
}
// Simplified demonstration rates. Production systems should query a versioned tax rules engine.
const DEMO_STATUTORY_RATES: Record<string, number> = {
  UK: 0.20, // 20% Standard VAT
  BD: 0.15, // 15% Standard VAT
};
const calculateTaxTool: AgentTool<CalculateTaxInput, TaxResult> = {
  name: "calculateTax",
  description:
    "Computes statutory tax and gross total for verified jurisdictions (UK, BD). Rejects unsupported regions.",
  schema: CalculateTaxSchema,
  isMutating: false,
  openAiDefinition: {
    type: "function",
    function: {
      name: "calculateTax",
      description:
        "Computes statutory tax for an invoice line. Only supports UK and BD in this demonstration environment.",
      parameters: {
        type: "object",
        properties: {
          subtotal: { type: "number", description: "Net amount before tax" },
          countryCode: { type: "string", description: "2-letter ISO code (e.g. 'UK', 'BD')" },
        },
        required: ["subtotal", "countryCode"],
        additionalProperties: false,
      },
      strict: true,
    },
  },
  handler: async (input: CalculateTaxInput): Promise<TaxResult> => {
    const rate = DEMO_STATUTORY_RATES[input.countryCode];
    if (rate === undefined) {
      throw new BusinessRuleError(
        "UNSUPPORTED_JURISDICTION",
        `Jurisdiction '${input.countryCode}' is not supported. Only UK and BD are configured.`
      );
    }
    const taxAmount = Number((input.subtotal * rate).toFixed(2));
    const grossTotal = Number((input.subtotal + taxAmount).toFixed(2));
    return {
      jurisdiction: input.countryCode,
      rateApplied: rate,
      taxAmount,
      grossTotal,
    };
  },
};
// ============================================================================
// 3. Mutating Side-Effect Tool (With Idempotency & Auth Boundary)
// ============================================================================
const RecordInvoiceSchema = z.object({
  clientName: z.string().min(1, "Client name is required"),
  amount: z.number().positive("Amount must be positive"),
  taxAmount: z.number().nonnegative(),
  currency: z.string().length(3).toUpperCase(),
});
type RecordInvoiceInput = z.infer<typeof RecordInvoiceSchema>;
const processedIdempotencyKeys = new Set<string>();
const recordInvoiceTool: AgentTool<RecordInvoiceInput, { invoiceId: string; status: string }> = {
  name: "recordInvoice",
  description: "Records an invoice in the ledger. Mutating operation requiring 'billing:write' permission.",
  schema: RecordInvoiceSchema,
  isMutating: true,
  requiredPermission: "billing:write",
  openAiDefinition: {
    type: "function",
    function: {
      name: "recordInvoice",
      description: "Persists an invoice into the financial accounting ledger.",
      parameters: {
        type: "object",
        properties: {
          clientName: { type: "string", description: "Customer or business entity name" },
          amount: { type: "number", description: "Gross invoice amount" },
          taxAmount: { type: "number", description: "Computed tax portion" },
          currency: { type: "string", description: "3-letter currency code (e.g. GBP, BDT)" },
        },
        required: ["clientName", "amount", "taxAmount", "currency"],
        additionalProperties: false,
      },
      strict: true,
    },
  },
  handler: async (input, context, idempotencyKey) => {
    if (!idempotencyKey) {
      throw new Error("Fatal: Mutating operations require an idempotency key.");
    }
    if (processedIdempotencyKeys.has(idempotencyKey)) {
      return {
        invoiceId: `inv_cached_${idempotencyKey.slice(0, 8)}`,
        status: "already_processed_idempotent",
      };
    }
    processedIdempotencyKeys.add(idempotencyKey);
    return {
      invoiceId: `inv_${randomUUID().slice(0, 8)}`,
      status: "recorded",
    };
  },
};
// ============================================================================
// 4. Strongly Typed Registry & Dispatcher
// ============================================================================
const toolRegistry = new Map<string, AgentTool<any, any>>([
  [calculateTaxTool.name, calculateTaxTool],
  [recordInvoiceTool.name, recordInvoiceTool],
]);
async function dispatchToolCall(
  toolCall: OpenAI.Chat.Completions.ChatCompletionMessageToolCall,
  agentRunId: string,
  securityContext: SecurityContext
): Promise<ToolResult> {
  const startTime = Date.now();
  const toolName = toolCall.function.name;
  const tool = toolRegistry.get(toolName);
  const idempotencyKey = tool?.isMutating ? `${agentRunId}:${toolCall.id}` : undefined;
  // 1. Tool Existence Check
  if (!tool) {
    return {
      status: "fatal_error",
      error: { code: "UNKNOWN_TOOL", message: `Tool '${toolName}' does not exist.` },
      metadata: { toolName, toolCallId: toolCall.id, latencyMs: Date.now() - startTime },
    };
  }
  // 2. Authorization Boundary Check
  if (tool.requiredPermission && !securityContext.roles.includes(tool.requiredPermission)) {
    return {
      status: "permission_denied",
      error: {
        code: "UNAUTHORIZED",
        message: `Execution rejected: Caller lacks required permission '${tool.requiredPermission}'.`,
      },
      metadata: { toolName, toolCallId: toolCall.id, latencyMs: Date.now() - startTime },
    };
  }
  // 3. Schema Boundary Check (Zod)
  let parsedArguments: unknown;
  try {
    parsedArguments = JSON.parse(toolCall.function.arguments);
  } catch {
    return {
      status: "validation_error",
      error: { code: "MALFORMED_JSON", message: "Arguments payload was not valid JSON." },
      metadata: { toolName, toolCallId: toolCall.id, latencyMs: Date.now() - startTime },
    };
  }
  const validationResult = tool.schema.safeParse(parsedArguments);
  if (!validationResult.success) {
    return {
      status: "validation_error",
      error: {
        code: "SCHEMA_VALIDATION_FAILED",
        message: "Tool arguments failed schema validation.",
        details: validationResult.error.flatten().fieldErrors,
      },
      metadata: { toolName, toolCallId: toolCall.id, latencyMs: Date.now() - startTime },
    };
  }
  // 4. Execution Boundary (Business logic & side effects)
  try {
    const data = await tool.handler(validationResult.data, securityContext, idempotencyKey);
    return {
      status: "success",
      data,
      metadata: {
        toolName,
        toolCallId: toolCall.id,
        latencyMs: Date.now() - startTime,
        idempotencyKey,
      },
    };
  } catch (error) {
    if (error instanceof BusinessRuleError) {
      return {
        status: "business_error",
        error: { code: error.code, message: error.message },
        metadata: { toolName, toolCallId: toolCall.id, latencyMs: Date.now() - startTime },
      };
    }
    return {
      status: "fatal_error",
      error: {
        code: "INTERNAL_EXECUTION_FAILURE",
        message: error instanceof Error ? error.message : "Unhandled execution crash.",
      },
      metadata: { toolName, toolCallId: toolCall.id, latencyMs: Date.now() - startTime },
    };
  }
}
// ============================================================================
// 5. Bounded Orchestration Loop with Observability & Limits
// ============================================================================
interface AgentRunConfig {
  maxCycles?: number;
  timeoutMs?: number;
  model?: string;
}
async function runReliableAgent(
  userPrompt: string,
  securityContext: SecurityContext,
  config: AgentRunConfig = {}
): Promise<string> {
  const {
    maxCycles = 5,
    timeoutMs = 30_000,
    model = process.env.OPENAI_MODEL || "gpt-4o-mini",
  } = config;
  const agentRunId = `run_${randomUUID()}`;
  const startTime = Date.now();
  const messages: OpenAI.Chat.Completions.ChatCompletionMessageParam[] = [
    {
      role: "system",
      content:
        "You are an enterprise accounting orchestrator. You do not calculate statutory taxes or book entries directly. You coordinate with deterministic tools. If a tool reports a validation error or business rule failure, analyze the issue and attempt to correct parameters or explain the limitation.",
    },
    { role: "user", content: userPrompt },
  ];
  const toolsPayload = Array.from(toolRegistry.values()).map((t) => t.openAiDefinition);
  let cycle = 0;
  while (cycle < maxCycles) {
    cycle += 1;
    // Guardrail: Wall-clock timeout
    if (Date.now() - startTime > timeoutMs) {
      throw new Error(`Agent run [${agentRunId}] aborted: Exceeded timeout of ${timeoutMs}ms.`);
    }
    const response = await openai.chat.completions.create({
      model,
      messages,
      tools: toolsPayload,
      tool_choice: "auto",
    });
    const choice = response.choices[0];
    const assistantMessage = choice.message;
    messages.push(assistantMessage);
    // Terminal condition: Orchestrator reached final text answer
    if (!assistantMessage.tool_calls || assistantMessage.tool_calls.length === 0) {
      return assistantMessage.content ?? "Agent completed without generating text.";
    }
    // Concurrently dispatch independent tool calls
    const toolPromises = assistantMessage.tool_calls.map(async (toolCall) => {
      const result = await dispatchToolCall(toolCall, agentRunId, securityContext);
      return {
        role: "tool" as const,
        tool_call_id: toolCall.id,
        content: JSON.stringify(result),
      };
    });
    const toolMessages = await Promise.all(toolPromises);
    messages.push(...toolMessages);
  }
  throw new Error(`Agent run [${agentRunId}] halted: Exceeded maximum iterations (${maxCycles}).`);
}
// ============================================================================
// 6. Test Scenario: Concurrency, Unsupported Jurisdictions & Auth
// ============================================================================
async function main() {
  const securityContext: SecurityContext = {
    userId: "usr_9918",
    tenantId: "tenant_uk_01",
    roles: ["billing:write"],
  };
  const prompt =
    "Calculate tax for two invoices: 1500 GBP for a client in the UK, and 500 EUR for a client in Germany (DE). If tax calculation succeeds, record the invoice for the UK client.";
  console.log("Executing Agent Run...\n");
  try {
    const finalAnswer = await runReliableAgent(prompt, securityContext);
    console.log("=== Agent Response ===");
    console.log(finalAnswer);
  } catch (err) {
    console.error("Execution failed:", err);
  }
}
main();

生产环境中的关键问题

1. 绝不能让系统无声无息地切换到备用方案。在金融或受监管的工程领域，若用默认利率悄悄替换无法识别的国家代码，就会引发严重的合规问题。在之前展示的实现方式中，系统未明确支持的任何司法管辖区都必须抛出类型为BusinessRuleError的异常。该异常会反馈给协调器，由其向用户准确说明遇到了何种限制，而非编造一个看似合理的数值。

2. 使用幂等键保护会修改状态的操作。仅读取数据的操作可以安全地重试。但任何会改变状态的操作，比如发送发票、调整账本余额或触发 webhook，都需要通过幂等键来控制。通过从 ${agentRunId}:${toolCallId} 导出该键，可以确保在模型为恢复错误而两次调用同一函数时，后端能够识别重复请求并跳过重新处理。

3. 将错误分类，以便模型知道如何应对。并非所有故障都应触发大语言模型的重试：

验证错误（validation_error）：参数格式不正确。模型可以查看 Zod 报告的字段级详细信息，并在下次尝试时修正其数据。

业务错误（business_error）：违反了既定规则，例如使用了不受支持的司法管辖区。模型可以尝试其他处理方式，或将该限制告知用户。

权限错误（permission_denied）：对于特定的工具调用而言，这类错误无法修复。由于模型无法自行获得额外权限，正确的处理方式是干净地停止执行，而非不断重试。