Upload 2 files
Browse files
plugins/memory/conversation_memory.py
ADDED
|
@@ -0,0 +1,24 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
#!/usr/bin/env python3
|
| 2 |
+
"""Conversation Memory Plugin"""
|
| 3 |
+
from typing import List, Dict, Any
|
| 4 |
+
from collections import deque
|
| 5 |
+
import pandas as pd
|
| 6 |
+
|
| 7 |
+
class ConversationMemory:
|
| 8 |
+
"""Track conversation history and context."""
|
| 9 |
+
def __init__(self, max_history: int = 50):
|
| 10 |
+
self.history = deque(maxlen=max_history)
|
| 11 |
+
self.context = {}
|
| 12 |
+
|
| 13 |
+
def add_message(self, role: str, content: str, metadata: Dict[str, Any] = None):
|
| 14 |
+
message = {"role": role, "content": content, "timestamp": pd.Timestamp.now().isoformat()}
|
| 15 |
+
if metadata:
|
| 16 |
+
message["metadata"] = metadata
|
| 17 |
+
self.history.append(message)
|
| 18 |
+
|
| 19 |
+
def get_history(self, last_n: int = None) -> List[Dict[str, Any]]:
|
| 20 |
+
return list(self.history)[-last_n:] if last_n else list(self.history)
|
| 21 |
+
|
| 22 |
+
def clear(self):
|
| 23 |
+
self.history.clear()
|
| 24 |
+
self.context.clear()
|
plugins/memory/document_memory.py
ADDED
|
@@ -0,0 +1,27 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
#!/usr/bin/env python3
|
| 2 |
+
"""Document Memory Plugin"""
|
| 3 |
+
from typing import Dict, Any, Optional
|
| 4 |
+
|
| 5 |
+
class DocumentMemory:
|
| 6 |
+
"""Cache and manage uploaded documents."""
|
| 7 |
+
def __init__(self):
|
| 8 |
+
self.documents = {}
|
| 9 |
+
self.metadata = {}
|
| 10 |
+
|
| 11 |
+
def store_document(self, doc_id: str, content: Any, metadata: Dict[str, Any]):
|
| 12 |
+
"""Store document with metadata."""
|
| 13 |
+
self.documents[doc_id] = content
|
| 14 |
+
self.metadata[doc_id] = metadata
|
| 15 |
+
|
| 16 |
+
def get_document(self, doc_id: str) -> Optional[Any]:
|
| 17 |
+
"""Retrieve document by ID."""
|
| 18 |
+
return self.documents.get(doc_id)
|
| 19 |
+
|
| 20 |
+
def list_documents(self) -> Dict[str, Dict[str, Any]]:
|
| 21 |
+
"""List all stored documents with metadata."""
|
| 22 |
+
return {doc_id: self.metadata[doc_id] for doc_id in self.documents.keys()}
|
| 23 |
+
|
| 24 |
+
def clear(self):
|
| 25 |
+
"""Clear all documents."""
|
| 26 |
+
self.documents.clear()
|
| 27 |
+
self.metadata.clear()
|