2.1 · Python API 参考（Python API Reference）

记忆管道与知识图谱构建 · 聚焦本章的模块关系、源码依据与实现要点。

项目Cognee 章节2.1 状态全文译文模块测试、发布与运维、接口与服务契约、入库与解析、系统架构

项目要点页2.5 参考项目项目章节目录Cognee DeepWiki 原始章节Python API Reference 上一章2 下一章2.2

源码线索

cognee/__init__.py
cognee/api/client.py
cognee/api/v1/add/add.py
cognee/api/v1/cognify/cognify.py
cognee/api/v1/search/search.py
cognee/api/v1/visualize/__init__.py
cognee/api/v1/visualize/visualize.py
cognee/modules/data/models/Data.py
cognee/modules/pipelines/operations/pipeline.py
cognee/modules/pipelines/operations/run_tasks.py

模块标签

测试、发布与运维
接口与服务契约
入库与解析
系统架构
存储与持久化

章节正文

Python API 参考

原始 DeepWiki 页面https://deepwiki.com/topoteretes/cognee/2.1-python-api-reference

Python API 参考

目的与范围

本文档提供了 Cognee Python API 的完整技术参考，该 API 是开发者构建知识引擎应用的主要编程接口。API 暴露了高层函数，支持数据入库、处理、查询和持久化内存管理。

Cognee 的 API 结构分为核心的"V1"管线函数（add、cognify、search）和简化的"V2"面向内存的 API（remember、recall、forget、improve），后者专为 AI 代理设计。

相关文档：

如需使用 CLI 而非 Python API，请参阅 CLI 接口
如需 REST API 端点，请参阅 REST API 服务器
如需数据管线内部实现，请参阅 核心数据管线
如需搜索实现细节，请参阅 搜索与检索系统

API 架构总览

Python API 提供了基于 Cognee 管线执行引擎的异步优先接口。所有函数都设计为使用 await 调用，并在关系型数据库、向量数据库和图数据库之间编排复杂任务。

自然语言到代码实体映射

下图将高层系统操作映射到具体的 Python 函数和实现它们的内部任务逻辑。

Cognee · 自然语言到代码实体映射 · 图 1

来源： cognee/__init__.py:21-61、cognee/api/v1/add/add.py:92-114、cognee/api/v1/cognify/cognify.py:43-59

核心管线函数

cognee.add()

将原始数据入库到 Cognee 的暂存区。这是 add → cognify → search 工作流的第一步。

函数签名： cognee/api/v1/add/add.py:92-114

async def add(
    data: Union[BinaryIO, list[BinaryIO], str, list[str], DataItem, list[DataItem], Any],
    dataset_name: str = "main_dataset",
    user: User = None,
    node_set: Optional[List[str]] = None,
    incremental_loading: bool = True,
    run_in_background: bool = False,
    **kwargs,
) -> PipelineRunInfo

关键实现细节：

数据解析：解析文件路径、URL 和 S3 路径。它处理 BinaryIO 流和 DataItem 对象 cognee/api/v1/add/add.py:92-101。
去重：使用哈希（内容哈希 + 所有者 ID）标识内容，避免同一用户产生重复数据点 cognee/tasks/ingestion/ingest_data.py:116-117。
元数据管理：使用 Data 模型在关系型数据库中记录文件元数据、扩展名和 MIME 类型 cognee/tasks/ingestion/ingest_data.py:153-165。
数据集授权：使用 resolve_authorized_user_dataset 确保用户对目标数据集拥有"写入"权限 cognee/api/v1/add/add.py:9-11。

来源： cognee/api/v1/add/add.py:92-175、cognee/tasks/ingestion/ingest_data.py:26-170

cognee.cognify()

通过分析内容并提取实体和关系，将已入库数据转换为结构化知识图谱。

函数签名： cognee/api/v1/cognify/cognify.py:43-59

async def cognify(
    datasets: Union[str, list[str], list[UUID]] = None,
    user: User = None,
    graph_model: BaseModel = KnowledgeGraph,
    chunker = TextChunker,
    run_in_background: bool = False,
    incremental_loading: bool = True,
    temporal_cognify: bool = False,
    **kwargs,
)

