如何构建可靠的人工智能系统。

如果任何验证环节失败，您不能默默接受错误的数据。此时您有三种选择：

1. 重新请求并提供更明确的指示：要求模型根据更详细的指令再次尝试生成数据。

2. 交由人工审核：记录失败情况，并将相关数据放入审核队列中等待人工处理。

3. 使用默认值：返回一个需要人工处理的默认值作为替代结果。

确定性规则

这里有一条必须严格遵守的规则：

如果某个问题可以用“if语句”来解决，那就不要使用人工智能。

验证电子邮件格式？使用正则表达式。解析日期？使用专门的日期处理库。检查字符串中是否包含特定关键词？使用字符串处理方法。进行数学计算？直接用传统的数学方法即可。

人工智能成本高昂，且其决策过程属于概率性判断；而传统代码则是免费、即时且结果具有确定性的。因此，只有在那些真正存在模糊性的任务中，才应该使用人工智能——比如从非结构化文本中提取信息、生成内容，或者对复杂的输入数据进行推理分析。其余的所有工作，都应该由确定性强的代码来处理。

故障模式#2：隐性故障

问题所在

在人工智能应用中，模型出现“幻觉”现象的情况相当普遍，这些问题的表现形式包括准确性下降、训练数据过时，或是分类结果错误等等。这种故障模式最为可怕，因为人们根本意识不到它的存在。

以准确性漂移为例：你使用2024年的数据训练了模型，但现在已经是2026年中期了，你的供应商改变了发票格式，导致模型的分类准确率从95%降到了71%。直到你进行季度审计时才会发现这个问题，而到那时，已经有成千上万的记录被错误地处理掉了。

道理很简单：你看不见的问题，就无法解决。

解决方案：可观测的流程体系

任何正式投入生产的人工智能系统，从一开始就必须具备可观测性。在实际的生产环境中，这一机制的具体运作方式如下：

在上述流程图中：

1. 输入数据到达：用户提出请求（无论是支持工单、文档还是查询信息），你会记录下请求ID、时间戳、用户ID以及输入数据的哈希值（用于去重处理）。

2. 大语言模型处理：请求会被传递给相应的人工智能模型。你需要记录下使用了哪个模型、处理耗时多少、消耗了多少计算资源，以及最重要的——模型的置信度评分。

3. 置信度判断环节：在这里你会决定如何处理该请求：

   • 置信度很高（>0.8）：自动执行相应操作。

   • 置信度中等（0.6–0.8）：将请求发送到人工审核队列中。

   • 置信度很低（<0.6）：立即上报问题并发出警报。

4. 监控面板：所有这些数据都会被传输到监控系统中，你可以通过这些数据来追踪系统运行中的各种趋势变化。

通过实施有效的监控机制，你可以及时发现系统中的问题并加以解决。监控不仅能帮助你发现问题，还能为你提供诊断和修复问题的所需数据，从而让你在短短几小时内就能解决问题，而无需花费数月时间。

您正在测量什么以及为何要测量这些内容：

我们的目标并不是让人类完全脱离这个流程，而是只有在系统确实无法做出准确判断时，才让人类介入。

故障模式#3：成本失控

问题所在

您用10个请求来测试您的开发流程，结果一切正常，且仅花费了50美分。但当您将这个系统部署到生产环境中后，如果有1,000个请求同时访问您的API，那么当天的费用就会达到500美元。

或者，如果您错误地编写了重试逻辑，导致系统会无限次地调用API，那么当天的费用可能会高达5,000美元。

又或者，您在所有情况下都使用最昂贵的模型，甚至包括那些可以用更便宜的模型来处理的简单任务，这样也会造成不必要的成本浪费。

事实就是：“适用于10个请求”的系统，并不等于“适用于10,000个请求”的系统。”规模的变化会带来巨大的影响。

解决方案：带有断路器的门控流程

要想将一个脆弱的原型开发成稳定的生产系统，就必须摒弃那种直接将用户输入连接到大型语言模型API的简单做法。相反，应该采用门控流程这种设计。

可以把这种架构想象成一排道闸。只有当请求成功通过每一道关卡后，才能继续产生费用。如果有任何一道关卡关闭，请求就会立即被拒绝，从而保护您的预算以及上游系统不受影响。

