import os import json import re import logging import time from model_logic import call_model_stream, MODELS_BY_PROVIDER, get_default_model_display_name_for_provider from memory_logic import retrieve_memories_semantic, retrieve_rules_semantic from tools.websearch import search_and_scrape_duckduckgo, scrape_url import prompts from utils import format_insights_for_prompt logger = logging.getLogger(__name__) WEB_SEARCH_ENABLED = os.getenv("WEB_SEARCH_ENABLED", "true").lower() == "true" TOOL_DECISION_PROVIDER = os.getenv("TOOL_DECISION_PROVIDER", "groq") TOOL_DECISION_MODEL_ID = os.getenv("TOOL_DECISION_MODEL", "llama3-8b-8192") MAX_HISTORY_TURNS = int(os.getenv("MAX_HISTORY_TURNS", 7)) def decide_on_tool(user_input: str, chat_history_for_prompt: list, initial_insights_ctx_str: str): user_input_lower = user_input.lower() if "http://" in user_input or "https://" in user_input: url_match = re.search(r'(https?://[^\s]+)', user_input) if url_match: return "scrape_url_and_report", {"url": url_match.group(1)} if len(user_input.split()) <= 3 and any(kw in user_input_lower for kw in ["hello", "hi", "thanks", "ok", "bye"]) and not "?" in user_input: return "quick_respond", {} if len(user_input.split()) > 3 or "?" in user_input or any(w in user_input_lower for w in ["what is", "how to", "explain", "search for"]): history_snippet = "\n".join([f"{msg['role']}: {msg['content'][:100]}" for msg in chat_history_for_prompt[-2:]]) guideline_snippet = initial_insights_ctx_str[:200].replace('\n', ' ') tool_user_prompt = prompts.get_tool_user_prompt(user_input, history_snippet, guideline_snippet) tool_decision_messages = [{"role":"system", "content": prompts.TOOL_SYSTEM_PROMPT}, {"role":"user", "content": tool_user_prompt}] tool_model_display = next((dn for dn, mid in MODELS_BY_PROVIDER.get(TOOL_DECISION_PROVIDER.lower(), {}).get("models", {}).items() if mid == TOOL_DECISION_MODEL_ID), None) if not tool_model_display: tool_model_display = get_default_model_display_name_for_provider(TOOL_DECISION_PROVIDER) if tool_model_display: try: tool_resp_raw = "".join(list(call_model_stream(provider=TOOL_DECISION_PROVIDER, model_display_name=tool_model_display, messages=tool_decision_messages, temperature=0.0, max_tokens=150))) json_match_tool = re.search(r"\{.*\}", tool_resp_raw, re.DOTALL) if json_match_tool: action_data = json.loads(json_match_tool.group(0)) action_type = action_data.get("action", "quick_respond") action_input = action_data.get("action_input", {}) if not isinstance(action_input, dict): action_input = {} return action_type, action_input except Exception as e: logger.error(f"Tool decision LLM error: {e}") return "quick_respond", {} def orchestrate_and_respond(user_input: str, provider_name: str, model_display_name: str, chat_history_for_prompt: list[dict], custom_system_prompt: str = None, ui_api_key_override: str = None): process_start_time = time.time() request_id = os.urandom(4).hex() logger.info(f"ORCHESTRATOR [{request_id}] Start. User: '{user_input[:50]}...'") history_str_for_prompt = "\n".join([f"{('User' if t['role'] == 'user' else 'AI')}: {t['content']}" for t in chat_history_for_prompt[-(MAX_HISTORY_TURNS * 2):]]) yield "status", "[Checking guidelines...]" initial_insights = retrieve_rules_semantic(f"{user_input}\n{history_str_for_prompt}", k=5) initial_insights_ctx_str, parsed_initial_insights_list = format_insights_for_prompt(initial_insights) yield "status", "[Choosing best approach...]" action_type, action_input_dict = decide_on_tool(user_input, chat_history_for_prompt, initial_insights_ctx_str) logger.info(f"ORCHESTRATOR [{request_id}]: Tool Decision: Action='{action_type}', Input='{action_input_dict}'") yield "status", f"[Path: {action_type}]" final_system_prompt_str = custom_system_prompt or prompts.DEFAULT_SYSTEM_PROMPT context_str, final_user_prompt_str = None, "" if action_type == "answer_using_conversation_memory": yield "status", "[Searching conversation memory...]" mems = retrieve_memories_semantic(f"User query: {user_input}\nContext:\n{history_str_for_prompt[-1000:]}", k=2) context_str = "Relevant Past Interactions:\n" + "\n".join([f"- User:{m.get('user_input','')}->AI:{m.get('bot_response','')} (Takeaway:{m.get('metrics',{}).get('takeaway','N/A')})" for m in mems]) if mems else "No relevant past interactions found." final_system_prompt_str += " Respond using Memory Context, guidelines, & history." elif WEB_SEARCH_ENABLED and action_type in ["search_duckduckgo_and_report", "scrape_url_and_report"]: query_or_url = action_input_dict.get("search_engine_query") or action_input_dict.get("url") if query_or_url: yield "status", f"[Web: '{query_or_url[:60]}'...]" web_results = [] try: if action_type == "search_duckduckgo_and_report": web_results = search_and_scrape_duckduckgo(query_or_url, num_results=2) elif action_type == "scrape_url_and_report": web_results = [scrape_url(query_or_url)] except Exception as e: web_results = [{"url": query_or_url, "error": str(e)}] context_str = "Web Content:\n" + "\n".join([f"Source {i+1}:\nURL:{r.get('url','N/A')}\nTitle:{r.get('title','N/A')}\nContent:\n{(r.get('content') or r.get('error') or 'N/A')[:3500]}\n---" for i,r in enumerate(web_results)]) if web_results else f"No results from {action_type} for '{query_or_url}'." yield "status", "[Synthesizing web report...]" final_system_prompt_str += " Generate report/answer from web content, history, & guidelines. Cite URLs as [Source X]." else: # quick_respond or fallback final_system_prompt_str += " Respond directly using guidelines & history." final_user_prompt_str = prompts.get_final_response_prompt(history_str_for_prompt, initial_insights_ctx_str, user_input, context_str) final_llm_messages = [{"role": "system", "content": final_system_prompt_str}, {"role": "user", "content": final_user_prompt_str}] streamed_response = "" try: for chunk in call_model_stream(provider=provider_name, model_display_name=model_display_name, messages=final_llm_messages, api_key_override=ui_api_key_override, temperature=0.6, max_tokens=2500): if isinstance(chunk, str) and chunk.startswith("Error:"): streamed_response += f"\n{chunk}\n"; yield "response_chunk", f"\n{chunk}\n"; break streamed_response += chunk; yield "response_chunk", chunk except Exception as e: streamed_response += f"\n\n(Error: {e})"; yield "response_chunk", f"\n\n(Error: {e})" final_bot_text = streamed_response.strip() or "(No response or error.)" logger.info(f"ORCHESTRATOR [{request_id}]: Finished. Total: {time.time() - process_start_time:.2f}s. Resp len: {len(final_bot_text)}") yield "final_response_and_insights", {"response": final_bot_text, "insights_used": parsed_initial_insights_list}