Class Session

tutorials/02_session_and_messages.py:223–302

对话会话 —— 包含一整段对话的所有消息。 注意这里没有 frozen=True，因为我们需要往里面添加消息。 对应源码: session.rs:42-46

@dataclass
class Session:
    """
    对话会话 —— 包含一整段对话的所有消息。

    注意这里没有 frozen=True，因为我们需要往里面添加消息。
    对应源码: session.rs:42-46
    """
    version: int = 1
    messages: list[ConversationMessage] = field(default_factory=list)

    # ---- 序列化（保存到文件）----
    # 序列化是什么？就是把 Python 对象变成可以写入文件的格式（比如 JSON）。
    # 反序列化就是反过来，从文件读取并还原成 Python 对象。

    def to_dict(self) -> dict:
        """把 Session 变成字典（方便转 JSON）"""
        return {
            "version": self.version,
            "messages": [self._message_to_dict(msg) for msg in self.messages],
        }

    def save_to_file(self, path: str) -> None:
        """保存到 JSON 文件（存档）"""
        with open(path, "w", encoding="utf-8") as f:
            json.dump(self.to_dict(), f, ensure_ascii=False, indent=2)

    @classmethod
    def load_from_file(cls, path: str) -> "Session":
        """从 JSON 文件加载（读档）"""
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        session = cls(version=data["version"])
        for msg_data in data["messages"]:
            session.messages.append(cls._dict_to_message(msg_data))
        return session

    @staticmethod
    def _message_to_dict(msg: ConversationMessage) -> dict:
        result = {"role": msg.role, "blocks": []}
        for block in msg.blocks:
            if isinstance(block, TextBlock):
                result["blocks"].append({"type": "text", "text": block.text})
            elif isinstance(block, ToolUseBlock):
                result["blocks"].append({
                    "type": "tool_use", "id": block.id,
                    "name": block.name, "input": block.input,
                })
            elif isinstance(block, ToolResultBlock):
                result["blocks"].append({
                    "type": "tool_result", "tool_use_id": block.tool_use_id,
                    "tool_name": block.tool_name, "output": block.output,
                    "is_error": block.is_error,
                })
        if msg.usage is not None:
            result["usage"] = {
                "input_tokens": msg.usage.input_tokens,
                "output_tokens": msg.usage.output_tokens,
            }