ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

从零手搓AI工程链路:切分、向量化与检索的底层实践

从零手搓AI工程链路:切分、向量化与检索的底层实践

1. 从零搭建AI工程能力:为什么“手搓一遍”比调包更值钱

这两年AI应用层的工具链成熟得吓人。LangChain、LlamaIndex、各种Agent框架、向量数据库托管服务,几乎把“做一个AI应用”的门槛压到了写几行配置的程度。但我自己带团队、也看过不少同行的项目之后,越来越确信一件事:只会调包的人,遇到线上问题基本是抓瞎的。模型输出突然不稳定、检索召回率掉了一半、成本莫名其妙翻了三倍、延迟从800ms飙到5s——这些问题在框架文档里根本找不到答案,因为答案藏在框架底下那层被封装掉的原理里。

ai-engineering-from-scratch这个标题,说的就是这件事:不依赖现成框架,从最底层把AI工程的核心环节自己实现一遍。它不是让你去训练一个大模型,那是另一回事;它聚焦的是“工程”二字——怎么把模型能力接进一个真实可用的系统里,包括文本切分、向量化、相似度检索、提示词组装、上下文管理、输出解析、失败重试、成本控制这一整条链路。

适合谁来参考?三类人最该动手做一遍。第一类是应用层开发者,平时用框架搭RAG或者Agent,但说不清底层发生了什么;第二类是准备转AI方向的后端工程师,有工程底子但缺AI这块的直觉;第三类是技术负责人,需要判断某个方案到底靠不靠谱,而不是听供应商讲故事。这篇文章我会把整条链路拆开,讲清楚每个环节为什么这么设计、自己实现时踩过哪些坑、参数怎么定,最后给出一套可以直接照着复现的最小实现思路。

需要先说明一点:下面涉及的具体实现细节,是基于“一个合格AI工程师在这个场景下最可能采用的合理方案”做的补全,不是某个特定开源仓库的逐行复刻。重点是思路和取舍逻辑,你换成任何语言、任何模型都能套用。

2. 整体架构设计:一条最小可用的AI工程链路长什么样

2.1 先想清楚“从零”到底指什么

很多人一听“from scratch”就以为是连神经网络都要自己写。不是的。在AI工程这个语境下,“从零”指的是不依赖高层编排框架,但底层的模型推理、向量计算这些基础设施该用还是用。你自己写一个Transformer出来没有任何工程价值,但你自己实现一遍检索增强的完整流程,价值巨大。

所以这条链路的边界我这样划:模型调用可以用官方SDK(比如直接发HTTP请求),向量计算可以用numpy,但切分策略、检索逻辑、提示词组装、上下文窗口管理、结果解析、重试与降级这些“工程决策密集”的部分,全部自己写。这些恰恰是框架帮你做了、但做得好不好你完全不知道的地方。

我见过太多项目,检索效果差,第一反应是换模型、换向量库,折腾一圈发现真正的问题是切分粒度不对,把一句完整的话从中间劈开了。这种问题,只有你自己实现过切分,才会有那个敏感度。

2.2 链路的五个核心环节

一条最小可用的AI工程链路,我把它拆成五段,每段都有明确的输入输出和失败模式:

环节输入输出典型失败模式
文档切分原始文本文本块列表语义被切断、块过大过小
向量化文本块向量矩阵维度不一致、归一化遗漏
检索查询向量Top-K文本块召回不准、相似度算错
提示词组装查询+检索结果完整prompt超上下文、指令冲突
输出解析模型原始输出结构化结果格式错乱、解析崩溃

这五段串起来就是一个RAG的最小闭环。但我要强调的是,每一段都不是“调个函数”那么简单,里面全是需要你自己拍板的参数和策略。下面逐个拆。

2.3 为什么不用现成框架起步

我的建议是:第一个版本一定自己写,跑通之后再考虑要不要上框架。理由有三个。

第一,调试能力。自己写的代码,出问题你能一路print到底,知道是哪一步的数据变了形。用框架,你只能看它抛出来的异常,很多时候异常信息毫无信息量。

