从 Demo 到产品：AI 落地的几个工程坑
一、Demo 跑通很容易，做成产品很难
很多团队都有一个共同经历：Demo 阶段一切顺利，输入规范，模型响应正常，用户也按预期操作。但一旦要变成正式产品，问题就来了——输入异常、模型幻觉、用户乱点、网络超时、并发冲突，这些“阴暗路径”在 Demo 里根本遇不到。从 0 到 Demo 的距离，往往远不如从 Demo 到产品的距离。
最头疼的是延迟的不确定性。本地跑模型，响应时间稳定在 2 秒左右；一旦部署到云端，受网络波动、模型排队、冷启动影响，P99 延迟可能直接飙到 15 秒以上。更麻烦的是长尾延迟——99% 的请求 3 秒内完成，但剩下的 1% 超过 10 秒。这种分布对用户体验的破坏，远比平均延迟上升更严重。
成本也是个容易被低估的问题。Demo 阶段免费 API 额度够用，产品化后日活一上来，Token 消耗线性增长，月度费用可能从几十美元跳到几万。还有那些“隐性成本”——向量数据库的存储费、GPU 推理的闲置浪费、Prompt 调试的人力投入，在 Demo 阶段根本看不出来。
二、三个必须解决的工程问题
要把 AI 做成产品，可靠性、可观测性和成本控制是绕不开的三件事。
flowchart TB subgraph 可靠性工程 R1[输入校验与清洗] R2[模型降级与熔断] R3[输出审核与过滤] R4[重试与超时策略] end subgraph 可观测性工程 O1[请求级追踪：TraceID 串联全链路] O2[指标监控：延迟/错误率/Token 消耗] O3[日志聚合：Prompt 与 Response 快照] O4[告警规则：P99 延迟/错误率阈值] end subgraph 成本治理工程 C1[Token 预算与限流] C2[模型路由：按任务复杂度分流] C3[缓存策略：相似请求复用结果] C4[用量计费：按 Token 消耗定价] end R1 & R2 & R3 & R4 --> STABLE[服务稳定] O1 & O2 & O3 & O4 --> OBSERVABLE[问题可定位] C1 & C2 & C3 & C4 --> SUSTAINABLE[商业可持续] STABLE & OBSERVABLE & SUSTAINABLE --> PRODUCTION[生产级 AI 产品] style STABLE fill:#e8f5e9 style OBSERVABLE fill:#e3f2fd style SUSTAINABLE fill:#fff3e0 style PRODUCTION fill:#ffe082这三者其实是环环相扣的：没有可观测性，问题发现不了，可靠性就无从谈起；没有可观测性的数据支撑，成本控制就是瞎猜；而可靠性如果不受成本约束，无限投入虽然能堆出高可用，但商业模式跑不通。
三、可靠性工程：把模型调用变成稳定服务
# production/reliability.py — AI 产品的可靠性工程组件 import time import hashlib import json from dataclasses import dataclass, field from typing import Optional, Callable from enum import Enum class ModelTier(Enum): PREMIUM = "premium" # 高质量模型：GPT-4, Claude STANDARD = "standard" # 标准模型：GPT-4o-mini FALLBACK = "fallback" # 兜底模型：本地部署 @dataclass class AIRequest: """AI 请求的标准化封装""" request_id: str prompt: str max_tokens: int = 2048 temperature: float = 0.7 tier: ModelTier = ModelTier.STANDARD metadata: dict = field(default_factory=dict) @dataclass class AIResponse: """AI 响应的标准化封装""" request_id: str content: str model: str tier: ModelTier input_tokens: int output_tokens: int latency_ms: float filtered: bool = False # 是否经过内容过滤 fallback_used: bool = False # 是否使用了降级模型 cached: bool = False # 是否命中缓存 class AIReliabilityLayer: """AI 可靠性层：输入校验、降级熔断、输出审核、重试超时""" def __init__( self, call_model_fn: Callable, content_filter_fn: Optional[Callable] = None, cache_ttl_seconds: int = 3600, ): self._call_model = call_model_fn self._content_filter = content_filter_fn self._cache_ttl = cache_ttl_seconds self._cache: dict[str, tuple[float, AIResponse]] = {} # 熔断器状态 self._failure_counts: dict[str, int] = {} self._circuit_open: dict[str, bool] = {} self._last_failure_time: dict[str, float] = {} # 降级链配置 self._fallback_chain = { ModelTier.PREMIUM: ModelTier.STANDARD, ModelTier.STANDARD: ModelTier.FALLBACK, } async def invoke(self, request: AIRequest) -> AIResponse: """执行 AI 调用，包含完整的可靠性保障流程""" start_time = time.perf_counter() # 1. 输入校验与清洗 validated_prompt = self._validate_and_sanitize(request.prompt) request.prompt = validated_prompt # 2. 缓存检查 cache_key = self._compute_cache_key(request) cached = self._check_cache(cache_key) if cached: cached.cached = True return cached # 3. 熔断检查 if self._is_circuit_open(request.tier): # 当前层级熔断，尝试降级 fallback_tier = self._fallback_chain.get(request.tier) if fallback_tier and not self._is_circuit_open(fallback_tier): request.tier = fallback_tier elif request.tier != ModelTier.FALLBACK: # 所有层级都熔断，使用兜底响应 return self._fallback_response(request, start_time) # 4. 执行模型调用（含重试） response = await self._call_with_retry(request, max_retries=2) # 5. 输出审核 if self._content_filter: filter_result = self._content_filter(response.content) if not filter_result.safe: response.content = filter_result.replacement response.filtered = True # 6. 写入缓存 latency_ms = (time.perf_counter() - start_time) * 1000 response.latency_ms = latency_ms self._write_cache(cache_key, response) return response def _validate_and_sanitize(self, prompt: str) -> str: """输入校验与清洗""" # 长度限制 if len(prompt) > 50000: prompt = prompt[:50000] # 去除控制字符 prompt = ''.join( c for c in prompt if c.isprintable() or c in '\n\r\t' ) # 检测注入攻击模式 injection_patterns = [ 'ignore previous instructions', '忽略以上指令', 'system:', ] prompt_lower = prompt.lower() for pattern in injection_patterns: if pattern in prompt_lower: # 标记但不拒绝，由输出审核兜底 break return prompt.strip() def _compute_cache_key(self, request: AIRequest) -> str: """计算缓存键：基于 Prompt 和参数的哈希""" key_data = f"{request.prompt}:{request.max_tokens}:{request.temperature}:{request.tier.value}" return hashlib.md5(key_data.encode()).hexdigest() def _check_cache(self, key: str) -> Optional[AIResponse]: """检查缓存""" if key in self._cache: timestamp, response = self._cache[key] if time.time() - timestamp < self._cache_ttl: return response del self._cache[key] return None def _write_cache(self, key: str, response: AIResponse) -> None: """写入缓存""" self._cache[key] = (time.time(), response) def _is_circuit_open(self, tier: ModelTier) -> bool: """检查熔断状态""" tier_name = tier.value if not self._circuit_open.get(tier_name, False): return False # 熔断 30 秒后半开 last_failure = self._last_failure_time.get(tier_name, 0) if time.time() - last_failure > 30: self._circuit_open[tier_name] = False self._failure_counts[tier_name] = 0 return False return True def _record_failure(self, tier: ModelTier) -> None: """记录失败，更新熔断状态""" tier_name = tier.value self._failure_counts[tier_name] = self._failure_counts.get(tier_name, 0) + 1 self._last_failure_time[tier_name] = time.time() # 连续 3 次失败触发熔断 if self._failure_counts[tier_name] >= 3: self._circuit_open[tier_name] = True def _record_success(self, tier: ModelTier) -> None: """记录成功，重置熔断计数""" self._failure_counts[tier.value] = 0 self._circuit_open[tier.value] = False async def _call_with_retry( self, request: AIRequest, max_retries: int, ) -> AIResponse: """带重试的模型调用""" last_error = None for attempt in range(max_retries + 1): try: response = await self._call_model(request) self._record_success(request.tier) return response except Exception as e: last_error = e self._record_failure(request.tier) # 最后一次尝试失败，尝试降级 if attempt == max_retries: fallback_tier = self._fallback_chain.get(request.tier) if fallback_tier and not self._is_circuit_open(fallback_tier): request.tier = fallback_tier try: response = await self._call_model(request) response.fallback_used = True self._record_success(fallback_tier) return response except Exception: pass # 指数退避 if attempt < max_retries: wait_time = min(2 ** attempt, 8) # 1s, 2s, 4s, 8s time.sleep(wait_time) # 所有尝试失败，返回兜底响应 return self._fallback_response(request, time.perf_counter()) def _fallback_response(self, request: AIRequest, start_time: float) -> AIResponse: """兜底响应：当所有模型不可用时返回""" return AIResponse( request_id=request.request_id, content='抱歉，服务暂时不可用，请稍后重试。', model='fallback', tier=ModelTier.FALLBACK, input_tokens=0, output_tokens=0, latency_ms=(time.perf_counter() - start_time) * 1000, fallback_used=True, )这套可靠性层把输入校验、缓存、熔断、重试、降级和输出审核封装成一个统一的调用接口。业务代码只需要调invoke方法，不用管底层的容错逻辑。熔断器采用“连续 3 次失败触发、30 秒后半开”的策略，降级链按 Premium→Standard→Fallback 的顺序逐级降级。
四、怎么不让钱烧得太快
Token 预算与限流：给每个用户设个日 Token 预算，超了就降级到轻量模型或者提示用户。企业客户可以买更高的预算。限流策略要区分模型层级——Premium 模型的并发限制更严，Standard 模型可以宽松些。
缓存策略：语义缓存（Semantic Cache）通过向量相似度匹配，把相似请求的响应复用。当新请求和缓存里的请求语义相似度超过 0.95 时，直接返回缓存结果。这对 FAQ 类请求效果不错，缓存命中率能到 40% 以上，但开放式对话基本没用。
用量计费：按 Token 消耗定价是主流，但得让用户清楚自己用了多少。建议在响应里带上input_tokens和output_tokens字段，在用户面板里提供日/月用量统计和费用预估。
五、总结
AI 产品化其实就是解决三个问题：怎么让服务稳定、怎么让问题可发现、怎么让成本可控。可靠性层把 Demo 级的模型调用升级成生产级服务；可观测性确保问题能发现能定位；成本治理通过 Token 预算、语义缓存和用量计费，让商业模式能跑通。这三件事，缺一不可。
