[KM-567][AI] Planner agent: prompt + few-shot examples

- prompt.py: build_planner_prompt assembles per-call human content (business
context + condensed catalog + tool list + constraints + examples + question,
plus prior error on retry).
- config/prompts/planner.md: system prompt encoding INV-1/6/7 and the planning
principles.
- examples.py: two few-shot TaskLists (A exploratory revenue-by-category;
B descriptive monthly-trend-by-region with date_trunc), built from the real
TaskList schema so they cannot drift from the output contract.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/agents/planner/examples.py

@@ -0,0 +1,227 @@
+"""Few-shot examples for the planner prompt.
+
+Two illustrative (question -> TaskList) pairs that teach the OUTPUT SHAPE:
+stages, dependency edges, parallelism, ordered tool-call chains, inline QueryIR,
+and "${t<id>}" placeholders. They reference a hypothetical sales catalog
+(`src_sales` / `t_orders`); these ids are part of the illustration and are not
+validated against the user's real catalog. v1 is descriptive/diagnostic — no
+modeling tasks.
+
+See AGENT_ARCHITECTURE_CONTEXT_new.md §7.3 (Examples A and B).
+"""
+
+from __future__ import annotations
+
+from .schemas import Task, TaskList, ToolCall
+
+# --------------------------------------------------------------------------- #
+# Example A — exploratory, no modeling.
+# "Which product categories drove last quarter's revenue?"
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_A = TaskList(
+    plan_id="example_a",
+    goal_restated="Identify which product categories contributed most to last quarter's revenue.",
+    assumptions=["'last quarter' = 2026-01-01 to 2026-03-31."],
+    open_questions=[],
+    tasks=[
+        Task(
+            id="t1",
+            stage="data_understanding",
+            objective="Confirm the sales source exposes category, revenue, and order date.",
+            tool_calls=[ToolCall(tool="describe_source", args={"source_id": "src_sales"})],
+            expected_output="source_shape",
+            success_criteria="describe_source returns the orders table with the 3 columns.",
+            depends_on=[],
+            parallelizable_with=[],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t2",
+            stage="evaluation",
+            objective="Sum last quarter's revenue per category, ranked high to low.",
+            tool_calls=[
+                ToolCall(
+                    tool="query_structured",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {"kind": "column", "column_id": "c_category", "alias": "category"},
+                                {
+                                    "kind": "agg",
+                                    "fn": "sum",
+                                    "column_id": "c_revenue",
+                                    "alias": "revenue",
+                                },
+                            ],
+                            "filters": [
+                                {
+                                    "column_id": "c_order_date",
+                                    "op": "between",
+                                    "value": ["2026-01-01", "2026-03-31"],
+                                    "value_type": "date",
+                                }
+                            ],
+                            "group_by": ["c_category"],
+                            "order_by": [{"column_id": "revenue", "dir": "desc"}],
+                            "limit": 20,
+                        }
+                    },
+                )
+            ],
+            expected_output="revenue_by_category",
+            success_criteria="Produced a ranked revenue figure per category.",
+            depends_on=["t1"],
+            parallelizable_with=["t3"],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t3",
+            stage="evaluation",
+            objective="Get total last-quarter revenue to contextualize each category's share.",
+            tool_calls=[
+                ToolCall(
+                    tool="query_structured",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {
+                                    "kind": "agg",
+                                    "fn": "sum",
+                                    "column_id": "c_revenue",
+                                    "alias": "total_revenue",
+                                }
+                            ],
+                            "filters": [
+                                {
+                                    "column_id": "c_order_date",
+                                    "op": "between",
+                                    "value": ["2026-01-01", "2026-03-31"],
+                                    "value_type": "date",
+                                }
+                            ],
+                        }
+                    },
+                )
+            ],
+            expected_output="total_revenue",
+            success_criteria="Produced a single total revenue figure for the quarter.",
+            depends_on=["t1"],
+            parallelizable_with=["t2"],
+            estimated_cost="low",
+        ),
+    ],
+)
+
+# --------------------------------------------------------------------------- #
+# Example B — descriptive / trend.
+# "How has monthly revenue trended by region this year, and what's unusual?"
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_B = TaskList(
+    plan_id="example_b",
+    goal_restated="Describe this year's monthly revenue trend and flag unusual months.",
+    assumptions=["'this year' starts 2026-01-01."],
+    open_questions=["'Unusual' is interpreted as months far from the typical monthly revenue."],
+    tasks=[
+        Task(
+            id="t1",
+            stage="data_understanding",
+            objective="Confirm the sales source exposes order date, revenue, and region.",
+            tool_calls=[ToolCall(tool="describe_source", args={"source_id": "src_sales"})],
+            expected_output="source_shape",
+            success_criteria="describe_source returns the orders table with the needed columns.",
+            depends_on=[],
+            parallelizable_with=[],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t2",
+            stage="data_preparation",
+            objective="Pull this year's order dates, revenue, and region.",
+            tool_calls=[
+                ToolCall(
+                    tool="query_structured",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {
+                                    "kind": "column",
+                                    "column_id": "c_order_date",
+                                    "alias": "order_date",
+                                },
+                                {"kind": "column", "column_id": "c_revenue", "alias": "revenue"},
+                                {"kind": "column", "column_id": "c_region", "alias": "region"},
+                            ],
+                            "filters": [
+                                {
+                                    "column_id": "c_order_date",
+                                    "op": ">=",
+                                    "value": "2026-01-01",
+                                    "value_type": "date",
+                                }
+                            ],
+                            "limit": 10000,
+                        }
+                    },
+                )
+            ],
+            expected_output="ytd_rows",
+            success_criteria="Produced this year's order-level rows with date, revenue, region.",
+            depends_on=["t1"],
+            parallelizable_with=[],
+            estimated_cost="medium",
+        ),
+        Task(
+            id="t3",
+            stage="evaluation",
+            objective="Bucket the order dates into months to form the monthly trend.",
+            tool_calls=[
+                ToolCall(
+                    tool="date_trunc",
+                    args={"values": "${t2}", "granularity": "month"},
+                )
+            ],
+            expected_output="monthly_series",
+            success_criteria="Produced a per-month revenue series.",
+            depends_on=["t2"],
+            parallelizable_with=[],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t4",
+            stage="evaluation",
+            objective="Quantify month-to-month spread to flag unusual months.",
+            tool_calls=[ToolCall(tool="compute_stddev", args={"values": "${t3}"})],
+            expected_output="monthly_volatility",
+            success_criteria="Produced a stddev figure that flags months above the typical spread.",
+            depends_on=["t3"],
+            parallelizable_with=[],
+            estimated_cost="low",
+        ),
+    ],
+)
+
+
+EXAMPLES: list[tuple[str, TaskList]] = [
+    ("Which product categories drove last quarter's revenue?", _EXAMPLE_A),
+    ("How has monthly revenue trended by region this year, and what's unusual?", _EXAMPLE_B),
+]
+
+
+def render_examples() -> str:
+    """Render the few-shots as text for the planner prompt."""
+    blocks: list[str] = []
+    for i, (question, plan) in enumerate(EXAMPLES, start=1):
+        blocks.append(
+            f"## Example {i}\n\n"
+            f"Question:\n{question}\n\n"
+            f"TaskList:\n{plan.model_dump_json(indent=2)}"
+        )
+    return "\n\n".join(blocks)

