Session Management

动机：为什么需要 Session Management

到目前为止，agent 每次对话都是从零开始的。关掉程序，所有历史都丢了。

但真实场景中，用户会跟 agent 进行多轮对话，可能跨越数小时甚至数天。而且同一个 agent 可能同时服务多个用户——每个用户需要独立的对话历史，不能互相污染。

这就需要两个能力：

• 持久化：对话历史保存到磁盘，重启后恢复
• 隔离：不同对话之间互不影响

Ground Truth：真实的 Session 实现

1@dataclass
2class Session:
3  key: str               # channel:chat_id
4  messages: list[dict] = field(default_factory=list)
5  created_at: datetime = field(default_factory=datetime.now)
6  last_consolidated: int = 0
7 
8  def get_history(self, max_messages=500):
9      """返回未压缩的消息用于 LLM 输入"""
10      unconsolidated = self.messages[self.last_consolidated:]
11      sliced = unconsolidated[-max_messages:]
12      # 跳过开头的非 user 消息，避免孤立的 tool_result
13      for i, m in enumerate(sliced):
14          if m.get("role") == "user":
15              return sliced[i:]
16      return sliced
17 
18class SessionManager:
19  def __init__(self, workspace: Path):
20      self.sessions_dir = workspace / "sessions"
21      self._cache: dict[str, Session] = {}
22 
23  def get_or_create(self, key: str) -> Session:
24      if key in self._cache:
25          return self._cache[key]
26      session = self._load(key) or Session(key=key)
27      self._cache[key] = session
28      return session
29 
30  def save(self, session: Session) -> None:
31      """JSONL 格式: 第一行 metadata, 后续每行一条消息"""
32      path = self._get_session_path(session.key)
33      with open(path, "w") as f:
34          f.write(json.dumps(metadata) + "\n")
35          for msg in session.messages:
36              f.write(json.dumps(msg) + "\n")

1// ironclaw 的 SessionManager (真实源码)
2 
3pub struct SessionManager {
4  // RwLock: 多个读取者 OR 一个写入者
5  sessions: RwLock<HashMap<String, Arc<Mutex<Session>>>>,
6  // 外部 thread ID → 内部 UUID 映射
7  thread_map: RwLock<HashMap<ThreadKey, Uuid>>,
8  // 每个 thread 的 undo 状态
9  undo_managers: RwLock<HashMap<Uuid, Arc<Mutex<UndoManager>>>>,
10}
11 
12impl SessionManager {
13  pub async fn get_or_create_session(
14      &self, user_id: &str
15  ) -> Arc<Mutex<Session>> {
16      // 快路径: 读锁检查是否存在
17      {
18          let sessions = self.sessions.read().await;
19          if let Some(s) = sessions.get(user_id) {
20              return Arc::clone(s);
21          }
22      }
23      // 慢路径: 写锁创建新 session
24      let mut sessions = self.sessions.write().await;
25      // 再次检查 (double-check locking)
26      if let Some(s) = sessions.get(user_id) {
27          return Arc::clone(s);
28      }
29      let session = Arc::new(Mutex::new(Session::new()));
30      sessions.insert(user_id.to_string(), Arc::clone(&session));
31      session
32  }
33}

关键观察：

• nanobot 用 JSONL 文件存储——简单直接，get_history() 做窗口化时会跳过开头的非 user 消息（避免孤立的 tool_result）
• ironclaw 用 RwLock<HashMap> + double-check locking——多线程安全的内存 session 管理，比文件 IO 快但不持久化
• 两者都有 get_or_create pattern——先查缓存，未命中则创建。这是从无状态 API 到有状态对话的标准桥梁
• nanobot 的 last_consolidated 字段标记了已压缩的消息边界——为 s05 的 context 压缩留了接口

构建：添加 SessionManager

本课新增两个类：

