【亲测免费】 推荐开源项目：Cognee——为AI应用提供可靠的数据层解决方案

项目介绍

Cognee 是一款专为开发者设计的开源项目，旨在为AI应用提供可靠、生产就绪的数据层。通过实现可扩展、模块化的ECL（提取、认知、加载）管道，Cognee能够帮助开发者轻松地互联和检索过去的对话、文档、音频转录等内容，同时减少幻觉现象、开发工作量和成本。

项目技术分析

Cognee 的核心架构基于ECL管道，通过模块化的任务组合，形成一个灵活且强大的数据处理框架。其主要技术亮点包括：

• 模块化设计：Cognee允许开发者自定义任务并将其组合成管道，满足不同的业务逻辑需求。
• 本地化设置：默认使用LanceDB和NetworkX进行本地数据处理，同时支持OpenAI等语言模型。
• 向量存储支持：支持Qdrant和Weaviate等向量存储服务，提升数据检索效率。
• 语言模型支持：兼容Anyscale和Ollama等多种语言模型提供商。
• 图存储支持：除了LanceDB，还支持Neo4j进行图存储，增强数据关联性。
• 用户管理：支持创建独立的用户图并管理权限，确保数据安全。

项目及技术应用场景

Cognee 适用于多种AI应用场景，包括但不限于：

• 智能对话系统：通过检索历史对话，提升对话系统的上下文理解和响应准确性。
• 文档管理与分析：对大量文档进行分类、检索和关联分析，提高信息提取效率。
• 音频转录处理：将音频转录为文本，并进行语义分析和信息检索。
• 知识图谱构建：利用图存储和向量检索技术，构建高效的知识图谱。

项目特点

• 易于安装和使用：支持通过pip和poetry进行安装，提供详细的文档和示例代码。
• 高度可定制：开发者可以根据需求自定义任务和管道，灵活应对各种业务场景。
• 强大的社区支持：拥有活跃的Discord社区，提供及时的技术支持和交流平台。
• 丰富的集成选项：支持多种向量存储、语言模型和图存储服务，满足不同技术栈需求。

安装示例

使用pip安装

pip install cognee

使用poetry安装

poetry add cognee

基本使用示例

设置API密钥

import os

os.environ["LLM_API_KEY"] = "YOUR_OPENAI_API_KEY"

或

import cognee
cognee.config.llm_api_key = "YOUR_OPENAI_API_KEY"

简单示例

import cognee

text = """自然语言处理（NLP）是计算机科学和信息检索的跨学科子领域"""

await cognee.add([text], "example_dataset")  # 添加新信息
await cognee.cognify()  # 使用LLM和cognee创建语义图
await search_results = cognee.search("SIMILARITY", {'query': '告诉我关于NLP的信息'})  # 查询cognee知识
print(search_results)

创建自定义记忆存储

Cognee框架由多个任务组成，这些任务可以组合成管道。每个任务都可以是独立的部分业务逻辑，与其他任务连接形成管道，并将数据持久化到记忆存储中，便于检索过去的对话、文档或其他存储数据。

示例：文档分类

1. 准备数据

text = """自然语言处理（NLP）是计算机科学和信息检索的跨学科子领域"""

await cognee.add([text], "example_dataset")  # 添加新信息

2. 创建任务

async def chunk_naive_llm_classifier(
    data_chunks: list[DocumentChunk],
    classification_model: Type[BaseModel]
):
    # 异步提取分类
    chunk_classifications = await asyncio.gather(
        *(extract_categories(chunk.text, classification_model) for chunk in data_chunks)
    )
    # 收集分类数据点
    classification_data_points = {
        uuid5(NAMESPACE_OID, cls.label.type)
        for cls in chunk_classifications
    } | {
        uuid5(NAMESPACE_OID, subclass.value)
        for cls in chunk_classifications
        for subclass in cls.label.subclass
    }
    vector_engine = get_vector_engine()
    collection_name = "classification"
    # 定义负载模式
    class Keyword(BaseModel):
        uuid: str
        text: str
        chunk_id: str
        document_id: str
    # 确保集合存在并检索现有数据点
    if not await vector_engine.has_collection(collection_name):
        await vector_engine.create_collection(collection_name, payload_schema=Keyword)
        existing_points_map = {}
    else:
        existing_points_map = {}
    return data_chunks

3. 组合任务成管道

Task(
    chunk_naive_llm_classifier,
    classification_model = cognee_config.classification_model,
)

pipeline = run_tasks(tasks, documents)

更多资源

• 文档：Cognee文档
• 社区：Discord社区
• 演示：演示笔记本

立即开始使用Cognee，为您的AI应用构建强大的数据层吧！🚀