src/agents/planner/prompt.py

@@ -0,0 +1,106 @@
+"""Builds the planner LLM human-message content.
+
+The system prompt (`config/prompts/planner.md`) carries the role, invariants,
+and planning principles. This module assembles the per-call human content:
+business context + condensed catalog + available tools + constraints + the
+few-shot examples + the question (+ the prior error on retry).
+
+Few-shot examples are rendered from `examples.py` (which builds them from the
+real `TaskList` schema) so they cannot drift from the output contract.
+"""
+
+from __future__ import annotations
+
+from .contracts import BusinessContext, ToolRegistry
+from .examples import render_examples
+from .inputs import CatalogSummary, Constraints
+
+
+def render_business_context(context: BusinessContext) -> str:
+    lines = [
+        f"Project: {context.project_id} (industry: {context.industry}, "
+        f"context completeness: {context.completeness})",
+        f"Business: {context.business_description}",
+        f"Scale & scope: {context.scale_and_scope}",
+    ]
+    if context.key_terms:
+        lines.append("Key terms:")
+        lines.extend(f"  - {kt.term}: {kt.meaning}" for kt in context.key_terms)
+    if context.data_overview:
+        lines.append("Data overview:")
+        lines.extend(
+            f"  - {n.table_name}: {n.row_represents}" for n in context.data_overview
+        )
+    if context.data_column_notes:
+        lines.append("Column notes:")
+        lines.extend(
+            f"  - {n.column_name}: {n.meaning}" for n in context.data_column_notes
+        )
+    if context.whats_normal:
+        lines.append(f"What's normal: {context.whats_normal}")
+    if context.recent_events:
+        lines.append(f"Recent events: {context.recent_events}")
+    if context.things_to_watch_for:
+        lines.append(f"Things to watch for: {context.things_to_watch_for}")
+    if context.open_questions:
+        lines.append("Known open questions:")
+        lines.extend(f"  - {q}" for q in context.open_questions)
+    return "\n".join(lines)
+
+
+def render_registry(tools: ToolRegistry) -> str:
+    if not tools.tools:
+        return "(no tools available)"
+    blocks: list[str] = []
+    for spec in tools.tools:
+        required = spec.input_schema.get("required", [])
+        blocks.append(
+            f"- {spec.name} (category: {spec.category}, returns: {spec.output_kind})\n"
+            f"  required args: {required}\n"
+            f"  {spec.description}"
+        )
+    return "\n".join(blocks)
+
+
+def render_constraints(constraints: Constraints) -> str:
+    lines = [
+        f"- max_tasks: {constraints.max_tasks}",
+        f"- modeling_allowed: {constraints.modeling_allowed} "
+        "(no modeling tools exist in v1 — do not emit modeling tasks)",
+        f"- row_budget: {constraints.row_budget}",
+    ]
+    if constraints.token_budget is not None:
+        lines.append(f"- token_budget: {constraints.token_budget}")
+    if constraints.time_budget_seconds is not None:
+        lines.append(f"- time_budget_seconds: {constraints.time_budget_seconds}")
+    return "\n".join(lines)
+
+
+def build_planner_prompt(
+    context: BusinessContext,
+    catalog: CatalogSummary,
+    tools: ToolRegistry,
+    query: str,
+    constraints: Constraints,
+    previous_error: str | None = None,
+) -> str:
+    """Return the human-message content for the planner LLM.
+
+    The system prompt (`config/prompts/planner.md`) is loaded separately by
+    `PlannerService`.
+    """
+    sections = [
+        f"# Business context\n\n{render_business_context(context)}",
+        f"# Catalog\n\n{catalog.render()}",
+        f"# Available tools\n\n{render_registry(tools)}",
+        f"# Constraints\n\n{render_constraints(constraints)}",
+        f"# Examples\n\n{render_examples()}",
+        f"# Question\n\n{query}",
+    ]
+    if previous_error:
+        sections.append(
+            "# Previous attempt failed validation\n\n"
+            f"{previous_error}\n\n"
+            "Emit a corrected TaskList. Do not repeat the same mistake."
+        )
+    return "\n\n".join(sections)

