企业官网建设流程全解析

1. 从零搭建AI工程能力：为什么我劝你别一上来就啃论文

这两年“AI工程师”这个岗位被炒得火热，招聘JD上动不动就是“熟悉Transformer、有LLM微调经验、掌握RAG架构”。很多人一看就慌了，转头去啃《Attention Is All You Need》，结果公式推导看了三遍，连一个能跑起来的推理服务都没搭出来。我自己带过几个刚入行的同学，也见过不少转岗的朋友，最大的误区就是把“AI工程”等同于“AI研究”。这两件事的差别，比“会开车”和“会造发动机”的差别还大。

ai-engineering-from-scratch这个标题，核心讲的其实是一件事：如何从工程视角，而不是学术视角，一步步把AI能力落地成可运行、可维护、可扩展的系统。它解决的不是“模型为什么有效”的问题，而是“模型怎么跑起来、怎么接业务、怎么扛住流量、怎么持续迭代”的问题。适合谁看？适合那些已经会写Python、懂基本后端开发，但面对AI项目不知道从哪下手的人；也适合已经在做AI应用、但总觉得自己的系统“能跑但不敢上线”的工程师。

我自己的经验是，AI工程能力可以拆成四层：环境与工具链、模型调用与推理、数据管道与检索、服务化与运维。这四层缺一层，系统就是瘸的。下面我就按这个顺序，把每一层里最容易踩坑的地方、最值得抄的配置、最容易被忽略的细节，全部摊开讲一遍。你不需要先成为算法专家，但你需要成为一个能把算法“用起来”的工程师。

2. 环境与工具链：别让配环境吃掉你三天时间

2.1 为什么我坚持用uv而不是pip

刚入门的人最容易在环境上翻车。我见过一个同学，光装PyTorch就折腾了两天，最后发现是CUDA版本和驱动对不上。这里我给一个非常明确的建议：用uv做Python包管理，用conda做CUDA环境隔离，两者分工明确。

uv是这两年崛起的包管理器，速度比pip快一个数量级，而且它自带虚拟环境管理。你不需要再单独装virtualenv，也不需要记source activate那一套。安装uv只需要一行：

curl -LsSf https://astral.sh/uv/install.sh | sh

装完之后，创建一个新项目：

uv init ai-eng-demo cd ai-eng-demo uv venv --python 3.11 source .venv/bin/activate

为什么强调Python 3.11？因为3.12对部分AI库的兼容性还不稳定，3.10又缺少一些新特性。3.11是目前生态最稳的版本，实测下来torch、transformers、vllm都能正常跑。

至于CUDA，我建议用conda单独建一个环境，因为conda能帮你把cudatoolkit、cudnn这些底层库一次性装好，不用自己手动配LD_LIBRARY_PATH。命令如下：

conda create -n cuda-env python=3.11 conda activate cuda-env conda install pytorch torchvision torchaudio pytorch-cuda=12.1 -c pytorch -c nvidia

注意：conda环境和uv环境不要混用。我的做法是conda只管CUDA和PyTorch，其他纯Python依赖全部用uv装。这样职责清晰，出问题好排查。

2.2 目录结构决定你后期改代码的痛苦程度

很多人写AI项目，所有代码堆在一个文件夹里，train.py、inference.py、utils.py混在一起。等到要加一个检索功能，发现import路径全乱了。我从实际项目里总结出一个最小可用的目录结构，你可以直接抄：

ai-eng-demo/ ├── configs/ # 配置文件，yaml格式 │ ├── model.yaml │ └── service.yaml ├── src/ │ ├── data/ # 数据加载、清洗、切分 │ ├── models/ # 模型定义、加载、推理封装 │ ├── retrieval/ # 检索相关，向量库、embedding │ ├── service/ # API服务，FastAPI路由 │ └── utils/ # 日志、监控、通用工具 ├── tests/ # 单元测试 ├── scripts/ # 一次性脚本，数据预处理等 ├── pyproject.toml # uv管理的依赖 └── README.md

这个结构的好处是：每一层职责单一，依赖方向清晰。service层可以importmodels和retrieval，但反过来不行。这样你后期想把模型从本地换成API调用，只需要改models层，service层完全不用动。

2.3 配置文件别硬编码，用YAML加环境变量

我见过太多项目把模型路径、API密钥、超时时间直接写在代码里。一旦要换环境，就得改代码重新部署。正确做法是用YAML管结构，用环境变量管敏感信息。比如configs/model.yaml：

