AI 工程化落地：从模型接入到可观测性体系的完整基建

AI 工程化落地：从模型接入到可观测性体系的完整基建

一、AI 功能上线的真实困境：Demo 能跑，生产不敢发

团队花了两周做了一个 AI 智能补全功能，Demo 演示效果很好。上线第一天：Token 费用超预算 3 倍、P99 延迟 8 秒、偶发返回空内容、用户反馈"补全建议完全不对"。更麻烦的是，出了问题不知道怎么排查——LLM 的输出是概率性的，同一个输入可能产生不同输出，传统的调试手段全部失效。

这不是个例，这是 AI 功能从 Demo 到生产的必经之路。AI 工程化的核心问题不是"模型够不够强"，而是"如何让 AI 功能在生产环境中可预测、可观测、可控制"。一个没有可观测性的 AI 功能，就是一个黑盒——你不知道它在做什么，不知道它什么时候会出错，不知道出了错该怎么修。

二、AI 工程化的三层架构：接入层、编排层、可观测层

AI 工程化不是"调一个 API"，而是三层架构的协同：接入层处理模型调用和降级，编排层管理 Prompt 和上下文，可观测层追踪全链路行为。

graph TB subgraph 接入层 CLIENT[客户端请求] --> GW[AI 网关] GW --> ROUTE[模型路由: 主模型/备用模型] ROUTE --> RATE[限流与配额控制] RATE --> CACHE[语义缓存] CACHE --> |命中| RESP[返回缓存结果] CACHE --> |未命中| MODEL[模型调用] end subgraph 编排层 MODEL --> PROMPT[Prompt 编排引擎] PROMPT --> CTX[上下文窗口管理] CTX --> TOOL[工具调用编排] TOOL --> GUARD[输出守卫: 格式/安全/质量] end subgraph 可观测层 GW & MODEL & GUARD --> LOG[结构化日志] LOG --> TRACE[全链路追踪] TRACE --> METRIC[指标聚合] METRIC --> ALERT[告警规则] ALERT --> DASH[监控仪表盘] end GUARD --> RESP style GW fill:#f9f,stroke:#333 style PROMPT fill:#bbf,stroke:#333 style GUARD fill:#f96,stroke:#333 style TRACE fill:#9cf,stroke:#333

关键设计：输出守卫是生产环境的底线。LLM 的输出不可预测，必须在返回给用户之前经过格式校验、安全过滤和质量评估。格式不对就重试，内容不安全就拦截，质量不达标就降级到备用模型。

三、生产级实现：AI 网关 + Prompt 编排 + 全链路可观测

AI 网关：限流、缓存与降级的统一入口

