【Agent工坊】LLM也需要睡觉：从顶会论文到Anthropic Dreams，AI Agent长时记忆实战指南

# Agent上下文管理配置示例

context_management:

  trigger_threshold: 0.65 # 窗口使用65%时触发压缩

  compression_strategy: hierarchical # 分层摘要策略

  keep_last_n_turns: 10 # 保留最近10轮完整对话（用户最关心当前语境）

  summary_categories: # 摘要的结构化分类

    - decisions_made # 已做出的决策（不可逆的）

    - tasks_pending # 未解决的任务

    - constraints_active # 当前生效的约束条件

    - user_preferences # 用户显式表达的偏好

你是一个对话摘要器。请将以上对话压缩为结构化摘要。

必须保留：

1. 所有已做出的决策（含决策理由）

2. 所有尚未完成的任务

3. 用户明确的偏好和约束

4. 关键数据（数字、日期、人名、文件名）

可以丢弃：

1. 中间推理过程

2. 已完成的简单确认

3. 重复内容

输出格式：JSON，每个分类一个字段，不超过500字

# session_tracker.py

import time

from datetime import datetime, timedelta

class SessionTracker:

    """

    跟踪Agent的会话状态，判断何时该"睡觉"

    """

    def __init__(self, idle_threshold_minutes=10):

        self.idle_threshold = timedelta(minutes=idle_threshold_minutes)

        self.last_activity = datetime.now()

        self.current_log = [] # 当前会话的完整记录

        self.task_queue = [] # 待完成任务队列

        self.min_active_time = timedelta(minutes=5) # 最小活跃时间

    def on_message(self, role: str, content: str):

        """记录每条消息"""

        self.last_activity = datetime.now()

        self.current_log.append({

            "ts": datetime.now().isoformat(),

            "role": role,

            "content": content[:1000] # 截断超长内容

        })

    def add_task(self, task: str):

        """添加待完成任务"""

        self.task_queue.append({

            "task": task,

            "added_at": datetime.now().isoformat()

        })

    def complete_task(self, task: str):

        """标记任务完成"""

        self.task_queue = [t for t in self.task_queue if t["task"] != task]

    def should_sleep(self) -> bool:

        """判断是否应该触发睡眠"""

        idle_time = datetime.now() - self.last_activity

        session_duration = datetime.now() - datetime.fromisoformat(

            self.current_log[0]["ts"]) if self.current_log else timedelta(0)

        # 条件1：空闲超过阈值

        # 条件2：至少活跃过一段时间（避免刚启动就睡）

        # 条件3：没有待完成任务

        return (idle_time > self.idle_threshold and

                session_duration > self.min_active_time and

                len(self.task_queue) == 0)

# sleep_processor.py

import json, os

from datetime import datetime

from typing import List, Dict

