# Install Cognee
pip install cognee

# 或者从源代码安装以获得最新功能
git 克隆 https://github.com/topotetes/cognee.git
cd科尼
pip install -e 。

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

# 使用 Neo4j 初始化 Cognee
科涅.配置（
graph_engine=Neo4jGraphEngine(
url =“螺栓：//本地主机：7687”，
用户名=“neo4j”，
密码=“你的密码”
)
)

# 将知识添加到记忆中
等待 cognee.add([
“Alice 在 TechCorp 担任高级工程师。”,
“TechCorp 开发人工智能驱动的代码分析工具。”,
“Alice 于 2024 年 1 月加入 TechCorp。”,
])

# 查询知识图谱
results = wait cognee.query("谁在 TechCorp 工作？")
打印（结果）
# 输出: [{'entity': 'Alice', 'role': '高级工程师', 'company': 'TechCorp'}]

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

<<<<<<< HEAD
# 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}"),
=======
# 使用内存初始化聊天机器人
提示模板 = ChatPromptTemplate.from_messages([
(“系统”,“”“你是一个有持久记忆的得力助手。
以下是您对用户的了解：
{内存上下文}

根据对话内容和你的记忆来回答。"""),
（“人类”，“{输入}”），
>>>>>>> 0f428019e6f21508f05fc402fc21585e618ed533
])

链=提示模板| ChatAnthropic（模型=“claude-sonnet-4-20250514”）

# 获取内存上下文的函数
异步 def get_memory_context(user_id):
记忆=等待cognee.search（
查询=f“用户：{user_id}”，
限制=10
)
return "\n".join([m["text"] 表示内存中的 m])

<<<<<<< HEAD
# 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
=======
# 带记忆的聊天功能
异步 def chat_with_memory(user_id, message):
内存 = 等待 get_memory_context(user_id)
响应 = chain.invoke({
“内存上下文”：内存，
“输入”：消息
})

# 将对话存储在内存中
等待 cognee.add([
f"用户 {user_id} 询问：{message}",
f“助理回复：{response.content}”
])

返回响应内容
>>>>>>> 0f428019e6f21508f05fc402fc21585e618ed533

import cognee
from cognee.infrastructure.ingestion import DocumentIngestionPipeline

# 创建摄取管道
管道 = DocumentIngestionPipeline(
来源=[
# PDF 文档
{"类型": "pdf", "路径": "./documents/"},
# 数据库查询
{“type”：“sql”，“query”：“从产品中选择*”}，
# API 端点
{"type": "api", "url": "https://api.example.com/data"},
# 用户对话
{"type": "对话", "channel": "slack"},
],
提取={
“实体”：确实，
“关系”：确实如此，
“情绪”：真实，
“主题”：确实，
},
存储={
“图”：“neo4j”，
“向量”：“pg向量”，
“文档”：“s3”，
}
)

# 运行管道
等待 pipeline.run()

# 查询所有来源
结果=等待cognee.query(
“显示有关 2024 年产品发布的所有信息”，
来源=[“pdf”、“sql”、“api”、“对话”]
)

import cognee

# 获取完整的知识图谱
图 = 等待 cognee.get_graph()

# 导出以进行可视化
graph.export（格式=“graphml”，路径=“./knowledge_graph.graphml”）

# 获取特定实体的子图
alice_graph = 等待 cognee.get_subgraph(
实体=“爱丽丝”，
深度=2，
最大节点数=50
)
alice_graph.export（格式=“点”，路径=“./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)
=======

# 步骤2：关系提取
关系= self._extract_relationships（实体，文本）

# 第 3 步：置信度评分
对于实体中的实体：
实体.置信度 = self._score_confidence(实体, 文本)

对于关系中的 rel：
rel.confidence = self._score_relationship_confidence(rel)

# 第 4 步：与现有图合并
返回 self._merge_with_graph(实体，关系)
>>>>>>> 0f428019e6f21508f05fc402fc21585e618ed533

class TemporalMemoryManager:
    def __init__(self, ttl_days=365):
        self.ttl = ttl_days
<<<<<<< HEAD
    
    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 MemoryDecay

衰减 = 记忆衰减（
half_life_days=90,
最小置信度=0.1，
Decay_function =“指数”
)

# 对所有记忆应用衰减
等待衰减。应用（user_id =“alice”）
# 超过 90 天的记忆失去 50% 的影响力
# 超过 180 天的记忆失去 75% 的影响力

from cognee.knowledge import KnowledgeShare

分享 = 知识分享(
共享策略=“匿名聚合”，
sensitive_data_filter=真，
同意_必需=真
)

# 分享非敏感知识模式
等待分享.分享（
source_agents=["agent-1", "agent-2"],
target_agents=["agent-3", "agent-4"],
knowledge_types=["best_practices", "common_patterns"]
)

from cognee.verify import MemoryVerifier

验证器 = 内存验证器(
verify_model =“克劳德-sonnet-4-20250514”，
置信度阈值=0.9
)

# 验证最近的记忆
最近 = 等待验证程序.verify_recent(
由于_小时=24，
最大内存=100
)

最近的记忆：
如果内存置信度 < 0.7：
print(f"低置信度：{memory.text}")
print(f"建议的操作：{memory.recommended_action}")

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

# 用LangChain内存包裹Cognee
cognee_memory = CogneeMemoryAdapter（
user_id =“用户123”，
最大上下文项=10,
相似度阈值=0.75
)

内存 = ConversationBufferMemory(
memory_key="聊天记录",
聊天内存=cognee_内存
)

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

# 为 CrewAI 代理添加内存
memory_plugin = CogneeMemoryPlugin(user_id="crew-1")

代理=[
代理人（
角色=“研究员”，
目标=“查找和分析信息”，
内存=内存_插件，
),
代理人（
角色=“作家”，
目标=“根据研究创建内容”，
内存=内存_插件，
),
]

from fastapi import FastAPI
from cognee.fastapi import CogneeMiddleware

应用程序 = FastAPI()
app.add_middleware(CogneeMiddleware, user_id_header="X-User-ID")

@app.post(“/聊天”)
异步 def 聊天（请求：ChatRequest）：
# 每个用户自动管理内存
响应 = 等待 process_message(request.message)
返回{“响应”：响应}