1. 为什么我要从零手搓一套AI工程栈
第一次看到ai-engineering-from-scratch这个标题,我脑子里蹦出来的不是“又一个教程仓库”,而是过去两年带团队踩过的那些坑。市面上讲AI的课程和文章多如牛毛,但绝大多数要么停留在调包层面——import openai然后写个聊天框就完事,要么直接跳到分布式训练和CUDA优化,中间那一大段“工程化落地”的空白地带几乎没人认真讲。而恰恰是这段空白,决定了你做的AI功能到底是能跑个demo,还是能扛住真实流量。
ai-engineering-from-scratch这个项目标题,核心诉求非常明确:从零开始,把AI工程化涉及的每一个环节都亲手实现一遍。它不是教你训练一个大模型,而是教你围绕模型构建一套完整的工程体系——数据管道、推理服务、缓存策略、评测体系、成本控制、可观测性。这套东西做出来,你才算真正具备了“AI工程师”的能力,而不是“调API工程师”。
我见过太多团队,模型选型很先进,prompt写得也漂亮,但一上线就崩:响应延迟忽高忽低、token成本失控、输出质量无法量化、出了问题不知道从哪查。这些问题的根源都不在模型本身,而在工程化能力缺失。所以这个项目的价值在于,它逼着你把每个环节都拆开、揉碎、自己动手搭一遍,搭完之后你对整个链路的理解会完全不一样。
这篇文章适合三类人:一是想从传统后端转AI工程方向的开发者,你需要补的不是算法课,而是这套工程思维;二是已经在做AI产品但总觉得“哪里不对劲”的工程师,你需要系统性地梳理一遍链路;三是技术负责人,你需要知道一个AI项目从零到一到底要经历哪些环节,才能合理排期和评估风险。我会按照实际搭建的顺序,把每个模块的设计思路、关键细节、实操步骤和踩坑经验都讲清楚,你可以直接照着复现。
2. 整体架构设计与技术选型思路
2.1 从需求倒推架构:先想清楚要解决什么问题
动手写代码之前,我习惯先把需求场景想透。ai-engineering-from-scratch这个项目要解决的核心问题是:让一个AI功能从本地脚本变成可对外服务的系统。这个过程中,最容易被低估的是“非功能性需求”——延迟、吞吐、成本、稳定性、可观测性。很多人一上来就纠结用哪个模型,其实模型只是其中一个可替换的组件,真正决定系统能不能用的,是围绕它的工程架构。
我的设计原则是“分层解耦,逐层可替换”。具体来说,把整个系统拆成五层:接入层负责请求路由和限流,编排层负责prompt组装和上下文管理,推理层负责模型调用和结果解析,缓存层负责降低重复计算,观测层负责全链路追踪和指标采集。每一层之间通过明确定义的接口通信,这样你换模型、换缓存方案、换监控工具,都不会牵一发动全身。
为什么这么设计?因为AI工程领域变化太快了。今天用这个模型,明天可能就换了;今天用Redis做缓存,明天可能换成语义缓存。如果架构是铁板一块,每次变更都是灾难。分层解耦的代价是前期多写一些接口定义和适配器代码,但收益是后期迭代速度提升一个数量级。我试过在一个没有分层的项目里换模型,改了三十多个文件,漏了一个导致线上事故;后来重构成分层架构,换模型只改了一个适配器文件,十分钟搞定。
2.2 技术栈选型:为什么是这些工具
技术选型这块,我的原则是“成熟优先,避免炫技”。AI工程不是搞科研,稳定性和可维护性比技术先进性重要得多。下面是我实际使用的技术栈和选型理由:
| 层级 | 选型 | 核心理由 | 替代方案 |
|---|---|---|---|
| 接入层 | FastAPI | 异步原生支持,Pydantic校验强,自动生成文档 | Flask(同步阻塞,不适合高并发) |
| 编排层 | 自研轻量框架 | 需要精细控制prompt模板和上下文窗口 | LangChain(抽象过重,调试困难) |
| 推理层 | 多provider适配器 | 避免供应商锁定,支持降级切换 | 直接调用单一SDK(风险集中) |
| 缓存层 | Redis + 向量索引 | 精确缓存和语义缓存结合 | 纯内存缓存(重启丢失,容量有限) |
| 观测层 | OpenTelemetry + Prometheus | 标准协议,生态完善 | 自研日志系统(重复造轮子) |
关于编排层,我要特别说一下为什么没有直接用LangChain。LangChain确实功能强大,但它的抽象层次太多,一个简单的prompt组装要经过好几层封装,出了问题很难定位。而且它的版本迭代快,API经常变,维护成本高。我选择自研一个轻量框架,核心就是三个类:PromptTemplate负责模板管理和变量填充,ContextManager负责上下文窗口的裁剪和压缩,Chain负责把多个步骤串起来。总共不到五百行代码,但完全可控,调试起来一目了然。
推理层的多provider适配器设计也值得展开说。我定义了一个LLMProvider抽象基类,包含generate、stream_generate、count_tokens三个核心方法。然后为每个模型供应商实现一个子类。这样做的好处是,业务代码只依赖抽象接口,切换供应商只需要改配置。更重要的是,可以实现自动降级:主供应商超时或报错时,自动切换到备用供应商,保证服务可用性。这个机制在一次供应商大面积故障时救了我们的命,主供应商挂了之后自动切到备用,用户几乎无感知。
2.3 数据流设计:一次请求的完整旅程
理解数据流是理解整个系统的关键。我拿一个典型的AI问答请求来举例,走一遍完整链路:
- 客户端发送请求到接入层,接入层做参数校验和限流检查
- 请求进入编排层,
ContextManager从会话存储中加载历史对话,根据token预算裁剪上下文 PromptTemplate把用户输入、历史对话、系统指令组装成最终prompt- 编排层先查缓存层,计算prompt的哈希值和语义向量,尝试命中精确缓存或语义缓存
- 缓存未命中,请求进入推理层,
LLMProvider调用模型API,支持流式返回 - 推理层解析模型输出,提取结构化数据,同时上报token消耗和延迟指标
- 结果写入缓存层,更新会话存储,返回给客户端
- 观测层在整个链路中埋点,记录每个环节的耗时和状态
这个数据流设计的关键在于缓存前置和观测贯穿。缓存前置意味着在调用模型之前就尝试命中缓存,能省下大量token成本。观测贯穿意味着每个环节都有埋点,出了问题能快速定位是哪个环节的瓶颈。我在实际运行中发现,缓存命中率能达到30%左右,对于重复问题较多的场景,这个比例还能更高,直接省下三成成本。
3. 核心模块拆解与关键实现细节
3.1 接入层:不只是路由,更是第一道防线
很多人觉得接入层就是写几个路由函数,没什么技术含量。但在我踩过的坑里,接入层出问题导致的事故占比相当高。接入层要做的远不止路由,它还是整个系统的第一道防线。
请求校验是基础但容易被忽视的环节。我用Pydantic定义严格的请求模型,对输入长度、格式、必填字段做校验。为什么要严格?因为AI模型的输入成本是按token算的,如果有人恶意发送超长文本,一次请求就能烧掉你几块钱。我在接入层设置了单次请求的最大token限制,超过直接拒绝,不进入后续流程。这个限制根据业务场景设定,一般对话场景设4000 token,文档处理场景可以放宽到32000 token。
限流策略我采用的是令牌桶算法,按用户维度和全局维度双重限流。用户维度防止单个用户刷爆系统,全局维度保护后端模型API不被打挂。令牌桶的参数需要根据实际压测结果调整,我一开始设得太宽松,结果一次营销活动涌入大量请求,模型API直接限流,所有用户都受影响。后来改成动态限流,根据后端响应时间自动调整令牌发放速率,效果好了很多。
超时控制是另一个关键点。AI请求的延迟波动很大,同一个prompt,有时2秒返回,有时20秒。如果不设超时,慢请求会堆积,拖垮整个服务。我在接入层设置了三级超时:连接超时5秒,首token超时10秒,总超时60秒。超过首token超时还没开始返回,直接断开并触发降级逻辑。这个参数需要根据模型和场景调整,流式场景可以放宽首token超时,非流式场景可以缩短总超时。
# 接入层核心配置示例 class RequestConfig: max_input_tokens: int = 4000 connect_timeout: float = 5.0 first_token_timeout: float = 10.0 total_timeout: float = 60.0 rate_limit_per_user: int = 20 # 每分钟请求数 rate_limit_global: int = 500 # 每分钟全局请求数注意:超时参数不要拍脑袋定,一定要用真实流量压测。我见过有人把首token超时设成3秒,结果正常请求大量被误杀,用户体验极差。
3.2 编排层:prompt工程的工程化
编排层是AI工程中最具“AI特色”的部分,也是传统后端工程师最容易懵的地方。它的核心任务是把用户输入、历史上下文、系统指令组装成模型能理解的prompt,同时管理上下文窗口不超限。
Prompt模板管理我采用“模板+变量”的方式,模板存在数据库或配置文件中,支持热更新。为什么要热更新?因为prompt是需要不断迭代优化的,如果每次改prompt都要发版,迭代速度太慢。我设计了一个简单的模板版本管理机制,每个模板有版本号,可以灰度发布,对比不同版本的效果。实测下来,一个好的prompt模板能把输出质量提升30%以上,而且迭代成本很低。
上下文窗口管理是编排层最复杂的部分。模型的上下文窗口是有限的,比如8K、32K、128K token。当历史对话很长时,必须做裁剪或压缩。我实现了三种策略:滑动窗口(保留最近N轮对话)、摘要压缩(把早期对话用模型总结成一段话)、关键信息提取(只保留实体和意图)。选择哪种策略取决于场景:客服对话适合滑动窗口,长文档问答适合摘要压缩,任务型对话适合关键信息提取。
class ContextManager: def __init__(self, max_tokens: int, strategy: str = "sliding_window"): self.max_tokens = max_tokens self.strategy = strategy def build_context(self, history: list, current_input: str) -> list: if self.strategy == "sliding_window": return self._sliding_window(history, current_input) elif self.strategy == "summarize": return self._summarize(history, current_input) elif self.strategy == "extract": return self._extract_key_info(history, current_input) def _sliding_window(self, history, current_input): # 从最近往前累加,直到接近token上限 selected = [] token_count = self._count_tokens(current_input) for msg in reversed(history): msg_tokens = self._count_tokens(msg["content"]) if token_count + msg_tokens > self.max_tokens * 0.9: break selected.insert(0, msg) token_count += msg_tokens return selected + [{"role": "user", "content": current_input}]提示:上下文窗口不要用满,留10%的余量给模型输出。我试过用满窗口,结果模型没有空间生成回复,直接返回空内容。
3.3 推理层:多provider适配与降级策略
推理层是直接和模型API打交道的地方,也是最容易出故障的环节。模型API可能超时、限流、返回格式错误、甚至服务不可用。如果推理层没有做好容错,整个系统就跟着挂。
多provider适配器的设计前面提过,这里展开讲实现细节。抽象基类定义三个核心方法,每个provider子类实现具体逻辑。关键是要统一输入输出格式,让上层业务代码不感知底层差异。比如有的模型返回choices[0].message.content,有的返回output.text,适配器要统一成{"content": "...", "usage": {...}}的格式。
降级策略我实现了三级:第一级是重试,同一个provider重试2次,间隔1秒和3秒;第二级是切换,重试失败后切换到备用provider;第三级是兜底,所有provider都失败时返回预设的兜底话术。降级触发条件包括:超时、HTTP错误码、返回内容为空、返回内容格式错误。这套机制在一次主provider大面积故障时发挥了关键作用,自动切换后用户几乎无感知。
Token计数是推理层另一个重要功能。不同模型的token计算方式不同,有的按字符,有的按词,有的用BPE。我封装了一个统一的count_tokens方法,内部根据provider类型调用对应的计数逻辑。准确的token计数对于成本控制和上下文管理都至关重要。我踩过的坑是:一开始用粗略估算(字符数除以4),结果实际消耗比预估高了40%,成本失控。后来接入准确的tokenizer,误差控制在5%以内。
| 降级级别 | 触发条件 | 处理动作 | 恢复策略 |
|---|---|---|---|
| 一级重试 | 超时/临时错误 | 同provider重试2次 | 自动恢复 |
| 二级切换 | 重试仍失败 | 切换备用provider | 定时探测主provider |
| 三级兜底 | 所有provider失败 | 返回兜底话术 | 人工介入 |
3.4 缓存层:省下的都是真金白银
缓存层是AI工程中ROI最高的模块,没有之一。模型调用是按token收费的,缓存命中一次就省一次钱。而且缓存还能显著降低延迟,提升用户体验。我实测下来,缓存命中率30%时,整体成本降低约25%,平均延迟降低约40%。
精确缓存是最基础的,用prompt的哈希值作为key,模型输出作为value。实现简单,但命中率有限,因为用户换个说法,哈希值就变了。我用的哈希算法是SHA256,key的构成包括:prompt内容、模型名称、温度参数、最大token数。这些参数任何一个不同,输出就可能不同,所以都要纳入key的计算。
语义缓存是进阶方案,用embedding模型把prompt转成向量,在向量数据库中查找相似度超过阈值的缓存项。这样即使用户换了说法,只要语义相同,就能命中缓存。阈值一般设0.92到0.95之间,太低会命中不相关的结果,太高命中率又上不去。我试过0.90的阈值,结果把“如何退款”和“如何退货”匹配到了一起,虽然都是售后问题,但答案完全不同,导致用户困惑。后来调到0.94,效果好很多。
class SemanticCache: def __init__(self, redis_client, embedding_model, threshold=0.94): self.redis = redis_client self.embedding_model = embedding_model self.threshold = threshold def get(self, prompt: str): # 先查精确缓存 exact_key = hashlib.sha256(prompt.encode()).hexdigest() exact_result = self.redis.get(f"exact:{exact_key}") if exact_result: return json.loads(exact_result) # 再查语义缓存 prompt_vector = self.embedding_model.encode(prompt) candidates = self.redis.ft("idx:cache").search( query_vector=prompt_vector, k=1 ) if candidates and candidates[0].score >= self.threshold: return json.loads(candidates[0].value) return None注意:语义缓存的阈值需要根据业务场景调优。事实型问答可以设高一些(0.95+),创意型生成可以设低一些(0.90+),因为创意型对精确度要求没那么高。
3.5 观测层:出了问题能查到根因
观测层是很多AI项目最容易忽略的模块,但它是系统稳定运行的保障。没有观测,出了问题就是盲人摸象,只能靠猜。我搭建的观测体系包括三个维度:指标(Metrics)、日志(Logs)、追踪(Traces)。
指标方面,我采集的核心指标有:请求量、成功率、延迟分布(P50/P95/P99)、token消耗量、缓存命中率、各provider调用占比、降级触发次数。这些指标用Prometheus采集,Grafana展示。我设置了几条关键告警:成功率低于95%告警、P99延迟超过10秒告警、缓存命中率低于20%告警、降级触发超过10次/分钟告警。
日志方面,我采用结构化日志,每条日志包含:请求ID、用户ID、prompt哈希、模型名称、输入token数、输出token数、延迟、状态、错误信息。日志用JSON格式,方便后续检索和分析。关键是要把请求ID贯穿整个链路,这样排查问题时能串起所有相关日志。
追踪方面,我用OpenTelemetry做全链路追踪,每个请求生成一个trace,包含多个span:接入层span、编排层span、缓存查询span、模型调用span、结果解析span。这样一眼就能看出时间花在哪个环节。我遇到过一次延迟飙升的问题,通过追踪发现是缓存查询环节慢了,原因是向量索引没有建好,全表扫描导致。如果没有追踪,这个问题很难定位。
4. 完整实操流程:从零搭建到上线运行
4.1 环境准备与项目初始化
动手之前,先把环境准备好。我用的Python版本是3.11,这个版本在异步性能和类型支持上比较平衡。依赖管理用Poetry,比pip+requirements.txt更清晰,能锁定依赖版本,避免“在我机器上能跑”的问题。
# 项目初始化 mkdir ai-engineering-from-scratch && cd ai-engineering-from-scratch poetry init --name ai-engineering-from-scratch --python "^3.11" poetry add fastapi uvicorn redis pydantic httpx opentelemetry-api opentelemetry-sdk prometheus-client poetry add --group dev pytest pytest-asyncio ruff mypy项目目录结构我按分层架构组织,每个层一个目录,层内按功能模块拆分文件。这样结构清晰,新人接手能快速理解代码组织。
ai-engineering-from-scratch/ ├── app/ │ ├── gateway/ # 接入层 │ │ ├── routes.py │ │ ├── middleware.py │ │ └── schemas.py │ ├── orchestration/ # 编排层 │ │ ├── prompt.py │ │ ├── context.py │ │ └── chain.py │ ├── inference/ # 推理层 │ │ ├── base.py │ │ ├── providers/ │ │ └── fallback.py │ ├── cache/ # 缓存层 │ │ ├── exact.py │ │ └── semantic.py │ └── observability/ # 观测层 │ ├── metrics.py │ ├── logging.py │ └── tracing.py ├── config/ │ └── settings.yaml ├── tests/ └── pyproject.toml环境变量管理我用pydantic-settings,所有配置项集中在一个Settings类中,支持从环境变量和配置文件读取。敏感信息如API密钥只从环境变量读取,不写入配置文件,避免泄露。
4.2 推理层适配器实现
推理层是第一个要实现的模块,因为其他层都依赖它。我先定义抽象基类,然后实现两个provider作为示例:一个通用HTTP provider,一个本地模型provider。
from abc import ABC, abstractmethod from typing import AsyncGenerator class LLMProvider(ABC): @abstractmethod async def generate(self, prompt: str, **kwargs) -> dict: """非流式生成,返回统一格式的结果""" pass @abstractmethod async def stream_generate(self, prompt: str, **kwargs) -> AsyncGenerator[str, None]: """流式生成,逐token返回""" pass @abstractmethod def count_tokens(self, text: str) -> int: """计算文本的token数""" pass class HTTPProvider(LLMProvider): def __init__(self, base_url: str, api_key: str, model: str): self.base_url = base_url self.api_key = api_key self.model = model self.client = httpx.AsyncClient(timeout=60.0) async def generate(self, prompt: str, **kwargs) -> dict: response = await self.client.post( f"{self.base_url}/v1/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model": self.model, "messages": [{"role": "user", "content": prompt}], "temperature": kwargs.get("temperature", 0.7), "max_tokens": kwargs.get("max_tokens", 2048), } ) response.raise_for_status() data = response.json() return { "content": data["choices"][0]["message"]["content"], "usage": { "input_tokens": data["usage"]["prompt_tokens"], "output_tokens": data["usage"]["completion_tokens"], } }实现完provider后,我写了一个简单的测试脚本,验证基本功能。测试内容包括:正常生成、流式生成、token计数、超时处理、错误处理。这一步不能省,我见过太多人跳过单元测试,结果集成时才发现基础功能有问题,排查成本高得多。
4.3 缓存层搭建与调优
缓存层依赖Redis,我用Redis Stack,因为它自带向量搜索功能,省得再部署一个向量数据库。Redis的安装我用Docker,一条命令搞定。
docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 redis/redis-stack:latest精确缓存的实现比较直接,关键是key的设计和过期时间的设置。过期时间我设的是24小时,因为大部分问答场景的时效性要求没那么高。对于时效性强的场景(如新闻问答),可以缩短到1小时。
语义缓存的实现需要embedding模型。我用的是一个轻量级的本地embedding模型,推理速度快,不依赖外部API。向量索引的创建需要注意维度和距离度量方式,我用的是余弦相似度。
# 创建向量索引 redis_client.execute_command( "FT.CREATE", "idx:cache", "ON", "HASH", "SCHEMA", "vector", "VECTOR", "FLAT", "6", "TYPE", "FLOAT32", "DIM", "384", "DISTANCE_METRIC", "COSINE", "prompt", "TEXT", "response", "TEXT" )缓存调优是个持续过程。我每周会分析缓存命中日志,看看哪些类型的请求命中率低,然后针对性优化。比如发现某类问题的表述方式特别多样,就调整语义缓存的阈值,或者增加同义表述的预处理。
4.4 观测层集成与告警配置
观测层的集成要趁早,不要等到出问题才加。我在项目初期就把OpenTelemetry和Prometheus集成进去,每个模块都埋点。这样从第一天起就有数据可看,能及时发现潜在问题。
from opentelemetry import trace from prometheus_client import Counter, Histogram # 定义指标 REQUEST_COUNT = Counter("ai_requests_total", "Total requests", ["status", "provider"]) REQUEST_LATENCY = Histogram("ai_request_latency_seconds", "Request latency", ["stage"]) CACHE_HIT = Counter("ai_cache_hits_total", "Cache hits", ["cache_type"]) TOKEN_USAGE = Counter("ai_token_usage_total", "Token usage", ["type", "provider"]) # 在关键位置埋点 async def generate_with_observability(prompt: str): tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("generate") as span: start = time.time() try: result = await provider.generate(prompt) REQUEST_COUNT.labels(status="success", provider=provider.name).inc() TOKEN_USAGE.labels(type="input", provider=provider.name).inc(result["usage"]["input_tokens"]) TOKEN_USAGE.labels(type="output", provider=provider.name).inc(result["usage"]["output_tokens"]) return result except Exception as e: REQUEST_COUNT.labels(status="error", provider=provider.name).inc() span.record_exception(e) raise finally: REQUEST_LATENCY.labels(stage="total").observe(time.time() - start)告警配置我用Prometheus的Alertmanager,规则写在YAML文件里。关键告警包括:错误率超过5%持续2分钟、P99延迟超过15秒持续5分钟、缓存命中率低于15%持续10分钟、token消耗速率超过预算阈值。告警通知我接的是企业微信机器人,方便团队及时响应。
4.5 压测与性能调优实录
上线前必须压测,这是铁律。我用Locust做压测,模拟不同并发下的表现。压测场景设计要贴近真实:混合长短请求、混合缓存命中和未命中、模拟provider超时。
第一轮压测结果很不理想:50并发时P99延迟就到了12秒,错误率3%。排查发现瓶颈在推理层,每个请求都新建HTTP连接,连接建立的开销很大。改成连接池后,P99降到6秒。第二轮压测发现缓存层是瓶颈,语义缓存的向量搜索耗时较长。优化方案是给向量索引加缓存,热门查询的向量结果缓存在内存中,P99进一步降到3秒。第三轮压测达到目标:100并发下P99延迟4秒,错误率0.5%。
| 压测轮次 | 并发数 | P99延迟 | 错误率 | 主要瓶颈 | 优化措施 |
|---|---|---|---|---|---|
| 第一轮 | 50 | 12s | 3% | HTTP连接建立 | 引入连接池 |
| 第二轮 | 80 | 6s | 1.2% | 向量搜索 | 向量结果内存缓存 |
| 第三轮 | 100 | 4s | 0.5% | 无明显瓶颈 | 参数微调 |
提示:压测时一定要监控各层指标,不要只看总体延迟。我第一轮压测只看了总体P99,排查了半天才发现是连接池的问题。后来把每层的延迟都打出来,一眼就能定位瓶颈。
5. 常见问题与排查技巧实录
5.1 模型输出不稳定怎么排查
模型输出不稳定是最高频的问题,表现为同一个prompt有时返回好结果,有时返回差结果。排查思路是分层定位:先确认是不是温度参数的问题,温度越高输出越随机,生产环境建议设0.3到0.7之间;再确认是不是上下文窗口的问题,上下文太长会导致模型“遗忘”早期信息;最后确认是不是prompt本身有歧义,换几种表述测试。
我遇到过一次输出不稳定的问题,排查后发现是上下文管理器的滑动窗口策略有问题:当历史对话刚好在窗口边界时,有时包含某条关键信息,有时不包含,导致输出差异。改成摘要压缩策略后,关键信息始终保留,输出稳定了很多。
5.2 缓存命中率低的优化路径
缓存命中率低,先分析原因。我用了一个简单的分析方法:抽样1000条未命中的请求,人工看它们的prompt有什么特点。发现三类问题:一是用户表述太口语化,和缓存中的标准表述差异大;二是多轮对话中,每轮的prompt都包含完整历史,导致哈希值不同;三是温度参数不同导致key不同。
针对性优化:对于口语化问题,在缓存查询前加一层query改写,把口语化表述转成标准表述;对于多轮对话,缓存key只用当前轮的用户输入,不含历史;对于温度参数,缓存key不包含温度,因为温度只影响生成随机性,不影响语义。优化后命中率从18%提升到35%。
5.3 成本失控的紧急止血方案
成本失控通常有几个信号:token消耗速率突然上升、缓存命中率下降、单请求平均token数增加。发现信号后,紧急止血方案分三步:第一步,在接入层临时降低单请求最大token限制,砍掉超长请求;第二步,提高缓存阈值,让更多请求命中缓存;第三步,切换到更便宜的模型provider作为默认,把贵模型作为备用。
长期控制方案是建立成本预算和告警机制。我给每个用户设了日token预算,超过预算后降级到便宜模型或直接拒绝。同时监控单位请求的平均成本,超过阈值就告警。这套机制运行半年,成本波动控制在10%以内。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 延迟突然飙升 | provider限流/网络抖动 | 查看各provider延迟指标 | 切换provider/增加重试 |
| 输出内容为空 | 上下文窗口用满 | 检查输入token数 | 裁剪上下文/增加窗口 |
| 缓存命中率骤降 | 缓存服务异常/索引损坏 | 检查Redis状态 | 重启缓存/重建索引 |
| token消耗异常 | prompt模板变更/上下文膨胀 | 对比历史token数据 | 回滚模板/优化上下文 |
| 错误率上升 | provider故障/代码bug | 查看错误日志和追踪 | 降级/修复代码 |
5.5 独家避坑经验分享
第一个坑:不要用生产环境的API密钥做压测。我试过一次,压测流量把生产配额用完了,导致真实用户请求全部失败。后来严格区分环境,压测用独立的密钥和配额。
第二个坑:prompt模板变更一定要灰度。我直接全量更新了一个prompt模板,结果新模板在某些边缘case下输出格式错误,导致下游解析失败。后来改成灰度发布,先放10%流量,观察24小时无问题再全量。
第三个坑:缓存过期时间不要设太长。我一开始设了7天,结果模型更新后,缓存里还是旧模型的输出,用户拿到的答案质量下降。后来改成24小时,平衡了成本和时效性。
第四个坑:观测数据要定期清理。Prometheus的指标数据如果不清理,几个月就占满磁盘。我设置了数据保留策略,原始指标保留15天,聚合指标保留90天。
6. 后续扩展方向与个人体会
这套系统搭完之后,我最大的体会是:AI工程的核心不是AI,是工程。模型能力固然重要,但决定产品体验的往往是工程细节——缓存策略、降级机制、上下文管理、成本控制。这些细节做不好,再好的模型也白搭。
后续可以扩展的方向有几个。一是多模态支持,把图像、音频的输入输出纳入统一框架,这需要扩展推理层的适配器和缓存层的key设计。二是A/B测试框架,支持同时运行多个prompt版本或模型版本,用数据驱动决策。三是自动化评测,用模型自己评测输出质量,减少人工评估成本。四是边缘部署,把轻量模型部署到离用户更近的地方,进一步降低延迟。
我个人在实际操作中的体会是,这套系统最有价值的部分不是代码本身,而是搭建过程中形成的工程思维。你会开始习惯性地思考:这个环节会不会成为瓶颈?出故障了怎么降级?成本怎么控制?质量怎么量化?这些问题意识,比任何具体技术都重要。如果你也在做AI工程化落地,建议不要跳过任何一个环节,亲手搭一遍,踩一遍坑,收获会远超你的预期。