CAMEL 多智能体框架中的 PgVectorStorage：基于 PostgreSQL pgvector 的向量存储实战指南

CAMEL 多智能体框架中的 PgVectorStorage：基于 PostgreSQL pgvector 的向量存储实战指南

【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel

导读

本文围绕 CAMEL 框架中基于 PostgreSQL pgvector 扩展实现的向量存储组件PgVectorStorage（API 参考见 docs/reference/camel.storages.vectordb_storages.pgvector.md），系统讲解其初始化参数、表结构与 HNSW 索引的自动管理、批量写入、按 ID 删除、三种距离度量下的相似性查询，以及状态统计与连接生命周期管理。读完本文，你将掌握如何在 CAMEL 的多智能体应用中直接使用 pgvector 承载 RAG 检索、语义缓存等场景下的向量数据，并理解每个方法背后的 SQL 实现与相似度换算逻辑。

一、PgVectorStorage 在 CAMEL 向量存储生态中的定位

CAMEL 的camel.storages模块以抽象基类BaseVectorStorage（定义于 camel/storages/vectordb_storages/base.py）统一了各类向量数据库的接入契约，其核心抽象方法包括：

• add(records)：批量保存VectorRecord
• delete(ids)：按 ID 删除向量
• query(query)：按相似度检索
• status()：返回维度与数量
• clear()：清空存储
• load()：加载云端集合（对本地型数据库通常为空操作）
• client属性：暴露底层客户端对象

在该契约之下，camel/storages/vectordb_storages/init.py 同时导出了 Chroma、Qdrant、Milvus、FAISS、Weaviate、TiDB、OceanBase、Surreal 以及本文主角PgVectorStorage等多种实现，并在 camel/storages/init.py 中对外统一暴露。PgVectorStorage是其中面向 PostgreSQL 生态的实现——它借助 pgvector 扩展，让开发者无需引入独立向量数据库即可在关系型数据库中完成向量检索，从而复用 PostgreSQL 成熟的备份、权限与运维体系。

二、运行环境与依赖准备

PgVectorStorage的构造函数通过@dependencies_required('psycopg', 'pgvector')装饰器（实现在 camel/utils/commons.py）在实例化时强制校验两个关键依赖，缺少任一模块都会抛出ImportError。

项目在 pyproject.toml 中对这两个依赖的版本约束为：

"psycopg[binary]>=3.1.18,<4", "pgvector>=0.2.4,<0.3",

其中psycopg是 PostgreSQL 的 Python 驱动（binary 版内置二进制依赖，免编译），pgvector则是 pgvector 扩展的 Python 适配层，用于把List[float]与数据库的vector类型互转。安装方式：

pip install "psycopg[binary]>=3.1.18,<4" "pgvector>=0.2.4,<0.3"

除 Python 依赖外，目标 PostgreSQL 实例本身需要启用 pgvector 扩展：

CREATE EXTENSION IF NOT EXISTS vector;

三、初始化：五个关键参数

PgVectorStorage.__init__的签名如下：

def __init__( self, vector_dim: int, conn_info: Dict[str, Any], table_name: Optional[str] = None, distance: VectorDistance = VectorDistance.COSINE, **kwargs: Any ) -> None:

各参数含义与源码行为（见 camel/storages/vectordb_storages/pgvector.py）如下表：

VectorDistance枚举提供三种距离度量：

class VectorDistance(Enum): DOT = "dot" # 点积（内积） COSINE = "cosine" # 余弦相似度 EUCLIDEAN = "euclidean" # 欧氏距离

初始化过程的内部顺序（源码第 68-86 行）为：建立psycopg.connect连接 → 调用register_vector(self._conn)注册 pgvector 类型适配 → 依次执行_ensure_table()与_ensure_index()自动建表建索引 → 任一环节失败则记录错误日志并向上抛出异常。

典型初始化示例：

