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

1. 从零搭建AI工程能力：为什么我选择用三种语言重走一遍基础链路

第一次看到ai-engineering-from-scratch这个项目名的时候，我脑子里蹦出来的不是“又一个教程仓库”，而是过去两年带团队时反复遇到的一个尴尬场景：面试者简历上写着“熟悉AI应用开发”，让他手写一个带重试和超时控制的推理请求封装，写出来的代码里try/except把KeyboardInterrupt也一起吞了；让他解释一下 token 是怎么被计费的，他说“调API的时候它自己算的”。这不是个例，这是整个行业在“AI应用层”快速膨胀之后留下的基本功断层。

ai-engineering-from-scratch这个标题里的关键词是from scratch，不是 from tutorial，也不是 from framework。它要解决的核心问题是：当 LangChain、LlamaIndex、各种 Agent 框架把一切都封装好之后，一个工程师到底还剩下多少对底层链路的掌控力？我的判断是，如果你只会调框架，那你做的是“配置工作”，不是“工程”。配置工作会被下一个更厚的封装替代，工程能力不会。

这个项目适合三类人：第一类是刚学完 Python 基础、想进入 AI 应用领域但不想只当“调包侠”的初学者；第二类是有后端经验、想搞清楚 AI 服务在生产环境里到底多脆弱的转型工程师；第三类是我这种带团队的人，需要一套可复现的、跨语言的最小实现来给新人做能力校准。它不教你训练大模型，那是另一条路，它教你的是：把一个大模型能力接进真实系统，中间要填多少坑。

我打算用 Python、TypeScript、Rust 三种语言各走一遍同一条链路，不是为了炫技，而是因为这三门语言在 AI 工程里扮演的角色完全不同。Python 是原型和数据处理的主场，TypeScript 是前端和边缘层交互的主力，Rust 则是当你要把推理延迟压到毫秒级、或者要把服务部署到资源受限环境时的答案。同一件事用三种语言做，你会被迫理解哪些是语言特性，哪些是工程本质。

2. 整体设计思路：一条最小可用链路拆成四层

2.1 为什么是“四层”而不是“一个Demo”

我见过太多所谓的 AI 入门项目，本质上就是一个requests.post加上print。这种代码跑通一次就结束了，学不到任何东西。真正有价值的 from scratch 项目，必须把一条完整链路拆成可独立验证的层，每一层都能单独测试、单独替换。

我的设计是四层：接入层、编排层、状态层、观测层。接入层负责和模型服务通信，处理认证、重试、超时、流式解析；编排层负责把用户输入变成模型能吃的 prompt，再把模型输出变成业务能用的结构；状态层负责多轮对话、上下文窗口管理、缓存；观测层负责日志、耗时统计、token 计量、错误分类。

这四层拆开之后，你会发现一个很有意思的现象：大部分线上事故不是出在模型本身，而是出在接入层和状态层。模型返回慢一点用户能忍，但你的重试逻辑把一次请求放大成五次、把账单打爆，用户忍不了。上下文窗口管理没做好，第十轮对话突然报错，用户更忍不了。

2.2 三种语言的分工逻辑

Python 版本我用来做“参考实现”，因为它生态最全，httpx、pydantic、tiktoken这些库能让我把注意力放在链路设计上而不是造轮子。但我会刻意限制自己只用标准库加两三个必要依赖，避免变成框架教学。

TypeScript 版本我用来做“边界验证”。TS 的类型系统在描述 API 契约时比 Python 的 type hint 更严格，尤其是zod这类运行时校验库，能让我把“模型返回的 JSON 到底长什么样”这件事从假设变成断言。而且 TS 版本天然适合跑在 Node 或边缘函数里，是很多 AI 产品前端的真实形态。

Rust 版本我用来做“性能基线”。Rust 的reqwest+tokio+serde组合，能让我精确控制连接池、超时、并发，而且编译期就能排除掉一大批空指针和生命周期问题。Rust 版本跑出来的延迟数字，是其他两个版本的参照系。如果你做的 AI 功能对延迟敏感，比如实时补全、语音交互，Rust 这条路的收益是实打实的。

2.3 一个被忽略的设计决策：错误分类先于错误处理

新手写 AI 调用代码，最常见的写法是一个大try包住所有逻辑，然后except Exception as e: print(e)。这种代码在开发环境能跑，上线必炸。因为 AI 服务的错误不是同质的：有网络层的连接超时，有服务端的 429 限流，有模型层的上下文超长，有业务层的输出格式不符。这四类错误的处理策略完全不同。

