pip install cognee

import cognee

# 1. 添加记忆
cognee.add("用户A的订单12345状态是运输中", user_id="user_001")
cognee.add("用户A的订单12345包含商品iPhone", user_id="user_001")

# 2. 检索记忆
response = cognee.search("用户A的订单状态是什么？", user_id="user_001")
print(response)  # 输出：订单12345状态是运输中，包含商品iPhone

你是一个客服助手。用户问：我的订单12345到哪了？

# 先检索用户历史记忆
memories = cognee.search("最近的订单信息", user_id="user_001")
# 注入到系统提示
prompt = f"""你是客服助手。以下是你已经知道的用户信息：
{memories}
用户问：我的订单到哪了？
"""

skills/memory_agent/
├── SKILL.md          # 元描述
├── memory.py         # 核心逻辑：记忆读写
├── injector.py       # 记忆注入器：将Cognee输出转为系统提示
└── test_agent.py     # 测试脚本

# SKILL.md
name: "Cognee Memory Agent"
description: >
  给LLM Agent注入跨Session长期记忆，基于知识图谱检索实体关系。
  Agent启动时自动加载用户历史记忆，每次对话结束时新增记忆。
trigger:
  - event: "new_session"
    action: "load_user_memory(user_id)"
  - event: "message_received"
    action: "inject_memory_into_prompt()"
  - event: "response_sent"
    action: "extract_and_store_memory(text)"
inputs:
  - user_id: str
  - message: str
outputs:
  - response: str
dependencies:
  - cognee >= 0.1.0
  - openai       # 可选，任何LLM均可

import cognee
from typing import List

class CogneeMemory:
    def __init__(self, user_id: str):
        self.user_id = user_id

    def load_memory(self, query: str = "历史对话关键信息") -> str:
        """检索用户记忆，返回结构化文本"""
        results = cognee.search(query, user_id=self.user_id)
        return results if results else "暂无历史记忆。"

    def save_memory(self, new_info: str):
        """将新信息存入知识图谱"""
        cognee.add(new_info, user_id=self.user_id)

    def extract_from_conversation(self, conversation: List[dict]):
        """从对话历史中抽取事实并存储"""
        for turn in conversation[-5:]:  # 只处理最近5轮，避免噪声
            if turn['role'] == 'assistant':
                continue
            # 让LLM提取关键事实（可选）
            facts = extract_facts(turn['content'])
            for fact in facts:
                self.save_memory(fact)

def build_memory_prompt(user_id: str) -> str:
    memory = CogneeMemory(user_id).load_memory("用户偏好和历史订单")
    return f"""【已记忆的用户信息】
{memory}

请基于以上已知信息回答用户问题。如果用户问的问题记忆中没有，请先回答“我现在不知道，请提供更多信息”。
"""

import openai
from cognee_memory import CogneeMemory

client = openai.OpenAI()

async def handle_message(user_id: str, user_message: str):
    # 加载记忆
    memory_manager = CogneeMemory(user_id)
    memory_text = memory_manager.load_memory("用户身份和最近操作")

    # 构建系统提示 + 记忆
    system_prompt = f"你是一个记忆增强客服。已知用户信息：\n{memory_text}"

    # 调用LLM
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message}
        ]
    )
    answer = response.choices[0].message.content

    # 从用户消息中提取新事实并存入
    facts_to_store = extract_key_facts(user_message)  # 需要实现或调用LLM
    for fact in facts_to_store:
        memory_manager.save_memory(fact)

    return answer

def extract_key_facts(text: str) -> list:
    # 简单示例：可用正则或LLM；这里用LLM演示
    prompt = f"从以下文本中提取判断性的、可长期记忆的事实（如用户偏好、购买物品、状态变化）。每行一个事实：\n\n{text}"
    result = openai.chat.completions.create(model="gpt-4o-mini", messages=[{"role":"user","content":prompt}])
    return result.choices[0].message.content.strip().split('\n')

# 测试
print(handle_message("user_001", "我昨天买了一件红色毛衣，查看物流"))
# 输出包含“您的订单XXX已发货”

from langchain.memory import BaseMemory

class CogneeLangChainMemory(BaseMemory):
    def __init__(self, user_id):
        self.inner = CogneeMemory(user_id)
    
    @property
    def memory_variables(self):
        return ["history"]
    
    def load_memory_variables(self, inputs):
        return {"history": self.inner.load_memory(inputs.get("query", ""))}
    
    def save_context(self, inputs, outputs):
        # 自动存储输入输出
        self.inner.extract_from_conversation([
            {"role": "user", "content": str(inputs)},
            {"role": "assistant", "content": str(outputs)}
        ])

class CogneeMemoryProvider:
    def add(self, text, metadata=None):
        cognee.add(text, user_id=metadata.get("user_id", "default"))
    def get(self, query, limit=5):
        # 使用图检索 + 排序
        return cognee.search(query, user_id="default")

import hashlib
user_id = hashlib.sha256("alice@example.com".encode()).hexdigest()[:16]