memory = []
memory.append({"role": "user", "content": "你好"})
memory.append({"role": "assistant", "content": "你好！有什么可以帮你的？"})

import java.util.*;

List<Map<String, String>> memory = new ArrayList<>();
memory.add(Map.of("role", "user", "content", "你好"));
memory.add(Map.of("role", "assistant", "content", "你好！有什么可以帮你的？"));

from collections import deque
from datetime import datetime, timedelta
from typing import Optional
import json

class EpisodicMemory:
    """分层情节记忆：当前轮次 → 滑动窗口 → 历史总结 → 向量检索"""

    def __init__(self,
                 sliding_window_size: int = 20,
                 summary_llm=None,
                 vector_store=None,
                 ttl_days: int = 30):
        self.window = deque(maxlen=sliding_window_size)
        self.summary: Optional[str] = None
        self.raw_store: list = []          # 完整日志，带时间戳
        self.summary_llm = summary_llm     # 用于总结的 LLM
        self.vector_store = vector_store   # 向量检索后端
        self.ttl = timedelta(days=ttl_days)

    def add(self, turn: dict):
        """写入一轮对话，附带时间戳"""
        turn["timestamp"] = datetime.utcnow().isoformat()
        self.raw_store.append(turn)
        self.window.append(turn)

    def build_context(self,
                      need_detail: bool = False,
                      query: Optional[str] = None) -> list:
        """组装最终送入 LLM 的上下文"""
        context = []

        # 1. 注入历史总结
        if self.summary:
            context.append({
                "role": "system",
                "content": f"[历史摘要] {self.summary}"
            })

        # 2. 滑动窗口（最近 K 轮）
        context.extend(list(self.window))

        # 3. 检索补充：按需从向量库召回
        if need_detail and query and self.vector_store:
            retrieved = self.vector_store.similarity_search(query, k=3)
            for doc in retrieved:
                context.append({
                    "role": "system",
                    "content": f"[历史记录] {doc}"
                })

        return context

    def summarize(self):
        """将窗口中较旧的内容压缩为摘要"""
        if not self.summary_llm:
            return
        # 保留最近 5 轮不压缩，压缩之前的
        turns_to_summarize = list(self.window)[:-5]
        if not turns_to_summarize:
            return

        prompt = (
            f"以下是对话历史，请压缩为一段摘要（保留关键决定、"
            f"用户偏好、任务进展）：\n{json.dumps(turns_to_summarize, ensure_ascii=False)}"
        )
        new_summary = self.summary_llm(prompt)

        # 增量合并摘要
        if self.summary:
            self.summary = (
                f"{self.summary}\n[增量] {new_summary}"
            )
        else:
            self.summary = new_summary

    def cleanup_expired(self):
        """删除超出 TTL 的原始日志"""
        cutoff = datetime.utcnow() - self.ttl
        self.raw_store = [
            t for t in self.raw_store
            if t["timestamp"] >= cutoff.isoformat()
        ]

import java.time.Duration;
import java.time.Instant;
import java.util.*;
import java.util.function.Function;

/**
 * 分层情节记忆 —— 滑动窗口 + 历史总结 + 向量检索 + TTL 过期
 */
public class EpisodicMemory {
    private final int slidingWindowSize;
    private final Deque<Map<String, Object>> window;
    private final List<Map<String, Object>> rawStore;
    private String summary;
    private final Function<String, String> summaryLlm;
    private final Function<String, List<String>> vectorStore;
    private final Duration ttl;

    public EpisodicMemory(int slidingWindowSize,
                          Function<String, String> summaryLlm,
                          Function<String, List<String>> vectorStore,
                          Duration ttl) {
        this.slidingWindowSize = slidingWindowSize;
        this.window = new ArrayDeque<>(slidingWindowSize);
        this.rawStore = new ArrayList<>();
        this.summary = null;
        this.summaryLlm = summaryLlm;
        this.vectorStore = vectorStore;
        this.ttl = ttl;
    }

    /** 写入一轮对话，附带时间戳 */
    public void add(Map<String, Object> turn) {
        turn.put("timestamp", Instant.now().toString());
        rawStore.add(new HashMap<>(turn));
        if (window.size() >= slidingWindowSize) {
            window.pollFirst();
        }
        window.addLast(turn);
    }