连接超时可以重试，429 要退避重试并且看Retry-After头，上下文超长重试一万次也没用必须截断，输出格式不符要重新生成或者降级。所以我在设计阶段就把错误分类做成了第一优先级，每种语言里都定义一个错误枚举，而不是用字符串。这个决策在后面写重试逻辑的时候省了我大量时间。

3. 接入层实操：把一次模型调用写扎实

3.1 Python 版本：httpx 加 tenacity 的最小组合

Python 版本我用httpx而不是requests，原因是httpx原生支持异步和 HTTP/2，而且超时配置更细粒度。依赖只加httpx、tenacity、pydantic三个，tiktoken可选用于 token 估算。

先定义配置和错误类型。配置用pydantic的BaseSettings，这样环境变量能自动加载，避免把密钥硬编码。错误类型我定义成一个枚举加一个异常类：

from enum import Enum from pydantic import BaseModel, Field import httpx class ErrorKind(str, Enum): TIMEOUT = "timeout" RATE_LIMIT = "rate_limit" CONTEXT_TOO_LONG = "context_too_long" BAD_OUTPUT = "bad_output" SERVER_ERROR = "server_error" class LLMError(Exception): def __init__(self, kind: ErrorKind, message: str, retryable: bool): self.kind = kind self.retryable = retryable super().__init__(message) class LLMConfig(BaseModel): base_url: str api_key: str model: str timeout: float = 30.0 max_retries: int = 3

这里的关键决策是retryable字段。它让重试逻辑不需要去猜错误类型，直接读这个布尔值就行。TIMEOUT和RATE_LIMIT是True，CONTEXT_TOO_LONG和BAD_OUTPUT是False。这个设计看起来简单，但它把“什么该重试”这个业务判断从重试框架里剥离出来了，后面换任何重试库都不用改判断逻辑。

接入函数本身用tenacity的retry装饰器，但退避策略要自定义，因为 429 的退避和超时的退避不一样：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception def is_retryable(exc: BaseException) -> bool: return isinstance(exc, LLMError) and exc.retryable @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=20), retry=retry_if_exception(is_retryable), reraise=True, ) async def call_llm(config: LLMConfig, messages: list[dict]) -> dict: async with httpx.AsyncClient(timeout=config.timeout) as client: try: resp = await client.post( f"{config.base_url}/chat/completions", headers={"Authorization": f"Bearer {config.api_key}"}, json={"model": config.model, "messages": messages}, ) except httpx.TimeoutException as e: raise LLMError(ErrorKind.TIMEOUT, str(e), retryable=True) except httpx.ConnectError as e: raise LLMError(ErrorKind.SERVER_ERROR, str(e), retryable=True) if resp.status_code == 429: raise LLMError(ErrorKind.RATE_LIMIT, resp.text, retryable=True) if resp.status_code >= 500: raise LLMError(ErrorKind.SERVER_ERROR, resp.text, retryable=True) if resp.status_code == 400 and "context" in resp.text.lower(): raise LLMError(ErrorKind.CONTEXT_TOO_LONG, resp.text, retryable=False) resp.raise_for_status() return resp.json()

注意：httpx.AsyncClient每次调用都新建是有性能代价的，生产环境应该复用。我这里为了代码清晰先这样写，后面观测层会讲怎么改成连接池复用。

3.2 TypeScript 版本：用 zod 把契约钉死

TS 版本我用原生fetch，因为 Node 18 之后fetch已经稳定，没必要引入axios。校验用zod，这是整个 TS 版本里最重要的一个依赖，因为它把“模型返回的 JSON 结构”从口头约定变成了运行时断言。

先定义 schema：

import { z } from "zod"; export const MessageSchema = z.object({ role: z.enum(["system", "user", "assistant"]), content: z.string(), }); export const CompletionSchema = z.object({ id: z.string(), choices: z.array( z.object({ message: MessageSchema, finish_reason: z.string(), }) ), usage: z.object({ prompt_tokens: z.number(), completion_tokens: z.number(), total_tokens: z.number(), }), }); export type Completion = z.infer<typeof CompletionSchema>;

这里有个细节值得说：finish_reason我用了z.string()而不是枚举，因为不同服务商的取值不完全一致，用枚举会导致解析失败。这是踩过坑之后的妥协，校验要严格，但不能严格到把合法响应拒之门外。

错误分类用 discriminated union：

export type LLMError = | { kind: "timeout"; retryable: true; message: string } | { kind: "rate_limit"; retryable: true; message: string; retryAfter?: number } | { kind: "context_too_long"; retryable: false; message: string } | { kind: "bad_output"; retryable: false; message: string } | { kind: "server_error"; retryable: true; message: string };

TS 的 discriminated union 在这里比 Python 的枚举加异常类更优雅，因为switch的时候编译器会帮你做穷尽性检查，漏掉一个分支直接编译报错。

调用函数用AbortController做超时，这是原生fetch没有超时参数的标准解法：

export async function callLLM( config: LLMConfig, messages: Message[] ): Promise<Completion> { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), config.timeoutMs); try { const resp = await fetch(`${config.baseUrl}/chat/completions`, { method: "POST", headers: { "Content-Type": "application/json", Authorization: `Bearer ${config.apiKey}`, }, body: JSON.stringify({ model: config.model, messages }), signal: controller.signal, }); if (resp.status === 429) { const retryAfter = Number(resp.headers.get("retry-after")) || undefined; throw { kind: "rate_limit", retryable: true, message: await resp.text(), retryAfter }; } if (resp.status >= 500) { throw { kind: "server_error", retryable: true, message: await resp.text() }; } if (!resp.ok) { throw { kind: "bad_output", retryable: false, message: await resp.text() }; } const json = await resp.json(); const parsed = CompletionSchema.safeParse(json); if (!parsed.success) { throw { kind: "bad_output", retryable: false, message: parsed.error.message }; } return parsed.data; } catch (e: any) { if (e.name === "AbortError") { throw { kind: "timeout", retryable: true, message: "request aborted" }; } throw e; } finally { clearTimeout(timer); } }

提示：retry-after头有时候是秒数有时候是 HTTP 日期，生产代码要两种都处理。我这里简化成数字，实际项目里要加日期解析分支。

3.3 Rust 版本：把超时和连接池握在自己手里

Rust 版本用reqwest+tokio+serde。Rust 的好处是这些配置在类型层面就强制你想清楚，reqwest::Client一旦构建就带着连接池和超时策略，复用它是零成本的。

use reqwest::Client; use serde::{Deserialize, Serialize}; use std::time::Duration; #[derive(Debug, Clone)] pub enum LLMError { Timeout, RateLimit { retry_after: Option<u64> }, ContextTooLong, BadOutput(String), ServerError(String), } impl LLMError { pub fn retryable(&self) -> bool { matches!(self, LLMError::Timeout | LLMError::RateLimit { .. } | LLMError::ServerError(_)) } } #[derive(Serialize)] struct ChatRequest<'a> { model: &'a str, messages: &'a [Message], } #[derive(Deserialize)] struct ChatResponse { choices: Vec<Choice>, usage: Usage, } pub struct LLMClient { client: Client, base_url: String, api_key: String, model: String, } impl LLMClient { pub fn new(base_url: String, api_key: String, model: String) -> Self { let client = Client::builder() .timeout(Duration::from_secs(30)) .pool_max_idle_per_host(10) .build() .expect("failed to build client"); Self { client, base_url, api_key, model } } pub async fn call(&self, messages: &[Message]) -> Result<ChatResponse, LLMError> { let resp = self .client .post(format!("{}/chat/completions", self.base_url)) .bearer_auth(&self.api_key) .json(&ChatRequest { model: &self.model, messages }) .send() .await .map_err(|e| if e.is_timeout() { LLMError::Timeout } else { LLMError::ServerError(e.to_string()) })?; let status = resp.status(); if status.as_u16() == 429 { let retry_after = resp .headers() .get("retry-after") .and_then(|v| v.to_str().ok()) .and_then(|s| s.parse().ok()); return Err(LLMError::RateLimit { retry_after }); } if status.is_server_error() { return Err(LLMError::ServerError(resp.text().await.unwrap_or_default())); } if !status.is_success() { let body = resp.text().await.unwrap_or_default(); if body.to_lowercase().contains("context") { return Err(LLMError::ContextTooLong); } return Err(LLMError::BadOutput(body)); } resp.json::<ChatResponse>() .await .map_err(|e| LLMError::BadOutput(e.to_string())) } }

Rust 版本里Client是Clone的，克隆它只是克隆一个Arc，所以可以放心地在多个任务间共享。这是 Rust 生态里一个很舒服的设计，连接池复用不需要你手动管理生命周期。

4. 编排层与状态层：上下文窗口管理才是真功夫

4.1 上下文窗口不是“能塞多少塞多少”

新手最容易犯的错，是把上下文窗口当成一个容量固定的桶，拼命往里塞历史消息，直到报错为止。这种做法有两个问题：一是成本，历史消息每一轮都要重新计费；二是质量，无关的历史会稀释当前指令的权重，模型表现反而下降。

