尧图网页设计 
1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年AI应用开发的门槛肉眼可见地降低了随便拉个框架、调个API就能跑出一个能对话的Demo。但我带过不少新人也面试过不少号称“做过AI项目”的候选人发现一个很普遍的问题模型能跑起来但一问到“为什么这么设计”“数据怎么流转”“线上挂了怎么排查”基本就卡壳了。这就是典型的“会调包但不懂工程”。ai-engineering-from-scratch这个标题我理解它想表达的核心诉求是从零开始把AI工程当作一门正经的软件工程来对待而不是把它当成魔法。它适合那些已经会写Python、想进入AI应用开发领域但不想只停留在“调API”层面的开发者也适合那些已经在做AI项目、但总觉得自己的系统“脆得像纸糊的”的工程师。这篇文章我会从整体设计思路、核心模块拆解、实操落地流程、常见坑排查四个维度把“从零搭建AI工程能力”这件事讲透。全文基于我过去几年在推荐系统、对话系统、内容审核等场景的落地经验结合常见工程实践补充细节不涉及任何特定平台或敏感内容。先说清楚一个前提AI工程不等于模型训练。大部分业务场景下你不需要自己训练大模型你需要的是把模型能力稳定、高效、可观测地集成到业务系统里。这中间涉及数据管道、推理服务、缓存策略、降级方案、监控告警、成本控制等一系列工程问题。这些才是“AI工程”的真正内涵也是这篇文章要展开的重点。2. 整体架构设计与技术选型思路2.1 为什么我坚持“分层解耦”而不是“一把梭”很多新手做AI项目习惯写一个巨大的脚本读数据、调模型、处理结果、写数据库全塞在一起。这种写法在Demo阶段没问题但一旦要上线、要迭代、要排查问题就是灾难。我踩过最惨的一次坑是一个内容分类服务模型调用和业务逻辑耦合在一起某天模型接口超时整个服务线程池被占满连带影响了同进程的其他接口。所以我的核心设计原则是分层解耦至少拆成四层接入层负责请求接收、参数校验、鉴权、限流。这一层不碰任何模型逻辑只做流量治理。编排层负责业务逻辑编排决定什么时候调模型、调哪个模型、要不要走缓存、失败了怎么降级。模型层封装模型调用统一输入输出格式屏蔽不同模型提供方的差异。数据层负责特征存储、结果缓存、日志落库、指标上报。这样拆的好处是模型层可以独立替换比如从A模型换到B模型编排层可以独立测试用Mock模型接入层可以独立扩容。每一层的职责边界清晰出问题时能快速定位是哪一层的问题。2.2 模型选型别只看效果先算清楚这三笔账选模型是AI工程里最容易拍脑袋的环节。我见过太多团队一上来就选最大最强的模型结果上线后发现成本扛不住、延迟太高、并发上不去。我的经验是选型前先算三笔账第一笔是成本账。按Token计费的模型你要估算日均请求量、平均输入输出长度算出日成本。举个例子假设日均10万次请求平均输入500 Token、输出200 Token某模型输入价格是每百万Token 10元、输出30元那日成本就是10万 × (500/100万 × 10 200/100万 × 30) 10万 × (0.005 0.006) 1100元。一个月就是3万多这个数字必须提前算清楚。第二笔是延迟账。模型推理延迟直接决定用户体验。一般来说首Token延迟TTFT控制在1秒以内、整体响应控制在3秒以内是比较舒服的。如果模型本身延迟就高你就要考虑流式输出、预加载、缓存等补偿手段。第三笔是稳定性账。模型服务不是100%可用的你要假设它随时可能超时、限流、返回异常。所以选型时要看提供方是否有SLA承诺、是否有备用模型、切换成本高不高。我的建议是主力模型选效果和成本平衡的备用模型选一个更便宜或更快的关键场景做双路调用结果择优。这样既保证效果又留了后路。2.3 数据流转设计把“特征”和“结果”分开管AI工程里数据分两类一类是输入特征比如用户历史行为、文本内容一类是输出结果比如分类标签、生成文本。这两类的生命周期和管理方式完全不同。输入特征往往需要实时拼接可能来自多个数据源对延迟敏感输出结果往往需要缓存复用对一致性要求没那么高。所以我的做法是特征拼接走实时计算用Redis或本地缓存做热点数据加速设置合理的过期时间。结果缓存走键值存储用输入内容的哈希值作为Key缓存模型输出。相同输入直接命中缓存省时省钱。这里有个细节缓存Key的设计要考虑模型版本。如果模型升级了旧缓存必须失效否则会出现“新旧结果混用”的诡异问题。我的做法是在Key里带上模型版本号比如model:v2:hash(input)升级时改版本号即可。3. 核心模块拆解与实操要点3.1 模型调用封装统一接口是稳定性的基石模型调用封装看起来简单其实是最容易埋雷的地方。不同模型提供方的接口格式、错误码、超时行为都不一样如果不做统一封装业务代码里会散落大量if-else。我的封装原则是统一入参、统一出参、统一异常。入参定义成一个标准结构包含prompt、max_tokens、temperature等字段出参定义成标准结构包含text、usage、finish_reason等字段异常统一成几类超时、限流、内容违规、服务不可用。class ModelResponse: def __init__(self, text, usage, finish_reason, model_version): self.text text self.usage usage self.finish_reason finish_reason self.model_version model_version class ModelClient: def __init__(self, timeout10, max_retries2): self.timeout timeout self.max_retries max_retries def call(self, prompt, **kwargs): for attempt in range(self.max_retries 1): try: raw self._do_request(prompt, **kwargs) return self._parse(raw) except TimeoutError: if attempt self.max_retries: raise ModelTimeoutError() except RateLimitError: time.sleep(2 ** attempt) raise ModelUnavailableError()这段代码的关键点是重试策略。我一般设置最多2次重试第一次失败后等1秒第二次失败后等2秒指数退避。但要注意不是所有错误都值得重试。超时和限流可以重试内容违规重试也没用直接返回错误即可。注意重试次数不是越多越好。重试会放大下游压力如果模型服务已经过载大量重试会让情况更糟。我的经验是重试不超过2次且要加随机抖动jitter避免所有请求同时重试。3.2 缓存策略省下的都是真金白银缓存是AI工程里性价比最高的优化手段。我做过统计在很多场景下20%的请求覆盖了80%的输入尤其是客服问答、内容分类这类场景用户问来问去就那些问题。缓存设计有三个关键决策第一缓存什么。我一般缓存完整的模型输出Key用输入内容的规范化哈希。规范化包括去空格、转小写、去除标点差异。这样“今天天气怎么样”和“今天天气怎么样”能命中同一个缓存。第二缓存多久。这取决于业务对时效性的要求。内容分类可以缓存几小时甚至几天实时问答可能只能缓存几分钟。我的做法是设置两级TTL热点数据长TTL冷数据短TTL用LRU淘汰。第三缓存穿透怎么办。如果大量请求查不到缓存全部打到模型会造成雪崩。我的做法是加布隆过滤器或空值缓存查不到的结果也缓存一个短TTL的空标记避免同一个不存在的Key反复穿透。缓存策略适用场景TTL建议注意事项全量缓存输入重复率高1-24小时模型升级需清空热点缓存长尾分布明显热点长、冷点短需LRU淘汰空值缓存防穿透1-5分钟标记要区分“无结果”和“未查询”语义缓存相似问法多视业务而定需向量检索成本较高语义缓存是进阶玩法把输入转成向量相似度超过阈值就复用结果。这个方案效果好但实现复杂建议先把精确缓存做扎实再考虑。3.3 降级与熔断系统挂了也要优雅AI服务最怕的就是“模型挂了整个系统跟着挂”。我经历过一次线上事故模型提供方突发故障所有请求超时我们的服务线程池被占满导致整个应用不可用。从那以后我给所有AI服务都加了降级和熔断。熔断的逻辑是统计最近一段时间的失败率超过阈值就“跳闸”后续请求直接走降级逻辑不再调用模型。过一段时间后放少量请求试探如果成功就恢复。降级的逻辑是模型不可用时返回什么。这取决于业务内容分类返回“未知”或默认分类记录日志后续补偿。对话生成返回预设话术比如“当前咨询人数较多请稍后再试”。搜索排序降级到规则排序或热门排序。class CircuitBreaker: def __init__(self, failure_threshold0.5, window_size100, cooldown60): self.failure_threshold failure_threshold self.window_size window_size self.cooldown cooldown self.failures deque(maxlenwindow_size) self.state CLOSED self.last_failure_time None def allow_request(self): if self.state OPEN: if time.time() - self.last_failure_time self.cooldown: self.state HALF_OPEN return True return False return True def record(self, success): self.failures.append(0 if success else 1) if self.state HALF_OPEN and success: self.state CLOSED elif sum(self.failures) / len(self.failures) self.failure_threshold: self.state OPEN self.last_failure_time time.time()提示熔断阈值不要设得太敏感。我一般设50%失败率、窗口100次请求、冷却60秒。太敏感会导致正常波动就跳闸太迟钝则失去保护意义。3.4 可观测性没有监控的AI系统等于裸奔AI系统的可观测性比普通系统更重要因为它的行为更不确定。我要求所有AI服务必须上报四类指标请求指标QPS、成功率、延迟分布P50/P95/P99。模型指标Token消耗量、缓存命中率、重试次数、熔断次数。业务指标分类准确率抽样、生成内容长度分布、用户反馈率。成本指标按天/按小时统计模型调用成本设置预算告警。日志方面我要求每次模型调用都记录请求ID、输入哈希、模型版本、耗时、Token用量、是否命中缓存、是否降级。这些日志是排查问题的命根子。有个细节容易被忽略输入内容可能包含敏感信息日志里不能明文记录。我的做法是记录输入哈希和长度需要复现时用哈希去缓存里捞原文或者做脱敏处理。4. 完整实操流程从零搭一个可用的AI服务4.1 环境准备与依赖管理假设你要搭一个文本分类服务从零开始的步骤是这样的。首先是环境准备我习惯用Python 3.10依赖管理用poetry或pip-tools保证依赖版本锁定。# 初始化项目 mkdir ai-classifier cd ai-classifier poetry init --name ai-classifier --python ^3.10 poetry add fastapi uvicorn redis httpx pydantic prometheus-client poetry add --dev pytest pytest-asyncio依赖选择说明fastapi做Web框架轻量且异步支持好redis做缓存httpx做异步HTTP客户端调模型prometheus-client做指标上报。这些都是经过大量项目验证的稳定选择。4.2 核心代码结构项目结构我一般这样组织ai-classifier/ ├── app/ │ ├── main.py # 入口FastAPI实例 │ ├── api/ │ │ └── routes.py # 路由定义 │ ├── core/ │ │ ├── config.py # 配置管理 │ │ └── metrics.py # 指标定义 │ ├── services/ │ │ ├── model_client.py # 模型调用封装 │ │ ├── cache.py # 缓存逻辑 │ │ └── classifier.py # 业务编排 │ └── models/ │ └── schemas.py # 请求响应结构 ├── tests/ └── pyproject.toml这个结构的核心思想是按职责分包services目录下每个文件负责一个独立能力互相通过明确接口调用。4.3 关键步骤请求处理全链路一个分类请求的完整处理链路是这样的接入层校验检查请求体格式、文本长度我一般限制在2000字符以内、鉴权Token。计算缓存Key对文本做规范化去空格、转小写计算SHA256哈希。查缓存用哈希去Redis查命中则直接返回记录缓存命中指标。熔断检查如果熔断器打开走降级逻辑返回默认分类。调用模型构造Prompt调用模型接口带超时和重试。解析结果从模型输出中提取分类标签做格式校验。写缓存把结果写入Redis设置TTL。上报指标记录耗时、Token用量、是否降级。返回响应返回标准结构。async def classify(text: str) - dict: normalized normalize(text) cache_key fcls:v1:{hashlib.sha256(normalized.encode()).hexdigest()} cached await cache.get(cache_key) if cached: metrics.cache_hit.inc() return {label: cached, source: cache} if not circuit_breaker.allow_request(): metrics.degraded.inc() return {label: unknown, source: degraded} try: response await model_client.call(build_prompt(text)) label parse_label(response.text) await cache.set(cache_key, label, ttl3600) circuit_breaker.record(True) metrics.model_success.inc() return {label: label, source: model} except Exception as e: circuit_breaker.record(False) metrics.model_failure.inc() logger.error(fmodel call failed: {e}) return {label: unknown, source: error}4.4 参数计算超时和并发怎么定超时时间不是拍脑袋定的。我的方法是先测模型P99延迟超时设为P99的1.5倍。比如模型P99是2秒超时设3秒。这样既能容忍正常波动又不会让慢请求拖垮系统。并发数计算稍微复杂。假设你的服务要支撑100 QPS模型平均延迟1秒那理论上需要100个并发。但实际要考虑峰值我一般按峰值QPS × 平均延迟 × 1.5安全系数来估算。100 QPS × 1秒 × 1.5 150并发。然后根据这个数字配置线程池或连接池大小。注意并发数不是越大越好。超过模型服务承载能力后延迟会急剧上升反而降低吞吐。一定要做压测找到拐点。4.5 上线前的检查清单上线前我必做这几件事压测用locust或wrk模拟峰值流量观察延迟和错误率。故障演练手动断开模型服务验证降级逻辑是否生效。缓存预热把高频输入提前灌入缓存避免上线瞬间穿透。监控看板确认所有指标都能正常上报和展示。告警配置错误率超过5%、P99延迟超过3秒、熔断触发时都要告警。5. 常见问题与排查技巧实录5.1 模型返回格式不稳定怎么办这是最常见的问题。模型有时候返回JSON有时候返回带解释的文字有时候多输出一段废话。我的应对策略是三层防护第一层是Prompt约束明确要求“只输出JSON不要任何解释”。第二层是解析容错用正则提取JSON部分解析失败时尝试修复常见问题比如单引号转双引号。第三层是兜底重试解析失败时用更严格的Prompt重试一次还失败就走降级。def parse_label(text: str) - str: # 尝试直接解析 try: data json.loads(text) return data.get(label, unknown) except json.JSONDecodeError: pass # 尝试提取JSON片段 match re.search(r\{.*?\}, text, re.DOTALL) if match: try: data json.loads(match.group()) return data.get(label, unknown) except json.JSONDecodeError: pass # 尝试关键词匹配 for label in [positive, negative, neutral]: if label in text.lower(): return label return unknown5.2 缓存命中率低怎么优化缓存命中率低先别急着加缓存先分析原因。我一般看三个维度输入分布是不是输入太分散如果是考虑语义缓存。Key设计是不是规范化不够比如大小写、空格、标点差异导致同一语义不同Key。TTL设置是不是太短适当延长TTL能提升命中率但要权衡时效性。我遇到过一个案例缓存命中率只有15%排查发现是用户输入里带了时间戳导致每次Key都不同。后来在规范化时去掉了时间戳命中率直接升到60%。5.3 成本突然飙升怎么排查成本飙升通常有四个原因按排查优先级排列排查项可能原因排查方法请求量流量突增或爬虫看QPS曲线和来源IP分布缓存缓存失效或穿透看命中率曲线检查Redis状态输入长度用户输入变长看输入Token分布重试模型不稳定导致重试看重试次数和失败率我有一次遇到成本翻倍最后发现是缓存Redis实例内存满了大量Key被淘汰导致穿透。解决办法是扩容Redis并优化TTL策略。5.4 模型升级后效果变差怎么办模型升级是个高风险操作。我的做法是灰度发布效果对比新模型先接10%流量同时记录新旧模型的结果。用离线评估集对比新旧模型的准确率、延迟、成本。如果新模型在关键指标上不劣于旧模型逐步扩大流量。全程保留回滚能力一旦发现问题立即切回旧模型。提示模型升级时一定要清空或隔离旧缓存否则会出现新旧结果混用。我一般用版本号隔离新模型用新的缓存前缀。5.5 高频问题速查表问题现象可能原因解决方向延迟突然升高模型服务过载或网络抖动检查模型方状态启用熔断错误率上升模型接口变更或限流查看错误码联系提供方内存持续增长缓存或日志未清理检查TTL和日志轮转结果不一致缓存新旧混用检查缓存版本隔离成本异常重试过多或输入变长检查重试策略和输入分布6. 我踩过的坑和给你的建议做AI工程这几年踩过的坑比写过的代码还多。有几个教训特别深刻分享出来让你少走弯路。第一个坑是过度依赖模型。早期我做一个分类任务把所有希望寄托在模型上结果模型效果不稳定整个系统跟着抖。后来我加了规则兜底高置信度的规则先过滤模型只处理规则覆盖不了的case。这样既降低了成本又提升了稳定性。规则和模型不是对立的而是互补的。第二个坑是忽视冷启动。新服务上线缓存是空的所有请求都打到模型瞬间把模型服务打挂。后来我养成了习惯上线前用历史数据预热缓存或者先小流量灰度让缓存慢慢暖起来。第三个坑是日志记太多。有段时间我把每次模型调用的完整输入输出都记下来结果日志量爆炸存储成本比模型成本还高。后来改成只记哈希和元数据需要复现时再去缓存捞原文。日志要记但要记对地方。第四个坑是没有成本意识。刚开始做AI项目时我只关注效果不关注成本。直到财务找上门才发现一个月烧了几十万。从那以后我给每个AI服务都加了成本监控和预算告警超预算自动降级到便宜模型。最后分享一个实用技巧给模型调用加一个“影子模式”。新模型上线前先让它和旧模型并行跑只记录结果不返回给用户对比一段时间后再决定是否切换。这个模式帮我避免了好几次线上事故。AI工程说到底是一门平衡的艺术效果和成本平衡、延迟和准确率平衡、稳定性和迭代速度平衡。没有银弹只有在具体场景下不断调优。希望这些经验能帮你少踩几个坑把AI系统做得更扎实。 从零搭建AI工程能力：分层解耦、缓存降级与可观测性实战
1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年AI应用开发的门槛肉眼可见地降低了随便拉个框架、调个API就能跑出一个能对话的Demo。但我带过不少新人也面试过不少号称“做过AI项目”的候选人发现一个很普遍的问题模型能跑起来但一问到“为什么这么设计”“数据怎么流转”“线上挂了怎么排查”基本就卡壳了。这就是典型的“会调包但不懂工程”。ai-engineering-from-scratch这个标题我理解它想表达的核心诉求是从零开始把AI工程当作一门正经的软件工程来对待而不是把它当成魔法。它适合那些已经会写Python、想进入AI应用开发领域但不想只停留在“调API”层面的开发者也适合那些已经在做AI项目、但总觉得自己的系统“脆得像纸糊的”的工程师。这篇文章我会从整体设计思路、核心模块拆解、实操落地流程、常见坑排查四个维度把“从零搭建AI工程能力”这件事讲透。全文基于我过去几年在推荐系统、对话系统、内容审核等场景的落地经验结合常见工程实践补充细节不涉及任何特定平台或敏感内容。先说清楚一个前提AI工程不等于模型训练。大部分业务场景下你不需要自己训练大模型你需要的是把模型能力稳定、高效、可观测地集成到业务系统里。这中间涉及数据管道、推理服务、缓存策略、降级方案、监控告警、成本控制等一系列工程问题。这些才是“AI工程”的真正内涵也是这篇文章要展开的重点。2. 整体架构设计与技术选型思路2.1 为什么我坚持“分层解耦”而不是“一把梭”很多新手做AI项目习惯写一个巨大的脚本读数据、调模型、处理结果、写数据库全塞在一起。这种写法在Demo阶段没问题但一旦要上线、要迭代、要排查问题就是灾难。我踩过最惨的一次坑是一个内容分类服务模型调用和业务逻辑耦合在一起某天模型接口超时整个服务线程池被占满连带影响了同进程的其他接口。所以我的核心设计原则是分层解耦至少拆成四层接入层负责请求接收、参数校验、鉴权、限流。这一层不碰任何模型逻辑只做流量治理。编排层负责业务逻辑编排决定什么时候调模型、调哪个模型、要不要走缓存、失败了怎么降级。模型层封装模型调用统一输入输出格式屏蔽不同模型提供方的差异。数据层负责特征存储、结果缓存、日志落库、指标上报。这样拆的好处是模型层可以独立替换比如从A模型换到B模型编排层可以独立测试用Mock模型接入层可以独立扩容。每一层的职责边界清晰出问题时能快速定位是哪一层的问题。2.2 模型选型别只看效果先算清楚这三笔账选模型是AI工程里最容易拍脑袋的环节。我见过太多团队一上来就选最大最强的模型结果上线后发现成本扛不住、延迟太高、并发上不去。我的经验是选型前先算三笔账第一笔是成本账。按Token计费的模型你要估算日均请求量、平均输入输出长度算出日成本。举个例子假设日均10万次请求平均输入500 Token、输出200 Token某模型输入价格是每百万Token 10元、输出30元那日成本就是10万 × (500/100万 × 10 200/100万 × 30) 10万 × (0.005 0.006) 1100元。一个月就是3万多这个数字必须提前算清楚。第二笔是延迟账。模型推理延迟直接决定用户体验。一般来说首Token延迟TTFT控制在1秒以内、整体响应控制在3秒以内是比较舒服的。如果模型本身延迟就高你就要考虑流式输出、预加载、缓存等补偿手段。第三笔是稳定性账。模型服务不是100%可用的你要假设它随时可能超时、限流、返回异常。所以选型时要看提供方是否有SLA承诺、是否有备用模型、切换成本高不高。我的建议是主力模型选效果和成本平衡的备用模型选一个更便宜或更快的关键场景做双路调用结果择优。这样既保证效果又留了后路。2.3 数据流转设计把“特征”和“结果”分开管AI工程里数据分两类一类是输入特征比如用户历史行为、文本内容一类是输出结果比如分类标签、生成文本。这两类的生命周期和管理方式完全不同。输入特征往往需要实时拼接可能来自多个数据源对延迟敏感输出结果往往需要缓存复用对一致性要求没那么高。所以我的做法是特征拼接走实时计算用Redis或本地缓存做热点数据加速设置合理的过期时间。结果缓存走键值存储用输入内容的哈希值作为Key缓存模型输出。相同输入直接命中缓存省时省钱。这里有个细节缓存Key的设计要考虑模型版本。如果模型升级了旧缓存必须失效否则会出现“新旧结果混用”的诡异问题。我的做法是在Key里带上模型版本号比如model:v2:hash(input)升级时改版本号即可。3. 核心模块拆解与实操要点3.1 模型调用封装统一接口是稳定性的基石模型调用封装看起来简单其实是最容易埋雷的地方。不同模型提供方的接口格式、错误码、超时行为都不一样如果不做统一封装业务代码里会散落大量if-else。我的封装原则是统一入参、统一出参、统一异常。入参定义成一个标准结构包含prompt、max_tokens、temperature等字段出参定义成标准结构包含text、usage、finish_reason等字段异常统一成几类超时、限流、内容违规、服务不可用。class ModelResponse: def __init__(self, text, usage, finish_reason, model_version): self.text text self.usage usage self.finish_reason finish_reason self.model_version model_version class ModelClient: def __init__(self, timeout10, max_retries2): self.timeout timeout self.max_retries max_retries def call(self, prompt, **kwargs): for attempt in range(self.max_retries 1): try: raw self._do_request(prompt, **kwargs) return self._parse(raw) except TimeoutError: if attempt self.max_retries: raise ModelTimeoutError() except RateLimitError: time.sleep(2 ** attempt) raise ModelUnavailableError()这段代码的关键点是重试策略。我一般设置最多2次重试第一次失败后等1秒第二次失败后等2秒指数退避。但要注意不是所有错误都值得重试。超时和限流可以重试内容违规重试也没用直接返回错误即可。注意重试次数不是越多越好。重试会放大下游压力如果模型服务已经过载大量重试会让情况更糟。我的经验是重试不超过2次且要加随机抖动jitter避免所有请求同时重试。3.2 缓存策略省下的都是真金白银缓存是AI工程里性价比最高的优化手段。我做过统计在很多场景下20%的请求覆盖了80%的输入尤其是客服问答、内容分类这类场景用户问来问去就那些问题。缓存设计有三个关键决策第一缓存什么。我一般缓存完整的模型输出Key用输入内容的规范化哈希。规范化包括去空格、转小写、去除标点差异。这样“今天天气怎么样”和“今天天气怎么样”能命中同一个缓存。第二缓存多久。这取决于业务对时效性的要求。内容分类可以缓存几小时甚至几天实时问答可能只能缓存几分钟。我的做法是设置两级TTL热点数据长TTL冷数据短TTL用LRU淘汰。第三缓存穿透怎么办。如果大量请求查不到缓存全部打到模型会造成雪崩。我的做法是加布隆过滤器或空值缓存查不到的结果也缓存一个短TTL的空标记避免同一个不存在的Key反复穿透。缓存策略适用场景TTL建议注意事项全量缓存输入重复率高1-24小时模型升级需清空热点缓存长尾分布明显热点长、冷点短需LRU淘汰空值缓存防穿透1-5分钟标记要区分“无结果”和“未查询”语义缓存相似问法多视业务而定需向量检索成本较高语义缓存是进阶玩法把输入转成向量相似度超过阈值就复用结果。这个方案效果好但实现复杂建议先把精确缓存做扎实再考虑。3.3 降级与熔断系统挂了也要优雅AI服务最怕的就是“模型挂了整个系统跟着挂”。我经历过一次线上事故模型提供方突发故障所有请求超时我们的服务线程池被占满导致整个应用不可用。从那以后我给所有AI服务都加了降级和熔断。熔断的逻辑是统计最近一段时间的失败率超过阈值就“跳闸”后续请求直接走降级逻辑不再调用模型。过一段时间后放少量请求试探如果成功就恢复。降级的逻辑是模型不可用时返回什么。这取决于业务内容分类返回“未知”或默认分类记录日志后续补偿。对话生成返回预设话术比如“当前咨询人数较多请稍后再试”。搜索排序降级到规则排序或热门排序。class CircuitBreaker: def __init__(self, failure_threshold0.5, window_size100, cooldown60): self.failure_threshold failure_threshold self.window_size window_size self.cooldown cooldown self.failures deque(maxlenwindow_size) self.state CLOSED self.last_failure_time None def allow_request(self): if self.state OPEN: if time.time() - self.last_failure_time self.cooldown: self.state HALF_OPEN return True return False return True def record(self, success): self.failures.append(0 if success else 1) if self.state HALF_OPEN and success: self.state CLOSED elif sum(self.failures) / len(self.failures) self.failure_threshold: self.state OPEN self.last_failure_time time.time()提示熔断阈值不要设得太敏感。我一般设50%失败率、窗口100次请求、冷却60秒。太敏感会导致正常波动就跳闸太迟钝则失去保护意义。3.4 可观测性没有监控的AI系统等于裸奔AI系统的可观测性比普通系统更重要因为它的行为更不确定。我要求所有AI服务必须上报四类指标请求指标QPS、成功率、延迟分布P50/P95/P99。模型指标Token消耗量、缓存命中率、重试次数、熔断次数。业务指标分类准确率抽样、生成内容长度分布、用户反馈率。成本指标按天/按小时统计模型调用成本设置预算告警。日志方面我要求每次模型调用都记录请求ID、输入哈希、模型版本、耗时、Token用量、是否命中缓存、是否降级。这些日志是排查问题的命根子。有个细节容易被忽略输入内容可能包含敏感信息日志里不能明文记录。我的做法是记录输入哈希和长度需要复现时用哈希去缓存里捞原文或者做脱敏处理。4. 完整实操流程从零搭一个可用的AI服务4.1 环境准备与依赖管理假设你要搭一个文本分类服务从零开始的步骤是这样的。首先是环境准备我习惯用Python 3.10依赖管理用poetry或pip-tools保证依赖版本锁定。# 初始化项目 mkdir ai-classifier cd ai-classifier poetry init --name ai-classifier --python ^3.10 poetry add fastapi uvicorn redis httpx pydantic prometheus-client poetry add --dev pytest pytest-asyncio依赖选择说明fastapi做Web框架轻量且异步支持好redis做缓存httpx做异步HTTP客户端调模型prometheus-client做指标上报。这些都是经过大量项目验证的稳定选择。4.2 核心代码结构项目结构我一般这样组织ai-classifier/ ├── app/ │ ├── main.py # 入口FastAPI实例 │ ├── api/ │ │ └── routes.py # 路由定义 │ ├── core/ │ │ ├── config.py # 配置管理 │ │ └── metrics.py # 指标定义 │ ├── services/ │ │ ├── model_client.py # 模型调用封装 │ │ ├── cache.py # 缓存逻辑 │ │ └── classifier.py # 业务编排 │ └── models/ │ └── schemas.py # 请求响应结构 ├── tests/ └── pyproject.toml这个结构的核心思想是按职责分包services目录下每个文件负责一个独立能力互相通过明确接口调用。4.3 关键步骤请求处理全链路一个分类请求的完整处理链路是这样的接入层校验检查请求体格式、文本长度我一般限制在2000字符以内、鉴权Token。计算缓存Key对文本做规范化去空格、转小写计算SHA256哈希。查缓存用哈希去Redis查命中则直接返回记录缓存命中指标。熔断检查如果熔断器打开走降级逻辑返回默认分类。调用模型构造Prompt调用模型接口带超时和重试。解析结果从模型输出中提取分类标签做格式校验。写缓存把结果写入Redis设置TTL。上报指标记录耗时、Token用量、是否降级。返回响应返回标准结构。async def classify(text: str) - dict: normalized normalize(text) cache_key fcls:v1:{hashlib.sha256(normalized.encode()).hexdigest()} cached await cache.get(cache_key) if cached: metrics.cache_hit.inc() return {label: cached, source: cache} if not circuit_breaker.allow_request(): metrics.degraded.inc() return {label: unknown, source: degraded} try: response await model_client.call(build_prompt(text)) label parse_label(response.text) await cache.set(cache_key, label, ttl3600) circuit_breaker.record(True) metrics.model_success.inc() return {label: label, source: model} except Exception as e: circuit_breaker.record(False) metrics.model_failure.inc() logger.error(fmodel call failed: {e}) return {label: unknown, source: error}4.4 参数计算超时和并发怎么定超时时间不是拍脑袋定的。我的方法是先测模型P99延迟超时设为P99的1.5倍。比如模型P99是2秒超时设3秒。这样既能容忍正常波动又不会让慢请求拖垮系统。并发数计算稍微复杂。假设你的服务要支撑100 QPS模型平均延迟1秒那理论上需要100个并发。但实际要考虑峰值我一般按峰值QPS × 平均延迟 × 1.5安全系数来估算。100 QPS × 1秒 × 1.5 150并发。然后根据这个数字配置线程池或连接池大小。注意并发数不是越大越好。超过模型服务承载能力后延迟会急剧上升反而降低吞吐。一定要做压测找到拐点。4.5 上线前的检查清单上线前我必做这几件事压测用locust或wrk模拟峰值流量观察延迟和错误率。故障演练手动断开模型服务验证降级逻辑是否生效。缓存预热把高频输入提前灌入缓存避免上线瞬间穿透。监控看板确认所有指标都能正常上报和展示。告警配置错误率超过5%、P99延迟超过3秒、熔断触发时都要告警。5. 常见问题与排查技巧实录5.1 模型返回格式不稳定怎么办这是最常见的问题。模型有时候返回JSON有时候返回带解释的文字有时候多输出一段废话。我的应对策略是三层防护第一层是Prompt约束明确要求“只输出JSON不要任何解释”。第二层是解析容错用正则提取JSON部分解析失败时尝试修复常见问题比如单引号转双引号。第三层是兜底重试解析失败时用更严格的Prompt重试一次还失败就走降级。def parse_label(text: str) - str: # 尝试直接解析 try: data json.loads(text) return data.get(label, unknown) except json.JSONDecodeError: pass # 尝试提取JSON片段 match re.search(r\{.*?\}, text, re.DOTALL) if match: try: data json.loads(match.group()) return data.get(label, unknown) except json.JSONDecodeError: pass # 尝试关键词匹配 for label in [positive, negative, neutral]: if label in text.lower(): return label return unknown5.2 缓存命中率低怎么优化缓存命中率低先别急着加缓存先分析原因。我一般看三个维度输入分布是不是输入太分散如果是考虑语义缓存。Key设计是不是规范化不够比如大小写、空格、标点差异导致同一语义不同Key。TTL设置是不是太短适当延长TTL能提升命中率但要权衡时效性。我遇到过一个案例缓存命中率只有15%排查发现是用户输入里带了时间戳导致每次Key都不同。后来在规范化时去掉了时间戳命中率直接升到60%。5.3 成本突然飙升怎么排查成本飙升通常有四个原因按排查优先级排列排查项可能原因排查方法请求量流量突增或爬虫看QPS曲线和来源IP分布缓存缓存失效或穿透看命中率曲线检查Redis状态输入长度用户输入变长看输入Token分布重试模型不稳定导致重试看重试次数和失败率我有一次遇到成本翻倍最后发现是缓存Redis实例内存满了大量Key被淘汰导致穿透。解决办法是扩容Redis并优化TTL策略。5.4 模型升级后效果变差怎么办模型升级是个高风险操作。我的做法是灰度发布效果对比新模型先接10%流量同时记录新旧模型的结果。用离线评估集对比新旧模型的准确率、延迟、成本。如果新模型在关键指标上不劣于旧模型逐步扩大流量。全程保留回滚能力一旦发现问题立即切回旧模型。提示模型升级时一定要清空或隔离旧缓存否则会出现新旧结果混用。我一般用版本号隔离新模型用新的缓存前缀。5.5 高频问题速查表问题现象可能原因解决方向延迟突然升高模型服务过载或网络抖动检查模型方状态启用熔断错误率上升模型接口变更或限流查看错误码联系提供方内存持续增长缓存或日志未清理检查TTL和日志轮转结果不一致缓存新旧混用检查缓存版本隔离成本异常重试过多或输入变长检查重试策略和输入分布6. 我踩过的坑和给你的建议做AI工程这几年踩过的坑比写过的代码还多。有几个教训特别深刻分享出来让你少走弯路。第一个坑是过度依赖模型。早期我做一个分类任务把所有希望寄托在模型上结果模型效果不稳定整个系统跟着抖。后来我加了规则兜底高置信度的规则先过滤模型只处理规则覆盖不了的case。这样既降低了成本又提升了稳定性。规则和模型不是对立的而是互补的。第二个坑是忽视冷启动。新服务上线缓存是空的所有请求都打到模型瞬间把模型服务打挂。后来我养成了习惯上线前用历史数据预热缓存或者先小流量灰度让缓存慢慢暖起来。第三个坑是日志记太多。有段时间我把每次模型调用的完整输入输出都记下来结果日志量爆炸存储成本比模型成本还高。后来改成只记哈希和元数据需要复现时再去缓存捞原文。日志要记但要记对地方。第四个坑是没有成本意识。刚开始做AI项目时我只关注效果不关注成本。直到财务找上门才发现一个月烧了几十万。从那以后我给每个AI服务都加了成本监控和预算告警超预算自动降级到便宜模型。最后分享一个实用技巧给模型调用加一个“影子模式”。新模型上线前先让它和旧模型并行跑只记录结果不返回给用户对比一段时间后再决定是否切换。这个模式帮我避免了好几次线上事故。AI工程说到底是一门平衡的艺术效果和成本平衡、延迟和准确率平衡、稳定性和迭代速度平衡。没有银弹只有在具体场景下不断调优。希望这些经验能帮你少踩几个坑把AI系统做得更扎实。