Cognee 深度调研报告

import cognee

# 添加文本
await cognee.add("Your document content here")

# 添加文件或目录
await cognee.add("papers/source/", dataset_name="research_papers")

# 添加多种格式
await cognee.add(["file1.pdf", "file2.txt", "http://example.com"])

# 对添加的数据进行认知化处理
await cognee.cognify()

# 指定处理特定数据集
await cognee.cognify(dataset_name="research_papers")

# 语义增强（即将推出）
await cognee.memify()

from cognee.api.v1.search import SearchType

# 默认综合搜索
results = await cognee.search(query_text="What is Agentic AI?")

# 指定搜索类型
results = await cognee.search(
    SearchType.INSIGHTS,
    query_text="How do knowledge graphs improve RAG?"
)

class DataPoint(BaseModel):
    id: UUID = Field(default_factory=uuid4)
    created_at: int = ...  # 毫秒时间戳
    updated_at: int = ...  # 毫秒时间戳
    version: int = 1
    topological_rank: Optional[int] = 0
    metadata: Optional[dict] = {"index_fields": []}
    type: str = "DataPoint"
    belongs_to_set: Optional[List["DataPoint"]] = None

# index_fields 中的字段 → 转换为嵌入向量
# 每个索引字段 → 独立的向量集合（Class_field）
# 非索引字段 → 保持为普通属性

from cognee.tasks import Task

class ExtractEntitiesTask(Task):
    async def run(self, chunks: List[Chunk]) -> List[Entity]:
        # 实体抽取逻辑
        entities = await self.llm.extract_entities(chunks)
        return entities

from cognee.pipelines import Pipeline

pipeline = Pipeline([
    ChunkTask(chunk_size=512),
    ExtractEntitiesTask(llm=claude),
    ExtractRelationsTask(llm=claude),
    EmbedTask(embedder=openai_embedding),
])

await pipeline.run(documents)

import cognee
from cognee.infrastructure.databases.graph import Neo4jConfig

cognee.config.set_graph_db_config(Neo4jConfig(
    uri="bolt://localhost:7687",
    username="neo4j",
    password="password"
))

from cognee.infrastructure.databases.graph import FalkorDBConfig

cognee.config.set_graph_db_config(FalkorDBConfig(
    host="localhost",
    port=6379
))

from cognee.infrastructure.databases.vector import QdrantConfig

cognee.config.set_vector_db_config(QdrantConfig(
    url="http://localhost:6333",
    api_key="your-api-key"
))

from cognee.infrastructure.databases.vector import WeaviateConfig

cognee.config.set_vector_db_config(WeaviateConfig(
    url="http://localhost:8080"
))

import cognee

# OpenAI
cognee.config.set_llm_api_key("sk-...")
cognee.config.set_llm_provider("openai")
cognee.config.set_llm_model("gpt-4o")

# Anthropic
cognee.config.set_llm_provider("anthropic")
cognee.config.set_llm_model("claude-3-5-sonnet-20241022")

# Azure OpenAI
cognee.config.set_llm_provider("azure")
cognee.config.set_azure_endpoint("https://your-resource.openai.azure.com")

# 为数据点分配 Node Set
datapoint.belongs_to_set = [paper_set, method_set]

# 按 Node Set 检索
results = await cognee.search(
    query_text="...",
    node_sets=["research_papers"]
)

# 加载外部本体
await cognee.load_ontology("https://example.org/ontology.owl")

# 知识接地
await cognee.ground_to_ontology(entities, ontology="schema.org")

# 设置数据权限
await cognee.add(
    data,
    permissions=["user:alice", "group:research"]
)

# 带权限检索
results = await cognee.search(
    query_text="...",
    user_id="alice"
)

from cognee.models import DataPoint
from pydantic import Field
from typing import List

class Paper(DataPoint):
    title: str = Field(..., description="论文标题")
    abstract: str = Field(..., description="摘要")
    authors: List[str] = Field(default_factory=list)
    year: int = Field(..., description="发表年份")
    venue: str = Field(None, description="发表期刊/会议")

    class Config:
        index_fields = ["title", "abstract"]

class Author(DataPoint):
    name: str
    affiliation: str = None
    email: str = None

class Method(DataPoint):
    name: str
    description: str
    category: str  # e.g., "Prompt Engineering", "Agent Framework"

    class Config:
        index_fields = ["description"]

