feat: enhance Knowledge Ingester with symbolic extraction

agent/gemini_adapter.py

@@ -0,0 +1,90 @@
+"""Native Gemini 3 Series adapter for Hermes Agent.
+
+Leverages the google-genai SDK to provide sovereign access to Gemini's
+unique capabilities: Thinking (Reasoning) tokens, Search Grounding,
+and Maps Grounding.
+"""
+
+import logging
+import os
+from typing import Any, Dict, List, Optional, Union
+
+try:
+    from google import genai
+    from google.genai import types
+except ImportError:
+    genai = None  # type: ignore
+    types = None  # type: ignore
+
+logger = logging.getLogger(__name__)
+
+class GeminiAdapter:
+    def __init__(self, api_key: Optional[str] = None):
+        self.api_key = api_key or os.environ.get("GEMINI_API_KEY")
+        if not self.api_key:
+            logger.warning("GEMINI_API_KEY not found in environment.")
+        
+        if genai:
+            self.client = genai.Client(api_key=self.api_key)
+        else:
+            self.client = None
+
+    def generate(
+        self,
+        model: str,
+        prompt: str,
+        system_instruction: Optional[str] = None,
+        thinking: bool = False,
+        thinking_budget: int = 16000,
+        grounding: bool = False,
+        **kwargs
+    ) -> Dict[str, Any]:
+        if not self.client:
+            raise ImportError("google-genai SDK not installed. Run 'pip install google-genai'.")
+
+        config = {}
+        if system_instruction:
+            config["system_instruction"] = system_instruction
+        
+        if thinking:
+            # Gemini 3 series thinking config
+            config["thinking_config"] = {"include_thoughts": True}
+            # max_output_tokens includes thinking tokens
+            kwargs["max_output_tokens"] = kwargs.get("max_output_tokens", 32000) + thinking_budget
+
+        tools = []
+        if grounding:
+            tools.append({"google_search": {}})
+        
+        if tools:
+            config["tools"] = tools
+
+        response = self.client.models.generate_content(
+            model=model,
+            contents=prompt,
+            config=types.GenerateContentConfig(**config, **kwargs)
+        )
+
+        result = {
+            "text": response.text,
+            "usage": {
+                "prompt_tokens": response.usage_metadata.prompt_token_count,
+                "candidates_tokens": response.usage_metadata.candidates_token_count,
+                "total_tokens": response.usage_metadata.total_token_count,
+            }
+        }
+
+        # Extract thoughts if present
+        thoughts = []
+        for part in response.candidates[0].content.parts:
+            if hasattr(part, 'thought') and part.thought:
+                thoughts.append(part.thought)
+        
+        if thoughts:
+            result["thoughts"] = "\n".join(thoughts)
+
+        # Extract grounding metadata
+        if response.candidates[0].grounding_metadata:
+            result["grounding"] = response.candidates[0].grounding_metadata
+
+        return result

agent/knowledge_ingester.py

@@ -1,13 +1,14 @@
 """Sovereign Knowledge Ingester for Hermes Agent.
 
 Uses Gemini 3.1 Pro to learn from Google Search in real-time and
-persists the knowledge to Timmy's sovereign memory.
+persists the knowledge to Timmy's sovereign memory (both Markdown and Symbolic).
 """
 
 import logging
 import base64
 from typing import Any, Dict, List, Optional
 from agent.gemini_adapter import GeminiAdapter
+from agent.symbolic_memory import SymbolicMemory
 from tools.gitea_client import GiteaClient
 
 logger = logging.getLogger(__name__)
@@ -16,6 +17,7 @@ class KnowledgeIngester:
     def __init__(self):
         self.adapter = GeminiAdapter()
         self.gitea = GiteaClient()
+        self.symbolic = SymbolicMemory()
 
     def learn_about(self, topic: str) -> str:
         """Searches Google, analyzes the results, and saves the knowledge."""
@@ -43,12 +45,14 @@ Include:
         
         knowledge_fragment = result["text"]
         