from camel.storages import PgVectorStorage from camel.types import VectorDistance storage = PgVectorStorage( vector_dim=1536, # 与你的 embedding 模型输出维度一致 conn_info={ "host": "localhost", "port": 5432, "dbname": "camel_db", "user": "postgres", "password": "your_password", }, table_name="agent_vectors", distance=VectorDistance.COSINE, )

四、表结构与索引的自动管理

4.1_ensure_table：幂等建表

构造函数会自动执行建表逻辑（源码第 88-110 行），使用psycopg.sql.SQL组合参数化 SQL，表名与维度均通过Identifier/Literal安全转义，避免 SQL 注入：

CREATE TABLE IF NOT EXISTS {table} ( id VARCHAR PRIMARY KEY, vector vector({dim}), payload JSONB )

可见每张向量表由三列构成：

• id：主键，字符串类型，对应VectorRecord.id；
• vector：pgvector 的vector类型列，维度在建表时固化；
• payload：JSONB 类型，用于存储任意元数据（来源文本、标题、标签等），查询时随结果一并返回。

4.2_ensure_index：自动创建 HNSW 索引

建表后紧接着创建近似最近邻搜索索引（源码第 112-132 行）：

CREATE INDEX IF NOT EXISTS {table}_vector_idx ON {table} USING hnsw (vector vector_cosine_ops)

需要说明的几点：

• 索引使用 HNSW（分层可导航小世界图）算法，适合大规模向量的近似检索；
• 索引与distance参数存在耦合：源码中_ensure_index固定使用vector_cosine_ops算子类，因此当前实现更匹配默认的COSINE度量。若你在query阶段改用欧氏或点积度量，从源码结构看索引算子类与度量之间可能不完全匹配，建议按实际度量手工调整索引，或在确定度量后保持配置一致；
• 与建表不同，索引创建失败仅记录logger.warning而不会中断初始化（建表失败则会抛出异常），这是因为索引属于性能优化手段，其缺失不应阻塞核心写入/查询能力。

五、数据写入：add

add(records: List[VectorRecord], **kwargs)用于新增或更新向量记录（源码第 134-181 行），行为要点：

1. 空列表短路：records为空时直接返回，不产生任何 SQL 执行；
2. 维度校验：逐条检查len(rec.vector) != self.vector_dim，不一致立即抛出ValueError；
3. 批量插入：将记录组装成(id, vector, payload_json)元组列表，payload 为None时写入None，否则json.dumps序列化；
4. UPSERT 语义：使用ON CONFLICT (id) DO UPDATE SET vector=EXCLUDED.vector, payload=EXCLUDED.payload，因此重复写入相同id会覆盖旧向量与元数据，天然支持"增改合一"；
5. 事务管理：批量执行executemany后统一commit()，失败时rollback()并抛出异常。

VectorRecord定义在 camel/storages/vectordb_storages/base.py，是一个 Pydantic 模型：vector: List[float]必填；id缺省时自动生成随机 UUID；payload: Optional[Dict[str, Any]]可选。写入示例：

from camel.storages import VectorRecord records = [ VectorRecord( id="doc-001", vector=[0.1, 0.2, 0.3, 0.4], # 长度必须等于 vector_dim payload={"title": "CAMEL 简介", "source": "docs/intro.md"}, ), VectorRecord( vector=[0.5, 0.6, 0.7, 0.8], # 未指定 id，自动生成 UUID ), ] storage.add(records)

六、数据删除：delete

delete(ids: List[str], **kwargs)按 ID 批量删除（源码第 183-204 行），使用 PostgreSQL 数组参数化删除，同样在空列表时短路：

DELETE FROM {table} WHERE id = ANY(%s)

调用方式：

storage.delete(["doc-001", "doc-002"])

删除失败会回滚事务并抛出异常，保证数据一致性。

七、相似性查询：query

query(query: VectorDBQuery, **kwargs)是检索核心（源码第 206-287 行）。VectorDBQuery封装了query_vector: List[float]与top_k: int（默认 1）两个字段。查询前会先校验查询向量维度，不一致即抛ValueError。

7.1 三种距离度量到 SQL 算子的映射

生成的查询 SQL 为：

SELECT id, vector, payload, (vector {metric} %s::vector) AS score FROM {table} ORDER BY score {order} LIMIT %s

其中查询向量以参数形式绑定，top_k控制返回条数。

7.2 距离分数到相似度（越高越好）的换算

数据库返回的是"距离"分数，CAMEL 通过_score_to_similarity（源码第 279-287 行）统一转换为 0~1 区间的高分即相似，方便上层 RAG 管线直接使用：

7.3 返回结果

查询返回List[VectorDBQueryResult]，每个结果由record: VectorRecord（含 id、原始向量、payload）与similarity: float组成，按相似度从高到低排序。完整查询示例：

from camel.storages import VectorDBQuery results = storage.query( VectorDBQuery(query_vector=[0.1, 0.2, 0.3, 0.4], top_k=5) ) for r in results: print(r.record.id, r.record.payload, r.similarity)

此外，基类还提供了便捷方法get_payloads_by_vector(vector, top_k)，它内部调用query并只返回非空 payload 列表，适合"只取元数据"的检索场景。

八、状态、清理与生命周期管理

8.1status：查询库内统计

status()返回VectorDBStatus对象（vector_dim与vector_count两个字段），内部执行SELECT COUNT(*) FROM {table}，例如：

status = storage.status() print(status.vector_dim, status.vector_count)

8.2clear：清空全部数据

clear()直接执行TRUNCATE TABLE {table}快速清空整张表（注意：TRUNCATE不可按条件筛选，会删除该表全部向量），随后提交事务。

8.3load：接口兼容的空操作

load()为空操作（源码第 332-337 行），注释明确说明"对于 PostgreSQL 本地/托管实例无需加载"，其存在仅是为了满足BaseVectorStorage的接口兼容性——这与面向云端集合的向量库（如需要显式加载 collection 的实现）形成对照。

8.4close与__del__：连接回收

close()安全关闭底层 psycopg 连接（带hasattr与异常防护）；析构函数__del__调用close()，确保对象被销毁时连接自动回收。配合with上下文或显式storage.close()使用更佳。

8.5client属性

client是只读属性，直接返回底层psycopg连接对象，便于在需要原生 SQL 操作时透传访问：

raw_conn = storage.client # psycopg.Connection

九、与检索器的集成实践

PgVectorStorage遵循BaseVectorStorage抽象，因此可以直接注入 CAMEL 的VectorRetriever完成"Embedding + 向量检索"的 RAG 闭环。参考 examples/agents/repo_agent.py 中 Qdrant 的接法，替换为 pgvector 只需：

from camel.retrievers import VectorRetriever from camel.embeddings import OpenAIEmbedding from camel.storages import PgVectorStorage storage = PgVectorStorage( vector_dim=1536, conn_info={"host": "localhost", "dbname": "camel_db", "user": "postgres"}, table_name="rag_chunks", ) vr = VectorRetriever( embedding_model=OpenAIEmbedding(), storage=storage, ) # vr.process(content, top_k=3) # 完成内容切分、向量化并检索

这样的组合让多智能体应用在保持关系型数据库统一管理的同时获得向量检索能力，尤其适合已重度使用 PostgreSQL 的团队。

十、测试用例验证

仓库在 test/storages/vector_storages/test_pgvector.py 中通过 mock 连接对象对PgVectorStorage的每个方法进行了单测覆盖，可作行为契约参考：

• test_pgvector_init：验证构造参数透传与register_vector、connect各被调用一次；
• test_pgvector_add：验证executemany批量插入被调用且事务提交；
• test_pgvector_query：给定返回行("1", [0.1,...], {"a": 1}, 0.01)，断言结果 id、vector、payload 正确，且余弦距离 0.01 被换算为相似度0.99；
• test_pgvector_score_to_similarity：参数化验证三种度量的换算（余弦1-0=1.0、余弦1-0.25=0.75、欧氏1/(1+1)=0.5、点积-(-0.8)=0.8）；
• test_pgvector_delete/test_pgvector_status/test_pgvector_clear：分别验证删除执行、COUNT统计与清空提交；
• test_pgvector_empty_add_delete：验证空列表add/delete不产生任何 SQL 执行。

结语

PgVectorStorage将 pgvector 的能力封装进 CAMEL 统一的BaseVectorStorage抽象中：自动建表建索引、UPSERT 批量写入、三种距离度量的一键切换、距离分数到相似度的标准化换算，以及完备的连接生命周期管理。对于希望在 PostgreSQL 之上构建多智能体 RAG 检索、语义缓存等能力的开发者，它是与现有数据库体系无缝衔接的轻量选择。进一步深入可阅读 camel/storages/vectordb_storages/pgvector.py 源码，以及对比 camel/storages/vectordb_storages 目录下其他向量存储实现，理解 CAMEL 如何在不同向量数据库之间保持一致的接入体验。

【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel