我花了三年时间在各类AI系统里打转，从最早的文本分类、关键词抽取，到后来做问答、知识库、智能体，越来越发现一个残酷的事实：大多数AI项目不是死在模型效果上，而是死在工程化上。这三年里，我反复验证了一件事——如果你真的想把AI工程做扎实，最可靠的方式不是找一个酷炫的框架往上堆，而是从零开始手写核心逻辑，把每一层拆开看清楚。这个项目“ai-engineering-from-scratch”就是在这样的背景下启动的：放弃对黑盒的依赖，用最朴素的代码把模型调用、向量检索、上下文管理、任务编排串成一条完整链路。
这篇文章我会把它的完整设计思路、技术选型、关键实现、评测方法和踩坑记录都摊开来讲。不是说框架不好，而是很多问题只有在你亲手从零实现一遍之后，才能理解框架为什么那样设计。你可能是刚入门AI开发的学生，也可能是已经在生产环境里被各种抽象概念折腾过的工程师，这篇内容都能给你一套可复现、可验证的落地路径。
1. 为什么我坚持从零开始搭建AI工程
1.1 框架带来的是什么，又偷走了什么
引入现成的AI框架时，最常见的情景是你还没有搞清楚业务需要什么，就已经被框架背后的抽象模型牵着走了。比如某个编排框架要求你定义节点、定义连接、定义状态机，这些概念本身没有问题，但问题在于：如果你没有亲手实现过一个状态循环，你根本无法判断这些抽象是必要的还是多余的。
框架偷走的是你对细节的感知。当你调用一个封装好的Agent执行函数时，你只知道它“工作”，不知道它内部如何处理上下文超长、工具返回格式异常、模型输出JSON非法。这些细节才是生产环境真正的拦路虎。我从零开始写第一版时，目标非常明确：不要任何框架级的运行时，只用HTTP请求、Python字典、SQLite和少量参数配置，把整个流程跑通。
1.2 这个项目到底在解决什么问题
我观察到很多团队在落地AI功能时反复卡在同一个点上：单点技术都会，串联起来就崩。有人能把大模型调用写得很好，但不知道如何管理多轮记忆；有人懂向量化，但不知道召回结果怎么跟模型输出连接起来；有人会调提示词，但无法建立一套离线评测集来度量改动的好坏。
这个项目就是以“最小完整AI系统”为目标，把下面这些能力逐个从零实现：
- 一个稳定的文本生成服务，支持流式输出和参数控制。
- 一个基于向量的语义检索模块，不依赖分布式数据库。
- 一个能管理多轮对话、自动压缩历史的上下文层。
- 一个支持工具调用的执行循环，让模型能触发外部函数。
- 一个离线评测和线上日志记录系统，能定位每次输出变差的原因。
范围听上去不小，但每个模块都刻意保持极简。目标是让读者用一天时间读懂全部代码，用一周时间改造成自己可用的版本。如果做不到这个标准，说明抽象层又加多了，需要继续砍。
2. 项目蓝图和目标拆解
2.1 先锁定边界：不做大而全的平台
项目启动时，我给自己列一条铁律：不做平台，只做“一个人能跑通的工具链”。没有用户管理系统，没有权限体系，没有任务队列，没有插件市场。这些东西等你有真实流量之后再考虑。
从零开始的另一个原则是每多一个组件，必须能回答“它解决了什么问题”。比如要不要上Redis？如果只用来存会话状态，那SQLite加一个缓存字典就够了，前提是你清楚自己的并发量级。如果每天只有几百次请求，微服务拆分也纯属给运维加负担。我当时明确服务形态是单进程多线程的Python服务，只在外面套了一个反向代理做TLS和负载均衡。
这种边界设定有两层价值。第一层是心理上的，它让你敢动手，因为目标足够小；第二层是工程上的，当你真的需要扩容时，你会知道瓶颈具体出现在模型的推理时间、向量检索的召回量，还是外部API的延迟，而不是对着分布式架构图猜。
2.2 验收标准和评估维度
只有目标不算验收，我按照下面几个维度给项目定了可量化的标准：
| 维度 | 目标值 | 测量方式 |
|---|---|---|
| 端到端延迟 | 单轮问答P95低于3秒 | 请求日志统计 |
| 上下文准确率 | 多轮追问后仍能正确引用上一轮信息 | 离线测试集人工标注 |
| 检索命中率 | 前5条召回中包含正确答案的比例高于80% | 检索评测集 |
| 工具调用成功率 | 模型输出的函数参数能被完美解析的比例高于95% | 线上日志统计 |
| 资源占用 | 单容器内存低于2GB | 监控面板 |
注意这里面没有“准确率”这个指标，因为开放式生成任务的“准确率”很难定义。我更关注它是否可靠、是否可用、是否经济。AI工程的基本功是把模糊的目标翻译成可测的指标，如果做不到这一点，后续所有优化都无从谈起。
3. 技术选型：该省的省，该造的造
3.1 模型与服务：自托管还是API
做AI工程的第一步就是选模型。我的选择标准有三个：能下载权重、有公开基准、社区反馈活跃。考虑到成本和数据安全，我选择把模型放在本地GPU上推理，用vLLM或transformers+pipeline跑起来，具体取决于部署机器显存大小。
如果你没有GPU，就老老实实用云厂商的模型API。这里有一个非常容易犯的错误：过早地优化模型调用层。有人一开始就封装了十几个厂商接口，做了复杂的fallback策略，结果业务还没跑通，先被兼容性问题拖死。正确的姿势是先写一个极简的ChatModel抽象，里面只有两个方法：generate(prompt)和generate_stream(prompt)，实现先接一个来源，跑起来再说。
3.2 向量数据库：让检索先简单起来
向量检索是AI工程里被炒作最严重的部分。事实上有80%的早期项目用numpy加暴力检索就够用，因为数据量不超过十万条。暴力检索在这个规模下延迟是毫秒级，而它带来的好处是完全可控、零依赖、方便调试。
我第一版实现方案非常简单：
- 用
text-embedding接口把文档切成小块并向量化。 - 全部向量存成一个
float32的numpy矩阵。 - 查询时把用户问题向量化，用余弦相似度或内积做top-k检索。
代码只有不到一百行。等数据量真的涨上来，再把底层换成sqlite-vss或者faiss，因为此时你已经清楚接口长什么样、需要索引哪些元数据，迁移成本非常低。先让逻辑正确，再让性能飞的理念在这里非常关键。
3.3 编排：代码优先而非工作流平台
现在很多AI工程强调用可视化工作流平台，把节点拖来拖去。我不反对工具本身，但我反对在项目初期使用它。原因是可视化编排会把逻辑关系藏在图形化界面里，代码评审、版本控制、单元测试全都变得非常别扭。
我的编排层是一段普通的Python代码：一个while循环里跑“生成→解析→调用工具→观察结果→再次生成”，最大轮数设置成8次。每一个步骤都写日志，记录模型输入输出和工具返回值。这种朴素实现有几个实打实的好处：调试时能打印出完整轨迹，问题定位精确到代码行；迭代时可以直接改函数逻辑；压测时能清楚看到循环里最耗时的环节。
3.4 配置与观测：从第一天就开始搭
观测能力的建设决定你能走多快。不是等系统上线后再加，而是在最基础的服务里就内置日志。每个请求分配一个request_id，日志格式统一为JSON，包含时间戳、模型名、请求参数、token用量、延迟、返回状态。
配置管理的核心是敏感信息不进代码库。我在项目根目录放一个.env.example，把需要的环境变量列清楚，然后本地维护真实的.env文件。另一个关键配置是提示词模板：提示词是易变资产，我单独放在prompts/目录下，用文件名做版本标识，不硬编码在Python代码里。这样每次改动提示词都能通过git diff看到变化，避免“为什么线上效果变了”的悬案。
4. 核心模块的从零实现
4.1 推理服务：一个能用的文本生成接口
先说明一下，我这里说的推理服务不是部署大模型本身的工程，而是把模型封装成一个对内部业务友好的接口。无论底层是本地GPU还是云端API，对外暴露的接口保持一致。
这是我的第一版最小实现：
# app/services/chat_model.py import json import time from dataclasses import dataclass, field @dataclass class ChatMessage: role: str content: str @dataclass class ChatRequestBody: messages: list[ChatMessage] temperature: float = 0.7 max_tokens: int = 512 tools: list[dict] = field(default_factory=list) class ChatModel: def __init__(self, base_url, api_key, model_name): self.base_url = base_url self.api_key = api_key self.model_name = model_name def generate(self, body: ChatRequestBody) -> str: payload = self._build_payload(body) start = time.time() # 这里是实际的HTTP调用，不同服务商地址不同 response = self._http_post(payload) self._log_usage(body, response, time.time() - start) return response["choices"][0]["message"]["content"] def _build_payload(self, body): payload = { "model": self.model_name, "messages": [ {"role": m.role, "content": m.content} for m in body.messages ], "temperature": body.temperature, "max_tokens": body.max_tokens, } if body.tools: payload["tools"] = body.tools payload["tool_choice"] = "auto" return payload注意几个细节。第一，我把temperature和max_tokens放进请求体，而不是在函数里写死，因为线上调参是很频繁的事，硬编码会逼着你频繁改代码。第二，generate返回的是纯字符串，工具调用场景需要额外解析，后面专门处理。第三，我在请求前打印完整请求摘要，调试AI系统时，看不到模型收到的原始内容等于盲人摸象。
4.2 记忆与向量检索：让AI持有上下文
要让模型记住多轮对话，最朴素的做法是把历史消息全部拼进prompt里。但token是成本，而且上下文过长会稀释模型对当前问题的注意力。所以需要一个记忆层来做取舍：短期对话直接携带，长期记忆靠向量检索召回。
我的实现思路是用“消息回收”机制：当会话历史总字数超过阈值，把最早的消息向量化存入记忆库，后续每次提问先从记忆库召回复述给模型。
# app/services/memory.py import numpy as np from typing import List class VectorMemory: def __init__(self): self.vectors = [] self.texts = [] self.metadata = [] def add(self, text: str, embedding: List[float], meta: dict): self.vectors.append(np.array(embedding, dtype=np.float32)) self.texts.append(text) self.metadata.append(meta) def search(self, query_embedding: List[float], top_k: int = 3): if not self.vectors: return [] q = np.array(query_embedding, dtype=np.float32) matrix = np.vstack(self.vectors) # 归一化后内积等价于余弦相似度 q_norm = q / (np.linalg.norm(q) + 1e-9) matrix_norm = matrix / (np.linalg.norm(matrix, axis=1, keepdims=True) + 1e-9) scores = (matrix_norm @ q_norm).tolist() top_indices = np.argsort(scores)[::-1][:top_k].tolist() return [ {"text": self.texts[i], "score": scores[i], "metadata": self.metadata[i]} for i in top_indices ]有个很关键的工程细节：检索结果必须带上得分和元数据一起返回，而不能只返回文本。因为你要在日志里追踪“这条记忆是为什么被召回的”，得分能帮助判断阈值是否合理，元数据能帮你排查是不是误召回了另一篇文章的内容。
4.3 把RAG串起来
RAG的完整流程不是“检索+拼接提示词”这么简单。一个稳健的RAG管线要处理检索结果的排序、过滤、引用标注、上下文窗口裁剪。
我的实现里有三个关键步骤：
- 将用户问题向量化，从
VectorMemory和文档库中分别召回候选。 - 对候选去重，按得分排序，过滤掉低于阈值的结果。
- 选择能挤进
max_tokens预算的top结果，按“问题+上下文参考+历史摘要+指令”的顺序拼接提示词。
上下文拼接顺序我试过很多种，最后固定为“系统指令→参考材料→对话历史→问题本身”。原因是模型对提示词末尾内容的注意力通常更强，把问题放最后能让它聚焦当前任务。而参考材料放在历史之前，能防止模型把资料内容误当做自己说过的话。
下面是核心实现示意：
# app/services/rag.py from typing import List from .memory import VectorMemory class RAGEngine: def __init__(self, memory, embedder): self.memory = memory self.embedder = embedder def build_prompt(self, query: str, history: List[dict], system_prompt: str) -> str: query_vec = self.embedder.embed(query) refs = self.memory.search(query_vec, top_k=5) # 过滤低分结果 filtered = [r for r in refs if r["score"] > 0.45] ref_text = "\n\n".join( f"[{i+1}] {r['text']}" for i, r in enumerate(filtered) ) if ref_text: prompt = f"{system_prompt}\n\n参考材料：\n{ref_text}\n\n对话历史：\n{self._format_history(history)}\n\n当前问题：{query}\n回答：" else: prompt = f"{system_prompt}\n\n对话历史：\n{self._format_history(history)}\n\n当前问题：{query}\n回答：" return prompt def _format_history(self, history): return "\n".join(f"{m['role']}: {m['content']}" for m in history[-6:])这个阈值0.45是不是最优？不一定。它需要在你的业务数据上多次测试后确定。我建议不要把它写死在代码里，而是放进配置项，方便线上快速调整。
5. 评测、迭代、成本平衡
5.1 离线评测集怎么做才有参考价值
没有评测集的AI工程就是闭眼开车。但评测集怎么建，直接决定它的参考价值。我的经验是：不要只用标准公开数据集，一定要加入你自己业务场景的真实案例。
我维护了一个eval_cases.jsonl，每行是一个带标注的测试样本：
query：用户真实问题expected_span：期望检索回来的文档片段expected_action：期望的最终回复要点category：用于区分不同业务场景
每次改动提示词或检索逻辑时，用同一批测试集跑一遍，输出比对结果。如果某一类问题效果退化，立刻能定位到是哪个环节导致的。这里有个实操技巧：评测集要持续补充线上出错的case。凡是在线上被用户反馈为错误答案的例子，都要分析原因后加入评测集，防止同类问题回归。
5.2 线上监控：不只是看延迟
延迟和错误率当然要监控，但AI系统最大的隐患是“看似正常实则不对”。只靠HTTP状态码完全不够，必须监控语义层面的指标。我在日志里额外记录了三项内容：
#ctx_tokens：每次请求消耗的上下文token数，用于观察上下文是否在膨胀。recalled_documents：实际召回文档的ID和得分，用于评估检索质量。parse_error：模型输出无法解析的报错计数，这往往是工具调用失败的前兆。
其中最容易被忽略的是“召回得分分布”。如果大量请求的召回得分都集中在0.3附近，说明你的embedding模型和业务语义不匹配，需要换模型或调整切块策略。如果得分普遍很高但答案还是不对，那问题可能出在提示词拼接顺序，而不是检索本身。
5.3 成本控制与延迟优化
做AI工程的人都会经历“一开始什么功能都敢接，月底被账单吓一跳”的阶段。成本大头不在调用次数，而在无脑地往上下文里塞内容。我的优化策略有几条：
- 对话历史优先截断：保留最近三条消息，更早的消息用摘要替代。摘要本身由模型生成，但只在会话空闲时异步执行。
- 让检索结果可被丢弃：很多系统把检索资料无条件拼进提示词，导致模型被无关信息干扰。我引入一个“相关性验证”步骤，让模型先判断参考资料是否与问题有关，无关就不拼。
- 批量处理嵌入式任务：如果你有大量离线文档要向量化，不要一条条循环调用接口，一次性打包成批次发送，可以节省大量时间。
延迟方面的记录是：在一次真实项目中，模型推理占了总延迟的75%，检索只占8%，其余是网络和解析。所以优先优化推理路径，比如使用更小的模型、量化模型、开启动态批处理，而不是在向量检索上疯狂加缓存。这个结论可能跟很多人的直觉相反，但数据不会骗人。
6. 踩坑实录：覆盖和遗漏
6.1 常见问题速查表
以下是这个项目从开发到上线之间遇到的典型问题，我把每一个都整理成了“现象→原因→解法”的表格。
| 现象 | 原因 | 解法 |
|---|---|---|
| 模型回答突然变成英文 | 系统提示词被上下文窗口截断 | 检查提示词拼接位置，确保系统指令在最前且不被截断 |
| 工具调用总是参数JSON解析失败 | 模型输出包含markdown代码块包围 | 解析前先剥离```围栏，再尝试json.loads |
| 多轮对话到第五轮就开始答非所问 | 对话历史全部累积，关键信息被稀释 | 加入历史消息的摘要压缩，保留最近消息 |
| 检索结果与问题完全无关 | 文档切块太细，单块信息量不足 | 调整切块长度，增加重叠窗口 |
| 同一问题两次回复差异巨大 | temperature设置过高 | 非创意任务将temperature降到0.2以下 |
| 服务频繁超时 | 单请求最大token数设置过大，推理排队 | 限流，降级为流式输出 |
每条背后都是真实的调试经历。比如JSON解析失败这个问题，一开始我很不理解，直到我打印了模型原始输出才发现它把工具参数用三反引号包住了。从那以后，所有工具调用解析都加一步预处理。
6.2 我反复踩的几个坑
第一个坑是过早抽象嵌入函数。最初我把嵌入调用封装成一个类，设计了几种不同的构造方式，结果业务变更是经常发生的事，但抽象层反而成为修改变动的障碍。后来改成普通函数，带两个默认参数，世界清爽了。记住：从零开始的代码不在于设计多优雅，而在于修改时能快速定位。
第二个坑是没有给提示词做版本化。我曾在线上直接改了提示词文本，没有提交到git仓库。结果效果变化无法追踪，回滚也没法做。现在所有提示词都作为独立文件管理，并且每次修改都提交commit记录，这已经是项目的强制流程。
第三个坑是低估了流式输出的复杂度。普通输出很简单，但一旦接上前端流式渲染，就要处理SSE协议、断线重连、半截JSON。如果业务对实时要求不高，完全可以先用非流式接口上线，后续再优化体验。
第四个坑是把检索阈值设得太刚。我曾经把召回阈值固定为0.6，发现很多有效信息被过滤掉。后来把阈值做成动态可配置，并在日志里保留被过滤样本。通过观察被过滤样本的真实相关度，我发现自己当时的阈值设定缺乏数据支撑，完全是拍脑袋。
7. 最后：这种折腾值不值
把这套从零开始的工程做完整之后，我回头看最大的收获不是代码本身，而是建立了一种排查问题的直觉。现在我接手任何一个AI项目，第一反应不再是“用什么框架”，而是“先看数据流，再定技术栈”。这种直觉没法通过阅读框架文档获得，只能靠在最基础的地方亲手写一遍、亲手踩一遍坑来积累。
如果你现在正被各种AI工程平台、低代码工具、大模型框架包围，我的建议是：别急着选型，先给自己一周时间，从零写一个最小闭环。不需要写多好，只需要跑通从输入到输出再到反馈的完整链路。等你做完，再回头看看那些框架，你会发现自己终于能看懂它们的设计初衷，也终于能判断哪些功能对你的业务是必需，哪些只是噪音。
如果让我总结一句，那就是：AI工程的核心不在模型，而在围绕模型建立的那套稳定、可控、可观测的系统。而这个系统，恰恰最应该从零开始盖一遍。
