1. 这不是“搭积木”，而是重建AI系统的底层认知框架
“AI Engineering from Scratch”——这个标题乍看像一句技术圈的时髦口号，实则藏着一个被严重低估的真相：当前90%标榜“AI工程化”的团队，其实只是在已有模型、已有平台、已有SDK的缝隙里做缝合与调参。他们用LangChain编排提示词，用LlamaIndex接入向量库，用FastAPI暴露接口，却从没真正问过：如果删掉所有现成框架，仅靠Python标准库、Linux基础工具链和一张白纸，我们还能不能把一个能响应用户提问、调用外部API、记住对话上下文、生成结构化输出的AI服务跑起来？这不是复古怀旧，而是对AI工程本质的一次压力测试。
我去年带一个三人小队做过一次极限验证：不装任何pip包（除了requests和json），不依赖任何LLM托管平台，不使用Docker或Kubernetes，只用一台4核8G的云服务器，从零开始构建一个可对外提供问答服务的AI系统。整个过程耗时17天，其中前5天卡在最基础的环节——如何让一个纯文本模型真正“理解”用户输入的意图边界。我们发现，所谓“工程化”，从来不是堆砌工具，而是持续回答三个问题：数据怎么来、状态怎么存、错误怎么流。这三个问题的答案，决定了你是在写脚本，还是在建系统。
关键词“ai-engineering”常被误读为“用AI做工程”，但真正的含义是“把AI当作一个需要被工程化对待的复杂系统”。它要求你像处理分布式数据库一样思考token缓存，像设计消息队列一样设计prompt流水线，像维护操作系统内核一样管理模型加载与卸载。而“from-scratch”不是拒绝轮子，而是先亲手造一个轮子，再决定要不要换。这正是本文要展开的：不讲LangChain怎么配置，不教如何微调Qwen，而是回到命令行、回到socket、回到sys.stdout，一砖一瓦地垒出AI服务的地基。适合两类人：一类是刚跳出教程陷阱、想看清AI系统全貌的开发者；另一类是技术决策者，需要判断团队是否真具备AI系统级交付能力，而非只会调用API。
2. 从零启动：用最简协议定义AI服务的最小可行单元
很多人以为“从零开始”意味着从训练模型开始，这是最大的认知偏差。真正的起点，是定义一个可交互、可验证、可拆解的最小服务单元。我们把它命名为ai-shell——一个基于HTTP协议、仅依赖Python内置模块的极简AI服务壳。
2.1 为什么选HTTP而不是gRPC或WebSocket？
- 调试友好性：curl就能触发，浏览器F12就能看请求/响应，无需额外客户端。
- 边界清晰性：HTTP天然定义了request/response语义，强制你思考“一次AI交互”的完整生命周期——输入是什么、输出必须包含什么、失败时返回什么。
- 运维兼容性：Nginx、Cloudflare、甚至家用路由器都能直接代理，不依赖特定运行时。
我们第一版ai-shell只有63行代码，核心逻辑如下：
# ai_shell.py import http.server import json import urllib.parse import sys class AIShellHandler(http.server.BaseHTTPRequestHandler): def do_POST(self): # 1. 解析路径：/v1/chat/completions → 提取版本与端点 path_parts = self.path.strip('/').split('/') if len(path_parts) < 3 or path_parts[0] != 'v1' or path_parts[1] != 'chat': self.send_error(404, "Invalid endpoint") return # 2. 读取原始body（不依赖json.loads自动解析，手动控制） content_length = int(self.headers.get('Content-Length', 0)) raw_body = self.rfile.read(content_length) try: # 3. 手动解析JSON，捕获格式错误 req_data = json.loads(raw_body.decode('utf-8')) except json.JSONDecodeError as e: self.send_error(400, f"Invalid JSON: {str(e)}") return # 4. 核心逻辑：这里才是AI服务的“心脏” response = self.handle_chat_completion(req_data) # 5. 构建标准OpenAI-style响应 self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(response, ensure_ascii=False).encode('utf-8')) def handle_chat_completion(self, req_data): # 模拟真实AI处理：提取messages，生成mock响应 messages = req_data.get('messages', []) if not messages: return {"error": "messages required"} # 真实场景下，这里会调用本地模型或远程API # 当前版本只做回声+简单规则 last_msg = messages[-1].get('content', '') if 'hello' in last_msg.lower(): reply = "Hello! I'm ai-shell — built from scratch." else: reply = f"You said: '{last_msg}'. This is a minimal AI service." return { "id": "chatcmpl-" + str(hash(last_msg))[:12], "object": "chat.completion", "created": int(time.time()), "model": "ai-shell-v0.1", "choices": [{ "index": 0, "message": {"role": "assistant", "content": reply}, "finish_reason": "stop" }] } if __name__ == '__main__': server = http.server.HTTPServer(('localhost', 8000), AIShellHandler) print("AI Shell running on http://localhost:8000") server.serve_forever()提示：这段代码刻意避开
flask或fastapi，因为它们隐藏了HTTP协议细节。当你亲手写self.rfile.read()和self.send_header()时，才会真正理解“请求体大小限制”“字符编码陷阱”“头部大小限制”这些在高级框架里被自动处理的痛点。
2.2 为什么坚持“手动解析JSON”而非依赖框架？
- 错误定位精准：当用户发送了非法JSON，你能明确告诉他是第几行第几个字符错了，而不是笼统报“Bad Request”。
- 内存可控：
rfile.read(content_length)避免了流式读取可能引发的OOM，尤其当用户恶意构造超大body时。 - 协议意识强化：HTTP Header里的
Content-Length不是可选项，它是服务健壮性的第一道防线。我们曾在线上环境遇到过因CDN未透传该Header导致的body截断问题，而自研解析器能立即捕获并记录。
实测下来，这个63行的服务在单核CPU上QPS稳定在120+，延迟中位数<15ms。它不智能，但足够“工程化”——每个环节都可监控、可日志、可压测、可替换。这才是AI工程化的起点：先有确定性，再谈智能化。
3. 数据管道：不用LangChain，如何构建可审计的Prompt流水线
当ai-shell能稳定接收请求后，下一个硬骨头是：如何把用户输入的原始文本，变成模型能理解的结构化指令？行业普遍用LangChain的PromptTemplate，但它的黑盒特性让我们在生产环境吃过亏——某次升级后，模板渲染逻辑变更，导致所有带变量的prompt多出一个空格，进而引发模型输出格式错乱，而日志里只显示“模型返回无效JSON”，排查耗时4小时。
于是我们回归本质：Prompt不是魔法字符串，而是一条有输入、有转换、有输出、有校验的数据流水线。
3.1 四层Prompt架构：从原始输入到模型就绪
我们把Prompt生成拆解为四个明确阶段，每个阶段独立可测试：
| 阶段 | 输入 | 处理逻辑 | 输出 | 可观测性 |
|---|---|---|---|---|
| 1. 清洗层 | 原始user input | 去除不可见字符、截断超长文本、标准化换行符 | 清洁文本 | 记录清洗前后长度比 |
| 2. 上下文注入层 | 清洁文本 + session history | 按时间倒序拼接历史消息，添加角色标识 | 带上下文的message list | 记录history token数 |
| 3. 指令编排层 | message list + system prompt | 插入system prompt，按规则格式化（如：`< | user | >{content}< |
| 4. 安全校验层 | 格式化prompt | 检查敏感词、长度超限、嵌套深度、特殊字符比例 | 通过/拒绝 + 拒绝原因 | 拒绝日志含具体违规项 |
关键设计点：
- 所有层均无状态：输入输出严格函数式，便于单元测试。例如清洗层的测试用例：
assert clean_text("\u200b\u200c\u200d hello \t\n\r") == "hello " assert clean_text("a" * 10000) == "a" * 4096 # 截断至4KB - 上下文注入采用滑动窗口而非全量保留：不是简单
history[-5:]，而是按token数动态计算。我们用transformers的AutoTokenizer（仅用于token计数，不加载模型）预估每条消息token数，确保总长度≤模型最大上下文的80%。这避免了固定条数导致的token浪费或溢出。
注意：我们坚持用
transformers的tokenizer而非正则表达式估算，因为中文分词、emoji、标点符号的token占用差异极大。曾用正则粗算，结果某条含10个emoji的消息实际占127 token，远超预估的30 token，导致批量请求失败。
3.2 实战中的“Prompt漂移”问题与应对
上线两周后，我们发现一个隐蔽问题：同一份用户输入，在不同时间段得到的回复略有差异。日志显示模型调用参数完全一致，问题出在系统提示词（system prompt）的动态注入上。原设计根据用户角色（普通用户/管理员）动态插入不同权限说明，但权限判断逻辑依赖外部API，当该API偶发延迟时，系统提示词生成超时，fallback为空字符串，导致模型失去约束。
解决方案不是加重试，而是将Prompt流水线彻底解耦：
- 权限判断提前到请求入口，结果存入请求上下文（context dict）
- 所有Prompt层只读取context，不发起新网络请求
- 新增
prompt_audit中间件，对每个生成的prompt计算MD5，与历史版本比对，异常波动实时告警
这个改动让Prompt一致性从92%提升至99.97%。它印证了一个经验：AI工程中最危险的不是模型不准，而是输入不可控。当你能把Prompt生成过程像数据库事务一样审计、回滚、压测时，才算真正掌控了AI服务的命脉。
4. 状态管理：告别Redis，用文件锁实现跨进程会话一致性
多数AI应用需要记住用户对话历史，行业方案是“Redis存储session”。但我们发现，当服务部署在无Redis的边缘设备（如树莓派）或客户私有云（无中间件审批权）时，这套方案直接失效。更关键的是，Redis引入了新的故障域——连接超时、序列化错误、内存淘汰策略误伤等。
于是我们回归POSIX标准，用文件锁（flock）+ JSON序列化实现轻量级会话存储，核心代码仅47行：
# session_store.py import os import json import time import fcntl from pathlib import Path SESSION_DIR = Path("/tmp/ai_session") def init_session_dir(): SESSION_DIR.mkdir(exist_ok=True) def get_session_path(session_id: str) -> Path: return SESSION_DIR / f"{session_id}.json" def load_session(session_id: str) -> dict: session_path = get_session_path(session_id) if not session_path.exists(): return {"messages": [], "created_at": time.time()} try: with open(session_path, 'r') as f: fcntl.flock(f.fileno(), fcntl.LOCK_SH) # 共享锁 data = json.load(f) fcntl.flock(f.fileno(), fcntl.LOCK_UN) return data except (IOError, json.JSONDecodeError, OSError): return {"messages": [], "created_at": time.time()} def save_session(session_id: str, session_data: dict): session_path = get_session_path(session_id) try: with open(session_path, 'w') as f: fcntl.flock(f.fileno(), fcntl.LOCK_EX) # 排他锁 json.dump(session_data, f, ensure_ascii=False, indent=2) fcntl.flock(f.fileno(), fcntl.LOCK_UN) except (IOError, OSError) as e: # 锁失败时降级为原子写入（风险：并发覆盖） temp_path = session_path.with_suffix('.tmp') with open(temp_path, 'w') as f: json.dump(session_data, f, ensure_ascii=False, indent=2) os.replace(temp_path, session_path)4.1 为什么文件锁比Redis更可靠？
- 无依赖：Linux内核原生支持，无需额外进程或网络。
- 强一致性：
flock在单机上提供严格的读写互斥，比Redis的SETNX+EXPIRE组合更不易出错。 - 故障自愈：进程崩溃后锁自动释放，无Redis的
redis-cli shutdown残留问题。
但文件锁有其局限：仅限单机部署。我们接受这个约束，并将其转化为架构优势——通过反向代理（如Nginx）的ip_hash策略，确保同一用户请求始终路由到同一台服务器，形成天然的“会话亲和性”。这反而简化了水平扩展逻辑：扩容只需增加机器，无需担心会话同步。
4.2 文件存储的性能实测与优化
质疑最多的是性能。我们在4核服务器上用ab压测：
- 100并发，持续1分钟：平均延迟23ms，99分位延迟41ms
- 500并发：平均延迟38ms，99分位延迟127ms（此时磁盘I/O成为瓶颈）
优化手段：
- 写操作异步化：
save_session改为写入内存队列，由单独线程批量刷盘 - 读操作缓存：为高频session ID建立LRU内存缓存（
functools.lru_cache），命中率92% - 文件分片：按session_id哈希值分16个子目录，避免单目录文件过多影响查找
最终在500并发下，99分位延迟降至63ms，磁盘写入量减少70%。这证明：传统方案未必过时，只是需要针对AI场景重新调优。当你的AI服务QPS低于200时，一个精心设计的文件系统，比一套配置不当的Redis集群更稳。
5. 错误治理：构建AI服务的“熔断-降级-归因”三阶防御体系
AI服务最棘手的不是宕机，而是“安静的失败”——模型返回看似合理但事实错误的答案，或以极低概率返回乱码，日志里却只有200 OK。我们曾因一个未捕获的UnicodeEncodeError（模型输出含罕见汉字），导致下游解析器崩溃，而上游服务仍返回200，问题潜伏3天才被用户投诉发现。
因此，我们为ai-shell设计了三层错误防御：
5.1 第一层：协议级熔断（Network Layer）
在HTTP handler最外层加入超时与重试控制：
import signal import functools def timeout(seconds=30): def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): def timeout_handler(signum, frame): raise TimeoutError(f"Function {func.__name__} timed out after {seconds}s") old_handler = signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(seconds) try: result = func(*args, **kwargs) return result finally: signal.alarm(0) signal.signal(signal.SIGALRM, old_handler) return wrapper return decorator # 在handle_chat_completion上装饰 @timeout(seconds=25) # 留5秒给HTTP层处理 def handle_chat_completion(self, req_data): ...提示：
signal.alarm在多线程环境下不安全，因此我们限定此熔断仅用于单进程模式。生产环境改用concurrent.futures.TimeoutError配合ThreadPoolExecutor，原理相同但线程安全。
5.2 第二层：语义级降级（Model Layer）
当模型调用失败或返回异常内容时，不直接抛错，而是启用降级策略：
| 触发条件 | 降级动作 | 用户感知 |
|---|---|---|
| 模型HTTP 5xx | 返回预设FAQ列表（静态JSON） | “暂时无法处理，请稍后再试” |
| 模型返回空字符串 | 用规则引擎生成回复（如：匹配关键词→固定话术） | 正常回复，但略显机械 |
| 模型输出JSON格式错误 | 提取首句作为回复，附加“（AI正在学习中）” | 保持交互连续性 |
关键点：所有降级策略的触发条件必须可配置、可热更新、可灰度。我们用config.json文件存储规则，服务启动时加载，同时监听文件修改事件，实现秒级生效。
5.3 第三层：根因归因（Observability Layer）
最难的是定位“为什么降级”。我们设计了统一错误分类码（AEC Code），每个错误附带结构化元数据：
{ "aec_code": "AEC-4027", "layer": "model", "cause": "llm_timeout", "context": { "model_name": "qwen2-7b", "input_tokens": 1247, "max_new_tokens": 512, "retry_count": 2 } }- AEC-4027：4000系为模型层错误，27为具体子类（超时）
- layer：标明错误发生层级（network/model/prompt/session）
- context：提供复现所需全部参数，无需翻日志
这套机制让故障排查时间从小时级降至分钟级。上周一次线上事故，运维同事根据AEC码直接定位到GPU显存不足，更换实例类型后5分钟恢复——全程无需开发介入。
6. 模型集成：不碰CUDA，用标准HTTP API对接本地模型
“From Scratch”不等于拒绝生态。我们的目标是可控地集成，而非重复造轮子。对于模型推理，我们选择绕过PyTorch/CUDA的复杂部署，直接对接Hugging Face TGI（Text Generation Inference）的HTTP API。
6.1 为什么TGI是当前最优解？
- 零CUDA依赖：TGI Docker镜像已预编译CUDA驱动，宿主机只需安装nvidia-docker，无需配置nvcc、cudnn版本。
- 工业级API：完全兼容OpenAI Chat Completions格式，
ai-shell的handle_chat_completion方法几乎无需修改。 - 弹性扩缩：TGI支持
--num-shard参数，单卡7B模型可切分为2 shard提升吞吐，无需改业务代码。
部署命令一行搞定：
docker run --gpus all -p 8080:80 -v /path/to/model:/data \ ghcr.io/huggingface/text-generation-inference:2.3.0 \ --model-id /data/qwen2-7b-instruct \ --num-shard 2 \ --max-input-length 4096 \ --max-total-tokens 81926.2 安全网关：在ai-shell与TGI之间加一层适配器
直接调用TGI存在风险：TGI的/generate端点返回格式与OpenAI不完全一致，且错误码混乱（如模型加载失败返回500而非400）。因此我们新增model_gateway.py作为适配层：
# model_gateway.py import requests import json from typing import Dict, Any class TGIAdapter: def __init__(self, tgi_url: str = "http://localhost:8080"): self.tgi_url = tgi_url.rstrip('/') def chat_completion(self, messages: list, **kwargs) -> Dict[str, Any]: # 1. 将OpenAI格式转为TGI格式 tgi_payload = { "inputs": self._format_tgi_input(messages), "parameters": { "max_new_tokens": kwargs.get("max_tokens", 1024), "temperature": kwargs.get("temperature", 0.7), "do_sample": True } } try: resp = requests.post( f"{self.tgi_url}/generate", json=tgi_payload, timeout=(5, 60) # connect=5s, read=60s ) resp.raise_for_status() # 2. 将TGI响应转为OpenAI格式 tgi_resp = resp.json() return self._parse_tgi_response(tgi_resp, messages) except requests.exceptions.Timeout: return {"error": "TGI timeout", "aec_code": "AEC-4011"} except requests.exceptions.RequestException as e: return {"error": f"TGI request failed: {str(e)}", "aec_code": "AEC-4012"} def _format_tgi_input(self, messages: list) -> str: # Qwen2专用格式："<|im_start|>system\n{system}<|im_end|><|im_start|>user\n{user}<|im_end|><|im_start|>assistant\n" formatted = "" for msg in messages: role = msg["role"] content = msg["content"] if role == "system": formatted += f"<|im_start|>{role}\n{content}<|im_end|>" elif role == "user": formatted += f"<|im_start|>{role}\n{content}<|im_end|>" elif role == "assistant": formatted += f"<|im_start|>{role}\n{content}<|im_end|>" formatted += "<|im_start|>assistant\n" return formatted def _parse_tgi_response(self, tgi_resp: dict, messages: list) -> Dict[str, Any]: generated_text = tgi_resp.get("generated_text", "") # 移除prompt部分，只取assistant回复 if "<|im_start|>assistant\n" in generated_text: reply = generated_text.split("<|im_start|>assistant\n", 1)[1] else: reply = generated_text return { "id": "tgi-" + str(hash(reply))[:10], "object": "chat.completion", "created": int(time.time()), "model": "qwen2-7b-instruct", "choices": [{ "index": 0, "message": {"role": "assistant", "content": reply.strip()}, "finish_reason": "stop" }] }这个适配器的价值在于：将模型供应商的变更成本，隔离在单一模块内。当我们要切换到DeepSeek-VL时，只需重写_format_tgi_input和_parse_tgi_response，ai-shell主逻辑完全不动。这正是工程化的核心——通过清晰的边界，让变化局部化。
7. 生产就绪：用12个检查项完成AI服务的“出厂质检”
一个能跑通Demo的服务，离生产就绪还有巨大鸿沟。我们制定了一套12项“AI服务出厂质检清单”，每次发布前必须全员签字确认：
| 序号 | 检查项 | 验证方式 | 不通过后果 |
|---|---|---|---|
| 1 | HTTP状态码规范 | curl -I所有端点，确认4xx/5xx返回正确code | 拒绝上线，修复协议层 |
| 2 | 错误响应JSON结构 | 发送非法JSON，检查返回是否含{"error": "...", "aec_code": "..."} | 拒绝上线，修复错误包装 |
| 3 | Token计数准确性 | 用相同输入对比transformerstokenizer与服务内计数器 | 差异>1 token需校准 |
| 4 | 会话持久性 | 启动服务→存session→重启服务→读session，验证数据不丢失 | 降级为内存存储，暂缓上线 |
| 5 | 熔断有效性 | timeout装饰器下故意sleep(35)，确认返回504 | 修复熔断逻辑，重新测试 |
| 6 | 降级策略覆盖率 | 模拟TGI宕机、模型超时、输出乱码，验证各降级路径 | 补充降级规则，重新测试 |
| 7 | AEC码唯一性 | 检查所有错误分支是否分配唯一AEC码 | 重新设计错误分类体系 |
| 8 | 日志字段完整性 | 抽样100条日志，确认含session_id,aec_code,input_tokens,output_tokens | 补全日志埋点，重新测试 |
| 9 | 配置热更新 | 修改config.json，验证降级规则秒级生效 | 修复文件监听逻辑 |
| 10 | 压力测试达标 | ab -n 10000 -c 200，确认99分位延迟≤100ms | 优化I/O或扩容 |
| 11 | 安全扫描通过 | bandit扫描无高危漏洞，trivy扫描镜像无CVE | 修复漏洞，重新构建 |
| 12 | 文档完备性 | README含部署命令、配置说明、AEC码表、降级策略说明 | 补充文档，暂缓上线 |
这份清单不是形式主义，而是血泪教训的结晶。第4项（会话持久性）曾让我们在灰度发布时发现：flock在某些NFS文件系统上行为异常，导致会话丢失。我们立即暂停上线，改为在本地SSD挂载专用分区，问题解决后才继续。真正的工程化，是把每一次侥幸，都变成下一次的确定性。
8. 经验沉淀：从“能跑”到“稳跑”的5个关键认知跃迁
做完这个项目，团队经历了五次认知刷新，这些不是技术细节，而是影响长期架构决策的底层思维：
8.1 认知一：AI工程化的首要敌人不是算力，而是不确定性
- 模型输出的随机性、网络延迟的波动性、用户输入的不可预测性，共同构成AI服务的“混沌三要素”。所有工程设计，本质都是在给混沌加约束。比如我们坚持手动解析JSON，不是为了炫技，而是把“输入不确定性”压缩到最小范围——只要JSON合法，后续流程就确定；而框架的自动解析，把不确定性扩散到了整个调用栈。
8.2 认知二：可观测性不是日志，而是结构化归因能力
- 传统日志（如
logger.info("Request processed")）在AI场景下价值有限。我们必须能回答：“为什么这个请求返回了错误答案？”答案不在INFO日志里，而在AEC码、token计数、prompt MD5、模型响应原始体中。我们后来把AEC码扩展为URL可访问的文档页（如/docs/aec-4027），点击即显示该错误的全部上下文、历史发生频率、修复方案，这才是真正的可观测性。
8.3 认知三：状态管理的终极形态，是让状态“可丢弃”
- 我们曾花两周优化Redis集群的高可用，最后发现：只要会话数据能在10秒内重建，用户根本感知不到中断。于是我们重构了会话逻辑——所有非核心状态（如用户偏好设置）允许丢失，核心状态（当前对话）通过客户端重传保障。这让我们敢于在K8s滚动更新时主动kill pod，而不必等待优雅退出。
8.4 认知四：模型集成的最佳实践，是“API契约优先”
- 不管底层是TGI、vLLM还是自研引擎，只要它遵守OpenAI Chat Completions API契约，上层代码就无需修改。我们甚至用
httpx.MockTransport为TGI适配器写单元测试，模拟各种网络异常，确保契约不变性。这比任何CI/CD流水线都更能保障长期稳定性。
8.5 认知五：从零开始的真正价值，是建立技术决策的“锚点”
- 当团队争论“要不要上Redis”时，有人会说：“我们试过文件锁，在200QPS下完全够用，加Redis是为未来，但未来需求还没出现。”这个“锚点”让讨论回归事实，而非技术潮流。现在每次选型，我们都会问：“如果回到第一天，没有这个工具，我们会怎么做？”
最后分享一个小技巧：在ai-shell的根目录放一个VERSION文件，每次发布时手动更新。不是为了CI/CD，而是为了让每个新成员第一次git clone时，看到的第一行代码就是v0.1.0——提醒他，这里没有魔法，只有人写的代码，和人踩过的坑。
