01 一个凌晨三点的不眠夜凌晨两点四十七分报警群炸了。“RAG服务超时率飙升到45%”“用户端大量反馈’AI答非所问’” “向量数据库连接池耗尽” “Token消耗量是平日的8倍”——运维同事在群里连续甩出四条报警每一个都指向同一个根因我们引以为傲的AI问答系统正在以肉眼可见的速度崩溃。我睡眼惺忪地打开电脑登录服务器开始了一场漫长的破案之旅。首先检查的是Python AI引擎的日志发现大量的超时错误。打开监控面板CPU使用率正常、内存正常、网络IO正常一切都显示正常——但系统就是不正常。这是所有AI工程师最怕的夜晚所有指标都显示正常但业务就是不可用。我花了四个小时才找到真相凌晨两点某业务团队通过脚本批量上传了3.2万份文档触发了向量库重索引重建。Milvus在重建索引期间查询性能急剧下降大量请求排队等待。与此同时某位实习生未做限流配置在大模型API调用中使用了串行请求导致单次会话阻塞长达47秒把整个服务的线程池拖垮了。修复方案并不复杂将批量索引任务迁移到离线集群在线集群采用双Buffer索引切换策略引入信号量限流单个会话的并发请求数上限设为3对所有外部API调用增加超时熔断。凌晨六点四十七分系统恢复平稳报警群安静下来。那个夜晚让我深刻地明白了一个道理在这个行业里能让AI跑起来的人很多但能让AI一直跑下去的人很少。02 为什么AI代码写完才是一切麻烦的开始在传统的软件开发中写代码是核心工作。写完、测试、部署稳定性问题大多是已知模式的可预测故障。但在AI应用开发中写代码只占了工作量的20%剩下的80%都花在怎么让它稳定跑起来上。挑战一不确定性输出带来的稳定性质疑普通后端服务的输出是可预测的输入11永远返回2。但大模型不一样同样的输入今天和明天的输出可能完全不同。# 同样的输入可能产生不同的输出print(llm.invoke(请介绍一下我们的退款政策))# 第一次输出尊敬的客户退款政策如下...# 第二次输出根据公司规定退款需要...# 第三次输出很抱歉我无法找到相关信息...这种不确定性在Demo阶段是惊喜在生产环境就是灾难。解决方案输出结构化与校验frompydanticimportBaseModel,ValidationErrorfromtypingimportOptional,LiteralimportjsonimportreclassRefundResponse(BaseModel):强制模型输出结构化数据确保字段完整is_eligible:boolreason:strrefund_amount:Optional[float]Noneestimated_days:intrequired_documents:list[str]next_action:Literal[在线提交,联系客服,前往门店]confidence_score:float# 模型自评置信度defsafe_llm_invoke(prompt:str,retry:int3)-RefundResponse: 带重试和校验的LLM调用 确保输出始终符合预期结构 forattemptinrange(retry):try:raw_responsellm.invoke(prompt)# 尝试解析JSON# 如果模型返回的不是合法JSON用正则提取json_matchre.search(r\{.*\},raw_response,re.DOTALL)ifjson_match:datajson.loads(json_match.group())else:# 模型可能纯文本回复手动解析关键信息dataextract_info_from_text(raw_response)# Pydantic会自动校验类型、枚举值、必填字段returnRefundResponse(**data)except(json.JSONDecodeError,ValidationError)ase:ifattemptretry-1:# 最后一次重试失败返回安全降级响应returnRefundResponse(is_eligibleFalse,reason系统暂时无法判断建议联系人工客服,estimated_days0,required_documents[],next_action联系客服,confidence_score0.0)# 增加重试提示到prompt中引导模型修正promptf\n\n上次输出格式不符合要求错误{e}。请严格按照JSON格式输出。continue挑战二流式响应中的断流问题AI应用普遍采用SSEServer-Sent Events流式响应来提升用户体验。但流式传输充满了陷阱网络抖动导致连接中断用户看到半截回答后端服务重启正在进行的流式会话全部失效前端关闭页面但后端还在继续生成浪费算力importasynciofromcontextlibimportasynccontextmanagerfromtypingimportAsyncGeneratorimportuuidclassResilientStreamManager: 弹性流式响应管理器 特性断点续传、超时保护、优雅关闭 def__init__(self,timeout_seconds:int120):self.timeouttimeout_seconds self.active_streams{}# stream_id - 状态asynccontextmanagerasyncdefstream_with_heartbeat(self,stream_id:str)-AsyncGenerator: 带心跳保活的流式响应 每30秒发送一个: heartbeat事件防止Nginx超时断开 self.active_streams[stream_id]{status:active,started_at:asyncio.get_event_loop().time()}try:# 使用超时包装器asyncdefgenerator_with_timeout():try:asyncwithasyncio.timeout(self.timeout):yieldexceptasyncio.TimeoutError:self.active_streams[stream_id][status]timeoutyielddata: [TIMEOUT] 请求处理超时请重试\n\nyieldgenerator_with_timeout()finally:# 清理资源self.active_streams.pop(stream_id,None)asyncdefping_heartbeat(self,stream_id:str):每隔30秒发送心跳保持连接活跃whilestream_idinself.active_streams:awaitasyncio.sleep(30)# 发送SSE心跳yield: heartbeat\n\n挑战三大模型API的不可靠依赖你的服务再稳定只要依赖的大模型API不稳定一切归零。我经历过OpenAI API 大规模宕机持续3小时某国产大模型API突然修改了输出格式没发任何通知网络抖动导致API超时但请求已经被扣费解决方案多模型降级与熔断fromtypingimportList,DictimporttimefromcollectionsimportdequeclassModelRouter: 多模型路由与熔断器 核心策略基于成功率动态切换实现自动降级 def__init__(self):self.models{primary:{endpoint:https://api.openai.com/v1/chat/completions,weight:10,# 优先级权重success_rate:1.0,latency_p99:0.8,is_healthy:True,error_window:deque(maxlen100)# 最近100次调用的错误率},fallback_1:{endpoint:https://api.azure.com/openai/deployments/...,weight:5,success_rate:0.98,latency_p99:1.2,is_healthy:True,error_window:deque(maxlen100)},fallback_2:{endpoint:https://dashscope.aliyuncs.com/api/v1/...,weight:2,success_rate:0.95,latency_p99:1.5,is_healthy:True,error_window:deque(maxlen100)}}self.circuit_breakers{}# 熔断器状态defselect_model(self)-str: 基于健康度和延迟选择最优模型 熔断逻辑如果错误率超过阈值暂时屏蔽该模型 healthy_models[(name,config)forname,configinself.models.items()ifconfig[is_healthy]andself._is_circuit_closed(name)]ifnothealthy_models:# 所有模型都熔断强制使用主模型风险自担returnprimary# 按延迟排序选最快的healthy_models.sort(keylambdax:x[1][latency_p99])returnhealthy_models[0][0]defrecord_error(self,model_name:str,error_type:str): 记录错误触发熔断条件判断 ifmodel_namenotinself.models:returnwindowself.models[model_name][error_window]window.append({error:error_type,time:time.time()})# 计算最近60秒的错误率recent[eforeinwindowiftime.time()-e[time]60]error_ratelen([eforeinrecentife[error]!timeout])/len(recent)ifrecentelse0timeout_ratelen([eforeinrecentife[error]timeout])/len(recent)ifrecentelse0# 熔断条件错误率30% 或 超时率50%iferror_rate0.3ortimeout_rate0.5:self._open_circuit(model_name,duration120)# 熔断120秒def_is_circuit_closed(self,model_name:str)-bool:检查熔断器是否闭合允许调用ifmodel_namenotinself.circuit_breakers:returnTrueopened_atself.circuit_breakers[model_name][opened_at]durationself.circuit_breakers[model_name][duration]returntime.time()-opened_atdurationdef_open_circuit(self,model_name:str,duration:int):打开熔断器self.circuit_breakers[model_name]{opened_at:time.time(),duration:duration}print(f[熔断] 模型{model_name}已熔断持续{duration}秒)asyncdefinvoke_with_fallback(self,prompt:str,max_attempts:int3)-str: 带降级的调用依次尝试可用模型 tried_models[]forattemptinrange(max_attempts):# 选择当前最优模型model_nameself.select_model()tried_models.append(model_name)try:starttime.time()resultawaitself._call_model(model_name,prompt)latencytime.time()-start# 记录成功self.models[model_name][latency_p99]0.9*self.models[model_name][latency_p99]0.1*latencyreturnresultexceptExceptionase:# 记录失败error_typetimeoutiftimeoutinstr(e).lower()elseotherself.record_error(model_name,error_type)# 如果还有模型没试继续remaining[namefornameinself.modelsifnamenotintried_models]ifremaining:continueelse:# 所有模型都失败了抛出最后的异常raiseRuntimeError(f所有模型调用失败已尝试:{tried_models})fromeraiseRuntimeError(f达到最大重试次数{max_attempts})03 稳定性不是加监控加出来的很多人对稳定的理解停留在加监控、加告警、加日志。但真正的稳定性是在设计阶段就被注入灵魂的。原则一防呆设计不要假设用户会正确使用也不要假设模型会正确输出。classInputSanitizer:输入清洗器——防御Prompt注入攻击INJECTION_PATTERNS[r忽略.*指令,rignore.*instruction,r之前.*内容,rprevious.*context,r系统提示词,rsystem prompt,r输出.*全部.*内容,r你是一个.*模型,]staticmethoddefsanitize(user_input:str)-str:过滤可疑的注入尝试importre# 长度限制iflen(user_input)2000:user_inputuser_input[:2000]# 检测注入模式forpatterninInputSanitizer.INJECTION_PATTERNS:ifre.search(pattern,user_input,re.IGNORECASE):# 发现注入尝试截断输入只保留安全部分user_inputre.sub(pattern,[已过滤],user_input,flagsre.IGNORECASE)# 移除不可见Unicode字符容易隐藏攻击代码user_inputre.sub(r[\u200b-\u200f\u2028-\u202f],,user_input)returnuser_input原则二优雅降级当AI无法正常工作时用户不应该看到500 Internal Server Error。classGracefulDegradation:优雅降级当AI服务不可用时返回用户可理解的信息staticmethoddeffallback_response(error_type:str,user_context:dict)-dict: 根据错误类型返回不同的降级响应 fallbacks{timeout:{reply:AI助手当前响应较慢已为您记录问题稍后将通过站内信回复您。,action:create_ticket,ticket_priority:P1},rate_limit:{reply:当前咨询量较大请稍后再试。如需紧急帮助请点击右下角转人工客服。,action:redirect_human},content_filter:{reply:您的问题涉及敏感内容系统无法处理。请联系人工客服处理。,action:human_only},unknown:{reply:系统遇到了一点小问题我们正在修复。您可以刷新页面重试或稍后再来。,action:retry}}templatefallbacks.get(error_type,fallbacks[unknown])return{reply:template[reply],action:template[action],timestamp:time.time(),session_persist:True# 保存会话状态方便后续跟进}04 写在最后我们找的不是代码工人是系统医生一个只关注写代码的工程师会在凌晨两点被报警吵醒后盯着日志发呆半小时。而一个懂得让AI稳定跑起来的工程师会在设计阶段就想到如果向量库重建索引怎么办如果大模型API挂了怎么办如果用户恶意注入怎么办如果流量突增10倍怎么办如果上下文超过窗口限制怎么办这些问题每一个都能写一篇论文。而我们需要的是——能在实战中解决这些问题的人。我们正在建设一个企业级AI智能体平台每天承载数十万次AI推理请求。代码写了超过20万行踩过的坑比走过的路还多。我们不缺会写CRUD的人我们缺的是能在高并发、高可用、高复杂度的场景下让AI系统持续稳定输出的工程师。如果你写代码之余习惯性思考如果这里挂了怎么办看到监控图比看到产品需求更兴奋对大模型API的各种异常码如数家珍能在十分钟内从报警群定位到根因坚信稳定不是测试出来的是设计出来的那么我们等你。简历投递recruitai-engineering.com标题请注明【AI稳定性方向】 姓名我们不看你用了多少先进技术我们看你在凌晨三点能救回多少请求。