    /** 组装最终送入 LLM 的上下文 */
    public List<Map<String, Object>> buildContext(
            boolean needDetail, String query) {
        List<Map<String, Object>> context = new ArrayList<>();

        // 1. 注入历史摘要
        if (summary != null && !summary.isEmpty()) {
            context.add(Map.of("role", "system",
                    "content", "[历史摘要] " + summary));
        }

        // 2. 滑动窗口
        context.addAll(window);

        // 3. 向量检索补充
        if (needDetail && query != null && vectorStore != null) {
            List<String> retrieved = vectorStore.apply(query);
            for (String doc : retrieved) {
                context.add(Map.of("role", "system",
                        "content", "[历史记录] " + doc));
            }
        }

        return context;
    }

    /** 将窗口中较旧的内容压缩为摘要 */
    public void summarize() {
        if (summaryLlm == null) return;

        List<Map<String, Object>> toSummarize = new ArrayList<>(window);
        // 保留最近 5 轮不压缩
        if (toSummarize.size() <= 5) return;
        toSummarize = toSummarize.subList(0, toSummarize.size() - 5);

        String prompt = "压缩以下对话历史为一段摘要（保留关键决定、"
                + "用户偏好、任务进展）：" + toSummarize;
        String newSummary = summaryLlm.apply(prompt);

        if (summary != null) {
            summary = summary + "\n[增量] " + newSummary;
        } else {
            summary = newSummary;
        }
    }

    /** 清理超出 TTL 的原始日志 */
    public void cleanupExpired() {
        Instant cutoff = Instant.now().minus(ttl);
        rawStore.removeIf(turn -> {
            String ts = (String) turn.get("timestamp");
            return ts != null && Instant.parse(ts).isBefore(cutoff);
        });
    }
}

{
  "fact": "user.preferred_database",
  "value": "MySQL",
  "confidence": 0.95,
  "source": "explicit_statement",
  "updated_at": "2026-04-20T10:30:00Z",
  "history": [
    {"value": "PostgreSQL", "updated_at": "2026-03-15T08:00:00Z"}
  ]
}

{
  "user_id": "zhangsan",
  "preferences": {
    "response_style": "concise_with_details",
    "code_review_mode": "per_file",
    "language": "zh",
    "code_comments": "en"
  }
}

from dataclasses import dataclass
from typing import Optional

@dataclass
class TokenBudget:
    """每类内容的 Token 预算软限制"""
    system_prompt: int = 1024
    procedural: int = 512
    semantic: int = 1024
    episodic: int = 4096
    current_turn: int = 1536

class ContextBuilder:
    """将三层记忆按策略组装为 LLM 上下文"""

    def __init__(self,
                 procedural_store,
                 semantic_store,
                 episodic_store,
                 budget: Optional[TokenBudget] = None):
        self.procedural_store = procedural_store
        self.semantic_store = semantic_store
        self.episodic_store = episodic_store
        self.budget = budget or TokenBudget()

    def build(self, request: str, user_id: str,
              scene: str = "general") -> list:
        """根据场景决定检索策略"""
        system_parts = []

        # 1. 程序记忆 —— 固定注入，体积小
        procedural = self.procedural_store.get(user_id)
        if procedural:
            system_parts.append({
                "role": "system",
                "content": f"[用户偏好] {self._truncate(procedural, self.budget.procedural)}"
            })

        # 2. 语义记忆 —— 按场景决定是否检索
        if self._should_retrieve(scene):
            semantic = self.semantic_store.query(
                user_id=user_id,
                query=request,
                top_k=5,
                min_score=0.7
            )
            if semantic:
                system_parts.append({
                    "role": "system",
                    "content": f"[相关事实] {self._truncate(semantic, self.budget.semantic)}"
                })

        # 3. 情节记忆 —— 分层加载
        need_detail = self._need_historical_detail(scene, request)
        episodic = self.episodic_store.build_context(
            need_detail=need_detail,
            query=request if need_detail else None
        )

        return system_parts + episodic

    def _should_retrieve(self, scene: str) -> bool:
        """按场景决定是否触发语义检索"""
        retrieval_scenes = {"task_continuation", "qa", "tool_call"}
        return scene in retrieval_scenes

    def _need_historical_detail(self, scene: str, request: str) -> bool:
        """判断是否需要检索历史细节"""
        if scene in {"task_continuation", "tool_call"}:
            return True
        if any(kw in request for kw in ["刚才", "之前", "上次", "还记得"]):
            return True
        return False

    @staticmethod
    def _truncate(text: str, max_tokens: int) -> str:
        """按 Token 预算截断（简易版，生产应使用 tokenizer）"""
        words = text.split()
        if len(words) * 1.3 > max_tokens:  # 粗略估算
            limit = int(max_tokens / 1.3)
            return " ".join(words[:limit]) + "..."
        return text

