Spaces:
Sleeping
Sleeping
| from datetime import datetime | |
| from typing import Dict, List, Optional, Any, Union | |
| from uuid import uuid4 | |
| import json | |
| class Context: | |
| def __init__(self, context_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, max_history_length: int = 100): | |
| self.context_id = context_id or str(uuid4()) | |
| self.metadata = metadata or {} | |
| self.max_history_length = max_history_length | |
| self.history: List[Dict[str, Any]] = [] | |
| self.state: Dict[str, Any] = {} | |
| self.created_at = datetime.utcnow() | |
| self.updated_at = self.created_at | |
| def add_interaction(self, input_data: Dict[str, Any], output_data: Dict[str, Any]) -> None: | |
| interaction = { | |
| "timestamp": datetime.utcnow().isoformat(), | |
| "input": input_data, | |
| "output": output_data | |
| } | |
| self.history.append(interaction) | |
| if len(self.history) > self.max_history_length: | |
| self.history = self.history[-self.max_history_length:] | |
| self.updated_at = datetime.utcnow() | |
| def update_state(self, new_state: Dict[str, Any], merge: bool = True) -> None: | |
| if merge: | |
| self.state.update(new_state) | |
| else: | |
| self.state = new_state | |
| self.updated_at = datetime.utcnow() | |
| def get_history(self, limit: Optional[int] = None) -> List[Dict[str, Any]]: | |
| if limit is None or limit >= len(self.history): | |
| return self.history | |
| return self.history[-limit:] | |
| def to_dict(self) -> Dict[str, Any]: | |
| return { | |
| "context_id": self.context_id, | |
| "metadata": self.metadata, | |
| "state": self.state, | |
| "history": self.history, | |
| "created_at": self.created_at.isoformat(), | |
| "updated_at": self.updated_at.isoformat(), | |
| } | |
| def to_json(self) -> str: | |
| return json.dumps(self.to_dict()) | |
| def from_dict(cls, data: Dict[str, Any]) -> 'Context': | |
| context = cls(context_id=data.get("context_id")) | |
| context.metadata = data.get("metadata", {}) | |
| context.state = data.get("state", {}) | |
| context.history = data.get("history", []) | |
| if "created_at" in data: | |
| context.created_at = datetime.fromisoformat(data["created_at"]) | |
| if "updated_at" in data: | |
| context.updated_at = datetime.fromisoformat(data["updated_at"]) | |
| return context | |
| def from_json(cls, json_str: str) -> 'Context': | |
| data = json.loads(json_str) | |
| return cls.from_dict(data) | |
| class ContextManager: | |
| def __init__(self): | |
| self.contexts: Dict[str, Context] = {} | |
| def create_context(self, metadata: Optional[Dict[str, Any]] = None, context_id: Optional[str] = None) -> Context: | |
| context = Context(context_id=context_id, metadata=metadata) | |
| self.contexts[context.context_id] = context | |
| return context | |
| def get_context(self, context_id: str) -> Optional[Context]: | |
| return self.contexts.get(context_id) | |
| def update_context(self, context_id: str, updates: Dict[str, Any]) -> Optional[Context]: | |
| context = self.get_context(context_id) | |
| if not context: | |
| return None | |
| if "metadata" in updates: | |
| context.metadata.update(updates["metadata"]) | |
| if "state" in updates: | |
| context.update_state(updates["state"]) | |
| return context | |
| def delete_context(self, context_id: str) -> bool: | |
| if context_id in self.contexts: | |
| del self.contexts[context_id] | |
| return True | |
| return False | |
| def list_contexts(self) -> List[str]: | |
| return list(self.contexts.keys()) | |