@dataclass
class Session:
    """一个对话会话"""
    key: str
    messages: list[dict] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)

    def add_message(self, role: str, content: str, **kwargs):
        self.messages.append({
            "role": role, "content": content,
            "timestamp": datetime.now().isoformat(), **kwargs
        })

    def get_history(self, max_messages: int = 100) -> list[dict]:
        """返回最近的消息用于 LLM 输入"""
        return self.messages[-max_messages:]


class SessionManager:
    """管理多个独立会话"""
    def __init__(self, workspace: Path):
        self.sessions_dir = workspace / "sessions"
        self.sessions_dir.mkdir(exist_ok=True)
        self._cache: dict[str, Session] = {}

    def get_or_create(self, key: str) -> Session:
        if key in self._cache:
            return self._cache[key]
        session = self._load(key) or Session(key=key)
        self._cache[key] = session
        return session

    def save(self, session: Session):
        path = self.sessions_dir / f"{session.key}.jsonl"
        with open(path, "w") as f:
            for msg in session.messages:
                f.write(json.dumps(msg, ensure_ascii=False) + "\n")

Agent loop 的变化很小——在调用 LLM 前加载历史，调用后保存：

def agent_loop(query: str, registry, context, sessions, session_key="default"):
    session = sessions.get_or_create(session_key)
    history = session.get_history()
    system = context.build_system_prompt()

    messages = [{"role": "system", "content": system}, *history,
                {"role": "user", "content": query}]
    # ... while loop 不变 ...

    # 循环结束后保存新消息
    sessions.save(session)

测试：多会话隔离

mgr = SessionManager(Path("."))

# 会话 A
session_a = mgr.get_or_create("user_alice")
session_a.add_message("user", "写一个排序算法")

# 会话 B
session_b = mgr.get_or_create("user_bob")
session_b.add_message("user", "写一个 HTTP server")

# 互不影响
assert len(session_a.messages) == 1
assert len(session_b.messages) == 1
assert session_a.messages[0]["content"] != session_b.messages[0]["content"]

变更内容

本课代码: agents/s04_session_management.py — 168 行 (新增 28 行)

试一试

cd public/code
python agents/s04_session_management.py "记住我叫小明"
python agents/s04_session_management.py "我叫什么名字？"

可以尝试的提示:

1. 运行一次告诉 agent 一些信息，退出后再运行一次看是否记住
2. 用不同的 session key 验证会话隔离
3. 检查 sessions/ 目录下生成的 JSONL 文件

距离生产

消息历史的微妙处理: get_history() 的 leading-message 清理

nanobot 的 Session.get_history() 有一个容易忽略的细节：它会跳过开头的非 user 消息。为什么？因为当历史被窗口化截断时，可能刚好从一个 assistant 消息或 tool_result 开始。LLM API 要求对话以 user 消息开头——孤立的 tool_result 没有对应的 tool_use，会导致 API 报错。这个小小的 for 循环解决了一个真实的 edge case。

并发安全: acquireSessionWriteLock()

OpenClaw 使用 acquireSessionWriteLock() 确保同一个 session 不会被并发写入。在 Web 环境中，用户可能快速连续发送多条消息，或者 sub-agent 和主 agent 同时操作同一个 session。没有锁的话，JSONL 文件会出现交错写入导致数据损坏。

其他差距

• JSONL 不适合大量 session——生产环境用 PostgreSQL/SQLite，支持索引和查询
• 无 session 过期清理——旧 session 会永远占用磁盘，生产系统有 TTL 和清理策略

第一性原理思考

Session 的本质是”有状态的对话”——而 LLM API 本身是无状态的。每次调用都需要把完整的对话历史发送过去。Session management 就是在无状态 API 上模拟有状态交互——这和 HTTP session/cookie 的设计动机完全一致。Web 应用用 session ID 映射到服务端状态，agent 用 session key 映射到消息历史。理解了这一点，就能理解为什么 get_or_create 模式在所有实现中都出现——它就是 “从无状态到有状态” 的标准桥梁。