Cognee: 26K+ Star Open-Source AI Memory Platform for Agents

# Install Cognee
pip install cognee

# Or install from source for latest features
git clone https://github.com/topoteretes/cognee.git
cd cognee
pip install -e .

import cognee
from cognee.infrastructure.databases.graph import Neo4jGraphEngine

# Initialize Cognee with Neo4j
cognee.configure(
<<<<<<< HEAD
    graph_engine=Neo4jGraphEngine(
        url="bolt://localhost:7687",
        username="neo4j",
        password="your_password"
    )
=======
graph_engine=Neo4jGraphEngine(
url="bolt://localhost:7687",
tên người dùng="neo4j",
mật khẩu="your_password"
)
>>>>>>> 0f428019e6f21508f05fc402fc21585e618ed533
)

# Add knowledge to memory
await cognee.add([
    "Alice works at TechCorp as a senior engineer.",
    "TechCorp develops AI-powered code analysis tools.",
    "Alice joined TechCorp in January 2024.",
])

# Query the knowledge graph
results = await cognee.query("Who works at TechCorp?")
print(results)
# Output: [{'entity': 'Alice', 'role': 'senior engineer', 'company': 'TechCorp'}]

from langchain_community.chat_models import ChatAnthropic
from langchain.prompts import ChatPromptTemplate
import cognee

