13.1 · Python API 完整参考（Python API Complete Reference）

Python API 完整参考

Python API 完整参考

相关源文件

本章引用的主要源码文件：

• cognee/__init__.py
• cognee/api/client.py
• cognee/api/v1/add/add.py
• cognee/api/v1/cognify/cognify.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
• cognee/shared/utils.py
• cognee/tasks/ingestion/ingest_data.py
• cognee/tasks/ingestion/save_data_item_to_storage.py
• cognee/tests/test_telemetry.py
• cognee/tests/unit/processing/utils/utils_test.py
• poetry.lock
• pyproject.toml
• uv.lock

目的与范围

本文档提供了 cognee 包导出的所有公共 Python 函数和类的完整参考。这些函数构成了将 Cognee 知识图谱引擎集成到 Python 应用程序中的主要接口。

如需了解以下信息，请参阅对应文档：

• CLI 命令，请参见 CLI 接口
• REST API 端点，请参见 REST API 服务器
• MCP 服务器工具，请参见 MCP 工具参考
• 自定义管线开发，请参见 自定义管线教程

除非另有说明，本文档中记录的所有函数均为异步函数，必须使用 await 调用。

来源：cognee/__init__.py:1-61

API 组织结构图

来源：cognee/__init__.py:21-61

核心工作流函数

这些函数实现了 Cognee 的主要工作流：数据入库、知识图谱生成和检索。

cognee.add()

签名：

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,
    vector_db_config: dict = None,
    graph_db_config: dict = None,
    dataset_id: Optional[UUID] = None,
    preferred_loaders: Optional[List[Union[str, dict[str, dict[str, Any]]]]] = None,
    incremental_loading: bool = True,
    data_per_batch: Optional[int] = 20,
    importance_weight: Optional[float] = 0.5,
    run_in_background: bool = False,
    **kwargs
) -> PipelineRunInfo

描述：将原始数据入库到 Cognee 中，以供后续知识图谱处理。这是所有将被系统处理的数据的入口点。

参数：

返回：PipelineRunInfo - 包含运行 ID、数据集 ID、状态和元数据的管线执行信息。

实现细节： add 函数会启动 ingest_data 过程 cognee/api/v1/add/add.py:16-17。它会解析文件路径，通过 save_data_item_to_storage cognee/tasks/ingestion/save_data_item_to_storage.py:27-97 处理各种输入类型，并使用 ingestion.identify cognee/tasks/ingestion/ingest_data.py:117 执行基于内容哈希的去重。它还支持使用 DataItem 对象进行自定义标签 cognee/tasks/ingestion/ingest_data.py:91-96。

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

cognee.cognify()

签名：

async def cognify(
    datasets: Union[str, list[str], list[UUID]] = None,
    user: User = None,
    graph_model: BaseModel = KnowledgeGraph,
    chunker = TextChunker,
    chunk_size: int = None,
    chunks_per_batch: int = None,
    config: Config = None,
    vector_db_config: dict = None,
    graph_db_config: dict = None,
    run_in_background: bool = False,
    incremental_loading: bool = True,
    custom_prompt: Optional[str] = None,
    temporal_cognify: bool = False,
    data_per_batch: int = 20,
    **kwargs
) -> Union[dict, list[PipelineRunInfo]]

描述：通过分类、片段切分、实体提取、关系检测和摘要生成，将入库数据转换为结构化的知识图谱。

处理管线： cognify 使用 run_pipeline cognee/api/v1/cognify/cognify.py:12 编排多个任务。

默认管线任务：

1. classify_documents cognee/api/v1/cognify/cognify.py:23
2. extract_chunks_from_documents cognee/api/v1/cognify/cognify.py:24
3. extract_graph_and_summarize cognee/api/v1/cognify/cognify.py:26
4. add_data_points cognee/api/v1/cognify/cognify.py:27

来源：cognee/api/v1/cognify/cognify.py:43-135、cognee/api/v1/cognify/cognify.py:269-331

cognee.检索()

签名：

async def search(
    query: str,
    query_type: SearchType = SearchType.GRAPH_COMPLETION,
    **kwargs
) -> list[dict]

描述：使用 SearchType 中定义的各种检索策略查询知识图谱。

来源：cognee/__init__.py:30

cognee.remember()

签名：

async def remember(
    data: Union[BinaryIO, list[BinaryIO], str, list[str], DataItem, list[DataItem], Any],
    dataset_name: str = "main_dataset",
    user: User = None,
    dataset_id: Optional[UUID] = None,
    node_set: Optional[List[str]] = None,
    run_in_background: bool = False,
    custom_prompt: Optional[str] = None,
    chunks_per_batch: Optional[int] = 10,
    **kwargs
) -> RememberResult

描述：一个高级便利函数，将 add() 和 cognify() 合并为单个原子操作。它是 V2 面向记忆 API 的一部分。

来源：cognee/__init__.py:48

数据模型参考

数据类（关系型）

Data 模型表示关系数据库中入库的内容。

来源：cognee/modules/data/models/Data.py:12-40

管线系统内部实现

管线系统使用 run_tasks 通过信号量进行并发控制，以编排跨数据项的执行。

实现细节：run_tasks 使用 asyncio.Semaphore(data_per_batch) 来限制数据项的并发处理数量 cognee/modules/pipelines/operations/run_tasks.py:97-100。如果设置了 COGNEE_DISTRIBUTED 环境变量，它还支持通过 run_tasks_distributed 进行分布式执行 cognee/modules/pipelines/operations/run_tasks.py:40-45。

来源：cognee/modules/pipelines/operations/pipeline.py:33-61、cognee/modules/pipelines/operations/run_tasks.py:56-156

可视化与用户界面

cognee.visualize_graph()

签名：

async def visualize_graph(
    destination_file_path: str = None
) -> str

描述：生成当前知识图谱的交互式网络可视化。它通过图引擎获取数据，并将其处理为 HTML/D3.js 格式 cognee/api/v1/visualize/visualize.py:11-30。

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

可观测性

Cognee 提供了内置的 OpenTelemetry 追踪功能，用于监控管线性能和大语言模型（LLM）使用情况。

来源：cognee/__init__.py:51-58、cognee/modules/observability/trace_context.py:16-93

数据库与迁移

Cognee 通过 run_startup_migrations 函数确保数据库模式在启动时是最新的，该函数处理关系型（Alembic）和向量存储的迁移。

来源：cognee/run_migrations.py:82-90、cognee/api/client.py:82-90

遥测系统

Cognee 包含一个匿名遥测系统，用于跟踪使用情况并改进库。它会生成三层身份标识：

1. 匿名 ID：基于项目根目录的 ID，存储在 .anon_id 文件中 cognee/shared/utils.py:49-73。
2. 持久 ID：机器级别的 ID，存储在 ~/.cognee/.persistent_id 文件中 cognee/shared/utils.py:76-104。
3. API 密钥追踪 ID：通过 PBKDF2 哈希从大语言模型（LLM）API 密钥派生的假名 ID cognee/shared/utils.py:139-168。

来源：cognee/shared/utils.py:49-190