Source code for langchain.memory.motorhead_memory
from typing import Any, Dict, List, Optional
import requests
from langchain.memory.chat_memory import BaseChatMemory
from langchain.schema.messages import get_buffer_string
MANAGED_URL = "https://api.getmetal.io/v1/motorhead"
# LOCAL_URL = "http://localhost:8080"
[docs]class MotorheadMemory(BaseChatMemory):
    """Chat message memory backed by Motorhead service."""
    url: str = MANAGED_URL
    timeout: int = 3000
    memory_key: str = "history"
    session_id: str
    context: Optional[str] = None
    # Managed Params
    api_key: Optional[str] = None
    client_id: Optional[str] = None
    def __get_headers(self) -> Dict[str, str]:
        is_managed = self.url == MANAGED_URL
        headers = {
            "Content-Type": "application/json",
        }
        if is_managed and not (self.api_key and self.client_id):
            raise ValueError(
                """
                You must provide an API key or a client ID to use the managed
                version of Motorhead. Visit https://getmetal.io for more information.
                """
            )
        if is_managed and self.api_key and self.client_id:
            headers["x-metal-api-key"] = self.api_key
            headers["x-metal-client-id"] = self.client_id
        return headers
[docs]    async def init(self) -> None:
        res = requests.get(
            f"{self.url}/sessions/{self.session_id}/memory",
            timeout=self.timeout,
            headers=self.__get_headers(),
        )
        res_data = res.json()
        res_data = res_data.get("data", res_data)  # Handle Managed Version