class Citation(DataPoint):
    source_paper_id: str
    target_paper_id: str
    context: str  # 引用上下文

# 安装 Cognee
pip install cognee

# 安装可选依赖（根据选用的存储）
pip install cognee[neo4j]     # Neo4j 支持
pip install cognee[qdrant]    # Qdrant 支持
pip install cognee[weaviate]  # Weaviate 支持

# .env 文件
OPENAI_API_KEY=sk-your-openai-key
ANTHROPIC_API_KEY=your-anthropic-key

# Neo4j 配置（可选）
NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=password

import cognee
import asyncio

async def paper_memory_demo():
    # 1. 重置环境（开发测试用）
    await cognee.prune.prune_data()
    await cognee.prune.prune_system(metadata=True)

    # 2. 配置 LLM
    cognee.config.set_llm_api_key("sk-...")
    cognee.config.set_llm_provider("openai")

    # 3. 添加论文内容
    papers = [
        """
        Title: ReAct: Synergizing Reasoning and Acting in Language Models
        Abstract: While large language models have demonstrated remarkable abilities...
        """,
        """
        Title: Chain-of-Thought Prompting Elicits Reasoning
        Abstract: We explore how generating a chain of thought...
        """
    ]

    for paper in papers:
        await cognee.add(paper, dataset_name="agent_papers")

    # 4. 构建知识图谱
    await cognee.cognify(dataset_name="agent_papers")

    # 5. 执行查询
    results = await cognee.search(
        query_text="What are the main reasoning approaches in LLMs?"
    )

    for result in results:
        print(f"Score: {result.score:.3f}")
        print(f"Content: {result.content[:200]}...")
        print("---")

if __name__ == "__main__":
    asyncio.run(paper_memory_demo())

import cognee
from cognee.api.v1.search import SearchType

async def paper_graph_demo():
    # 1. 添加论文目录
    await cognee.add(
        "papers/source/llm-agents/",
        dataset_name="agent_papers"
    )

    # 2. 构建图谱
    await cognee.cognify()

    # 3. 图遍历查询：找相关论文
    results = await cognee.search(
        SearchType.GRAPH_TRAVERSAL,
        query_text="ReAct framework",
        max_hops=2  # 最多两跳关系
    )

    # 4. 自然语言转 Cypher 查询
    cypher_results = await cognee.search(
        SearchType.NATURAL_LANGUAGE,
        query_text="Find all papers that cite ReAct and use chain-of-thought"
    )

    # 5. 直接 Cypher 查询（高级）
    raw_cypher = await cognee.search(
        SearchType.CYPHER,
        query_text="""
        MATCH (p1:Paper)-[:CITES]->(p2:Paper)
        WHERE p2.title CONTAINS 'ReAct'
        RETURN p1.title, p1.abstract
        LIMIT 10
        """
    )

asyncio.run(paper_graph_demo())

from agents.core.memory import CogneeMemory
from agents.heartfelt.agent import HeartfeltAgent

class CogneeMemory:
    def __init__(self):
        cognee.config.set_llm_provider("anthropic")
        cognee.config.set_llm_model("claude-3-5-sonnet-20241022")

    async def add_paper(self, content: str, metadata: dict):
        """添加论文到记忆"""
        await cognee.add(content, dataset_name="papers")
        await cognee.cognify()

    async def search(self, query: str, search_type: str = "INSIGHTS"):
        """检索相关知识"""
        type_map = {
            "INSIGHTS": SearchType.INSIGHTS,
            "GRAPH": SearchType.GRAPH_TRAVERSAL,
            "SEMANTIC": SearchType.CHUNKS
        }
        return await cognee.search(
            type_map.get(search_type, SearchType.INSIGHTS),
            query_text=query
        )

# Agent 使用示例
async def agent_with_memory():
    memory = CogneeMemory()
    agent = HeartfeltAgent(memory=memory)

    # Agent 查询时自动使用记忆
    response = await agent.analyze(
        "比较 ReAct 和 Chain-of-Thought 在表格推理任务上的表现"
    )
    print(response)

import cognee

async def visualize_knowledge_graph():
    # 构建图谱后获取可视化
    await cognee.cognify()

    # 生成交互式图谱可视化
    from cognee.visualization import render_graph

    # 导出为 HTML（可交互拖拽、缩放）
    render_graph(
        output_path="paper_knowledge_graph.html",
        title="论文知识图谱"
    )