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

@@ -0,0 +1,73 @@
+"""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 (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__)
+
+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."""
+        logger.info(f"Learning about: {topic}")
+        
+        # 1. Search and Analyze
+        prompt = f"""
+Please perform a deep dive into the following topic: {topic}
+
+Use Google Search to find the most recent and relevant information.
+Analyze the findings and provide a structured 'Knowledge Fragment' in Markdown format.
+Include:
+- Summary of the topic
+- Key facts and recent developments
+- Implications for Timmy's sovereign mission
+- References (URLs)
+"""
+        result = self.adapter.generate(
+            model="gemini-3.1-pro-preview",
+            prompt=prompt,
+            system_instruction="You are Timmy's Sovereign Knowledge Ingester. Your goal is to find and synthesize high-fidelity information from Google Search.",
+            grounding=True,
+            thinking=True
+        )
+        
+        knowledge_fragment = result["text"]
+        
+        # 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:
+            sha = None
+            try:
+                existing = self.gitea.get_file(repo, filename)
+                sha = existing.get("sha")
+            except:
+                pass
+            
+            content_b64 = base64.b64encode(knowledge_fragment.encode()).decode()
+            
+            if sha:
+                self.gitea.update_file(repo, filename, content_b64, f"Update knowledge on {topic}", sha)
+            else:
+                self.gitea.create_file(repo, filename, content_b64, f"Initial knowledge on {topic}")
+                
+            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 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

skills/research/realtime_learning.py

@@ -0,0 +1,22 @@
+"""
+---
+title: Real-time Learning
+description: Allows Timmy to learn about any topic in real-time using Google Search and persist it to his sovereign memory.
+conditions:
+  - New information required
+  - Real-time events or trends
+  - Knowledge base expansion
+---
+"""
+
+from agent.knowledge_ingester import KnowledgeIngester
+
+def learn(topic: str) -> str:
+    """
+    Performs real-time learning on a topic and updates Timmy's memory.
+    
+    Args:
+        topic: The topic to learn about (e.g., 'recent advancements in quantum computing').
+    """
+    ingester = KnowledgeIngester()
+    return ingester.learn_about(topic)

tools/gitea_client.py

@@ -0,0 +1,59 @@
+"""
+Gitea API Client — typed, sovereign, zero-dependency.
+
+Enables the agent to interact with Timmy's sovereign Gitea instance
+for issue tracking, PR management, and knowledge persistence.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.request
+import urllib.error
+import urllib.parse
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Optional, Dict, List
+
+class GiteaClient:
+    def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None):
+        self.base_url = base_url or os.environ.get("GITEA_URL", "http://143.198.27.163:3000")
+        self.token = token or os.environ.get("GITEA_TOKEN")
+        self.api = f"{self.base_url.rstrip('/')}/api/v1"
+
+    def _request(self, method: str, path: str, data: Optional[dict] = None) -> Any:
+        url = f"{self.api}{path}"
+        body = json.dumps(data).encode() if data else None
+        req = urllib.request.Request(url, data=body, method=method)
+        if self.token:
+            req.add_header("Authorization", f"token {self.token}")
+        req.add_header("Content-Type", "application/json")
+        req.add_header("Accept", "application/json")
+
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                raw = resp.read().decode()
+                return json.loads(raw) if raw else {}
+        except urllib.error.HTTPError as e:
+            raise Exception(f"Gitea {e.code}: {e.read().decode()}") from e
+
+    def get_file(self, repo: str, path: str, ref: str = "main") -> Dict[str, Any]:
+        return self._request("GET", f"/repos/{repo}/contents/{path}?ref={ref}")
+
+    def create_file(self, repo: str, path: str, content: str, message: str, branch: str = "main") -> Dict[str, Any]:
+        data = {
+            "branch": branch,
+            "content": content, # Base64 encoded
+            "message": message
+        }
+        return self._request("POST", f"/repos/{repo}/contents/{path}", data)
+
+    def update_file(self, repo: str, path: str, content: str, message: str, sha: str, branch: str = "main") -> Dict[str, Any]:
+        data = {
+            "branch": branch,
+            "content": content, # Base64 encoded
+            "message": message,
+            "sha": sha
+        }
+        return self._request("PUT", f"/repos/{repo}/contents/{path}", data)

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