在做 AI 应用产品化的那段时间我最大的感受是模型能力早就不是瓶颈了真正的瓶颈是把 Embeddings、向量检索、知识库、推荐这些零散能力串成一条可靠的产品链路。恰好前阵子我负责的一个内部项目就是用 Ace Data Cloud 接入 OpenAI Embeddings把 RAG 知识问答、语义搜索和“猜你喜欢”式的推荐一次性落地。这篇文章就围绕这个项目展开从底层的方案选型、数据准备到 RAG、语义搜索、向量化推荐的具体实现再到线上性能和成本控制完整梳理一遍。如果你正准备做一个带“知识库问答”或“语义搜索”的 AI 产品这篇文章应该能帮你少走不少弯路。1. 先把问题想清楚为什么是“Ace Data Cloud OpenAI Embeddings”这个组合1.1 底层逻辑AI 应用真正缺的是“记忆系统”把 OpenAI Embeddings 接进来这件事本身并不复杂复杂的是它之后的存储、检索和更新。Embeddings 接口做的事情很纯粹你给它一段文本它返回一串浮点数。比如text-embedding-3-small默认返回 1536 维向量。关键问题是这 1536 个浮点数拿到之后放哪、怎么查、怎么和业务数据关联如果只是把它们丢在一个普通数据库的 JSON 字段里那每来一个查询请求就得全表扫描一遍再算余弦相似度数据量上千条就开始卡了。Ace Data Cloud 在这个链路里扮演的角色就是那个“记忆系统”。它本质上是一个云数据库服务支持标准 SQL同时内置了向量类型和相似度检索能力。也就是说你可以像建一张普通业务表一样建一张向量表然后直接执行“按语义相似度排序”的查询。这就把“向量化”和“业务数据”放到了同一个存储底座上不用再单独引一套向量数据库也不用自己搭 faiss 服务。我习惯用一个类比来解释这个组合Embeddings 接口是“图书管理员”它负责给每本书文本写索引卡Ace Data Cloud 则是“图书馆”它决定这些索引卡怎么归档、怎么按主题快速找出相关书籍。光有图书管理员没有图书馆书还是堆在地上光有图书馆没有索引卡找书依然只能靠肉眼翻。1.2 为什么不用“纯向量数据库”或“自建方案”很多团队一提到向量检索第一反应是上 Pinecone、Milvus 或者 Weaviate 这类专用向量数据库。确实它们在千万级向量检索上做得非常极致但如果你的项目还处于“快速产品化”阶段引入一个独立向量库意味着多一套基础设施、多一种运维视角、多一层数据同步。数据先存在业务库里再异步同步到向量库两边容易不一致出了问题排查链路还长。自建 faiss 的问题更直接faiss 只是一个计算库不是存储系统。索引保存在内存里服务一重启索引就没了得重建多副本、高可用、权限控制、慢查询监控全要自己写。我见过不少团队花了三周把 faiss 服务跑起来又花了一个月补各种周边能力最后还是回到云数据库。Ace Data Cloud 最打动我的点是它把向量检索做成了 SQL 语法的一部分。这意味着业务团队不需要学习一套全新的查询语言写WHERE category 财经 ORDER BY embedding - :query_vec LIMIT 10就能带上结构化过滤条件做语义检索。尤其在 RAG 场景里文档往往要按部门、时间、标签做权限过滤纯向量库要做到这种“结构化条件 向量相似度”混查通常要额外维护过滤器而 SQL 天然支持这一点。这也是我坚持“数据库优先而不是向量库优先”的根本原因。2. 动手之前环境准备、数据清洗与 Embeddings 批量生成2.1 需要准备的材料项目开始前我先把所需资源列了一个清单避免做一半发现缺这缺那一个 Ace Data Cloud 实例。在控制台创建实例后会拿到连接信息。Ace Data Cloud 同时支持 JDBC 和 HTTP API 方式访问我用的是 JDBC 方式配合 Python 的ace-db-connector包也可以直接用psycopg或pymysql风格的驱动取决于你创建实例时的兼容模式。一个 OpenAI API Key。注意调用 Embeddings 接口和 Chat 接口建议使用两个独立的 Key方便分别统计用量和设置预算上限。如果公司有统一的网关也可以走网关但需要确认网关是否支持text-embedding-3-small这类模型。Python 3.10 以上的开发环境安装openai、pandas、ace-db-connector这几个依赖。一份待处理的数据集。我这里是一批产品说明文档和 FAQ大概 1200 篇每篇平均 2500 字左右。后面为了测试语义搜索又加了一批商品描述数据。2.2 数据清洗的几条硬规矩不要以为拿到文档就能直接扔给 Embeddings 接口。数据质量直接决定检索效果我在这个项目里最深的体会就是Embeddings 很诚实你喂给它一段混着 HTML 标签、乱码、广告尾巴的文本它就把这些噪音一并编码进向量里检索时这些噪音反而会成为干扰。清洗阶段我只做了四件事但每一件都很关键。第一统一编码把所有的GBK、ISO-8859-1内容转成UTF-8。第二去除网页标签用BeautifulSoup把 HTML 标签、script块、style块、导航栏文案全部剥离。第三合并断行很多从 PDF 或 Word 导出的文本有大量人工换行不合并的话切块会切得乱七八糟。第四去重和去敏感词尤其是把手机号、身份证号、内部系统路径这些信息提前筛掉避免 Embedding 后这些信息被检索出来流到外部。清洗完之后我还做了一次人工抽样检查。1200 篇文档里随机抽了 80 篇人工快速浏览确认没有明显乱码和错位。这个步骤很费时间但必须做因为一旦脏文本进入了向量库后面清理的成本是清洗阶段的十几倍。2.3 批量生成 Embeddings 的代码实现批处理部分我写了一个独立脚本放在scripts/embed_docs.py。核心逻辑是读清洗后的文本切块调用 OpenAI Embeddings 接口把结果存成本地 parquet 文件。为什么不直接写进数据库因为我想先保留一份原始向量结果方便后面反复调试建表语句和检索参数不用每次重算 Embeddings。import time import hashlib import pandas as pd from openai import OpenAI client OpenAI(api_keysk-你的key) def embed_texts(texts: list[str], model: str text-embedding-3-small) - list[list[float]]: vectors [] batch_size 64 # 实测 64 个一批比较稳太大容易超时 for i in range(0, len(texts), batch_size): batch texts[i:i batch_size] for attempt in range(5): try: resp client.embeddings.create(modelmodel, inputbatch) # 注意返回顺序与输入顺序一致 vectors.extend([item.embedding for item in resp.data]) break except Exception as e: wait 2 ** attempt print(f第 {attempt} 次重试等待 {wait}s{e}) time.sleep(wait) else: raise RuntimeError(f批次 {i} 在 5 次重试后仍失败) time.sleep(0.2) # 轻微限速避免触发 429 return vectors # 读取清洗后的文档 df pd.read_parquet(cleaned_docs.parquet) # 切块逻辑见下一节这里先占位 chunks [文档1的第1块, 文档1的第2块, ...] result embed_texts(chunks) # 保存结果留作后续入库 vec_df pd.DataFrame({ chunk_id: [hashlib.md5(c.encode()).hexdigest() for c in chunks], content: chunks, embedding: [json.dumps(v) for v in result], }) vec_df.to_parquet(doc_vectors.parquet, indexFalse)openai库在 1.x 版本里已经全面转向client.embeddings.create注意不要拿 0.x 版本的旧代码硬套。批量参数我一开始设了 128结果经常触发超时改成 64 之后稳定很多。另外虽然 5 次重试看起来足够但在实际运行中遇到了两次连续 429我后来在重试逻辑里加了随机抖动time.sleep(wait random.uniform(0, 1))效果好不少。3. 核心落地之一RAG 知识库的 Chat 问答3.1 RAG 到底是什么RAGRetrieval-Augmented Generation的中文叫“检索增强生成”。它解决的核心问题是大模型不懂你的私有知识但你可以把相关知识临时“喂”给它让它看着资料回答。这和让模型凭空背诵是不同的逻辑。你不需要把整本产品手册塞进模型上下文而是先判断用户问题涉及哪一部分再把那一部分检索出来拼接成 Prompt 交给模型。完整的链路是这样文档切块 → 向量化 → 存入 Ace Data Cloud → 用户提问 → 问题向量化 → 向量检索 top-k 相关块 → 拼装 Prompt → 调用 OpenAI Chat 模型生成回答。唯一稍复杂的环节就是“检索出相关块”这一步其他环节都比较常规。这个方案最大的优点是天然支持知识更新。文档改了只需要重新写入对应块不需要重新训练模型。这就是为什么 RAG 项目在企业内部那么受欢迎——产品文档每天都在变但模型不可能每天重新训练。3.2 在 Ace Data Cloud 中创建向量表向量表的设计直接影响后续的检索和过滤效率。我的建表语句长这样CREATE TABLE doc_chunks ( id VARCHAR(64) PRIMARY KEY, content TEXT NOT NULL, source VARCHAR(255) NOT NULL, -- 来源文档 category VARCHAR(64), -- 分类标签 author VARCHAR(64), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, embedding VECTOR(1536) NOT NULL ); CREATE INDEX idx_doc_chunks_embedding ON doc_chunks USING hnsw (embedding vector_cosine_ops);Ace Data Cloud 对向量的支持是原生的类型叫VECTOR(n)其中n必须和 Embeddings 模型的输出维度一致。text-embedding-3-small是 1536 维text-embedding-3-large是 3072 维。如果你之前有人用过 2.x 版本的模型它们的维度可能是 1536 或 2048最好在代码里显式声明模型版本避免混乱。索引部分我选择了hnsw和余弦距离。HNSW 是一种近似最近邻索引算法检索速度快适合在线查询场景。如果你的数据量在一个小范围里百万以内完全够用。写入数据时可以直接从 parquet 文件读出来逐批插入import ace_db_connector conn ace_db_connector.connect(jdbc:ace://your-instance-id, useradmin, password***) cursor conn.cursor() for _, row in vec_df.iterrows(): cursor.execute( INSERT INTO doc_chunks (id, content, source, category, embedding) VALUES (%s, %s, %s, %s, %s), (row[chunk_id], row[content], row[source], row[category], row[embedding]) ) conn.commit() cursor.close() conn.close()插入时需要注意一个坑很多数据库驱动默认不会把 Python 的list[float]自动转成向量类型必须显式传成字符串形式比如[0.001, -0.002, ...]否则会报类型不匹配。这个问题我在项目里卡了快一个小时后来查看连接器源码才发现它只识别字符串格式。3.3 检索与问答的代码实现检索的 SQL 看起来很简单但里面的细节不少SELECT id, content, source, 1 - (embedding :query_vec) AS similarity FROM doc_chunks WHERE category :category ORDER BY embedding :query_vec LIMIT 5;在 Ace Data Cloud 中表示余弦距离值越小越相似。1 - 距离就转换成了“相似度”方便业务侧直接判断阈值。上面这段 SQL 用到了category过滤这是 RAG 场景里非常实用的能力。用户问“报销流程”时我可以先在 SQL 里限定category 财务避免从整个知识库里去捞检索速度和准确率都会提升。拿到 top-k 块之后Prompt 拼装我采用的模板如下你是一个企业内部助手。请严格根据下面提供的资料回答问题。 如果资料中没有答案请直接说“根据当前知识库无法回答”。 资料 1. [来源财务手册] 报销流程分为三步... 2. [来源财务FAQ] 发票金额超过5000元需要... 问题差旅费报销需要提供哪些材料之所以要在每条资料前加“来源”是为了让模型在回答时能区分不同文档减少串味。最后设置temperature0.2让回答更忠实于资料不要自由发挥。3.4 切块大小的经验值RAG 效果不好一半的问题出在切块策略上。我一开始偷懒把整篇文章作为一个向量块结果用户问具体细节时检索出来的内容太长塞进 Prompt 后模型容易被无关信息干扰。后来我改成按“固定 token 窗口 重叠”的方式每块约 512 个 token约合中文 350 到 450 个字符。相邻块重叠 128 个 token避免关键信息恰好被切在边界上。用tiktoken库来做切块比较方便import tiktoken enc tiktoken.encoding_for_model(text-embedding-3-small) def split_text(text: str, chunk_size: int 512, overlap: int 128): tokens enc.encode(text) chunks [] for i in range(0, len(tokens), chunk_size - overlap): chunk tokens[i:i chunk_size] chunks.append(enc.decode(chunk)) return chunks切块之后最好人工检查几条边界内容看看是否出现语义被切断的情况。我遇到过一份合同文档因为段落标题刚好落在重叠区导致两个块各自都缺了半句话检索时这块内容基本不可用。后来我在切块函数里加了一个简单规则如果切出来的块以冒号或“第”字开头就前移几个 token让块以完整句子开头。这个小规则处理了 80% 的切块质量问题。4. 核心落地之二语义搜索4.1 语义搜索和关键词搜索的区别如果只是做关键词搜索用数据库里的LIKE %关键词%就行为什么还要引入向量我用一个真实例子说明。用户搜索“怎么报销手机话费”关键词搜索的逻辑是找包含“报销”“手机话费”这几个字的内容。但如果文档里写的是“通讯费用补贴申请流程”字面上一个关键词都不匹配关键词搜索就漏掉了。语义搜索不一样。它把“手机话费报销”编码成向量把“通讯费用补贴申请流程”也编码成向量两个向量在空间里距离很近于是能正确召回。这是语义搜索和关键词搜索最本质的区别它匹配的是“意思”不是“字面”。很多搜索产品最后是两者结合。“语义召回 关键词精排”是我比较推荐的做法。先通过向量把候选集扩大到 50 条再用 BM25 或者简单的关键词命中规则把真正包含用户查询词且语义也相关的结果排到前面。Ace Data Cloud 支持标准 SQL所以关键词和向量可以放在同一条 SQL 里配合使用。4.2 相似度计算方式向量之间的距离算法有三种常见选择余弦距离、欧几里得距离、内积。OpenAI 官方的 Embeddings 模型建议用余弦相似度因为该模型的向量经过了 L2 归一化处理余弦相似度和内积在数值上等价但余弦更直观取值范围在 -1 到 1 之间。在 Ace Data Cloud 里是余弦距离运算符。如果换成-运算符表示欧几里得距离#表示内积不同运算符对应不同的索引类型。因为我的建表语句里写的是vector_cosine_ops所以查询里也必须用否则索引会失效。4.3 让搜索结果更准的做法项目上线后我陆续优化了三轮检索效果。第一阶段是纯向量召回准确率大约 70%第二阶段加上元数据过滤比如限制发布时间、商品分类、作者准确率到了 82%第三阶段引入混合检索用关键词召回一批候选再做向量重排准确率稳定在 90% 左右。元数据过滤非常重要。比如用户搜索“降噪耳机”如果不过滤分类很可能把“隔音耳塞”“降噪风扇”的内容也召回。加上category 耳机的过滤条件后搜索结果立刻干净很多。SQL 的长处在这里体现得很充分你把过滤条件直接拼进 WHERE 子句就行。SELECT id, product_name, price, 1 - (embedding :query_vec) AS score FROM products WHERE category 耳机 AND price BETWEEN 200 AND 2000 AND is_on_sale true ORDER BY embedding :query_vec LIMIT 20;阈值也要设。我一般把相似度阈值设在 0.65 到 0.75 之间。相似度低于 0.65 的检索结果大概率是不相关的内容没必要展示给用户。这个阈值需要根据数据分布调不是固定值。方法也很简单抽 100 条用户真实查询人工标注“是否相关”画出准确率随阈值变化的曲线选曲线拐点。5. 核心落地之三向量化推荐5.1 从“物相似”到“人相似”推荐系统的本质是相似度匹配这和语义搜索其实是同一套数学工具。我在这个项目里实现了两种推荐逻辑Item-to-Item物品到物品和 User-to-Item用户到物品。Item-to-Item 最简单也最容易落地。把每个商品描述转成向量当用户点开一个商品详情页时系统查询该商品向量的最近邻返回相似商品。比如用户在看“便携式咖啡机”系统会推荐“手冲壶”“磨豆机”“咖啡滤纸”。这里没有用到任何用户历史行为纯靠商品文本的语义相似度冷启动阶段特别管用。User-to-Item 更进一步。把用户的历史点击、收藏、购买记录拼成一段文本比如“用户近期浏览了便携咖啡机、手冲壶、耶加雪菲咖啡豆”把这段行为日志文本向量化得到“用户兴趣向量”然后用这个向量去商品向量表里查询最近邻。关键在于行为日志的文本质量要高不能把偶尔误点的内容也拼进去否则用户兴趣向量会非常飘。5.2 实现 Item-to-Item 推荐的完整流程推荐部分的 SQL 和语义搜索几乎一样只是查询条件换成“当前商品 ID”SELECT p2.id, p2.product_name, p2.price, 1 - (p1.embedding p2.embedding) AS similarity FROM products p1 JOIN products p2 ON p1.id ! p2.id WHERE p1.id :current_product_id AND p2.is_on_sale true ORDER BY p1.embedding p2.embedding LIMIT 10;这段 SQL 的关键点是用self-join的方式把目标商品的向量和全部商品向量做距离计算。在小数据集几万条下这个查询可以接受但如果商品量达到百万级这种全表笛卡尔积式扫描会非常慢。优化方案有两种一是提前把每个商品的最近邻集合物化到一张表里离线算好线上直接读二是结合商品分类做预过滤只在同类目或同品牌下做最近邻搜索。我实际项目里的商品量只有 4 万左右所以直接用了self-join查询耗时在 300ms 左右完全能接受。物化最近邻的方案虽然查询速度快但商品上新后需要重新跑一次离线任务对于更新频繁的电商场景反而麻烦。推荐结果上线之前一定要做一次“相关性人工抽检”。模型说“相似”不一定是用户认为的“相似”。我抽检后发现一个典型问题所有黑色商品放在一起向量会把“黑色背包”和“黑色手机壳”判为相似因为它们共享了“黑色”“便携”“大容量”这些泛化语义。后来我在商品描述里增加了更细的品类词并在向量化之前把品牌词、颜色词做了权重调整这个问题才明显缓解。6. 产品化加速性能优化、成本控制与模型更新6.1 线上环境一定要做的性能优化从原型到产品最大的变化是并发请求多了。demo 阶段一个查询 1 秒钟没人关心线上每秒钟来 50 个查询排序就开始排队。我在性能优化上做了四件事。第一确认走索引。利用EXPLAIN看查询计划确认hnsw索引真正被用到。第二给高频查询加缓存。用户搜索词在短时间内往往会重复出现我用 Redis 缓存了查询向量和 top-10 结果缓存时间 5 分钟命中率大约 35%极大缓解了数据库压力。第三设置查询超时时间避免慢查询拖垮整个连接池。第四把 Embedding 接口调用从同步改成异步。用户提问时如果每次都实时调用 OpenAI Embeddings 接口那一次回答要等两次外部 API 调用一次向量化、一次生成体验很差。我的做法是维护一个“已向量化查询缓存表”把最近 10 万条用户查询和对应的向量存起来命中缓存就直接查库。关于 HNSW 参数Ace Data Cloud 提供了m和ef_construction的配置项。m越大索引精度越高但内存占用越大ef_construction越大建索引时搜索越充分。线上我用的是m 16, ef_construction 64查询时设置ef_search 40。如果你的数据更新频繁建议定期重建索引否则 HNSW 的图结构会因为大量删除操作而劣化。6.2 成本控制Embeddings 接口虽然便宜但积少成多。以text-embedding-3-small为例按官方定价的月度账单估算每百万 token 大约零点几美元到几美元不等具体取决于账号套餐和区域。我们的语料总量大约是 2 亿 token一次性全量向量化的成本可以接受但如果没有离线批处理改为公有业务每次访问都实时调用一个月下来费用会非常难看。我采用的成本控制策略有三个。第一离线批量任务和线上实时调用分离。文档更新定时跑批一次性算好向量存入库用户查询时优先走缓存只有新查询才实时调用 Embeddings。第二向量复用。同一段文本如果已经向量化过就直接从缓存表里取不重复调用。第三控制上下文长度。RAG 拼 Prompt 时如果 top-5 个块每个 512 token那就是 2560 token再加历史和指令一次请求轻松超过 3000 token。实际运营下来大部分问题其实用 top-3 就能回答得不错能省下不少 token 费用。6.3 模型升级与重算迁移OpenAI 会不定期更新 Embeddings 模型新模型的向量维度、质量都可能变化。最稳妥的做法是在新模型上重新全量向量化然后切换查询侧。我的迁移流程分三步。第一步在库里新建一张表doc_chunks_v2把新模型的向量写进去。第二步写一个对比脚本用 100 个典型查询分别打新旧两版库人工对比检索结果。第三步把应用程序里的模型名和查询表名一起切换最后删掉旧表。这套流程听起来麻烦但做一次之后以后模型再更新你就知道怎么操作了。有个容易被忽略的点新旧两个模型同时存在期间不要把旧表的数据同步到新表。因为向量维度和分布不同混着用会导致检索结果非常诡异。哪怕旧模型检索效果更好新模型也已经不支持了该迁移就得迁移。7. 常见问题与避坑经验7.1 问题速查表我整理了项目期间遇到的高频问题做成速查表问题现象原因解决办法插入向量报类型错误column embedding is of type vector but expression is of type text连接器把 list 当作字符串处理将向量列表显式转成[0.001, -0.002]再传入查询很慢数据量不大但每次检索要几百毫秒建索引时用了错误的距离算子确认vector_cosine_ops对应查询中的检索结果不相关返回内容与问题无关文档块切得太粗或太细检查切块长度调整为 512 token、重叠 128 并验证边界中文切块乱码回答中出现半个词或语义断裂tiktoken按 BPE 分词中文场景需要额外处理按句子边界调整切块位置避免从词中截断用户查询缓存 miss 太高实时调用 Embeddings 次数过多缓存维度单一没有区分大小写和长尾词对查询做归一化缓存键用归一化后的文本哈希推荐结果“表面相似”黑色手机壳和黑色背包互相推荐向量被颜色、材质等泛化词主导在文本中加强品类词权重降低泛化词权重HNSW 索引不生效EXPLAIN显示全表扫描查询条件或排序顺序导致索引无法使用将相似度排序放在最外层并把过滤条件改写为可索引形式7.2 从项目里总结的产品化经验这四条经验不是从某个文档里看来的是实际踩坑换来的。第一先打通全链路再优化细节。我见过很多团队一上来就纠结“切块是 256 还是 512”结果链路还跑不通做了一堆没用的调参。正确顺序是先实现一个粗糙但完整的链路哪怕检索准确率只有 60%也要让用户问上第一句话。链路打通之后再根据数据逐项优化。第二离线任务永远要比线上接口多一步验证。批量跑 Embeddings 时我遇到过源文档更新导致部分向量过期的脏数据问题。后来我加了一个校验步骤每个批次入库前抽样 10 条在测试环境验证“查得到、排得对”再继续。第三上线前一定要做人工抽检而且要让业务方参与。技术团队认为“语义相近”和业务团队认为“相关”往往不是一回事。我在做电商推荐时技术团队觉得两个商品描述相似业务看一眼就说“这款是男装那款是女装不能推荐”。这类业务常识只能靠业务方人工把关。第四日志和监控要早点接入。向量检索最怕的是“用户搜不到”和“搜错”但如果没有查询日志你根本发现不了问题。我在每个搜索请求里打了query_id、检索 top-10、用户点击了第几个结果、有没有采纳回答这样后面可以用一个简单的“检索召回率”指标持续评估线上效果。最后再说一个我个人的操作习惯不要把所有业务都塞进一张向量表。RAG 知识库、商品搜索、推荐系统这三种场景的查询模式和过滤条件差异很大混在一张表里会让索引参数、缓存策略和权限控制都变得混乱。拆成三张表各自建索引各自配缓存虽然看起来多建了表但线上出问题时排查起来省心得多。如果你也是第一次把 Embeddings 接入业务系统我建议你也从一个小场景开始先把 RAG 问答跑通再加语义搜索最后叠推荐能力。一条链路走通之后其他的都是复用。