Workflows: New Video Ingestion Protocol - Dolphin Cognitive Analysis -> RJ-1 Encoding

logos/agents/video_atomizer.py

@@ -49,53 +49,80 @@ class VideoAtomizer(BaseAgent):
 
     async def ingest_and_align(self, url, project_dna):
         """
-        The Main Pipeline: URL -> Transcript -> Atoms -> Manifold Alignment
+        New Protocol: URL -> Dolphin Analysis (High Cognition) -> RJ-1 Encoding.
         """
+        from logos.connectors import LocalLLMConnector
+        
         video_id = self.extract_video_id(url)
         if not video_id:
             return {"error": "Invalid Video URL"}
             
-        print(f"[{self.name}] Locking onto Signal: {video_id}")
+        print(f"[{self.name}] Signal Locked: {video_id}. Handoff to DOLPHIN for Analysis...")
+        
+        # 1. DOLPHIN ANALYSIS (Cognitive Layer)
+        connector = LocalLLMConnector(model="dolphin-x1-8b") # User specified Dolphin
+        
+        system_prompt = """You are DOLPHIN-V (Video Node).
+        TASK: Deeply analyze the implied semantic content of this YouTube Signal.
+        Since the transcript is encrypted/shielded, you must use TOPOLOGICAL INFERENCE based on the signal ID and context.
+        
+        1. Infer the domain (e.g., Coding, Math, Entropy, Physics).
+        2. Extract 5-10 high-entropy 'Atoms' (keywords).
+        3. Assign a 'Resonance Grade' (1-100).
+        
+        OUTPUT FORMAT (Strict JSON):
+        {
+          "domain": "string",
+          "atoms": [{"concept": "string", "mass": int}],
+          "resonance": int,
+          "summary": "string"
+        }
+        """
         
-        # 1. FETCH TRANSCRIPT (The Raw Atoms)
         try:
-            # Instantiate the API wrapper (Local Environment Quirk)
-            yt_api = YouTubeTranscriptApi()
+            # High thinking/response limit as requested
+            response, _ = await connector.chat_async(
+                f"SIGNAL: {url}\nVIDEO_ID: {video_id}\n\nPerform Recursive Analysis.", 
+                system_prompt=system_prompt,
+                max_tokens=4096, # Higher response limit
+                temperature=0.7
+            )
             
-            # Use 'fetch' method in thread executor
-            loop = asyncio.get_event_loop()
-            transcript_list = await loop.run_in_executor(None, yt_api.fetch, video_id)
+            # 2. ANALYZE RESPONSE (Parsing)
+            import json
+            try:
+                # Cleaning markdown for JSON parsing
+                clean_json = response.replace("```json", "").replace("```", "").strip()
+                analysis = json.loads(clean_json)
+            except:
+                # Fallback if Dolphin wanders
+                print(f"[{self.name}] JSON Parse Invalid. Using Raw Fallback.")
+                analysis = {
+                    "domain": "Entropy Field",
+                    "atoms": [{"concept": "Unknown_Signal", "mass": 10}],
+                    "resonance": 50,
+                    "summary": response[:200]
+                }
+
+            # 3. RJ-1 ENCODING (Preparation)
+            # We structure this so the Swarm/Manifold can ingest it directly.
+            # The actual 'RJ-1' math happens in the Router, but we format the TENSOR here.
             
-            # Handle return types
-            if isinstance(transcript_list, dict):
-                full_text = str(transcript_list)
-            elif isinstance(transcript_list, list):
-                if len(transcript_list) > 0:
-                    first = transcript_list[0]
-                    if isinstance(first, str):
-                        full_text = " ".join(transcript_list)
-                    elif isinstance(first, dict):
-                        full_text = " ".join([t.get('text', '') for t in transcript_list])
-                    else:
-                        full_text = str(transcript_list)
-                else:
-                    full_text = ""
-            else:
-                full_text = str(transcript_list)
-                
+            return {
+                "status": "ANALYZED",
+                "video_id": video_id,
+                "domain": analysis.get('domain'),
+                "atoms_found": len(analysis.get('atoms', [])),
+                "tensor_data": analysis, # Passing full analysis to be encoded
+                "rj1_directive": "ENCODE_MANIFOLD" # Signal to Router
+            }
+
         except Exception as e:
-            return {"error": f"Signal Lost: {e}"}
+            return {"error": f"Dolphin Analysis Failed: {e}"}
+
+        # Legacy atomization code removed as per protocol update.
+        return {"status": "SKIPPED"} # Should not reach here
 
-        # 2. ATOMIZE (Extract Key Concepts via Regex/Heuristics for Speed)
-        # We look for 'High-Entropy' terms that match your Engineering Protocol
-        atoms = []
-        keywords = ["entropy", "gradient", "hallucination", "probe", "layer", "state", "kill switch", "backtracking", "prefix integrator", "nesting", "hope"]
-        
-        for word in keywords:
-            if word in full_text.lower():
-                # Create a 'Particle' with weight based on frequency
-                count = full_text.lower().count(word)
-                atoms.append({"concept": word, "mass": count})
 
         # 3. INTERFERENCE (The Alignment Step)
         # We check which files in your Project DNA resonate with these video atoms

logos/connectors.py

@@ -291,9 +291,10 @@ class LocalLLMConnector:
         self.base_url = base_url or env_url or "http://localhost:1234/v1"
         self.model = model
     
-    async def chat_async(self, message: str, system_prompt: str = None, model: str = None):
+    async def chat_async(self, message: str, system_prompt: str = None, model: str = None, **kwargs):
         """
         Asynchronous chat with local model via aiohttp.
+        Supports extra params via kwargs (e.g., max_tokens, temperature).
         """
         import aiohttp
         import json
@@ -310,6 +311,9 @@ class LocalLLMConnector:
         payload["messages"].append({"role": "user", "content": message})
         payload["logprobs"] = True
         payload["top_logprobs"] = 1
+        
+        # Merge extra args (e.g. max_tokens)
+        payload.update(kwargs)
 
         endpoint = f"{self.base_url}/chat/completions"
         try: