对企业客户的交付复盘：Demo 跑通了，生产环境才刚刚开始

对企业客户的交付复盘Demo 跑通了生产环境才刚刚开始一、个性化深度引言每个 AI 项目都有两个版本演示版和生产版。它们之间的差距比产品和研发最初预估的要大一个数量级。去年最深刻的教训来自一个内部文档问答项目。Demo 跑得行云流水——上传三份 PDF问上季度销售额是多少3秒出答案数据准确。会议室里客户频频点头当场确认了二期合同。然后进入交付部署阶段。生产环境的第一周就出了状况。客户实际上传的不是3份文档是3000份。响应时间从3秒飙升到45秒。有两个文件夹里的文档文件名是新建文件夹(5)-副本(3).docx——OCR 对这些文件名的处理逻辑直接崩溃了。用户上传了一份带有彩色背景的扫描件预处理模块把背景当作文本区域输出了3000个乱码字符。Demo 验证的是这条路能走通生产环境验证的是这条路能长期走、稳定走、在恶劣条件下走。这篇文章复盘一下交付过程中的真实问题。二、个性化原理剖析AI 项目从 Demo 到生产的差距图Demo 到生产的五大差距数据规模3 vs 3000、异常输入无 vs 百种、并发压力单用户 vs 多用户、部署环境本地 vs 容器化、运维能力手动排查 vs 自动监控。三、个性化代码实践生产环境的健壮性处理代码import asyncio import time import logging import traceback from dataclasses import dataclass, field from typing import List, Dict, Optional, Any, Callable from collections import defaultdict from enum import Enum import psutil import signal # 生产级日志配置——设计原因调试靠日志不能靠print logging.basicConfig( levellogging.INFO, format%(asctime)s | %(levelname)s | %(name)s | %(message)s, handlers[ logging.FileHandler(prod.log), logging.StreamHandler() ] ) logger logging.getLogger(ai_service) class ServiceStatus(Enum): HEALTHY healthy DEGRADED degraded # 降级运行 OVERLOADED overloaded # 过载 DOWN down dataclass class RequestContext: 请求上下文——设计原因把每个请求的状态封装便于追踪和调试 request_id: str start_time: float input_size: int timeout: float 30.0 # 默认30秒超时 property def elapsed(self) - float: return time.time() - self.start_time class ProductionAIEngine: 生产级AI服务引擎——设计原因Demo里没有的超时/限流/降级/监控全在这 # 配置参数——设计原因集中管理方便运维调整 MAX_INPUT_SIZE 50 * 1024 * 1024 # 50MB——设计原因单文件超大直接拒防止OOM MAX_CONCURRENT 10 # 最大并发 REQUEST_TIMEOUT 30 # 秒——设计原因30秒没结果用户也等不下去了 MAX_RETRIES 2 # 最大重试次数 def __init__(self): self._semaphore asyncio.Semaphore(self.MAX_CONCURRENT) self._request_count 0 self._error_count 0 self._latency_records: List[float] [] self._start_time time.time() # 注册优雅退出——设计原因收到SIGTERM时完成当前请求再退出 self._shutting_down False signal.signal(signal.SIGTERM, self._handle_shutdown) def _handle_shutdown(self, signum, frame): 优雅退出处理——设计原因K8s发SIGTERM时给30秒完成当前请求 logger.info(收到退出信号等待当前请求完成...) self._shutting_down True async def process(self, input_data: Any, request_id: str None) - Dict[str, Any]: 主处理入口——设计原因统一入口方便加中间件限流/超时/监控 ctx RequestContext( request_idrequest_id or freq_{int(time.time()*1000)}, start_timetime.time(), input_sizelen(str(input_data)) ) # 第一层输入校验——设计原因入口处拦截异常输入不让脏数据进模型 validation_error self._validate_input(input_data, ctx) if validation_error: return self._error_response(ctx, validation_error) # 第二层并发控制——设计原因信号量限流比线程池更轻量 async with self._semaphore: return await self._process_with_retry(input_data, ctx) def _validate_input(self, data: Any, ctx: RequestContext) - Optional[str]: 输入校验——设计原因生产环境异常输入种类远超Demo时的想象 # 空输入检查 if data is None: return 输入为空 # 文件大小检查——设计原因有人传过800MB的图片OOM重启了三次才发现 if ctx.input_size self.MAX_INPUT_SIZE: return f输入过大({ctx.input_size/1024/1024:.1f}MB)最大{self.MAX_INPUT_SIZE/1024/1024}MB # 文件签名检查——设计原因文件名是pdf但实际是exe这类异常在生产环境很常见 if isinstance(data, bytes) and len(data) 4: if data[:4] bPK\x03\x04: return 看起来像是ZIP/DOCX文件请先解压并转换为PDF return None async def _process_with_retry(self, data: Any, ctx: RequestContext) - Dict[str, Any]: 带重试的处理——设计原因临时性故障网络抖动重试后大概率恢复 last_error None for attempt in range(self.MAX_RETRIES 1): try: result await asyncio.wait_for( self._do_process(data, ctx), timeoutself.REQUEST_TIMEOUT ) # 记录成功指标 self._record_success(ctx) return self._success_response(ctx, result) except asyncio.TimeoutError: logger.warning(f请求超时: {ctx.request_id} (尝试{attempt1})) last_error 处理超时 # 超时也重试——设计原因GPU可能刚好在执行其他推理 continue except MemoryError: # 内存不足不重试——设计原因重试也大概率再次OOM logger.error(f内存溢出: {ctx.request_id}) self._record_error(ctx) return self._error_response(ctx, 服务资源不足请稍后重试) except Exception as e: logger.error(f处理异常: {ctx.request_id} | {e}\n{traceback.format_exc()}) last_error str(e) if attempt self.MAX_RETRIES: # 指数退避——设计原因避免风暴重试打垮下游 backoff 2 ** attempt logger.info(f重试等待{backoff}秒...) await asyncio.sleep(backoff) self._record_error(ctx) return self._error_response(ctx, f处理失败(已重试{self.MAX_RETRIES}次): {last_error}) async def _do_process(self, data: Any, ctx: RequestContext) - Any: 实际处理逻辑——设计原因在Demo代码外面包装重试/超时/监控 # 阶段一预处理 preprocessed await self._preprocess(data) # 阶段二模型推理——设计原因最关键的一步单独监控耗时 inference_start time.time() model_output await self._model_inference(preprocessed) inference_time time.time() - inference_start logger.info(f推理完成 | {ctx.request_id} | 耗时{inference_time:.2f}s) # 阶段三后处理 result await self._postprocess(model_output) return result async def _preprocess(self, data: Any) - Any: 预处理——设计原因统一的预处理入口各个模型复用 # 实际预处理逻辑 return data async def _model_inference(self, data: Any) - Any: 模型推理——设计原因抽象接口切换模型不改上层 # 实际推理逻辑 return {output: inference_result, confidence: 0.95} async def _postprocess(self, data: Any) - Any: 后处理——设计原因模型输出校验拦截明显异常的结果 return data def _record_success(self, ctx: RequestContext): 记录成功指标——设计原因没有度量就没有优化 self._request_count 1 self._latency_records.append(ctx.elapsed) # 只保留最近1000条的延迟记录——设计原因防止内存无限增长 if len(self._latency_records) 1000: self._latency_records self._latency_records[-500:] def _record_error(self, ctx: RequestContext): 记录错误指标 self._error_count 1 def _success_response(self, ctx: RequestContext, result: Any) - Dict[str, Any]: 成功响应——设计原因统一格式前端不用区分正常/异常 return { status: success, request_id: ctx.request_id, elapsed_ms: round(ctx.elapsed * 1000, 2), data: result } def _error_response(self, ctx: RequestContext, message: str) - Dict[str, Any]: 错误响应——设计原因给用户明确错误信息而非「出错了」三个字 return { status: error, request_id: ctx.request_id, elapsed_ms: round(ctx.elapsed * 1000, 2), error: message } def health_check(self) - Dict[str, Any]: 健康检查——设计原因K8s liveness probe需要15秒检查一次 uptime time.time() - self._start_time error_rate ( self._error_count / max(self._request_count, 1) ) # 计算P50/P95/P99延迟——设计原因平均值没意义长尾决定用户体验 p50 p95 p99 0.0 if self._latency_records: sorted_latency sorted(self._latency_records) n len(sorted_latency) p50 sorted_latency[int(n * 0.5)] p95 sorted_latency[int(n * 0.95)] p99 sorted_latency[int(n * 0.99)] # 内存使用——设计原因OOM前预警提前扩容 memory_mb psutil.Process().memory_info().rss / 1024 / 1024 status ServiceStatus.HEALTHY if error_rate 0.05: status ServiceStatus.DEGRADED if memory_mb 30 * 1024: # 超过30GB status ServiceStatus.OVERLOADED return { status: status.value, uptime_hours: round(uptime / 3600, 1), total_requests: self._request_count, error_rate_pct: round(error_rate * 100, 2), latency_p50_ms: round(p50 * 1000, 2), latency_p95_ms: round(p95 * 1000, 2), latency_p99_ms: round(p99 * 1000, 2), memory_mb: round(memory_mb, 2), concurrent: self.MAX_CONCURRENT - self._semaphore._value } # Web服务入口FastAPI示例模式 async def create_app(): 创建Web应用——设计原因统一入口初始化所有组件 engine ProductionAIEngine() return engine生产代码里最容易被 Demo 阶段忽略的信号量限流——Demo 是单用户单请求永远不会超限流但生产环境100个用户同时上传文件没有限流就是100个 OOM。优雅退出也是——K8s 的默认行为是 SIGTERM 后30秒强制 SIGKILL不捕获 SIGTERM 就等着正在处理的请求全部中断。四、个性化边界权衡容错力度 vs 问题暴露把所有异常都 catch 掉、返回友好错误信息——用户体验好了但问题被掩盖了。需要区分短暂故障网络超时、第三方 API 限流静默处理并重试系统故障OOM、逻辑错误必须报警并记录完整堆栈。内存 vs 延迟输入文件缓存到内存处理快但大文件多了会 OOM。写磁盘安全但在高并发时磁盘 IO 成为瓶颈。折中方案10MB 以下的文件走内存流10MB 以上走磁盘临时文件处理完毕立即删除。降级策略 vs 用户体验模型某个模块挂了是返回错误还是降级输出降级输出如表格解析失败仅提取了文字内容信息不完整但可用对于业务连续性来说优于完全不可用。但需要明确标注哪些字段是降级产出的。五、总结AI 项目从 Demo 到生产的差距涵盖数据规模、异常输入处理、并发与性能、部署与运维四个维度。生产代码需要增加并发信号量限流、输入多层校验、带指数退避的重试、统一异常处理、健康检查端点、优雅退出机制。监控指标应包括 P50/P95/P99 延迟、错误率、内存使用、并发数。实施中需权衡容错力度与问题暴露、内存使用与延迟、降级策略与用户体验的关系。Demo 验证功能可行性生产验证系统鲁棒性两者的工程量差5-10倍是正常范围。