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, fInvalid 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_asciiFalse).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! Im ai-shell — built from scratch. else: reply fYou 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_okTrue) 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_asciiFalse, indent2) 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_asciiFalse, indent2) os.replace(temp_path, session_path)4.1 为什么文件锁比Redis更可靠无依赖Linux内核原生支持无需额外进程或网络。强一致性flock在单机上提供严格的读写互斥比Redis的SETNXEXPIRE组合更不易出错。故障自愈进程崩溃后锁自动释放无Redis的redis-cli shutdown残留问题。但文件锁有其局限仅限单机部署。我们接受这个约束并将其转化为架构优势——通过反向代理如Nginx的ip_hash策略确保同一用户请求始终路由到同一台服务器形成天然的“会话亲和性”。这反而简化了水平扩展逻辑扩容只需增加机器无需担心会话同步。4.2 文件存储的性能实测与优化质疑最多的是性能。我们在4核服务器上用ab压测100并发持续1分钟平均延迟23ms99分位延迟41ms500并发平均延迟38ms99分位延迟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(seconds30): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): def timeout_handler(signum, frame): raise TimeoutError(fFunction {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(seconds25) # 留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-40274000系为模型层错误27为具体子类超时layer标明错误发生层级network/model/prompt/sessioncontext提供复现所需全部参数无需翻日志这套机制让故障排查时间从小时级降至分钟级。上周一次线上事故运维同事根据AEC码直接定位到GPU显存不足更换实例类型后5分钟恢复——全程无需开发介入。6. 模型集成不碰CUDA用标准HTTP API对接本地模型“From Scratch”不等于拒绝生态。我们的目标是可控地集成而非重复造轮子。对于模型推理我们选择绕过PyTorch/CUDA的复杂部署直接对接Hugging Face TGIText 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, jsontgi_payload, timeout(5, 60) # connect5s, read60s ) 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: fTGI 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_responseai-shell主逻辑完全不动。这正是工程化的核心——通过清晰的边界让变化局部化。7. 生产就绪用12个检查项完成AI服务的“出厂质检”一个能跑通Demo的服务离生产就绪还有巨大鸿沟。我们制定了一套12项“AI服务出厂质检清单”每次发布前必须全员签字确认序号检查项验证方式不通过后果1HTTP状态码规范curl -I所有端点确认4xx/5xx返回正确code拒绝上线修复协议层2错误响应JSON结构发送非法JSON检查返回是否含{error: ..., aec_code: ...}拒绝上线修复错误包装3Token计数准确性用相同输入对比transformerstokenizer与服务内计数器差异1 token需校准4会话持久性启动服务→存session→重启服务→读session验证数据不丢失降级为内存存储暂缓上线5熔断有效性timeout装饰器下故意sleep(35)确认返回504修复熔断逻辑重新测试6降级策略覆盖率模拟TGI宕机、模型超时、输出乱码验证各降级路径补充降级规则重新测试7AEC码唯一性检查所有错误分支是否分配唯一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——提醒他这里没有魔法只有人写的代码和人踩过的坑。