在构建AI应用时，基于RAG（检索增强生成）的方式常常感到不足。尽管相关信息存储在向量数据库中，AI仍难以建立正确的关联。例如，当询问“SpaceX 的创始人还创办了哪些公司”时，AI可能检索到SpaceX的文档，却无法理解“创始人”与“其他公司”之间的关系链条。这正是传统RAG的局限所在——它侧重于相似性，而非深层关系理解。Cognee 正是为解决这一问题而生。
它到底是什么？
简而言之，Cognee 是一个开源的端到端知识图谱构建框架。但其独特之处在于，它将向量搜索的语义理解能力与知识图谱的关系推理能力完美融合，从而创建了一个真正能够“理解”数据的AI记忆层。更值得一提的是，整个构建过程仅需5行代码即可完成。
近期，Cognee与Redis、Kuzu等数据库的集成案例展示了显著效果：它不仅维持了向量搜索70%以上的准确率基线，更通过图结构将准确率提升至90%以上。这不仅仅是简单的性能优化，更是AI数据理解能力上的质的飞跃。
核心概念
Cognee 的设计理念深受人类认知科学的启发，旨在模仿大脑构建“心智地图”的方式：
ECL 流水线
ECL（Extract, Cognify, Load）是 Cognee 的核心处理流程：
- Extract（提取）：从各种数据源（API、数据库、文档）摄取原始数据
- Cognify（认知化）：这是 Cognee 的独特之处，将数据转化为结构化知识
- Load（加载）：将处理后的数据同时存储到向量和图数据库
DataPoints
DataPoints 是 Cognee 知识图谱的基本构建块。每个 DataPoint 不仅定义了实体（节点），还定义了它们之间的关系（边）：
from cognee import DataPoint
class Person(DataPoint):
__tablename__ = "person"
name: str
age: int
works_at: Optional["Company"] = None
class Company(DataPoint):
__tablename__ = "company"
name: str
employees: List[Person] = []
双存储架构
Cognee 采用“多模态存储”策略：
- 向量数据库：存储语义嵌入，支持模糊搜索
- 图数据库：存储实体关系，支持精确推理
- 关系数据库：存储元数据，支持结构化查询
理论框架
知识图谱构建流程
Cognee 的知识图谱构建遵循以下核心流程：
1. 信息提取与结构化
原始文本 → LLM 分析 → 实体识别 → 关系抽取 → DataPoint 实例化
系统使用 LLM 对输入内容进行深度分析，自动识别：
- 实体（人物、地点、组织、概念）
- 关系（所属、关联、依赖、引用）
- 属性（特征、标签、元数据）
2. 认知化（Cognify）过程
这是Cognee的核心创新，包含以下三个关键步骤：
去重与合并：
- 识别相同实体的不同表述
- 合并重复信息
- 解决冲突数据
关系推理：
- 推断隐含关系
- 构建多跳关联
- 创建知识网络
向量化与索引：
- 生成语义嵌入
- 创建倒排索引
- 构建图结构
3. GraphRAG 检索机制
Cognee 的 GraphRAG 结合了两种检索范式：
# 向量检索：找到语义相关的内容
vector_results = vector_store.search(query_embedding, top_k=10)
# 图检索：探索实体关系
graph_results = graph_store.traverse(
start_nodes=vector_results,
max_depth=3,
relationship_types=["FOUNDED","WORKS_AT"])
# 融合结果：结合语义和结构化信息
final_results = merge_and_rank(vector_results, graph_results)
认知科学基础
Cognee 的设计深受认知心理学的“语义网络理论”影响：
- 扩散激活：从一个概念出发，激活相关联的其他概念
- 层次组织：知识按照抽象程度分层存储
- 关联强度：频繁共现的概念建立更强的连接
实践
环境搭建
1. 安装 Cognee
# 使用 pip 安装
pip install cognee
# 或使用 poetry
poetry add cognee
# 如需 PostgreSQL 支持
pip install cognee[postgres]
2. 配置 LLM 和嵌入模型
创建.env文件：
# LLM 配置（支持 OpenAI、DeepSeek、Ollama 等）
LLM_API_KEY=sk-your-api-key
LLM_PROVIDER=openai # 或 deepseek、ollama
LLM_MODEL=gpt-4o-mini
# 嵌入模型配置
EMBEDDING_PROVIDER=ollama
EMBEDDING_MODEL=mxbai-embed-large
# 向量存储配置
VECTOR_STORE_PROVIDER=qdrant
QDRANT_URL=http://localhost:6333
QDRANT_COLLECTION_NAME=cognee_vectors
# 图存储配置
GRAPH_STORE_PROVIDER=neo4j
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=password
3. 安装存储后端
Qdrant（向量存储）：
docker run -p 6333:6333 qdrant/qdrant
Neo4j（图存储）：
docker run -p 7474:7474 -p 7687:7687 \
-e NEO4J_AUTH=neo4j/password \
neo4j:latest
或者使用轻量级的 FalkorDB：
docker run -p 6379:6379 falkordb/falkordb
基本使用示例
1. 最简单的 5 行代码
import cognee
import asyncio
async def main():
# 添加数据
await cognee.add("Elon Musk 创立了 SpaceX。SpaceX 是一家航天公司，致力于火星殖民。")
# 构建知识图谱
await cognee.cognify()
# 查询
results = await cognee.search("Elon Musk 的公司在做什么？")
for result in results:
print(result)
asyncio.run(main())
2. 使用 DataPoints 构建结构化知识
from cognee import DataPoint, add_data_points
from typing import List, Optional
import asyncio
# 定义数据模型
class Person(DataPoint):
__tablename__ = "person"
name: str
role: str
founded: Optional[List["Company"]] = []
works_at: Optional["Company"] = None
class Company(DataPoint):
__tablename__ = "company"
name: str
industry: str
founded_year: int
founder: Optional[Person] = None
employees: List[Person] = []
class Product(DataPoint):
__tablename__ = "product"
name: str
company: Company
description: str
launch_year: int
async def build_tech_knowledge_graph():
# 创建实体
elon = Person(
name="Elon Musk",
role="Entrepreneur"
)
spacex = Company(
name="SpaceX",
industry="Aerospace",
founded_year=2002,
founder=elon
)
tesla = Company(
name="Tesla",
industry="Electric Vehicles",
founded_year=2003,
founder=elon
)
# 建立关系
elon.founded = [spacex, tesla]
elon.works_at = spacex
# 创建产品
falcon9 = Product(
name="Falcon 9",
company=spacex,
description="可重复使用的轨道级火箭",
launch_year=2010
)
model3 = Product(
name="Model 3",
company=tesla,
description="大众化电动轿车",
launch_year=2017
)
# 添加到知识图谱
datapoints = [elon, spacex, tesla, falcon9, model3]
await add_data_points(datapoints)
# 构建图谱
await cognee.cognify()
# 复杂查询
results = await cognee.search(
"Elon Musk 创立的航天公司有什么产品？",
search_type="graph_traversal"
)
return results
# 运行
asyncio.run(build_tech_knowledge_graph())
与 LangChain 集成
Cognee 可以作为 LangChain 的记忆层，增强 Agent 的推理能力：
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
import cognee
import asyncio
class CogneeMemory(ConversationBufferMemory):
"""基于 Cognee 的增强记忆"""
def __init__(self, user_id: str):
super().__init__()
self.user_id = user_id
async def save_context(self, inputs: dict, outputs: dict):
"""保存对话到 Cognee"""
# 保存到传统记忆
super().save_context(inputs, outputs)
# 提取并存储到知识图谱
conversation = f"User: {inputs['input']}
AI: {outputs['response']}"
await cognee.add(conversation, user_id=self.user_id)
await cognee.cognify()
async def load_memory_variables(self, inputs: dict):
"""从 Cognee 加载相关记忆"""
# 获取传统记忆
memory = super().load_memory_variables(inputs)
# 从知识图谱检索
query = inputs.get("input", "")
graph_memories = await cognee.search(
query,
user_id=self.user_id,
search_type="hybrid"# 同时使用向量和图检索
)
# 合并记忆
if graph_memories:
context = "
".join([m.get("content", "") for m in graph_memories])
memory["graph_context"] = context
return memory
# 使用示例
async def chat_with_memory():
llm = ChatOpenAI(model="gpt-4")
memory = CogneeMemory(user_id="alice")
chain = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
# 第一轮对话
response1 = await chain.apredict(input="我正在学习知识图谱技术")
await memory.save_context(
{"input":"我正在学习知识图谱技术"},
{"response": response1}
)
# 第二轮对话（会自动检索相关记忆）
response2 = await chain.apredict(input="有什么好的学习资源推荐吗？")
print(response2)
asyncio.run(chat_with_memory())
与 Dify 集成
Cognee 也可以作为 Dify 的知识库后端：
# cognee_dify_adapter.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import cognee
import asyncio
app = FastAPI()
class QueryRequest(BaseModel):
query: str
dataset_id: str
top_k: int = 5
class AddDocumentRequest(BaseModel):
content: str
dataset_id: str
metadata: dict = {}
@app.post("/add_document")
async def add_document(request: AddDocumentRequest):
"""添加文档到 Cognee"""
try:
await cognee.add(
request.content,
dataset_id=request.dataset_id,
metadata=request.metadata
)
await cognee.cognify()
return {"status": "success"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/query")
async def query_knowledge(request: QueryRequest):
"""查询知识图谱"""
try:
results = await cognee.search(
request.query,
dataset_id=request.dataset_id,
limit=request.top_k,
search_type="hybrid"
)
return {"results": results}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# 在 Dify 中配置自定义工具指向这个 API
一些坑和建议
使用 Cognee 一段时间后，总结几个关键点：
-
选对图数据库：如果数据量大，推荐 Neo4j；需要嵌入式部署，用 KuzuDB；追求性能，试试 FalkorDB。
-
DataPoint 设计要慎重：一开始就要想清楚实体和关系，后期修改成本很高。建议先在纸上画出领域模型。
-
增量更新 vs 全量重建：对于小规模更新，可采用增量方式；而大规模变更则建议直接重建，因为增量更新可能导致知识图谱碎片化。
-
LLM 成本控制：Cognify 过程涉及大量LLM调用，因此建议采取以下措施控制成本：
-
在开发阶段使用成本较低的模型
-
在生产环境中，根据重要性对不同任务分级，并选用相应模型
-
实施结果缓存机制以减少重复调用
-
混合检索策略：不应过度依赖图检索，因为在特定场景下，向量检索可能更为高效。应根据查询类型动态选择合适的检索策略。