model: name: "bge-small-zh" path: "${MODEL_PATH:-./models/bge-small-zh}" device: "cuda" max_length: 512 batch_size: 32

然后在代码里用os.environ读取，配合pydantic做校验。这样本地开发时用默认路径，线上部署时通过环境变量覆盖，不用改一行代码。

实操心得：我习惯在configs里放一个local.yaml和prod.yaml，通过ENV=prod来切换。这样配置差异一目了然，不会出现“本地能跑线上挂”的情况。

3. 模型调用与推理：从“能跑”到“跑得稳”的关键细节

3.1 本地推理和API调用的选择逻辑

很多人一上来就问“我该用本地模型还是调API”。这个问题没有标准答案，但有一个判断框架：看你的数据敏感度、调用频率、延迟要求和预算。

如果数据不能出内网，那必须本地部署。如果调用频率低、延迟要求不严、预算充足，那调API更省事。我自己的做法是：开发阶段用API快速验证，生产阶段根据数据合规要求决定是否本地化。

本地推理目前最稳的方案是vllm，它支持连续批处理，吞吐量比HuggingFace的pipeline高好几倍。安装：

uv add vllm

启动一个OpenAI兼容的服务：

python -m vllm.entrypoints.openai.api_server \ --model ./models/Qwen2.5-7B-Instruct \ --dtype auto \ --max-model-len 8192 \ --gpu-memory-utilization 0.9

这里有几个参数值得解释。--dtype auto让vllm自动选择float16还是bfloat16，省得你手动试。--max-model-len控制上下文长度，设太大显存扛不住，设太小长文本会被截断。--gpu-memory-utilization 0.9表示用90%的显存，留10%给系统，避免OOM。

注意：vllm启动时会预分配显存，如果你同时跑多个模型，一定要算好总显存。我试过在一张24G的卡上同时跑一个7B模型和一个embedding模型，结果第二个直接OOM。后来改成embedding用CPU跑，才稳住。

3.2 推理封装的三个必备能力

不管你用本地模型还是API，推理层必须封装三个能力：重试、超时、降级。这三个能力不做好，线上就是定时炸弹。

重试的逻辑是：遇到网络抖动或服务暂时不可用，自动重试2到3次，每次间隔指数退避。超时的逻辑是：设置一个合理的超时时间，比如30秒，超过就放弃。降级的逻辑是：主模型不可用时，自动切到备用模型或返回缓存结果。

我用tenacity做重试，代码大概长这样：

from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def call_model(prompt: str) -> str: response = client.chat.completions.create( model="qwen2.5-7b", messages=[{"role": "user", "content": prompt}], timeout=30 ) return response.choices[0].message.content

这段代码的意思是：最多重试3次，第一次等2秒，第二次等4秒，第三次等8秒。为什么用指数退避？因为如果服务是过载导致的失败，你立刻重试只会加重过载。等一会儿再试，成功率更高。

3.3 批处理与流式输出的取舍

批处理能提高吞吐，流式输出能降低首字延迟。这两个怎么选？我的经验是：离线任务用批处理，在线对话用流式。

批处理的实现很简单，把多个请求攒成一个batch，一次性送给模型。vllm的LLM.generate支持传入一个prompt列表，返回一个结果列表。但要注意，batch size不是越大越好。我实测下来，batch size超过32之后，吞吐提升就不明显了，但显存占用线性增长。所以32是一个比较安全的默认值。

流式输出用stream=True，然后逐块读取。这里有个坑：流式输出时，如果客户端断开连接，服务端要能感知并停止生成。否则模型会一直跑下去，浪费算力。FastAPI里可以用request.is_disconnected()来检测。

from fastapi import Request from fastapi.responses import StreamingResponse @app.post("/chat") async def chat(request: Request, body: ChatRequest): async def generate(): async for chunk in model.stream(body.prompt): if await request.is_disconnected(): break yield chunk return StreamingResponse(generate(), media_type="text/event-stream")

实操心得：流式输出一定要加心跳。如果模型生成很慢，客户端可能以为连接断了。我习惯每5秒发一个空行作为心跳，保持连接活跃。

4. 数据管道与检索：RAG系统的命脉在这里

4.1 文档切分的颗粒度怎么定

