|
1 | 1 | import json
|
2 | 2 | from pathlib import Path
|
3 |
| -from typing import List |
| 3 | +from typing import List, Optional |
4 | 4 |
|
5 | 5 | from langchain_core.chat_history import (
|
6 | 6 | BaseChatMessageHistory,
|
|
11 | 11 | class FileChatMessageHistory(BaseChatMessageHistory):
|
12 | 12 | """Chat message history that stores history in a local file."""
|
13 | 13 |
|
14 |
| - def __init__(self, file_path: str) -> None: |
| 14 | + def __init__( |
| 15 | + self, |
| 16 | + file_path: str, |
| 17 | + *, |
| 18 | + encoding: Optional[str] = None, |
| 19 | + ensure_ascii: bool = True, |
| 20 | + ) -> None: |
15 | 21 | """Initialize the file path for the chat history.
|
16 |
| -
|
17 | 22 | Args:
|
18 | 23 | file_path: The path to the local file to store the chat history.
|
| 24 | + encoding: The encoding to use for file operations. Defaults to None. |
| 25 | + ensure_ascii: If True, escape non-ASCII in JSON. Defaults to True. |
19 | 26 | """
|
20 | 27 | self.file_path = Path(file_path)
|
| 28 | + self.encoding = encoding |
| 29 | + self.ensure_ascii = ensure_ascii |
| 30 | + |
21 | 31 | if not self.file_path.exists():
|
22 | 32 | self.file_path.touch()
|
23 |
| - self.file_path.write_text(json.dumps([])) |
| 33 | + self.file_path.write_text( |
| 34 | + json.dumps([], ensure_ascii=self.ensure_ascii), encoding=self.encoding |
| 35 | + ) |
24 | 36 |
|
25 | 37 | @property
|
26 | 38 | def messages(self) -> List[BaseMessage]: # type: ignore
|
27 | 39 | """Retrieve the messages from the local file"""
|
28 |
| - items = json.loads(self.file_path.read_text()) |
| 40 | + items = json.loads(self.file_path.read_text(encoding=self.encoding)) |
29 | 41 | messages = messages_from_dict(items)
|
30 | 42 | return messages
|
31 | 43 |
|
32 | 44 | def add_message(self, message: BaseMessage) -> None:
|
33 | 45 | """Append the message to the record in the local file"""
|
34 | 46 | messages = messages_to_dict(self.messages)
|
35 | 47 | messages.append(messages_to_dict([message])[0])
|
36 |
| - self.file_path.write_text(json.dumps(messages)) |
| 48 | + self.file_path.write_text( |
| 49 | + json.dumps(messages, ensure_ascii=self.ensure_ascii), encoding=self.encoding |
| 50 | + ) |
37 | 51 |
|
38 | 52 | def clear(self) -> None:
|
39 | 53 | """Clear session memory from the local file"""
|
40 |
| - self.file_path.write_text(json.dumps([])) |
| 54 | + self.file_path.write_text( |
| 55 | + json.dumps([], ensure_ascii=self.ensure_ascii), encoding=self.encoding |
| 56 | + ) |
0 commit comments