# app/modules/build_prompt.py from runtime.session_store import get_session_controller def build_planning_prompt(message, rag_context=""): """ Planner prompt: Always include the RAG context. - If rag_context is empty, an empty string is used. - 1st: Unconditionally generate 1 to 5 queries. - 2nd: Dynamically control 0 to 5 queries after FT. """ system_message = ( "You are a planning assistant. Your role is to generate search queries " "based on the user message and RAG context." ) return ( f"{system_message}\n" f"User question (multilingual): {message}\n\n" f"Context from RAG DB (may be empty):\n{rag_context}\n\n" "Task:\n" "- Generate 1 to 5 concise search queries.\n" "- If the RAG context already fully answers the question, you may generate 0 queries.\n" "- Prefer official documentation, standards, academic or authoritative sources.\n" "- Prefer English queries for global/academic sources.\n" "- If the user question is strongly local or language-specific, also generate queries in the same language.\n" "- Avoid redundant queries; use broader queries if they cover multiple facets.\n" "- Output only the queries, one per line." ) def build_generation_prompt(message, docs, history=None, hf_user=None, session_id=None, top_k=5): """ Answer generation prompt with context from retrieved documents. - Include up to top_k document snippets as context. - Include recent conversation history if available. - System message is fixed internally for answer generation role. """ system_message = ( "You are an answer generation assistant. Your role is to synthesize " "information from context and produce a clear, concise, Markdown-formatted answer " "with citations in (title)[url] format." ) # Sort by score if available, then take Top-K sorted_docs = sorted(docs or [], key=lambda d: d.get("score", 0), reverse=True) top_docs = sorted_docs[:top_k] # Build structured context string context_entries = [] for d in top_docs: q = d.get("query", "") title = d.get("title", "") url = d.get("url", "") snippet = d.get("snippet") or d.get("summary") or d.get("content") or "" score = d.get("score", 0) entry = ( f"Query: {q}\n" f"Title: {title}\n" f"URL: {url}\n" f"Snippet: {snippet}\n" f"Score: {score}\n" ) context_entries.append(entry.strip()) context = "\n---\n".join(context_entries) # Conversation history (optional) history_text = "" if history: if history == "history_ctr": history_controller = get_session_controller(hf_user, session_id) try: ctx = history_controller.build_prompt_history(message, max_tokens=1024, top_k=10) history_text = "\nConversation history (relevant with user message):\n" + ctx except Exception: history_text = "" else: try: last_items = history[-4:] if isinstance(history, list) else [str(history)] history_text = "\nConversation history (recent):\n" + "\n".join(f"{r}: {c}" for r, c in last_items) except Exception: history_text = "" # Final prompt assembly return ( f"{system_message}\n" f"{history_text}\n" "Context (structured search results):\n" f"{context}\n\n" "Instructions:\n" "- Provide the answer to the user message below using the context above.\n" "- Answer in the same language as the user message.\n" "- Follow Markdown format strictly.\n" "- Include citations using (title)[url] format whenever referencing sources.\n" "- Use the score field to prioritize higher relevance results first.\n" "- Cross-check and summarize multiple sources when possible.\n" "- If context is insufficient, state limitations clearly.\n\n" f"User message: {message}\n" "Answer:" )