RAG系统里，文档切分是最容易被忽视但影响最大的环节。切得太碎，检索出来的片段缺乏上下文；切得太粗，检索精度下降。我的经验是：中文文档按300到500字切分，英文按200到300词切分，重叠50字。

为什么要有重叠？因为一句话可能被切断，重叠能保证语义完整。比如“AI工程的核心是落地”这句话，如果正好在“核心”后面切断，检索“AI工程落地”时就匹配不上。重叠50字能解决大部分这类问题。

切分工具我推荐langchain-text-splitters，它支持按字符、按token、按递归分割。递归分割最智能，它会先按段落切，段落太长再按句子切，句子太长再按字符切。配置如下：

from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=400, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？", "；", "，", ""] )

注意separators的顺序，中文标点要放在英文标点前面，否则中文句子会被错误切分。

4.2 向量库选型：Chroma、Milvus还是Qdrant

向量库的选择取决于数据量和部署复杂度。我列一个对比表：

我自己的项目里，原型阶段用Chroma，上线后切到Qdrant。Chroma的API极其简单：

import chromadb client = chromadb.PersistentClient(path="./chroma_db") collection = client.get_or_create_collection("docs") collection.add(documents=chunks, ids=[str(i) for i in range(len(chunks))]) results = collection.query(query_texts=["AI工程"], n_results=5)

但Chroma的过滤能力弱，如果你需要按时间、按标签过滤，就得换Qdrant。Qdrant的过滤语法更灵活，而且支持payload索引，查询速度快很多。

注意：向量库的维度必须和embedding模型一致。比如bge-small-zh是512维，text-embedding-3-small是1536维。换模型时一定要重建索引，否则查询结果全是乱的。

4.3 检索策略：向量检索不够，还得加关键词

纯向量检索有个问题：对专有名词和数字不敏感。比如你搜“Qwen2.5”，向量检索可能返回一堆“Qwen”相关的文档，但精确匹配不到“Qwen2.5”。解决办法是混合检索：向量检索加BM25关键词检索，然后融合排序。

融合排序最简单的方法是RRF（Reciprocal Rank Fusion），公式是：

score = sum(1 / (k + rank_i))

其中k通常取60，rank_i是文档在第i路检索中的排名。RRF的好处是不需要调权重，两路检索的分数直接融合。

def rrf_fusion(vector_results, bm25_results, k=60): scores = {} for rank, doc_id in enumerate(vector_results): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1) for rank, doc_id in enumerate(bm25_results): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1) return sorted(scores.items(), key=lambda x: x[1], reverse=True)

实测下来，混合检索比纯向量检索的召回率高15%到20%，尤其是对技术文档和产品手册这类专有名词多的场景。

5. 服务化与运维：让系统敢上线的最后一步

5.1 FastAPI的异步陷阱

FastAPI是AI服务最常用的框架，但很多人用错了异步。最常见的错误是在async def路由里调用同步的阻塞函数，比如requests.get或model.generate。这会导致整个事件循环被阻塞，并发能力直接归零。

正确做法是：阻塞操作放到线程池里跑。用run_in_executor或者anyio.to_thread.run_sync：

import anyio from fastapi import FastAPI app = FastAPI() @app.post("/infer") async def infer(body: InferRequest): result = await anyio.to_thread.run_sync(model.generate, body.prompt) return {"result": result}

这样模型推理在线程池里跑，事件循环不被阻塞，其他请求还能正常处理。

实操心得：我见过一个项目，QPS上不去，排查了半天发现是日志写文件用了同步IO。改成异步日志后，QPS直接翻倍。所以任何IO操作都要检查是不是阻塞的。

5.2 监控指标：别只看QPS

AI服务的监控和普通Web服务不一样。除了QPS、延迟、错误率，还要看token吞吐量、显存占用、队列长度。这几个指标能提前预警。

token吞吐量突然下降，可能是模型遇到了长文本，生成变慢。显存占用持续上涨，可能是内存泄漏，需要重启。队列长度超过阈值，说明服务过载，要扩容或限流。

我用prometheus-client暴露指标：

from prometheus_client import Counter, Histogram, Gauge REQUEST_COUNT = Counter("ai_requests_total", "Total requests", ["endpoint", "status"]) LATENCY = Histogram("ai_latency_seconds", "Request latency", ["endpoint"]) GPU_MEMORY = Gauge("ai_gpu_memory_bytes", "GPU memory usage")

然后在推理前后打点。Grafana面板上把这三个指标放在一起看，基本能判断服务健康度。