// ai-gateway/index.ts —— AI 调用网关 interface AIGatewayConfig { models: Array<{ id: string; provider: 'openai' | 'anthropic' | 'local'; endpoint: string; maxTokens: number; rpmLimit: number; // 每分钟请求限制 costPerToken: number; // 每千 Token 成本 }>; cacheConfig: { enabled: boolean; ttlMs: number; similarityThreshold: number; // 语义相似度阈值 }; fallbackConfig: { maxRetries: number; fallbackModelId: string; timeoutMs: number; }; } interface AIRequest { prompt: string; model?: string; temperature?: number; maxTokens?: number; metadata?: Record<string, string>; // 追踪用元数据 } interface AIResponse { content: string; model: string; tokensUsed: { input: number; output: number }; latencyMs: number; cached: boolean; traceId: string; } class AIGateway { private rateLimiter: Map<string, { count: number; windowStart: number }> = new Map(); private cache: Map<string, { response: AIResponse; expiresAt: number }> = new Map(); constructor(private config: AIGatewayConfig) {} async request(req: AIRequest): Promise<AIResponse> { const traceId = generateTraceId(); const startTime = Date.now(); // 1. 限流检查 const modelId = req.model ?? this.config.models[0].id; if (!this.checkRateLimit(modelId)) { throw new Error(`模型 ${modelId} 已超过速率限制`); } // 2. 语义缓存查询 if (this.config.cacheConfig.enabled) { const cached = await this.queryCache(req.prompt); if (cached) { this.emitMetric('cache_hit', { model: modelId, traceId }); return { ...cached, cached: true, traceId }; } } // 3. 模型调用（含超时和重试） const model = this.config.models.find((m) => m.id === modelId); if (!model) { throw new Error(`模型 ${modelId} 未配置`); } let lastError: Error | null = null; let attemptCount = 0; const maxAttempts = 1 + this.config.fallbackConfig.maxRetries; while (attemptCount < maxAttempts) { attemptCount++; try { const response = await this.callModelWithTimeout(model, req, traceId); // 4. 输出守卫 const guarded = this.guardOutput(response.content, req); if (!guarded.valid) { this.emitMetric('output_guard_reject', { model: modelId, reason: guarded.reason, traceId, }); // 输出不合格，重试 continue; } const latencyMs = Date.now() - startTime; const aiResponse: AIResponse = { content: guarded.content, model: modelId, tokensUsed: response.tokensUsed, latencyMs, cached: false, traceId, }; // 5. 写入缓存 if (this.config.cacheConfig.enabled) { this.writeCache(req.prompt, aiResponse); } // 6. 记录指标 this.emitMetric('request_success', { model: modelId, latencyMs, tokensUsed: response.tokensUsed, traceId, }); return aiResponse; } catch (err) { lastError = err instanceof Error ? err : new Error(String(err)); this.emitMetric('request_error', { model: modelId, attempt: attemptCount, error: lastError.message, traceId, }); } } // 6. 所有重试失败，降级到备用模型 const fallbackModel = this.config.models.find( (m) => m.id === this.config.fallbackConfig.fallbackModelId ); if (fallbackModel && fallbackModel.id !== modelId) { this.emitMetric('fallback_triggered', { fromModel: modelId, toModel: fallbackModel.id, traceId, }); try { const response = await this.callModelWithTimeout(fallbackModel, req, traceId); const latencyMs = Date.now() - startTime; return { content: response.content, model: fallbackModel.id, tokensUsed: response.tokensUsed, latencyMs, cached: false, traceId, }; } catch { // 降级也失败，抛出原始错误 } } throw lastError ?? new Error('AI 请求失败'); } // 限流检查：滑动窗口算法 private checkRateLimit(modelId: string): boolean { const model = this.config.models.find((m) => m.id === modelId); if (!model) return false; const now = Date.now(); const windowMs = 60_000; const key = modelId; const current = this.rateLimiter.get(key); if (!current || now - current.windowStart > windowMs) { this.rateLimiter.set(key, { count: 1, windowStart: now }); return true; } if (current.count >= model.rpmLimit) { return false; } current.count++; return true; } // 输出守卫：格式校验 + 安全过滤 + 质量评估 private guardOutput( content: string, req: AIRequest ): { valid: boolean; content: string; reason?: string } { // 空内容检查 if (!content || content.trim().length === 0) { return { valid: false, content, reason: 'empty_output' }; } // 安全过滤：检测敏感信息泄露 const sensitivePatterns = [ /sk-[a-zA-Z0-9]{32,}/, // API Key /\b\d{3}-\d{4}-\d{4}\b/, // 电话号码 /\b[\w.-]+@[\w.-]+\.\w+\b/, // 邮箱地址 ]; for (const pattern of sensitivePatterns) { if (pattern.test(content)) { return { valid: false, content, reason: 'sensitive_info_leak' }; } } // JSON 格式校验（如果期望 JSON 输出） if (req.metadata?.expectJson) { try { JSON.parse(content); } catch { return { valid: false, content, reason: 'invalid_json' }; } } // 长度校验：过短的输出通常质量不高 if (content.length < 10) { return { valid: false, content, reason: 'output_too_short' }; } return { valid: true, content }; } private async callModelWithTimeout( model: AIGatewayConfig['models'][0], req: AIRequest, traceId: string ): Promise<{ content: string; tokensUsed: { input: number; output: number } }> { const controller = new AbortController(); const timeout = setTimeout( () => controller.abort(), this.config.fallbackConfig.timeoutMs ); try { const response = await fetch(model.endpoint, { method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${process.env.AI_API_KEY}`, 'X-Trace-Id': traceId, }, body: JSON.stringify({ model: model.id, messages: [{ role: 'user', content: req.prompt }], temperature: req.temperature ?? 0.1, max_tokens: req.maxTokens ?? model.maxTokens, }), signal: controller.signal, }); if (!response.ok) { throw new Error(`模型调用失败: ${response.status}`); } const data = await response.json(); return { content: data.choices[0].message.content, tokensUsed: { input: data.usage.prompt_tokens, output: data.usage.completion_tokens, }, }; } finally { clearTimeout(timeout); } } private emitMetric(event: string, data: Record<string, unknown>): void { // 结构化日志输出，供可观测系统采集 console.log(JSON.stringify({ type: 'ai_metric', event, timestamp: Date.now(), ...data, })); } private async queryCache(prompt: string): Promise<AIResponse | null> { // 简化实现：精确匹配缓存 const key = this.cacheKey(prompt); const entry = this.cache.get(key); if (entry && entry.expiresAt > Date.now()) { return entry.response; } return null; } private writeCache(prompt: string, response: AIResponse): void { const key = this.cacheKey(prompt); this.cache.set(key, { response, expiresAt: Date.now() + this.config.cacheConfig.ttlMs, }); } private cacheKey(prompt: string): string { // 使用 prompt 的哈希作为缓存键 const crypto = require('crypto'); return crypto.createHash('sha256').update(prompt).digest('hex').slice(0, 16); } } function generateTraceId(): string { return `ai-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; }