从上面的图示中可以看出，这些关卡包括：

1. 速率限制机制

2. 缓存检查功能

3. 请求队列系统

4. 断路器装置

让我们来逐一了解这些关卡的作用。

关卡1：速率限制

第一道防线的作用是在恶意请求进入系统之前就将其阻止。在标准的网络开发中，速率限制主要是为了保护服务器的CPU资源；而在人工智能开发中，它的作用则是保护您的预算不受损失。

关卡2：缓存检查

最节省成本的AI API调用，就是那些根本不需要被执行的请求。许多AI请求其实是重复性的，或者内容非常相似。因此，应该积极利用缓存机制来减少不必要的计算开销。

关卡3：请求队列

请求队列的作用是确保请求能够按照合理的顺序被处理。通过设置队列长度和优先级，可以避免系统因为过多的并发请求而陷入混乱。

大语言模型API与标准的REST API不同，请求的处理通常需要10到30秒才能完成。如果同时有500名用户点击“提交”按钮，服务器将无法同时建立500个连接，否则就会崩溃或超出服务提供商规定的并发限制。通过使用请求队列，可以批量处理这些请求，并以受控的速度逐一执行它们，从而解决这个问题。

第四道屏障：断路器

对于短暂的网络故障来说，重试机制是必要的；但在发生真正的服务中断时，这种机制反而会造成更大的破坏。如果某个大语言模型服务提供商出现了故障，并返回了500个错误信息，那么一个简单的重试循环就会不断反复地向他们的API发送请求，从而导致资源浪费。

如何实现带屏障的请求处理流程

下面是一个示例，展示了这四道屏障是如何协同工作的：

步骤1：速率限制器（使用Redis实现）

import { RateLimiterRedis } from "rate-limiter-flexible";
import Redis from "ioredis";

const redis = new Redis({
  host: process.env.REDIS_HOST,
  port: 6379
});

// 为每个用户设置速率限制
const userLimiter = new RateLimiterRedis({
  storeClient: redis,
  keyPrefix: "rl:user",
  points: 100,        
  duration: 3600,     
  blockDuration: 60   
});

// 全局范围设置速率限制
const globalLimiter = new RateLimiterRedis({
  storeClient: redis,
  keyPrefix: "rl:global",
  points: 1000,       
  duration: 3600      
});

步骤2：缓存层

import { createHash } from "crypto";

class AICache {
  private redis: Redis;
  private ttl: number = 3600; 

  hashInput(input: string): string {
    return createHash("sha256").update(input).digest("hex");
  }

  async get(input: string): Promise {
    const key = `ai:cache:${this.hashInput(input)}`;
    const cached = await this.redis.get(key);
    
    if (cached) {
      // 缓存命中，直接返回结果！
      await metrics.increment("ai.cachehits");
      return JSON.parse(cached);
    }
    
    await metrics.increment("ai.cache.misses");
    return null;
  }

  async set(input: string, result: T): Promise {
    const key = `ai:cache:${this.hashInput(input)}`;
    await this.redis.setex(key, this.ttl, JSON.stringify(result));
  }
}

步骤3：请求队列

import Queue from "bull";

const aiQueue = new Queue("ai-requests", {
  redis: {
    host: process.env.REDIS_HOST,
    port: 6379
  }
});

aiQueue.process(5, async (job) => {
  // 同时最多只能有5个请求在处理中
  const { ticket } = job.data;
  return await callLLM(ticket);
});

async function enqueueRequest.ticket: Ticket) {
  const job = await aiQueue.add(
    { ticket },
    {
      attempts: 3,
      backoff: {
        type: "exponential",
        delay: 2000
      }
    }
  );
  
  return job_finished(); 
}

步骤4：断路器

enum CircuitState {
  CLOSED,   
  OPEN,     
  HALF_OPEN 
}

class CircuitBreaker {
  private state = CircuitState.CLOSED;
  private failures = 0;
  private lastFailureTime?: Date;
  private successesInHalfOpen = 0;

  private readonly failureThreshold = 3;
  private readonly openDurationMs = 5 * 60 * 1000; 
  private readonly halfOpenSuccesses = 2;