import java.util.*;

/**
 * 将三层记忆按策略组装为 LLM 上下文。
 * 支持 Token 预算控制和场景驱动的检索策略。
 */
public class ContextBuilder {
    private final ProceduralStore proceduralStore;
    private final SemanticStore semanticStore;
    private final EpisodicMemory episodicStore;
    private final TokenBudget budget;

    public ContextBuilder(ProceduralStore proceduralStore,
                          SemanticStore semanticStore,
                          EpisodicMemory episodicStore,
                          TokenBudget budget) {
        this.proceduralStore = proceduralStore;
        this.semanticStore = semanticStore;
        this.episodicStore = episodicStore;
        this.budget = budget;
    }

    /** 按场景组装上下文 */
    public List<Map<String, Object>> build(String request, String userId, String scene) {
        List<Map<String, Object>> parts = new ArrayList<>();

        // 1. 程序记忆 —— 固定注入
        String procedural = proceduralStore.get(userId);
        if (procedural != null && !procedural.isEmpty()) {
            parts.add(Map.of("role", "system",
                    "content", "[用户偏好] " + truncate(procedural, budget.procedural)));
        }

        // 2. 语义记忆 —— 按场景检索
        if (shouldRetrieve(scene)) {
            List<SemanticResult> semantic = semanticStore.query(userId, request, 5, 0.7);
            if (!semantic.isEmpty()) {
                parts.add(Map.of("role", "system",
                        "content", "[相关事实] " + truncate(semantic.toString(), budget.semantic)));
            }
        }

        // 3. 情节记忆 —— 分层加载
        boolean needDetail = needHistoricalDetail(scene, request);
        List<Map<String, Object>> episodic = episodicStore.buildContext(needDetail, request);
        parts.addAll(episodic);

        return parts;
    }

    private boolean shouldRetrieve(String scene) {
        return Set.of("task_continuation", "qa", "tool_call").contains(scene);
    }

    private boolean needHistoricalDetail(String scene, String request) {
        if (Set.of("task_continuation", "tool_call").contains(scene)) return true;
        return request.contains("刚才") || request.contains("之前")
                || request.contains("上次") || request.contains("还记得");
    }

    private String truncate(String text, int maxTokens) {
        String[] words = text.split("\\s+");
        if (words.length * 1.3 > maxTokens) {
            int limit = (int) (maxTokens / 1.3);
            return String.join(" ", Arrays.copyOf(words, limit)) + "...";
        }
        return text;
    }

    /** Token 预算配置 */
    public record TokenBudget(
            int systemPrompt,
            int procedural,
            int semantic,
            int episodic,
            int currentTurn
    ) {
        public TokenBudget() {
            this(1024, 512, 1024, 4096, 1536);
        }
    }

    // 以下为依赖接口定义
    public interface ProceduralStore {
        String get(String userId);
    }
    public interface SemanticStore {
        List<SemanticResult> query(String userId, String query, int topK, double minScore);
    }
    public record SemanticResult(String fact, double score, String updatedAt) {}
}

import re
from typing import Set, Optional

