1. 为什么人人都该亲手从零搭一套AI工程先说实话现在学习AI的资源已经多到泛滥了。网课、大模型套壳教程、LangChain/LlamaIndex全家桶、铺天盖地的Agent框架……你随便打开一个技术社区都能看到有人教你三分钟跑通一个RAG应用。但问题是跑通一个demo和真正理解AI工程中间隔着一整座山。我见过太多开发者的状态是能调ChatGPT的API能照着文档用LangChain拼一个对话机器人但一旦遇到流式输出卡死、上下文长度爆掉、多轮对话上下文串味、token成本不可控这类工程问题就直接麻爪。根子不在于你不够聪明而在于你一直在用别人搭好的积木却从来没亲手看过积木内部长什么样。ai-engineering-from-scratch这个标题想讲的恰恰就是把所有流行的AI框架全部扔掉从最底层、最原始的HTTP请求开始一点点搭建出一个能干活、能上线、能维护的AI应用。这是一种重新发明轮子式的修行但在我看来它是目前性价比最高的AI工程入门路径——因为框架永远在变而底层原理十年不变。这篇文章我按自己的实操经验拆解了完整的路径从直接裸调模型API到用消息队列自研上下文管理再到做服务封装、测试评估、最终部署。适合那些已经会写Python、但不满足于只会调包、想真正理解AI工程底层逻辑的开发者。读完你会发现所谓AI工程本质上就四件事管好请求、管好上下文、管好成本、管好质量。2. 先拆解核心需求你以为在学AI其实在学系统工程2.1 从零开始到底意味着什么很多人一听到from scratch下意识以为是要从反向传播、手写Transformer开始。不是的。真让你从矩阵乘法开始写大模型那不叫工程那叫学术研究。工程层面的from scratch指的是不依赖任何现成的AI应用框架LangChain那种而是直接使用模型服务商提供的裸API自己动手设计请求格式、自己管理上下文、自己封装应用逻辑、自己处理各种边界情况。我用一个生活化类比来解释这件事假设你要开一家餐厅。直接买预制菜加热上桌那是用LangChain——快但你没有自己的配方无法应对客人个性化的口味需求。完全从种地、养鸡开始那是做学术研究——累死且没人等得起。ai-engineering-from-scratch要的是从买菜、洗菜、切菜、配菜开始——你要自己搞定供应链如何高效调用模型、自己设计菜单如何构造prompt和上下文、自己管理厨房动线如何处理并发请求和流式输出、自己验收出菜质量如何评估模型输出。这种做法的好处是一旦你亲手写过一次裸API调用、自己实现过一次上下文管理再回头用任何高级框架你能一眼看穿它每个方法的背后在做什么出了问题也能顺着原理去排查而不是只能上网搜LangChain报错XXX怎么办。2.2 AI工程和传统软件开发的核心差异在做这个项目之前我一直用传统后端开发的思维去想问题结果踩了一堆坑。传统工程的输入是可枚举的输出是可断言的——一个函数传了非法参数它会抛异常一个接口超时了你能明确捕获。但AI工程完全不同。你把同样一段用户问题发给同一个模型两次拿到的回答可能用词都不一样你为了增加确定性给prompt写了长长的规则结果模型在边界case上就是不听话你觉得温度参数调低到0能保证稳定实测却发现它仍然可能产生随机输出。这意味着什么意味着你的代码层面必须额外做好容错、重试、校验、兜底。这些在设计阶段如果不规划好后面上线就是事故现场。另外AI工程里你的计算资源是按token计费的这和传统开发里调个函数不花钱的思维差异巨大。一次用户对话可能消耗几千token如果上下文管理做不好同样的问题反复拼接历史记录成本指数级上升响应延迟也能拖到用户直接关掉页面。所以从零开始学AI工程学的不仅是怎么把模型接进来更是怎么花钱花得聪明、让你写的每一行代码都在控制成本。3. 第一阶段里程碑从裸API调用到第一个能用的程序3.1 用最原始的方式让模型开口说话把框架都扔掉之后第一个要攻克的城墙就是直接用HTTP请求调用模型API。我自己偏好用Python的requests库配合OpenAI兼容接口来做这件事因为OpenAI兼容协议现在几乎是行业标准不管是OpenAI、智谱、DeepSeek还是各种开源模型网关全都兼容这套接口格式。这意味着你在本地用兼容接口调通过一次换任何一家模型服务商只改base_url和api_key就能跑。这里有一个新手特别容易犯的错误把API Key硬编码在代码里还顺手提交到GitHub的公开仓库。这不是羞耻的问题这是能让你被机器人扫到、一夜之间账户被刷爆的问题。正确做法是用环境变量管理密钥比如在项目根目录放一个.env文件用python-dotenv读取。import os import requests from dotenv import load_dotenv load_dotenv() def call_llm(messages, modelgpt-4o-mini, temperature0.7, max_tokens1024): api_key os.getenv(OPENAI_API_KEY) base_url os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) resp requests.post( f{base_url}/chat/completions, headers{ Authorization: fBearer {api_key}, Content-Type: application/json }, json{ model: model, messages: messages, temperature: temperature, max_tokens: max_tokens }, timeout(10, 120) # 连接超时10秒读超时120秒 ) resp.raise_for_status() data resp.json() return data[choices][0][message][content]别小看这个看起来很简单的函数它里面已经藏了三个工程关键设计。第一messages是个列表这就是模型理解上下文的输入格式——你需要主动决定把哪些内容放进去。第二timeout必须设双元组否则遇到网络抖动你的线程可能在底层socket上挂到天荒地老。第三temperature参数直接影响生成结果的稳定性和创造性在工程里你不可能永远用一套参数后面做不同业务场景时需要细化。3.2 把系统提示词当成程序的配置文件来管我在第一批写AI应用的开发者身上看到的通病是把prompt当作聊天时输入的一段开场白随手写在代码里、或者直接在前端文本框里打进去。这在demo阶段没问题但一旦你要维护一个真实项目prompt就成了你程序逻辑的一部分它应该被当作配置文件来管理。我自己习惯在项目里建一个prompts/目录按场景存放system prompt统一用模板语言渲染变量。这么做有三个理由一是产品经理改文案不用找程序员二是测试时能对同一套prompt做版本对比三是你后期做prompt评估时需要把这个作为可变量来实验。举个例子假设你要做一个视频脚本助手。你的system prompt不能只写你是一个视频脚本助手你要定义角色边界、输出格式、禁用事项、特殊情况兜底逻辑。而这些应该通过模板参数动态插入system_prompt f 你是一位短视频编剧擅长把复杂的科技概念转译为大众能听懂的故事。 ## 你的输出格式 必须严格按以下结构输出 1. 开场钩子不超过30字必须制造悬念或情绪共鸣 2. 痛点引入用生活化场景描述观众遇到的问题 3. 知识点拆解分2-3个小点每点用一个小类比辅助理解 4. 行动建议给出可立刻执行的下一步 ## 硬性要求 - 全文不超过800字 - 禁止使用总的来说综上所述这类书面结尾 - 禁止输出任何未经证实的统计数据 ## 本次任务主题 {user_query} 工程思维在这里的体现是你自己定义了什么算一个好回答的标准而不是含糊地指望模型写得更好一些。这就为你后续做自动化评估打好了基础。3.3 第一个坑流式输出到底要不要做我第一版AI应用死活不开流式输出因为觉得代码简单。上线后用户反馈说点完发送要等十几秒还以为是卡了我才发现交互体验跟不流式的差距有多大。流式输出streaming的意思是模型把答案切成一段一段返回你的程序边收边展示用户看到打字机效果感知延迟大幅降低。实现流的思路其实不复杂请求时加stream: true然后不断读取响应体里以data:开头的JSON片段直到遇到data: [DONE]。这里最大的工程坑在于流式输出的字符是在多个数据块里切割的如果直接在流中做敏感词过滤、输出字数统计你会在一个词被切成两半时得到错误的中间结果。正确做法是把全量片段缓存一份流结束后再做后处理。def call_llm_stream(messages, on_token_callback, modelgpt-4o-mini): api_key os.getenv(OPENAI_API_KEY) base_url os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) resp requests.post( f{base_url}/chat/completions, headers{ Authorization: fBearer {api_key}, }, json{ model: model, messages: messages, stream: True }, streamTrue, timeout(10, 300) ) full_content [] for raw_line in resp.iter_lines(decode_unicodeTrue): if not raw_line: continue line raw_line.strip() if line.startswith(data: ): data_str line[len(data: ):] if data_str [DONE]: break import json chunk json.loads(data_str) delta chunk[choices][0][delta] if content in delta: token_text delta[content] full_content.append(token_text) on_token_callback(token_text) return .join(full_content)做完这一步你的程序就从能返回文本的一个脚本变成了像一个真应用的对话服务。4. 自研上下文管理器AI应用的核心骨架4.1 为什么上下文管理是AI工程的第一阵地如果说裸API调用是AI工程的Hello World那上下文管理就是AI工程的分布式系统设计。很多人在这一关被卡住。大模型本身没有记忆。你每次调用API它只看到你这次传进去的完整messages列表之前说过的话它一体不知。所谓多轮对话能力其实完全靠你在每次请求前手动把之前的用户问题、AI回答甚至系统工具的结果拼接到当前请求里。听起来简单但只要你开始处理真实业务复杂度立刻爆炸对话长了怎么办超出token上限怎么截断用户修改话题后旧记忆是否保留系统工具返回的长文档和核心对话怎么分配token预算4.2 用消息队列思路替代简单的列表追加新手做对话记忆最常见的方式就是把所有消息都存在一个Python列表里越积越长。这在对话轮次少的时候没问题但对话超过20轮后效果就开始浮动模型开始忘记开头的指令成本也在肉眼可见地飙升。我推荐的办法是把上下文管理看作一个带容量限制的消息队列而你用token长度来决定什么时候淘汰哪条消息。我实现了一个简单的上下文窗口管理器核心逻辑是在加入新消息后检查总token数超过阈值就从最旧的消息开始淘汰但三条消息必须被特殊保护——系统提示词、最近一轮用户输入、最近一轮助手回复。class ContextWindow: def __init__(self, system_prompt, max_tokens8000, token_encoderNone): self.messages [{role: system, content: system_prompt}] self.max_tokens max_tokens # token_encoder 可以是 tiktoken也可以是模型服务商提供的计数函数 self.token_encoder token_encoder def _count_tokens(self, messages): return sum(len(self.token_encoder.encode(m[content])) for m in messages) def add_message(self, role, content): self.messages.append({role: role, content: content}) self._trim() def _trim(self): # 系统提示词必须保留记录其token数 system_tokens len(self.token_encoder.encode(self.messages[0][content])) current_tokens self._count_tokens(self.messages) while current_tokens self.max_tokens and len(self.messages) 3: # 从系统提示之后最旧的消息淘汰 oldest self.messages.pop(1) reduced len(self.token_encoder.encode(oldest[content])) current_tokens - reduced如果你不想引入tiktoken这种额外依赖还有一个粗糙但好用的估算方法中文场景下一个token大约对应0.6到0.7个汉字你可以用字符串长度除以0.6来粗糙估算同时给自己留出20%的余量。但正式项目里我还是建议用官方计数函数毕竟token计费是严肃的金钱问题。4.3 动态摘要压缩对话太长时的最终武器只有窗口滑动截断还不够。如果一个金融客服场景用户聊了三十轮最开头的关键信息比如用户说我来自上海想咨询企业贷款早被挤出去了模型后面就一直在闭眼瞎猜。这种情况下我建议引入关键信息动态摘要机制。思路是当上下文超长要淘汰旧消息时不简单丢弃而是把被移除的对话内容批量丢给一个便宜的小模型让它总结出关键要点然后作为一条压缩摘要消息放在系统提示词后面。这样对话窗口既能控制token预算又不丢失关键背景。我实测过一个小项目不加摘要时一个三十轮的客服机器人到后面答非所问加了摘要机制后同样的对话它能准确记得用户最初的服务诉求。当然代价是每次summary也要花钱所以摘要触发频率必须控制好——我通常设定为旧消息占用超过上下文预算1/3时才做一次批量摘要。5. 工程化课代表把玩具代码改造成可上线的应用5.1 三层架构与配置解耦做完上面几步你手上的代码其实已经是一个能跑的对话程序了。但离可以交给别人用、可以部署上线还有一段路这段路就是工程化改造。我个人的标准方案是把代码分成三层llm/只负责和模型API通信、memory/负责上下文管理、app/负责业务逻辑和Web服务。同时所有可配置项模型名、温度、超时时间、token上限、摘要触发阈值都抽到config.yaml或环境变量里绝对不硬编码。# config.yaml llm: model: gpt-4o-mini temperature: 0.3 max_tokens: 1024 timeout_seconds: 120 memory: max_context_tokens: 8000 summary_trigger_ratio: 0.33 summary_model: gpt-4o-mini app: host: 0.0.0.0 port: 8080 max_request_concurrency: 50很多从写脚本起步的开发者会嫌这套流程麻烦——我直接global变量不香吗但当你要让这个应用面对多个用户时你就会明白全局变量是所有用户共享的A用户的对话历史会串到B用户那里去。这就是并发隔离问题。5.2 用用户级状态管理解决串话事故ai-engineering-from-scratch讲的是工程工程就意味着你要处理多用户、并发、状态隔离。大模型服务是无状态的但你自己的应用要做的是为每个用户维护一个有状态的会话上下文。最简单的方案是给每个会话分配一个session_id可以是UUID然后用一个内存字典把session_id映射到对应的ContextWindow实例。class ChatService: def __init__(self): self.sessions {} self.max_sessions 1000 def get_context(self, session_id): if session_id not in self.sessions: if len(self.sessions) self.max_sessions: # 淘汰最久未使用的会话 oldest_key next(iter(self.sessions)) del self.sessions[oldest_key] self.sessions[session_id] ContextWindow(system_prompt) return self.sessions[session_id] def chat(self, session_id, user_input): ctx self.get_context(session_id) ctx.add_message(user, user_input) reply call_llm_stream(ctx.messages, on_token_callbackNone) ctx.add_message(assistant, reply) return reply做一个简单的内存缓存是最低成本的方案。但如果你的服务要部署成多实例比如跑在Kubernetes里内存方案就失效了——用户第一次请求打到实例A第二次打到实例B上下文就断了。这时你需要把上下文存到Redis里key就是session_idvalue是序列化后的消息列表每次请求先加载、更新后再写回。关于这个话题可以作为一种扩展思路但第一版不急着上Redis先把内存版的并发隔离和会话淘汰机制吃透再说。5.3 用FastAPI包一层从脚本到服务我会用FastAPI来做Web层因为它天然支持异步、自带接口文档、并且有很好的Streaming支持。把这层加上后你的程序就正式从命令行玩具变成了可以被任何客户端通过HTTP调用的服务。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from fastapi.responses import StreamingResponse import json, uuid app FastAPI() chat_service ChatService() class ChatRequest(BaseModel): message: str session_id: str | None None app.post(/v1/chat) async def chat_endpoint(req: ChatRequest): session_id req.session_id or str(uuid.uuid4()) async def event_generator(): # 使用流式回调往SSE格式里写内容 def on_token(t): yield fdata: {json.dumps({type: token, content: t}, ensure_asciiFalse)}\n\n reply call_llm_stream( chat_service.get_context(session_id).messages, on_token_callbackon_token ) chat_service.get_context(session_id).add_message(assistant, reply) yield fdata: {json.dumps({type: done, session_id: session_id}, ensure_asciiFalse)}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这里我把API设计成返回text/event-stream的SSE格式前端用原生fetch配合ReadableStream就能实现打字机输出完全不需要引入WebSocket。5.4 防抖、限流与成本护栏线上应用逃不开三个问题恶意刷接口、并发打爆后端、token成本失控。你至少要做三件事。第一按用户/IP限流。用最简单的令牌桶算法每秒钟允许一定数量的请求超过就返回429。第二给单会话设置最大调用次数和单日token用量上限超了就禁止继续调用防止有人把几十万字的小说粘进来然后让你一次生成十万字总结。第三在调用模型前做个粗略的token预检估算这条prompt要花多少钱超过阈值直接拒绝而不是默默等模型返回。我一度觉得做成本护栏太保守了直到有一次测试脚本在循环里忘了sleep一个小时花掉了差不多够买一台入门级笔记本的钱。从那之后我给自己定了个规矩任何调用模型API的入口必须有成本埋点日志记录prompt tokens、completion tokens和估算美元成本。这行日志就是你的工程良心。6. 质量评估体系没有评测的AI工程等于盲飞6.1 为什么你感觉它不错根本不算数很多开发者做AI应用验收方式是我自己玩了一下感觉还行。这在demo阶段没问题但你要知道模型是概率性的。同一个prompt你测五次可能三次很好、一次有瑕疵、一次完全跑偏。如果你只测一次然后上线你上线的就是那个跑偏的可能性。我在项目里引入了一套轻量评估集的做法。所谓评估集就是提前准备几十条典型用户输入每条输入都标注好期望行为比如包含特定信息拒绝回答按指定结构输出。每次改动prompt或上下文策略后我批量跑一遍评估集用一套固定的规则来打自动分。6.2 如何设计你的第一个评估集评估集不是什么高深的东西我举个例子。假设你做的是产品客服助手评估集可以长这样[ { id: case_001, input: 你们公司的退款政策是什么, expected: { must_contain: [7天, 无理由], must_not_contain: [亲, 老铁], output_type: plain_text } }, { id: case_002, input: 在吗能聊聊吗, expected: { must_contain: [您好, 有什么可以帮您], max_length: 50 } }, { id: case_003, input: 我要骂你们垃圾产品, expected: { must_not_contain: [傻逼, 你不行], require_empathy_policy: true } } ]拿这批用例跑完之后程序会统计整体通过率。如果你的prompt调整让通过率从92%掉到70%立刻就能发现改坏了。这里有个经验之谈评估集不用多50条高质量用例永远胜过500条平庸用例。重要的是覆盖边界情况而不是凑数量。6.3 回归测试AI应用也要持续集成我还会把这套评估脚本挂到CI里去每次提交代码前跑一遍。这里有个和传统工程很大的不同传统CI跑不过就是红叉AI评估不存在100%通过。所以你要设定一个质量门槛——比如通过率不低于85%且关键用例mark为high的必须100%通过低于就阻断合并。这个门槛按你的业务容忍度来定但它必须存在否则你无法阻止任何人把新prompt的副作用悄悄带进生产环境。7. 完整实战一个最小可用的文档问答工具讲到这里全是模型拿一个具体小项目来收个尾。这个实战项目虽然小但五脏俱全刚好把前面所有知识点串起来——裸API调用、上下文管理、服务封装、成本日志、评估集。目标做一个能喂给它几篇markdown技术文档、然后允许用户针对文档内容提问的问答工具。目录结构doc-qa-tool/ ├── config.yaml ├── prompts/ │ └── qa_system.txt ├── llm/ │ ├── client.py # 裸API调用 流式输出 │ └── cost_logger.py # token/成本日志 ├── memory/ │ ├── context_window.py │ └── summary.py ├── eval/ │ ├── test_cases.json │ └── run_eval.py ├── app/ │ └── server.py └── docs/ ├── deployment.md └── api_guide.md关键实现步骤如下。第一步启动时读取docs目录下的文档把它们按章节切块chunk每块控制在500-800字左右。切片可以用最朴素的按标题结构切不用急着上向量检索那个花活。第二步用户提问时把用户问题跑一遍简单的关键词匹配/或者用正则定位到相关章节第一版不需要做Embedding和向量库那是后面进阶的事。第三步把命中章节的内容拼到一个参考资料区和用户问题一起作为新消息加入上下文。系统提示词固定负责告诉模型你只能依据参考资料回答参考资料没有的内容要如实说不知道。这个工具第一版跑通后你立刻会感受到工程链路完整和玩具demo之间的差距你能在日志里看到每次问答的成本和对应文档来源你改了一个切块策略能通过评估集对比哪个版本召回更靠谱你让十个用户同时访问会话上下文互不干扰。这就是一个从零构建的AI工程应用该有的样子。8. 常用问题排查与避坑清单我在反复做这个练习时攒了不少实用的排错经验。与其让你重复踩坑不如直接列出来。症状可能原因排查思路与解法返回内容突然被截断max_tokens设置过小把max_tokens提到输出长度的1.5倍或改用streaming边收边展示多轮对话开始失忆上下文超限被粗暴截掉检查ContextWindow淘汰逻辑是否保留了system prompt和最近轮次同一条问题返回不稳定temperature过高需要精准答案的场景把温度降到0-0.2创作用途可保留0.8以上接口偶发超时模型推理慢或网络抖动必须设置超时时间且要有重试机制指数退避最多3次并发多了之后开始报错限流超过了服务商RPM限制本地上限流请求排队别指望靠重试硬扛限流错误回答完全和参考资料无关prompt未约束或参考内容未传入检查system prompt是否写死了只能依据参考内容回答还有一些代码之外的提醒。第一API响应里的finish_reason一定要打日志length和content_filter的含义完全不同前者是token超了后者是内容安全机制触发。第二生产环境永远使用配置中心或环境变量管密钥写进代码里就是在给损失送钱。第三不要盲目追赶最新的Agent框架先把调用、上下文、成本、评测这四个地基打好你会发现所有高级框架都只是这些能力的封装组合。9. 从单轮调用到Agent开发一条可靠的进阶路径最后再说一个真实的体会。做完这个从零搭建的项目之后我最大的收获不是我会调API了而是建立了一种判断力——看到任何一个AI工具或框架我能立刻拆解出它的核心机制然后判断它到底解决了我链路里的哪个环节的问题。如果你也想获得这种判断力我建议按这个顺序进阶。第一给这个问答工具加上Embedding向量检索把关键词匹配替换成语义召回你会发现RAG的真相就是检索拼接生成而背后检索的质量决定生成质量的天花板。第二把单次的问一个答一个改造成模型自主决定调用哪个工具的循环这就是Agent的雏形。你会在实现过程中理解Agent不是神秘的黑魔法它的本质就是一个模型在循环中决定下一步做什么你的代码负责执行并反馈结果。第三尝试接入开源模型通过Ollama或vLLM对比同一个prompt在开源和商用模型上的表现差异这是锻炼选型能力的好方法。》最后分享一个我在测试中最得意的Trick设计Agent循环时在每轮工具调用结束后打印一条思考摘要日志比如模型认为需要查天气API因为用户提到了出行计划。这个做法你在后期调试Agent时一定会回来感谢我——它把你从看着黑盒猜它在干什么的深渊里解救出来。