# Initialize the chatbot with memory
prompt_template = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful assistant with persistent memory.
    Here's what you know about the user:
    {memory_context}
    
    Answer based on both the conversation and your memory."""),
    ("human", "{input}"),
])

chain = prompt_template | ChatAnthropic(model="claude-sonnet-4-20250514")

# Function to get memory context
async def get_memory_context(user_id):
<<<<<<< HEAD
    memories = await cognee.search(
        query=f"user:{user_id}",
        limit=10
    )
    return "\n".join([m["text"] for m in memories])

# Chat function with memory
async def chat_with_memory(user_id, message):
    memory = await get_memory_context(user_id)
    response = chain.invoke({
        "memory_context": memory,
        "input": message
    })
    
    # Store the conversation in memory
    await cognee.add([
        f"User {user_id} asked: {message}",
        f"Assistant responded: {response.content}"
    ])
    
    return response.content
=======
ký ức = đang chờ cognee.search(
query=f"người dùng:{user_id}",
giới hạn=10
)
return "\n".join([m["text"] for m trong ký ức])

# Chức năng trò chuyện với bộ nhớ
async def chat_with_memory(user_id, tin nhắn):
bộ nhớ = đang chờ get_memory_context(user_id)
phản hồi = chain.invoke({
"memory_context": bộ nhớ,
"đầu vào": tin nhắn
})

# Lưu cuộc trò chuyện vào bộ nhớ
đang chờ cognee.add([
f"Người dùng {user_id} đã hỏi: {message}",
f"Trợ lý đã trả lời: {response.content}"
])

trả lại phản hồi.content
>>>>>>> 0f428019e6f21508f05fc402fc21585e618ed533

import cognee
from cognee.infrastructure.ingestion import DocumentIngestionPipeline

<<<<<<< HEAD
# Create ingestion pipeline
pipeline = DocumentIngestionPipeline(
    sources=[
        # PDF documents
        {"type": "pdf", "path": "./documents/"},
        # Database queries
        {"type": "sql", "query": "SELECT * FROM products"},
        # API endpoints
        {"type": "api", "url": "https://api.example.com/data"},
        # User conversations
        {"type": "conversation", "channel": "slack"},
    ],
    extraction={
        "entities": True,
        "relationships": True,
        "sentiment": True,
        "topics": True,
    },
    storage={
        "graph": "neo4j",
        "vector": "pgvector",
        "document": "s3",
    }
=======
# Tạo đường dẫn nhập
đường ống = DocumentIngestionPipeline(
nguồn=[
# tài liệu PDF
{"type": "pdf", "path": "./documents/"},
# Truy vấn cơ sở dữ liệu
{"type": "sql", "query": "CHỌN * TỪ sản phẩm"},
# điểm cuối API
{"type": "api", "url": "https://api.example.com/data"},
# Cuộc trò chuyện của người dùng
{"loại": "cuộc trò chuyện", "kênh": "chậm"},
],
trích xuất={
"thực thể": Đúng,
"mối quan hệ": Đúng,
“tình cảm”: Đúng,
"chủ đề": Đúng,
},
lưu trữ={
"biểu đồ": "neo4j",
"vectơ": "pgvector",
"tài liệu": "s3",
}
>>>>>>> 0f428019e6f21508f05fc402fc21585e618ed533
)

# Run the pipeline
await pipeline.run()

# Query across all sources
results = await cognee.query(
    "Show me all information about product launches in 2024",
    sources=["pdf", "sql", "api", "conversation"]
)

import cognee

# Get the full knowledge graph
graph = await cognee.get_graph()

# Export for visualization
graph.export(format="graphml", path="./knowledge_graph.graphml")

# Get subgraph for a specific entity
alice_graph = await cognee.get_subgraph(
    entity="Alice",
    depth=2,
    max_nodes=50
)
alice_graph.export(format="dot", path="./alice_network.dot")

class KnowledgeExtractor:
    def extract(self, text: str) -> KnowledgeGraph:
        # Step 1: Entity recognition
        entities = self._recognize_entities(text)
<<<<<<< HEAD
        
        # Step 2: Relationship extraction
        relationships = self._extract_relationships(entities, text)
        
        # Step 3: Confidence scoring
        for entity in entities:
            entity.confidence = self._score_confidence(entity, text)
        
        for rel in relationships:
            rel.confidence = self._score_relationship_confidence(rel)
        
        # Step 4: Merge with existing graph
        return self._merge_with_graph(entities, relationships)
=======

# Bước 2: Trích xuất mối quan hệ
mối quan hệ = self._extract_relationships(thực thể, văn bản)

# Bước 3: Chấm điểm tin cậy
cho thực thể trong thực thể:
thực thể.sự tự tin = self._score_trust(thực thể, văn bản)

cho mối quan hệ trong các mối quan hệ:
rel.sự tự tin = self._score_relationship_trust(rel)

# Bước 4: Hợp nhất với biểu đồ hiện có
trả về self._merge_with_graph(thực thể, mối quan hệ)
>>>>>>> 0f428019e6f21508f05fc402fc21585e618ed533

class TemporalMemoryManager:
    def __init__(self, ttl_days=365):
        self.ttl = ttl_days
    
    def manage(self, memories):
        # Mark memories for expiration
        for memory in memories:
            age = datetime.now() - memory.created_at
            if age.days > self.ttl:
                memory.status = "expired"
            elif age.days > self.ttl * 0.8:
                memory.status = "aging"
        
        # Consolidate related memories
        consolidated = self._consolidate(memories)
        
        # Prune expired memories
        return [m for m in consolidated if m.status != "expired"]

from cognee.memory import MemoryConsolidator

consolidator = MemoryConsolidator(
    similarity_threshold=0.85,
    max_memories_per_topic=50,
    consolidation_strategy="semantic_merge"
)

# Consolidate memories older than 30 days
await consolidator.consolidate(
    older_than_days=30,
    output_dir="./consolidated_memory"
)

from cognee.memory import MemoryDecay

decay = MemoryDecay(
    half_life_days=90,
    minimum_confidence=0.1,
    decay_function="exponential"
)

# Apply decay to all memories
await decay.apply(user_id="alice")
# Memories older than 90 days lose 50% influence
# Memories older than 180 days lose 75% influence

from cognee.knowledge import KnowledgeShare

share = KnowledgeShare(
    sharing_policy="anonymous_aggregate",
    sensitive_data_filter=True,
    consent_required=True
)

# Share non-sensitive knowledge patterns
await share.share(
    source_agents=["agent-1", "agent-2"],
    target_agents=["agent-3", "agent-4"],
    knowledge_types=["best_practices", "common_patterns"]
)

from cognee.verify import MemoryVerifier

verifier = MemoryVerifier(
    verification_model="claude-sonnet-4-20250514",
    confidence_threshold=0.9
)

# Verify recent memories
recent = await verifier.verify_recent(
    since_hours=24,
    max_memories=100
)

for memory in recent:
    if memory.confidence < 0.7:
        print(f"Low confidence: {memory.text}")
        print(f"Suggested action: {memory.recommended_action}")

from langchain.memory import ConversationBufferMemory
from cognee.langchain import CogneeMemoryAdapter

# Wrap Cognee with LangChain memory
cognee_memory = CogneeMemoryAdapter(
    user_id="user-123",
    max_context_items=10,
    similarity_threshold=0.75
)

memory = ConversationBufferMemory(
    memory_key="chat_history",
    chat_memory=cognee_memory
)

from crewai import Agent, Task, Crew
from cognee.crewai import CogneeMemoryPlugin

# Add memory to CrewAI agents
memory_plugin = CogneeMemoryPlugin(user_id="crew-1")

<<<<<<< HEAD
agents = [
    Agent(
        role="Researcher",
        goal="Find and analyze information",
        memory=memory_plugin,
    ),
    Agent(
        role="Writer",
        goal="Create content based on research",
        memory=memory_plugin,
    ),
=======
đại lý = [
Đại lý(
role="Nhà nghiên cứu",
target="Tìm và phân tích thông tin",
bộ nhớ=bộ nhớ_plugin,
),
Đại lý(
vai trò="Nhà văn",
target="Tạo nội dung dựa trên nghiên cứu",
bộ nhớ=bộ nhớ_plugin,
),
>>>>>>> 0f428019e6f21508f05fc402fc21585e618ed533
]

from fastapi import FastAPI
from cognee.fastapi import CogneeMiddleware

app = FastAPI()
app.add_middleware(CogneeMiddleware, user_id_header="X-User-ID")

@app.post("/chat")
async def chat(request: ChatRequest):
    # Memory is automatically managed per user
    response = await process_message(request.message)
    return {"response": response}