src/config/prompts/planner.md

@@ -0,0 +1,54 @@
+You are the Planner for Data Eyond, an AI data scientist that works within the
+CRISP-DM lifecycle. Your single job is to turn a business question into a
+**static analysis plan**: one `TaskList` that downstream deterministic code
+executes exactly as written.
+
+You plan. You do not execute, and you do not write prose for the user. You emit
+only a `TaskList` object that conforms to the provided schema.
+
+# Hard rules (non-negotiable)
+
+1. **Emit intent, never code.** Never write SQL, pandas, or any code. The only
+   query you express is an inline `QueryIR` (a JSON intent object) inside a
+   `query_structured` tool call's `args.ir`.
+2. **The plan is static.** There is no replanning and no execution feedback. Plan
+   the whole analysis up front; assume each task runs once, in dependency order.
+3. **Use only tools from the "Available tools" list.** Never invent a tool name.
+   Every `tool_calls[].tool` must be one of the listed tool names.
+4. **Reference only data that exists.** Every `source_id`, `table_id`, and
+   `column_id` you put in an inline `QueryIR` must come from the "Catalog"
+   section. Copy the stable ids verbatim — downstream validation does a literal
+   id lookup, so a paraphrased name fails.
+5. **No modeling in v1.** There are no modeling tools. Do not emit `modeling`
+   tasks. The product is descriptive/diagnostic only — no predictions, no charts.
+
+# How to plan
+
+- **Smallest plan that answers the question.** Do not exceed `Constraints.max_tasks`.
+- **A task is an ordered chain of tool calls** with fully-specified arguments. A
+  simple question is one task with one tool call; a step that needs a follow-up
+  computation is a short chain or a dependent task.
+- **Wire data between tasks with placeholders.** When a task needs an upstream
+  task's output as an argument, use the string `"${t<id>}"` (e.g. `"${t2}"`) as
+  the argument value. Set `depends_on` accordingly.
+- **Built-in aggregation vs compute_* tools.** Use `query_structured` for
+  count/sum/avg/min/max/count_distinct, filtering, and grouping. For statistics
+  the IR cannot express (median, percentile, mode, standard deviation), run
+  `query_structured` to fetch the series, then a `compute_*` tool on its output.
+- **Mixing structured + unstructured.** If qualitative context helps, add a
+  `retrieve_documents` task against an unstructured source listed in the catalog.
+- **Parallelism.** List sibling tasks that have no data dependency on each other
+  in `parallelizable_with` (must be mutually consistent with `depends_on`).
+- **CRISP-DM stages.** Tag each task with the stage it serves:
+  `data_understanding`, `data_preparation`, or `evaluation`. (Never `modeling`.)
+- **success_criteria is a reporting signal**, not a control trigger. State, in
+  checkable terms (counts, rates, "produced", "above"/"below"), what a good
+  result looks like. It never causes a retry.
+- **Surface uncertainty, don't guess.** If the question is ambiguous or the
+  catalog can't fully answer it, record it in `open_questions` and plan the best
+  defensible analysis anyway. Record interpretation choices in `assumptions`.
+
+# Output
+
+Return exactly one `TaskList`. The "Examples" section in the human message shows
+the required shape. Match it.