13.2 · 数据库 Adapter 接口（Database Adapter Interfaces）

记忆管道与知识图谱构建 · 本章是 Cognee DeepWiki 中文译文的独立章节页，保留原始链接、源码锚点、模块标签和章节层级。

项目Cognee 章节13.2 状态全文译文模块系统架构、存储与持久化、记忆与上下文、检索、召回与索引

项目要点页2.5 参考项目项目章节目录Cognee DeepWiki 原始章节Database Adapter Interfaces 上一章13.1 下一章13.3

源码线索

cognee/infrastructure/databases/dataset_database_handler/supported_dataset_database_handlers.py
cognee/infrastructure/databases/graph/graph_db_interface.py
cognee/infrastructure/databases/graph/kuzu/adapter.py
cognee/infrastructure/databases/graph/neo4j_driver/Neo4jAuraDevDatasetDatabaseHandler.py
cognee/infrastructure/databases/graph/neo4j_driver/adapter.py
cognee/infrastructure/databases/graph/postgres/PostgresGraphDatasetDatabaseHandler.py
cognee/infrastructure/databases/relational/config.py
cognee/infrastructure/databases/relational/create_relational_engine.py
cognee/infrastructure/databases/relational/get_relational_engine.py
cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py

模块标签

系统架构
存储与持久化
记忆与上下文
检索、召回与索引
界面与交互

中文译文

数据库 Adapter 接口（中文译文）

原始 DeepWiki 页面：https://deepwiki.com/topoteretes/cognee/13.2-database-adapter-interfaces

翻译时间：2026-05-27T08:45:25.803Z

翻译模型：deepseek-chat

原文字符数：14357

项目：Cognee (cognee)

---

数据库适配器接口

架构总览

Cognee 采用适配器模式，为多种数据库类型提供统一接口。每个数据库类别都定义了一个抽象基类或协议（Protocol），具体适配器必须实现该接口。

适配器模式结构

标题：数据库适配器层级结构

graph TB
    subgraph "应用层"
        APP["cognee.api.v1.search.search"]
    end

    subgraph "抽象接口"
        GDI["GraphDBInterface"]
        VDI["VectorDBInterface"]
        RDI["SQLAlchemyAdapter"]
    end

    subgraph "图数据库适配器"
        LADY["LadybugAdapter (Kuzu)"]
        NEO4J["Neo4jAdapter"]
    end

    subgraph "向量数据库适配器"
        LANCE["LanceDBAdapter"]
        PGVEC["PGVectorAdapter"]
        CHROMA["ChromaDBAdapter"]
    end

    subgraph "关系型数据库后端"
        SQLITE["SQLite (aiosqlite)"]
        POSTGRES["PostgreSQL (asyncpg)"]
    end

    APP --> GDI
    APP --> VDI
    APP --> RDI

    GDI --> LADY
    GDI --> NEO4J

    VDI --> LANCE
    VDI --> PGVEC
    VDI --> CHROMA

    RDI --> SQLITE
    RDI --> POSTGRES

来源：cognee/infrastructure/databases/graph/graph_db_interface.py:17-40、cognee/infrastructure/databases/vector/vector_db_interface.py:9-13、cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:27-31

图数据库接口

GraphDBInterface 抽象基类定义了所有图数据库操作的契约。该接口位于 cognee/infrastructure/databases/graph/graph_db_interface.py:17-40。

核心接口定义

标题：GraphDBInterface 与具体适配器