-        # 2. Persist to Timmy's Memory
+        # 2. Extract Symbolic Triples
+        self.symbolic.ingest_text(knowledge_fragment)
+        
+        # 3. Persist to Timmy's Memory (Markdown)
         repo = "Timmy_Foundation/timmy-config"
         filename = f"memories/realtime_learning/{topic.lower().replace(' ', '_')}.md"
         
         try:
-            # Check if file exists to get SHA
             sha = None
             try:
                 existing = self.gitea.get_file(repo, filename)
@@ -63,7 +67,7 @@ Include:
             else:
                 self.gitea.create_file(repo, filename, content_b64, f"Initial knowledge on {topic}")
                 
-            return f"Successfully learned about {topic} and updated Timmy's memory at {filename}"
+            return f"Successfully learned about {topic}. Updated Timmy's Markdown memory and Symbolic Knowledge Graph."
         except Exception as e:
             logger.error(f"Failed to persist knowledge: {e}")
-            return f"Learned about {topic}, but failed to save to memory: {e}\n\n{knowledge_fragment}"
+            return f"Learned about {topic}, but failed to save to Markdown memory: {e}\n\n{knowledge_fragment}"

agent/meta_reasoning.py

@@ -0,0 +1,47 @@
+"""Meta-Reasoning Layer for Hermes Agent.
+
+Implements a sovereign self-correction loop where a 'strong' model (Gemini 3.1 Pro)
+critiques the plans generated by the primary agent loop before execution.
+"""
+
+import logging
+from typing import Any, Dict, List, Optional
+from agent.gemini_adapter import GeminiAdapter
+
+logger = logging.getLogger(__name__)
+
+class MetaReasoningLayer:
+    def __init__(self):
+        self.adapter = GeminiAdapter()
+
+    def critique_plan(self, goal: str, proposed_plan: str, context: str) -> Dict[str, Any]:
+        """Critiques a proposed plan using Gemini's thinking capabilities."""
+        prompt = f"""
+Goal: {goal}
+
+Context:
+{context}
+
+Proposed Plan:
+{proposed_plan}
+
+Please perform a deep symbolic and neuro-symbolic analysis of this plan.
+Identify potential risks, logical fallacies, or missing steps.
+Suggest improvements to make the plan more sovereign, cost-efficient, and robust.
+"""
+        try:
+            result = self.adapter.generate(
+                model="gemini-3.1-pro-preview",
+                prompt=prompt,
+                system_instruction="You are a Senior Meta-Reasoning Engine for the Hermes Agent. Your goal is to ensure the agent's plans are flawless and sovereign.",
+                thinking=True,
+                thinking_budget=8000
+            )
+            return {
+                "critique": result["text"],
+                "thoughts": result.get("thoughts", ""),
+                "grounding": result.get("grounding")
+            }
+        except Exception as e:
+            logger.error(f"Meta-reasoning failed: {e}")
+            return {"critique": "Meta-reasoning unavailable.", "error": str(e)}

agent/symbolic_memory.py