第二,成本感知。框架往往默认帮你做很多“贴心”的事,比如自动重试、自动截断、自动加系统提示词,这些都会悄悄增加token消耗。自己写,每一分钱花在哪你清清楚楚。

第三,迁移自由。框架的抽象一旦泄漏,你想换掉某个组件会非常痛苦。自己写的代码,换个模型就是改一个函数的事。

当然,自己写也有代价:开发慢、要处理很多边角情况。所以我的实际做法是——核心链路自己写,外围的监控、日志、部署用成熟工具。这个平衡点后面会细说。

3. 核心环节逐个拆解:切分、向量化与检索的实操要点

3.1 文本切分:最被低估、最容易翻车的一步

切分是整个链路里最不起眼、但影响最大的环节。我踩过的坑里,至少一半跟切分有关。

为什么不能按固定字数硬切?假设你按每500字切一刀,很可能把“因此,我们得出以下结论:”和后面的结论切到两个块里。检索的时候,用户问结论是什么,命中的是前半块,模型拿到的是半句话,输出自然驴唇不对马嘴。

我常用的策略是递归切分 + 语义边界优先。具体做法:

  1. 先按段落(连续两个换行)切,这是最自然的语义边界。
  2. 如果某个段落还是太长,按句子切(中文按。!?,英文按. ! ?)。
  3. 如果单句还是超长(比如代码块、表格),才退化成按字符硬切。

这里有个关键参数:块大小(chunk size)和重叠(overlap)。我的经验值是这样的:

  • 块大小:中文300-500字,英文200-400词。太小则上下文不足,太大则检索精度下降、token浪费。
  • 重叠:块大小的10%-20%。重叠的作用是防止边界信息丢失,比如一个概念横跨两个块,重叠部分能让它至少在一个块里完整出现。

注意:重叠不是越大越好。我试过30%的重叠,结果检索时经常返回两个高度相似的块,白白占用上下文窗口,还稀释了真正有用的信息。

还有一个细节:切分时要保留元数据。每个块除了文本本身,还要带上它来自哪个文档、第几段、原始位置。这样检索回来之后,你能在提示词里告诉模型“这段来自《XX文档》第3节”,模型的可信度判断会好很多。

3.2 向量化:归一化和维度对齐是命门

向量化这一步,用现成的embedding模型就行,没必要自己训。但有两个坑必须自己处理。

第一个坑:归一化。不同embedding模型输出的向量,有的已经归一化了(模长为1),有的没有。如果你混用了不同来源的向量,或者自己算相似度时忘了归一化,余弦相似度会算错。我的做法是:拿到向量后统一做一次L2归一化,这样余弦相似度就等价于点积,计算也快。

import numpy as np def normalize(vec): norm = np.linalg.norm(vec) if norm == 0: return vec return vec / norm

第二个坑:维度对齐。如果你中途换了embedding模型,新旧向量的维度可能不一样,直接算相似度会报错或者算出垃圾结果。所以换模型必须重建整个向量库,没有捷径。我见过有人想“只对新文档用新模型”,结果检索时新旧向量混在一起,效果惨不忍睹。

批量处理也很重要。embedding接口一般支持批量输入,一次传几十条比一条条传快得多,也省钱。但要注意接口的批量上限,超了会被截断或者报错。我一般设成32或64一批,具体看接口文档。

3.3 检索:Top-K不是拍脑袋定的

检索这一步,核心是相似度计算 + Top-K选择。相似度用余弦,前面归一化之后就是点积,numpy一行搞定:

def search(query_vec, doc_matrix, top_k=5): scores = doc_matrix @ query_vec # 归一化后等价于余弦相似度 top_indices = np.argsort(scores)[::-1][:top_k] return top_indices, scores[top_indices]

Top-K怎么定?这是很多人拍脑袋的地方。K太小,召回不足,模型没材料可用;K太大,噪声多,还费token。我的经验:

  • 如果块比较小(300字左右),K取5-8。
  • 如果块比较大(500字以上),K取3-5。
  • 如果加了重排序(rerank),初筛可以取20-30,重排后留3-5。

