from __future__ import annotations
from typing import Any, Dict, Optional
from langchain_community.chat_message_histories import ZepChatMessageHistory
try:
from langchain.memory import ConversationBufferMemory
class ZepMemory(ConversationBufferMemory):
"""将您的链历史持久化到Zep MemoryStore。
Zep返回的消息数量以及Zep服务器何时总结聊天历史是可配置的。有关更多详细信息，请参阅Zep文档。
文档：https://docs.getzep.com
示例：
.. code-block:: python
memory = ZepMemory(
session_id=session_id, # 用于标识您的用户或用户的会话
url=ZEP_API_URL, # 您的Zep服务器的URL
api_key=<your_api_key>, # 可选
memory_key="history", # 确保与链的提示模板中使用的键匹配
return_messages=True, # 您的提示模板是否期望一个字符串或一个消息列表？
)
chain = LLMChain(memory=memory,...) # 配置您的链以使用ZepMemory实例
注意：
要在聊天历史记录旁边持久化元数据，您需要创建一个自定义Chain类，覆盖`prep_outputs`方法，以在调用`self.memory.save_context`时包含元数据。
Zep - 用于LLM应用程序的快速、可扩展的构建块
=========
Zep是一个用于将LLM应用程序投入生产的开源平台。在几分钟内从在LangChain或LlamaIndex中构建的原型或自定义应用程序转到生产环境，无需重写代码。
有关服务器安装说明和更多信息，请参阅：
https://docs.getzep.com/deployment/quickstart/
有关zep-python软件包的更多信息，请参见：
https://github.com/getzep/zep-python""" # noqa: E501
chat_memory: ZepChatMessageHistory
def __init__(
self,
session_id: str,
url: str = "http://localhost:8000",
api_key: Optional[str] = None,
output_key: Optional[str] = None,
input_key: Optional[str] = None,
return_messages: bool = False,
human_prefix: str = "Human",
ai_prefix: str = "AI",
memory_key: str = "history",
):
"""初始化ZepMemory。
参数:
session_id (str): 标识您的用户或用户的会话
url (str, optional): 您的Zep服务器的URL。默认为"http://localhost:8000"。
api_key (Optional[str], optional): 您的Zep API密钥。默认为None。
output_key (Optional[str], optional): 用于输出消息的键。
默认为None。
input_key (Optional[str], optional): 用于输入消息的键。
默认为None。
return_messages (bool, optional): 您的提示模板是否期望一个字符串
还是一个消息列表？默认为False
即返回一个字符串。
human_prefix (str, optional): 用于人类消息的前缀。
默认为"Human"。
ai_prefix (str, optional): 用于AI消息的前缀。
默认为"AI"。
memory_key (str, optional): 用于存储的键。
默认为"history"。
确保与链的提示模板中使用的键匹配。
""" # noqa: E501
chat_message_history = ZepChatMessageHistory(
session_id=session_id,
url=url,
api_key=api_key,
)
super().__init__(
chat_memory=chat_message_history,
output_key=output_key,
input_key=input_key,
return_messages=return_messages,
human_prefix=human_prefix,
ai_prefix=ai_prefix,
memory_key=memory_key,
)
[docs] def save_context(
self,
inputs: Dict[str, Any],
outputs: Dict[str, str],
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""将此对话的上下文保存到缓冲区中。
参数：
inputs（Dict[str, Any]）：链的输入。
outputs（Dict[str, str]）：链的输出。
metadata（Optional[Dict[str, Any]，可选）：要与上下文一起保存的任何元数据。默认为None
返回：
无
"""
input_str, output_str = self._get_input_output(inputs, outputs)
self.chat_memory.add_user_message(input_str, metadata=metadata)
self.chat_memory.add_ai_message(output_str, metadata=metadata)
except ImportError:
# Placeholder object
[docs] class ZepMemory: # type: ignore[no-redef]
pass