class SleepProcessor:

    """

    核心记忆消化逻辑

    """

    def __init__(self, memory_path: str = "/tmp/agent_memory.json",

                 max_memories: int = 50):

        self.memory_path = memory_path

        self.max_memories = max_memories

        os.makedirs(os.path.dirname(memory_path), exist_ok=True)

    def consolidate(self, session_log: List[Dict], llm) -> Dict:

        """

        核心"睡眠"逻辑：

        1. 生成会话摘要

        2. 提取模式

        3. 更新记忆库

        4. 清理过期记忆

        """

        # 1. 生成结构化摘要

        summary = llm.summarize(session_log, categories=[

            "decisions", "errors", "successes", "preferences"

        ])

        # 2. 提取可复用的模式

        patterns = llm.extract_patterns(session_log)

        # 3. 构建记忆条目

        entry = {

            "date": datetime.now().isoformat(),

            "summary": summary,

            "patterns": patterns,

            "message_count": len(session_log),

            "duration_minutes": self._calc_duration(session_log)

        }

        # 4. 更新并裁剪记忆库

        memory = self._load()

        memory["entries"].append(entry)

        memory["total_sessions"] += 1

        memory["last_sleep"] = datetime.now().isoformat()

        # 保留最近N条，其余归档

        if len(memory["entries"]) > self.max_memories:

            archived = memory["entries"][:-self.max_memories]

            self._archive(archived)

            memory["entries"] = memory["entries"][-self.max_memories:]

        self._save(memory)

        return memory

    def get_wake_context(self, max_entries: int = 5) -> str:

        """

        唤醒时加载记忆，注入系统提示词

        """

        memory = self._load()

        recent = memory["entries"][-max_entries:]

        if not recent:

            return ""

        ctx = "## 来自过去的记忆\n\n"

        ctx += f"已积累 {memory['total_sessions']} 次会话的经验。"

        ctx += f"以下是最近的关键经验：\n\n"

        for i, entry in enumerate(recent, 1):

            ctx += f"### 记忆 #{memory['total_sessions'] - len(recent) + i}"

            ctx += f"（{entry['date'][:10]}，{entry['message_count']}条消息）\n"

            for pattern in entry.get("patterns", [])[:3]:

                ctx += f"- {pattern}\n"

            ctx += "\n"

        return ctx

    def _load(self) -> Dict:

        if os.path.exists(self.memory_path):

            with open(self.memory_path) as f:

                return json.load(f)

        return {"entries": [], "total_sessions": 0, "last_sleep": None}

    def _save(self, memory: Dict):

        with open(self.memory_path, 'w') as f:

            json.dump(memory, f, indent=2, ensure_ascii=False)

    def _archive(self, old_entries: List[Dict]):

        archive_path = self.memory_path.replace('.json', '_archive.json')

        existing = []

        if os.path.exists(archive_path):

            with open(archive_path) as f:

                existing = json.load(f)

        existing.extend(old_entries)

        with open(archive_path, 'w') as f:

            json.dump(existing, f, indent=2, ensure_ascii=False)

    def _calc_duration(self, log: List[Dict]) -> float:

        if len(log) < 2:

            return 0

        start = datetime.fromisoformat(log[0]["ts"])

        end = datetime.fromisoformat(log[-1]["ts"])

        return round((end - start).total_seconds() / 60, 1)

# agent_loop.py - 集成"觉醒-睡眠"循环

def create_agent_with_sleep(base_system_prompt: str):

    tracker = SessionTracker(idle_threshold_minutes=10)

    processor = SleepProcessor()

    # 启动时加载记忆

    wake_ctx = processor.get_wake_context()

    full_prompt = f"{base_system_prompt}\n\n{wake_ctx}" if wake_ctx else base_system_prompt

    class AgentWithSleep:

        def __init__(self):

            self.system_prompt = full_prompt

            self.conversation = [{"role": "system", "content": full_prompt}]

        def process(self, user_input: str) -> str:

            # 记录用户输入

            tracker.on_message("user", user_input)

            self.conversation.append({"role": "user", "content": user_input})

            # 调用LLM

            response = call_llm(self.conversation)

            # 记录Agent响应

            tracker.on_message("assistant", response)

            self.conversation.append({"role": "assistant", "content": response})

            # 检查是否需要压缩上下文

            if estimate_tokens(self.conversation) > CONTEXT_LIMIT * 0.65:

                self.conversation = compress_conversation(self.conversation)

            return response

        def try_sleep(self):

            """在空闲时调用，检查是否应该睡眠"""

            if tracker.should_sleep():

                memory = processor.consolidate(tracker.current_log, call_llm)

                tracker.current_log = []

                print(f"Agent已睡眠 | 总会话数: {memory['total_sessions']} | "

                      f"最近模式: {memory['entries'][-1]['patterns'][:2]}")

                return True

            return False

    return AgentWithSleep()

以下信息绝对不能丢失：

- 用户明确表达的偏好（"我更喜欢方案A"）

- 已做出的不可逆决策（"确定使用PostgreSQL"）

- 当前会话的约束条件（"预算上限$500"）