Cognee测评：为AI应用赋予记忆力的智能引擎

# 使用pip安装
pip install cognee

# 或使用poetry安装
poetry add cognee

import os
os.environ["LLM_API_KEY"] = "YOUR_OPENAI_API_KEY"  # 设置API密钥

import cognee
import asyncio

async def main():
    # 添加文本到cognee
    await cognee.add("自然语言处理(NLP)是计算机科学和信息检索的交叉学科领域。")
    
    # 生成知识图谱
    await cognee.cognify()
    
    # 查询知识图谱
    results = await cognee.search("告诉我关于NLP的知识")
    
    # 显示结果
    for result in results:
        print(result)

if __name__ == '__main__':
    asyncio.run(main())

# 克隆cognee仓库
git clone https://github.com/topoteretes/cognee.git

# 进入cognee-mcp子目录
cd cognee/cognee-mcp

# 安装uv（如果还没有安装）
pip install uv

# 使用uv安装cognee mcp服务器所需的所有依赖项
uv sync --dev --all-extras --reinstall

# 激活虚拟环境
source .venv/bin/activate

# 设置OpenAI API密钥（创建.env文件）
echo 'LLM_API_KEY="YOUR_OPENAI_API_KEY"' > .env

# 从Docker Hub直接拉取镜像运行（最简单）
docker run --env-file ./.env -p 8000:8000 --rm -it cognee/cognee-mcp:main

# 或本地构建镜像
docker rmi cognee/cognee-mcp:main || true
docker build --no-cache -f cognee-mcp/Dockerfile -t cognee/cognee-mcp:main .

# 运行容器
docker run --env-file ./.env -p 8000:8000 --rm -it cognee/cognee-mcp:main

# Ubuntu/Debian
sudo apt update
sudo apt install python3 python3-pip git

# CentOS/RHEL
sudo yum install python3 python3-pip git

import os
import asyncio
import cognee
from cognee import config

# 设置API密钥
os.environ["LLM_API_KEY"] = "sk-your-openai-api-key"

# 准备示例数据（实际应用中可从文件读取）
documents = [
    {
        "content": "自然语言处理(NLP)是计算机科学和人工智能的一个领域，致力于让计算机能够理解、解释和生成人类语言。",
        "metadata": {"title": "NLP介绍", "department": "AI研发", "date": "2024-01-15"}
    },
    {
        "content": "我们的API接口采用RESTful设计原则，支持JSON格式的请求和响应。所有API调用都需要通过OAuth 2.0认证。",
        "metadata": {"title": "API设计指南", "department": "后端开发", "date": "2024-02-20"}
    },
    {
        "content": "项目Alpha阶段将于本季度结束前完成，主要里程碑包括核心模块开发、测试套件实现和文档编写。",
        "metadata": {"title": "项目Alpha更新", "department": "项目管理", "date": "2024-03-10"}
    }
]

from typing import List, Dict, Any
from pydantic import BaseModel
from cognee import BaseModel, DocumentChunk
import uuid
from uuid import uuid5, NAMESPACE_OID

# 定义自定义数据结构
class Department(BaseModel):
    id: str
    name: str
    description: str

class Employee(BaseModel):
    id: str
    name: str
    department_id: str
    skills: List[str]

# 创建自定义处理任务
async def process_company_documents(
    documents: List[Dict[str, Any]], 
    dataset_name: str = "company_docs"
):
    """处理公司文档并提取关键信息"""
    
    # 添加文档到cognee
    for doc in documents:
        await cognee.add(
            doc["content"], 
            dataset_name,
            metadata=doc.get("metadata", {})
        )
    
    # 认知化处理（构建知识图谱）
    print("开始构建知识图谱...")
    await cognee.cognify()
    
    # 提取部门信息
    departments = await extract_entities("department", Department)
    
    # 提取员工技能信息
    skills = await extract_entities("skill", None)
    
    return {
        "departments": departments,
        "skills": skills,
        "dataset": dataset_name
    }

async def extract_entities(entity_type: str, model: Type[BaseModel]):
    """从内容中提取特定类型的实体"""
    search_results = await cognee.search(
        "GRAPH_COMPLETION", 
        {"query": f"列出所有{entity_type}及相关信息"}
    )
    return search_results

async def query_company_knowledge():
    """查询公司知识库"""
    
    # 1. 搜索特定技术信息
    technical_results = await cognee.search(
        "RAG_COMPLETION", 
        {"query": "我们的API认证机制是什么？"}
    )
    
    print("API认证信息:", technical_results)
    
    # 2. 搜索项目信息
    project_results = await cognee.search(
        "GRAPH_COMPLETION", 
        {"query": "项目Alpha的当前状态和里程碑"}
    )
    
    print("项目信息:", project_results)
    
    # 3. 搜索部门相关信息
    department_results = await cognee.search(
        "SIMILARITY", 
        {"query": "AI研发部门的相关文档"}
    )
    
    print("部门文档:", department_results)
    
    return {
        "technical": technical_results,
        "projects": project_results,
        "departments": department_results
    }

async def main():
    """主执行函数"""
    
    # 初始化cognee
    print("初始化Cognee知识管理系统...")
    
    # 处理公司文档
    processing_result = await process_company_documents(documents)
    print(f"处理完成，数据集: {processing_result['dataset']}")
    
    # 查询知识库
    knowledge = await query_company_knowledge()
    
    # 打印摘要报告
    print("\n=== 知识管理摘要报告 ===")
    print(f"处理文档数: {len(documents)}")
    print(f"提取部门数: {len(processing_result['departments'])}")
    print(f"提取技能数: {len(processing_result['skills'])}")
    
    return knowledge

if __name__ == "__main__":
    # 运行主程序
    result = asyncio.run(main())

async def advanced_analysis():
    """高级知识图谱分析"""
    
    # 获取知识图谱子图
    subgraph = await cognee.search(
        "GRAPH_COMPLETION",
        {
            "query": "显示AI研发与后端开发部门之间的关系",
            "graph_type": "department_relations"
        }
    )
    
    # 时序分析（基于文档日期）
    timeline_analysis = await cognee.search(
        "INSIGHTS",
        {
            "query": "分析项目进展随时间的变化",
            "time_field": "date",
            "time_range": ["2024-01-01", "2024-03-31"]
        }
    )
    
    # 跨文档关联分析
    cross_document_analysis = await cognee.search(
        "RAG_COMPLETION",
        {
            "query": "找出NLP技术与API开发之间的潜在关联",
            "depth": 2,
            "max_relationships": 5
        }
    )
    
    return {
        "department_relations": subgraph,
        "timeline_analysis": timeline_analysis,
        "cross_document_insights": cross_document_analysis
    }

# 执行高级分析
advanced_results = asyncio.run(advanced_analysis())