@@ -0,0 +1,74 @@
+"""Sovereign Intersymbolic Memory Layer.
+
+Bridges Neural (LLM) and Symbolic (Graph) reasoning by extracting
+structured triples from unstructured text and performing graph lookups.
+"""
+
+import logging
+import json
+from typing import List, Dict, Any
+from agent.gemini_adapter import GeminiAdapter
+from tools.graph_store import GraphStore
+
+logger = logging.getLogger(__name__)
+
+class SymbolicMemory:
+    def __init__(self):
+        self.adapter = GeminiAdapter()
+        self.store = GraphStore()
+
+    def ingest_text(self, text: str):
+        """Extracts triples from text and adds them to the graph."""
+        prompt = f"""
+Extract all meaningful entities and their relationships from the following text.
+Format the output as a JSON list of triples: [{{"s": "subject", "p": "predicate", "o": "object"}}]
+
+Text:
+{text}
+
+Guidelines:
+- Use clear, concise labels for entities and predicates.
+- Focus on stable facts and structural relationships.
+- Predicates should be verbs or descriptive relations (e.g., 'is_a', 'works_at', 'collaborates_with').
+"""
+        try:
+            result = self.adapter.generate(
+                model="gemini-3.1-pro-preview",
+                prompt=prompt,
+                system_instruction="You are Timmy's Symbolic Extraction Engine. Extract high-fidelity knowledge triples.",
+                response_mime_type="application/json"
+            )
+            
+            triples = json.loads(result["text"])
+            if isinstance(triples, list):
+                count = self.store.add_triples(triples)
+                logger.info(f"Ingested {count} new triples into symbolic memory.")
+                return count
+        except Exception as e:
+            logger.error(f"Symbolic ingestion failed: {e}")
+            return 0
+
+    def get_context_for(self, topic: str) -> str:
+        """Performs a 2-hop graph search to find related context for a topic."""
+        # 1. Find direct relations
+        direct = self.store.query(subject=topic) + self.store.query(object=topic)
+        
+        # 2. Find 2nd hop
+        related_entities = set()
+        for t in direct:
+            related_entities.add(t['s'])
+            related_entities.add(t['o'])
+        
+        extended = []
+        for entity in related_entities:
+            if entity == topic: continue
+            extended.extend(self.store.query(subject=entity))
+        
+        all_triples = direct + extended
+        if not all_triples:
+            return ""
+            
+        context = "Symbolic Knowledge Graph Context:\n"
+        for t in all_triples:
+            context += f"- {t['s']} --({t['p']})--> {t['o']}\n"
+        return context

pyproject.toml

@@ -13,7 +13,7 @@ license = { text = "MIT" }
 dependencies = [
   # Core — pinned to known-good ranges to limit supply chain attack surface
   "openai>=2.21.0,<3",
-  "anthropic>=0.39.0,<1",
+  "anthropic>=0.39.0,<1",\n  "google-genai>=1.2.0,<2",
   "python-dotenv>=1.2.1,<2",
   "fire>=0.7.1,<1",
   "httpx>=0.28.1,<1",

skills/creative/sovereign_thinking.py

@@ -0,0 +1,47 @@
+"""
+---
+title: Sovereign Thinking
+description: Pauses the agent to perform deep reasoning on complex problems using Gemini 3.1 Pro.
+conditions:
+  - Complex logic required
+  - High-stakes decision making
+  - Architecture or design tasks
+---
+"""
+
+from agent.gemini_adapter import GeminiAdapter
+
+def think(problem: str, effort: str = "medium") -> str:
+    """
+    Performs deep reasoning on a complex problem.
+    
+    Args:
+        problem: The complex problem or question to analyze.
+        effort: The reasoning effort ('low', 'medium', 'high', 'xhigh').
+    """
+    adapter = GeminiAdapter()
+    
+    budget_map = {
+        "low": 4000,
+        "medium": 16000,
+        "high": 32000,
+        "xhigh": 64000
+    }
+    
+    budget = budget_map.get(effort, 16000)
+    
+    result = adapter.generate(
+        model="gemini-3.1-pro-preview",
+        prompt=problem,
+        system_instruction="You are the internal reasoning engine of the Hermes Agent. Think deeply and provide a structured analysis.",
+        thinking=True,
+        thinking_budget=budget
+    )
+    
+    output = []
+    if result.get("thoughts"):
+        output.append("### Internal Monologue\n" + result["thoughts"])
+    
+    output.append("### Conclusion\n" + result["text"])
+    
+    return "\n\n".join(output)

skills/memory/intersymbolic_graph.py

@@ -0,0 +1,27 @@
+"""
+---
+title: Intersymbolic Graph Query
+description: Queries Timmy's sovereign knowledge graph to find connections and structured facts.
+conditions:
+  - Complex relationship analysis
+  - Fact checking against structured memory
+  - Finding non-obvious connections
+---
+"""
+
+from agent.symbolic_memory import SymbolicMemory
+
+def query_graph(topic: str) -> str:
+    """
+    Queries the knowledge graph for a specific topic and returns structured context.
+    
+    Args:
+        topic: The entity or topic to search for in the graph.
+    """
+    memory = SymbolicMemory()
+    context = memory.get_context_for(topic)
+    
+    if not context:
+        return f"No symbolic connections found for '{topic}' in the knowledge graph."
+        
+    return context

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