全链路可观测：结构化日志 + 指标聚合 + 告警规则

// observability/ai-metrics.ts —— AI 功能的可观测性指标定义 interface AIMetrics { // 请求指标 requestTotal: number; requestSuccess: number; requestError: number; requestLatencyP50: number; requestLatencyP99: number; // Token 消耗指标 tokensInputTotal: number; tokensOutputTotal: number; costTotal: number; // 美元 // 缓存指标 cacheHitRate: number; // 质量指标 guardRejectRate: number; fallbackRate: number; userAcceptRate: number; // 用户采纳率（需要前端埋点） } // 告警规则定义 const AI_ALERT_RULES = [ { name: 'AI 请求错误率过高', condition: (m: AIMetrics) => m.requestError / m.requestTotal > 0.05, severity: 'critical' as const, message: 'AI 请求错误率超过 5%，请检查模型服务状态', }, { name: 'AI 响应延迟过高', condition: (m: AIMetrics) => m.requestLatencyP99 > 5000, severity: 'warning' as const, message: 'AI P99 延迟超过 5 秒，影响用户体验', }, { name: 'Token 消耗异常', condition: (m: AIMetrics) => m.costTotal > 100, // 日消耗超过 $100 severity: 'warning' as const, message: 'AI Token 日消耗超过预算阈值，请检查是否有异常调用', }, { name: '输出守卫拦截率过高', condition: (m: AIMetrics) => m.guardRejectRate > 0.1, severity: 'warning' as const, message: 'AI 输出被守卫拦截超过 10%，模型输出质量可能下降', }, { name: '降级率过高', condition: (m: AIMetrics) => m.fallbackRate > 0.2, severity: 'critical' as const, message: 'AI 降级率超过 20%，主模型可能不可用', }, ];

四、AI 工程化的边界：成本、延迟与质量的三角约束

成本-延迟-质量的不可能三角

实际项目中必须在三者之间做出取舍。推荐策略：对高频简单请求优化成本和延迟，对低频复杂请求优化质量。具体来说：80% 的请求走小模型 + 缓存（低成本低延迟），20% 的请求走大模型 + 校验（高质量）。

语义缓存的适用边界

语义缓存（相似问题命中同一缓存）在 FAQ 类场景下效果很好，但在创意性场景下是灾难——用户不希望每次得到相同的回答。判断标准：如果用户期望确定性答案（如"这个 API 怎么用"），适合缓存；如果用户期望多样性（如"帮我写一段文案"），不适合缓存。

可观测性的维护成本

全链路追踪和结构化日志不是免费的。每次 AI 请求产生的日志量是普通 API 请求的 5-10 倍（包含 Prompt、输出、Token 统计、延迟分解等）。在高频场景下，日志存储和检索的成本不可忽视。建议：生产环境保留 7 天详细日志，超过 7 天聚合为指标数据。

五、总结

AI 工程化的核心是三层架构：接入层处理限流、缓存与降级，编排层管理 Prompt 和上下文，可观测层追踪全链路行为。输出守卫是生产环境的底线，必须在返回给用户之前完成格式校验、安全过滤和质量评估。成本、延迟、质量三者不可兼得，推荐按请求复杂度分级处理：简单请求走小模型+缓存，复杂请求走大模型+校验。语义缓存适用于确定性场景，不适用于创意性场景。可观测性的维护成本需要纳入预算，详细日志保留 7 天，历史数据聚合为指标。