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

# Install Cognee
pip install cognee

# 또는 최신 기능을 위해 소스에서 설치
자식 클론 https://github.com/topoteretes/cognee.git
CD 코니
pip 설치 -e .

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

# Neo4j로 Cognee 초기화
코니.구성(
graph_engine=Neo4jGraphEngine(
url="bolt://localhost:7687",
사용자 이름="neo4j",
비밀번호="귀하의_비밀번호"
)
)

# 기억에 지식을 더하다
cognee.add([
"Alice는 TechCorp에서 수석 엔지니어로 일하고 있습니다.",
"TechCorp는 AI 기반 코드 분석 도구를 개발합니다.",
"Alice는 2024년 1월 TechCorp에 합류했습니다.",
])

# 지식 그래프 쿼리
results = wait cognee.query("TechCorp에서 일하는 사람은 누구입니까?")
인쇄(결과)
# 출력: [{'entity': 'Alice', 'role': 'senior Engineer', 'company': 'TechCorp'}]

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

# 메모리로 챗봇 초기화
프롬프트_템플릿 = ChatPromptTemplate.from_messages([
("시스템", """당신은 영구 기억력을 가진 도움이 되는 조수입니다.
사용자에 대해 알고 있는 정보는 다음과 같습니다.
{memory_context}

대화 내용과 기억력을 바탕으로 답변해 주세요."""),
("인간", "{입력}"),
])

체인 = 프롬프트_템플릿 | ChatAnthropic(모델="claude-sonnet-4-20250514")

# 메모리 컨텍스트를 얻는 함수
비동기 def get_memory_context(user_id):
추억 = cognee.search를 기다립니다(
쿼리=f"사용자:{user_id}",
한도=10
)
return "\n".join([m["text"] for m in 메모리])

# 메모리와 채팅 기능
비동기 def chat_with_memory(user_id, 메시지):
메모리 = get_memory_context(user_id)를 기다립니다.
응답 = chain.invoke({
"memory_context": 메모리,
"입력": 메시지
})

# 대화 내용을 메모리에 저장
cognee.add([
f"사용자 {user_id}가 질문했습니다: {message}",
f"어시스턴트가 응답했습니다: {response.content}"
])

응답.내용 반환

import cognee
from cognee.infrastructure.ingestion import DocumentIngestionPipeline

# 수집 파이프라인 생성
파이프라인 = DocumentIngestionPipeline(
출처=[
# PDF 문서
{"유형": "pdf", "경로": "./documents/"},
# 데이터베이스 쿼리
{"type": "sql", "query": "SELECT * FROM 제품"},
# API 엔드포인트
{"유형": "api", "url": "https://api.example.com/data"},
# 사용자 대화
{"유형": "대화", "채널": "느슨함"},
],
추출={
"엔티티": 사실,
"관계": 사실,
"감정": 사실,
"주제": 사실,
},
저장공간={
"그래프": "neo4j",
"벡터": "pg벡터",
"문서": "s3",
}
)

# 파이프라인 실행
파이프라인을 기다립니다.run()

# 모든 소스에 대한 쿼리
결과 = cognee.query를 기다립니다(
"2024년 제품 출시 정보를 모두 보여주세요",
소스=["pdf", "sql", "api", "대화"]
)

import cognee

# 전체 지식 그래프를 가져옵니다
그래프 = cognee.get_graph()를 기다립니다.

# 시각화를 위해 내보내기
graph.export(format="graphml", path="./knowledge_graph.graphml")

# 특정 엔터티에 대한 하위 그래프를 가져옵니다.
alice_graph = cognee.get_subgraph를 기다립니다(
엔터티="앨리스",
깊이=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)

# 2단계: 관계 추출
관계 = self._extract_relationships(엔티티, 텍스트)

# 3단계: 신뢰도 점수
엔터티의 엔터티:
엔터티.confidence = self._score_confidence(엔티티, 텍스트)

관계의 상대에 대해:
rel.confidence = self._score_relationship_confidence(rel)

# 4단계: 기존 그래프와 병합
self._merge_with_graph(엔티티, 관계)를 반환합니다.

class TemporalMemoryManager:
    def __init__(self, ttl_days=365):
        self.ttl = ttl_days

def 관리(자기, 기억):
# 만료할 추억을 표시합니다.
추억 속의 기억을 위해:
나이 = datetime.now() - memory.created_at
age.days > self.ttl인 경우:
memory.status = "만료됨"
elif age.days > self.ttl * 0.8:
memory.status = "노화"

# 관련된 추억을 통합하세요
통합 = self._consolidate(메모리)

# 만료된 기억 정리하기
[m.status != "만료됨"인 경우 통합된 m의 경우 m]을 반환합니다.

from cognee.memory import MemoryConsolidator

통합자 = MemoryConsolidator(
유사성_임계값=0.85,
max_memories_per_topic=50,
consolidation_strategy="semantic_merge"
)

# 30일이 지난 추억을 통합하세요
consolidator.consolidate(를 기다립니다.
이전_보다_일=30,
출력_디렉터리="./consolidated_memory"
)

from cognee.memory import MemoryDecay

부패 = 메모리 부패(
half_life_days=90,
최소_신뢰도=0.1,
decay_function="지수"
)

# 모든 메모리에 부패 적용
부패를 기다리십시오.적용(user_id="alice")
# 90일이 지난 추억은 영향력이 50% 감소합니다.
# 180일이 지난 추억은 영향력이 75% 감소합니다.

from cognee.knowledge import KnowledgeShare

공유 = 지식공유(
Sharing_policy="anonymous_aggregate",
sensitive_data_filter=참,
동의_필수=참
)

# 민감하지 않은 지식 패턴 공유
공유를 기다려주세요.공유(
source_agents=["에이전트-1", "에이전트-2"],
target_agents=["에이전트-3", "에이전트-4"],
Knowledge_types=["best_practices", "common_patterns"]
)

from cognee.verify import MemoryVerifier

검증자 = 메모리 검증자(
verify_model="claude-sonnet-4-20250514",
Confidence_threshold=0.9
)

# 최근 추억 확인하기
최근 = verifier.verify_recent(를 기다립니다.
since_hours=24,
max_memories=100
)

최근 기억을 위해:
memory.confidence < 0.7인 경우:
print(f"낮은 신뢰도: {memory.text}")
print(f"권장 조치: {memory.recommended_action}")

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

# Cognee를 LangChain 메모리로 래핑
cognee_memory = CogneeMemoryAdapter(
user_id="사용자-123",
max_context_items=10,
유사성_임계값=0.75
)

메모리 = ConversationBufferMemory(
memory_key="chat_history",
chat_memory=cognee_memory
)

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

# CrewAI 에이전트에 메모리 추가
memory_plugin = CogneeMemoryPlugin(user_id="crew-1")

에이전트 = [
대리인(
역할="연구원",
goal="정보 찾기 및 분석",
메모리=memory_plugin,
),
대리인(
역할="작가",
goal="연구를 바탕으로 콘텐츠 제작",
메모리=memory_plugin,
),
]

from fastapi import FastAPI
from cognee.fastapi import CogneeMiddleware

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

@app.post("/chat")
비동기 def 채팅(요청: ChatRequest):
# 메모리는 사용자별로 자동으로 관리됩니다.
응답 = process_message(request.message)를 기다립니다.
{"응답": 응답}을 반환합니다.