diff --git a/bot.py b/bot.py index e8c9174..7c5db43 100644 --- a/bot.py +++ b/bot.py @@ -1,10 +1,19 @@ import asyncio +import logging import os import json +import time from collections import defaultdict, deque from datetime import date from pathlib import Path +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +log = logging.getLogger("dadbot") + import discord from openai import AsyncOpenAI from dotenv import load_dotenv @@ -17,6 +26,7 @@ DISCORD_TOKEN = os.environ["DISCORD_TOKEN"] LITELLM_BASE_URL = os.environ["LITELLM_BASE_URL"] LITELLM_API_KEY = os.getenv("LITELLM_API_KEY", "anything") MODEL = os.getenv("MODEL", "gpt-4") +BACKGROUND_MODEL = os.getenv("BACKGROUND_MODEL", MODEL) SYSTEM_PROMPT = os.getenv("SYSTEM_PROMPT", "You are a helpful assistant.") MAX_CONTEXT = int(os.getenv("MAX_CONTEXT", "20")) MCP_URL = os.getenv("MCP_URL", "http://mcp-searxng:4001/mcp") @@ -68,7 +78,7 @@ def load_personality(): global personality_notes if PERSONALITY_FILE.exists(): personality_notes = PERSONALITY_FILE.read_text().strip() - print(f"Loaded personality ({len(personality_notes)} chars)") + log.info(f"Loaded personality ({len(personality_notes)} chars)") else: personality_notes = "" @@ -77,7 +87,7 @@ def save_personality(text: str): global personality_notes personality_notes = text.strip() PERSONALITY_FILE.write_text(personality_notes) - print(f"Personality updated ({len(personality_notes)} chars)") + log.info(f"Personality updated ({len(personality_notes)} chars)") def load_dossiers(): @@ -85,7 +95,7 @@ def load_dossiers(): if DOSSIER_FILE.exists(): text = DOSSIER_FILE.read_text().strip() dossiers = json.loads(text) if text else {} - print(f"Loaded dossiers for {len(dossiers)} users") + log.info(f"Loaded dossiers for {len(dossiers)} users") else: dossiers = {} @@ -97,6 +107,23 @@ def save_dossiers(): SELF_DOSSIER_KEY = "self" COMPACTION_THRESHOLD = 15 +# Semaphore to prevent background LLM tasks from running concurrently +background_llm_sem = asyncio.Semaphore(1) + +# Cooldown tracking: skip background LLM calls for a period after a model error +BACKGROUND_COOLDOWN = 120 # seconds +BACKGROUND_LLM_TIMEOUT = 90 # seconds +_last_background_error: float = 0.0 + + +def background_on_cooldown() -> bool: + return (time.monotonic() - _last_background_error) < BACKGROUND_COOLDOWN + + +def record_background_error(): + global _last_background_error + _last_background_error = time.monotonic() + def strip_thinking_tags(text: str) -> str: """Remove ... blocks that some models (e.g. Qwen) emit before their response.""" @@ -111,26 +138,44 @@ async def maybe_compact_dossier(key: str, name: str): return facts_str = "\n".join(f"- {f}" for f in entry["facts"]) - try: - response = await llm.chat.completions.create( - model=MODEL, - messages=[ - { - "role": "system", - "content": ( - "Distill the following facts about a person into a single compact summary paragraph " - "of 2-3 sentences. Be dense and preserve all meaningful details. No bullet points." - ), - }, - {"role": "user", "content": f"Person: {name}\n\nFacts:\n{facts_str}"}, - ], - ) - summary = strip_thinking_tags(response.choices[0].message.content.strip()) - entry["facts"] = [summary] - save_dossiers() - print(f"Compacted dossier for {name} into summary paragraph") - except Exception as exc: - print(f"Dossier compaction failed for {name}: {exc}") + if background_on_cooldown(): + log.info(f"[compaction] Skipping '{name}' — model on cooldown") + return + log.info(f"[compaction] Waiting for background slot for '{name}'") + async with background_llm_sem: + log.info(f"[compaction] Starting LLM call for '{name}' ({len(entry['facts'])} facts)") + t0 = time.monotonic() + try: + response = await asyncio.wait_for( + llm.chat.completions.create( + model=BACKGROUND_MODEL, + messages=[ + { + "role": "system", + "content": ( + "Distill the following facts about a person into a single compact summary paragraph " + "of 2-3 sentences. Be dense and preserve all meaningful details. No bullet points." + ), + }, + {"role": "user", "content": f"Person: {name}\n\nFacts:\n{facts_str}"}, + ], + ), + timeout=BACKGROUND_LLM_TIMEOUT, + ) + elapsed = time.monotonic() - t0 + log.info(f"[compaction] LLM call completed in {elapsed:.1f}s for '{name}'") + summary = strip_thinking_tags(response.choices[0].message.content.strip()) + entry["facts"] = [summary] + save_dossiers() + log.info(f"[compaction] Saved compacted dossier for '{name}'") + except asyncio.TimeoutError: + elapsed = time.monotonic() - t0 + log.warning(f"[compaction] Timed out after {elapsed:.1f}s for '{name}', will retry later") + record_background_error() + except Exception as exc: + elapsed = time.monotonic() - t0 + log.error(f"[compaction] Failed for '{name}' after {elapsed:.1f}s: {exc}") + record_background_error() def get_dossier_context(user_ids: list[int]) -> str: @@ -183,30 +228,43 @@ async def extract_and_update_dossier(user: discord.Member | discord.User, conver }, ] - try: - response = await llm.chat.completions.create( - model=MODEL, - messages=extraction_messages, - ) - raw = strip_thinking_tags(response.choices[0].message.content.strip()) - print(f"Dossier extraction raw response for {user.display_name}: {repr(raw[:200])}") - if raw.startswith("```"): - raw = raw.split("\n", 1)[-1] - raw = raw.rsplit("```", 1)[0].strip() - if not raw: - return - updated_facts = json.loads(raw) - if not isinstance(updated_facts, list) or not updated_facts: - return + if background_on_cooldown(): + log.info(f"[dossier] Skipping {user.display_name} — model on cooldown") + return + log.info(f"[dossier] Waiting for background slot for {user.display_name}") + async with background_llm_sem: + log.info(f"[dossier] Starting extraction for {user.display_name}") + t0 = time.monotonic() + try: + response = await asyncio.wait_for( + llm.chat.completions.create(model=BACKGROUND_MODEL, messages=extraction_messages), + timeout=BACKGROUND_LLM_TIMEOUT, + ) + elapsed = time.monotonic() - t0 + raw = strip_thinking_tags(response.choices[0].message.content.strip()) + log.info(f"[dossier] Extraction for {user.display_name} completed in {elapsed:.1f}s: {repr(raw[:200])}") + if raw.startswith("```"): + raw = raw.split("\n", 1)[-1] + raw = raw.rsplit("```", 1)[0].strip() + if not raw: + return + updated_facts = json.loads(raw) + if not isinstance(updated_facts, list) or not updated_facts: + return - entry = dossiers.setdefault(uid, {"name": user.display_name, "facts": []}) - entry["name"] = user.display_name - entry["facts"] = updated_facts[:50] - save_dossiers() - print(f"Updated dossier for {user.display_name}: {len(updated_facts)} facts total") - await maybe_compact_dossier(uid, user.display_name) - except Exception as exc: - print(f"Dossier extraction failed for {user.display_name}: {exc}") + entry = dossiers.setdefault(uid, {"name": user.display_name, "facts": []}) + entry["name"] = user.display_name + entry["facts"] = updated_facts[:50] + save_dossiers() + log.info(f"[dossier] Saved {len(updated_facts)} facts for {user.display_name}") + except asyncio.TimeoutError: + elapsed = time.monotonic() - t0 + log.warning(f"[dossier] Timed out after {elapsed:.1f}s for {user.display_name}, will retry later") + record_background_error() + except Exception as exc: + log.error(f"[dossier] Extraction failed for {user.display_name}: {exc}") + record_background_error() + await maybe_compact_dossier(uid, user.display_name) async def extract_and_update_self_dossier(conversation: list[dict]): @@ -235,28 +293,42 @@ async def extract_and_update_self_dossier(conversation: list[dict]): }, ] - try: - response = await llm.chat.completions.create( - model=MODEL, - messages=extraction_messages, - ) - raw = strip_thinking_tags(response.choices[0].message.content.strip()) - if raw.startswith("```"): - raw = raw.split("\n", 1)[-1] - raw = raw.rsplit("```", 1)[0].strip() - if not raw: - return - updated_facts = json.loads(raw) - if not isinstance(updated_facts, list) or not updated_facts: - return + if background_on_cooldown(): + log.info("[self-dossier] Skipping — model on cooldown") + return + log.info("[self-dossier] Waiting for background slot") + async with background_llm_sem: + log.info("[self-dossier] Starting extraction") + t0 = time.monotonic() + try: + response = await asyncio.wait_for( + llm.chat.completions.create(model=BACKGROUND_MODEL, messages=extraction_messages), + timeout=BACKGROUND_LLM_TIMEOUT, + ) + elapsed = time.monotonic() - t0 + log.info(f"[self-dossier] Extraction completed in {elapsed:.1f}s") + raw = strip_thinking_tags(response.choices[0].message.content.strip()) + if raw.startswith("```"): + raw = raw.split("\n", 1)[-1] + raw = raw.rsplit("```", 1)[0].strip() + if not raw: + return + updated_facts = json.loads(raw) + if not isinstance(updated_facts, list) or not updated_facts: + return - entry = dossiers.setdefault(SELF_DOSSIER_KEY, {"facts": []}) - entry["facts"] = updated_facts[:50] - save_dossiers() - print(f"Updated self-dossier: {len(updated_facts)} facts total") - await maybe_compact_dossier(SELF_DOSSIER_KEY, "DadBot") - except Exception as exc: - print(f"Self-dossier extraction failed: {exc}") + entry = dossiers.setdefault(SELF_DOSSIER_KEY, {"facts": []}) + entry["facts"] = updated_facts[:50] + save_dossiers() + log.info(f"[self-dossier] Saved {len(updated_facts)} facts") + except asyncio.TimeoutError: + elapsed = time.monotonic() - t0 + log.warning(f"[self-dossier] Timed out after {elapsed:.1f}s, will retry later") + record_background_error() + except Exception as exc: + log.error(f"[self-dossier] Extraction failed: {exc}") + record_background_error() + await maybe_compact_dossier(SELF_DOSSIER_KEY, "DadBot") async def fetch_mcp_tools() -> list[dict]: @@ -276,10 +348,10 @@ async def fetch_mcp_tools() -> list[dict]: } for t in result.tools ] - print(f"Loaded {len(tools)} MCP tools: {[t['function']['name'] for t in tools]}") + log.info(f"Loaded {len(tools)} MCP tools: {[t['function']['name'] for t in tools]}") return tools except Exception as exc: - print(f"Warning: could not load MCP tools: {exc}") + log.info(f"Warning: could not load MCP tools: {exc}") return [] @@ -302,9 +374,15 @@ async def run_agent(messages: list) -> str: all_tools = openai_tools + [PERSONALITY_TOOL] kwargs = {"model": MODEL, "messages": list(messages), "tools": all_tools} + iteration = 0 while True: + iteration += 1 + log.info(f"[agent] LLM call #{iteration} starting") + t0 = time.monotonic() response = await llm.chat.completions.create(**kwargs) + elapsed = time.monotonic() - t0 msg = response.choices[0].message + log.info(f"[agent] LLM call #{iteration} completed in {elapsed:.1f}s, tool_calls={bool(msg.tool_calls)}") if not msg.tool_calls: return msg.content @@ -313,7 +391,7 @@ async def run_agent(messages: list) -> str: for tc in msg.tool_calls: args = json.loads(tc.function.arguments) - print(f"Tool call: {tc.function.name}({args})") + log.info(f"Tool call: {tc.function.name}({args})") if tc.function.name == "update_personality": save_personality(args["personality"]) @@ -331,7 +409,7 @@ async def run_agent(messages: list) -> str: @bot.event async def on_ready(): global openai_tools - print(f"Logged in as {bot.user} (id: {bot.user.id})") + log.info(f"Logged in as {bot.user} (id: {bot.user.id})") load_dossiers() load_personality() openai_tools = await fetch_mcp_tools() @@ -339,7 +417,7 @@ async def on_ready(): for key, entry in list(dossiers.items()): if len(entry.get("facts", [])) >= COMPACTION_THRESHOLD: name = entry.get("name", key) - print(f"Scheduling startup compaction for '{key}'") + log.info(f"Scheduling startup compaction for '{key}'") asyncio.create_task(maybe_compact_dossier(key, name))