nanobot 会话与记忆系统深度分析

@dataclass
class Session:
    key: str  # channel:chat_id
    messages: list[dict[str, Any]] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
    metadata: dict[str, Any] = field(default_factory=dict)

{
    "role": "user" | "assistant",
    "content": "消息内容",
    "timestamp": "2024-01-15T10:30:00.123456",
    # 可选的额外字段通过 **kwargs 注入
}

def __init__(self, workspace: Path):
    self.workspace = workspace
    self.sessions_dir = ensure_dir(Path.home() / ".nanobot" / "sessions")
    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)
    if session is None:
        session = Session(key=key)

    self._cache[key] = session
    return session

def _load(self, key: str) -> Session | None:
    path = self._get_session_path(key)

    if not path.exists():
        return None

    try:
        messages = []
        metadata = {}
        created_at = None

        with open(path) as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue

                data = json.loads(line)

                if data.get("_type") == "metadata":
                    metadata = data.get("metadata", {})
                    created_at = datetime.fromisoformat(data["created_at"])
                else:
                    messages.append(data)

        return Session(
            key=key,
            messages=messages,
            created_at=created_at or datetime.now(),
            metadata=metadata
        )
    except Exception as e:
        logger.warning(f"Failed to load session {key}: {e}")
        return None

def save(self, session: Session) -> None:
    path = self._get_session_path(session.key)

    with open(path, "w") as f:
        # 先写入元数据行
        metadata_line = {
            "_type": "metadata",
            "created_at": session.created_at.isoformat(),
            "updated_at": session.updated_at.isoformat(),
            "metadata": session.metadata
        }
        f.write(json.dumps(metadata_line) + "\n")

        # 写入所有消息
        for msg in session.messages:
            f.write(json.dumps(msg) + "\n")

    self._cache[session.key] = session

def get_history(self, max_messages: int = 50) -> list[dict[str, Any]]:
    # 获取最近的 max_messages 条
    recent = self.messages[-max_messages:] if len(self.messages) > max_messages else self.messages

    # 转换为 LLM 格式（只保留 role 和 content）
    return [{"role": m["role"], "content": m["content"]} for m in recent]

def append_today(self, content: str) -> None:
    today_file = self.get_today_file()

    if today_file.exists():
        existing = today_file.read_text(encoding="utf-8")
        content = existing + "\n" + content
    else:
        # 新建文件时添加日期标题
        header = f"# {today_date()}\n\n"
        content = header + content

    today_file.write_text(content, encoding="utf-8")

def write_long_term(self, content: str) -> None:
    self.memory_file.write_text(content, encoding="utf-8")

def get_recent_memories(self, days: int = 7) -> str:
    from datetime import timedelta

    memories = []
    today = datetime.now().date()

    for i in range(days):
        date = today - timedelta(days=i)
        date_str = date.strftime("%Y-%m-%d")
        file_path = self.memory_dir / f"{date_str}.md"

        if file_path.exists():
            content = file_path.read_text(encoding="utf-8")
            memories.append(content)

    return "\n\n---\n\n".join(memories)

def _get_session_path(self, key: str) -> Path:
    safe_key = safe_filename(key.replace(":", "_"))
    return self.sessions_dir / f"{safe_key}.jsonl"

def safe_filename(name: str) -> str:
    """Convert a string to a safe filename."""
    # 替换不安全字符
    unsafe = '<>:"/\\|?*'
    for char in unsafe:
        name = name.replace(char, "_")
    return name.strip()

@property
def session_key(self) -> str:
    """Unique key for session identification."""
    return f"{self.channel}:{self.chat_id}"

async def _process_message(self, msg: InboundMessage) -> OutboundMessage | None:
    # 获取或创建会话
    session = self.sessions.get_or_create(msg.session_key)

    # 更新工具上下文
    message_tool = self.tools.get("message")
    if isinstance(message_tool, MessageTool):
        message_tool.set_context(msg.channel, msg.chat_id)

    # 构建初始消息
    messages = self.context.build_messages(
        history=session.get_history(),  # 从会话获取历史
        current_message=msg.content,
        media=msg.media if msg.media else None,
    )

    # Agent Loop 处理...
    # ...

    # 保存到会话
    session.add_message("user", msg.content)
    session.add_message("assistant", final_content)
    self.sessions.save(session)

    return OutboundMessage(
        channel=msg.channel,
        chat_id=msg.chat_id,
        content=final_content
    )

def build_messages(
    self,
    history: list[dict[str, Any]],  # 来自 session.get_history()
    current_message: str,
    skill_names: list[str] | None = None,
    media: list[str] | None = None,
) -> list[dict[str, Any]]:
    messages = []

    # 1. System prompt (包含记忆)
    system_prompt = self.build_system_prompt(skill_names)
    messages.append({"role": "system", "content": system_prompt})

    # 2. Conversation history
    messages.extend(history)

    # 3. Current user message
    user_content = self._build_user_content(current_message, media)
    messages.append({"role": "user", "content": user_content})

    return messages

def build_system_prompt(self, skill_names: list[str] | None = None) -> str:
    parts = []

    # 1. 核心身份
    parts.append(self._get_identity())

    # 2. Bootstrap 文件 (AGENTS.md, SOUL.md, USER.md, etc.)
    bootstrap = self._load_bootstrap_files()
    if bootstrap:
        parts.append(bootstrap)

    # 3. Memory 上下文
    memory = self.memory.get_memory_context()  # ← 这里集成记忆
    if memory:
        parts.append(f"# Memory\n\n{memory}")

    # 4. Skills (always-loaded 和 available)
    # ...

    return "\n\n---\n\n".join(parts)

def get_memory_context(self) -> str:
    """
    Get memory context for the agent.

    Returns:
        Formatted memory context including long-term and recent memories.
    """
    parts = []

    # 长期记忆
    long_term = self.read_long_term()
    if long_term:
        parts.append("## Long-term Memory\n" + long_term)

    # 今天笔记
    today = self.read_today()
    if today:
        parts.append("## Today's Notes\n" + today)

    return "\n\n".join(parts) if parts else ""