Agent三层防护：循环检测、API 容错、Token预算

agent实战

agent 三层防护

给 Agent 加上几层防护

前面的 agent 已经可以调用工具了：查天气、多步推理等。

但，你让它帮你查个文件，它 read_file 读完发现路径不对，于是它决定……再读一次。同样的路径，同样的参数，同样的错误结果。然后它"恍然大悟"——再试一次。

--- Step 1 ---
  [调用: read_file({"path":"/tmp/data.csv"})]
  [结果: "文件不存在"]
 
--- Step 2 ---
  [调用: read_file({"path":"/tmp/data.csv"})]
  [结果: "文件不存在"]
 
--- Step 3 ---
  [调用: read_file({"path":"/tmp/data.csv"})]
  [结果: "文件不存在"]
 
...（一直到 Step 10，MAX_STEPS 救了你一命）
 
// 10 步到了，强制停了。
// 但如果把 MAX_STEPS 设得大一点呢？200?
// 生产环境里，一个不受控的 Agent 跑 200 轮，每轮上下文越滚越大，token 消耗是指数级增长
// 账单飙升

生产级 Agent 需要多层防护机制

三层防护

循环检测 - 模型反复做同样的事且没有进展，检测并打断
API 容错 - API 限流、超时、网络断开，自动重试
Token 预算 - 累计追踪 token 消耗，超预算自动停止

循环检测

死循环

模型很忙，不断做事，但没有任何进展。每一轮都在调用工具，但在原地打转。

常见的三种模式：

通用重复：同一个工具，同样的参数，同样的结果，反复调
乒乓循环：两个操作来回交替，A→B→A→B，每一步都在“做事”，整体没进展
轮询无进展：不断poll检查状态，结果一直是 running

核心思路：哈希 + 滑动窗口

给每一次的工具调用算哈希：把工具名+参数做一次确定性的 JSON 序列化（key排序），然后哈希。这样不管参数顺序怎么变，签名都一样。
维护滑动窗口（近 30 条）：只看最近行为，早期正常行为不太有参考意义，主要看最近若干轮有没有出现重复。
同样的输入+同样的输出=无进展：看参数相同不够。模型调了 10 次 read_file 但每次读的都是不同文件，这是正常探索。只有调用指纹和结果指纹都一样，才算真的没进展。

检测到重复后不是一刀切，而是给模型自救机会，三级响应：

级别	阈值	行为
Warning	5 次	注入系统提醒消息，让模型"醒过来"换策略
Critical	8 次	阻断工具调用，强制停止循环
全局熔断	10 次	无论什么情况，强制停止

循环检测工具

import { createHash } from 'node:crypto';
 
// --- 类型定义 ---
 
export interface ToolCallRecord {
  toolName: string;
  argsHash: string;
  resultHash?: string;
  timestamp: number;
}
 
export type DetectorKind = 'generic_repeat' | 'ping_pong' | 'global_circuit_breaker';
 
export type DetectionResult =
  | { stuck: false }
  | { stuck: true; level: 'warning' | 'critical'; detector: DetectorKind; count: number; message: string };
 
// --- 配置 ---
 
const HISTORY_SIZE = 30;       // 滑动窗口大小
const WARNING_THRESHOLD = 5;   // 警告阈值（演示用，生产环境通常是 10）
const CRITICAL_THRESHOLD = 8;  // 严重阈值（演示用，生产环境通常是 20）
const BREAKER_THRESHOLD = 10;  // 熔断阈值（演示用，生产环境通常是 30）
 
// --- 指纹计算 ---
 
function stableStringify(value: unknown): string {
  if (value === null || typeof value !== 'object') return JSON.stringify(value);
  if (Array.isArray(value)) return `[${value.map(stableStringify).join(',')}]`;
  const keys = Object.keys(value as Record<string, unknown>).sort();
  return `{${keys.map(k => `${JSON.stringify(k)}:${stableStringify((value as any)[k])}`).join(',')}}`;
}
 
function hash(input: string): string {
  return createHash('sha256').update(input).digest('hex').slice(0, 16);
}
 
export function hashToolCall(toolName: string, params: unknown): string {
  return `${toolName}:${hash(stableStringify(params))}`;
}
 
export function hashResult(result: unknown): string {
  return hash(stableStringify(result));
}
 
// --- 滑动窗口 ---
 
const history: ToolCallRecord[] = [];
 
export function recordCall(toolName: string, params: unknown): void {
  history.push({
    toolName,
    argsHash: hashToolCall(toolName, params),
    timestamp: Date.now(),
  });
  if (history.length > HISTORY_SIZE) history.shift();
}
 
export function recordResult(toolName: string, params: unknown, result: unknown): void {
  const argsHash = hashToolCall(toolName, params);
  const resultH = hashResult(result);
  for (let i = history.length - 1; i >= 0; i--) {
    if (history[i].toolName === toolName && history[i].argsHash === argsHash && !history[i].resultHash) {
      history[i].resultHash = resultH;
      break;
    }
  }
}
 
