3.3 · 知识 Graph 生成 (cognee.cognify)（Knowledge Graph Generation (cognee.cognify)）

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

项目Cognee 章节3.3 状态全文译文模块图谱与关系、测试、发布与运维、入库与解析、检索、召回与索引

项目要点页2.5 参考项目项目章节目录Cognee DeepWiki 原始章节Knowledge Graph Generation (cognee.cognify) 上一章3.2 下一章3.4

源码线索

cognee/api/v1/add/add.py
cognee/api/v1/cognify/cognify.py
cognee/eval_framework/corpus_builder/task_getters/get_cascade_graph_tasks.py
cognee/eval_framework/corpus_builder/task_getters/get_default_tasks_by_indices.py
cognee/infrastructure/llm/prompts/chunk_association_system.txt
cognee/infrastructure/llm/prompts/chunk_association_user.txt
cognee/modules/chunking/TextChunker.py
cognee/modules/data/models/Data.py
cognee/modules/data/processing/document_types/AudioDocument.py
cognee/modules/data/processing/document_types/Document.py

模块标签

图谱与关系
测试、发布与运维
入库与解析
检索、召回与索引
系统架构

章节正文

知识 Graph 生成 (cognee.cognify)

原始 DeepWiki 页面https://deepwiki.com/topoteretes/cognee/3.3-knowledge-graph-generation-cognee-cognify

知识图谱生成（cognee.cognify）

目的与范围

本文档描述了 cognee.cognify() 流程及其相关的处理管线，该管线将原始入库数据转换为结构化的知识图谱。cognify 流程是 Cognee 中的核心转换层，它接收通过 cognee.add() 添加的数据（参见 cognee/api/v1/add/add.py:114-116），并将其转换为包含节点、边和向量嵌入向量的语义图谱，这些数据分别存储在向量数据库和图数据库中。

关于 cognify 之前的数据入库流程，请参见 数据入库（cognee.add）（3.1）。关于管线任务编排机制的详细信息，请参见 管线任务与执行（3.2）。关于基于时间事件的知识图谱生成，请参见 时序知识图谱（3.4）。关于定义自定义图谱模式，请参见 自定义图谱模型（3.5）。

cognify 函数

cognify() 函数是知识图谱生成的主要入口点。它编排了一个多阶段管线，将数据集处理为可查询的知识结构。

函数签名与参数

cognee/api/v1/cognify/cognify.py:43-59

参数	类型	默认值	描述
`datasets`	`Union[str, list[str], list[UUID]]`	`None`	要处理的数据集名称或 UUID。如果为 None，则处理所有数据集。
`user`	`User`	`None`	用于认证和数据访问的用户上下文。
`graph_model`	`BaseModel`	`KnowledgeGraph`	定义知识图谱结构的 Pydantic 模型。
`chunker`	可调用对象	`TextChunker`	文本片段切分策略（TextChunker、LangchainChunker）。
`chunk_size`	`int`	`None`	每个片段的最大 Token 数。如果为 None，则自动计算。
`chunks_per_batch`	`int`	`None`	任务中每批处理的片段数量。
`config`	`Config`	`None`	配置，包括本体解析器设置。
`vector_db_config`	`dict`	`None`	自定义向量数据库配置。
`graph_db_config`	`dict`	`None`	自定义图数据库配置。
`run_in_background`	`bool`	`False`	如果为 True，则异步运行并立即返回。
`incremental_loading`	`bool`	`True`	启用数据项的增量处理。
`custom_prompt`	`Optional[str]`	`None`	用于实体提取的自定义大语言模型提示词。
`temporal_cognify`	`bool`	`False`	启用时序事件提取模式。
`data_per_batch`	`int`	`20`	每批并发处理的数据项数量。

执行流程

Cognee · 执行流程 · 图 1

来源： cognee/api/v1/cognify/cognify.py:168-233

返回值

该函数返回 Union[dict, list[PipelineRunInfo]]：

阻塞模式（run_in_background=False）：字典，将 dataset_id 映射到 PipelineRunInfo，包含处理状态、实体/关系数量、持续时间和错误信息。
后台模式（run_in_background=True）：PipelineRunInfo 对象列表，包含用于跟踪进度的 pipeline_run_id。

来源： cognee/api/v1/cognify/cognify.py:124-134

处理管线架构

cognify 管线由一系列按顺序执行的 Task 对象组成。存在两种管线变体：默认管线和时序管线。

默认管线

Cognee · 默认管线 · 图 2

默认任务组装

cognee/api/v1/cognify/cognify.py:237-297

get_default_tasks() 函数创建以下任务序列：

