ARTICLE DETAIL

资讯详情

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

从零构建AI工程能力:Token管理、RAG与Agent系统设计实战

从零构建AI工程能力:Token管理、RAG与Agent系统设计实战 1. 从零搭建AI工程能力这个项目到底在解决什么问题第一次看到 ai-engineering-from-scratch 这个标题我脑子里蹦出来的第一个念头是又是一个教人调包的教程但仔细琢磨了一下 from scratch 这几个字再结合现在市面上 AI 学习资源的现状我意识到这个项目想做的事情其实很不一样。现在学 AI 的人大概分两类。一类是搞算法研究的天天跟论文和数学公式打交道目标是发顶会、刷榜另一类是应用层的开发者想用大模型 API 做点产品出来但往往卡在不知道从哪下手和做出来的东西跑不稳这两个坎上。这两类人之间有一个巨大的断层——从模型原理到工程落地之间的那一段路几乎没人系统性地讲清楚。ai-engineering-from-scratch这个项目瞄准的就是这个断层。它要解决的核心问题是一个有一定编程基础的人如何在不依赖大量现成框架封装的前提下从最底层开始理解并构建一套完整的 AI 工程能力。这里说的从零不是让你从写矩阵乘法开始造轮子而是指从工程视角出发把 AI 应用开发中每一个关键环节都拆开来看清楚然后自己动手搭一遍。适合谁来参考这个项目我的判断是三类人第一类是有后端或全栈开发经验、想转型做 AI 应用的工程师第二类是刚入门的算法工程师模型会训但不知道怎么把模型变成能上线的服务第三类是技术负责人需要评估 AI 项目的工程复杂度和团队能力缺口。如果你属于这三类中的任何一类这个项目的内容框架都值得你花时间吃透。我之所以对这个方向特别有感触是因为过去两年我在实际项目里踩过的坑几乎全都集中在工程这两个字上。模型选型、Prompt 设计、上下文管理、检索增强、评测体系、成本控制、延迟优化——这些东西没有任何一门课会系统教你但每一个都直接决定你的 AI 应用能不能活下来。下面我就按照这个项目的逻辑脉络把我理解的 AI 工程能力建设路径完整拆解一遍。2. 核心能力拆解AI 工程师到底需要掌握什么2.1 为什么不能只学调 API很多人对 AI 工程的理解停留在会调 OpenAI 接口就行。我刚开始也这么想直到第一次做一个客服问答系统发现事情远没有那么简单。你调一次 API 返回一段文本很容易但你要让这个系统在真实场景下稳定运行需要解决的问题包括用户输入超长了怎么办模型返回格式不对怎么解析多轮对话的上下文怎么管理才不会爆 token并发请求上来之后延迟飙升怎么优化模型偶尔胡说八道怎么兜底这些问题没有一个能靠调 API解决。它们属于工程问题需要你有系统设计的能力。ai-engineering-from-scratch这个项目的价值就在于它把这些工程问题一个一个拎出来让你从原理层面理解之后再动手实现。我个人的经验是AI 工程能力可以拆成四个层次基础层理解 Transformer 架构的基本原理知道 tokenization、embedding、attention 这些概念在工程上意味着什么应用层掌握 Prompt 工程、上下文管理、输出解析、函数调用等核心技能系统层能够设计 RAG 系统、Agent 系统、多模型路由等复杂架构运维层具备评测、监控、成本优化、灰度发布等生产级能力大部分教程只覆盖了第二层的一部分而真正让 AI 应用能跑起来、跑得稳的是第三层和第四层的能力。这个项目从标题来看目标就是把这四层全部打通。2.2 从零构建的零到底在哪里这里需要澄清一个很容易被误解的点。from scratch 不是让你不用任何框架从零实现一个 Transformer。那样做除了学习目的之外没有任何工程价值。这里的零我理解是指认知上的零——假设你对 AI 工程一无所知从最基础的概念开始一步一步建立起完整的知识体系和动手能力。具体来说这个零包含几个层面的含义第一不预设你有 AI 背景。你不需要先学完吴恩达的机器学习课程才能开始。项目应该会从什么是 token为什么模型有上下文长度限制这些最基础的问题讲起让你先建立直觉再深入细节。第二不预设你有特定的技术栈。不管你之前是写 Python 的还是写 Java 的是做前端的还是做后端的都应该能跟上。AI 工程的核心概念是跨语言的Python 只是当前最方便的工具而已。第三不跳过任何关键步骤。很多教程为了显得高级会直接给你一个封装好的库让你调用你跑通了但不知道里面发生了什么。这个项目应该会反其道而行之把每个环节都拆开让你看到内部机制之后再决定要不要用现成的轮子。我特别欣赏这种思路因为我自己就是这么学过来的。当初学 RAG 的时候直接用 LangChain 的封装确实五分钟就能跑通一个 demo但后来遇到检索效果不好的问题我完全不知道从哪里排查——因为我不知道 LangChain 在背后做了什么。后来我花了一个周末用最朴素的方式手写了一遍 RAG 流程手动做文本分块、手动调 embedding 接口、手动算余弦相似度、手动拼 Prompt。写完之后所有问题都变得清晰了。2.3 工程视角和学术视角的根本差异这一点我觉得值得单独拿出来说因为它决定了你学习 AI 工程的方式。学术视角关心的是这个模型在 benchmark 上刷了多少分这个算法的理论复杂度是多少这个方法的创新点在哪里工程视角关心的是这个模型在我这个场景下够不够用调一次要多少钱响应时间能不能接受出错了怎么降级数据怎么更新怎么评估效果好不好这两个视角没有高下之分但如果你要做 AI 应用开发你必须切换到工程视角。举个例子学术界可能花大量精力研究如何把模型准确率从 95% 提升到 96%但工程上你可能更关心的是为了这 1 个百分点推理成本要翻倍延迟要增加 300ms这笔账划不划算ai-engineering-from-scratch这个项目如果做得好应该会始终贯穿这种工程思维。不是教你追求最先进的模型而是教你在约束条件下做出最合理的工程决策。3. 实操路径从环境搭建到第一个可用系统3.1 开发环境与工具链的选型逻辑动手之前先把环境搞利索这是我做了这么多年项目养成的习惯。AI 工程的环境搭建有几个特殊之处值得单独说一下。Python 版本的选择。我的建议是直接用 3.10 或 3.11。原因很简单AI 生态里很多库对 Python 版本有要求3.9 以下很多新库装不了3.12 又太新部分库还没适配。3.10/3.11 是目前兼容性最好的区间。我实测下来3.11 在性能和兼容性之间平衡得最好。虚拟环境管理。别用系统 Python别用 conda 一把梭。我推荐用uv或者poetry来管理依赖。uv是这两年新起来的工具速度快得离谱装依赖的时间能从几分钟缩短到几秒。如果你还没用过强烈建议试一下# 安装 uv curl -LsSf https://astral.sh/uv/install.sh | sh # 创建虚拟环境 uv venv # 激活环境 source .venv/bin/activate # 安装依赖 uv pip install openai tiktoken numpy核心依赖的选择。AI 工程不需要一上来就装一堆重型框架。我的建议是最开始只装这几个依赖用途为什么选它openai调用模型接口事实标准兼容性最好tiktokentoken 计数精确计算成本的基础numpy数值计算向量运算必备python-dotenv环境变量管理避免密钥硬编码httpx异步 HTTP 请求高并发场景必备注意不要一上来就装 LangChain、LlamaIndex 这些重型框架。先用最基础的工具把流程跑通理解每一步在做什么之后再决定要不要用框架提效。3.2 第一个核心模块Token 管理与成本控制Token 是 AI 工程里最基础也最容易被忽视的概念。很多人做 AI 应用亏钱就是因为没搞清楚 token 是怎么算的。简单来说token 是模型处理文本的最小单位。一个英文单词大约是 1-2 个 token一个中文字大约是 1-2 个 token。模型 API 的计费、上下文长度限制、生成速度全都跟 token 数量直接相关。我建议你做的第一件事就是写一个 token 计数和成本估算的工具函数import tiktoken def count_tokens(text: str, model: str gpt-4) - int: 计算文本的 token 数量 encoding tiktoken.encoding_for_model(model) return len(encoding.encode(text)) def estimate_cost(input_text: str, output_text: str, model: str gpt-4) - dict: 估算单次调用的成本 # 价格表美元/1K tokens实际使用时需更新 pricing { gpt-4: {input: 0.03, output: 0.06}, gpt-3.5-turbo: {input: 0.0015, output: 0.002}, } input_tokens count_tokens(input_text, model) output_tokens count_tokens(output_text, model) price pricing.get(model, pricing[gpt-4]) cost (input_tokens / 1000 * price[input] output_tokens / 1000 * price[output]) return { input_tokens: input_tokens, output_tokens: output_tokens, total_tokens: input_tokens output_tokens, estimated_cost_usd: round(cost, 6) }这个工具看起来简单但它是你做所有成本优化的基础。没有它你根本不知道钱花在哪里了。我踩过的一个坑早期做项目的时候没有做 token 计数上线一个月后发现账单比预期高了 5 倍。排查之后发现是系统 Prompt 写得太长每次请求都带了一大段没用的说明文字。后来把 Prompt 精简了 60%成本直接降下来了。3.3 上下文管理的工程实现上下文管理是 AI 工程里最考验设计能力的部分。模型有上下文长度限制你不能把所有历史对话都塞进去但你又不能丢掉关键信息。怎么平衡我总结了几种常见的策略按复杂度从低到高排列策略一滑动窗口。最简单的做法只保留最近 N 轮对话。优点是实现简单缺点是会丢失早期的重要信息。策略二摘要压缩。当对话轮数超过阈值时用模型把之前的对话总结成一段摘要然后只保留摘要和最近的对话。这个策略在客服场景下特别有效。策略三关键信息提取。从对话中提取出结构化的关键信息比如用户的名字、订单号、问题类型单独存储每次请求时按需注入。这个策略实现复杂度最高但效果最好。class ContextManager: def __init__(self, max_tokens: int 4000, reserve_tokens: int 1000): self.max_tokens max_tokens self.reserve_tokens reserve_tokens self.history [] self.summary def add_message(self, role: str, content: str): self.history.append({role: role, content: content}) self._compress_if_needed() def _compress_if_needed(self): 当历史记录超过阈值时进行压缩 total sum(count_tokens(m[content]) for m in self.history) if total self.max_tokens - self.reserve_tokens: # 保留最近 4 轮其余压缩成摘要 recent self.history[-8:] old self.history[:-8] if old: old_text \n.join( f{m[role]}: {m[content]} for m in old ) # 这里调用模型生成摘要 self.summary self._summarize(old_text) self.history recent def get_context(self) - list: 获取当前应该发送给模型的上下文 messages [] if self.summary: messages.append({ role: system, content: f之前的对话摘要{self.summary} }) messages.extend(self.history) return messages实操心得上下文压缩的阈值不要设得太激进。我一开始把阈值设得很低结果模型经常忘记用户之前说过的关键信息体验很差。后来调整为保留最近 8-10 轮对话只在确实超限时才压缩效果好很多。3.4 输出解析与结构化数据提取模型返回的是自然语言文本但你的程序需要的是结构化数据。这个转换过程就是输出解析它是 AI 工程里最容易出问题的环节之一。最朴素的做法是用正则表达式去匹配但模型输出的格式往往不完全可控正则很容易失效。更可靠的做法是在 Prompt 里明确要求模型输出 JSON 格式然后用容错解析器处理。import json import re def parse_json_response(text: str) - dict: 容错解析模型返回的 JSON # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 尝试提取 json ... 代码块 pattern r(?:json)?\s*([\s\S]*?) matches re.findall(pattern, text) for match in matches: try: return json.loads(match.strip()) except json.JSONDecodeError: continue # 尝试找到第一个 { 和最后一个 } start text.find({) end text.rfind(}) if start ! -1 and end ! -1: try: return json.loads(text[start:end1]) except json.JSONDecodeError: pass raise ValueError(f无法从响应中解析出 JSON: {text[:200]})这个容错解析器的思路是层层降级先试最理想的直接解析不行就找代码块再不行就找花括号范围。实测下来这套组合拳能覆盖 95% 以上的情况。但更重要的是你要在 Prompt 层面就做好约束。我常用的做法是在 System Prompt 里明确写清楚输出格式并给一个示例你是一个信息提取助手。请从用户输入中提取以下字段 以 JSON 格式返回不要输出任何其他内容。 字段说明 - name: 人名字符串 - age: 年龄整数 - city: 城市字符串 示例输出 {name: 张三, age: 28, city: 北京}4. 进阶系统设计RAG 与 Agent 的工程落地4.1 RAG 系统的核心环节与常见陷阱RAG检索增强生成是当前 AI 应用中最主流的架构模式。它的核心思路很简单先从知识库里检索出相关内容再把内容和用户问题一起发给模型让模型基于检索到的内容来回答。但简单只是表面现象。我在实际项目中做过好几个 RAG 系统每一个都遇到了不同的问题。下面我把 RAG 的完整流程拆开逐个环节讲清楚。环节一文档处理与分块。这是最容易被低估的环节。很多人直接把整篇文档丢进去做 embedding结果检索效果一塌糊涂。正确的做法是根据文档结构做智能分块。分块的核心矛盾是块太小语义不完整块太大检索精度下降。我的经验值是300-500 个 token 一块块之间保留 50-100 token 的重叠。重叠的作用是避免关键信息刚好被切在边界上导致丢失。def chunk_text(text: str, chunk_size: int 400, overlap: int 80) - list: 按 token 数量分块保留重叠 encoding tiktoken.get_encoding(cl100k_base) tokens encoding.encode(text) chunks [] start 0 while start len(tokens): end start chunk_size chunk_tokens tokens[start:end] chunk_text encoding.decode(chunk_tokens) chunks.append(chunk_text) start end - overlap return chunks环节二向量化与存储。把文本块转成向量存起来。这里的选择很多可以用 API 做 embedding也可以用本地模型。我的建议是数据量小的时候用 API数据量大且对成本敏感的时候用本地模型。环节三检索策略。最基础的是余弦相似度检索但实际项目中纯向量检索往往不够。我通常会组合三种检索方式向量检索处理语义相似的问题关键词检索处理专有名词、型号等精确匹配混合排序把两种检索结果合并后重新排序环节四Prompt 组装与生成。把检索到的内容拼进 Prompt 里让模型基于这些内容回答。这里的关键是明确告诉模型只基于检索内容回答不要自己编。踩坑记录我做过一个法律咨询的 RAG 系统早期没有在 Prompt 里加只基于以下内容回答的约束结果模型经常把检索到的法条和它自己训练数据里的内容混在一起给出了错误的法条引用。加上约束之后准确率提升了将近 30%。4.2 Agent 系统的设计模式与实现要点Agent 是比 RAG 更进一步的架构。RAG 是检索-生成的单向流程Agent 则是思考-行动-观察的循环流程。模型不仅能回答问题还能调用工具、执行操作、根据结果调整策略。Agent 的核心组件包括规划模块决定下一步做什么工具集可调用的外部函数执行器实际执行工具调用记忆模块存储历史交互和中间结果最简单的 Agent 实现是一个 ReAct 循环class SimpleAgent: def __init__(self, tools: dict, max_steps: int 10): self.tools tools self.max_steps max_steps def run(self, task: str) - str: history [] for step in range(self.max_steps): # 构建 Prompt包含任务、工具描述和历史 prompt self._build_prompt(task, history) response call_llm(prompt) # 解析模型的输出判断是调用工具还是给出答案 action self._parse_action(response) if action[type] final_answer: return action[content] if action[type] tool_call: tool_name action[tool] tool_input action[input] if tool_name in self.tools: result self.tools[tool_name](**tool_input) history.append({ step: step, tool: tool_name, input: tool_input, result: result }) else: history.append({ step: step, error: f未知工具: {tool_name} }) return 达到最大步数限制未能完成任务Agent 系统最大的挑战是稳定性。模型可能会陷入循环、调用错误的工具、或者产生幻觉。我在实际项目中总结了几条经验第一工具描述要极其清晰。不要写查询天气要写根据城市名称查询当前天气输入参数为城市名称字符串返回温度、湿度和天气状况。第二设置最大步数限制。没有限制的 Agent 可能会无限循环烧钱又浪费时间。第三关键操作要加人工确认。比如发送邮件、修改数据库这类不可逆的操作一定要在 Agent 执行前让用户确认。第四做好日志记录。Agent 的每一步决策都要记录下来出问题的时候才能排查。4.3 多模型路由与降级策略在实际生产环境中只用一个模型是很危险的。API 可能超时、可能限流、可能涨价。你需要一套多模型路由和降级机制。我的做法是维护一个模型池根据任务类型和当前状态动态选择class ModelRouter: def __init__(self): self.models [ {name: gpt-4, priority: 1, cost: high, capability: high}, {name: gpt-3.5-turbo, priority: 2, cost: low, capability: medium}, {name: local-model, priority: 3, cost: free, capability: low}, ] self.failure_count {} def select_model(self, task_complexity: str) - str: 根据任务复杂度选择模型 if task_complexity high: candidates [m for m in self.models if m[capability] high] elif task_complexity medium: candidates [m for m in self.models if m[capability] in (high, medium)] else: candidates self.models # 过滤掉失败次数过多的模型 available [m for m in candidates if self.failure_count.get(m[name], 0) 3] if not available: available candidates # 按优先级排序返回最优选择 available.sort(keylambda m: m[priority]) return available[0][name] def report_failure(self, model_name: str): 报告模型调用失败 self.failure_count[model_name] \ self.failure_count.get(model_name, 0) 1 def report_success(self, model_name: str): 报告模型调用成功重置失败计数 self.failure_count[model_name] 0这套机制的核心思想是优先用最好的模型但如果它出问题了自动降级到次优选择。同时记录每个模型的失败次数连续失败多次就暂时跳过它。实操心得降级策略一定要提前测试。我见过很多团队写了降级逻辑但从来没测过真到需要降级的时候发现降级后的模型输出格式不兼容整个系统直接崩了。建议在开发阶段就模拟主模型不可用的情况确保降级路径是通的。5. 评测、监控与持续优化5.1 为什么评测是 AI 工程的生命线做 AI 应用最怕的一件事是你改了一个 Prompt感觉效果变好了但实际上在你看不到的角落其他场景的效果变差了。没有评测体系你就是在盲人摸象。评测体系的核心是建立一套可重复运行的测试集。这个测试集应该包含典型场景的输入和期望输出边界情况的输入之前出过问题的 caseclass Evaluator: def __init__(self, test_cases: list): self.test_cases test_cases def run(self, system_fn) - dict: 运行评测返回各项指标 results [] for case in self.test_cases: try: output system_fn(case[input]) score self._score(case, output) results.append({ case_id: case[id], input: case[input], expected: case[expected], actual: output, score: score, status: success }) except Exception as e: results.append({ case_id: case[id], status: error, error: str(e), score: 0 }) # 汇总指标 total len(results) success sum(1 for r in results if r[status] success) avg_score sum(r[score] for r in results) / total return { total_cases: total, success_rate: success / total, average_score: avg_score, details: results } def _score(self, case: dict, output: str) - float: 评分逻辑根据具体场景定制 # 简单示例关键词匹配 expected_keywords case.get(keywords, []) if not expected_keywords: return 1.0 if output case[expected] else 0.0 matched sum(1 for kw in expected_keywords if kw in output) return matched / len(expected_keywords)评测的频率建议是每次修改 Prompt 或更换模型后必须跑一遍日常每周跑一次全量评测。评测结果要存档方便对比不同版本的效果。5.2 生产环境的监控指标上线之后你需要持续监控几个核心指标指标含义警戒线建议请求延迟 P9595% 请求的响应时间超过 5s 需排查错误率调用失败的比例超过 2% 需告警Token 消耗每日 token 使用量突增 50% 需排查用户反馈率用户主动反馈的比例低于 1% 说明体验尚可缓存命中率缓存生效的比例低于 20% 需优化这些指标不需要一开始就全部搭建但至少要把延迟和错误率监控起来。我见过太多团队上线之后完全不看监控直到用户投诉才发现问题。5.3 持续优化的几个方向AI 应用的优化是一个持续的过程没有一劳永逸的方案。我通常从这几个方向入手Prompt 优化。这是成本最低、见效最快的方式。定期回顾失败 case分析是 Prompt 哪里写得不够清楚然后针对性修改。检索优化。如果是 RAG 系统检索质量直接决定最终效果。优化方向包括改进分块策略、增加重排序环节、调整检索数量。缓存策略。很多请求是重复的加一层语义缓存可以大幅降低成本。注意是语义缓存不是简单的字符串匹配缓存——意思相近的问题应该命中同一个缓存。模型微调。当 Prompt 优化到极限之后可以考虑用积累的数据做微调。但微调的成本和门槛都比较高建议先把前面的优化做到位再考虑。6. 常见问题与排查技巧实录6.1 模型输出不稳定的排查思路这是被问得最多的问题。同一个输入模型有时候回答得好有时候回答得差。怎么排查首先确认是不是温度参数的问题。温度越高输出越随机。如果你的场景需要稳定输出把温度调到 0 或接近 0。如果温度已经是 0 了还不稳定检查Prompt 是否有歧义。模型对 Prompt 的理解可能和你的预期不一样。把 Prompt 给同事看一下如果同事理解有偏差模型大概率也会有偏差。还有一个容易被忽视的原因是上下文污染。如果多轮对话中前面的内容质量不高会影响后续的回答。这时候需要在 Prompt 里明确告诉模型忽略之前的无关内容。6.2 成本失控的常见原因成本突然飙升通常有这几个原因System Prompt 太长每次请求都带着一大段文字上下文没有做压缩历史对话无限增长没有做缓存重复请求反复计费用了过于昂贵的模型处理简单任务输出没有做长度限制模型生成了大量无用内容排查的时候先把每个请求的 token 消耗打出来看看是输入多还是输出多然后针对性优化。6.3 常见问题速查表问题现象可能原因排查方向响应时间突然变长API 限流或网络问题检查 API 状态增加超时重试输出格式不符合预期Prompt 约束不够明确增加格式示例使用结构化输出检索结果不相关分块策略或 embedding 质量问题调整分块大小更换 embedding 模型多轮对话丢失上下文上下文压缩过于激进调整压缩阈值保留更多历史并发请求报错超过 API 速率限制增加请求队列实现限流模型回答包含幻觉缺少事实约束增加只基于给定内容回答的约束最后分享一个我自己的习惯每次遇到新问题解决之后我都会把问题和解决方案记到一个文档里。时间长了这个文档就成了我自己的故障排查手册。下次遇到类似问题翻一下就能找到答案。这个习惯帮我省了大量的重复排查时间。AI 工程这个领域变化很快新的模型、新的工具、新的模式层出不穷。但底层的工程能力——如何管理上下文、如何设计检索、如何做评测、如何控制成本——这些东西是相对稳定的。把ai-engineering-from-scratch这个项目里涉及的核心能力一个一个吃透比追着新框架跑要扎实得多。我在实际项目中最大的体会就是基础打得越牢学新东西越快。那些看起来过时的手写实现恰恰是你理解新框架的钥匙。
返回列表