| """Builds the Assembler LLM human-message content. |
| |
| The system prompt (`config/prompts/assembler.md`) carries the role and rules. This |
| module assembles the per-call human content: the business context + the executed |
| `RunState` (task objectives, statuses, and structured tool outputs) + the original |
| question. Tool outputs are rendered compactly as data — the model turns them into |
| prose and markdown tables. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from ..planner.contracts import BusinessContext, ToolOutput |
| from ..planner.prompt import render_business_context |
| from .schemas import RunState, TaskResult |
|
|
| _MAX_ROWS = 20 |
|
|
|
|
| def render_run_state(run_state: RunState) -> str: |
| lines = [f"Plan: {run_state.plan_id}"] |
| if run_state.open_questions: |
| lines.append("Open questions carried from the plan:") |
| lines.extend(f" - {q}" for q in run_state.open_questions) |
| lines.append("") |
| lines.append("Task results (in execution order):") |
| for task_id, result in run_state.results.items(): |
| lines.append(_render_task(task_id, result)) |
| return "\n".join(lines) |
|
|
|
|
| def _render_task(task_id: str, result: TaskResult) -> str: |
| lines = [f"- [{result.status}] {task_id}: {result.objective}"] |
| if result.error: |
| lines.append(f" note: {result.error}") |
| for output in result.outputs: |
| lines.append(f" {_render_output(output)}") |
| return "\n".join(lines) |
|
|
|
|
| def _render_output(output: ToolOutput) -> str: |
| if output.kind == "error": |
| return f"({output.tool}) error: {output.error}" |
| if output.kind == "table" and output.columns is not None: |
| header = ", ".join(output.columns) |
| rows = output.rows or [] |
| preview = "; ".join( |
| " | ".join(str(cell) for cell in row) for row in rows[:_MAX_ROWS] |
| ) |
| more = "" if len(rows) <= _MAX_ROWS else f" … (+{len(rows) - _MAX_ROWS} more rows)" |
| return f"({output.tool}) table [{header}]: {preview}{more}" |
| meta = f" meta={output.meta}" if output.meta else "" |
| return f"({output.tool}) {output.kind}: {output.value}{meta}" |
|
|
|
|
| def build_assembler_prompt( |
| run_state: RunState, |
| context: BusinessContext, |
| question: str | None = None, |
| reply_language: str | None = None, |
| ) -> str: |
| sections = [ |
| f"# Business context\n\n{render_business_context(context)}", |
| f"# Analysis results\n\n{render_run_state(run_state)}", |
| ] |
| if question: |
| sections.append(f"# Original question\n\n{question}") |
| if reply_language: |
| |
| |
| |
| sections.append( |
| f"# Reply language (MANDATORY)\n\n" |
| f"Write `chat_answer` AND every narrative field entirely in **{reply_language}**. " |
| f"The results above use English column names and labels — do NOT let that switch " |
| f"your reply to English. (If the user explicitly asked for a different language, " |
| f"follow that instead.)" |
| ) |
| return "\n\n".join(sections) |
|
|