classDiagram
    class GraphDBInterface {
        <<abstract>>
        +is_empty() bool
        +query(query: str, params: dict) List~Any~
        +add_node(node: Union~DataPoint, str~, properties: dict) None
        +add_nodes(nodes: Union~List~Node~, List~DataPoint~~) None
        +delete_node(node_id: str) None
        +delete_nodes(node_ids: List~str~) None
        +get_node(node_id: str) Optional~NodeData~
        +get_nodes(node_ids: List~str~) List~NodeData~
        +add_edge(source_id, target_id, relationship_name, properties) None
        +add_edges(edges: List~EdgeData~) None
        +has_edge(source_id, target_id, relationship_name) bool
        +has_edges(edges: List~EdgeData~) List~EdgeData~
        +get_edges(node_id: str) List~EdgeData~
        +get_neighbors(node_id: str) List~NodeData~
        +get_connections(node_id: UUID) List
        +delete_graph() None
        +get_graph_data() Tuple
        +get_graph_metrics(include_optional: bool) Dict
    }

    class LadybugAdapter {
        +db_path: str
        +db: Database
        +connection: Connection
    }

    class Neo4jAdapter {
        +driver: AsyncGraphDatabase
        +graph_database_name: str
        +BASE_LABEL: str
    }

    GraphDBInterface <|-- LadybugAdapter
    GraphDBInterface <|-- Neo4jAdapter

来源：cognee/infrastructure/databases/graph/graph_db_interface.py:17-40、cognee/infrastructure/databases/graph/kuzu/adapter.py:10-17、cognee/infrastructure/databases/graph/neo4j_driver/adapter.py:52-67

节点操作

所有图数据库适配器必须实现以下节点管理方法：

方法	参数	返回值	描述
`is_empty()`	无	`bool`	检查图中是否包含任何节点 `cognee/infrastructure/databases/graph/graph_db_interface.py:43-46`
`add_node(node, properties)`	`node: Union[DataPoint, str]`、`properties: Optional[dict]`	`None`	向图中添加单个节点 `cognee/infrastructure/databases/graph/graph_db_interface.py:62-75`
`add_nodes(nodes)`	`nodes: Union[List[Node], List[DataPoint]]`	`None`	批量添加多个节点 `cognee/infrastructure/databases/graph/graph_db_interface.py:78-87`
`delete_node(node_id)`	`node_id: str`	`None`	根据 ID 删除单个节点 `cognee/infrastructure/databases/graph/graph_db_interface.py:90-99`
`delete_nodes(node_ids)`	`node_ids: List[str]`	`None`	批量删除多个节点 `cognee/infrastructure/databases/graph/graph_db_interface.py:102-111`
`get_node(node_id)`	`node_id: str`	`Optional[NodeData]`	根据 ID 检索单个节点 `cognee/infrastructure/databases/graph/graph_db_interface.py:135-144`
`get_nodes(node_ids)`	`node_ids: List[str]`	`List[NodeData]`	根据 ID 检索多个节点 `cognee/infrastructure/databases/graph/graph_db_interface.py:147-156`

来源：cognee/infrastructure/databases/graph/graph_db_interface.py:42-156

向量数据库接口

VectorDBInterface 是一个协议（Protocol），定义了向量存储和语义搜索的操作。

核心向量接口方法

方法	参数	返回值	描述
`embed_data(data)`	`data: list[str]`	`list[list[float]]`	将文本转换为嵌入向量 `cognee/infrastructure/databases/vector/vector_db_interface.py:188-193`
`has_collection(name)`	`name: str`	`bool`	检查集合是否存在 `cognee/infrastructure/databases/vector/vector_db_interface.py:16-30`
`create_collection(name, payload_schema)`	`name: str`、`payload_schema: Optional[Any]`	`None`	创建新集合 `cognee/infrastructure/databases/vector/vector_db_interface.py:33-48`
`create_data_points(name, points)`	`name: str`、`points: List[DataPoint]`	`None`	向集合中添加向量 `cognee/infrastructure/databases/vector/vector_db_interface.py:53-64`
`search(name, text, vector, limit, ...)`	`name: str`、`text: Optional[str]`、`vector: Optional[list]`、`limit: Optional[int]`	`List[ScoredResult]`	语义搜索 `cognee/infrastructure/databases/vector/vector_db_interface.py:83-115`

来源：cognee/infrastructure/databases/vector/vector_db_interface.py:9-193

数据点模式转换

向量适配器将 DataPoint 对象转换为数据库特定的格式：