我的做法是分层截断加摘要。把消息分成三类：系统指令（永不截断）、最近 N 轮对话（完整保留）、更早的历史（压缩成摘要）。N 的取值取决于你的场景，客服类场景 N 取 5 到 8 轮，代码助手类场景 N 取 3 到 5 轮，因为代码上下文本身就很长。

Python 版本里我用一个简单的滑动窗口加摘要函数：

def build_context( system: str, history: list[dict], max_tokens: int, keep_recent: int = 6, ) -> list[dict]: recent = history[-keep_recent:] older = history[:-keep_recent] messages = [{"role": "system", "content": system}] if older: summary = summarize(older) messages.append({ "role": "system", "content": f"以下是更早对话的摘要：{summary}", }) messages.extend(recent) return trim_to_budget(messages, max_tokens)

summarize函数本身也是一次模型调用，但它用的是便宜的小模型，而且只在历史超过阈值时才触发。这个设计的关键是摘要的触发条件要基于 token 数而不是轮数，因为一轮对话可能是十个字，也可能是三千字。

4.2 状态层的三种存储选择

状态层要解决的是“多轮对话的状态放哪”。我试过三种方案，各有适用场景。

第一种是内存字典，用会话 ID 做 key。优点是零依赖、零延迟，缺点是进程重启就丢，多实例部署不共享。适合本地开发和单实例小规模服务。

第二种是Redis，用HSET存消息列表，EXPIRE设过期时间。优点是共享、快、支持原子操作，缺点是引入一个外部依赖。这是生产环境最常用的方案，我大部分项目都用这个。

第三种是数据库，用 PostgreSQL 的JSONB字段存消息。优点是持久化、可查询、可审计，缺点是慢。适合需要合规审计或者对话历史要长期保留的场景。

Rust 版本里我用rediscrate 的异步接口，这里有个坑要注意：rediscrate 的连接不是Sync的，多任务共享要用ConnectionManager而不是MultiplexedConnection，后者在高并发下会有 head-of-line blocking。

4.3 流式输出的解析：SSE 没那么简单

流式输出用 Server-Sent Events，格式是data: {...}\n\n。看起来简单，但实际解析的时候有几个坑：一是data:后面可能没有空格，二是最后会有一个data: [DONE]的哨兵，三是网络分片可能把一个事件切成两半。

TypeScript 版本里我用ReadableStream的 reader 手动处理缓冲：