数据流图： 下图展示了数据从原始输入经过管线到达持久化存储的流动过程。

Cognee · cognee.cognify() · 图 2

来源： cognee/api/v1/cognify/cognify.py:82-89、cognee/api/v1/cognify/cognify.py:22-33、cognee/modules/pipelines/operations/run_tasks.py:56-160

cognee.检索()

使用多种检索策略查询已处理的知识图谱。

函数签名： cognee/api/v1/search/search.py:28-52

async def search(
    query_text: str,
    query_type: SearchType = SearchType.GRAPH_COMPLETION,
    user: Optional[User] = None,
    datasets: Optional[Union[list[str], str]] = None,
    top_k: int = 10,
    **kwargs,
) -> List[SearchResult]

关键搜索类型：

GRAPH_COMPLETION：使用完整图谱上下文的自然语言问答。
RAG_COMPLETION：使用文档片段的传统检索增强生成（RAG）。
CHUNKS：针对特定文本片段的语义搜索。
CODE：针对代码结构和逻辑的专用搜索。

来源： cognee/api/v1/search/search.py:88-126、cognee/modules/search/methods/search.py:37-57

内存 API（V2）

V2 API 提供了封装复杂管线任务的高层内存操作。

函数	导入位置	描述
`remember()`	`cognee/__init__.py:48`	一步完成数据入库并处理到知识图谱中。
`recall()`	`cognee/__init__.py:48`	使用语义搜索和图遍历检索信息。
`forget()`	`cognee/__init__.py:48`	从内存中删除特定数据点或整个数据集。
`improve()`	`cognee/__init__.py:48`	根据新见解或规则优化现有图谱结构。

来源： cognee/__init__.py:45-50

可观测性与追踪

Cognee 通过 OpenTelemetry 和内部遥测提供了广泛的可观测性。

追踪函数

enable_tracing() / disable_tracing()：切换 OpenTelemetry 收集 cognee/__init__.py:52-54。
get_last_trace() / get_all_traces()：检索捕获的追踪数据以进行调试 cognee/__init__.py:55-56。

遥测实现

匿名 ID：基于项目根目录的 UUID，存储在 .anon_id 中，用于历史分析 cognee/shared/utils.py:49-73。
持久化 ID：存储在 ~/.cognee/.persistent_id 中的机器级 ID，在数据删除后仍然保留 cognee/shared/utils.py:76-101。
数据消毒：在传输前自动将 URL 等敏感字段哈希为 UUID5 哈希值 cognee/shared/utils.py:107-124。

来源： cognee/shared/utils.py:49-105、cognee/shared/utils.py:176-194

管线执行与并发

API 使用集中式任务运行器（run_tasks）管理执行流程和并发。

并发控制：使用 asyncio.Semaphore 限制同时处理的数据项数量（由 data_per_batch 控制）cognee/modules/pipelines/operations/run_tasks.py:97-100。
增量加载：如果启用，系统会解析数据目录，仅处理新增或变更的文件 cognee/modules/pipelines/operations/run_tasks.py:92-93。
分布式模式：如果 COGNEE_DISTRIBUTED 为 True，run_tasks 会通过装饰器将执行重定向到 run_tasks_distributed cognee/modules/pipelines/operations/run_tasks.py:36-55。
持久化钩子：在管线完成后，自动触发 push_to_s3，用于支持远程同步的数据库引擎 cognee/modules/pipelines/operations/run_tasks.py:162-168。

来源： cognee/modules/pipelines/operations/run_tasks.py:36-186

会话与可视化

会话管理

session API 处理代理交互的生命周期，跟踪模型使用情况和会话记录 cognee/__init__.py:36、cognee/__init__.py:64。

可视化

visualize_graph()：生成当前知识图谱的交互式可视化 cognee/__init__.py:31。
start_visualization_server()：启动专用 Web 服务器以托管图谱 UI cognee/__init__.py:31。

来源： cognee/__init__.py:31-36、cognee/api/v1/visualize/visualize.py