| """Builds the planner LLM prompt from question + catalog. |
| |
| Renders the catalog into a compact textual form that fits the LLM context |
| window. For users with ≤50 tables the full catalog goes in verbatim. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from ...catalog.models import Catalog |
| from ...catalog.render import render_source |
|
|
|
|
| def render_catalog(catalog: Catalog) -> str: |
| """Render every Source in the catalog as text. One blank line between sources.""" |
| if not catalog.sources: |
| return "(catalog is empty — the user has not registered any structured data yet)" |
| return "\n\n".join(render_source(s) for s in catalog.sources) |
|
|
|
|
| def build_planner_prompt( |
| question: str, |
| catalog: Catalog, |
| previous_error: str | None = None, |
| ) -> str: |
| """Return the human-message content for the planner LLM. |
| |
| Composed of three sections in order: |
| 1. The user's question. |
| 2. The user's full catalog (rendered). |
| 3. (optional) The previous attempt's error, on retry. |
| |
| The system prompt (`config/prompts/query_planner.md`) is loaded |
| separately by `QueryPlannerService`. |
| """ |
| sections = [ |
| f"# Question\n\n{question}", |
| f"# Catalog\n\n{render_catalog(catalog)}", |
| ] |
| if previous_error: |
| sections.append( |
| "# Previous attempt failed validation\n\n" |
| f"{previous_error}\n\n" |
| "Emit a corrected IR. Do not repeat the same mistake." |
| ) |
| return "\n\n".join(sections) |
|
|
