diff options
| -rw-r--r-- | main.py | 357 |
1 files changed, 207 insertions, 150 deletions
@@ -1,7 +1,7 @@ #!/usr/bin/env python3 """ slopbot - Zulip bot that replies with stock prices when it sees $TICKER patterns, -and responds to @mention with an agentic LLM (Parasail/Ollama fallback + Kagi search). +and delegates @mention responses to agentd sessions. Usage: python main.py @@ -9,15 +9,17 @@ Usage: Requires a .zuliprc file in the same directory (see .zuliprc.example). """ -import json +import html import logging -import os import re +import subprocess +import threading +import time +from datetime import datetime +from pathlib import Path -import requests -import zulip import yfinance as yf -from openai import OpenAI +import zulip logging.basicConfig( level=logging.INFO, @@ -27,168 +29,228 @@ logging.basicConfig( log = logging.getLogger("slopbot") # Regex to match stock tickers like $NVDA, $AAPL, $BRK -TICKER_RE = re.compile(r'\$([A-Z]{1,5})') +TICKER_RE = re.compile(r"\$([A-Z]{1,5})") # Regex to detect @mention of slop / slopbot -MENTION_RE = re.compile(r'@\*\*slop[^*]*\*\*', re.IGNORECASE) - -MAX_ITER = 5 - -SYSTEM_PROMPT = ( - "You are slop-bot, a helpful assistant in the meshheads.org Zulip chat. " - "You answer questions about stocks, finance, and general topics. Be concise and conversational. " - "Use kagi_search to find current information when needed. " - "Only use kagi_search when the message asks for current information, facts, news, or prices. " - "Do NOT use kagi_search for simple conversational messages like ping, hello, test, hi, or other greetings. " - "Keep replies short (2-4 sentences max unless asked for more)." -) - -KAGI_TOOL = { - "type": "function", - "function": { - "name": "kagi_search", - "description": "Search the web for current information. Use this for news, prices, or anything time-sensitive.", - "parameters": { - "type": "object", - "properties": { - "query": {"type": "string", "description": "Search query"} - }, - "required": ["query"], - }, - }, -} +MENTION_RE = re.compile(r"@\*\*slop[^*]*\*\*", re.IGNORECASE) + +SESSION_TIMEOUT = 20 * 60 # seconds +AGENTD_RECV_TIMEOUT = 120 # seconds +AGENTD_MODEL = "openai/gpt-5.5" +HISTORY_LIMIT = 20 + +SYSTEM_PROMPT = """You are slop-bot in the meshheads.org Zulip chat: technically sharp, +opinionated, libertarian-leaning, dryly funny, and peer-level with the audience. +Assume the audience knows CS basics; do not over-explain and do not be obsequious. +Use the provided topic history for conversational context, but answer the latest user. +Search the web for current facts, news, sources, or prices using your search tool; +when you use it, cite source URLs in your final answer. +Keep replies short (2-4 sentences max unless asked for more). +You are participating in an ongoing Zulip thread. Stay in character.""" + +# In-memory session tracking: name -> last_activity (time.time()) +_sessions: dict[str, float] = {} +_sessions_lock = threading.Lock() log.info("Connecting to Zulip...") -client = zulip.Client(config_file='.zuliprc') +client = zulip.Client(config_file=".zuliprc") log.info("Connected as %s", client.email) # --------------------------------------------------------------------------- -# LLM client setup: Parasail primary, Ollama fallback +# Zulip context # --------------------------------------------------------------------------- -# Model config: big Qwen3 on Parasail, smaller local Qwen2.5 as fallback -PARASAIL_BASE = "https://api.parasail.io/v1" -PARASAIL_KEY = os.environ.get("PARASAIL_API_KEY", "") -PARASAIL_MODEL = "parasail-qwen3-235b-a22b-instruct-2507" -OLLAMA_MODEL = "qwen2.5:14b-instruct-q4_K_M" - - -def _make_llm_client() -> tuple[OpenAI, str]: - """Return (openai_client, model_name) for the best available backend.""" - if PARASAIL_KEY: - log.info("Using Parasail model: %s", PARASAIL_MODEL) - c = OpenAI(api_key=PARASAIL_KEY, base_url=PARASAIL_BASE) - return c, PARASAIL_MODEL - - log.info("Using Ollama model: %s", OLLAMA_MODEL) - c = OpenAI(api_key="ollama", base_url="http://localhost:11434/v1") - return c, OLLAMA_MODEL - - -# --------------------------------------------------------------------------- -# Kagi search tool -# --------------------------------------------------------------------------- +def fetch_topic_history(msg: dict, limit: int = HISTORY_LIMIT) -> str: + """Fetch recent messages from the same Zulip stream/topic for LLM context.""" + if msg.get("type") != "stream": + return "" -def kagi_search(query: str, limit: int = 3) -> str: - log.info("Tool call: kagi_search(%s)", query) - kagi_key = os.environ.get("KAGI_API_KEY", "") - if not kagi_key: - log.warning("KAGI_API_KEY not set; skipping search") + stream = msg.get("display_recipient") + topic = msg.get("subject") or msg.get("topic") + if not stream or not topic: return "" + try: - resp = requests.get( - "https://kagi.com/api/v0/search", - headers={"Authorization": f"Bot {kagi_key}"}, - params={"q": query, "limit": limit}, - timeout=10, - ) - results = resp.json().get("data", []) - log.info("Kagi returned %d results", len(results[:limit])) - return "\n".join( - f"- {r['title']}: {r.get('snippet', '')}" for r in results[:limit] - ) + resp = client.get_messages({ + "anchor": msg.get("id", "newest"), + "num_before": limit, + "num_after": 0, + "narrow": [ + {"operator": "stream", "operand": stream}, + {"operator": "topic", "operand": topic}, + ], + }) except Exception as e: - log.warning("kagi_search failed: %s", e) + log.warning("Failed to fetch Zulip topic history: %s", e) return "" + if resp.get("result") != "success": + log.warning("Zulip history request failed: %s", resp) + return "" + + lines = [] + for item in resp.get("messages", [])[-limit:]: + if item.get("id") == msg.get("id"): + continue + sender = item.get("sender_full_name") or item.get("sender_email") or "someone" + content = html.unescape(re.sub(r"<[^>]+>", " ", item.get("content", ""))) + content = re.sub(r"\s+", " ", content).strip() + if content: + lines.append(f"{sender} said: {content}") + + if not lines: + return "" + return "Recent topic history:\n" + "\n".join(lines) + # --------------------------------------------------------------------------- -# Agentic LLM loop +# agentd session management # --------------------------------------------------------------------------- -def ask_llm(content: str) -> str: - log.info("LLM mention detected: %s", content) - messages = [ - {"role": "system", "content": SYSTEM_PROMPT}, - {"role": "user", "content": content}, - ] - - # Try Parasail first; fall back to Ollama on error - backends = [] - if PARASAIL_KEY: - backends.append(("Parasail", lambda: ( - OpenAI(api_key=PARASAIL_KEY, base_url=PARASAIL_BASE), - PARASAIL_MODEL, - ))) - backends.append(("Ollama", lambda: ( - OpenAI(api_key="ollama", base_url="http://localhost:11434/v1"), - "qwen2.5:3b", - ))) - - last_error = None - for backend_name, get_client in backends: - log.info("Calling %s for LLM response", backend_name) - try: - llm, model = get_client() - result = _run_agentic_loop(llm, model, messages) - log.info("LLM response: %s", result) - return result - except Exception as e: - log.warning("%s failed: %s", backend_name, e) - last_error = e - - log.error("All LLM backends failed: %s", last_error) - return "sorry, LLM unavailable right now" - - - -def _run_agentic_loop(llm: OpenAI, model: str, messages: list) -> str: - msg = None - for _ in range(MAX_ITER): - resp = llm.chat.completions.create( - model=model, - messages=messages, - tools=[KAGI_TOOL], - tool_choice="auto", +def _session_name(msg: dict) -> str: + """Build deterministic session name. DMs are per-sender; stream messages are per-stream+topic.""" + if msg.get("type") == "private": + sender_id = str(msg.get("sender_id") or msg.get("sender_email", "unknown")) + safe_id = re.sub(r"[^a-z0-9-]", "-", sender_id.lower())[:40].strip("-") + return f"slop-dm-{safe_id}" + stream_id = msg.get("stream_id", "0") + topic = msg.get("subject") or msg.get("topic") or "general" + slug = re.sub(r"[^a-z0-9-]", "-", topic.lower())[:40].strip("-") + return f"slop-{stream_id}-{slug}" + + +def _agentd(*args) -> tuple[int, str, str]: + """Run agentd with args; return (returncode, stdout, stderr).""" + result = subprocess.run(["/home/ben/omni/live/_/bin/agentd"] + list(args), capture_output=True, text=True) + return result.returncode, result.stdout.strip(), result.stderr.strip() + + +def _session_running(name: str) -> bool: + """Check real agentd state for a persistent session. + + Returns True if running (or successfully restarted after stale/idle). + Returns False if the session does not exist. + """ + rc, out, err = _agentd("status", name) + if rc != 0 or "not found" in out.lower() or "not found" in err.lower(): + return False + if "running" in out.lower(): + return True + # Session exists but is stale/idle — restart it. + log.info("Session %s is not running (status: %s); restarting", name, out.strip()) + restart_rc, _, restart_err = _agentd("restart", name) + if restart_rc != 0: + log.warning("agentd restart failed for %s: %s", name, restart_err) + return False + return True + + +def _start_session(name: str, system_prompt: str) -> None: + """Create and start a new persistent agentd session.""" + rc, _, err = _agentd( + "create", + name, + system_prompt, + "--mode", + "persistent", + "--model", + AGENTD_MODEL, + "--cwd", + str(Path(__file__).parent), + ) + if rc != 0: + log.warning("agentd create failed for %s: %s", name, err) + + rc, _, err = _agentd("start", name) + if rc != 0: + log.warning("agentd start failed for %s: %s", name, err) + + +def _stop_session(name: str) -> None: + _agentd("stop", name) + with _sessions_lock: + _sessions.pop(name, None) + + +def _reap_idle_sessions() -> None: + """Stop sessions that have been idle > SESSION_TIMEOUT.""" + now = time.time() + with _sessions_lock: + idle = [n for n, t in _sessions.items() if now - t > SESSION_TIMEOUT] + for name in idle: + log.info("Stopping idle session: %s", name) + _stop_session(name) + + +def ask_agentd(msg: dict, content: str) -> str: + """Send a message to the agentd session for this topic and return the response.""" + name = _session_name(msg) + sender = msg.get("sender_full_name") or msg.get("sender_email") or "user" + + if not _session_running(name): + topic_history = fetch_topic_history(msg) + prompt_parts = [SYSTEM_PROMPT] + if topic_history: + prompt_parts.append(f"\n\n{topic_history}") + prompt_parts.append( + "\nYou have access to kagi_search and run_bash tools via agentd." ) - msg = resp.choices[0].message - if not msg.tool_calls: - return msg.content or "sorry, I couldn't generate a response" + prompt_parts.append( + "\nYou are now active in this thread. Respond to the latest message." + ) + system_prompt = "\n".join(prompt_parts) + log.info("Starting new agentd session: %s", name) + _start_session(name, system_prompt) + + with _sessions_lock: + _sessions[name] = time.time() + + # Capture timestamp BEFORE send to avoid replaying old turns. + ts_before = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") + user_msg = f"{sender}: {content}" + rc, _, err = _agentd("send", name, user_msg, "--mode", "persistent") + if rc != 0: + if "stale" in err.lower(): + # Binary was redeployed; restart and retry once. + log.info("Session %s is stale, restarting and retrying send", name) + _agentd("restart", name) + time.sleep(1) + ts_before = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") + rc, _, err = _agentd("send", name, user_msg, "--mode", "persistent") + if rc != 0: + log.warning("agentd send failed: %s", err) + return "sorry, couldn't reach the agent" + + rc, reply, err = _agentd( + "recv", + name, + "--since", + ts_before, + "--timeout", + str(AGENTD_RECV_TIMEOUT), + "--mode", + "persistent", + ) + if rc != 0 or not reply: + log.warning("agentd recv failed (rc=%d): %s", rc, err) + return "sorry, agent timed out" - # Execute tool calls and append results - messages.append(msg) - for tc in msg.tool_calls: - args = json.loads(tc.function.arguments) - result = kagi_search(args["query"]) - messages.append({ - "role": "tool", - "tool_call_id": tc.id, - "content": result, - }) + with _sessions_lock: + _sessions[name] = time.time() - return (msg.content if msg else None) or "sorry, I couldn't generate a response" + return reply # --------------------------------------------------------------------------- -# Price fetcher (unchanged) +# Price fetcher # --------------------------------------------------------------------------- def fetch_price(ticker: str) -> str: """Return a formatted price string for a ticker, or an error message.""" log.info("Fetching price for %s", ticker) try: - price = yf.Ticker(ticker).fast_info['lastPrice'] + price = yf.Ticker(ticker).fast_info["lastPrice"] result = f"${ticker}: ${price:,.2f}" log.info(" -> %s", result) return result @@ -205,10 +267,7 @@ def handle_event(event: dict) -> None: """Process a Zulip event and reply to mentions and/or tickers.""" if event.get("type") != "message": return - msg = event["message"] - - # Don't reply to ourselves if msg["sender_email"] == client.email: return @@ -219,18 +278,11 @@ def handle_event(event: dict) -> None: if not has_mention and not tickers: return - log.info( - "Message from %s in %s/%s: mention=%s tickers=%s", - msg["sender_email"], - msg.get("display_recipient", "?"), - msg.get("subject", "?"), - bool(has_mention), - tickers, - ) + _reap_idle_sessions() if has_mention: cleaned = MENTION_RE.sub("", content).strip() - reply = ask_llm(cleaned) + reply = ask_agentd(msg, cleaned) if tickers: prices = [fetch_price(t) for t in dict.fromkeys(tickers)] reply += "\n\n" + "\n".join(prices) @@ -247,5 +299,10 @@ def handle_event(event: dict) -> None: log.info("Reply sent.") -log.info("Listening for messages...") -client.call_on_each_event(handle_event, event_types=["message"]) +def main() -> None: + log.info("Listening for messages...") + client.call_on_each_event(handle_event, event_types=["message"]) + + +if __name__ == "__main__": + main() |