class ImportanceScorer:
    """记忆重要性评分 —— 多信号加权评估"""

    def __init__(self, current_goal: Optional[str] = None):
        self.current_goal = current_goal

    def score(self, content: str) -> float:
        signals = []

        # 用户明确强调
        if self._has_emphasis(content):
            signals.append((0.9, "user_emphasis"))

        # 包含个人身份信息
        if self._contains_pii(content):
            signals.append((0.85, "pii"))

        # 涉及当前任务目标
        if self._relates_to_goal(content):
            signals.append((0.75, "goal_relevant"))

        # 包含具体数值或时间
        if self._contains_specifics(content):
            signals.append((0.7, "has_specifics"))

        # 情感强烈
        if self._has_strong_sentiment(content):
            signals.append((0.6, "strong_sentiment"))

        # 默认识别
        signals.append((0.3, "default"))

        best_score, best_reason = max(signals, key=lambda x: x[0])
        return best_score

    # ---- 信号检测方法 ----
    _EMPHASIS_KEYWORDS: Set[str] = {"记住", "重要的是", "一定要注意", "千万别", "务必"}

    def _has_emphasis(self, text: str) -> bool:
        return any(kw in text for kw in self._EMPHASIS_KEYWORDS)

    def _contains_pii(self, text: str) -> bool:
        # 简单 PII 模式匹配
        patterns = [
            r'\b\d{3,4}-?\d{4}-?\d{4}\b',  # 手机号/银行卡
            r'[我](?:叫|是|的(?:名字|邮箱|电话)(?:是|为))[一-鿿]+',
        ]
        return any(re.search(p, text) for p in patterns)

    def _relates_to_goal(self, text: str) -> bool:
        if not self.current_goal:
            return False
        goal_keywords = set(self.current_goal.split())
        text_keywords = set(text.split())
        return len(goal_keywords & text_keywords) > 0

    _SPECIFICS_PATTERNS = [
        r'\d+',                             # 数字
        r'\d{4}-\d{2}-\d{2}',              # 日期
        r'https?://\S+',                    # 链接
    ]

    def _contains_specifics(self, text: str) -> bool:
        return any(re.search(p, text) for p in self._SPECIFICS_PATTERNS)

    def _has_strong_sentiment(self, text: str) -> bool:
        strong_words = {"糟糕", "太好了", "气死了", "完美", "绝对不行", "非常满意"}
        return any(w in text for w in strong_words)

import java.util.*;
import java.util.regex.Pattern;

/**
 * 记忆重要性评分 —— 多信号加权评估。
 * 用于判断哪些记忆值得长期保留。
 */
public class ImportanceScorer {

    private final String currentGoal;

    public ImportanceScorer(String currentGoal) {
        this.currentGoal = currentGoal;
    }

    /** 对一段记忆内容进行重要性评分 [0, 1] */
    public double score(String content) {
        List<ScoredSignal> signals = new ArrayList<>();

        if (hasEmphasis(content))
            signals.add(new ScoredSignal(0.9, "user_emphasis"));
        if (containsPii(content))
            signals.add(new ScoredSignal(0.85, "pii"));
        if (relatesToGoal(content))
            signals.add(new ScoredSignal(0.75, "goal_relevant"));
        if (containsSpecifics(content))
            signals.add(new ScoredSignal(0.70, "has_specifics"));
        if (hasStrongSentiment(content))
            signals.add(new ScoredSignal(0.60, "strong_sentiment"));

        signals.add(new ScoredSignal(0.30, "default"));

        return signals.stream()
                .mapToDouble(ScoredSignal::score)
                .max()
                .orElse(0.3);
    }

    // ---- 信号检测 ----

    private static final Set<String> EMPHASIS_KEYWORDS = Set.of(
            "记住", "重要的是", "一定要注意", "千万别", "务必");

    boolean hasEmphasis(String text) {
        return EMPHASIS_KEYWORDS.stream().anyMatch(text::contains);
    }

    private static final Pattern PII_PATTERN = Pattern.compile(
            "\\b\\d{3,4}-?\\d{4}-?\\d{4}\\b");

    boolean containsPii(String text) {
        return PII_PATTERN.matcher(text).find();
    }

    boolean relatesToGoal(String text) {
        if (currentGoal == null || currentGoal.isEmpty()) return false;
        // 简易关键词重叠判断
        Set<String> goalWords = new HashSet<>(Arrays.asList(currentGoal.split("\\s+")));
        Set<String> textWords = new HashSet<>(Arrays.asList(text.split("\\s+")));
        goalWords.retainAll(textWords);
        return !goalWords.isEmpty();
    }

    private static final Pattern SPECIFICS_PATTERN = Pattern.compile(
            "\\d+|\\d{4}-\\d{2}-\\d{2}|https?://\\S+");

    boolean containsSpecifics(String text) {
        return SPECIFICS_PATTERN.matcher(text).find();
    }

    private static final Set<String> STRONG_SENTIMENT_WORDS = Set.of(
            "糟糕", "太好了", "气死了", "完美", "绝对不行", "非常满意");

    boolean hasStrongSentiment(String text) {
        return STRONG_SENTIMENT_WORDS.stream().anyMatch(text::contains);
    }

    private record ScoredSignal(double score, String reason) {}
}

from langchain.memory import ConversationSummaryBufferMemory
from langchain_community.chat_message_histories import SQLChatMessageHistory

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,
    chat_memory=SQLChatMessageHistory(
        session_id="user_123",
        connection_string="sqlite:///memory.db"
    ),
    return_messages=True
)