export function resetHistory(): void {
  history.length = 0;
}
 
// --- 检测器 ---
 
function getNoProgressStreak(toolName: string, argsHash: string): number {
  let streak = 0;
  let lastResultHash: string | undefined;
  for (let i = history.length - 1; i >= 0; i--) {
    const r = history[i];
    if (r.toolName !== toolName || r.argsHash !== argsHash) continue;
    if (!r.resultHash) continue;
    if (!lastResultHash) { lastResultHash = r.resultHash; streak = 1; continue; }
    if (r.resultHash !== lastResultHash) break;
    streak++;
  }
  return streak;
}
 
function getPingPongCount(currentHash: string): number {
  if (history.length < 3) return 0;
  const last = history[history.length - 1];
  let otherHash: string | undefined;
  for (let i = history.length - 2; i >= 0; i--) {
    if (history[i].argsHash !== last.argsHash) { otherHash = history[i].argsHash; break; }
  }
  if (!otherHash) return 0;
  let count = 0;
  for (let i = history.length - 1; i >= 0; i--) {
    const expected = count % 2 === 0 ? last.argsHash : otherHash;
    if (history[i].argsHash !== expected) break;
    count++;
  }
  if (currentHash === otherHash && count >= 2) return count + 1;
  return 0;
}
 
// --- 主检测函数 ---
 
export function detect(toolName: string, params: unknown): DetectionResult {
  const argsHash = hashToolCall(toolName, params);
  const noProgress = getNoProgressStreak(toolName, argsHash);
 
  if (noProgress >= BREAKER_THRESHOLD) {
    return { stuck: true, level: 'critical', detector: 'global_circuit_breaker', count: noProgress,
      message: `[熔断] ${toolName} 已重复 ${noProgress} 次且无进展，强制停止` };
  }
 
  const pingPong = getPingPongCount(argsHash);
  if (pingPong >= CRITICAL_THRESHOLD) {
    return { stuck: true, level: 'critical', detector: 'ping_pong', count: pingPong,
      message: `[熔断] 检测到乒乓循环（${pingPong} 次交替），强制停止` };
  }
  if (pingPong >= WARNING_THRESHOLD) {
    return { stuck: true, level: 'warning', detector: 'ping_pong', count: pingPong,
      message: `[警告] 检测到乒乓循环（${pingPong} 次交替），建议换个思路` };
  }
 
  const recentCount = history.filter(h => h.toolName === toolName && h.argsHash === argsHash).length;
  if (recentCount >= CRITICAL_THRESHOLD) {
    return { stuck: true, level: 'critical', detector: 'generic_repeat', count: recentCount,
      message: `[熔断] ${toolName} 相同参数已调用 ${recentCount} 次，强制停止` };
  }
  if (recentCount >= WARNING_THRESHOLD) {
    return { stuck: true, level: 'warning', detector: 'generic_repeat', count: recentCount,
      message: `[警告] ${toolName} 相同参数已调用 ${recentCount} 次，你可能陷入了重复` };
  }
 
  return { stuck: false };
}

API容错

API 出错在生产环境很常见，不能一个 try-catch 一兜了事。

429 限流等一会就好，400 参数错误重试一万次也没用。

核心是分类

哪些值得重试，哪些直接抛。

指数退避加抖动

指数退避：每次重试等的时间翻倍——500ms → 1000ms → 2000ms → 4000ms。这样避免了连续重试轰炸服务端，给它喘息的时间

随机抖动（±25%）：在退避的基础上加一个随机偏移，分散请求，避免"惊群效应"。

Exponential Backoff And Jitter

retry

// --- 错误分类 ---
 
export function isRetryable(error: unknown): boolean {
  if (!(error instanceof Error)) return false;
 
  const message = error.message || '';
 
  // HTTP 状态码判断
  const statusMatch = message.match(/(\d{3})/);
  if (statusMatch) {
    const status = parseInt(statusMatch[1]);
    if ([429, 529, 408].includes(status)) return true;
    if (status >= 500 && status < 600) return true;
    if (status >= 400 && status < 500) return false;
  }
 
  // 网络错误
  if (message.includes('ECONNRESET') || message.includes('EPIPE')) return true;
  if (message.includes('ETIMEDOUT') || message.includes('timeout')) return true;
  if (message.includes('fetch failed') || message.includes('network')) return true;
  // AI SDK 会把流式错误包装成 NoOutputGeneratedError
  if (message.includes('No output generated')) return true;
 
  return false;
}
 
// --- 指数退避 + 随机抖动 ---
 
export function calculateDelay(attempt: number, baseMs = 500, maxMs = 30000): number {
  const exponential = baseMs * Math.pow(2, attempt - 1);
  const capped = Math.min(exponential, maxMs);
  const jitterRange = capped * 0.25;
  const jittered = capped + (Math.random() * 2 - 1) * jitterRange;
  return Math.max(0, Math.round(jittered));
}
 