相似度阈值也值得设一个。有时候用户问的问题跟知识库完全不相关,硬检索也会返回一堆低分块,这些块塞进提示词只会干扰模型。我一般设一个阈值(比如0.3,具体看模型),低于阈值的直接丢弃,宁可告诉用户“没找到相关资料”。

实操心得:检索效果不好时,先别急着换模型。把query和命中的块打印出来,人工看一眼。十有八九是切分问题,或者query本身表述跟文档差太远,需要做query改写。

4. 提示词组装与上下文管理:token预算怎么花

4.1 提示词的结构化组装

检索回来的块,不能直接拼一拼就丢给模型。我习惯用一个固定的模板,把各部分职责分清楚:

[系统指令] 你是一个严谨的助手,只根据下面提供的资料回答问题。 如果资料中没有相关信息,直接说“资料中未提及”,不要编造。 [参考资料] 1. {chunk_1} 2. {chunk_2} ... [用户问题] {query} [回答要求] - 用中文回答 - 引用资料时标注序号

这个模板有几个讲究。系统指令里明确“只根据资料回答”,能大幅降低幻觉。给每个块编号,方便模型引用,也方便你事后核对。明确说“没有就说没有”,比让模型自由发挥安全得多。

4.2 上下文窗口的预算分配

上下文窗口是有限资源,得精打细算。假设模型窗口是8K token,我的分配大概是:

  • 系统指令:200 token
  • 参考资料:4000 token(大约5-8个块)
  • 用户问题:200 token
  • 回答预留:2000 token
  • 缓冲:剩余

参考资料超预算怎么办?两个策略。一是按相似度截断,从低分的块开始丢,直到塞得下。二是对块做压缩,比如只保留跟query最相关的句子。第二种效果更好但实现复杂,我一般先用第一种。

这里有个容易忽略的点:token计数不能用字符数估算。中文一个字大约1-2个token,英文一个词大约1-1.3个token,代码和特殊符号更没谱。一定要用对应模型的tokenizer实际数一遍。我早期用字符数估算,结果线上频繁超窗口,被坑得很惨。

4.3 多轮对话的上下文管理

如果是多轮对话,还要管理历史消息。我的做法是滑动窗口 + 摘要:最近3轮对话保留原文,更早的对话用模型压缩成一句话摘要。这样既保留了上下文连贯性,又不会让历史消息无限膨胀。

注意:摘要本身也要花token,而且摘要会丢信息。所以只在对话确实很长的时候才启用,短对话直接全量保留就行。

5. 输出解析、重试与降级:让系统真正能上线

5.1 输出解析:别假设模型一定听话

你让模型输出JSON,它十次里有八次给你JSON,剩下两次可能给你一段解释加JSON,或者JSON外面包了markdown代码块。解析器必须能容错。

我的解析策略是分层的:

  1. 先尝试直接json.loads。
  2. 失败则用正则提取第一个{...}或[...]。
  3. 再失败则去掉markdown代码块标记再试。
  4. 全失败就触发重试。
