从零构建AI工程能力：Python、TypeScript、Rust三语言实战

1. 从零搭建 AI 工程能力为什么我决定自己造一遍轮子这两年 AI 应用开发的门槛肉眼可见地降低了调个 API、拼几段提示词就能跑出一个能用的 Demo。但真到了要上生产、要控成本、要排查线上问题时很多人包括我自己会突然发现底层那套东西我其实没真正搞明白。模型怎么加载、推理怎么调度、向量检索为什么慢、上下文怎么裁剪、流式输出怎么保证不丢包——这些在框架里被封装成一行代码的东西一旦出问题就抓瞎。ai-engineering-from-scratch这个项目就是冲着这个痛点去的。它的核心主张很直接不依赖现成的高层框架用 Python、TypeScript、Rust 三种语言从最基础的组件开始把 AI 工程里那些关键环节亲手实现一遍。它不是教你调库而是让你理解每一层抽象背后到底发生了什么。适合谁看有一定编程基础、想把 AI 应用从“能跑”做到“跑得稳、跑得省、跑得明白”的开发者尤其是那些被框架黑盒坑过、想补底层认知的人。我自己带过几个 AI 相关的项目踩过的坑基本都集中在“框架帮你做了决定但你没意识到”这件事上。比如默认的向量检索用的是暴力搜索数据量一上来延迟直接爆炸比如流式响应在弱网下会莫名其妙断掉查半天发现是缓冲区没处理好。这些问题的解法其实都藏在“从零实现”的过程里。所以这个项目的价值不在于产出什么生产级代码而在于把黑盒拆开让你看清楚每个齿轮是怎么咬合的。下面我会按“整体设计思路 → 核心细节解析 → 实操过程 → 常见问题排查”这条线把我在复现这个项目时的思考、取舍和踩坑经验完整讲一遍。三种语言的选型不是随便定的每一层用什么语言、为什么这么选我都会说清楚背后的逻辑。2. 整体设计与技术选型三种语言各管一摊2.1 为什么是 Python TypeScript Rust 这个组合很多人第一反应是做一个项目为什么要用三种语言这不是给自己找麻烦吗我一开始也这么想但真正动手之后发现这个组合恰恰是这个项目最聪明的地方——它对应的是 AI 工程里三个天然不同的层次每个层次对语言特性的诉求完全不一样。Python 负责的是模型侧和数据处理侧。原因很实际主流模型生态、数据处理库、科学计算栈几乎都长在 Python 上。你要做 tokenization、要处理 embedding、要跑推理用 Python 能直接对接现成的生态开发效率最高。这一层的核心诉求是“快速验证 生态丰富”性能不是第一优先级。TypeScript 负责的是应用侧和交互侧。AI 应用最终要落到 Web 前端或者 Node 服务上流式输出、SSE、WebSocket、UI 状态管理这些活儿TypeScript 的类型系统能帮你把数据结构的边界卡得很死。尤其是流式响应的分片处理类型定义清晰能省掉大量运行时调试。这一层的核心诉求是“类型安全 异步编排”。Rust 负责的是性能敏感的基础设施侧。向量检索、相似度计算、高并发请求调度、内存管理这些地方 Python 的性能瓶颈非常明显。Rust 的零成本抽象和内存安全特性让它成为重写这些核心组件的最佳选择。这一层的核心诉求是“极致性能 内存可控”。提示不要试图用一种语言打通所有层。我见过有人硬要用 Python 写高性能向量检索最后靠 numpy 向量化勉强撑住但并发一上来还是崩。分层选型不是炫技是让每层用最合适的工具。2.2 分层架构的核心思路整个项目我理解下来是一个清晰的四层结构从下往上依次是基础层tokenizer、embedding 计算、向量相似度算法。这一层是纯计算用 Rust 实现核心算法Python 通过绑定调用。检索层向量索引、近似最近邻搜索ANN、元数据过滤。这一层是 AI 应用区别于普通应用的关键也是性能瓶颈最集中的地方。编排层上下文组装、提示词模板、多轮对话状态管理、流式响应控制。这一层用 TypeScript 实现因为它本质是异步流程编排。应用层API 接口、前端交互、错误处理、可观测性。这一层是用户直接接触的部分。这么分的好处是每一层可以独立测试、独立替换。比如你不想用 Rust 的向量检索可以换成 Python 的 faiss 实现上层编排逻辑完全不用动。这种解耦在实际项目里非常值钱因为 AI 领域的技术栈变化太快今天用的库明天可能就过时了分层能让你把替换成本控制在一层之内。2.3 从零实现的边界在哪里这里要说清楚一个容易误解的点“from scratch”不等于“什么都自己写”。你不可能从晶体管开始造计算机。这个项目的边界划得很清楚自己实现的部分向量相似度算法、简单的 ANN 索引结构、tokenizer 的核心逻辑、流式响应的分片与重组、上下文窗口的裁剪策略、请求的重试与降级逻辑。直接使用的部分HTTP 服务器框架、JSON 序列化库、基础的数学运算库、模型推理的底层运行时。这个边界的原则是凡是影响 AI 应用核心行为和性能的自己实现凡是通用的、与 AI 无关的基础设施直接用成熟的。比如你不会自己去写一个 HTTP 服务器但你应该自己实现向量检索的索引结构因为这直接决定了你的应用能扛多少数据、延迟有多高。3. 核心细节解析那些框架不会告诉你的关键点3.1 向量检索为什么暴力搜索一定会崩向量检索是 AI 应用里最容易被低估的环节。很多人用框架的时候默认就是暴力搜索brute force数据量小的时候感觉不到问题一旦向量数量超过十万级别延迟就会从毫秒级跳到秒级。暴力搜索的本质是对每一个查询向量都要和库里所有向量算一遍相似度。假设你有 N 个向量每个向量维度是 D那么单次查询的计算量就是 O(N×D)。N100000、D768 的时候单次查询要做 7680 万次浮点运算。就算用 SIMD 加速也很难压到 100ms 以内。所以从零实现的时候必须引入ANN近似最近邻索引。最经典的是 HNSW分层可导航小世界图它的核心思想是把向量组织成一个多层图结构查询的时候从顶层开始快速定位到大致区域再逐层往下细化。这样查询复杂度能从 O(N) 降到 O(log N) 级别。我在 Rust 里实现 HNSW 的时候有几个参数特别关键参数含义推荐值影响M每个节点的最大连接数16-32越大召回率越高内存占用越大ef_construction建图时的候选集大小200越大图质量越好建图越慢ef_search查询时的候选集大小50-100越大召回率越高查询越慢这几个参数的取舍逻辑是M 决定图的连通性ef_construction 决定建图质量ef_search 决定查询精度。实际调的时候先用 M16、ef_construction200 建图然后根据召回率要求调 ef_search。如果召回率不够优先加 ef_search因为它只影响查询速度不影响内存。注意HNSW 的召回率不是 100%它是用精度换速度。如果你的场景要求精确检索比如去重那还是得用暴力搜索或者精确索引。别盲目上 ANN。3.2 流式响应分片、缓冲与断线重连流式响应是 AI 应用体验的关键。用户等 10 秒看到完整答案和逐字蹦出来的感觉完全不一样。但流式响应的坑特别多框架通常只帮你处理了最顺利的情况。从零实现流式响应核心要解决三个问题第一是分片边界。模型输出的 token 流不是按字符切的可能一个 UTF-8 字符被切在两个分片里。如果你直接按分片解码就会出现乱码。正确做法是维护一个缓冲区每次收到分片先追加到缓冲区然后尝试解码如果解码失败就等下一个分片。TypeScript 里可以用TextDecoder的stream: true选项来处理。第二是背压处理。如果客户端消费速度慢于服务端生产速度数据会堆积在内存里。Node 的 Stream 有内置的背压机制但如果你自己实现 SSE就要手动处理。我的做法是给缓冲区设一个上限超过就暂停读取上游等下游消费完再恢复。第三是断线重连。流式连接很容易因为网络抖动断掉。如果断了就从头发用户体验很差。正确做法是给每个分片编号客户端记录已收到的最大编号重连时带上这个编号服务端从下一个分片继续发。// 流式响应的分片处理核心逻辑 class StreamBuffer { private buffer ; private decoder new TextDecoder(utf-8, { stream: true }); private lastChunkId 0; push(chunk: Uint8Array, chunkId: number): string { // 处理乱序和重复 if (chunkId this.lastChunkId) return ; this.lastChunkId chunkId; this.buffer this.decoder.decode(chunk, { stream: true }); // 按句子边界切分避免半个句子发给用户 const boundary this.buffer.lastIndexOf(。); if (boundary -1) return ; const output this.buffer.slice(0, boundary 1); this.buffer this.buffer.slice(boundary 1); return output; } }这段代码的关键在于stream: true和句子边界切分。前者解决 UTF-8 跨分片问题后者解决语义完整性问题。用户看到的永远是完整的句子而不是半个字。3.3 上下文窗口管理裁剪策略决定成本和质量上下文窗口是 AI 应用里最贵的资源。每次请求都要把历史对话、检索结果、系统提示词拼进去token 数直接决定成本。框架通常给你一个“最大 token 数”的配置但怎么裁剪、裁剪什么它不管。从零实现的时候我总结了三种裁剪策略按优先级从高到低策略一按相关性裁剪检索结果。检索回来的文档按相似度排序从高到低累加 token 数超过预算就截断。这是最直接的但要注意保留至少一条结果否则模型没有上下文可用。策略二按时间窗口裁剪对话历史。多轮对话里越早的消息越不重要。可以保留最近 N 轮或者按时间衰减给旧消息降权。我的做法是保留最近 5 轮完整对话再往前只保留用户的关键问题去掉助手的冗长回复。策略三系统提示词压缩。系统提示词通常很长但很多内容是固定的。可以把不变的部分缓存起来只拼接变化的部分。有些模型支持 prompt caching能省不少钱。裁剪策略适用场景风险相关性截断检索增强问答可能丢掉关键但相似度低的结果时间窗口多轮对话可能丢掉早期的重要约定提示词压缩固定系统提示压缩过度可能改变语义实际用的时候这三种策略是组合使用的。先压缩系统提示词再按时间窗口裁剪对话最后按相关性截断检索结果。每一步都要留出 token 预算不能把窗口占满要留 10%-20% 给模型输出。3.4 错误处理与降级AI 应用比普通应用更脆弱AI 应用的错误率天然比普通应用高。模型服务可能超时、可能限流、可能返回格式错误的结果。从零实现的时候错误处理不是可选项是核心功能。我总结的降级链路是这样的重试对于超时和限流先重试。重试要带指数退避第一次等 1 秒第二次等 2 秒第三次等 4 秒。别用固定间隔否则会把下游打垮。降级模型如果主模型连续失败切到备用模型。备用模型可以小一点、慢一点但至少能返回结果。降级检索如果向量检索失败降级到关键词检索。虽然召回率低但至少能用。兜底回复如果全挂了返回一个预设的兜底话术别让用户看到 500 错误。提示重试一定要设上限并且要区分可重试错误和不可重试错误。参数错误、鉴权失败这种重试多少次都没用直接失败。超时、限流、5xx 才值得重试。4. 实操过程从环境搭建到跑通全链路4.1 环境准备三种语言的工具链配置这部分我踩了不少坑尤其是 Rust 和 Python 的混合调用。先把环境说清楚。Python 侧建议用 3.10 以上版本因为要用到一些新的类型语法。虚拟环境用 venv 或者 conda 都行我习惯用 venv轻量。核心依赖是 numpy 做数值计算其他尽量少装因为我们的目标是“从零实现”装太多库就失去意义了。python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install numpyTypeScript 侧Node 18 以上因为要用到原生的 fetch 和 Stream API。TypeScript 配置里target设成 ES2022module设成 NodeNext。注意别用那些已经废弃的配置项比如moduleResolution: node10新版本会报错。{ compilerOptions: { target: ES2022, module: NodeNext, moduleResolution: NodeNext, strict: true, esModuleInterop: true } }Rust 侧用 rustup 装最新稳定版。如果国内下载慢可以配置镜像源改~/.cargo/config.toml里的 registry 地址。核心依赖是pyo3Python 绑定和serde序列化。pyo3的版本要和 Python 版本匹配这个很容易搞错。[ dependencies ] pyo3 { version 0.20, features [extension-module] } serde { version 1.0, features [derive] }注意pyo3 编译出来的扩展模块Python 版本必须和编译时用的版本一致。我试过用 3.11 编译、3.10 运行直接报 ABI 不兼容。建议用 maturin 来管理构建它会自动处理版本匹配。4.2 第一步实现 tokenizer 的核心逻辑tokenizer 是 AI 工程的入口所有文本都要先变成 token 才能喂给模型。从零实现一个完整的 BPE tokenizer 工作量不小但核心逻辑其实不复杂。BPEByte Pair Encoding的核心思想是从字符级别开始不断合并出现频率最高的相邻对直到达到目标词表大小。实现的时候分两步训练阶段统计所有相邻字符对的频率每次合并频率最高的一对记录合并规则。重复这个过程直到词表达到目标大小。编码阶段把新文本按字符切开然后按训练时记录的合并规则顺序应用。注意合并顺序很重要必须按训练时的顺序来否则结果不一致。def train_bpe(texts, vocab_size): # 初始化所有字符作为基础词表 vocab set() for text in texts: vocab.update(text) merges [] while len(vocab) vocab_size: # 统计相邻对频率 pairs {} for text in texts: tokens list(text) for i in range(len(tokens) - 1): pair (tokens[i], tokens[i1]) pairs[pair] pairs.get(pair, 0) 1 if not pairs: break # 合并频率最高的一对 best max(pairs, keypairs.get) merges.append(best) vocab.add(best[0] best[1]) # 更新文本 texts [t.replace(best[0] best[1], best[0] best[1]) for t in texts] return vocab, merges这段代码是简化版实际生产用的 tokenizer 还要处理特殊 token、字节级回退、多语言等问题。但核心逻辑就是“统计频率 → 合并 → 重复”。理解了这个你就能明白为什么同一个词在不同上下文里会被切成不同的 token。4.3 第二步用 Rust 实现向量相似度计算向量相似度是检索的核心。最常用的是余弦相似度公式是点积除以模长乘积。用 Rust 实现的时候关键是要利用 SIMD 指令做并行计算。pub fn cosine_similarity(a: [f32], b: [f32]) - f32 { let mut dot 0.0; let mut norm_a 0.0; let mut norm_b 0.0; for i in 0..a.len() { dot a[i] * b[i]; norm_a a[i] * a[i]; norm_b b[i] * b[i]; } dot / (norm_a.sqrt() * norm_b.sqrt()) }这是标量版本编译器在开启优化后会自动向量化。如果你想手动控制可以用std::simd或者packed_simd库。实测下来手动 SIMD 能比标量版本快 3-4 倍但代码复杂度高很多。我的建议是先用标量版本跑通确认是瓶颈之后再优化。提示计算余弦相似度之前如果向量已经归一化模长为 1那点积就等于余弦相似度可以省掉除法和开方。很多 embedding 模型输出的向量本身就是归一化的用之前先确认一下。4.4 第三步TypeScript 编排流式响应流式响应的编排是 TypeScript 层的核心。我用的是 SSEServer-Sent Events因为它比 WebSocket 简单而且天然支持断线重连。服务端的核心逻辑是从模型拿到 token 流按句子边界切分通过 SSE 推给客户端。客户端的核心逻辑是接收 SSE 事件拼接分片渲染到界面。// 服务端SSE 推送 async function streamResponse(prompt: string, res: Response) { res.setHeader(Content-Type, text/event-stream); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); const buffer new StreamBuffer(); let chunkId 0; for await (const chunk of modelStream(prompt)) { const text buffer.push(chunk, chunkId); if (text) { res.write(id: ${chunkId}\n); res.write(data: ${JSON.stringify({ text })}\n\n); } } res.write(data: [DONE]\n\n); res.end(); }客户端用EventSource接收注意EventSource会自动重连但重连后是从头发还是从断点发取决于服务端有没有实现Last-Event-ID的处理。我在服务端记录了每个连接的 chunkId重连时根据Last-Event-ID头从下一个分片继续发。4.5 第四步把三层串起来跑通三层串起来的时候最容易出问题的是数据格式的转换。Python 的 numpy 数组、Rust 的 Vec 、TypeScript 的 Float32Array三者之间的转换要特别小心字节序和内存布局。我的做法是统一用 JSON 做跨语言的数据交换虽然性能有损耗但调试方便。等跑通之后再考虑用二进制格式优化。具体流程是Python 侧把文本转成 embedding序列化成 JSON 数组。Rust 侧接收 JSON反序列化成 Vec 做检索返回结果的 ID 和分数。TypeScript 侧接收检索结果组装上下文调用模型流式返回。这个链路跑通一次之后后面就是优化性能了。我建议先用小数据量几百条向量跑通确认逻辑没问题再逐步加数据量测试性能瓶颈。5. 常见问题与排查技巧实录5.1 跨语言调用的典型坑跨语言调用是这类项目最容易出问题的地方。我整理了几个高频问题和排查方法问题现象可能原因排查方法Python 导入 Rust 模块报 ImportError扩展模块名不对或路径不在 sys.path检查 .so/.pyd 文件名是否和 import 名一致调用时崩溃无报错内存越界或空指针用 Rust 的 panic 捕获加日志返回的数组全是 0字节序或类型不匹配打印原始字节确认 float32 布局性能比纯 Python 还慢每次调用都重新初始化把初始化逻辑放到模块加载时注意pyo3 的 GIL 处理很容易出错。如果你的 Rust 函数会长时间运行记得释放 GIL否则会阻塞 Python 的其他线程。用py.allow_threads(|| { ... })包住耗时逻辑。5.2 向量检索的精度与速度权衡ANN 检索的召回率调优是个经验活。我的做法是先建一个小规模的测试集比如 1000 条向量用暴力搜索的结果作为 ground truth然后调 ANN 的参数看召回率能到多少。一般来说ef_search 从 50 开始调每次翻倍观察召回率和延迟的变化。如果召回率到 95% 以上、延迟还能接受就停。如果延迟已经很高但召回率还不够那可能是 M 太小图的质量不行需要重建索引。还有一个容易被忽略的点向量归一化。如果你的 embedding 没有归一化余弦相似度和欧氏距离的结果会不一致。检索的时候用哪种距离度量建索引的时候就要用同一种不能混。5.3 流式响应的断线与乱序流式响应在弱网环境下问题特别多。我遇到过的典型场景场景一分片乱序。SSE 是基于 TCP 的理论上不会乱序但如果中间有代理或者负载均衡可能会。解决办法是给每个分片编号客户端按编号排序后再拼接。场景二连接被中间层缓冲。有些反向代理会缓冲响应导致流式变成一次性返回。解决办法是设置X-Accel-Buffering: no头或者用 chunked transfer encoding。场景三客户端消费太慢。如果客户端渲染速度跟不上数据会堆积。解决办法是客户端做节流比如每 50ms 渲染一次而不是每收到一个分片就渲染。5.4 上下文超长的处理上下文超长是 AI 应用最常见的错误之一。模型有最大 token 限制超了就直接报错。排查的时候要分清楚是输入超长还是输出超长。输入超长的解决办法就是前面说的裁剪策略。输出超长的话要设置max_tokens参数并且在提示词里明确要求模型简洁回答。还有一个技巧是如果模型输出被截断可以在下一次请求里把已输出的内容作为前缀让模型接着写。提示不同模型的 token 计算方式不一样中文和英文的 token 比例也不同。估算的时候中文大概 1 个字 1-2 个 token英文大概 1 个词 1-2 个 token。实际用的时候要用对应模型的 tokenizer 精确计算别靠估算。5.5 性能瓶颈的定位方法AI 应用的性能瓶颈通常在这几个地方embedding 计算、向量检索、模型推理、网络传输。定位的方法是分段计时。我在每一层都加了计时日志记录每个环节的耗时。跑一批请求之后看哪个环节的 P99 延迟最高就优化哪个。实测下来向量检索和模型推理通常是大头embedding 计算如果做了缓存耗时可以忽略。优化的优先级是先加缓存再换算法最后才考虑换语言重写。缓存能解决 80% 的性能问题因为很多请求的输入是重复的。embedding 缓存、检索结果缓存、甚至模型输出缓存都能大幅降低延迟和成本。6. 我在这个项目里踩过的坑和总结的经验最后说几个我觉得最有价值的经验都是实际动手之后才明白的。第一别一上来就追求性能。我一开始就想用 Rust 把所有东西都重写一遍结果卡在跨语言调用上花了两周。后来改成先用 Python 跑通逻辑确认没问题再用 Rust 替换热点效率高多了。正确的顺序是先跑通再优化最后才重写。第二类型定义要前置。三种语言之间传数据最容易出问题的就是类型不匹配。我的做法是先定义好 JSON schema三种语言都按这个 schema 生成类型定义这样编译期就能发现大部分问题。TypeScript 的 interface、Python 的 dataclass、Rust 的 struct三者要一一对应。第三日志要打全。AI 应用的调试比普通应用难因为中间有模型这个不确定因素。我在每个环节都打了结构化日志包括输入、输出、耗时、token 数。出问题的时候看日志就能定位到是哪一层的问题不用瞎猜。第四测试要用真实数据。我一开始用构造的假数据测试跑得好好的一上真实数据就崩。真实数据的分布、长度、语言混合程度都和假数据不一样。建议尽早用真实数据做端到端测试别等到最后才发现问题。这个项目后续还可以往几个方向扩展加一个简单的模型推理层把本地小模型也接进来加一个可观测性面板实时看各层延迟和成本加一个 A/B 测试框架对比不同检索策略的效果。但这些都是后话了先把核心链路跑通、跑稳比什么都重要。