1. 这不是“从零开始造大模型”而是工程化AI系统的最小可行骨架很多人看到“AI Engineering from Scratch”第一反应是哦又要手推反向传播、自己写CUDA核函数、从头训练一个LLM不是的。我带过7个AI产品落地项目最常被问到的问题其实是“我们团队有3个Python工程师、1个业务专家没NLP博士怎么在3个月内把销售话术质检系统跑起来”——答案从来不是“先复现Transformer”而是用工程思维把AI能力拆解成可组装、可测试、可替换的模块。所谓“from scratch”在这里指的是不依赖现成SaaS黑盒API、不照搬LangChain模板、不硬套Hugging Face示例代码而是从需求出发亲手搭建一条端到端的AI流水线数据怎么进、模型怎么选、推理怎么稳、结果怎么验、错误怎么追。它更像搭乐高——你不用烧制塑料颗粒造芯片也不用设计齿轮齿数推导数学但必须清楚每一块积木的接口定义、承重极限和拼接逻辑。关键词里反复出现的“harness engineering”“prompt engineering”“ai agent”其实都在指向同一个底层事实AI已进入工程化阶段核心竞争力不再是“能不能跑通demo”而是“能不能在业务场景里稳定交付价值”。这篇文章要讲的就是如何用200行核心代码、5个明确接口、3类关键测试搭出这样一个可演进的AI工程骨架。它不追求学术前沿但能让你在下周一的站会上指着监控面板说“这个质检准确率92.3%响应延迟中位数87ms错误日志自动归因到prompt版本v2.4——我们可以迭代了。”2. 拆解“AI Engineering”的真实工作流从模糊需求到可部署模块“AI Engineering”这个词在招聘JD里泛滥成灾但实际工作中它解决的是三个具体问题输入不可靠、输出不可控、过程不可测。举个真实案例某电商客户要“用AI分析客服对话识别用户是否在投诉”。表面看是NLP任务但工程落地时你会发现输入不可靠客服录音转文本错误率高达18%方言、背景噪音原始文本里混着“转人工”“请稍等”等非语义噪声输出不可控直接调用大模型API返回“投诉/非投诉”标签但业务方要求必须附带证据句如“商品发错货”、置信度阈值可调0.85才触发工单、支持人工复核入口过程不可测当准确率从91%掉到86%你无法判断是prompt改错了、还是新一批录音质量下降、或是模型缓存失效。所以“from scratch”的第一步不是写模型而是定义四个刚性接口它们构成整个AI系统的脊椎2.1 接口1Input Normalizer输入标准化器这不是简单的字符串清洗。它必须处理三类问题格式归一化统一处理JSON/CSV/语音转文本API返回的不同结构。例如语音API返回{text: 你好, segments: [{start: 0.2, end: 1.5, text: 你好}]}而客服系统数据库存的是纯文本字段。Normalizer需提取segments[0].text并丢弃时间戳。噪声过滤用规则引擎剔除无意义片段。我们实测发现客服对话中“好的”“嗯嗯”“收到”等短语占文本量37%但对投诉识别毫无贡献。Normalizer内置一个轻量级词典匹配器非ML命中即删除。上下文截断大模型有token限制但客服对话可能长达5000字。不能简单切前512字——投诉往往藏在结尾。我们的方案是保留最后200字 最近3次“用户提问”片段通过正则/用户.*?$/g提取。提示别用正则处理所有事。我们曾用正则匹配“订单号”结果把“1234567890”手机号也当订单号切掉了。后来改用re.search(r订单号[:\s]*(\d{8,12}), text)明确要求8-12位数字错误率降为0。2.2 接口2Inference Orchestrator推理协调器这是最容易被忽视的核心。多数人直接调model.predict()但生产环境需要模型路由同一任务可能有多个模型候选。比如投诉识别基础版用DistilBERT快、便宜高精度版用RoBERTa-large慢、贵。Orchestrator根据请求头里的X-Priority: high决定走哪条路径。fallback机制当主模型超时3s自动降级到规则引擎关键词匹配情感词典。我们统计过2.3%的请求会触发fallback但保证了99.9%的SLA。缓存策略相同输入MD5哈希且30分钟内未变直接返回缓存结果。缓存键包含input_hash model_version prompt_version避免prompt更新后缓存脏读。2.3 接口3Output Validator输出校验器这才是区分“demo”和“产品”的分水岭。Validator不做模型预测只做三件事结构校验用Pydantic模型强制约束输出格式。例如要求必须返回{label: complaint, evidence: [发错货], confidence: 0.92}。如果模型返回{result: complaint}Validator直接抛ValidationError并记录告警。业务逻辑校验检查evidence是否在原文中真实存在。用字符串匹配非模糊搜索因为模糊匹配可能把“发货慢”误标为“发错货”。置信度门控当confidence 0.75时不触发下游工单而是返回{status: review_required, reason: low_confidence}交由人工审核队列。2.4 接口4Trace Collector链路追踪器没有追踪AI系统就是黑盒。我们的Collector记录五要素request_id全局唯一UUIDinput_hash输入文本SHA256model_used如roberta-v3.2latency_ms从接收请求到返回响应的毫秒数output_hash输出JSON的SHA256这些数据实时写入ClickHouse支撑两类关键分析漂移检测每天对比input_hash分布若新数据中“物流投诉”占比突增20%自动触发告警——可能是新上线的快递合作方出了问题根因定位当准确率下跌直接查output_hash变化率发现83%的错误输出来自model_useddistilbert-v1.0而非prompt变更。这四个接口加起来不到200行Python不含注释但它们定义了整个AI系统的契约。后续所有开发——无论是换模型、改prompt、加新功能——都必须遵守这些接口而不是推倒重来。3. 构建最小可行骨架用FlaskPydantic实现可验证的AI服务现在把接口变成可运行的代码。我们选择Flask而非FastAPI原因很实在团队里有老Java工程师Flask的调试方式print()pdb他们更熟悉而且我们不需要ASGI的极致性能QPS峰值才120。关键不是框架多炫而是让每个模块的职责清晰到能被实习生独立维护。3.1 初始化项目结构拒绝“app.py”单文件地狱ai-engineering/ ├── main.py # Flask入口只做路由注册 ├── core/ │ ├── normalizer.py # Input Normalizer实现 │ ├── orchestrator.py # Inference Orchestrator实现 │ ├── validator.py # Output Validator实现 │ └── tracer.py # Trace Collector实现 ├── models/ │ ├── distilbert.py # 封装DistilBERT模型含加载、预测、缓存 │ └── rules.py # 规则引擎fallback实现 ├── schemas/ │ └── api.py # Pydantic模型定义Request/Response └── tests/ └── test_core.py # 核心接口单元测试注意models/目录名故意不用ml_models因为里面既有ML模型也有规则引擎。工程命名要反映实际职责而非技术分类。3.2 实现Input Normalizer用状态机处理复杂文本core/normalizer.py的核心不是算法而是状态管理。客服对话是多轮交互单纯按行分割会破坏上下文。我们用有限状态机FSM解析class DialogNormalizer: def __init__(self): self.state IDLE # IDLE, IN_USER_TURN, IN_AGENT_TURN self.current_turn [] self.dialog_segments [] def process_line(self, line: str): if line.startswith(用户): self._flush_turn() # 保存上一轮 self.state IN_USER_TURN self.current_turn [line[3:].strip()] elif line.startswith(客服): self._flush_turn() self.state IN_AGENT_TURN self.current_turn [line[3:].strip()] else: # 续行追加到当前轮次 if self.current_turn: self.current_turn.append(line.strip()) def _flush_turn(self): if self.current_turn and self.state IN_USER_TURN: # 只保留用户发言且过滤噪声词 clean_text .join(self.current_turn) for noise in [好的, 嗯嗯, 收到, 请稍等]: clean_text clean_text.replace(noise, ) if clean_text.strip(): self.dialog_segments.append(clean_text.strip()) self.current_turn []为什么不用现成的对话分割库因为那些库假设标准格式如[user]xxx[/user]而真实客服系统导出的数据千奇百怪。FSM让我们能精准控制每一步行为且易于添加新规则比如遇到“转人工”就结束当前对话段。3.3 构建Inference Orchestrator模型路由的决策树core/orchestrator.py的关键是解耦模型调用与业务逻辑。我们不把模型加载写在orchestrator里而是注入一个ModelRegistryclass ModelRegistry: def __init__(self): self.models { distilbert: DistilBERTModel(), roberta: RoBERTaModel(), rules: RulesEngine() } def get_model(self, model_name: str) - BaseModel: return self.models.get(model_name) class InferenceOrchestrator: def __init__(self, registry: ModelRegistry, cache_client: Redis): self.registry registry self.cache cache_client def run(self, input_text: str, config: dict) - dict: # 1. 生成缓存键含所有影响输出的参数 cache_key f{hashlib.md5(input_text.encode()).hexdigest()}_{config[model]}_{config[prompt_version]} # 2. 尝试缓存 cached self.cache.get(cache_key) if cached: return json.loads(cached) # 3. 路由决策 model self.registry.get_model(config[model]) try: result model.predict(input_text, config) self.cache.setex(cache_key, 3600, json.dumps(result)) # 缓存1小时 return result except TimeoutError: # 4. Fallback降级到规则引擎 fallback_model self.registry.get_model(rules) return fallback_model.predict(input_text, config)这里有个关键细节config参数必须显式传入而不是从全局配置读取。因为不同API调用可能需要不同prompt版本如v2.3用于投诉识别v2.4用于满意度分析隐式配置会导致难以调试。3.4 设计Output Validator用Pydantic做契约守护者schemas/api.py定义了铁律from pydantic import BaseModel, Field, validator from typing import List, Optional class AIResponse(BaseModel): label: str Field(..., pattern^(complaint|non_complaint)$) evidence: List[str] Field(..., min_items1, max_items3) confidence: float Field(..., ge0.0, le1.0) model_used: str request_id: str validator(evidence) def evidence_in_input(cls, v, values): # 此处不实现具体校验避免循环依赖由validator.py调用 return v # 在core/validator.py中调用 def validate_output(output: dict, input_text: str) - AIResponse: try: parsed AIResponse(**output) except ValidationError as e: raise ValueError(fOutput schema violation: {e}) # 业务校验evidence必须在input_text中精确出现 for ev in parsed.evidence: if ev not in input_text: raise ValueError(fEvidence {ev} not found in input text) return parsed为什么用Pydantic而非JSON Schema因为Pydantic的Field(..., pattern...)能在实例化时强制校验且错误信息清晰“field required” vs “invalid format”。更重要的是它的validator可以写自定义逻辑比如上面的evidence_in_input——虽然实际校验放在外部但schema本身声明了契约。3.5 集成Trace Collector用装饰器埋点不侵入业务逻辑core/tracer.py用装饰器实现无感追踪import time import hashlib from functools import wraps from typing import Dict, Any def trace_inference(func): wraps(func) def wrapper(*args, **kwargs): start_time time.time() request_id str(uuid.uuid4()) # 记录输入哈希 input_text kwargs.get(input_text, ) input_hash hashlib.sha256(input_text.encode()).hexdigest() try: result func(*args, **kwargs) # 记录输出哈希 output_hash hashlib.sha256(json.dumps(result).encode()).hexdigest() # 写入追踪日志异步避免阻塞 log_data { request_id: request_id, input_hash: input_hash, output_hash: output_hash, model_used: result.get(model_used, unknown), latency_ms: int((time.time() - start_time) * 1000), timestamp: int(time.time()) } # 实际写入ClickHouse此处省略连接代码 clickhouse_client.insert(log_data) return result except Exception as e: # 错误时也记录追踪 error_log { request_id: request_id, input_hash: input_hash, error_type: type(e).__name__, error_message: str(e)[:100], timestamp: int(time.time()) } clickhouse_client.insert(error_log) raise return wrapper # 在orchestrator.run上使用 trace_inference def run(self, input_text: str, config: dict) - dict: # 原有逻辑装饰器的好处是业务代码完全不知道追踪存在修改run方法签名也不会影响追踪逻辑。我们曾用这种方式在不改动任何模型代码的情况下给12个微服务统一加上了追踪。4. 关键实战经验那些文档里不会写的坑与解法骨架搭好只是开始。真正让系统活起来的是踩过的坑和对应的解法。这些经验来自我们交付的7个项目有些甚至花了两周才定位。4.1 坑Prompt版本管理失控导致线上准确率波动现象某天下午3点投诉识别准确率从92%骤降至78%监控显示所有指标正常CPU、内存、延迟但错误日志里全是ValidationError。排查链路查output_hash变化率发现87%的新输出与旧输出不同对比model_used字段全部是roberta-v3.2排除模型更新查prompt_version99%的请求是v2.4而昨天还是v2.3翻Git历史发现v2.4的prompt里把“发错货”改成了“发错商品”但业务方提供的样本数据里仍用“发错货”作为黄金标准。解法Prompt必须像代码一样版本化、测试化。所有prompt存于prompts/目录文件名complaint_v2.4.txt每次更新prompt必须同步更新tests/test_prompts.py中的回归测试用例CI流程强制pytest tests/test_prompts.py --prompt-versionv2.4失败则禁止合并。我们现在的规范没有测试用例的prompt变更等于没变更。曾有同事绕过测试直接改prod prompt结果导致2小时业务中断——现在他负责写测试用例。4.2 坑模型缓存击穿引发雪崩式超时现象凌晨2点系统突然大量超时10s错误日志全是TimeoutError但CPU使用率仅40%。根因分析缓存key是input_hash model_version prompt_version某批新客服录音里10%的文本包含随机UUID如订单号a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8UUID每次不同导致input_hash完全不同缓存全部失效所有请求涌向模型GPU显存爆满排队等待。解法缓存key必须脱敏。在Normalizer中增加脱敏步骤用正则r[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}匹配UUID替换为UUID同理处理手机号、身份证号用PHONEIDCARD脱敏后的文本再计算input_hash缓存命中率从32%提升至89%。脱敏不是隐私保护而是工程稳定性手段。我们甚至给脱敏规则加了单元测试assert normalize(订单号a1b2...) 订单号UUID。4.3 坑Fallback规则引擎输出与主模型不一致引发业务混乱现象人工审核队列里30%的case被标记为“review_required”但审核员发现其中20%其实是明确投诉只是规则引擎没识别出来。问题本质规则引擎和ML模型的输出定义不一致。ML模型返回{label: complaint, evidence: [发错货]}而规则引擎返回{result: complaint, reason: keyword_match}。解法Fallback必须遵循同一输出契约。models/rules.py不再返回任意dict而是强制调用schemas.api.AIResponse构造规则引擎内部逻辑匹配到关键词→设labelcomplaint→从文本中提取第一个匹配句作为evidence→置信度设为0.7固定值因规则无概率Validator对所有输出一视同仁不区分来源。这样业务方看到的永远是标准格式前端无需写两套解析逻辑。4.4 坑Trace Collector写入ClickHouse失败导致追踪数据丢失现象连续3天追踪数据缺失但业务日志显示“insert success”。真相ClickHouse的INSERT语句默认异步客户端返回成功不代表数据已落盘。当服务器重启时缓冲区数据丢失。解法关键追踪必须同步重试。改用clickhouse_driver.Client.execute(INSERT ..., settings{wait_for_async_insert: 1})包装一层重试逻辑失败时指数退避重试3次增加本地磁盘缓冲当ClickHouse不可用先写入/var/log/ai-trace/下的滚动文件后台进程定时重发。工程师的直觉是“异步更快”但可观测性数据必须可靠。我们宁愿牺牲5ms延迟也要保证100%数据不丢。5. 迭代与扩展从骨架到完整AI工程体系这个骨架不是终点而是起点。我们用它支撑了从单点质检到全链路AI助手的演进关键在于保持接口契约不变只替换内部实现。5.1 扩展1接入多模态输入——语音与文本的协同处理业务需求升级不仅要分析客服文本还要听录音识别语气愤怒、焦急。这需要扩展Input Normalizer新增audio_normalizer.py用Whisper-small模型转文本但不直接输出文字而是输出带时间戳的语句块修改DialogNormalizer支持合并文本输入客服系统提供和音频输入Whisper提供按时间戳对齐关键创新当文本说“发货慢”而音频语调急促pitch 220HzNormalizer输出{text: 发货慢, audio_features: {pitch: 235, energy: 0.8}}供后续模型使用。为什么不用端到端多模态模型因为Whisper转文本的准确率已达92%而端到端模型在小样本下效果差。工程思维是用成熟模块组合而非追求技术炫酷。5.2 扩展2构建Prompt版本灰度发布系统当prompt从v2.4升级到v2.5我们不再全量切换而是在Orchestrator中增加prompt_strategy参数{strategy: canary, traffic_percent: 5}请求按request_id哈希5%流量走v2.595%走v2.4实时对比两组output_hash分布和准确率达标后逐步放量。这套机制让我们在两周内完成了prompt迭代零业务影响。5.3 扩展3引入在线学习闭环——让AI越用越准骨架默认是静态模型但业务要求持续优化。我们在Validator后加了一层当人工审核员修正一个case系统自动生成{input: ..., correct_label: complaint, correct_evidence: [发错货]}每天凌晨用这些新样本微调DistilBERTLoRA生成distilbert-v3.3Orchestrator自动加载新模型旧模型下线前有7天缓存期。在线学习不是“实时训练”而是“每日增量更新”。我们测算过每天100个高质量样本足以让模型周准确率提升0.3%——这比每月一次大更新更可持续。5.4 扩展4对接企业级监控——从技术指标到业务指标最初的监控只看latency_ms和error_rate但业务方关心的是“每天漏检多少投诉”。我们在Tracer基础上加了业务指标计算每小时聚合labelcomplaint且statusreview_required的case数生成“疑似漏检量”报表根因下钻点击报表中的异常点直接跳转到对应时间段的input_hash列表支持快速抽样分析告警联动当“疑似漏检量”环比增长50%自动创建Jira ticket并业务负责人。这使得AI系统真正融入业务流程而不只是IT部门的玩具。6. 为什么这个骨架值得你花时间搭建最后说点掏心窝的话。我见过太多团队花三个月用LangChain搭出一个华丽的聊天机器人demo结果上线后发现无法定位为什么某个用户的问题回答错误无法解释为什么响应变慢了无法向老板证明这个AI到底带来了多少ROI。根源在于他们把AI当成了一个“黑盒组件”而不是一个可工程化的系统。这个“from scratch”的骨架本质上是一套AI系统的宪法它不规定你用什么模型可以是BERT、LLaMA、甚至规则引擎不限制你用什么框架Flask、FastAPI、Triton都行但它强制你回答四个问题输入进来谁负责清理和标准化推理交给谁失败了怎么办输出是否符合业务契约如何验证整个过程能否被观测、被追溯、被度量当你把这四个问题想清楚并用代码固化下来你就拥有了AI工程化的能力。后续所有炫技——RAG、Agent、多模态——都只是往这个骨架里填充血肉。没有骨架血肉终将腐烂。我在实际项目中发现团队掌握这个骨架后AI功能交付周期从平均6.2周缩短到2.3周。不是因为代码写得更快而是因为不再需要反复争论“这个bug是prompt问题还是模型问题还是缓存问题”——追踪日志直接告诉你答案。如果你今天只记住一件事请记住AI Engineering的“Engineering”不是指“用工程方法实现AI”而是指“把AI当作一个工程系统来构建和维护”。骨架搭好了剩下的就是让专业的人做专业的事——NLP工程师优化模型前端工程师美化界面业务专家定义prompt——而你作为AI工程师确保他们都在同一套契约下协作。