用 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 不仅定义了实体（节点），还定义了它们之间的关系（边）：

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. 扩散激活：从一个概念出发，激活相关联的其他概念
2. 层次组织：知识按照抽象程度分层存储
3. 关联强度：频繁共现的概念建立更强的连接

实践

环境搭建

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

asyncdef 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

asyncdef 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
        
    asyncdef save_context(self, inputs: dict, outputs: dict):
        """保存对话到 Cognee"""
        # 保存到传统记忆
        super().save_context(inputs, outputs)
        
        # 提取并存储到知识图谱
        conversation = f"User: {inputs['input']}\nAI: {outputs['response']}"
        await cognee.add(conversation, user_id=self.user_id)
        await cognee.cognify()
    
    asyncdef 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 = "\n".join([m.get("content", "") for m in graph_memories])
            memory["graph_context"] = context
            
        return memory

# 使用示例
asyncdef chat_with_memory():
    llm = ChatOpenAI(model="gpt-4")
    memory = CogneeMemory(user_id="alice")
    
    chain = ConversationChain(
        llm=llm,
        memory=memory,
        verbose=True
    )
    
    # 第一轮对话
    response1 = chain.predict(input="我正在学习知识图谱技术")
    await memory.save_context(
        {"input": "我正在学习知识图谱技术"}, 
        {"response": response1}
    )
    
    # 第二轮对话（会自动检索相关记忆）
    response2 = chain.predict(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")
asyncdef 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")
asyncdef 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 一段时间后，总结几个关键点：

1. 选对图数据库：如果数据量大，推荐 Neo4j；需要嵌入式部署，用 KuzuDB；追求性能，试试 FalkorDB。

2. DataPoint 设计要慎重：一开始就要想清楚实体和关系，后期修改成本很高。建议先在纸上画出领域模型。

3. 增量更新 vs 全量重建：小规模更新用增量，大规模变更直接重建。增量更新可能导致图谱碎片化。

4. LLM 成本控制：Cognify 过程会大量调用 LLM，建议：

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

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