21 Embedding 向量化
向量入库：把 Embedding 稳定写入 PostgreSQL + pgvector
向量入库的核心不是简单执行一次
UPDATE，而是保证 chunk、向量、模型信息和文档状态一致，并且在失败时能够完整回滚、重新执行和排查问题。
[TOC]
1. 本节要解决什么
前面的流程已经得到每个 chunk 对应的 1024 维向量：
chunkId
chunkIndex
vector
inputHash
embedding metadata
本节负责把这些数据可靠地写入 PostgreSQL：
向量 -> rag_chunk.embedding模型信息 -> rag_chunk.metadata.embedding完成状态 -> rag_document.status失败原因 -> rag_document.error_message
完成本节后，应能够理解：
pgvector 字段如何创建和校验
Java 如何把浮点数组写入 vector 字段
为什么使用 JDBC batch向量和 metadata 如何保持一致
为什么写入必须使用事务
重复执行是否会产生重复数据
失败后数据库会处于什么状态
如何验收整篇文档是否入库完整
HNSW 索引什么时候生效
生产环境还可以做哪些改进
本文示例采用以下配置：
Embedding 模型：text-embedding-v4
向量维度：1024
向量数据库：PostgreSQL + pgvector
距离度量：cosine distance
向量字段：rag_chunk.embedding
模型血缘：rag_chunk.metadata.embedding
2. 向量入库整体流程
graph TD;
A["收到当前文档的全部向量"] --> B["校验向量数量和维度"];
B --> C["打开数据库连接"];
C --> D["关闭自动提交"];
D --> E["JDBC batch 更新 rag_chunk"]; E --> F["检查每个 chunk 是否成功更新"];
F --> G["更新 rag_document.status = EMBEDDED"]; G --> H["提交事务"];
E -- "失败" --> I["回滚事务"];
F -- "chunk 不存在" --> I;
G -- "文档不存在" --> I;
模型调用和数据库写入应该分开：先得到完整向量结果，再把向量交给入库服务统一持久化。
3. pgvector 数据库准备
3.1 启用 vector 扩展
CREATE EXTENSION IF NOT EXISTS vector;
检查扩展：
SELECT extname, extversion
FROM pg_extension
WHERE extname = 'vector';
如果没有查询结果，说明当前 database 没有启用扩展。
注意：
pgvector 必须先安装到 PostgreSQL 服务
CREATE EXTENSION 需要在当前使用的 database 中执行
不同 database 的扩展启用状态相互独立
应用账号可能没有创建扩展的权限
3.2 创建向量字段
本文示例使用：
ALTER TABLE rag_chunk
ADD COLUMN IF NOT EXISTS embedding vector(1024);
vector(1024) 表示：
每条向量必须包含1024个浮点数
不能写入1023维或1536维向量
NULL表示这个chunk尚未完成向量化
维度必须在所有位置保持一致：
EMBEDDING_DIMENSION=1024
Python请求 dimensions=1024Python返回向量长度=1024
Java响应校验=1024
PostgreSQL字段 vector(1024)
3.3 检查字段实际维度
ADD COLUMN IF NOT EXISTS 只能保证字段存在，不能保证已有字段维度正确。
可以使用下面的系统表查询字段类型：
SELECT format_type(attribute.atttypid, attribute.atttypmod) AS column_type
FROM pg_attribute attribute
JOIN pg_class table_info
ON table_info.oid = attribute.attrelidWHERE table_info.relname = 'rag_chunk'
AND attribute.attname = 'embedding' AND attribute.attnum > 0 AND NOT attribute.attisdropped;```
本文示例的期望结果：
```text
vector(1024)
如果数据库是 vector(1536)，配置却是 1024，应用会拒绝继续启动，避免把错误留到入库阶段。
3.4 创建 HNSW 索引
CREATE INDEX IF NOT EXISTS rag_chunk_embedding_hnsw_idx
ON rag_chunk
USING hnsw (embedding vector_cosine_ops);
这里的 vector_cosine_ops 表示索引用于 cosine distance。
后续查询应使用匹配的运算符：
ORDER BY embedding <=> :query_embedding
LIMIT :top_k;
4. 入库数据结构
Java 可以使用一个简单的 record 表示待入库向量：
public record EmbeddingWriteItem(
UUID chunkId, int chunkIndex, List<Double> vector, String inputHash, String embeddingMetadataJson) {
}
字段含义：
| 字段 | 作用 |
|---|---|
chunkId |
定位 rag_chunk.id |
chunkIndex |
表示 chunk 在文档中的顺序 |
vector |
1024 维 Embedding 数组 |
inputHash |
最终向量化输入文本的 SHA-256 |
embeddingMetadataJson |
模型、维度、版本、时间等血缘信息 |
4.1 为什么需要 chunkId
向量必须与生成它的文本严格一一对应：
rag_chunk.id = chunkId
rag_chunk.content = 生成该向量的正文
rag_chunk.embedding = 该正文对应的向量
如果向量顺序错位，数据库仍可能写入成功，但后续检索会返回错误文本。这类错误比 SQL 报错更危险。
因此不能只依赖列表位置，必须携带稳定的 chunkId。
4.2 inputHash 的作用
向量化输入可能由标题和正文共同组成：
标题路径：产品手册 > 上传限制
正文：单个文件最大20MB。
Python 对最终输入计算 SHA-256：
inputHash = sha256(最终输入文本)
它可以用于：
确认向量对应的是哪一版文本
发现标题或正文是否发生变化
排查旧向量是否已经失效
以后实现未变化chunk跳过向量化
5. 向量写入 SQL
每个 chunk 使用下面的 SQL 更新：
UPDATE rag_chunk
SET embedding = CAST(? AS vector),
metadata = metadata || jsonb_build_object( 'embedding', CAST(? AS jsonb) ), updated_at = now()WHERE id = ?
AND document_id = ?;
四个参数依次是：
1. 向量文本
2. Embedding metadata JSON
3. chunkId
4. documentId
5.1 为什么同时使用 chunkId 和 documentId
只按 chunkId 更新虽然也能找到数据，但无法验证该 chunk 是否属于当前文档。
推荐条件：
WHERE id = ? AND document_id = ?
可以阻止下面这种错误：
当前任务处理文档A
请求数据中却混入文档B的chunkId
如果更新行数为 0，入库服务应认为 chunk 已不存在或归属不正确，并回滚整个事务。
5.2 Java 如何写入 vector
一种简单做法是先把 List<Double> 转成 pgvector 支持的文本格式：
[0.012,-0.035,0.087,...]
然后使用参数化 SQL：
chunkStatement.setString(1, vectorText(embedding.vector()));
数据库端转换：
CAST(? AS vector)
这样做的优点：
实现简单
仍然使用PreparedStatement参数绑定
不会把完整向量直接拼接进SQL语句
不需要额外的pgvector JDBC类型
生产项目也可以评估 pgvector Java/JDBC 类型或框架提供的 VectorStore 封装。
6. Embedding metadata 如何保存
本文示例没有单独建立以下列：
embedding_model
embedding_dimension
embedding_version
embedded_at
这些信息保存在 rag_chunk.metadata 的 embedding 对象中。
示例：
{
"source": "upload", "fileType": "pdf", "pageStart": 3, "pageEnd": 4, "embedding": { "provider": "dashscope-openai-compatible", "model": "text-embedding-v4", "dimension": 1024, "version": "v1", "normalized": false, "inputHash": "c0f2...", "embeddedAt": "2026-06-15T10:30:00+08:00" }}
6.1 JSONB 合并行为
写入 SQL：
metadata = metadata || jsonb_build_object(
'embedding', CAST(? AS jsonb))
它的行为是：
保留metadata中已有的source、页码、标题等顶层字段
新增embedding对象
如果embedding已经存在，则替换整个embedding对象
它不是把整个 metadata 清空后重写。
6.2 查询模型信息
查询模型名：
SELECT metadata #>> '{embedding,model}' AS embedding_model
FROM rag_chunk
WHERE id = :chunk_id;
查询维度：
SELECT (metadata #>> '{embedding,dimension}')::int AS embedding_dimension
FROM rag_chunk
WHERE id = :chunk_id;
查询输入摘要：
SELECT metadata #>> '{embedding,inputHash}' AS input_hash
FROM rag_chunk
WHERE id = :chunk_id;
6.3 JSONB 和普通列如何选择
学习项目可以使用 JSONB，便于扩展和减少改表。
如果以后经常按模型、版本或时间过滤，可以考虑拆成普通列：
ALTER TABLE rag_chunk
ADD COLUMN embedding_model varchar(128), ADD COLUMN embedding_version varchar(64), ADD COLUMN embedded_at timestamptz;
选择原则：
经常过滤、排序、索引的字段：普通列
主要用于追踪和展示的扩展信息：JSONB
本文后续示例仍以 metadata.embedding 为准。
7. 为什么使用 JDBC batch
如果一篇文档有 200 个 chunk，逐条执行会产生 200 次独立数据库往返。
批量更新可以这样实现：
for (EmbeddingWriteItem embedding : embeddings) {
chunkStatement.setString(1, vectorText(embedding.vector())); chunkStatement.setString(2, embedding.embeddingMetadataJson()); chunkStatement.setObject(3, embedding.chunkId()); chunkStatement.setObject(4, documentId); chunkStatement.addBatch();}
int[] updateCounts = chunkStatement.executeBatch();
优点：
减少数据库网络往返
统一管理参数
提高批量更新吞吐
便于在一个事务中检查结果
7.1 检查批量更新结果
还应检查 executeBatch() 返回的更新数量：
for (int updateCount : updateCounts) {
if (updateCount == 0) { throw new IllegalStateException("写入向量时发现 chunk 已不存在");
}}
目的是防止：
模型调用期间chunk被删除
documentId和chunkId归属不一致
上游传入了过期chunk
发现异常时不会继续把文档标记为 EMBEDDED。
8. 事务设计
向量入库应使用一个数据库事务：
connection.setAutoCommit(false);
// 批量更新全部 chunkchunkStatement.executeBatch();
// 更新文档状态
documentStatement.executeUpdate();
connection.commit();
出现异常：
connection.rollback();
8.1 事务保护什么
事务保证以下操作要么全部成功，要么全部失败：
所有rag_chunk.embedding写入
所有metadata.embedding写入
rag_document.status更新为EMBEDDED
rag_document.error_message清空
不会出现：
只写入一半chunk，文档却已经是EMBEDDED
文档状态成功，但向量仍然为空
向量写入成功，但metadata没有对应模型信息
8.2 为什么模型调用不放进事务
Embedding API 可能需要数秒甚至更久，还可能发生限流和网络重试。
如果先开启事务再调用模型：
数据库连接会被长时间占用
事务持续时间过长
发生锁竞争的概率增加
外部接口失败会让事务白白等待
推荐顺序是：
先在事务外生成全部向量
再打开短数据库事务完成持久化
8.3 事务粒度
第一版可以把事务粒度设为一整篇文档：
一篇文档的全部chunk -> 一个写入事务
适合：
学习项目
普通大小文档
需要简单一致性保证
不希望半篇文档进入检索
大文档或高吞吐生产环境可以采用版本隔离、分批提交和最终切换可见版本，但设计会复杂很多。
9. 文档状态更新
全部 chunk 写入成功后执行：
UPDATE rag_document
SET status = 'EMBEDDED',
error_message = NULL, updated_at = now()WHERE id = ?;
如果文档不存在，更新行数为 0，事务会回滚。
9.1 第一版可以不增加 chunk 级状态
第一版可以不增加以下 chunk 级状态：
PENDING_EMBED
EMBEDDING
EMBEDDED
EMBED_FAILED
判断单个 chunk 是否有向量，使用：
embedding IS NOT NULL
判断整篇文档是否允许参与检索，还要结合：
rag_document.status = 'EMBEDDED'
推荐检索条件：
WHERE d.status = 'EMBEDDED'
AND c.embedding IS NOT NULL
9.2 为什么两个条件都需要
重新处理一个已经有旧向量的文档时，如果新模型调用失败：
旧embedding可能仍然存在
文档状态会被标记为FAILED
如果检索只检查 embedding IS NOT NULL，可能继续召回旧向量。
所以文档状态是控制整篇文档是否对检索可见的重要开关。
10. 幂等和重复执行
本文使用 UPDATE，而不是向量表 INSERT：
UPDATE rag_chunk
SET embedding = ...
WHERE id = ? AND document_id = ?;
重复执行同一个文档时：
不会新增重复chunk
不会新增第二条向量记录
原embedding被新embedding覆盖
原metadata.embedding被新版本替换
其他metadata顶层字段继续保留
因此这种数据库写入方式具有覆盖式幂等性。
10.1 模型调用去重
如果每次重新处理都直接请求 Embedding 模型，即使 chunk 文本没有变化，也会产生重复调用。
已经保存的 inputHash 可以用于后续优化：
重新计算当前输入Hash
读取旧metadata.embedding.inputHash
比较模型名、版本、维度和Hash
完全一致则跳过模型调用
判断条件不能只看 Hash，还要包含：
provider
model
dimension
version
normalize配置
文本相同但模型变化，仍然必须重新生成向量。
11. 失败处理
11.1 数据库写入失败
常见原因：
数据库连接失败
vector扩展不存在
向量维度不匹配
metadata不是合法JSON
chunk已被删除
文档已被删除
账号没有UPDATE权限
发生异常时：
向量写入事务回滚
不会保留本次部分写入
上层服务捕获异常
尝试把rag_document标为FAILED
error_message记录根异常摘要
11.2 标记失败 SQL
UPDATE rag_document
SET status = 'FAILED',
error_message = ?, updated_at = now()WHERE id = ?;
标记失败本身使用新的数据库连接，不属于已经回滚的写入事务。
11.3 错误信息长度
建议限制写入数据库的错误信息长度，避免异常堆栈或远程响应无限增长。
数据库中建议保存：
稳定错误码
简洁错误摘要
最后失败时间
必要的requestId
完整堆栈放日志系统，不要全部塞进业务字段。
12. 入库状态查询
可以使用下面的 SQL 汇总一篇文档：
SELECT count(*) AS total_count,
count(embedding) AS embedded_count, max(metadata #>> '{embedding,model}') AS model, max((metadata #>> '{embedding,dimension}')::int) AS dimension, max(updated_at) FILTER (WHERE embedding IS NOT NULL) AS last_embedded_atFROM rag_chunk
WHERE document_id = ?;
返回：
totalChunkCount
embeddedChunkCount
model
dimension
lastEmbeddedAt
12.1 如何判断入库完整
最基本条件：
totalChunkCount > 0
totalChunkCount == embeddedChunkCount
model == text-embedding-v4
dimension == 1024
rag_document.status == EMBEDDED
不能只判断文档状态，也不能只判断向量数量。
12.2 为什么使用 count(embedding)
PostgreSQL：
count(*)
统计所有行，而：
count(embedding)
只统计 embedding IS NOT NULL 的行。
两者差值就是尚未写入向量的 chunk 数量。
13. 向量样本查询
为了调试，可以查询少量向量样本：
SELECT id,
chunk_index, content, embedding::text AS vector_text, metadata #>> '{embedding,inputHash}' AS input_hashFROM rag_chunk
WHERE document_id = ?
AND embedding IS NOT NULLORDER BY chunk_index
LIMIT ?;
13.1 为什么只返回向量预览
一条向量有 1024 个数字。完整返回多条向量会：
增加接口响应体
降低前端可读性
刷满日志
没有实际展示价值
接口可以只取前 8 个数字作为预览，用于确认：
向量不是空值
格式可以解析
chunk和向量已经关联
完整向量应保留在数据库中，不应作为普通业务接口数据频繁返回。
14. HNSW 索引与写入的关系
建立 HNSW 后，每次写入或更新 embedding 时，PostgreSQL 还需要维护向量索引。
因此 HNSW 会带来：
更快的近似向量查询
更多磁盘占用
更高的写入成本
更长的初始索引构建时间
14.1 学习阶段是否必须建索引
不是。
没有 HNSW，下面的查询仍然可以执行精确检索：
SELECT id, content,
embedding <=> :query_embedding AS distanceFROM rag_chunk
WHERE embedding IS NOT NULL
ORDER BY embedding <=> :query_embedding
LIMIT 5;
数据少时，顺序扫描甚至可能更合适。
14.2 大批量导入策略
大量历史数据首次导入时，常见策略是：
先写入全部向量
再创建HNSW索引
执行ANALYZE
测试查询计划和召回质量
学习项目为了使用方便，可以提前创建索引。
14.3 检查查询计划
EXPLAIN (ANALYZE, BUFFERS)
SELECT id, content
FROM rag_chunk
WHERE embedding IS NOT NULL
ORDER BY embedding <=> :query_embedding
LIMIT 5;
表很小时没有使用 HNSW 不一定是问题，优化器可能认为顺序扫描成本更低。
15. 手动验收 SQL
15.1 检查 vector 扩展
SELECT extname, extversion
FROM pg_extension
WHERE extname = 'vector';
15.2 检查向量字段
SELECT format_type(attribute.atttypid, attribute.atttypmod) AS column_type
FROM pg_attribute attribute
JOIN pg_class table_info
ON table_info.oid = attribute.attrelidWHERE table_info.relname = 'rag_chunk'
AND attribute.attname = 'embedding' AND attribute.attnum > 0 AND NOT attribute.attisdropped;
期望：
vector(1024)
15.3 检查索引
SELECT indexname, indexdef
FROM pg_indexes
WHERE tablename = 'rag_chunk'
AND indexname = 'rag_chunk_embedding_hnsw_idx';
15.4 检查向量完整度
SELECT
document_id, count(*) AS total_chunks, count(embedding) AS embedded_chunks, count(*) - count(embedding) AS missing_embeddingsFROM rag_chunk
WHERE document_id = :document_id
GROUP BY document_id;
15.5 检查实际维度
SELECT
chunk_index, vector_dims(embedding) AS dimensionFROM rag_chunk
WHERE document_id = :document_id
AND embedding IS NOT NULLORDER BY chunk_index;
所有结果都应为：
1024
15.6 检查模型血缘
SELECT
chunk_index, metadata #>> '{embedding,provider}' AS provider, metadata #>> '{embedding,model}' AS model, (metadata #>> '{embedding,dimension}')::int AS dimension, metadata #>> '{embedding,version}' AS version, metadata #>> '{embedding,inputHash}' AS input_hash, metadata #>> '{embedding,embeddedAt}' AS embedded_atFROM rag_chunk
WHERE document_id = :document_id
ORDER BY chunk_index;
15.7 检查文档状态
SELECT
id, original_name, status, error_message, updated_atFROM rag_document
WHERE id = :document_id;
成功标准：
status = EMBEDDED
error_message IS NULL
missing_embeddings = 0
所有维度 = 1024所有模型 = text-embedding-v4所有inputHash非空
16. 生产化改进
前面的实现适合学习和普通文档流程。生产环境可以继续增强。
16.1 使用 DataSource 和连接池
基础示例可能直接使用：
DriverManager.getConnection(...)
生产环境更适合复用 Spring Boot 的：
DataSource
HikariCP
JdbcTemplate
NamedParameterJdbcTemplate
这样可以获得连接复用、超时控制、连接池指标和统一事务管理。
16.2 使用 Spring 事务
基础 JDBC 示例通常手动控制：
setAutoCommit(false)
commit
rollback
后续可以通过：
@Transactional
配合 Spring JDBC 管理事务，但要确保外部 Embedding 调用仍然发生在数据库事务之外。
16.3 使用 Flyway 管理 DDL
学习阶段可以在应用启动时检查扩展、字段和索引。
生产环境建议：
V1 创建基础RAG表
V2 启用vector并增加embedding字段
V3 创建HNSW索引
应用启动只做校验，不自动修改数据库结构。
16.4 增量向量化
利用 inputHash、模型和版本信息，只重新处理发生变化的 chunk：
输入Hash相同 + 模型相同 + 维度相同 + 版本相同
-> 复用旧向量
任一不同
-> 重新生成并覆盖向量
16.5 向量版本表
如果需要同时保留多个模型或版本，不应继续覆盖一个字段。
可以建立：
CREATE TABLE rag_chunk_embedding (
id uuid PRIMARY KEY, chunk_id uuid NOT NULL REFERENCES rag_chunk(id) ON DELETE CASCADE, provider varchar(64) NOT NULL, model varchar(128) NOT NULL, version varchar(64) NOT NULL, dimension integer NOT NULL, embedding vector(1024) NOT NULL, input_hash char(64) NOT NULL, created_at timestamptz NOT NULL DEFAULT now(), UNIQUE (chunk_id, model, version));
这是多模型版本需求下的设计，不是第一版必须完成的内容。
16.6 大文档分批持久化
整篇文档一次提交比较简单，也容易保证一致性。
超大文档可以考虑：
按批次写入新版本向量
旧版本继续对外可见
全部批次完成后原子切换当前版本
失败时删除未完成的新版本
不要简单地“每批写一点并立即让它参与检索”，否则用户可能搜到半篇已更新、半篇未更新的数据。
17. 常见问题
17.1 向量入库会再次调用模型吗
入库层本身不负责模型推理。它接收已经生成的向量，完成校验、持久化和状态更新。
业务接口可以连续完成模型调用和向量入库，但入库组件本身仍然只负责存储。
17.2 embedding 为什么允许 NULL
chunk 创建时还没有向量，所以字段需要允许 NULL。
NULL：尚未向量化
非NULL：已有可用于距离计算的向量
17.3 为什么不用 INSERT
向量直接属于已有 rag_chunk 时，使用 UPDATE 最简单，并且不会产生重复向量行。
只有需要保留多模型、多版本时，才更适合拆分独立向量表并使用 INSERT。
17.4 metadata 合并会丢失页码吗
不会。示例 SQL 只替换顶层 embedding 键，其他顶层 metadata 保留。
17.5 修改为其他维度后为什么不能写入
模型输出、Java 配置和数据库字段必须完全一致。更换维度需要：
停止旧写入
迁移或重建embedding字段
重新生成所有chunk向量
重建HNSW索引
确保查询向量使用相同模型和维度
17.6 为什么不用每个 chunk 一个事务
每条一个事务会增加提交开销，并可能让整篇文档进入部分完成状态。第一版建议整篇文档原子写入，优先保证一致性。
17.7 HNSW 索引创建后为什么查询仍然顺序扫描
可能原因：
表数据量太少
查询没有ORDER BY距离并LIMIT
距离运算符与索引operator class不匹配
统计信息尚未更新
过滤条件改变了优化器的成本判断
应使用 EXPLAIN ANALYZE 判断，不要只看索引是否存在。
18. 练习清单
完成以下练习：
执行vector扩展和vector(1024)字段初始化
查询并确认字段实际维度
解释List<Double>如何转换并写入pgvector
解释metadata || jsonb_build_object的合并行为
写出文档向量完整度检查SQL
写出模型血缘查询SQL
解释JDBC batch的作用
解释事务保护了哪些数据一致性
说明重复执行为什么不会增加重复行
说明为什么检索要同时检查文档状态和embedding非空
使用EXPLAIN ANALYZE观察向量查询计划
19. 小结
向量入库的完整数据链路：
1024维向量
-> 按chunkId定位rag_chunk
-> CAST(? AS vector) -> JDBC batch批量更新
-> metadata.embedding保存模型血缘
-> 检查所有chunk更新结果
-> rag_document.status更新为EMBEDDED
-> 同一事务提交
本节最重要的结论：
向量入库不是把一串数字塞进数据库，而是让向量、原文、模型版本、处理状态和事务结果始终保持一致。