""" RUBRA v4 - Agent Module Main agent class to orchestrate brain.py and tools """ import re import json import logging import math from typing import Optional # Define log at the very top — before any code that uses it try: from rubra_logging import get_logger log = get_logger("rubra.agent") except ImportError: log = logging.getLogger("rubra.agent") from brain import ( detect_lang, lang_instr, RUBRA_CORE, RUBRA_CORE_MINIMAL, TUTOR_PROMPT, VISION_PROMPT, build_msgs, llm, hermes_stream, classify_coding_task, NCTB_2026 ) try: from context_engine import PromptSection, assemble_budgeted_sections BUDGET_OK = True except ImportError: BUDGET_OK = False # Phase 1 hardening: a message this short/trivial doesn't need the full # ~750-token persona essay before any real context is even added — see # context_engine.assemble_budgeted_sections and brain.RUBRA_CORE_MINIMAL. _TRIVIAL_MSG_RE = re.compile( r'^\s*(hi|hello|hey|salam|assalam|hola|yo|ok|okay|thanks|thank you|' r'thik ache|dhonnobad|kemon acho|kemon achen|bye|good\s*(morning|night|evening))' r'[\s!.?,]*$', re.IGNORECASE) def _is_trivial_message(msg: str) -> bool: return bool(_TRIVIAL_MSG_RE.match(msg)) or len(msg.split()) <= 2 # Phase 1 hardening: default per-section token budgets for the assembler. # "identity" is sized to fit RUBRA_CORE_MINIMAL comfortably; when the full # RUBRA_CORE is used instead (non-trivial messages) it's marked required # so it's truncated rather than dropped if something else pushes it over. FAST_CHAT_TOTAL_BUDGET = 1_800 # generous vs. the ~500-tok acceptance # target for a bare greeting — trivial # messages use RUBRA_CORE_MINIMAL and # skip memory/RAG sections entirely, so # they land well under this ceiling; this # is the CEILING for non-trivial fast-chat # turns that do carry memory/RAG context. GENERAL_AGENT_TOTAL_BUDGET = 2_500 # GeneralAgent carries more optional # context (tool_ctx/RAG on top of # memory) than FastChatAgent by design try: from code_executor import execute_with_autofix, format_execution_result, detect_language EXECUTOR_OK = True except ImportError: EXECUTOR_OK = False try: from ponytail_ruleset import PONYTAIL_RULESET, should_apply_ponytail PONYTAIL_OK = True except ImportError: PONYTAIL_OK = False try: from personality import detect_emotion, detect_expertise_level, build_personality_instruction PERSONALITY_OK = True except ImportError: PERSONALITY_OK = False try: from evolution_engine import get_memory_context, after_conversation MEMORY_OK = True except ImportError: MEMORY_OK = False def get_memory_context(query, session_id): return "" def after_conversation(message, response, session_id): pass try: from video_tool import tool_render_video, is_video_request, VIDEO_RENDER_ENABLED VIDEO_TOOL_OK = True except ImportError: VIDEO_TOOL_OK = False VIDEO_RENDER_ENABLED = False def is_video_request(msg): return False _URL_RE = re.compile(r'https?://[^\s\)\]\>"\']+') try: import sys as _sys, pathlib as _pathlib _sys.path.insert(0, str(_pathlib.Path(__file__).resolve().parent / "services" / "tools")) from computer_use import engine as _cu_engine, ActionRequest as _CUActionRequest COMPUTER_USE_OK = True except ImportError: COMPUTER_USE_OK = False # ===================================================== # SPECIALIST AGENTS LIBRARY (Agency Agents — 232 personas) # ===================================================== # Bundled as data files under agents_library//.md rather than # inlined as Python string literals — 232 personas average ~15KB each, so # inlining them would make this module ~3.7MB of string constants that get # re-parsed on every cold start. Instead: a tiny in-memory INDEX (names/ # descriptions only, built once) for matching, and the full persona body is # read from disk and cached (lru_cache) only the first time it's actually used. # skipped: hot-reloading on file edits — these are static, shipped with the # repo; redeploy is the upgrade path if a persona file changes. import functools as _functools _LIBRARY_DIR = _pathlib.Path(__file__).resolve().parent / "agents_library" def _parse_frontmatter(text: str): """Tiny flat key:value frontmatter parser — no PyYAML dependency needed since every persona file uses simple `key: value` lines, no nesting.""" m = re.match(r'^---\s*\n(.*?)\n---\s*\n(.*)$', text, re.DOTALL) if not m: return None, text fm, body = {}, m.group(2) for line in m.group(1).splitlines(): line = line.strip() if not line or ':' not in line: continue k, _, v = line.partition(':') fm[k.strip()] = v.strip().strip('"').strip("'") return (fm if 'name' in fm else None), body def _build_specialist_index() -> dict: """slug -> {category, path, name, description, color, emoji, vibe}""" index = {} if not _LIBRARY_DIR.is_dir(): return index for path in _LIBRARY_DIR.rglob("*.md"): try: text = path.read_text(encoding="utf-8", errors="replace") except Exception: continue fm, _ = _parse_frontmatter(text) if not fm: continue slug = path.stem category = path.relative_to(_LIBRARY_DIR).parts[0] index[slug] = { "category": category, "path": str(path), "name": fm.get("name", slug), "description": fm.get("description", ""), "color": fm.get("color", ""), "emoji": fm.get("emoji", "🧠"), "vibe": fm.get("vibe", ""), } return index SPECIALIST_INDEX = _build_specialist_index() SPECIALISTS_OK = len(SPECIALIST_INDEX) > 0 if not SPECIALISTS_OK: log.warning(f"Specialist agents library not found/empty at {_LIBRARY_DIR}") # ── Matching: IDF-weighted keyword overlap with light stemming ───────────── # Deliberately NOT an embedding/semantic search (no new ML dependency) — a # bag-of-words match is the "stdlib-first" option here, but a *naive* one # badly misroutes (e.g. "weather today" coincidentally matching an unrelated # persona on the single word "what"). Three corrections made that workable: # 1) a much larger stopword list (wh-words, tense/aux verbs), # 2) crude suffix stripping so "researcher"/"research"/"researching" count # as the same word — without pulling in a real stemmer dependency, # 3) IDF weighting PLUS a minimum-overlap-count floor, so one coincidental # rare-word hit can't outscore real multi-word matches. _SPECIALIST_STOPWORDS = { "the","a","an","i","me","my","you","your","yours","is","are","was","were","be","been", "to","for","of","in","on","at","with","and","or","please","can","could","should", "would","need","want","help","make","get","got","this","that","these","those","it","its", "do","does","did","have","has","had","will","shall","just","also","about","into", "what","where","when","why","how","who","which","whom","today","now","currently", "new","plan","review","write","create","build","give","tell","show","let","lets", } _SPECIALIST_SUFFIXES = ("ing", "ers", "er", "ed", "es", "s") def _stem(word: str) -> str: for suf in _SPECIALIST_SUFFIXES: if len(word) > len(suf) + 3 and word.endswith(suf): return word[: -len(suf)] return word def _specialist_words(text: str) -> set: raw = re.findall(r'[a-z]{2,}', text.lower()) return {_stem(w) for w in raw if w not in _SPECIALIST_STOPWORDS} def _build_specialist_corpus(): """Precompute each persona's word-set + global IDF table, once.""" word_sets = { slug: _specialist_words(f"{e['name']} {e['description']} {e['category']}") for slug, e in SPECIALIST_INDEX.items() } n = max(len(word_sets), 1) df: dict = {} for ws in word_sets.values(): for w in ws: df[w] = df.get(w, 0) + 1 idf = {w: math.log(n / (1 + c)) for w, c in df.items()} return word_sets, idf _SPECIALIST_WORDS, _SPECIALIST_IDF = _build_specialist_corpus() @_functools.lru_cache(maxsize=40) def load_specialist_prompt(slug: str) -> str: """Read+cache a specialist's full persona body (frontmatter stripped). Bounded cache (40 of 232) — popular specialists stay warm, the rest are a single fast local disk read away; never holds all 3.7MB in memory.""" entry = SPECIALIST_INDEX.get(slug) if not entry: return "" try: text = _pathlib.Path(entry["path"]).read_text(encoding="utf-8", errors="replace") except Exception: return "" _, body = _parse_frontmatter(text) return body.strip() def list_specialist_categories() -> list: return sorted({e["category"] for e in SPECIALIST_INDEX.values()}) def list_specialists(category: str = None) -> list: items = SPECIALIST_INDEX.items() if category: items = [(s, e) for s, e in items if e["category"] == category] return [{"slug": s, **e} for s, e in items] def find_specialist(message: str, min_overlap: int = 2, min_score: float = 5.0) -> Optional[str]: """ Best-matching specialist slug, or None if nothing clears the bar. Deliberately conservative — this is an OPT-IN router, not a replacement for the main intent engine. A weak/ambiguous match should fall through to GeneralAgent rather than hijack the conversation with an oddly specific persona. Requires EITHER >= min_overlap distinct shared words OR one very high-IDF (very distinctive) single-word hit. """ if not SPECIALISTS_OK: return None qwords = _specialist_words(message) if not qwords: return None best_slug, best_score, best_overlap = None, 0.0, 0 for slug, doc_words in _SPECIALIST_WORDS.items(): overlap = qwords & doc_words if not overlap: continue score = sum(_SPECIALIST_IDF.get(w, 0) for w in overlap) entry = SPECIALIST_INDEX[slug] if entry["category"].replace("-", " ") in message.lower(): score += 1.0 if score > best_score: best_slug, best_score, best_overlap = slug, score, len(overlap) if best_slug and (best_overlap >= min_overlap or best_score >= min_score + 2): return best_slug return None class SpecialistAgent: """ Runs one of the 232 Agency Agents personas as RUBRA's system prompt for this turn. RUBRA_CORE stays layered underneath — a specialist persona adds expertise and voice, it never overrides RUBRA's actual safety/ identity baseline. """ name = "SpecialistAgent" slug = None # callers using the generic agent.run(...) dispatch can set # `instance.slug = "..."` before calling instead of passing # a kwarg the shared call site doesn't know about. async def run(self, msg, hist, sid="", lang="en", img=None, slug: str = None): li = lang_instr(lang) slug = slug or self.slug or find_specialist(msg) entry = SPECIALIST_INDEX.get(slug) if slug else None if not entry: yield {"type": "error", "message": "No matching specialist found."} return persona = load_specialist_prompt(slug) yield {"type": "status", "text": f"{entry['emoji']} Bringing in the {entry['name']} specialist..."} sys_p = RUBRA_CORE + f"\n\n=== SPECIALIST MODE: {entry['name']} ===\n" + persona if li: sys_p += f"\n\n{li}" msgs = build_msgs(sys_p, hist, msg, img) try: mode = "vision" if img else "general" async for tok in llm(msgs, mode): yield {"type": "token", "content": tok} except Exception as e: yield {"type": "error", "message": str(e)[:200]} async def fetch_urls_in_message(msg: str, max_urls: int = 2) -> str: """ Pre-fetch any URL(s) found in the user's message and return their content as a context block. This closes the gap where a "build this from this link" request never actually visited the link — agents only wrote code/replies, they never browsed, so the link sat in the prompt as inert text. Two-tier fetch: 1. Plain HTTP (browse_url) — fast, works for static HTML. 2. If that returns no readable text (JS-rendered SPA — exactly the Google Business share-link case), fall back to the real headless browser in computer_use.py, which executes JS and can actually see the rendered DOM/screenshot. Every URL still passes through link_security via browse_url/computer_use's own checks before either tier touches the network. """ urls = _URL_RE.findall(msg)[:max_urls] if not urls: return "" blocks = [] for url in urls: url = url.rstrip('.,;)') try: page = browse_url(url, max_chars=4000) except Exception as e: page = {"error": str(e)[:150], "text": "", "title": url} got_text = page.get("text", "").strip() if not page.get("error") and got_text: blocks.append(f"[FETCHED {url} — \"{page.get('title','')}\"]\n{got_text[:3500]}") continue # Tier 2: real browser — covers exactly the JS-rendered case that # broke before (RUBRA insisting "I can't fetch links" on a share link). if COMPUTER_USE_OK: try: result = await _cu_engine.goto(url) if result.success and result.dom_snapshot.strip(): blocks.append( f"[FETCHED {url} via browser — rendered page]\n" f"Interactive elements and text found:\n{result.dom_snapshot[:3500]}" ) continue elif result.blocked_reason: blocks.append( f"[FETCHED {url}]\nBlocked by security filter: {result.blocked_reason}. " f"Tell the user this link looks unsafe and ask them to confirm the " f"real destination, or paste the content directly." ) continue except Exception as e: log.warning(f"[FETCH] Browser fallback failed for {url}: {e}") # Both tiers failed — be honest, never invent placeholder content. reason = page.get("error") or "page returned no readable text (likely JavaScript-rendered)" blocks.append( f"[FETCHED {url}]\nCould not retrieve usable content: {reason}. " f"Do NOT invent placeholder details (name/phone/address/etc) — " f"tell the user this link couldn't be read and ask them to paste " f"the actual text/details instead." ) return "\n\n".join(blocks) from database import ( mem_add, mem_get, session_load, session_update, rag_search, live_feed, get_session_owner ) def feed_get(limit=5, category=None): try: items = live_feed() if category: items = [i for i in items if i.get("category") == category] return items[:limit] except: return [] from tools import ( tool_weather, tool_crypto, tool_currency, tool_wikipedia, tool_arxiv, tool_books, tool_books_2026, tool_calc, ocr_image_space, ocr_image_tesseract, browse_url, search_and_browse, browse_profile, to_base64, pdf_to_text, IMAGE_EXTS ) # ===================================================== # AGENT CLASSES # ===================================================== class GeneralAgent: name = "GeneralAgent" async def run(self, msg, hist, sid="", lang="en", img=None): li = lang_instr(lang) tool_ctx = "" rag_ctx = "" # Session memory sess = session_load(sid) or {} sess["preferred_lang"] = lang session_update(sid, "user_intent_history", {"msg": msg[:120], "ts": __import__("time").time()}) # Auto Wikipedia if not img and re.search(r"\b(what is|who is|how does|explain|define|history|overview|what are)\b", msg, re.IGNORECASE): q = re.sub(r"\b(what is|who is|how does|explain|define|tell me|about|the|a|an|please)\b", "", msg, flags=re.IGNORECASE).strip()[:60] if len(q) > 3: page = tool_wikipedia(q) if page: tool_ctx = f"[WIKIPEDIA: {page['title']}]\n{page['text']}" # Live feed if re.search(r"\b(latest|recent|trending|news|today|2025|2026|current)\b", msg, re.IGNORECASE): items = feed_get(limit=5) if items: feed_ctx = "[LIVE KNOWLEDGE]\n" + "\n".join(f"• {f['title']} ({f['source']})" for f in items[:4]) tool_ctx = feed_ctx + ("\n\n" + tool_ctx if tool_ctx else "") # URL in message → actually fetch it (with browser fallback for JS-rendered # pages). Previously missing here specifically, which is why RUBRA replied # "I can't fetch links" instead of trying — that claim is now false on every # agent path, not just CodingAgent. if _URL_RE.search(msg): yield {"type": "status", "text": "🔗 Reading linked page..."} fetched_ctx = await fetch_urls_in_message(msg) if fetched_ctx: tool_ctx = fetched_ctx + ("\n\n" + tool_ctx if tool_ctx else "") # RAG — returns list of dicts {"role","content","time"} # Phase 12 hardening: scoped to this session's owner — was # searching every user's messages globally (see database.py:: # rag_search's docstring for the full finding). hits = rag_search(msg, limit=3, user_id=get_session_owner(sid) if sid else None) if hits: rag_ctx = "\n".join( f"[memory] {h.get('content','')[:200]}" for h in hits[:2] if h.get('content') ) # Session topic context topic_ctx = "" if sess.get("topic_memory"): recent = list(sess["topic_memory"].items())[-3:] topic_ctx = "[SESSION CONTEXT]\n" + "\n".join(f"• {k}" for k, v in recent) # Mood/expertise awareness — read the user, don't just answer the words mood_ctx = "" if PERSONALITY_OK: emotion = detect_emotion(msg) level = detect_expertise_level(msg, hist) mood_ctx = build_personality_instruction(msg, hist, lang, emotion, level) # Cross-session memory — people, facts, past conversations actually persist cross_mem_ctx = "" if MEMORY_OK and sid: cross_mem_ctx = get_memory_context(msg, sid) # Build context-aware system prompt — Phase 1 hardening: budgeted # assembly instead of raw concatenation (same fix as FastChatAgent; # GeneralAgent had the identical unbounded-concatenation pattern). if BUDGET_OK: sections = [ PromptSection("identity", RUBRA_CORE, max_tokens=800, required=True), PromptSection("rules", """[CONVERSATION RULES — FOLLOW EXACTLY] 1. You have the FULL conversation history above. USE IT. 2. If user says "run this" or "run it" → they mean the code from your PREVIOUS message 3. If user says "change X" or "update it" → they mean your PREVIOUS response 4. If user asks casual question (hi, how are you, what are you doing) → casual reply ONLY 5. NEVER say you cannot run/execute code — you CAN via the execution system 6. NEVER lose track of what was discussed before 7. If user uploads file → analyze IT, remember it for the whole conversation""", max_tokens=150, required=True), ] if li: sections.append(PromptSection("language", li, max_tokens=100, required=True)) if mood_ctx: sections.append(PromptSection("mood", f"[READ THE USER]\n{mood_ctx}", max_tokens=150)) if cross_mem_ctx: sections.append(PromptSection("memory", f"[WHAT YOU REMEMBER ABOUT THIS PERSON]\n{cross_mem_ctx}", max_tokens=400)) if tool_ctx: sections.append(PromptSection("tool_ctx", tool_ctx, max_tokens=600)) if rag_ctx: sections.append(PromptSection("rag", f"[KNOWLEDGE FROM MEMORY]\n{rag_ctx}", max_tokens=300)) sys_p = assemble_budgeted_sections(sections, total_budget=GENERAL_AGENT_TOTAL_BUDGET) else: parts = [RUBRA_CORE, """[CONVERSATION RULES — FOLLOW EXACTLY] 1. You have the FULL conversation history above. USE IT."""] if li: parts.append(li) if mood_ctx: parts.append(f"[READ THE USER]\n{mood_ctx}") if cross_mem_ctx: parts.append(f"[WHAT YOU REMEMBER ABOUT THIS PERSON]\n{cross_mem_ctx}") if tool_ctx: parts.append(tool_ctx) if rag_ctx: parts.append(f"[KNOWLEDGE FROM MEMORY]\n{rag_ctx}") sys_p = "\n\n".join(parts) msgs = build_msgs(sys_p, hist, msg, img) full_raw = "" try: mode = "vision" if img else "general" async for tok in llm(msgs, mode): full_raw += tok yield {"type": "token", "content": tok} if MEMORY_OK and sid and full_raw: try: after_conversation(msg, full_raw, sid) except Exception as e: log.warning(f"after_conversation failed: {e}") except Exception as e: yield {"type": "error", "message": str(e)[:200]} class CodingAgent: name = "CodingAgent" async def _handle_video(self, msg, lang): """ Honest, single-path video handling — no contradictory replies. Either it actually renders (HyperFrames CLI via subprocess), or it says clearly and immediately that rendering is not enabled yet. Never silently writes canvas-recording JS as a substitute. """ li = lang_instr(lang) if not VIDEO_RENDER_ENABLED: msg_out = ( "Video rendering isn\u2019t turned on for this RUBRA instance yet " "(HyperFrames needs Node.js installed in the Space + a feature flag). " "I can instead write you the HTML/CSS scene for it, or describe the " "shot list / script if that helps." ) yield {"type": "status", "text": "Video rendering unavailable"} yield {"type": "token", "content": msg_out} return yield {"type": "status", "text": "🎬 Writing HTML scene for video..."} scene_prompt = ( "Write a single self-contained HTML file (inline CSS, inline JS, no external " "assets) describing a short animated scene for this request, suitable for " "rendering to MP4 by HyperFrames. Output ONLY the HTML, no explanation.\n\n" f"Request: {msg}" ) msgs = [ {"role": "system", "content": "You write minimal, self-contained HTML/CSS/JS animation scenes."}, {"role": "user", "content": scene_prompt}, ] html_scene = "" async for tok in llm(msgs, mode="coding", temperature=0.2): html_scene += tok html_scene = re.sub(r'^```\w*\n?', '', html_scene.strip()) html_scene = re.sub(r'\n?```$', '', html_scene.strip()) yield {"type": "status", "text": "🎬 Rendering to MP4 (this can take up to 2 min)..."} result = tool_render_video(html_scene) if result.get("success"): yield {"type": "status", "text": f"✅ Rendered ({result['size_bytes']:,} bytes)"} yield {"type": "tool_result", "tool": "hyperframes", "data": {"size_bytes": result["size_bytes"]}} yield {"type": "token", "content": "Video rendered successfully. Use the download action to save the MP4."} else: yield {"type": "status", "text": "❌ Render failed"} yield {"type": "token", "content": f"Rendering failed: {result.get('error','unknown error')}. " f"Here\u2019s the HTML scene instead, in case you want to adjust and retry:\n\n```html\n{html_scene[:2000]}\n```"} async def run(self, msg, hist, sid="", lang="en", img=None): # ── Video requests go through HyperFrames, not raw code generation ──── if VIDEO_TOOL_OK and is_video_request(msg): async for evt in self._handle_video(msg, lang): yield evt return task = classify_coding_task(msg) li = lang_instr(lang) # ── Pre-fetch any URL the user referenced ("build this from this link") # before any code is written — not after, not never. fetched_ctx = "" if _URL_RE.search(msg): yield {"type": "status", "text": "🔗 Reading linked page..."} fetched_ctx = await fetch_urls_in_message(msg) # Build coding prompt sys_p = "You are RUBRA's Hermes Core Engine - world-class coding intelligence.\n\n" sys_p += "=== CORE MISSION ===\n" sys_p += "Produce complete, production-ready code that works on the first run.\n" sys_p += "No truncation. No placeholders. No TODOs.\n\n" sys_p += "=== ABSOLUTE RULES ===\n" sys_p += "X NEVER: // rest of code..., # TODO, ...\n" sys_p += "X NEVER: Truncate a function, class, or file mid-way\n" sys_p += "V ALWAYS: Complete every bracket, tag, function, class\n" sys_p += "V IF LONG: Finish current block -> write: \n\n" sys_p += f"Mode: {'ARCHITECT+EDITOR' if task['needs_architect'] else 'BUILD'}\n" sys_p += "\n=== NO STALLING ===\n" sys_p += "If asked to build/make/create something, do NOT respond with a list of\n" sys_p += "clarifying questions before writing any code. Make sensible, modern\n" sys_p += "assumptions and build the complete thing now. Mention an assumption in\n" sys_p += "one short line if it matters, never block on the user answering first.\n" if fetched_ctx: sys_p += f"\n=== CONTENT FROM USER'S LINK ===\n{fetched_ctx}\n=== END LINKED CONTENT ===\n" sys_p += "Use the REAL details above (name/phone/address/etc). If the fetch " \ "failed or returned no text, say so plainly instead of inventing " \ "placeholder info like 'example.com' or '[Your Name]'.\n" # Ponytail: stdlib/native-first, no unrequested abstractions — primary model only if PONYTAIL_OK and should_apply_ponytail("coding"): sys_p += f"\n{PONYTAIL_RULESET}\n" if task["is_frontend"]: sys_p += "\n[FRONTEND RULES]\n" sys_p += "V Mobile-first responsive\n" sys_p += "V Smooth transitions 150-300ms\n" sys_p += "V WCAG AA contrast\n" sys_p += "V Lucide/Heroicons SVG only (no emoji)\n" if task["is_game"]: sys_p += "\n[GAME RULES]\n" sys_p += "V Core loop design FIRST\n" sys_p += "V No magic numbers -> constants\n" sys_p += "V Delta time on ALL movement\n" sys_p += "V State machine for entities\n" if task["is_backend"]: sys_p += "\n[BACKEND RULES]\n" sys_p += "V Type hints / TypeScript strict\n" sys_p += "V Input validation (Pydantic/Zod)\n" sys_p += "V Error handling on all async ops\n" sys_p += "V Parameterized SQL only\n" if li: sys_p += f"\n\n{li}\n(Code in English always. Explanations in user's language.)" # Smart context trim if len(sys_p) > 13000: sys_p = sys_p[:13000] + "\n[...apply remaining best practices]" # Continuation mode if msg.startswith("[RUBRA_CONTINUE]"): last = msg.replace("[RUBRA_CONTINUE]", "").strip() msg = f"Continue EXACTLY from here - no intro, no repetition:\n...{last}" if img: msg = f"[Screenshot/Image provided]\n{msg}" msgs = build_msgs(sys_p, hist, msg, img) try: if img: async for tok in llm(msgs, "vision", max_tokens=8192, temperature=0.1): yield {"type": "token", "content": tok} else: async for tok in hermes_stream(msgs): yield {"type": "token", "content": tok} except Exception as e: yield {"type": "error", "message": str(e)[:200]} class SearchAgent: name = "SearchAgent" async def run(self, msg, hist, sid="", lang="en", img=None): lower = msg.lower() li = lang_instr(lang) tool_ctx = "" if re.search(r"\b(weather|temperature|forecast|rain|cold|hot|humid|wind)\b", lower): city = "Dhaka" for c in ["dhaka", "london", "new york", "tokyo", "paris", "sydney", "dubai", "singapore"]: if c in lower: city = c.title() break m = re.search(r"weather\s+(?:in\s+)?([a-z\s]{3,20})(?:\?|$|\.|\!)", lower) if m: city = m.group(1).strip().title() w = tool_weather(city) if w: tool_ctx = f"[LIVE WEATHER - {w['city']}]\nTemp: {w['temp']}C (feels {w['feels']}C) | {w['condition']} | Humidity: {w['humidity']}% | Wind: {w['wind']} km/h" yield {"type": "tool_result", "tool": "weather", "data": w} elif re.search(r"\b(bitcoin|ethereum|btc|eth|solana|crypto|coin|binance)\b", lower): cm = {"btc": "bitcoin", "eth": "ethereum", "sol": "solana", "bnb": "binancecoin"} found = [cm.get(k, k) for k in cm if k in lower] d = tool_crypto(",".join(found or ["bitcoin", "ethereum", "solana"])) if d: lines = [f"{'UP' if i.get('usd_24h_change', 0) >= 0 else 'DOWN'} {c.capitalize()}: ${i.get('usd', 0):,.2f} ({i.get('usd_24h_change', 0):+.2f}%)" for c, i in d.items()] tool_ctx = "[LIVE CRYPTO]\n" + "\n".join(lines) yield {"type": "tool_result", "tool": "crypto", "data": d} elif re.search(r"\b(exchange rate|forex|usd to|eur to|taka|bdt|currency)\b", lower): bases = re.findall(r"\b(USD|EUR|GBP|JPY|BDT|INR|CAD|AUD|CNY)\b", msg.upper()) d = tool_currency(bases[0] if bases else "USD") if d: lines = [f"1 {d['base']} = {r} {c}" for c, r in list(d["rates"].items())[:8]] tool_ctx = f"[LIVE RATES - {d['base']}]\n" + "\n".join(lines) yield {"type": "tool_result", "tool": "currency", "data": d} elif re.search(r"\b(latest news|trending|what.{0,10}happening|current events|today)\b", lower): cat = None if re.search(r"\b(tech|ai|software)\b", lower): cat = "tech" elif re.search(r"\b(bangladesh|dhaka)\b", lower): cat = "bangladesh" items = feed_get(category=cat, limit=8) if items: lines = [f"• **{f['title']}** - _{f['source']}_" for f in items] tool_ctx = "[LIVE NEWS]\n" + "\n".join(lines) yield {"type": "tool_result", "tool": "news", "count": len(items)} elif re.search(r"\b(2025 book|2026 book|new book|latest book|recent book)\b", lower): q = re.sub(r"\b(2025|2026|new|latest|recent|book|recommend|best|read|novel)\b", "", lower).strip()[:50] books = tool_books_2026(q) if books: lines = [f"**{b['title']}** ({b.get('year', '?')}) - {', '.join(b.get('authors', b.get('author_name', []))[:2])}" for b in books] tool_ctx = "[RECENT BOOKS 2024-2026]\n" + "\n".join(lines) elif re.search(r"\b(recommend.{0,10}book|best books|reading list)\b", lower): books = tool_books(re.sub(r"\b(recommend|book|about|best|read)\b", "", lower).strip()[:50]) if books: tool_ctx = "[BOOKS]\n" + "\n".join(f"{b['title']} ({b.get('year', '?')}) - {', '.join(b['authors'][:2])}" for b in books) elif re.search(r"\b(research papers?|arxiv|academic|scientific)\b", lower): q = re.sub(r"\b(research|papers?|arxiv|find|latest)\b", "", lower).strip()[:70] papers = tool_arxiv(q or msg, n=4) if papers: tool_ctx = "[ARXIV]\n" + "\n\n".join(f"• **{p['title']}** - {', '.join(p['authors'][:2])}\n {p['summary'][:200]}..." for p in papers) else: q = re.sub(r"\b(who is|what is|tell me about|history of|the|a|an)\b", "", lower).strip()[:60] page = tool_wikipedia(q or msg) if page: tool_ctx = f"[WIKIPEDIA: {page['title']}]\n{page['text']}" yield {"type": "tool_result", "tool": "wikipedia", "title": page["title"]} parts = [RUBRA_CORE, "\n[LIVE SEARCH MODE] Answer directly using retrieved data."] if li: parts.append(li) if tool_ctx: parts.append(tool_ctx) msgs = build_msgs("\n\n".join(parts), hist, msg) try: async for tok in llm(msgs, "general"): yield {"type": "token", "content": tok} except Exception as e: yield {"type": "error", "message": str(e)[:200]} class SmartTutorAgent: name = "SmartTutorAgent" async def run(self, msg, hist, sid="", lang="en", img=None): li = lang_instr(lang) sys_p = TUTOR_PROMPT if li: sys_p += f"\n\n{li}" # Inject relevant NCTB content for class_key, data in NCTB_2026.items(): if any(kw in msg.lower() for kw in [class_key.lower().replace("_", " "), class_key.lower()]): sys_p += f"\n\n[NCTB 2026 - {class_key}]\n{json.dumps(data, ensure_ascii=False, indent=2)[:1000]}" break hits = rag_search(msg, limit=2, user_id=get_session_owner(sid) if sid else None) if hits: sys_p += "\n\n[STUDY MATERIAL]\n" + "\n".join(f"[{s}:{t}]\n{c}" for _, t, c, s in hits) msgs = build_msgs(sys_p, hist, msg, img) try: mode = "vision" if img else "general" async for tok in llm(msgs, mode): yield {"type": "token", "content": tok} except Exception as e: yield {"type": "error", "message": str(e)[:200]} class VisionAgent: """Multi-model vision: OCR.space -> Groq Vision -> Qwen2.5-VL""" name = "VisionAgent" async def run(self, msg, hist, sid="", lang="en", img=None): li = lang_instr(lang) sys_p = VISION_PROMPT if li: sys_p += f"\n\n{li}" if not img: yield {"type": "error", "message": "No image provided"} return msgs = build_msgs(sys_p, hist, msg, img) try: async for tok in llm(msgs, "vision"): yield {"type": "token", "content": tok} except Exception as e: # Fallback: OCR log.warning(f"Vision LLM failed: {e}, falling back to OCR") import base64 img_bytes = base64.b64decode(img["data"]) ocr_text = ocr_image_space(img_bytes, img["mime"]) if not ocr_text: ocr_text = ocr_image_tesseract(img_bytes) if ocr_text: fallback_sys = VISION_PROMPT + f"\n\n[OCR EXTRACTED TEXT FROM IMAGE]\n{ocr_text}\n[/OCR]" if li: fallback_sys += f"\n\n{li}" msgs2 = build_msgs(fallback_sys, hist, msg or "Analyze this content") try: async for tok in llm(msgs2, "general"): yield {"type": "token", "content": tok} except Exception as e2: yield {"type": "error", "message": str(e2)[:200]} else: yield {"type": "error", "message": "Could not process image. Try a clearer photo."} class FileAgent: name = "FileAgent" def _read(self, fp): ext = fp.suffix.lower() try: if ext == ".pdf": return pdf_to_text(fp) if ext in (".xlsx", ".xls"): try: import openpyxl wb = openpyxl.load_workbook(fp, read_only=True, data_only=True) out = [] for name in wb.sheetnames[:4]: ws = wb[name] out.append(f"## Sheet: {name}") for row in ws.iter_rows(max_row=80, values_only=True): out.append(" | ".join(str(c) if c is not None else "" for c in row)) return "\n".join(out)[:12000] except ImportError: return "[openpyxl not installed]" if ext == ".csv": import csv with open(fp, "r", encoding="utf-8", errors="replace") as f: return "\n".join(", ".join(r) for r in csv.reader(f))[:10000] if ext in (".docx", ".doc"): try: import docx doc = docx.Document(str(fp)) return "\n".join(p.text for p in doc.paragraphs if p.text.strip())[:14000] except ImportError: return "[python-docx not installed]" return fp.read_text(encoding="utf-8", errors="replace")[:14000] except Exception as e: return f"[Read error: {e}]" async def analyze(self, fp, fname, question, sid="", lang="en", img_data=None): li = lang_instr(lang) ext = fp.suffix.lower() sys_p = RUBRA_CORE + "\n\n[FILE ANALYSIS] Extract insights. Answer question clearly. Use structure." if li: sys_p += f"\n\n{li}" if ext in IMAGE_EXTS or img_data: b64, mime = to_base64(fp) if not img_data else (img_data["data"], img_data["mime"]) img_d = {"data": b64, "mime": mime} vis = VisionAgent() async for evt in vis.run(question or f"Analyze: {fname}", [], sid, lang, img_d): yield evt return content = self._read(fp) sys_p += f"\n\n[FILE: {fname}]\n{content}\n[/FILE]" msgs = build_msgs(sys_p, [], question or f"Analyze: {fname}") try: async for tok in llm(msgs, "general"): yield {"type": "token", "content": tok} except Exception as e: yield {"type": "error", "message": str(e)[:200]} async def run(self, msg, hist, sid="", lang="en", img=None): yield {"type": "error", "message": "Use /api/upload for file analysis"} class TextAgent: """ Dedicated writing and text-editing agent. Handles: essays, blog posts, emails, cover letters, resumes, social media posts, proofreading, grammar fixing, paraphrasing, summarisation, tone-shifting, translation, and any task whose primary output is polished written text (not code, not factual research). """ name = "TextAgent" _TASK_MAP = { "proofread": "Fix grammar, spelling, punctuation, and clarity. Preserve the author's voice. Return only the corrected text.", "grammar": "Correct all grammar and punctuation errors. Keep the original meaning and style intact.", "rewrite": "Rewrite the text to be clearer, more concise, and more engaging while keeping the core meaning.", "paraphrase": "Rewrite using different words and sentence structures without changing the meaning.", "shorten": "Shorten the text significantly. Keep only the essential information.", "expand": "Expand the text with more detail, examples, and explanation. Keep the style consistent.", "formal": "Rewrite in a formal, professional tone.", "casual": "Rewrite in a casual, friendly, conversational tone.", "professional": "Rewrite in a polished, business-professional tone suitable for workplace communication.", "summarize": "Write a concise summary capturing the key points only.", "tldr": "Write a very short TL;DR — 1-3 sentences maximum.", "bullet": "Summarise as a clean bullet-point list of the key points.", "story": "Write an engaging, well-structured story with vivid detail and a clear narrative arc.", "poem": "Write an evocative poem matching the mood and theme requested.", "essay": "Write a well-structured essay with a clear thesis, supporting arguments, and a conclusion.", "blog": "Write a compelling blog post with a hook opening, clear sections, and a call to action.", "email": "Write a clear, professional email with subject line, greeting, body, and sign-off.", "cover_letter": "Write a strong cover letter tailored to the role described. Highlight relevant skills and enthusiasm.", "resume": "Write or improve a resume section. Be achievement-focused, use strong action verbs, quantify where possible.", "social": "Write an engaging social media post. Keep it punchy, human, and platform-appropriate.", "caption": "Write a short, catchy caption.", "headline": "Write 5 strong headline options. Each should be concise, compelling, and accurate.", "translate": "Translate accurately, preserving tone and nuance.", "write": "Write exactly what was requested. Match the requested format, tone, and length.", } def _detect_task(self, msg: str) -> str: m = msg.lower() if any(w in m for w in ("proofread", "proof read", "check my writing", "fix my writing")): return "proofread" if any(w in m for w in ("grammar", "spelling", "punctuation", "typo")): return "grammar" # tone/length checks BEFORE rewrite so "rewrite more formally" → formal, not rewrite if any(w in m for w in ("shorten", "make it shorter", "make this shorter", "condense", "trim")): return "shorten" if any(w in m for w in ("expand", "make it longer", "add more detail", "elaborate", "flesh out")): return "expand" if any(w in m for w in ("formal", "more professional", "professional tone", "business tone", "more formally")): return "formal" if any(w in m for w in ("casual", "informal", "friendly tone", "conversational", "more casual")): return "casual" if any(w in m for w in ("rewrite", "rephrase", "reword")): return "rewrite" if "paraphrase" in m: return "paraphrase" if any(w in m for w in ("summarize", "summarise", "summary")): return "summarize" if "tl;dr" in m or "tldr" in m or "in one sentence" in m: return "tldr" if "bullet" in m and any(w in m for w in ("summarize", "list", "points")): return "bullet" if "story" in m or "fiction" in m: return "story" if any(w in m for w in ("poem", "poetry", "haiku", "sonnet")): return "poem" if "essay" in m: return "essay" if "blog" in m: return "blog" if any(w in m for w in ("email", "e-mail")): return "email" if "cover letter" in m: return "cover_letter" if any(w in m for w in ("resume", " cv ", "curriculum vitae")): return "resume" if any(w in m for w in ("tweet", "instagram", "linkedin post", "social media", "facebook post")): return "social" if "caption" in m: return "caption" if "headline" in m: return "headline" if "translat" in m: return "translate" return "write" async def run(self, msg, hist, sid="", lang="en", img=None): li = lang_instr(lang) task = self._detect_task(msg) task_instr = self._TASK_MAP.get(task, self._TASK_MAP["write"]) sys_p = RUBRA_CORE + f""" [TEXT & WRITING AGENT] You are a senior editor and professional writer. Your job for this request: {task_instr} RULES: 1. Deliver the final written output directly — no lengthy preamble. 2. Match the tone, register, and format the user asks for. 3. For editing tasks: show the corrected/improved version; briefly note major changes at the end only if helpful. 4. For creative tasks: prioritize quality, voice, and engagement. 5. For summarisation: be ruthlessly concise. Don't pad. 6. Write in the user's language unless translation is requested. 7. Do NOT add unsolicited feedback or lecture the user about their content. """ if PERSONALITY_OK: try: emotion = detect_emotion(msg) level = detect_expertise_level(msg, hist) mood_instr = build_personality_instruction(msg, hist, lang, emotion, level) if mood_instr: sys_p += f"\n\n[USER CONTEXT]\n{mood_instr}" except Exception: pass if MEMORY_OK and sid: try: mem_ctx = get_memory_context(msg, sid) if mem_ctx: sys_p += f"\n\n[WHAT YOU KNOW ABOUT THIS PERSON]\n{mem_ctx}" except Exception: pass if li: sys_p += f"\n\n{li}" msgs = build_msgs(sys_p, hist, msg) full_reply = "" try: async for tok in llm(msgs, "general"): full_reply += tok yield {"type": "token", "content": tok} except Exception as e: yield {"type": "error", "message": str(e)[:200]} return if MEMORY_OK and sid and full_reply: try: after_conversation(msg, full_reply, sid) except Exception: pass class FastChatAgent: name = "FastChatAgent" async def run(self, msg, hist, sid="", lang="en", img=None): li = lang_instr(lang) trivial = _is_trivial_message(msg) if BUDGET_OK: sections = [ PromptSection("identity", RUBRA_CORE_MINIMAL if trivial else RUBRA_CORE, max_tokens=200 if trivial else 800, required=True), PromptSection("rules", """[FAST CHAT RULES] 1. Casual message → casual reply. Simple. Human. 2. "run this/it" → they mean previous code → say you'll run it and trigger execution 3. "make it better" / "change X" → they mean previous response 4. Use conversation history — never forget what was said before 5. Short friendly answers for small talk. Detailed for real questions.""", max_tokens=150, required=True), PromptSection("language", li, max_tokens=100, required=True), ] # Trivial messages skip mood/memory/RAG/URL context entirely — # there's nothing in "hi" for any of that to meaningfully act # on, and skipping it is what keeps the assembled prompt small # rather than relying on the budget cap to trim it after the # fact. Non-trivial short messages still get full context. if not trivial: if PERSONALITY_OK: emotion = detect_emotion(msg) level = detect_expertise_level(msg, hist) mood_instr = build_personality_instruction(msg, hist, lang, emotion, level) if mood_instr: sections.append(PromptSection("mood", f"[READ THE USER]\n{mood_instr}", max_tokens=150)) if MEMORY_OK and sid: mem_ctx = get_memory_context(msg, sid) if mem_ctx: sections.append(PromptSection("memory", f"[WHAT YOU REMEMBER ABOUT THIS PERSON]\n{mem_ctx}", max_tokens=400)) if _URL_RE.search(msg): yield {"type": "status", "text": "🔗 Reading linked page..."} fetched_ctx = await fetch_urls_in_message(msg) if fetched_ctx: sections.append(PromptSection("url", f"[CONTENT FROM USER'S LINK]\n{fetched_ctx}", max_tokens=600)) sys_p = assemble_budgeted_sections(sections, total_budget=FAST_CHAT_TOTAL_BUDGET) else: # Fallback if context_engine isn't importable for some reason — # old unbudgeted behavior, better than crashing the agent. sys_p = RUBRA_CORE_MINIMAL if trivial else RUBRA_CORE sys_p += "\n\n[FAST CHAT RULES]\n1. Casual message → casual reply." if li: sys_p += f"\n\n{li}" msgs = build_msgs(sys_p, hist, msg) full_reply = "" try: async for tok in llm(msgs, "fast"): full_reply += tok yield {"type": "token", "content": tok} except Exception as e: yield {"type": "error", "message": str(e)[:200]} return if MEMORY_OK and sid and full_reply: try: after_conversation(msg, full_reply, sid) except Exception as e: log.warning(f"after_conversation failed: {e}") class BrowseAgent: name = "BrowseAgent" async def run(self, msg, hist, sid="", lang="en", img=None): li = lang_instr(lang) lower = msg.lower() browse_context = "" yield {"type": "tool_result", "tool": "browse", "status": "searching"} try: # Direct URL browse url_match = re.search(r'https?://[^\s\)\]\>"\']+', msg) if url_match: url = url_match.group(0).rstrip('.,;)') page = browse_url(url, max_chars=7000) if page.get('error'): browse_context = f"[BROWSE ERROR: {page['error']}]" else: browse_context = f"[PAGE: {page['title']}]\n[URL: {page['url']}]\n\n{page['text'][:6000]}" # Professional profile lookup elif re.search(r'\b(who is|profile of|find.*profile|contact.*of|linkedin|github profile|email of|phone of|details about|professional info)\b', lower): name_match = re.search(r'(?:who is|profile of|about|contact of|details about|find)\s+(.{3,50}?)(?:\?|$|\.)', msg, re.IGNORECASE) target = name_match.group(1).strip() if name_match else msg[:60] profile = browse_profile(target) if profile['pages']: browse_context = f"[PROFILE RESEARCH: {profile['target']}]\n[SOURCES: {', '.join(profile['sources'][:3])}]\n\n" + "\n\n---\n\n".join(profile['pages'][:3]) else: browse_context = f"[No profile data found for: {target}]" # General web search else: query = re.sub(r'\b(browse|search|find|look up|check|get|fetch|what is|tell me about|latest|recent|current|today)\b', '', msg, flags=re.IGNORECASE).strip() if len(query) < 4: query = msg.strip() pages = search_and_browse(query, n_results=3) if pages: parts = [] for i, p in enumerate(pages, 1): parts.append(f"[SOURCE {i}: {p.get('title', '')}]\n[URL: {p.get('url', '')}]\n\n{p.get('text', '')[:2500]}") browse_context = "\n\n---\n\n".join(parts) else: browse_context = "[No browseable results found]" except Exception as e: browse_context = f"[Browse error: {str(e)[:120]}]" log.warning(f"BrowseAgent: {e}") sys_p = RUBRA_CORE + "\n\n[WEB BROWSER MODE - Real-time browsing results below]\n\n" sys_p += "You have just browsed the web. Present the findings as clean, structured Markdown.\n" sys_p += "FORMAT RULES:\n" sys_p += "- Use ## headers for sections\n" sys_p += "- Use **bold** for names, titles, key facts\n" sys_p += "- Use bullet lists for contact info, skills, recent posts\n" sys_p += "- Include source URLs as [Source](url) links\n" sys_p += "- Be factual - only state what was found\n" sys_p += "- NEVER fabricate contact details or credentials\n\n" if li: sys_p += f"{li}\n\n" sys_p += f"[BROWSED CONTENT]\n{browse_context}\n[/BROWSED CONTENT]" msgs = build_msgs(sys_p, hist, msg) try: async for tok in llm(msgs, "general"): yield {"type": "token", "content": tok} except Exception as e: yield {"type": "error", "message": str(e)[:200]}