第三层：会话管理

📌 核心文件：nanobot/session/manager.py (~100 行)

概述

会话管理（Session Management）负责持久化对话历史，让 Agent 能够记住之前的对话内容。

会话模型

from pydantic import BaseModel

class Session(BaseModel):
    """会话模型"""
    key: str                           # 会话标识（如 "telegram:123456"）
    history: list[dict[str, Any]] = [] # 对话历史
    
    def add_message(self, role: str, content: str):
        """添加消息到历史"""
        self.history.append({
            "role": role,
            "content": content
        })
    
    def get_history(self) -> list[dict[str, Any]]:
        """获取历史（供 LLM 使用的格式）"""
        return self.history

SessionManager

class SessionManager:
    """会话管理器"""
    
    def __init__(self, workspace: Path):
        self.workspace = workspace
        self.sessions_dir = workspace / "sessions"
        self.sessions_dir.mkdir(exist_ok=True)
    
    def get_or_create(self, session_key: str) -> Session:
        """获取或创建会话"""
        session_file = self.sessions_dir / f"{session_key}.json"
        
        if session_file.exists():
            data = json.loads(session_file.read_text())
            return Session(**data)
        
        return Session(key=session_key)
    
    def save(self, session: Session):
        """保存会话"""
        session_file = self.sessions_dir / f"{session.key}.json"
        session_file.write_text(session.model_dump_json(indent=2))

使用方式

# 在 AgentLoop 中
session = self.sessions.get_or_create(msg.session_key)

# 构建消息时使用历史
messages = self.context.build_messages(
    history=session.get_history(),
    current_message=msg.content
)

# 处理完成后保存
session.add_message("user", msg.content)
session.add_message("assistant", final_content)
self.sessions.save(session)

会话文件示例

~/.nanobot/sessions/telegram:123456789.json

{
  "key": "telegram:123456789",
  "history": [
    {
      "role": "user",
      "content": "你好"
    },
    {
      "role": "assistant",
      "content": "你好！有什么可以帮助你的？"
    },
    {
      "role": "user",
      "content": "读取 config.json 文件"
    },
    {
      "role": "assistant",
      "content": "文件内容：..."
    }
  ]
}

优化建议

1. 限制历史长度

def add_message(self, role: str, content: str, max_history: int = 100):
    self.history.append({"role": role, "content": content})
    
    # 保留最近 N 条
    if len(self.history) > max_history:
        self.history = self.history[-max_history:]

2. 定期清理旧会话

def cleanup_old_sessions(self, days: int = 30):
    """删除超过 N 天未使用的会话"""
    cutoff = datetime.now() - timedelta(days=days)
    
    for session_file in self.sessions_dir.glob("*.json"):
        if session_file.stat().st_mtime < cutoff.timestamp():
            session_file.unlink()

小结

✅ 简单的本地 JSON 存储
✅ 支持多渠道独立会话
✅ 易于扩展（可换成数据库）

下一步：09-内置工具详解.md