LanceDB 模式： LanceDB 使用类级别的记忆化缓存来存储载荷模式，以防止动态类创建导致的内存增长 cognee/infrastructure/databases/vector/lancedb/LanceDBAdapter.py:85-105。

PGVector 模式：

class PGVectorDataPoint(Base):
    __tablename__ = collection_name
    id: Mapped[data_point_types["id"]] = mapped_column(primary_key=True)
    payload = Column(JSON)
    vector = Column(self.Vector(vector_size))

cognee/infrastructure/databases/vector/pgvector/PGVectorAdapter.py:163-185

关系型数据库接口

SQLAlchemyAdapter 类使用 async_sessionmaker 为关系型数据库提供异步操作。

适配器初始化

标题：SQLAlchemyAdapter 初始化流程

graph LR
    INIT["SQLAlchemyAdapter.__init__"] --> CHECK{SQLite？}
    CHECK -->|是| PARSE["解析 db_path"]
    CHECK -->|否| POSTGRES["PostgreSQL 配置"]
    PARSE --> S3{S3 存储？}
    S3 -->|是| TEMP["创建临时文件"]
    S3 -->|否| LOCAL["本地路径"]
    TEMP --> ENGINE["create_async_engine"]
    LOCAL --> ENGINE
    POSTGRES --> ENGINE["create_async_engine (asyncpg)"]
    ENGINE --> SESSMAKER["async_sessionmaker"]

来源：cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:33-104、cognee/infrastructure/databases/relational/create_relational_engine.py:49-79

会话管理

方法	类型	描述
`get_async_session()`	`@asynccontextmanager`	生成 `AsyncSession` 用于异步操作 `cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:123-132`
`get_session()`	上下文管理器	生成同步会话（包装 sessionmaker） `cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:134-143`

来源：cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:123-143

数据操作

方法	参数	返回值	描述
`insert_data(table, data, schema)`	`table: str`、`data: list`、`schema: str`	`int`	向表中插入行 `cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:188-232`
`delete_entity_by_id(table, id, schema)`	`table: str`、`id: UUID`、`schema: str`	`None`	根据 ID 删除实体 `cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:441-469`
`create_table(schema, name, config)`	`schema: str`、`name: str`、`config: list`	`None`	使用模式创建表 `cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:161-181`

来源：cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:161-469

多租户数据库处理器

Cognee 通过 DatasetDatabaseHandlerInterface 支持多租户数据库隔离。

PGVector 租户管理

PGVectorDatasetDatabaseHandler 为每个数据集创建独立的 Postgres 数据库：

创建：create_pg_database 使用到 postgres 数据库的维护连接来执行 CREATE DATABASE cognee/infrastructure/databases/vector/pgvector/create_db_and_tables.py:32-55。
删除：delete_pg_database 在删除数据库之前终止活动的后端连接 cognee/infrastructure/databases/vector/pgvector/create_db_and_tables.py:91-101。

来源：cognee/infrastructure/databases/vector/pgvector/PGVectorDatasetDatabaseHandler.py:15-58、cognee/infrastructure/databases/vector/pgvector/create_db_and_tables.py:20-106

错误处理

适配器会抛出标准异常，以确保跨提供商的错误处理一致性：

异常	使用场景
`EntityNotFoundError`	当关系型实体缺失时抛出 `cognee/infrastructure/databases/exceptions.py:18`
`CollectionNotFoundError`	当向量集合缺失时抛出 `cognee/infrastructure/databases/vector/exceptions.py:20`
`MissingQueryParameterError`	当搜索参数不足时抛出 `cognee/infrastructure/databases/vector/lancedb/LanceDBAdapter.py:15`
`DatabaseCredentialsError`	当 Neo4j 认证信息不完整时抛出 `cognee/infrastructure/databases/graph/neo4j_driver/adapter.py:20`

来源：cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py:18、cognee/infrastructure/databases/vector/lancedb/LanceDBAdapter.py:15-20、cognee/infrastructure/databases/graph/neo4j_driver/adapter.py:77-90