  async execute(
    fn: () => Promise,
    fallback?: () => T
  ): Promise {
    if (this.state === CircuitState.OPEN) {
      const elapsed = Date.now() - (this.lastFailureTime?.getTime() || 0);
      
      if (elapsed < this.openDurationMs) {
        // 仍处于开启状态——使用回退处理方式或抛出异常
        if (fallback) {
          logger.warn("电路处于开启状态——使用回退处理方式");
          return fallback();
        }
        throw new Error("电路断路器处于开启状态——服务不可用");
      }
      
      // 将状态切换为半开状态
      this.state = CircuitState.HALF_OPEN;
      logger.info("电路正在切换到半开状态");
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  private onSuccess() {
    if (this.state === CircuitState.HALF_OPEN) {
      this.successesInHalfOpen++;
      
      if (this.successesInHalfOpen >= this.halfOpenSuccesses) {
        // 服务已恢复——关闭电路
        this.state = CircuitState.CLOSED;
        this.failures = 0;
        this.successesInHalfOpen = 0;
        logger.info("电路已关闭——服务已恢复");
      }
    } else {
      this FAILures = 0;
    }
  }

  private onFailure() {
    thisfails++;
    this.lastFailureTime = new Date();

    if (this.state === CircuitState.HALF_OPEN) {
      // 在测试过程中出现故障——将电路状态切换回开启状态
      this.state = CircuitState OPEN;
      this.successesInHalfOpen = 0;
      logger.error("在半开状态测试期间电路被重新打开");
    } else if (thisfails >= this.failureThreshold) {
      // 故障次数过多——将电路状态切换为开启状态
      this.state = CircuitState.OPEN;
      logger.error(`在发生${this.failures}次故障后，电路被重新打开`);
    }
  }
}

步骤5：将所有环节整合在一起

const cache = new AICache();
const circuitBreaker = new CircuitBreaker();

async function processWithGatedPipeline(ticket: Ticket) {
  try {
    await userLimiterconsume(ticket.userId);
    await globalLimiterconsume("global");
  } catch (error) {
    throw new Error("达到了速率限制。请稍后再试。”);
  }

  const cacheKey = ticket.body;
  const cached = await cache.get(cacheKey);
  if (cached) {
    logger.info("从缓存中找到了结果，直接返回缓存值.");
    return cached;
  }

  const queuedResult = await enqueueRequest(ticket);

  const result = await circuitBreaker.execute(
    async () => {
      const classification = await callLLMticket);
      await cache.set(cacheKey, classification);
      return classification;
    },
    () => ({
      category: "other",
      confidence: 0,
      requiresHumanReview: true,
      reason: "service_unavailable"
    })
  );

  return result;
}

这种方式能够实现以下目标：

• 速率限制：防止滥用系统资源导致成本失控。

• 缓存机制

  : 对重复请求进行缓存处理，从而将成本降低30%至40%。

• 队列管理

  : 在流量激增时避免服务器过载。

• 断路器机制

  : 在系统出现故障时能够迅速停止相关操作，从而避免不必要的重复尝试和资源浪费。

class TicketWorkflow {
  async processTicket(rawInput: unknown): Promise {
    const requestId = generateId();
    const startTime = Date.now();

    try {
      // 第一层：输入验证 + 速率限制 + 缓存
      const ticket = validateTicketInput(rawInput);
      await rateLimiterconsume(ticket.userId);
      
      const cached = await cache.get(ticket.body);
      if (cached) return { ...cached, source: "cache" };

      // 第二层：使用断路器进行AI处理
      const classification = await circuitBreaker.execute(() => 
        classifyTicket(ticket)
      );

      // 第三层：输出验证 + 根据置信度分配处理方式
      const validated = validateClassification(classification);
      
      let action: string;
      if (validated.confidence >= 0.8) {
        await sendToAgentticket, validated);
        action = "auto_assigned";
      } else {
        await sendToReviewQueue(ticket, validated);
        action = "needs_review";
      }

      // 第四层：记录所有操作以便进行可观测性分析
      await logger.log({
        requestId,
        userId: ticket.userId,
        confidence: validated.confidence,
        action,
        latencyMs: Date.now() - startTime,
        cost: calculateCost(classification.tokensUsed)
      });

      await cache.set(ticket.body, validated);
      return { classification: validated, action };
    } catch (error) {
      await logger.logError(requestId, error);
      throw error;
    }
  }
}