# 第一次对话
messages = [{"role": "user", "content": "我叫小明，我喜欢喝咖啡"}]

# 第二次对话：如果不把第一次的话带上，模型完全不知道小明是谁
messages = [{"role": "user", "content": "我叫什么名字？"}]

from collections import deque

class SlidingWindowMemory:
    """只保留最近 max_messages 条消息"""
    def __init__(self, max_messages: int = 10):
        self.history = deque(maxlen=max_messages)

    def add(self, role: str, content: str):
        self.history.append({"role": role, "content": content})

    def get_messages(self):
        return list(self.history)

memory = SlidingWindowMemory(max_messages=4)
memory.add("user", "我喜欢喝咖啡")
memory.add("assistant", "好的，记住了")
memory.add("user", "我的猫叫咪咪")
memory.add("assistant", "可爱！")
# 超过 4 条后，最老的自动被丢掉
memory.add("user", "刚才说的两个信息是什么？")
print(memory.get_messages())

from openai import OpenAI

client = OpenAI()

def summarize(history: list) -> str:
    """把对话历史压缩成一段摘要"""
    text = "\n".join(f"{m['role']}: {m['content']}" for m in history)
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是对话摘要助手，用 100 字以内总结对话中的关键事实和用户偏好，只输出摘要。"},
            {"role": "user", "content": text},
        ],
    )
    return resp.choices[0].message.content

class SummaryMemory:
    def __init__(self, keep_raw: int = 6):
        self.raw = []          # 最近几轮原文
        self.summary = ""      # 压缩后的摘要
        self.keep_raw = keep_raw

    def add(self, role: str, content: str):
        self.raw.append({"role": role, "content": content})
        if len(self.raw) > self.keep_raw:
            old = self.raw[: len(self.raw) - self.keep_raw]
            self.summary = summarize(old)   # 老消息变摘要
            self.raw = self.raw[-self.keep_raw:]

    def get_messages(self):
        base = [{"role": "system", "content": f"历史摘要：{self.summary}"}] if self.summary else []
        return base + self.raw

def extract_facts(dialog: str) -> list:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "从对话中抽取值得长期记住的事实，输出 JSON 数组，例如 [{\"fact\": \"用户叫小明\", \"type\": \"preference\"}]。没有就输出 []。"},
            {"role": "user", "content": dialog},
        ],
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content).get("facts", [])

import json
import os

class FileMemory:
    def __init__(self, path: str = "memory.json"):
        self.path = path
        self.data = {"facts": [], "history": []}
        if os.path.exists(path):
            with open(path, "r", encoding="utf-8") as f:
                self.data = json.load(f)

    def save(self):
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(self.data, f, ensure_ascii=False, indent=2)

    def add_fact(self, fact: str, type_: str = "general"):
        self.data["facts"].append({"fact": fact, "type": type_})
        self.save()

    def get_facts(self):
        return [item["fact"] for item in self.data["facts"]]

mem = FileMemory()
mem.add_fact("用户叫小明", "preference")
print(mem.get_facts())  # 下次启动程序，记忆还在！

import sqlite3

class SqliteMemory:
    def __init__(self, db_path: str = "agent.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS memories (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                content TEXT NOT NULL,
                kind TEXT DEFAULT "fact",
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        """)
        self.conn.commit()

    def add(self, user_id: str, content: str, kind: str = "fact"):
        self.conn.execute(
            "INSERT INTO memories (user_id, content, kind) VALUES (?, ?, ?)",
            (user_id, content, kind),
        )
        self.conn.commit()

    def search(self, user_id: str, keyword: str):
        cur = self.conn.execute(
            "SELECT content, kind FROM memories WHERE user_id=? AND content LIKE ? ORDER BY id DESC LIMIT 10",
            (user_id, f"%{keyword}%"),
        )
        return cur.fetchall()

mem = SqliteMemory()
mem.add("u-001", "用户喜欢简洁的回答", "preference")
print(mem.search("u-001", "简洁"))

import json
from openai import OpenAI

client = OpenAI()

def embed(text: str) -> list:
    resp = client.embeddings.create(model="text-embedding-3-small", input=text)
    return resp.data[0].embedding

# 1. 记忆库：文本 + 向量（演示用列表，真实项目用向量数据库）
memory_store = []

def remember(text: str, kind: str = "fact"):
    memory_store.append({"text": text, "kind": kind, "vector": embed(text)})

def recall(query: str, top_k: int = 3):
    """找和 query 意思最接近的 top_k 条记忆"""
    q_vec = embed(query)
    scored = []
    for item in memory_store:
        # 余弦相似度：越接近 1 越相似
        a, b = q_vec, item["vector"]
        dot = sum(x * y for x, y in zip(a, b))
        norm = (sum(x * x for x in a) ** 0.5) * (sum(y * y for y in b) ** 0.5)
        scored.append((dot / norm, item["text"], item["kind"]))
    scored.sort(reverse=True)
    return scored[:top_k]

remember("用户喜欢简洁的回答")
remember("用户早上 9 点上班")
remember("用户养了一只叫咪咪的猫")

for score, text, kind in recall("这个人讨厌长篇大论"):
    print(score, text, kind)

import chromadb

client = chromadb.Client()
collection = client.get_or_create_collection("agent_memory")

# 写入（embedding 可以自己传，也可以让 Chroma 用默认模型）
collection.add(
    ids=["1", "2"],
    documents=["用户喜欢简洁的回答", "用户养了一只猫"],
    metadatas=[{"kind": "preference"}, {"kind": "fact"}],
)

# 语义检索
results = collection.query(query_texts=["这个人讨厌啰嗦"], n_results=2)
print(results["documents"])