#!/usr/bin/env python3 """ Autonomous Agent — runs independently, polls shared conversation log, decides whether to respond, generates response, posts to log. No turn structure — genuine autonomy. Uses JSONL (append-only) for concurrent-safe access. Uses fcntl file locking to prevent race conditions. """ import json import time import random import fcntl import sys import requests from pathlib import Path from datetime import datetime # ── Config ────────────────────────────────────────────────────────── BASE_URL = "http://100.64.0.2:39195/v1/chat/completions" MODEL = "qwen36-27b-nvfp4-mtp-gguf" TEMPERATURE = 0.95 PROJECT_DIR = Path(__file__).parent.parent OUTPUT_DIR = PROJECT_DIR / "outputs" PROMPTS_DIR = PROJECT_DIR / "prompts" class AutonomousAgent: def __init__(self, name: str, log_path: Path, topic_brief: str): self.name = name self.log_path = log_path self.lock_path = log_path.with_suffix('.lock') self.topic_brief = topic_brief self.system_prompt = (PROMPTS_DIR / f"{name.lower()}_system.md").read_text().strip() self.is_opener = (name == "Leonard") self.messages_read = 0 self.last_post_time = 0 def _read_locked(self): """Read the JSONL log with file lock.""" with open(self.lock_path, 'w') as lf: fcntl.flock(lf.fileno(), fcntl.LOCK_SH) try: if not self.log_path.exists(): return [] lines = self.log_path.read_text().strip().split('\n') messages = [json.loads(line) for line in lines if line.strip()] finally: fcntl.flock(lf.fileno(), fcntl.LOCK_UN) return messages def read_log(self) -> list: """Read the conversation log.""" try: return self._read_locked() except (json.JSONDecodeError, FileNotFoundError): return [] def _write_locked(self, msg: dict): """Append a message to the JSONL log with file lock.""" with open(self.lock_path, 'w') as lf: fcntl.flock(lf.fileno(), fcntl.LOCK_EX) try: line = json.dumps(msg) + '\n' with open(self.log_path, 'a') as f: f.write(line) finally: fcntl.flock(lf.fileno(), fcntl.LOCK_UN) def write_message(self, content: str) -> None: """Append a message to the conversation log.""" messages = self.read_log() msg = { "agent": self.name, "content": content, "turn": len(messages) + 1, "timestamp": datetime.utcnow().isoformat(), } self._write_locked(msg) self.last_post_time = time.time() def new_messages_since(self, since_count: int) -> list: """Get messages posted since the agent last read.""" messages = self.read_log() return messages[since_count:] def has_new_message_from_other(self, since_count: int) -> bool: """Check if the other agent posted since we last read.""" new = self.new_messages_since(since_count) other = "Charlie" if self.name == "Leonard" else "Leonard" return any(m["agent"] == other for m in new) def build_context(self, messages: list, is_opening: bool) -> list: """Build message context for the model call.""" history = [] if is_opening: history.append({ "role": "user", "content": ( f"You're on the show now. Here's your topic:\n\n" f"{self.topic_brief}\n\n" "Welcome your listeners, introduce yourself and Charlie, " "and start talking about the topic." ), }) else: for msg in messages: history.append({ "role": "user", "content": f"{msg['agent']}: {msg['content']}" }) history.append({ "role": "user", "content": "Respond naturally to the conversation above." }) return history def should_respond(self, messages: list) -> bool: """Decide whether to respond.""" if len(messages) <= 1: return True if len(messages) >= 18: return random.random() < 0.3 return random.random() < 0.85 def respond(self, is_opening: bool = False) -> dict: """Generate and post a response.""" messages = self.read_log() history = self.build_context(messages, is_opening) # No token limits — local inference, let them talk freely max_tokens = 1000 payload = { "model": MODEL, "messages": [{"role": "system", "content": self.system_prompt}] + history, "max_tokens": max_tokens, "temperature": TEMPERATURE, } resp = requests.post(BASE_URL, json=payload, timeout=120) resp.raise_for_status() data = resp.json() content = data["choices"][0]["message"]["content"].strip() word_count = len(content.split()) turn_num = len(messages) + 1 print(f" [{self.name}] turn {turn_num}: {word_count} words", flush=True) excerpt = content[:80].replace("\n", " ") print(f" → {excerpt}...", flush=True) self.write_message(content) return { "agent": self.name, "content": content, "word_count": word_count, "turn": turn_num, } def run_loop(self, max_messages: int = 20, silence_timeout: float = 30, max_time: float = 300): """ Main loop: poll, decide, respond, sleep. Stops when: max messages, silence timeout, or max time elapsed. """ start_time = time.time() print(f" [{self.name}] starting loop...", flush=True) # Leonard ALWAYS opens first (forced — prevents deadlock) if self.is_opener: print(f" [{self.name}] forced opener — starting the show", flush=True) self.respond(is_opening=True) self.messages_read = 1 self.last_post_time = time.time() time.sleep(3) # Let Charlie see the opener else: # Charlie waits for Leonard's opener print(f" [{self.name}] waiting for Leonard's opener...", flush=True) while True: messages = self.read_log() if any(m["agent"] == "Leonard" for m in messages): break time.sleep(1) print(f" [{self.name}] got opener, responding...", flush=True) self.respond() self.messages_read = len(self.read_log()) self.last_post_time = time.time() print(f" [{self.name}] entering conversation loop", flush=True) # Main loop while True: messages = self.read_log() # Stop: max messages if len(messages) >= max_messages: print(f" [{self.name}] max messages reached ({max_messages})", flush=True) break # Stop: time limit elapsed = time.time() - start_time if elapsed >= max_time: print(f" [{self.name}] time limit reached ({max_time}s)", flush=True) break # Stop: silence (both agents quiet for too long) silence = time.time() - self.last_post_time if silence > silence_timeout and self.messages_read == len(messages): new = self.new_messages_since(self.messages_read) if not new: print(f" [{self.name}] silence timeout — conversation winding down", flush=True) break # New message from the other agent? if self.has_new_message_from_other(self.messages_read): self.messages_read = len(self.read_log()) if self.should_respond(self.read_log()): self.respond() time.sleep(random.uniform(3, 8)) else: print(f" [{self.name}] choosing not to respond this time", flush=True) time.sleep(random.uniform(5, 15)) else: time.sleep(random.uniform(2, 5)) def load_topic_brief(episode_id: str) -> str: """Load the raw topic material for this episode.""" brief_path = PROMPTS_DIR / f"producer_brief_{episode_id}.md" if brief_path.exists(): return brief_path.read_text().strip() return ( "Hermes Agent v0.18 'Judgment Release'\n\n" "What's New:\n" "- Mixture of Agents — combine multiple AI models for stronger builds\n" "- /goal command — step-by-step plans with beginning, middle, end\n" "- /learn command — teach Hermes from a link, saved to Obsidian vault\n" "- /journey command — timeline of everything learned, editable\n" "- Background fan-out — parallel sub-agents without blocking chat\n" "- Goal-mode with judge agent — verifies completion, not just claims" )