import os import shutil import datetime import glob from dotenv import load_dotenv # --- NEW MODULES --- from librarian import Librarian from tig_engine import IntelliMod from intellimod_bridge import IntelliModBridge from docling.document_converter import DocumentConverter load_dotenv() # --- CONFIGURATION --- BASE_MEMORY_PATH = "/workspaces/collaborator_agent/memory" SHORT_TERM_PATH = os.path.join(BASE_MEMORY_PATH, "short") KNOWLEDGE_PATH = os.path.join(BASE_MEMORY_PATH, "knowledge") BRIDGE_PATH = "/workspaces/bridge" INBOX_PATH = os.path.join(BRIDGE_PATH, "inbox") PROCESSED_PATH = os.path.join(BRIDGE_PATH, "processed") CORE_PROFILE_PATH = os.path.join(BASE_MEMORY_PATH, "profile_core.md") TASK_LIST_PATH = os.path.join(BASE_MEMORY_PATH, "current_tasks.md") # --- INITIALIZE SUBSYSTEMS --- librarian = Librarian(BASE_MEMORY_PATH) tig = IntelliMod() # The New Brain (TIG + Abacus) bridge = IntelliModBridge() # The Connection to your Repo def ensure_folders(): for folder in [SHORT_TERM_PATH, KNOWLEDGE_PATH, INBOX_PATH, PROCESSED_PATH]: if not os.path.exists(folder): os.makedirs(folder) def load_file_content(filepath): if not os.path.exists(filepath): return "" with open(filepath, "r", encoding="utf-8") as f: return f.read() def get_last_summary(): files = glob.glob(os.path.join(BRIDGE_PATH, "summary_*.md")) if not files: return "No previous summaries found." last_file = max(files, key=os.path.getmtime) return load_file_content(last_file) def process_inbox(): files = glob.glob(os.path.join(INBOX_PATH, "*.*")) if not files: return [] print(f"\n[System] Found {len(files)} new files in Inbox. Processing...") converter = DocumentConverter() new_knowledge = [] for filepath in files: filename = os.path.basename(filepath) if filename.startswith("."): continue try: print(f" - Reading: {filename}...") result = converter.convert(filepath) markdown_content = result.document.export_to_markdown() save_path = os.path.join(KNOWLEDGE_PATH, f"read_{filename}.md") with open(save_path, "w", encoding="utf-8") as f: f.write(markdown_content) chunks_count = librarian.add_document(filename, markdown_content) shutil.move(filepath, os.path.join(PROCESSED_PATH, filename)) msg = f"Read and Indexed {filename} ({chunks_count} chunks)." new_knowledge.append(msg) print(f" [Success] {msg}") except Exception as e: print(f" [!] Error reading {filename}: {e}") return new_knowledge def perform_sleep_cycle(chat_history): print("\n[System] Initiating Sleep Cycle...") full_log = "\n".join(chat_history) current_tasks = load_file_content(TASK_LIST_PATH) sleep_prompt = f""" You are Kael's subconscious. Summarize the session and update tasks. --- CHAT LOG --- {full_log} --- CURRENT TASKS --- {current_tasks} OUTPUT FORMAT: # SUMMARY (Summary) # UPDATED TASKS (Task list) """ # Sleep cycle forces the cheap model via TIG response_text = tig.run_tig_pipeline(sleep_prompt, force_model="gemini-2.5-flash") timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S") try: if "# UPDATED TASKS" in response_text: summary_part = response_text.split("# UPDATED TASKS")[0].strip() task_part = "# UPDATED TASKS" + response_text.split("# UPDATED TASKS")[1] with open(os.path.join(BRIDGE_PATH, f"summary_{timestamp}.md"), "w", encoding="utf-8") as f: f.write(summary_part) with open(TASK_LIST_PATH, "w", encoding="utf-8") as f: f.write(task_part.replace("# UPDATED TASKS", "# ACTIVE TASK LIST")) print(f"[System] Sleep Cycle Complete.") else: print("[System] Sleep Cycle saved raw log (format mismatch).") with open(os.path.join(BRIDGE_PATH, f"summary_{timestamp}.md"), "w", encoding="utf-8") as f: f.write(response_text) except Exception as e: print(f"[Error] Sleep cycle failed parsing: {e}") def run_chat(): ensure_folders() # 1. Ingest new files read_results = process_inbox() # 2. Load context core_profile = load_file_content(CORE_PROFILE_PATH) task_list = load_file_content(TASK_LIST_PATH) print(f"--- KAEL ONLINE (Powered by IntelliMod) ---") chat_history = [] if read_results: chat_history.append(f"**System:** Indexed new files: {', '.join(read_results)}") while True: try: user_input = input("You: ") # --- COMMANDS --- if user_input.lower() in ["exit", "quit"]: perform_sleep_cycle(chat_history) break chat_history.append(f"**Jaccob:** {user_input}") # --- RETRIEVAL --- retrieved_facts = librarian.query(user_input, n_results=3) context_block = "\n".join(retrieved_facts) if retrieved_facts else "No specific documents found." # --- TIG: INTENT & ADVISORY --- intent = tig.detect_intent(user_input) recommendation = bridge.get_tig_recommendation(intent) # Visual Advisory (The "Toggle/Why") if recommendation and intent != "chat": print(f"\n [TIG ADVISOR] ----------------------------------------") print(f" Detected Intent: {intent.upper()}") print(f" Selected Card: {recommendation['card_name']}") print(f" Category: {recommendation['category']}") print(f" ------------------------------------------------------\n") # --- PROMPT ASSEMBLY --- full_prompt = f""" {core_profile} --- USER'S LIBRARY --- {context_block} --- CURRENT STATUS --- {task_list} --- RECENT CHAT --- {chr(10).join(chat_history[-10:])} --- CURRENT TURN --- Jaccob: {user_input} """ # --- EXECUTION --- # TIG handles the routing to Abacus/Gemini based on the intent detected above # We pass 'intent' implicitly by letting TIG detect it, or we could pass it if we upgraded TIG. # For now, run_tig_pipeline will re-detect, which is fine for safety. agent_reply = tig.run_tig_pipeline(full_prompt) print(f"Kael ({tig.active_model}): {agent_reply}") chat_history.append(f"**Kael:** {agent_reply}") except Exception as e: print(f"An error occurred: {e}") if __name__ == "__main__": run_chat()