Blame for modules/processing/prompt_builder.py

# app/modules/processing/prompt_builder.py
from typing import Any, List, Dict
from runtime.session_store import get_session_controller

def build_planning_prompt(message: str, rag_context: str = "") -> List[Dict[str, str]]:

    system_content = (
        "You are an elite query planner for a search-augmented LLM. Your task is to determine "
        "if an external web search is absolutely mandatory to answer the user's question.\n\n"
        "TRIAE STRATEGY (Priority):\n"
        "1. INTERNAL KNOWLEDGE: If the question is about historical facts, static scientific truths, "
        "unchanging concepts, or general logic that you already know perfectly, rely entirely on your own knowledge.\n"
        "2. LOCAL RAG CONTEXT: Use the provided RAG context (Wikipedia) to complement or verify your knowledge.\n"
        "3. WEB SEARCH: Generate web search queries ONLY if the question requires real-time information (after 2025/2026), "
        "highly localized niche data not found in Wikipedia, or breaking updates.\n\n"
        "CRITICAL PERFORMANCE REQUIREMENT:\n"
        "- Keep your internal thinking (<think>...</think>) extremely short (under 3 lines). Do not ramble.\n"
        "- If internal knowledge or local RAG is sufficient, immediately output exactly 0 queries.\n"
        "- Output ONLY the raw queries, one per line, no numbers, no bullet points, no markdown markup outside thinking tags."
    )
    
    rag_section = rag_context.strip() if rag_context else "None (No relevant local database records found)."

    user_content = (
        f"User Question: {message}\n\n"
        f"Available Local RAG Context (Wikipedia snippets):\n{rag_section}\n\n"
        "Task:\n"
        "- Output 1 to 3 highly concise search queries ONLY if external web search is required.\n"
        "- Output exactly '0' if no external search is needed."
    )

    return [
        {"role": "system", "content": system_content},
        {"role": "user", "content": user_content}
    ]


def build_generation_prompt(
    message: str, 
    docs: list, 
    history: Any = None, 
    hf_user: str = None, 
    session_id: str = None, 
    top_k: int = 5
) -> List[Dict[str, str]]:
    
    system_content = (
        "You are an advanced, direct response assistant. Your objective is to synthesize your internal "
        "expertise with the provided context to answer the user comprehensively and clearly in Markdown format.\n\n"
        "CRITICAL INSTRUCTIONS FOR SPEED & TIMEOUT PREVENTION:\n"
        "- ABSOLUTELY NO INTERNAL REASONING OR THINKING PROCESS ALLOWED. Do not output <think> tags or chains of thought.\n"
        "- Start generating the final answer directly from the first token.\n"
        "- Prioritize your comprehensive internal knowledge for historical or logical accuracy, and use the provided context to add specific references or URLs.\n"
        "- Cite sources accurately using the exact (Title)[URL] format when referencing the provided context elements."
    )

    sorted_docs = sorted(docs or [], key=lambda d: d.get("score", 0), reverse=True)
    top_docs = sorted_docs[:top_k]

    context_entries = []
    for d in top_docs:
        entry = (
            f"Query Source: {d.get('query', 'N/A')}\n"
            f"Title: {d.get('title', 'Untitled')}\n"
            f"URL: {d.get('url', '')}\n"
            f"Content: {d.get('snippet') or d.get('summary') or d.get('content') or ''}\n"
        )
        context_entries.append(entry.strip())

    context_str = "\n---\n".join(context_entries) if context_entries else "No external or local context provided. Rely entirely on your internal parameters."

    history_str = ""
    if history:
        if history == "history_ctr":
            history_controller = get_session_controller(hf_user, session_id)
            try:
                history_str = history_controller.build_prompt_history(message, max_tokens=1024, top_k=10)
            except Exception:
                pass
        else:
            try:
                last_items = history[-4:] if isinstance(history, list) else [str(history)]
                history_str = "\n".join(f"{r}: {c}" for r, c in last_items)
            except Exception:
                pass
                
    user_content = ""
    if history_str:
        user_content += f"Relevant Context from Past Conversation turns:\n{history_str}\n\n"

    user_content += (
        f"Retrieved Reference Context (Local RAG & Web Search):\n{context_str}\n\n"
        f"Instructions:\n"
        f"- Answer the user message using your internal knowledge and the reference context above.\n"
        f"- Respond in the same language as the user's message.\n"
        f"- Do not mention that you were forbidden to think. Just start the answer.\n\n"
        f"User Message: {message}"
    )

    return [
        {"role": "system", "content": system_content},
        {"role": "user", "content": user_content}
    ]