|
|
|
|
|
"""Conversation Memory Plugin""" |
|
|
from typing import List, Dict, Any |
|
|
from collections import deque |
|
|
import pandas as pd |
|
|
|
|
|
class ConversationMemory: |
|
|
"""Track conversation history and context.""" |
|
|
def __init__(self, max_history: int = 50): |
|
|
self.history = deque(maxlen=max_history) |
|
|
self.context = {} |
|
|
|
|
|
def add_message(self, role: str, content: str, metadata: Dict[str, Any] = None): |
|
|
message = {"role": role, "content": content, "timestamp": pd.Timestamp.now().isoformat()} |
|
|
if metadata: |
|
|
message["metadata"] = metadata |
|
|
self.history.append(message) |
|
|
|
|
|
def get_history(self, last_n: int = None) -> List[Dict[str, Any]]: |
|
|
return list(self.history)[-last_n:] if last_n else list(self.history) |
|
|
|
|
|
def clear(self): |
|
|
self.history.clear() |
|
|
self.context.clear() |
|
|
|
