从零构建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 的reqwesttokioserde组合能让我精确控制连接池、超时、并发而且编译期就能排除掉一大批空指针和生命周期问题。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是TrueCONTEXT_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( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max20), retryretry_if_exception(is_retryable), reraiseTrue, ) async def call_llm(config: LLMConfig, messages: list[dict]) - dict: async with httpx.AsyncClient(timeoutconfig.timeout) as client: try: resp await client.post( f{config.base_url}/chat/completions, headers{Authorization: fBearer {config.api_key}}, json{model: config.model, messages: messages}, ) except httpx.TimeoutException as e: raise LLMError(ErrorKind.TIMEOUT, str(e), retryableTrue) except httpx.ConnectError as e: raise LLMError(ErrorKind.SERVER_ERROR, str(e), retryableTrue) if resp.status_code 429: raise LLMError(ErrorKind.RATE_LIMIT, resp.text, retryableTrue) if resp.status_code 500: raise LLMError(ErrorKind.SERVER_ERROR, resp.text, retryableTrue) if resp.status_code 400 and context in resp.text.lower(): raise LLMError(ErrorKind.CONTEXT_TOO_LONG, resp.text, retryableFalse) resp.raise_for_status() return resp.json()注意httpx.AsyncClient每次调用都新建是有性能代价的生产环境应该复用。我这里为了代码清晰先这样写后面观测层会讲怎么改成连接池复用。3.2 TypeScript 版本用 zod 把契约钉死TS 版本我用原生fetch因为 Node 18 之后fetch已经稳定没必要引入axios。校验用zod这是整个 TS 版本里最重要的一个依赖因为它把“模型返回的 JSON 结构”从口头约定变成了运行时断言。先定义 schemaimport { 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.infertypeof CompletionSchema;这里有个细节值得说finish_reason我用了z.string()而不是枚举因为不同服务商的取值不完全一致用枚举会导致解析失败。这是踩过坑之后的妥协校验要严格但不能严格到把合法响应拒之门外。错误分类用 discriminated unionexport 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[] ): PromiseCompletion { 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 版本用reqwesttokioserde。Rust 的好处是这些配置在类型层面就强制你想清楚reqwest::Client一旦构建就带着连接池和超时策略复用它是零成本的。use reqwest::Client; use serde::{Deserialize, Serialize}; use std::time::Duration; #[derive(Debug, Clone)] pub enum LLMError { Timeout, RateLimit { retry_after: Optionu64 }, ContextTooLong, BadOutput(String), ServerError(String), } impl LLMError { pub fn retryable(self) - bool { matches!(self, LLMError::Timeout | LLMError::RateLimit { .. } | LLMError::ServerError(_)) } } #[derive(Serialize)] struct ChatRequesta { model: a str, messages: a [Message], } #[derive(Deserialize)] struct ChatResponse { choices: VecChoice, 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]) - ResultChatResponse, 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[] ): AsyncGeneratorstring { 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 用structlogTS 用pinoRust 用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], ) - ResultChatResponse, 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 排查速查表现象可能原因排查动作首 token 延迟高但总延迟正常模型排队或 prompt 过长检查 prompt token 数考虑精简系统提示词随机丢字符SSE 缓冲处理有 bug检查是否保留了不完整的行429 频繁触发并发过高或重试无抖动降低并发加随机抖动成本超预期历史消息重复计费检查上下文截断逻辑统计每轮 token输出格式不符模型版本变更或 prompt 漂移加运行时校验记录原始输出多实例状态不一致用了内存存储迁移到 Redis 或数据库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 工程的失败路径比成功路径多得多把失败路径覆盖好了成功路径自然就稳了。这个习惯让我在后面的项目里少熬了很多夜。