from crewai import Agent, Task, Crew, Process
from crewai.memory import LongTermMemory

agent = Agent(
    role="数据分析师",
    goal="分析用户数据并提供洞察",
    memory=True,                     # 开启记忆
    long_term_memory=LongTermMemory() # 使用长期记忆
)

from mem0 import Memory

m = Memory(
    vector_store="qdrant",          # 向量存储
    graph_store="neo4j",            # 图存储（可选）
    llm="gpt-4",                    # 用于事实提取
)

# 写入记忆
m.add("用户说：我喜欢 PostgreSQL，但最近换成了 MySQL", user_id="zhangsan")

# 检索记忆
results = m.search("用户用什么数据库", user_id="zhangsan")
print(results)
# → "用户偏好数据库从 PostgreSQL 切换到了 MySQL（更新于 2026-04-20）"

import org.springframework.ai.chat.memory.ChatMemory;
import org.springframework.ai.chat.memory.InMemoryChatMemory;
import org.springframework.ai.chat.memory.jdbc.JdbcChatMemory;

// 内存实现 —— 开发和测试
ChatMemory memory = new InMemoryChatMemory();

// JDBC 实现 —— 生产环境，自动建表
ChatMemory jdbcMemory = new JdbcChatMemory(dataSource);

// 使用：在 ConversationChain 中注入
ConversationChain chain = new ConversationChain(
    chatClient,
    "user-session-001",
    jdbcMemory
);

ReActAgent agent = ReActAgent.builder()
        .name("analyst")
        .sysPrompt("你是研究员，先澄清问题，再调用工具")
        .memory(windowMemory)    // 短期记忆
        .maxTurns(8)
        .build();

agent.observe(Msg.user("请分析 A 公司利润变化")).block();  // 只写记忆，不触发模型
Msg answer = agent.call().block();                         // 读取记忆，执行推理

// 写入路径
protected Mono<Void> doObserve(Msg msg) {
    memory.addMessage(msg);   // 追加到滑动窗口
    return Mono.empty();
}

// 读取路径 — call() 时构建快照
public Mono<Msg> call() {
    return plannerLoop(memory.snapshot());  // 快照不可变，避免并发问题
}

try (MsgHub hub = MsgHub.builder()
        .name("analysis_hub")
        .participants(researcher, riskReviewer, writer)
        .announcement(Msg.system("请围绕同一主题协作"))
        .enableAutoBroadcast(true)
        .build()) {

    hub.enter().block();
    researcher.call().block();   // call 完成后，结果自动 broadcast
    riskReviewer.call().block(); // 上一轮结果已通过 observe 写入记忆
    writer.call().block();
}

for (int i = 0; i < maxTurns; i++) {
    ModelOutput out = llm.generate(memory.snapshot());  // 读快照
    if (out.hasToolCall()) {
        ToolResult tr = toolGateway.invoke(out.toolCall());
        memory.addMessage(Msg.tool(tr));                // 写回记忆
        continue;
    }
    return Msg.assistant(out.finalAnswer());
}

# MEMORY.md 存放的内容
- 持久事实（用户的长期偏好、项目约定）
- 环境信息（开发环境、API Key 位置）
- 已确定的决策记录

# USER.md 存放的内容
- 用户基本信息
- 交互风格偏好
- 高频需求模式

# Hermes 通过 session_search 工具按需检索历史
result = agent.search_session("我们之前讨论过什么数据库方案？")
# → FTS5 全文匹配，返回相关历史片段

# 默认：memory-core —— 本地 Markdown + 混合检索
from openclaw.memory import MemoryCore
memory = MemoryCore()  # BM25 + 向量混合搜索

# 可选：memory-lancedb —— 本地向量数据库
from openclaw.memory import LanceDBMemory
memory = LanceDBMemory()  # 自动召回 + 自动捕获

from autogen import ConversableAgent, AssistantAgent

# Agent 1：用户代理
user = ConversableAgent("user", 
    max_consecutive_auto_reply=0,
    human_input_mode="ALWAYS")

# Agent 2：带记忆的助手
assistant = AssistantAgent("assistant",
    llm_config={"config_list": [{"model": "gpt-4"}]},
    system_message="你是有记忆的助手")
    
# 多轮对话 —— 记忆在 chat_history 中累积
user.initiate_chat(assistant, message="我上周说的项目方案还记得吗？")

# 查看历史
print(len(assistant.chat_history))  # 累计消息数