| 106 | |
| 107 | @dataclass |
| 108 | class ChatSession: |
| 109 | id: str |
| 110 | created_at: str |
| 111 | updated_at: str |
| 112 | model: str |
| 113 | language: str |
| 114 | title: str |
| 115 | turn_count: int |
| 116 | history: list[dict[str, Any]] |
| 117 | user_turns: list[str] |
| 118 | assistant_texts: list[str] |
| 119 | path: Path |
| 120 | |
| 121 | @classmethod |
| 122 | def new(cls, kb_dir: Path, model: str, language: str) -> "ChatSession": |
| 123 | now = _utcnow_iso() |
| 124 | sid = _gen_id() |
| 125 | return cls( |
| 126 | id=sid, |
| 127 | created_at=now, |
| 128 | updated_at=now, |
| 129 | model=model, |
| 130 | language=language, |
| 131 | title="", |
| 132 | turn_count=0, |
| 133 | history=[], |
| 134 | user_turns=[], |
| 135 | assistant_texts=[], |
| 136 | path=chats_dir(kb_dir) / f"{sid}.json", |
| 137 | ) |
| 138 | |
| 139 | def to_dict(self) -> dict[str, Any]: |
| 140 | return { |
| 141 | "id": self.id, |
| 142 | "created_at": self.created_at, |
| 143 | "updated_at": self.updated_at, |
| 144 | "model": self.model, |
| 145 | "language": self.language, |
| 146 | "title": self.title, |
| 147 | "turn_count": self.turn_count, |
| 148 | "history": self.history, |
| 149 | "user_turns": self.user_turns, |
| 150 | "assistant_texts": self.assistant_texts, |
| 151 | } |
| 152 | |
| 153 | def save(self) -> None: |
| 154 | self.path.parent.mkdir(parents=True, exist_ok=True) |
| 155 | tmp = self.path.with_suffix(".json.tmp") |
| 156 | tmp.write_text( |
| 157 | json.dumps(self.to_dict(), ensure_ascii=False, indent=2, default=str), |
| 158 | encoding="utf-8", |
| 159 | ) |
| 160 | os.replace(tmp, self.path) |
| 161 | |
| 162 | def record_turn( |
| 163 | self, |
| 164 | user_message: str, |
| 165 | assistant_text: str, |