用 Cognee 構(gòu)建端到端知識圖譜，實現(xiàn)當前效果最好的AI Agent記憶層 原創(chuàng)
用 RAG 構(gòu)建 AI 應用時，總感覺差了點什么。明明相關(guān)的信息都在向量數(shù)據(jù)庫里，但 AI 就是找不到正確的關(guān)聯(lián)。問它"SpaceX 的創(chuàng)始人還創(chuàng)辦了哪些公司"，它能找到關(guān)于 SpaceX 的文檔，卻理解不了"創(chuàng)始人"和"其他公司"之間的關(guān)系鏈條。這就是傳統(tǒng) RAG 的軟肋——它只懂相似性，不懂關(guān)系。Cognee 就是來解決這個問題的。
它到底是什么？
簡單說，Cognee 是個開源的端到端知識圖譜構(gòu)建框架。但它不僅僅是知識圖譜——它將向量搜索的語義理解能力和知識圖譜的關(guān)系推理能力完美融合，創(chuàng)造了一個真正能"理解"數(shù)據(jù)的 AI 記憶層。更重要的是，整個過程只需要 5 行代碼。
最近看到他們和 Redis、Kuzu 等數(shù)據(jù)庫的集成案例，效果確實讓人眼前一亮：不僅保留了向量搜索 70%+ 的準確率基線，還通過圖結(jié)構(gòu)將準確率提升到了 90%+ 。這不是簡單的性能優(yōu)化，而是質(zhì)的飛躍。
核心概念
Cognee 的設計理念受人類認知科學啟發(fā)，模仿我們大腦構(gòu)建"心智地圖"的方式：
ECL 流水線
ECL（Extract, Cognify, Load）是 Cognee 的核心處理流程：
- Extract（提取）：從各種數(shù)據(jù)源（API、數(shù)據(jù)庫、文檔）攝取原始數(shù)據(jù)
- Cognify（認知化）：這是 Cognee 的獨特之處，將數(shù)據(jù)轉(zhuǎn)化為結(jié)構(gòu)化知識
- Load（加載）：將處理后的數(shù)據(jù)同時存儲到向量和圖數(shù)據(jù)庫
DataPoints
DataPoints 是 Cognee 知識圖譜的基本構(gòu)建塊。每個 DataPoint 不僅定義了實體（節(jié)點），還定義了它們之間的關(guān)系（邊）：
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] = []雙存儲架構(gòu)
Cognee 采用"多模態(tài)存儲"策略：
- 向量數(shù)據(jù)庫：存儲語義嵌入，支持模糊搜索
- 圖數(shù)據(jù)庫：存儲實體關(guān)系，支持精確推理
- 關(guān)系數(shù)據(jù)庫：存儲元數(shù)據(jù)，支持結(jié)構(gòu)化查詢
理論框架
知識圖譜構(gòu)建流程
Cognee 的知識圖譜構(gòu)建遵循以下核心流程：
1. 信息提取與結(jié)構(gòu)化
原始文本 → LLM 分析 → 實體識別 → 關(guān)系抽取 → DataPoint 實例化系統(tǒng)使用 LLM 對輸入內(nèi)容進行深度分析，自動識別：
- 實體（人物、地點、組織、概念）
- 關(guān)系（所屬、關(guān)聯(lián)、依賴、引用）
- 屬性（特征、標簽、元數(shù)據(jù)）
2. 認知化（Cognify）過程
這是 Cognee 的核心創(chuàng)新，包含三個關(guān)鍵步驟：
去重與合并：
- 識別相同實體的不同表述
- 合并重復信息
- 解決沖突數(shù)據(jù)
關(guān)系推理：
- 推斷隱含關(guān)系
- 構(gòu)建多跳關(guān)聯(lián)
- 創(chuàng)建知識網(wǎng)絡
向量化與索引：
- 生成語義嵌入
- 創(chuàng)建倒排索引
- 構(gòu)建圖結(jié)構(gòu)
3. GraphRAG 檢索機制
Cognee 的 GraphRAG 結(jié)合了兩種檢索范式：
# 向量檢索：找到語義相關(guān)的內(nèi)容
vector_results = vector_store.search(query_embedding, top_k=10)
# 圖檢索：探索實體關(guān)系
graph_results = graph_store.traverse(
start_nodes=vector_results,
max_depth=3,
relationship_types=["FOUNDED", "WORKS_AT"]
)
# 融合結(jié)果：結(jié)合語義和結(jié)構(gòu)化信息
final_results = merge_and_rank(vector_results, graph_results)認知科學基礎
Cognee 的設計深受認知心理學的"語義網(wǎng)絡理論"影響：
- 擴散激活：從一個概念出發(fā)，激活相關(guān)聯(lián)的其他概念
- 層次組織：知識按照抽象程度分層存儲
- 關(guān)聯(lián)強度：頻繁共現(xiàn)的概念建立更強的連接
實踐
環(huán)境搭建
1. 安裝 Cognee
# 使用 pip 安裝
pip install cognee
# 或使用 poetry
poetry add cognee
# 如需 PostgreSQL 支持
pip install cognee[postgres]2. 配置 LLM 和嵌入模型
創(chuàng)建 ??.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=password3. 安裝存儲后端
Qdrant（向量存儲）：
docker run -p 6333:6333 qdrant/qdrantNeo4j（圖存儲）：
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():
# 添加數(shù)據(jù)
await cognee.add("Elon Musk 創(chuàng)立了 SpaceX。SpaceX 是一家航天公司，致力于火星殖民。")
# 構(gòu)建知識圖譜
await cognee.cognify()
# 查詢
results = await cognee.search("Elon Musk 的公司在做什么？")
for result in results:
print(result)
asyncio.run(main())2. 使用 DataPoints 構(gòu)建結(jié)構(gòu)化知識
from cognee import DataPoint, add_data_points
from typing import List, Optional
import asyncio
# 定義數(shù)據(jù)模型
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():
# 創(chuàng)建實體
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
)
# 建立關(guān)系
elon.founded = [spacex, tesla]
elon.works_at = spacex
# 創(chuàng)建產(chǎn)品
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)
# 構(gòu)建圖譜
await cognee.cognify()
# 復雜查詢
results = await cognee.search(
"Elon Musk 創(chuàng)立的航天公司有什么產(chǎn)品？",
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"""
# 保存到傳統(tǒng)記憶
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 加載相關(guān)記憶"""
# 獲取傳統(tǒng)記憶
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="我正在學習知識圖譜技術(shù)")
await memory.save_context(
{"input": "我正在學習知識圖譜技術(shù)"},
{"response": response1}
)
# 第二輪對話（會自動檢索相關(guān)記憶）
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 一段時間后，總結(jié)幾個關(guān)鍵點：
- 選對圖數(shù)據(jù)庫：如果數(shù)據(jù)量大，推薦 Neo4j；需要嵌入式部署，用 KuzuDB；追求性能，試試 FalkorDB。
- DataPoint 設計要慎重：一開始就要想清楚實體和關(guān)系，后期修改成本很高。建議先在紙上畫出領(lǐng)域模型。
- 增量更新 vs 全量重建：小規(guī)模更新用增量，大規(guī)模變更直接重建。增量更新可能導致圖譜碎片化。
- LLM 成本控制：Cognify 過程會大量調(diào)用 LLM，建議：
- 開發(fā)時用便宜的模型
- 生產(chǎn)環(huán)境按重要性分級使用不同模型
- 實現(xiàn)結(jié)果緩存機制
- 混合檢索策略：不要過度依賴圖檢索，向量檢索在某些場景下更高效。根據(jù)查詢類型動態(tài)選擇策略。
本文轉(zhuǎn)載自???AI 博物院?? 作者：longyunfeigu
?著作權(quán)歸作者所有，如需轉(zhuǎn)載，請注明出處，否則將追究法律責任
贊
1
收藏 1
回復
分享
微博
QQ
微信
舉報
回復
相關(guān)推薦