classify_documents：识别文档类型（PDF、文本、音频等）并设置元数据 cognee/tasks/documents/classify_documents.py:99-146。
extract_chunks_from_documents：使用指定的 chunker 将内容分割为语义片段 cognee/tasks/documents/extract_chunks_from_documents.py:17-43。
extract_graph_from_data：基于大语言模型的实体和关系提取，使用 extract_content_graph cognee/tasks/graph/extract_graph_from_data.py:129-183。
summarize_text：创建分层内容摘要，以便更好地进行检索 cognee/tasks/summarization/summarize_text.py:12-25。
add_data_points：将节点/边存储到数据库，并进行向量索引 cognee/tasks/storage/add_data_points.py:30-149。
extract_dlt_fk_edges：从结构化 DLT 源中提取外键关系 cognee/tasks/ingestion/extract_dlt_fk_edges.py:10-35。

任务 2：文本片段切分

extract_chunks_from_documents 任务将文档分割为可处理的单元，同时保持语义连贯性。

片段切分策略

片段大小计算

如果未提供 chunk_size，则会使用 get_max_chunk_tokens() 自动计算：

chunk_size = min(
    embedding_model_max_tokens,
    llm_model_max_tokens // 2
)

来源： cognee/infrastructure/llm/get_max_chunk_tokens.py:10-25、cognee/api/v1/cognify/cognify.py:273-274

片段切分器实现

切分器	策略	使用场景
`TextChunker`	基于段落的分割	默认选项，对通用文本最可靠 `cognee/modules/chunking/TextChunker.py:10-20`
`chunk_by_paragraph`	迭代句子分组	段落分割的核心实现 `cognee/tasks/chunks/chunk_by_paragraph.py:7-11`

任务 3：实体与关系提取

extract_graph_from_data 任务使用大语言模型从文本片段中识别实体、关系和属性。

提取过程

cognee/tasks/graph/extract_graph_from_data.py:129-183

输入校验：确保 data_chunks 是有效的 DocumentChunk 对象 cognee/tasks/graph/extract_graph_from_data.py:142-145。
DLT 处理：跳过 DltRowDocument 片段的大语言模型提取，因为其图谱是根据模式元数据确定性构建的 cognee/tasks/graph/extract_graph_from_data.py:149-156。
大语言模型提取：对每个片段调用 extract_content_graph，生成图谱模型实例 cognee/tasks/graph/extract_graph_from_data.py:166-173。
集成：通过 integrate_chunk_graphs() 将特定片段的图谱集成到统一结构中，该函数处理本体映射 cognee/tasks/graph/extract_graph_from_data.py:56-125。

本体校验

实体会根据本体解析器（例如 RDFLibOntologyResolver）进行校验，以确保命名和分类的一致性 cognee/modules/graph/utils/expand_with_nodes_and_edges.py:115-121。

来源： cognee/tasks/graph/extract_graph_from_data.py:110-112、cognee/modules/graph/utils/expand_with_nodes_and_edges.py:11-18

任务 4：基于 DataPoint 的图谱构建

get_graph_from_model() 函数递归遍历 DataPoint 对象，构建完整的节点和边图谱结构。

图谱提取算法

Cognee · 图谱提取算法 · 图 3

实现细节

节点创建 节点通过复制 DataPoint 模型并排除关系字段来创建，以保持节点的"扁平化"以便存储 cognee/modules/graph/utils/get_graph_from_model.py:223-227。

边表示 边表示为包含 source_node_id、target_node_id 和 relationship_name 的属性 cognee/modules/graph/utils/get_graph_from_model.py:99-104。

循环预防 该算法使用唯一键 f"{data_point_id}_{relationship_key}_{target_id}" 跟踪已访问的属性，以防止在循环图中出现无限递归 cognee/modules/graph/utils/get_graph_from_model.py:130-132。

来源： cognee/modules/graph/utils/get_graph_from_model.py:178-293

任务 5：数据点存储与索引

add_data_points 任务将提取的图谱元素存储到向量数据库和图数据库中。

存储架构

Cognee · 存储架构 · 图 4

混合存储能力

如果引擎支持 HYBRID_WRITE（例如 PostgresHybridAdapter），则节点和向量可以在单次操作中存储 cognee/tasks/storage/add_data_points.py:87-88。

来源： cognee/tasks/storage/add_data_points.py:30-149、cognee/infrastructure/databases/unified/capabilities.py:5-10

管线执行与批处理

数据项批处理

run_tasks() 函数使用 asyncio.Semaphore 控制正在处理的数据项的并发数，该数量受 data_per_batch 限制 cognee/modules/pipelines/operations/run_tasks.py:97-100。

增量加载

当 incremental_loading=True 时，系统会检查 check_pipeline_run_qualification()，以跳过已处理到知识图谱中的数据项 cognee/modules/pipelines/operations/pipeline.py:79-86。

来源： cognee/modules/pipelines/operations/run_tasks.py:56-122、cognee/modules/pipelines/operations/pipeline.py:66-107