企业官网建设流程全解析

1. 从零搭建AI工程能力：为什么我劝你别急着调包

这两年AI应用开发的门槛肉眼可见地降低了，随便拉个框架、调个API就能跑出一个能对话的Demo。但我见过太多团队，Demo阶段惊艳四座，一上生产环境就原形毕露——推理延迟飙到十几秒、显存说爆就爆、模型更新一次整个服务抖三抖。问题的根子不在模型本身，而在于AI工程能力的缺失。

ai-engineering-from-scratch这个项目标题，说的就是从零开始把AI工程这套东西搭起来。它不是教你训一个更大的模型，也不是教你调某个闭源接口，而是把AI系统落地过程中那些脏活累活——数据处理、推理服务、性能压测、监控告警、版本管理——一层一层剥开讲清楚。适合谁看？我认为有三类人最该认真读：一是刚转行做AI应用、只会写Prompt不会写服务的开发者；二是在团队里负责把算法模型推上线的工程同学；三是技术负责人，需要判断一套AI系统到底能不能扛住真实流量。

我自己带过几个从零到一的AI项目，踩过的坑足够写一本小册子。这篇文章就把ai-engineering-from-scratch这个主题拆透，从整体设计思路到核心环节实现，再到常见问题排查，尽量给出一套可以直接抄作业的方案。全文不聊虚的，只讲能落地的工程细节。

2. 整体设计与思路拆解

2.1 为什么"从零"反而比"调包"更靠谱

很多人一上来就问：现在框架这么多，为什么还要从零搭？直接用现成的不香吗？我的回答是：框架解决的是通用问题，而AI工程的核心矛盾往往出在你的业务特有场景上。举个真实例子，我们做过一个文档问答系统，用现成的RAG框架跑Demo没问题，但上线后发现两个致命问题：一是长文档切分后检索召回率断崖式下跌，二是并发上来后向量检索的延迟波动极大。这两个问题框架文档里根本没提，因为它们是业务数据分布和部署环境共同决定的。

从零搭建的意义，不是让你重复造轮子，而是让你清楚每个环节的边界和代价。你知道数据怎么流、显存怎么分、请求怎么排队，出问题时才能定位到具体那一层。这就像开车，你可以不会修发动机，但至少得知道仪表盘上哪个灯亮了代表什么。

2.2 分层架构：把AI系统拆成五块积木

我习惯把一套AI工程系统拆成五层，从下往上分别是：

• 数据层：负责原始数据的采集、清洗、切分、向量化，以及向量库的构建和更新。
• 模型层：负责模型的加载、推理、批处理、量化，以及多模型的路由调度。
• 服务层：负责API网关、请求队列、限流熔断、超时重试。
• 观测层：负责日志、指标、链路追踪，尤其是推理延迟和显存占用的监控。
• 迭代层：负责模型版本管理、A/B测试、灰度发布、回滚机制。

这五层不是拍脑袋分的，而是对应了AI系统从"能跑"到"跑得稳"再到"跑得久"的三个阶段。很多团队只做了数据层和模型层，服务层和观测层基本空白，结果就是线上出问题只能靠重启大法。

2.3 技术选型的取舍逻辑

选型这块我踩过最大的坑就是过早追求"先进"。比如向量库，一开始就上了分布式方案，结果数据量才几十万条，运维复杂度却翻了三倍。后来我总结了一个原则：按数据规模和并发量选型，而不是按技术热度选型。

具体来说，数据量在百万级以下、并发在几十QPS以内，单机向量库加内存索引完全够用；到了千万级再考虑分片；上亿级别才需要认真设计分布式架构。模型推理也是同理，小模型用CPU加ONNX Runtime就能跑，没必要上来就上GPU集群。这个判断逻辑后面在实操部分会展开讲。

3. 核心细节解析与实操要点

3.1 数据层：切分策略决定检索上限

数据层最容易被低估的环节就是文本切分。很多人直接用固定长度切，比如每500字一段，结果把一句话切成两半，检索出来的片段语义不完整，模型回答自然驴唇不对马嘴。

我的做法是按语义边界切分，再按长度兜底。具体步骤：

1. 先按段落、标题、列表项等自然边界切分，保留结构信息。
2. 对超长段落，按句子边界二次切分，优先在句号、问号、分号处断开。
3. 设置重叠窗口，一般取切分长度的10%到20%，保证跨段语义连续。
4. 每个片段附带元数据：来源文档、章节标题、位置索引。

