Source code for langchain_community.chat_message_histories.upstash_redis

import json
import logging
from typing import List, Optional

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import (
    BaseMessage,
    message_to_dict,
    messages_from_dict,
)

logger = logging.getLogger(__name__)


[docs] class UpstashRedisChatMessageHistory(BaseChatMessageHistory): """Chat message history stored in an Upstash Redis database."""
[docs] def __init__( self, session_id: str, url: str = "", token: str = "", key_prefix: str = "message_store:", ttl: Optional[int] = None, ): try: from upstash_redis import Redis except ImportError: raise ImportError( "Could not import upstash redis python package. " "Please install it with `pip install upstash_redis`." ) if url == "" or token == "": raise ValueError( "UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN are needed." ) try: self.redis_client = Redis(url=url, token=token) except Exception: logger.error("Upstash Redis instance could not be initiated.") self.session_id = session_id self.key_prefix = key_prefix self.ttl = ttl
@property def key(self) -> str: """Construct the record key to use""" return self.key_prefix + self.session_id @property def messages(self) -> List[BaseMessage]: # type: ignore """Retrieve the messages from Upstash Redis""" _items = self.redis_client.lrange(self.key, 0, -1) items = [json.loads(m) for m in _items[::-1]] messages = messages_from_dict(items) return messages
[docs] def add_message(self, message: BaseMessage) -> None: """Append the message to the record in Upstash Redis""" self.redis_client.lpush(self.key, json.dumps(message_to_dict(message))) if self.ttl: self.redis_client.expire(self.key, self.ttl)
[docs] def clear(self) -> None: """Clear session memory from Upstash Redis""" self.redis_client.delete(self.key)