diff --git a/tools/graph_store.py b/tools/graph_store.py new file mode 100644 index 000000000..e89a3e52f --- /dev/null +++ b/tools/graph_store.py @@ -0,0 +1,64 @@ +"""Sovereign Knowledge Graph Store for Hermes Agent. + +Provides a simple triple-store (Subject, Predicate, Object) persisted +to Timmy's sovereign Gitea instance. +""" + +import json +import base64 +import logging +from typing import List, Dict, Any, Optional +from tools.gitea_client import GiteaClient + +logger = logging.getLogger(__name__) + +class GraphStore: + def __init__(self, repo: str = "Timmy_Foundation/timmy-config", path: str = "memories/knowledge_graph.json"): + self.repo = repo + self.path = path + self.gitea = GiteaClient() + + def _load_graph(self) -> Dict[str, Any]: + try: + content = self.gitea.get_file(self.repo, self.path) + raw = base64.b64decode(content["content"]).decode() + return json.loads(raw) + except Exception: + return {"triples": [], "entities": {}} + + def _save_graph(self, graph: Dict[str, Any], message: str): + sha = None + try: + existing = self.gitea.get_file(self.repo, self.path) + sha = existing.get("sha") + except: + pass + + content_b64 = base64.b64encode(json.dumps(graph, indent=2).encode()).decode() + if sha: + self.gitea.update_file(self.repo, self.path, content_b64, message, sha) + else: + self.gitea.create_file(self.repo, self.path, content_b64, message) + + def add_triples(self, triples: List[Dict[str, str]]): + """Adds a list of triples: [{'s': '...', 'p': '...', 'o': '...'}]""" + graph = self._load_graph() + added_count = 0 + for t in triples: + if t not in graph["triples"]: + graph["triples"].append(t) + added_count += 1 + + if added_count > 0: + self._save_graph(graph, f"Add {added_count} triples to knowledge graph") + return added_count + + def query(self, subject: Optional[str] = None, predicate: Optional[str] = None, object: Optional[str] = None) -> List[Dict[str, str]]: + graph = self._load_graph() + results = [] + for t in graph["triples"]: + if subject and t['s'] != subject: continue + if predicate and t['p'] != predicate: continue + if object and t['o'] != object: continue + results.append(t) + return results