这里有个细节：重叠窗口不是越大越好。我实测过，重叠超过30%后，检索结果重复率明显上升，反而挤占了有效片段的召回空间。10%到20%是个比较稳的区间。

注意：切分后的片段一定要做去重和长度过滤，太短的片段（比如少于50字）往往是噪声，会拉低检索质量。

3.2 模型层：批处理与量化的平衡点

模型推理的性能瓶颈通常不在计算本身，而在显存带宽和批处理效率。我见过一个服务，单条请求推理要800毫秒，改成批处理32条后，单条平均延迟降到120毫秒，吞吐量翻了六倍多。

批处理的核心参数是最大批大小和最大等待时间。这两个参数是一对矛盾：批越大吞吐越高，但单条延迟也越高；等待时间越长越容易凑大批，但用户感知的延迟也越大。我的经验值是：

量化方面，INT8量化通常能带来2到3倍的推理加速，精度损失在1%以内，性价比很高。但要注意，量化后的模型对输入长度更敏感，长文本推理时精度下降会更明显。我的做法是：短文本场景用INT8，长文本场景用FP16，极端性能要求下才考虑INT4。

3.3 服务层：限流和熔断是保命符

服务层最关键的三个机制是限流、熔断、超时。这三个东西平时看不出价值，一旦流量突增或模型异常，就是保命符。

限流我推荐用令牌桶算法，因为它允许一定程度的突发流量。配置上，令牌生成速率按平均QPS设置，桶容量设为平均QPS的2到3倍。这样正常流量下不会误杀，突发流量下也能平滑削峰。

熔断的触发条件要结合业务定。我的经验是：连续10次请求中失败超过5次，或者平均延迟超过阈值3倍，就触发熔断。熔断后进入半开状态，放少量请求试探，成功则恢复，失败则继续熔断。

超时设置有个坑：不要只设一个总超时。我一般设三层：连接超时2秒、首字节超时5秒、总超时30秒。这样能区分是网络问题、模型加载问题还是推理本身慢。

3.4 观测层：没有监控的AI系统等于裸奔

观测层我重点讲推理延迟的分解监控。一个请求的总延迟可以拆成：排队时间、预处理时间、推理时间、后处理时间。这四段分别监控，出问题时才能快速定位。

我用的方案是Prometheus加Grafana，核心指标包括：

• request_queue_duration：请求在队列中的等待时间
• preprocess_duration：文本清洗、向量化等预处理耗时
• inference_duration：模型推理耗时
• postprocess_duration：结果组装、格式化耗时
• gpu_memory_used：显存占用
• batch_size_actual：实际批大小

这些指标按分钟粒度聚合，设置告警阈值。比如推理时间P99超过500毫秒就告警，显存占用超过80%就告警。

提示：日志里一定要记录请求ID和对应的批大小，这样排查问题时能把慢请求和批处理参数关联起来。

4. 实操过程与核心环节实现

4.1 环境准备与依赖安装

先说明一下，这套方案基于Python生态，推理框架用ONNX Runtime，向量库用FAISS，服务框架用FastAPI。选ONNX Runtime是因为它跨平台、部署简单、对量化支持好；选FAISS是因为单机性能足够强，百万级数据毫无压力。

环境准备步骤：

# 创建虚拟环境 python -m venv ai-env source ai-env/bin/activate # 安装核心依赖 pip install onnxruntime-gpu==1.16.0 pip install faiss-gpu==1.7.4 pip install fastapi==0.104.0 pip install uvicorn==0.24.0 pip install transformers==4.35.0 pip install prometheus-client==0.18.0

这里有个细节：ONNX Runtime的GPU版本要和CUDA版本匹配。我用的CUDA 11.8，对应onnxruntime-gpu 1.16.0。版本不匹配会直接报错，而且报错信息很不友好，建议先查官方兼容性表。

4.2 数据管道搭建

数据管道负责把原始文档变成向量库。核心代码如下：