5.3 限流与降级：保护自己不被流量打死

AI服务的特点是单次请求消耗大，一个长文本推理可能占用几秒的GPU时间。如果不限流，几个并发请求就能把服务打满。限流我推荐用slowapi，基于令牌桶算法：

from slowapi import Limiter from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) @app.post("/chat") @limiter.limit("10/minute") async def chat(request: Request, body: ChatRequest): ...

这个配置表示每个IP每分钟最多10次请求。超过就返回429。

降级策略是：当GPU显存不足或队列过长时，自动切到小模型或返回缓存。我习惯在服务里维护一个fallback_model，主模型不可用时自动切换。切换逻辑要记录日志，方便事后分析。

注意：限流阈值要根据实际压测结果来定。我一般先用locust压测，找到服务能稳定支撑的最大QPS，然后限流阈值设为这个值的80%，留20%的余量。

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

6.1 模型加载慢、首次推理慢怎么办

这是新手最常问的问题。模型加载慢是因为要从磁盘读权重，首次推理慢是因为要初始化CUDA kernel。解决办法有两个：预热和缓存。

预热是在服务启动后，立刻跑一次推理，把CUDA kernel初始化好。这样第一个真实请求就不会慢。代码很简单：

@app.on_event("startup") async def warmup(): model.generate("warmup")

缓存是把模型权重放到内存文件系统，比如/dev/shm，读取速度比磁盘快很多。但要注意/dev/shm的大小限制，默认是内存的一半。

6.2 检索结果不相关怎么调

RAG系统检索不准，通常有三个原因：切分粒度不对、embedding模型不匹配、检索策略单一。排查顺序是：先看切分后的片段是否语义完整，再看embedding模型是否适合中文，最后看是否加了混合检索。

我遇到过一个案例，用户搜“如何退款”，检索出来的全是“退款政策”的文档，但用户想要的是“退款操作步骤”。原因是切分时把操作步骤和 policy 混在一起了。后来改成按标题切分，每个标题下的内容单独成块，问题就解决了。

6.3 显存泄漏怎么排查

显存泄漏的表现是：服务跑一段时间后OOM。排查方法是：在每次推理前后打印torch.cuda.memory_allocated()，看是否持续增长。如果增长，说明有张量没释放。

常见原因是：把中间结果存到了全局变量，或者用了torch.no_grad()但没加del。解决办法是：推理函数里用with torch.no_grad():，结束后手动del中间变量，并调用torch.cuda.empty_cache()。

实操心得：我习惯在服务里加一个定时任务，每处理1000个请求就调一次empty_cache()。虽然会稍微降低性能，但能有效防止显存碎片化导致的OOM。

6.4 常见问题速查表

7. 我踩过的坑和给你的三条建议

第一条建议：不要追求一步到位。我见过太多人想一次性把RAG、微调、Agent全做完，结果哪个都没做好。正确的做法是先跑通一个最小闭环：一个模型、一个向量库、一个API，能回答一个问题就行。然后再逐步加检索、加缓存、加监控。

第二条建议：日志要打全，但别打敏感信息。AI服务的日志里经常包含用户输入和模型输出，这些可能含隐私。我习惯在日志里只打请求ID、耗时、token数，不打具体内容。需要调试时，用请求ID去查专门的调试日志。

第三条建议：压测要趁早。很多人等到上线前才压测，结果发现一堆问题来不及改。我的做法是：服务能跑通后就立刻压测，用locust模拟10个并发用户，看延迟和错误率。如果10个并发就扛不住，那说明架构有问题，早发现早改。

最后分享一个小技巧：用nvidia-smi的--query-gpu参数做实时监控，比看默认输出清晰得多。

nvidia-smi --query-gpu=utilization.gpu,memory.used,memory.total --format=csv -l 1

这个命令每秒刷新一次，输出GPU利用率、已用显存、总显存。配合watch命令，可以一直挂在终端里看。我调试推理性能时，这个命令基本不离手。

AI工程这个方向，说到底就是把不确定性管起来。模型输出不确定，那就加校验和降级；流量不确定，那就加限流和弹性；数据不确定，那就加检索和过滤。你不需要成为算法专家，但你需要成为那个能让系统稳定跑起来的人。这条路我走了好几年，踩过的坑比写过的代码还多，但每填一个坑，系统就稳一分。希望这些经验能帮你少走点弯路。