用 Cognee 构建端到端知识图谱，实现当前效果最好的AI Agent记忆层

显示全部楼层

用 RAG 构建 AI 应用时，总感觉差了点什么。明明相关的信息都在向量数据库里，但 AI 就是找不到正确的关联。问它"SpaceX 的创始人还创办了哪些公司"，它能找到关于 SpaceX 的文档，却理解不了"创始人"和"其他公司"之间的关系链条。这就是传统 RAG 的软肋——它只懂相似性，不懂关系。Cognee 就是来解决这个问题的。

它到底是什么？

简单说，Cognee 是个开源的端到端知识图谱构建框架。但它不仅仅是知识图谱——它将向量搜索的语义理解能力和知识图谱的关系推理能力完美融合，创造了一个真正能"理解"数据的 AI 记忆层。更重要的是，整个过程只需要 5 行代码。

最近看到他们和 Redis、Kuzu 等数据库的集成案例，效果确实让人眼前一亮：不仅保留了向量搜索 70%+ 的准确率基线，还通过图结构将准确率提升到了 90%+ 。这不是简单的性能优化，而是质的飞跃。

核心概念

Cognee 的设计理念受人类认知科学启发，模仿我们大脑构建"心智地图"的方式：

ECL 流水线

ECL（Extract, Cognify, Load）是 Cognee 的核心处理流程：

Extract（提取）：从各种数据源（API、数据库、文档）摄取原始数据
Cognify（认知化）：这是 Cognee 的独特之处，将数据转化为结构化知识
Load（加载）：将处理后的数据同时存储到向量和图数据库

DataPoints

DataPoints 是 Cognee 知识图谱的基本构建块。每个 DataPoint 不仅定义了实体（节点），还定义了它们之间的关系（边）：

fromcogneeimportDataPoint

classPerson(DataPoint):
  __tablename__ ="person"
  name: str
  age: int
  works_at: Optional["Company"] =None
 
classCompany(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 行代码

importcognee
importasyncio

asyncdefmain():
 # 添加数据
 awaitcognee.add("Elon Musk 创立了 SpaceX。SpaceX 是一家航天公司，致力于火星殖民。")
 
 # 构建知识图谱
 awaitcognee.cognify()
 
 # 查询
  results =awaitcognee.search("Elon Musk 的公司在做什么？")
 
 forresultinresults:
    print(result)

asyncio.run(main())

2. 使用 DataPoints 构建结构化知识

fromcogneeimportDataPoint, add_data_points
fromtypingimportList, Optional
importasyncio

# 定义数据模型
classPerson(DataPoint):
  __tablename__ ="person"
  name: str
  role: str
  founded: Optional[List["Company"]] = []
  works_at: Optional["Company"] =None

classCompany(DataPoint):
  __tablename__ ="company"
  name: str
  industry: str
  founded_year: int
  founder: Optional[Person] =None
  employees: List[Person] = []

classProduct(DataPoint):
  __tablename__ ="product"
  name: str
  company: Company
  description: str
  launch_year: int

asyncdefbuild_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]
 awaitadd_data_points(datapoints)
 
 # 构建图谱
 awaitcognee.cognify()
 
 # 复杂查询
  results =awaitcognee.search(
   "Elon Musk 创立的航天公司有什么产品？",
    search_type="graph_traversal"
  )
 
 returnresults

# 运行
asyncio.run(build_tech_knowledge_graph())

与 LangChain 集成

Cognee 可以作为 LangChain 的记忆层，增强 Agent 的推理能力：

fromlangchain.memoryimportConversationBufferMemory
fromlangchain.chainsimportConversationChain
fromlangchain_openaiimportChatOpenAI
importcognee
importasyncio

classCogneeMemory(ConversationBufferMemory):
 """基于 Cognee 的增强记忆"""
 
 def__init__(self, user_id: str):
    super().__init__()
    self.user_id = user_id
   
 asyncdefsave_context(self, inputs: dict, outputs: dict):
   """保存对话到 Cognee"""
   # 保存到传统记忆
    super().save_context(inputs, outputs)
   
   # 提取并存储到知识图谱
    conversation =f"User:{inputs['input']}\nAI:{outputs['response']}"
   awaitcognee.add(conversation, user_id=self.user_id)
   awaitcognee.cognify()
 
 asyncdefload_memory_variables(self, inputs: dict):
   """从 Cognee 加载相关记忆"""
   # 获取传统记忆
    memory = super().load_memory_variables(inputs)
   
   # 从知识图谱检索
    query = inputs.get("input","")
    graph_memories =awaitcognee.search(
      query,
      user_id=self.user_id,
      search_type="hybrid"# 同时使用向量和图检索
    )
   
   # 合并记忆
   ifgraph_memories:
      context ="\n".join([m.get("content","")formingraph_memories])
      memory["graph_context"] = context
     
   returnmemory

# 使用示例
asyncdefchat_with_memory():
  llm = ChatOpenAI(model="gpt-4")
  memory = CogneeMemory(user_id="alice")
 
  chain = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
  )
 
 # 第一轮对话
  response1 = chain.predict(input="我正在学习知识图谱技术")
 awaitmemory.save_context(
    {"input":"我正在学习知识图谱技术"},
    {"response": response1}
  )
 
 # 第二轮对话（会自动检索相关记忆）
  response2 = chain.predict(input="有什么好的学习资源推荐吗？")
 
  print(response2)

asyncio.run(chat_with_memory())

与 dify 集成

Cognee 也可以作为 Dify 的知识库后端：

# cognee_dify_adapter.py
fromfastapiimportFastAPI, HTTPException
frompydanticimportBaseModel
importcognee
importasyncio

app = FastAPI()

classQueryRequest(BaseModel):
  query: str
  dataset_id: str
  top_k: int =5

classAddDocumentRequest(BaseModel):
  content: str
  dataset_id: str
  metadata: dict = {}

@app.post("/add_document")
asyncdefadd_document(request: AddDocumentRequest):
 """添加文档到 Cognee"""
 try:
   awaitcognee.add(
      request.content,
      dataset_id=request.dataset_id,
      metadata=request.metadata
    )
   awaitcognee.cognify()
   return{"status":"success"}
 exceptExceptionase:
   raiseHTTPException(status_code=500, detail=str(e))

@app.post("/query")
asyncdefquery_knowledge(request: QueryRequest):
 """查询知识图谱"""
 try:
    results =awaitcognee.search(
      request.query,
      dataset_id=request.dataset_id,
      limit=request.top_k,
      search_type="hybrid"
    )
   return{"results": results}
 exceptExceptionase:
   raiseHTTPException(status_code=500, detail=str(e))

# 在 Dify 中配置自定义工具指向这个 API

一些坑和建议

使用 Cognee 一段时间后，总结几个关键点：

选对图数据库：如果数据量大，推荐 Neo4j；需要嵌入式部署，用 KuzuDB；追求性能，试试 FalkorDB。
DataPoint 设计要慎重：一开始就要想清楚实体和关系，后期修改成本很高。建议先在纸上画出领域模型。
增量更新 vs 全量重建：小规模更新用增量，大规模变更直接重建。增量更新可能导致图谱碎片化。
LLM 成本控制：Cognify 过程会大量调用 LLM，建议：

开发时用便宜的模型
生产环境按重要性分级使用不同模型
实现结果缓存机制

混合检索策略：不要过度依赖图检索，向量检索在某些场景下更高效。根据查询类型动态选择策略。