export function sleep(ms: number): Promise<void> {
  return new Promise(resolve => setTimeout(resolve, ms));
}

Token 预算

Agent 每多跑一步，上下文就多一截，token 消耗是累积的。

一个长对话跑下来，可能都没意识到已经烧了多少钱。

先做最小可用版本——把每步的 token 用量累加起来，超了就停。

budget

export interface BudgetState {
  used: number;
  limit: number;
}
 
// agentLoop 只是消费 budget，不拥有它。
const budget: BudgetState = { used: 0, limit: 15000 };

Agent-loop

import { streamText, type ModelMessage } from 'ai';
import { detect, recordCall, recordResult, resetHistory } from './loop-detection';
import { isRetryable, calculateDelay, sleep } from './retry';
 
const MAX_STEP = 10;
const MAX_RETRIES = 3;
 
export interface BudgetState {
    used: number;
    limit: number;
}
 
export async function agentLoop(
    model: any,
    tools: any,
    messages: ModelMessage[],
    system: string,
    budget: BudgetState,
) {
    let step = 0;
    resetHistory();
 
    while (step < MAX_STEP) {
        step++;
        console.log(`\n--- Step ${step} ---`);
 
 
        let hasToolCall = false;
        let fullText = '';
        let shouldBreak = false;
        let lastToolCall: { name: string; input: unknown } | null = null;
        let stepResponse: Awaited<ReturnType<typeof streamText>['response']>;
        let stepUsage: Awaited<ReturnType<typeof streamText>['usage']>;
 
        for (let attempt = 1; ; attempt++) {
            try {
                const result = await streamText({ model, system, tools, messages, maxRetries: 0, onError: () => { } });
                for await (const part of result.fullStream) {
                    switch (part.type) {
                        case 'text-delta':
                            process.stdout.write(part.text);
                            fullText += part.text;
                            break;
 
                        case 'tool-call': {
                            hasToolCall = true;
                            lastToolCall = { name: part.toolName, input: part.input };
                            console.log(`  [调用: ${part.toolName}(${JSON.stringify(part.input)})]`);
 
                            const detection = detect(part.toolName, part.input);
                            if (detection.stuck) {
                                console.log(`  ${detection.message}`);
                                if (detection.level === 'critical') {
                                    shouldBreak = true;
                                } else {
                                    messages.push({
                                        role: 'user' as const,
                                        content: `[系统提醒] ${detection.message}。请换一个思路解决问题，不要重复同样的操作。`,
                                    });
                                }
                            }
                            recordCall(part.toolName, part.input);
                            break;
                        }
 
                        case 'tool-result':
                            console.log(`  [结果: ${JSON.stringify(part.output)}]`);
                            if (lastToolCall) {
                                recordResult(lastToolCall.name, lastToolCall.input, part.output);
                            }
                            break;
                    }
                }
 
                stepResponse = await result.response;
                stepUsage = await result.usage;
                break;
            } catch (error) {
                if (attempt > MAX_RETRIES || !isRetryable(error as Error)) throw error;
                const delay = calculateDelay(attempt);
                console.log(`  [重试] 第 ${attempt}/${MAX_RETRIES} 次失败，${delay}ms 后重试...`);
                await sleep(delay);
                hasToolCall = false;
                fullText = '';
                shouldBreak = false;
                lastToolCall = null;
            }
        }
 
 
        if (shouldBreak) {
            console.log('\n[循环检测触发，Agent 已停止]');
            break;
        }
 
        // 拿到这一步的完整结果，追加到消息历史
        messages.push(...stepResponse!.messages);
 
        // Token 预算追踪：budget 由调用方持有，跨轮持续累计
        const inp = typeof stepUsage?.inputTokens === 'number' ? stepUsage.inputTokens : (stepUsage?.inputTokens?.total ?? 0);
        const out = typeof stepUsage?.outputTokens === 'number' ? stepUsage.outputTokens : (stepUsage?.outputTokens?.total ?? 0);
        budget.used += inp + out;
        const pct = Math.round(budget.used / budget.limit * 100);
        console.log(`  [Token] ${budget.used}/${budget.limit} (${pct}%)`);
        if (budget.used > budget.limit) {
            console.log('\n[Token 预算耗尽，强制停止]');
            break;
        }
 
 
        // 退出条件：模型没有工具调用
        if (!hasToolCall) {
            if (fullText) {
                console.log();
            }
            break;
        }
 
        // 还有工具调用 ➡️ 继续循环，模型看到工具结果后继续思考
        console.log('   → 模型还在工作,继续下一步...')
    }
 
    if (step >= MAX_STEP) {
        console.log(`\n[达到最大步数限制，强制停止]`)
    }
}