export async function* streamLLM( config: LLMConfig, messages: Message[] ): AsyncGenerator<string> { const resp = await fetch(`${config.baseUrl}/chat/completions`, { method: "POST", headers: { "Content-Type": "application/json", Authorization: `Bearer ${config.apiKey}`, }, body: JSON.stringify({ model: config.model, messages, stream: true }), }); if (!resp.body) throw new Error("no body"); const reader = resp.body.getReader(); const decoder = new TextDecoder(); let buffer = ""; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); const lines = buffer.split("\n"); buffer = lines.pop() ?? ""; for (const line of lines) { const trimmed = line.trim(); if (!trimmed.startsWith("data:")) continue; const payload = trimmed.slice(5).trim(); if (payload === "[DONE]") return; try { const parsed = JSON.parse(payload); const delta = parsed.choices?.[0]?.delta?.content; if (delta) yield delta; } catch { // 分片不完整，等下一个 chunk } } } }

注意：buffer = lines.pop() ?? ""这一行是关键。它把最后一个可能不完整的行留在缓冲区里，等下一个 chunk 拼上再解析。没有这一行，你会随机丢字符，而且这种 bug 极难复现。

5. 观测层：没有度量就没有优化

5.1 三个必须记录的指标

AI 服务的观测和传统服务不太一样，我固定记录三个指标：首 token 延迟、总 token 数、错误分类计数。

首 token 延迟是流式场景下用户体验的核心指标，它和总延迟是两回事。一个请求可能总耗时 8 秒，但首 token 在 300 毫秒就出来了，用户感觉是“快”的。反过来，总耗时 2 秒但首 token 在 1.8 秒才出来，用户感觉是“卡”的。

总 token 数直接对应成本，必须按会话、按用户、按接口维度统计。我见过一个团队因为没统计 token，上线一个月才发现某个功能的成本是预期的二十倍，原因是系统提示词写得太长，每轮都重复计费。

错误分类计数是排查问题的入口。如果rate_limit突然飙升，说明你的并发策略有问题；如果context_too_long飙升，说明截断逻辑有 bug；如果bad_output飙升，说明模型版本变了或者 prompt 需要调整。

5.2 用结构化日志替代 print

三种语言里我都用结构化日志，Python 用structlog，TS 用pino，Rust 用tracing。核心是把每次调用记录成一个 JSON 对象，包含session_id、model、prompt_tokens、completion_tokens、first_token_ms、total_ms、error_kind。

Rust 的tracing在这里体验最好，因为它的 span 机制能自动把嵌套调用的上下文串起来：

use tracing::{info, instrument}; #[instrument(skip(self, messages), fields(session_id = %session_id))] pub async fn call_with_trace( &self, session_id: &str, messages: &[Message], ) -> Result<ChatResponse, LLMError> { let start = std::time::Instant::now(); let result = self.call(messages).await; let elapsed = start.elapsed().as_millis(); match &result { Ok(resp) => info!( prompt_tokens = resp.usage.prompt_tokens, completion_tokens = resp.usage.completion_tokens, total_ms = elapsed, "llm call succeeded" ), Err(e) => info!( error_kind = ?e, total_ms = elapsed, "llm call failed" ), } result }

#[instrument]宏会自动记录函数进入和退出，fields里声明的字段会附加到所有日志上。这种能力在排查跨服务问题时价值巨大，因为你不用手动传递 trace ID。

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

6.1 那些让我熬夜的坑

坑一：重试放大了限流。早期我用固定间隔重试，结果 429 的时候所有请求同时重试，把服务商彻底打挂，触发了更长时间的封禁。后来改成指数退避加随机抖动，抖动范围是退避时间的正负 20%。这个改动之后，限流恢复时间从平均 40 秒降到 8 秒。

坑二：超时设置比上游短。我把客户端超时设成 30 秒，但服务商的网关超时是 60 秒。结果是我的客户端先超时断开，但服务端还在处理，请求被计费了但我没拿到结果。后来我把客户端超时设成比上游长 5 秒，让上游先超时，我这边能拿到明确的错误响应。

坑三：JSON 解析吞掉了错误。有一次模型返回的 JSON 里有个字段是null，我的 TS 代码用z.string()校验直接抛错，但错误被上层一个宽泛的catch吞了，用户看到的是“服务异常”。后来我把所有catch都改成先判断错误类型再处理，bad_output单独走降级逻辑。

坑四：上下文摘要本身超长。摘要函数用模型生成，但摘要的输入是完整历史，如果历史本身超长，摘要调用也会失败。后来我改成先按 token 截断再摘要，摘要的输入永远不超过窗口的 30%。

6.2 排查速查表

6.3 三条我反复验证的经验

第一条，永远给模型调用加一个“熔断开关”。当错误率超过阈值时，直接返回降级结果，不要再打模型服务。这个开关救过我两次，一次是服务商故障，一次是我自己的代码 bug 导致死循环调用。

第二条，prompt 要版本化。把 prompt 当成代码一样管理，每次修改都记录版本号和变更原因。我见过太多团队 prompt 改来改去，出了问题不知道是哪次改的。

第三条，本地要有一个 mock 服务。不要每次都打真实 API，开发阶段用 mock 返回固定响应，测试阶段用 mock 模拟各种错误。这样既省钱又快，而且能覆盖真实服务很难触发的边界情况。

7. 三种语言版本的取舍与组合建议

走完三遍之后，我对三种语言的定位更清晰了。Python 适合做原型验证和数据处理，它的生态让你能快速试错，但部署到生产要注意 GIL 和异步的坑。TypeScript 适合做前端集成和边缘计算，它的类型系统在描述 API 契约时优势明显，但要注意 Node 的版本兼容性。Rust 适合做性能敏感的核心服务，它的编译期检查能排除大量运行时错误，但开发速度确实慢。

实际项目里我经常组合使用：用 Python 做数据预处理和 prompt 实验，用 TypeScript 做前端和 BFF 层，用 Rust 做高并发的推理网关。三者之间通过 HTTP 或消息队列通信，各自发挥优势。

如果你只想学一门，我的建议是先把 Python 版本吃透，因为它的概念最通用，迁移到其他语言只是语法问题。如果你已经有 Python 基础想进阶，那就用 Rust 重写一遍接入层，你会对超时、连接池、错误处理有全新的理解。

最后分享一个我在这个项目里养成的习惯：每写一个功能，先写它的失败路径测试，再写成功路径。AI 工程的失败路径比成功路径多得多，把失败路径覆盖好了，成功路径自然就稳了。这个习惯让我在后面的项目里少熬了很多夜。