import faiss import numpy as np from sentence_transformers import SentenceTransformer class DataPipeline: def __init__(self, model_name='BAAI/bge-small-zh-v1.5'): self.encoder = SentenceTransformer(model_name) self.index = None self.chunks = [] def split_text(self, text, max_len=500, overlap=100): """按语义边界切分文本""" paragraphs = text.split('\n\n') chunks = [] for para in paragraphs: if len(para) <= max_len: chunks.append(para) else: # 按句子切分 sentences = para.replace('。', '。\n').split('\n') current = '' for sent in sentences: if len(current) + len(sent) <= max_len: current += sent else: if current: chunks.append(current) current = sent if current: chunks.append(current) # 添加重叠 overlapped = [] for i, chunk in enumerate(chunks): if i > 0: prev_tail = chunks[i-1][-overlap:] overlapped.append(prev_tail + chunk) else: overlapped.append(chunk) return overlapped def build_index(self, documents): """构建向量索引""" all_chunks = [] for doc in documents: all_chunks.extend(self.split_text(doc)) self.chunks = all_chunks # 向量化 embeddings = self.encoder.encode( all_chunks, batch_size=32, show_progress_bar=True, normalize_embeddings=True ) # 构建FAISS索引 dim = embeddings.shape[1] self.index = faiss.IndexFlatIP(dim) # 内积索引，配合归一化向量等价于余弦相似度 self.index.add(embeddings.astype('float32')) return self.index

这段代码有几个关键点：向量归一化配合内积索引，等价于余弦相似度，比直接算余弦快很多；批大小设为32，兼顾显存和速度；重叠窗口取100字，对应500字切分长度的20%。

4.3 推理服务实现

推理服务的核心是批处理调度。我用一个异步队列加后台批处理线程实现：

import asyncio import onnxruntime as ort from queue import Queue from threading import Thread import numpy as np class InferenceServer: def __init__(self, model_path, max_batch=32, max_wait=0.2): self.session = ort.InferenceSession( model_path, providers=['CUDAExecutionProvider', 'CPUExecutionProvider'] ) self.max_batch = max_batch self.max_wait = max_wait self.queue = Queue() self.results = {} self._start_worker() def _start_worker(self): def worker(): while True: batch = [] batch_ids = [] # 收集批次 while len(batch) < self.max_batch: try: item = self.queue.get(timeout=self.max_wait) batch.append(item['input']) batch_ids.append(item['id']) except: break if not batch: continue # 批推理 inputs = np.stack(batch) outputs = self.session.run(None, {'input': inputs})[0] # 分发结果 for bid, out in zip(batch_ids, outputs): self.results[bid] = out t = Thread(target=worker, daemon=True) t.start() async def infer(self, input_data): req_id = id(input_data) self.queue.put({'id': req_id, 'input': input_data}) # 等待结果 while req_id not in self.results: await asyncio.sleep(0.01) return self.results.pop(req_id)

这个实现里，max_wait设为0.2秒对应文档问答场景。如果是实时对话，改成0.05秒。注意self.results这个字典在高并发下会有竞争问题，生产环境建议用线程安全的队列或者Redis。

4.4 监控埋点接入

监控埋点用Prometheus客户端库，在关键路径上打点：

from prometheus_client import Histogram, Gauge, Counter import time # 定义指标 queue_duration = Histogram('request_queue_duration_seconds', 'Queue wait time') inference_duration = Histogram('inference_duration_seconds', 'Inference time') gpu_memory = Gauge('gpu_memory_used_bytes', 'GPU memory usage') request_count = Counter('request_total', 'Total requests') class MonitoredInference: def __init__(self, server): self.server = server async def infer(self, input_data): request_count.inc() start = time.time() # 记录排队时间 queue_start = time.time() result = await self.server.infer(input_data) queue_duration.observe(time.time() - queue_start) # 记录推理时间 inference_duration.observe(time.time() - start) return result

这里有个实操心得：Histogram的桶设置要贴合实际延迟分布。默认桶是0.005到10秒，但AI推理延迟通常在0.1到2秒之间，默认桶太粗。我一般自定义桶为[0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0]，这样P99计算更准确。

5. 常见问题与排查技巧实录

5.1 推理延迟突然飙升怎么查

这是最常见的问题。我的排查顺序是：先看队列，再看批大小，最后看显存。

第一步，看request_queue_duration指标。如果排队时间涨了，说明请求积压，要么是并发太高，要么是推理变慢导致消费不过来。第二步，看batch_size_actual。如果批大小突然变小，说明请求间隔变长，可能是上游调用变慢。第三步，看gpu_memory_used。如果显存接近上限，GPU会开始频繁换页，延迟自然飙升。

我遇到过一次诡异的情况：延迟飙升但所有指标都正常。最后发现是向量检索的索引文件被操作系统缓存淘汰了，每次检索都要重新读盘。解决办法是把索引文件放在tmpfs内存盘上，或者用mlock锁定内存。

5.2 检索结果不相关怎么调

