ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

AI工程从零开始:拆解完整链路与落地实践

AI工程从零开始:拆解完整链路与落地实践 最近不管是社区还是工作群大家都在聊 ai-engineering-from-scratch 这种自建学习路线。我自己也被问过很多次“AI 工程从零开始第一步到底学什么”问这话的人往往已经收藏了一堆教程看过 Transformer 结构也调过一些大模型接口但真要让他们把一个想法变成可给别人使用的工具就卡住了。原因不是模型知识不够而是缺少一条完整的工程链路。我在这篇文章里不打算再把注意力放在某个热门模型上而是从一个从业者的角度拆解 AI 工程这个命题它到底在解决什么问题、需要哪些底层认知、怎么从零做一个能跑通的小系统、怎么把它变成服务以及遇到问题时按什么顺序排查。适合正在转型的开发者也适合已有算法基础但没上过生产环境的人。1. 把“AI 工程”拆开来看它不是算法的附属品1.1 工程能力不是写几个模型接口而是让整个系统可交付很多人一听到 AI 工程第一反应是“我要手写一个神经网络”。但实际工作里AI 工程的核心不是从零实现模型结构而是把一个模型能力放到真实环境里让它持续稳定地提供价值。算法工程师关心的是“在评测集上指标提升多少”AI 工程师关心的是“在不可控的输入、波动的流量、不可靠的第三方依赖面前系统还能不能正常响应”。我自己带人时常用一个类比算法研究像做菜谱开发在厨房里把味道调到最好AI 工程像开餐厅要考虑供应链、服务员、翻台率、顾客投诉和厨房着火时怎么办。菜谱只是餐厅的一部分而工程要解决的是整个系统能持续运转的问题。所以你如果真的要“from scratch”就别从构建 Transformer 开始而应该从构建一条完整数据-模型-服务链路开始。这条链路里任何一个环节断了模型能力再强也落不了地。1.2 一条完整链路里的六个必要环节我习惯把 AI 工程链路拆成六个环节从零开始做项目时按这六个环节走一遍比什么都重要需求定义谁在什么场景下用什么样的输入换取什么样的输出。数据工程把原始文本、图片、日志变成可批量消费的标准化数据。模型选型与推理在效果、成本、延迟之间找一个可接受的模型方案。评估定量地知道系统当前是什么水平而不是“看起来还行”。服务化把脚本变成接口处理并发、超时、限流和异常。观测与迭代记录线上真实行为把失败案例变成下一版改进依据。这六个环节并不严格线性。在实际项目里你经常要回到前面的环节去改数据、换模型、调提示词。但第一次做的时候最好完整走一遍哪怕每个环节都做得粗糙。因为你缺的不是某个知识点而是“全局图景”。等你完整做过一遍再回头精修任何一个环节都会比一开始就钻进模型细节要高效得多。1.3 第一个项目怎么选要“全”不要“新”第一个从零开始的 AI 项目选什么方向特别关键。我不建议一上来就做多模态、Agent、语音交互这些听起来很酷的东西也不建议只做一个“用大模型写周报”的提示词玩具。最好的第一个项目是文档问答也就是 RAGRetrieval-Augmented Generation里的最小闭环。原因很朴素文档问答场景中你既要处理真实文档的脏数据要把长文本切成合适片段要构建索引要写检索逻辑要设计生成提示词还要把结果包装成接口给用户用。这刚好覆盖了一条完整工程链路而且不需要你有算法创新能力。数据就用公司的产品说明书或公开的百科条目问题就设计成“怎么申请报销”“XX 功能怎么开启”这类有明确答案的问题。当你把这样一个项目做完你对 AI 工程的理解会超过刷十门课。2. 动工前先把四个底层认知焊死2.1 Python 工程习惯让数据流动可以“看得见”AI 项目里最常见的痛点不是模型效果差而是中间数据一团乱。初学者通常习惯写脚本时把所有东西塞进 dict字段名随手写调着调着就忘了。等到链路一长某个中间变量变成 None或者字段名拼错程序不报错只在最终结果里悄悄变差你排查半天都找不到原因。我踩过这个坑现在每做一个新项目都会先让数据流动“可见”。具体做法有三个一是用 dataclass 定义中间产物比如一个 Chunk 对象包含 doc_id、page、text而不是到处传 dict二是统一用 logging 输出关键节点信息别用 print三是在数据读取和模型调用前后加断言重要字段缺失就直接报错。这些习惯看起来不性感但它们决定了你后期的排错效率。工程化的本质不是用什么高级框架而是让系统每一个环节都有明确的输入、输出和错误边界。2.2 token 与采样模型行为是可以预测的如果你不了解 token你很难做好 AI 工程。模型看到的不是“字符”而是被切分成 token 的符号序列。同样的提示词中文和英文可能消耗不同数量的 token请求参数里限制 max_tokens影响的不只是长度还有成本、延迟和输出的完整性。很多人会在 prompt 里写“请简短回答”然后抱怨模型不听。倒是更可靠的办法是把输出长度直接限制在接口参数里并且尽量要求模型输出结构化内容比如 JSON。采样参数也要理解到位temperature 控制随机性数值越高输出越不稳定。做工程时我建议把温度压低尤其是需要程序解析输出内容的任务。否则你前一脚还在测试“每轮都稳定”后一脚上线就发现偶尔返回了一堆无法解析的废话。模型行为不是玄学是概率分布你要用控制参数的方式去约束它。2.3 向量检索的最小知识距离、归一化、暴力搜索进入 RAG 项目前先别急着上向量数据库。你只需要理解一个很小的核心embedding 把文本变成一串数字数字之间可以算距离距离越近语义越接近。常用度量是余弦相似度它对向量长度不敏感适合文本场景。如果模型输出的向量已经做了归一化那么内积就等于余弦相似度运算上会更快。我第一次做检索时先用 numpy 手写了一个暴力搜索几十个文档算一遍也就几毫秒。这个“笨办法”让我明白了索引的本质把所有向量存成一个矩阵把 query 向量和矩阵做点积排序取 top-k。后来的向量数据库无非是在大规模数据下做了更高效的近邻搜索和过滤能力但核心逻辑并没有变。带着这个认知去用工具你才知道为什么有时候结果不准为什么要在索引里额外存元数据为什么过滤条件会影响召回质量。2.4 日志先行先有排错地图再写功能代码从零开始做项目的时候最忌讳的是“把功能写完了再回来补日志”。你应该在写第一个数据处理函数的时候就把日志打好。比如读取原始文件后记录文件数量和总字符数清洗后记录丢弃了哪些异常样本模型调用前记录输入摘要和当前时间模型返回后记录状态码、耗时和内容长度。有了这些日志出问题时你能很快回答“数据是否进来了、处理是否正确、模型是否返回、下游解析是否失败”。没有日志你只能靠肉眼盯着报错信息猜改一个地方跑一次反复试错。我一般会在项目开始前花 10 分钟搭一个最简单的 logging 配置输出到文件和终端。这 10 分钟在后期的排查阶段通常会帮你省下几个小时。工程能力很大程度就是“在安静的时候先修好路而不是在着火的时候才找灭火器”。3. 从零搭一套文档问答系统从裸数据到可验证 Demo3.1 范围定义输入输出先说清楚这一步看起来最不“技术”但恰恰最影响后续所有开发。我们要做的文档问答系统输入是一个用户问题输出应该包含两部分回答文本和回答所依据的来源片段。为什么要返回来源因为用户需要可信度也因为你后续做问题定位时可以知道答案是来自检索还是来自模型瞎编。我习惯把这个问题写成一句话“给一批产品文档用户用自然语言提问系统返回有依据的回答并标明依据来自哪几个片段。”定义清楚之后目录结构也顺势出来了project/ data/raw_docs/ # 原始文档 data/processed/ # 处理后的切块结果 src/embedder.py # 向量化 src/retriever.py # 检索 src/generator.py # 生成回答 app.py # 服务入口 requirements.txt不一定一上来就按这个目录写但至少脑子里要有一条“数据在哪些环节流动”的路径。否则写到最后你会发现自己把切块、检索、生成全都堆在一个文件里改一处就要牵动全身。3.2 搭建最小环境venv、依赖和目录这一步很基础但很多人会跳过去直接 pip install 到全局环境后面依赖冲突到想哭。我建议每一步都按工程习惯来先建虚拟环境。依赖先保持最小能跑通再加。python -m venv .venv source .venv/bin/activate # Windows 下是 .venv\Scripts\activate pip install sentence-transformers fastapi uvicorn numpy这里选 sentence-transformers 做本地向量化是因为学习阶段不需要申请在线向量化服务的密钥而且本地模型可以离线跑结果可复现。生成部分可以先用一个兼容开放协议的大模型 API 封装成一个函数后续即使换模型也只改这一个函数。依赖越少出问题的面越小这对新手尤其重要。3.3 数据清理与切块第一个真正的坑原始文档一定不是干干净净给你切块的。常见的问题包括全角半角混用、空行过多、表格被导出成乱序文字、PDF 里把一句话断成两行。所以清洗这一步不能省。我自己会先写一个清洗函数把多余空行和不可见字符去掉再按段落拆分。接着才是切块。切块看起来简单实际上很影响检索效果。方案之一是按固定字符数切同时加入重叠。中文可以按字符数处理不容易乱码。代码大概长这样from dataclasses import dataclass dataclass class Chunk: doc_id: str page: int text: str def clean_text(text: str) - str: # 去掉被导入的空白符号再把连续空行压缩成一个换行 lines [line.strip() for line in text.splitlines() if line.strip()] return \n.join(lines) def split_by_chars(text: str, chunk_size: int 256, overlap: int 32) - list[str]: # 按字符数切分保留重叠避免关键信息恰好落在两个块的边界 chunks [] start 0 while start len(text): end start chunk_size chunks.append(text[start:end]) if end len(text): break start end - overlap return chunks为什么要加重叠因为用户问题问到的内容可能正好跨在切块边界上。没有重叠时一个完整句子可能被切成两半两半分别都不像能回答问题的片段。增加 32 或 64 个字符的重叠能明显提升召回稳定性。切块大小也需要测试。256 个字符对中文来说不算长适合大多数说明文档。文档主题比较独立、每段本身较长时可以把 chunk_size 调大主题密集、句子短时调小一点往往效果更好。这不是死参数后面要围绕验证集的数字去调整。3.4 索引与检索用十几行代码理解向量搜索清洗和切块完成之后下一步是把每个 Chunk 文本编码成向量。这里先不引入向量数据库先用 sentence-transformers 生成向量再用 numpy 做暴力检索from sentence_transformers import SentenceTransformer embedder SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) def build_index(chunks: list[Chunk]): texts [c.text for c in chunks] # normalize_embeddingsTrue 表示对向量做归一化后面点积等价于余弦相似度 matrix embedder.encode(texts, normalize_embeddingsTrue) return matrix def search(query: str, matrix, chunks: list[Chunk], top_k: int 5): q_vec embedder.encode([query], normalize_embeddingsTrue)[0] scores matrix q_vec top_indices np.argsort(-scores)[:top_k] return [(chunks[i], float(scores[i])) for i in top_indices]这个版本在小规模文档集上足够快而且没有额外依赖。用内积代替余弦计算前提是向量已经归一化。理解这一点比你直接调 es 或 milvus 更重要因为你知道工具内部在算什么。检索结果里带着 Chunk 对象后续生成阶段要引用来源这个设计从第一步就保留。3.5 生成把检索结果变得有依据检索只是过程生成回答才是用户看到的结果。生成阶段的核心不是让模型“自由发挥”而是让它严格基于检索到的片段回答。我常用的提示词结构是把多段资料拼接成上下文再明确要求模型只能基于上下文回答。def build_prompt(query: str, hits: list[tuple[Chunk, float]]) - str: context_parts [] for i, (chunk, score) in enumerate(hits): context_parts.append(f[{i}] {chunk.text}) context \n---\n.join(context_parts) return ( 请只根据下面的资料回答用户的问题。\n 如果资料中没有答案请直接回答资料中未提到。\n 不要编造。\n\n f资料\n{context}\n\n f问题{query}\n\n 回答 )生成函数是这样的def ask_question(query: str, matrix, chunks: list[Chunk]): hits search(query, matrix, chunks, top_k5) prompt build_prompt(query, hits) answer call_llm(prompt) # 这里封装大模型服务可能来自云端 API 或本地模型 return { answer: answer, sources: [h.chunk.doc_id for h in hits], }“不要编造”并不能完全消除幻觉但能把模型行为约束到安全方向。真正消除幻觉要靠两件事一是检索真的把包含答案的片段找出来二是生成阶段只给少量高相关片段避免无关信息干扰。这也是为什么检索阶段不能草草了事它决定了生成质量的上限。3.6 粗糙但必要的评估用数字代替感觉很多人在这一步会看一眼几个回答“嗯不错”然后就说完成了。不行。工程要求可评估。你至少要做一个二十条问题的小验证集并且尽量覆盖不同类型的提问方式。然后统计一个最简单的指标检索召回率。eval_questions [ (报销流程需要哪些材料, doc_报销.md), (如何修改个人邮箱, doc_账号设置.md), (申请休假的审批需要多久, doc_人事政策.md), ] hit_count 0 for question, gold_doc_id in eval_questions: hits search(question, matrix, chunks, top_k3) if gold_doc_id in [h.chunk.doc_id for h in hits]: hit_count 1 retrieval_recall hit_count / len(eval_questions) print(retrieval recall3:, retrieval_recall)这个指标不完美但你有了一个“锚点”。后面改了切块参数改了 embedding 模型再跑一遍同样的验证集就知道改动是变好还是变坏。这比凭感觉调参数可靠一百倍。等系统上线后验证集还要不断补充线上失败 case它会变成你的回归测试集。4. 从脚本到服务工程化的第二课4.1 用 FastAPI 把功能包成接口脚本能跑和能被使用是两回事。为了让别人能通过 HTTP 调用你的问答系统用 FastAPI 封装一个接口是成本比较低的方式。先定义请求和响应结构from fastapi import FastAPI, HTTPException from pydantic import BaseModel class AskRequest(BaseModel): question: str class AskResponse(BaseModel): answer: str sources: list[str] app FastAPI() app.post(/ask, response_modelAskResponse) def ask(request: AskRequest): if not request.question.strip(): raise HTTPException(status_code400, detailquestion 不能为空) result ask_question(request.question) return AskResponse(**result)这里有个细节接口层不要直接暴露内部数据结构。用户只需要 answer 和 sources不需要知道内部用了什么 embedding、什么检索库。把接口作为稳定的边界内部重构时只要接口不变调用方就不受影响。这是工程化的第一层意义。4.2 外部调用的超时、重试和错误传播一旦把大模型服务作为外部依赖就必须处理超时和错误。最常见的问题是模型接口偶尔响应很慢或者返回 429/503。程序如果只是傻等用户会以为系统坏了。正确做法是给外部调用设置超时并在捕获异常时记录上下文。我建议的最小容错策略设置合理的超时时间比如 30 秒超过就认为失败。对瞬时错误做最多两次重试重试间隔用指数退避。不要把重试用在“请求本身有问题”的场景比如提示词格式错了。最终失败时接口层要返回对应的错误码而不是让一条长 stack trace 直接裸露给用户。尤其要注意的是不要把解析失败也当成重试理由。如果大模型返回的内容不是期望的 JSON你重试同样的 prompt 大概率还是失败。这时候要做的是在提示词里加约束并在代码里 catch 解析异常必要时提示模型重新只输出 JSON最多一次。重试机制要精准乱重试只会放大成本。4.3 并发约束与有限资源下的排队策略从单用户到多用户很多新手会发现系统一下子变得很不稳定。原因通常很直接你的模型服务没法同时处理那么多请求。如果 GPU 只有一个或者 API 配额有限所有请求就会互相争抢资源。工程上的解法是显式控制并发。比如在异步接口里加一个信号量限制同时进入模型推理的请求数import asyncio semaphore asyncio.Semaphore(4) async def handle_ask(question: str): async with semaphore: return await call_llm_async(question)超出并发上限的请求可以排队也可以直接返回 429 让客户端稍后重试。这取决于你的产品形态。如果是内部工具排队通常体验更好如果是对公服务限流更合理。核心原则是系统的吞吐量取决于最弱的一环你需要在入口层就把超过承受能力的流量挡住否则所有请求一起失败体验更差。4.4 上线后的指标埋点别在漆黑里回溯服务上线之后你就不能只在本地打印日志了。你需要记录三类信息流量指标、性能指标、质量指标。流量指标包括请求量、来源、是否拒绝性能指标包括检索耗时、生成耗时、总耗时、超时和重试次数质量指标包括返回的答案长度、用户是否点击了“有帮助/无帮助”、检索来源分布。最简单的埋点就是在接口函数里打结构化日志logger.info( { event: ask_success, question_length: len(question), matches: len(sources), latency_ms: latency_ms, upstream_error: has_error, } )这些日志积累起来你就知道这个系统在真实使用中是什么状态。你会突然发现原来有 30% 的问题集中在某个文档主题上原来某个时间段模型服务特别慢。没有这些数据你后续优化全都靠猜。AI 工程和手艺活最大的区别就在这里每个决定都要有据可依。5. 常见问题排查实录从现象到根因5.1 回答质量差先在检索和生成之间定位用户反馈“回答不对”时不要直接改提示词。先做一次二分定位。把用户问题输入到系统里打印出 top_k 检索片段人肉眼看一下正确答案在不在这些片段里。如果不在问题出在检索侧。你后续动作应该是调整切块、换 embedding、改查询改写。如果在但生成答案还是不对问题出在生成侧。你后续动作应该是改提示词、加强约束或换更强的生成模型。这一步看起来基础但大多数新手都会跳过去直接调 prompt调了半天没效果最后发现检索到的片段根本不包含答案。先定位再动手省掉大量无用功。5.2 检索召回低切块、模型和查询改写三板斧检索结果不理想时我会按三个顺序排查。第一检查切块。如果 chunk 是 1000 个字符一个正确答案被淹没在大量无关内容里向量相似度会被摊薄如果 chunk 太短又会丢失上下文。第二换 embedding 模型。通用模型对特定领域术语、简称、中文口语化表达不一定友好可以对比一个多语言模型或领域微调模型。第三做查询改写。比如用户问“怎么申请报销”检索时直接匹配“报销 申请 流程”可能更准。简单做法是把原问题扩写成几个关键词变体分别检索后再合并结果。但记住每一步改动都要用验证集数据去验证不能只跑一两个 case 就说有效。人工看 3 个例子只能给你直觉给你数字的才是工程。5.3 接口超时与 JSON 解析失败绝大多数不是玄学接口超时先看日志里是发生在“检索”还是“生成”阶段。如果发生在检索阶段问题多半是向量化模型在单机 CPU 上太慢考虑 GPU 推理或给 embedding 模型做缓存。如果发生在生成阶段要去看外部模型服务的响应状态。大量 429 说明你并发打太高或者服务配额不够解决办法是降低并发和启用退避重试而不是把超时时间从 30 秒改成 60 秒那样只会让更多请求挂在队列里。JSON 解析失败也是高频问题。原因通常是提示词要求模型输出 JSON但模型返回前带了说明文字或者 JSON 被截断。工程上要做两层防御一层是提示词明确“只输出 JSON 对象不要解释”另一层是解析失败时用正则把首次出现的{到最后出现的}截取出来再解析。如果还失败才发起一次修复请求。5.4 文档更新后答案没变索引的一致性怎么维护RAG 系统里最容易忽略的问题之一是文档更新后索引还是旧的。用户改了说明书重新提问模型却还在回答旧内容。这个问题如果在已经上线的系统里出现会让用户对系统失去信任。简单解法是给每个文档一个唯一 doc_id并记录其版本号或修改时间。重建索引时先按 doc_id 删除旧 chunk再插入新 chunk。如果改动很大也可以定期全量重建。全量重建期间不能让服务断掉常见做法是先用两个索引副本一个在后台重建完成后切换。这个思路比“每次直接覆盖线上索引”安全得多。你不需要一开始就做得特别复杂但要意识到“数据”和“索引”是两套系统索引必须能跟随数据更新。6. 最后我陪跑下来最想留下的几条硬经验6.1 一次只动一个变量让验证集当裁判我在带人做项目时看到最多的习惯是一次性同时换 embedding 模型、调切块大小、改 prompt然后发现效果变好了却说不清是哪一个改动起了作用。这个习惯很危险因为你无法积累有效经验。正确的做法是一次只改一个变量。改完跑验证集记录结果再决定要不要保留。这个方法慢但每一步都是扎实的。AI 工程的很多改进都是“可叠加”的前提是你知道哪个改动确实有效。把验证集跑一遍的时间很短完全值得你坚持下去。6.2 把失败案例存成回归集别让 Bug 重演系统上线后一定会遇到一批你没有预测到的坏案例。我当时犯过的错误是手动修完就给忘了结果下个版本又出现同样的问题。正确做法是把每一次失败案例收集起来连同它的预期答案一起加入验证集。以后每次改动不仅要跑原始验证集还要跑这些历史失败案例。这个习惯本质上是给系统建立“回归防线”。你不一定能让模型 100% 不犯错但至少你已知的 bug 不能在下个版本里卷土重来。长期下来这个回归集比任何指标都要宝贵因为它记录了系统在真实环境里最真实的长短板。6.3 稳定大于炫技先让外行能用再谈优化从零开始做 AI 工程很容易被各种新框架、新模型吸引想着一上来就上最先进的架构。但我见过太多半成品检索用了最新工具生成用了最新模型结果连“用户提问后能不能稳定返回答案”都没做到。我的建议是先做减法。用最简单的暴力检索用最普通的 prompt 结构把接口、日志、错误处理全部跑通让一个完全不了解你的系统的人也能顺利用起来。确认系统稳定之后再往里面加 fancy 的东西。稳定是一个系统的最低门槛也是 AI 工程里最难练的基本功。你把这个门槛守住了后面所有优化才有意义。
返回列表