Add Qwen3.5 support: logging, semaphore, cooldown, background timeouts
- Switch model to qwen3.5-35b-a3b - Add structured logging with timestamps and per-task prefixes - Add background LLM semaphore to prevent concurrent background calls - Add 120s cooldown after background task failures - Add 90s timeout on background LLM calls to prevent semaphore deadlock - Strip <think> tags from Qwen model responses - Tune prompt: more aggressive stance, no folding under pressure Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Suppai_NekoandClaude Sonnet 4.6 committed
1 parent
ca825feb95
commit
ca02574719
1 file changed
+149
-72
@@ -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
|
||||
@@ -68,7 +77,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 +86,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 +94,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 +106,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 <think>...</think> blocks that some models (e.g. Qwen) emit before their response."""
|
||||
@@ -111,26 +137,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=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 +227,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=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 +292,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=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 +347,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 +373,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 +390,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 +408,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 +416,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))
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user