core[patch],community[patch]: Move file chat history back to community (#20834)

Marking as patch since we haven't had releases in between. This just reverting part of a PR from yesterday.
This commit is contained in:
Eugene Yurtsev 2024-04-24 12:47:25 -04:00 committed by GitHub
parent 6debadaa70
commit 30e48c9878
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 41 additions and 42 deletions

View File

@ -1,5 +1,40 @@
from langchain_core.chat_history import FileChatMessageHistory import json
from pathlib import Path
from typing import List
__all__ = [ from langchain_core.chat_history import (
"FileChatMessageHistory", BaseChatMessageHistory,
] )
from langchain_core.messages import BaseMessage, messages_from_dict, messages_to_dict
class FileChatMessageHistory(BaseChatMessageHistory):
"""Chat message history that stores history in a local file."""
def __init__(self, file_path: str) -> None:
"""Initialize the file path for the chat history.
Args:
file_path: The path to the local file to store the chat history.
"""
self.file_path = Path(file_path)
if not self.file_path.exists():
self.file_path.touch()
self.file_path.write_text(json.dumps([]))
@property
def messages(self) -> List[BaseMessage]: # type: ignore
"""Retrieve the messages from the local file"""
items = json.loads(self.file_path.read_text())
messages = messages_from_dict(items)
return messages
def add_message(self, message: BaseMessage) -> None:
"""Append the message to the record in the local file"""
messages = messages_to_dict(self.messages)
messages.append(messages_to_dict([message])[0])
self.file_path.write_text(json.dumps(messages))
def clear(self) -> None:
"""Clear session memory from the local file"""
self.file_path.write_text(json.dumps([]))

View File

@ -3,10 +3,10 @@ from pathlib import Path
from typing import Generator from typing import Generator
import pytest import pytest
from langchain_core.chat_history import FileChatMessageHistory
from langchain_core.messages import AIMessage, HumanMessage from langchain_core.messages import AIMessage, HumanMessage
from langchain_community.chat_message_histories import FileChatMessageHistory
@pytest.fixture @pytest.fixture
def file_chat_message_history() -> Generator[FileChatMessageHistory, None, None]: def file_chat_message_history() -> Generator[FileChatMessageHistory, None, None]:

View File

@ -16,9 +16,7 @@
""" # noqa: E501 """ # noqa: E501
from __future__ import annotations from __future__ import annotations
import json
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from pathlib import Path
from typing import List, Sequence, Union from typing import List, Sequence, Union
from langchain_core.messages import ( from langchain_core.messages import (
@ -26,8 +24,6 @@ from langchain_core.messages import (
BaseMessage, BaseMessage,
HumanMessage, HumanMessage,
get_buffer_string, get_buffer_string,
messages_from_dict,
messages_to_dict,
) )
from langchain_core.pydantic_v1 import BaseModel, Field from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import run_in_executor from langchain_core.runnables import run_in_executor
@ -215,35 +211,3 @@ class InMemoryChatMessageHistory(BaseChatMessageHistory, BaseModel):
async def aclear(self) -> None: async def aclear(self) -> None:
self.clear() self.clear()
class FileChatMessageHistory(BaseChatMessageHistory):
"""Chat message history that stores history in a local file."""
def __init__(self, file_path: str) -> None:
"""Initialize the file path for the chat history.
Args:
file_path: The path to the local file to store the chat history.
"""
self.file_path = Path(file_path)
if not self.file_path.exists():
self.file_path.touch()
self.file_path.write_text(json.dumps([]))
@property
def messages(self) -> List[BaseMessage]: # type: ignore
"""Retrieve the messages from the local file"""
items = json.loads(self.file_path.read_text())
messages = messages_from_dict(items)
return messages
def add_message(self, message: BaseMessage) -> None:
"""Append the message to the record in the local file"""
messages = messages_to_dict(self.messages)
messages.append(messages_to_dict([message])[0])
self.file_path.write_text(json.dumps(messages))
def clear(self) -> None:
"""Clear session memory from the local file"""
self.file_path.write_text(json.dumps([]))