import json, re def parse_json_safe(text): try: return json.loads(text) except json.JSONDecodeError: pass match = re.search(r'\{.*\}', text, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass return None

5.2 重试策略:指数退避 + 上限

模型调用失败(超时、限流、返回空)是常态,必须有重试。但重试不能无脑循环,我用指数退避:第一次失败等1秒,第二次等2秒,第三次等4秒,最多重试3次。超过就降级。

import time def call_with_retry(fn, max_retries=3): for i in range(max_retries): try: return fn() except Exception as e: if i == max_retries - 1: raise time.sleep(2 ** i)

5.3 降级方案:系统不能因为一个环节挂了就全挂

降级是上线系统的必修课。我的降级链是这样的:

  • 检索失败 → 退化成纯模型问答(明确告诉用户没有参考资料)。
  • 模型调用失败 → 返回缓存结果或友好提示。
  • 解析失败 → 返回原始文本,让用户自己看。

关键原则:任何单点失败都不能让整个请求500。用户宁可看到一个不完美的回答,也不想看到报错页面。

6. 常见问题与排查技巧实录

6.1 检索不准的排查顺序

检索不准是最常见的问题。我的排查顺序是固定的,从下往上查:

  1. 先看切分:把命中的块打印出来,看是不是被切碎了。
  2. 再看query:用户的问题表述是不是跟文档差太远,需不需要query改写。
  3. 再看embedding:是不是归一化漏了,或者维度对不上。
  4. 最后才怀疑模型:换embedding模型是最后手段,因为要重建整个库。

6.2 常见问题速查表

现象可能原因排查方法
回答答非所问检索块不相关打印命中块人工检查
回答不完整块被切断检查切分边界
频繁超上下文token估算不准用tokenizer实际计数
成本异常高重试过多或块太大加日志统计每次token消耗
输出格式错乱解析器太严格用容错解析
延迟高串行调用太多批量embedding、并行检索

6.3 几个独家避坑技巧

技巧一:给检索加日志。每次检索把query、命中的块、相似度分数都记下来。上线后你会发现,很多问题从日志里一眼就能看出来,比调试代码快得多。

技巧二:准备一个“黄金测试集”。挑20-30个典型问题,人工标注正确答案。每次改动链路(换模型、调参数),跑一遍测试集,看效果是涨是跌。没有测试集,你根本不知道改动是好是坏。

技巧三:成本监控要按请求维度。别只看月度总账单,要能看到每个请求花了多少token。我见过一个bug导致某个接口疯狂重试,一天烧掉平时一周的钱,就是因为没有按请求监控。

技巧四:embedding结果要缓存。同一段文本反复embedding是纯浪费。用文本的hash做key缓存向量,能省不少钱,尤其是文档没变、只是重新索引的时候。

7. 从能跑到好用:性能与成本的平衡

7.1 批量与并行

链路里最耗时的两件事是embedding和模型生成。embedding可以批量,模型生成可以流式。流式输出对用户体验提升巨大,用户不用干等十几秒,看到字一个个蹦出来心理感受完全不一样。

检索本身很快(numpy矩阵乘法),但如果文档量到了百万级,暴力检索就慢了。这时候才需要考虑向量索引(比如HNSW)。但我要提醒:十万条以下的文档,暴力检索完全够用,别过早优化。

7.2 缓存策略

三层缓存我都在用:

  • embedding缓存:文本hash → 向量。
  • 检索缓存:query hash → 命中块ID。
  • 回答缓存:query hash → 完整回答(适合高频重复问题)。

回答缓存要小心,如果知识库更新了,缓存得失效。我一般给缓存设个TTL,或者知识库更新时主动清空。

7.3 什么时候该上框架

自己写跑通之后,如果项目要快速扩展,可以考虑引入框架。但我的建议是只引入你需要的部分,比如只用它的向量库抽象,编排还是自己写。全盘交给框架,等于把前面辛苦建立的调试能力和成本感知又交出去了。

8. 我个人的一点体会

这套东西我从头到尾自己实现过两遍,第一遍是为了搞懂原理,第二遍是为了优化线上效果。最大的收获不是代码本身,而是对每个环节的“手感”——切分到什么粒度合适、Top-K取多少、阈值设在哪,这些没有标准答案,只有自己动手调过才有直觉。

如果你也想做一遍,我的建议是:先跑通最小闭环,再逐个环节优化。别一上来就追求完美,先让“文档进、答案出”这条链路通起来,哪怕效果一般。通了之后,用黄金测试集量化效果,然后一个环节一个环节地调。这个过程走完,你对AI工程的理解会跟只会调包的人拉开一个身位。

最后分享一个小技巧:把你调参的过程记成笔记,每个参数改动前后的效果都记下来。过一段时间回头看,你会发现哪些参数是真正重要的,哪些只是心理安慰。这份笔记,比任何教程都值钱。

返回列表