检索质量差通常有三个原因：切分不合理、向量模型不匹配、检索参数不当。

切分问题前面讲过了，重点说向量模型。中文场景我推荐BGE系列，英文场景推荐E5系列。但要注意，向量模型必须和检索任务匹配。比如你做的是问答检索，就要用问答对训练的模型，而不是通用语义模型。

检索参数方面，top_k不是越大越好。我实测下来，top_k从5增加到20，召回率提升不到5%，但延迟增加了一倍多。一般设5到10就够了，配合重排序模型效果更好。

5.3 显存溢出怎么预防

显存溢出基本是三个原因：批太大、序列太长、模型太多。

预防措施我总结了一个检查清单：

• 启动时先跑一次最大批大小的推理，确认显存够用
• 设置显存上限，超过就拒绝新请求而不是硬撑
• 长序列场景单独走一个队列，避免和短序列混批
• 多模型场景用懒加载，不用的时候释放显存

注意：ONNX Runtime的显存管理比较粗放，不会主动释放。如果模型频繁切换，建议用独立的Session，用完就销毁。

5.4 常见问题速查表

5.5 几个我踩过的坑

第一个坑：用默认参数跑生产。ONNX Runtime默认用CPU，忘了指定CUDA，结果推理慢十倍。一定要显式指定providers=['CUDAExecutionProvider']。

第二个坑：向量归一化忘了做。用内积索引但不归一化，检索结果完全乱套。归一化就一行代码，但忘了就是灾难。

第三个坑：监控指标没设告警。有次显存泄漏，跑了三天才被发现，期间服务一直降级运行。后来我加了显存告警，超过80%就发通知。

第四个坑：版本管理混乱。模型更新后没记录版本，出问题回滚都不知道回滚到哪个版本。现在我用模型文件的哈希值作为版本号，每次更新都记录。

6. 迭代层：让系统能持续进化

6.1 模型版本管理与灰度发布

迭代层是很多团队忽略的一层，但它是系统能否长期运行的关键。模型版本管理我推荐用文件哈希加元数据的方式，每个模型文件对应一个JSON元数据，记录训练数据、评估指标、上线时间。

灰度发布的策略我一般分三步：先放1%流量观察一天，再放10%观察三天，最后全量。观察指标包括推理延迟、错误率、业务指标（比如问答的满意度）。任何一项异常就自动回滚。

6.2 A/B测试的工程实现

A/B测试的关键是流量分割要稳定。同一个用户每次请求都应该落到同一个版本，否则体验会割裂。实现上用用户ID哈希取模，而不是随机数。

def route_version(user_id, versions): """根据用户ID稳定路由到某个版本""" hash_val = hash(user_id) % 100 cumulative = 0 for version, ratio in versions.items(): cumulative += ratio if hash_val < cumulative: return version return list(versions.keys())[-1]

这个函数保证同一用户始终路由到同一版本，同时整体流量比例符合配置。

6.3 反馈闭环的搭建

AI系统和其他系统最大的区别是需要持续迭代。我建议在服务层加一个反馈收集接口，用户可以对回答点赞或点踩。这些反馈数据定期回流到数据层，用于优化检索策略或微调模型。

反馈数据的处理要注意去偏。点赞多的回答不一定质量高，可能只是位置靠前。我一般会做位置校正，把位置因素从反馈中剥离出来。

7. 一些个人体会

这套从零搭建的方案，我前后迭代了三个版本。第一版只做了数据层和模型层，上线两周就崩了；第二版加了服务层和观测层，稳定运行了半年；第三版补上迭代层，才算真正能持续运转。

最大的体会是：AI工程的难点从来不在AI本身，而在工程。模型可以换、框架可以换，但数据怎么流、请求怎么调度、故障怎么定位，这些底层能力是换不掉的。把这几层搭扎实了，上面用什么模型都是锦上添花。

另外一个小技巧：压测一定要用真实数据。我用合成数据压测时一切正常，换成真实数据后检索延迟翻了三倍，因为真实数据的向量分布更分散，索引效率更低。压测数据从生产环境采样，脱敏后使用，这样测出来的结果才有参考价值。

最后说个扩展方向：这套架构可以平滑迁移到多模态场景。图片、音频的向量化和文本类似，只是编码器不同。服务层的批处理和调度逻辑基本不用改，观测层的指标加几个模态相关的维度就行。我最近在做一个图文混合检索的项目，就是在这套架构上直接扩展的，改动量不到20%。