☰
从零搭建AI工程:裸API调用、上下文管理与工程化实战
2026/10/2 11:27:28 网站建设 项目流程

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, model="gpt-4o-mini", temperature=0.7, max_tokens=1024): 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": f"Bearer {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, model="gpt-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": f"Bearer {api_key}", }, json={ "model": model, "messages": messages, "stream": True }, stream=True, timeout=(10, 300) ) full_content = [] for raw_line in resp.iter_lines(decode_unicode=True): 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_tokens=8000, token_encoder=None): 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_callback=None) ctx.add_message("assistant", reply) return reply

做一个简单的内存缓存是最低成本的方案。但如果你的服务要部署成多实例(比如跑在Kubernetes里),内存方案就失效了——用户第一次请求打到实例A,第二次打到实例B,上下文就断了。这时你需要把上下文存到Redis里,key就是session_id,value是序列化后的消息列表,每次请求先加载、更新后再写回。关于这个话题,可以作为一种扩展思路,但第一版不急着上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 f"data: {json.dumps({'type': 'token', 'content': t}, ensure_ascii=False)}\n\n" reply = call_llm_stream( chat_service.get_context(session_id).messages, on_token_callback=on_token ) chat_service.get_context(session_id).add_message("assistant", reply) yield f"data: {json.dumps({'type': 'done', 'session_id': session_id}, ensure_ascii=False)}\n\n" return StreamingResponse(event_generator(), media_type="text/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时一定会回来感谢我——它把你从"看着黑盒猜它在干什么"的深渊里解救出来。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询