from __future__ import annotations from dataclasses import dataclass from core.ontology import axis_options, partition_sectors from core.ports import ChatModel from core.schemas import RouteResponse @dataclass class Step1Result: """UI-facing bundle: the faithful route plus the in/out-of-scope partition.""" route: RouteResponse in_scope: list[str] out_of_scope: list[str] def route_from_chips(sectors: list[str], lens: list[str], type_: list[str]) -> RouteResponse: """Chips path — the analyst's selections ARE the route. Pure identity, no LLM.""" return RouteResponse(sectors=list(sectors), lens=list(lens), type=list(type_)) def _render_options() -> str: opts = axis_options() sections = [ ("SECTORS (domain of need)", opts["sectors"]), ("LENS (analytical 2D pillar)", opts["lens"]), ("TYPE (information 1D pillar)", opts["type"]), ] lines = [] for title, items in sections: lines.append(title + ":") for it in items: syn = ", ".join(it["synonyms"][:6]) lines.append(f"- {it['label']}" + (f" — {syn}" if syn else "")) lines.append("") return "\n".join(lines).strip() def _build_messages(question: str) -> list[dict]: system = ( "You classify a humanitarian analyst's question into a fixed framework of " "three axes: sectors, lens, type. Return ONLY labels drawn from the lists " "below, copied verbatim. A question may map to multiple sectors. Lens and " "type give analytical context — return an empty list for either if the " "question does not clearly indicate one. Sectors should reflect the " "domain(s) of need.\n\n" + _render_options() ) return [ {"role": "system", "content": system}, {"role": "user", "content": question}, ] def route_from_message(question: str, model: ChatModel) -> RouteResponse: """Message path — the one fenced LLM call in Step 1. Classifies the question into {sectors, lens, type} as a structured RouteResponse.""" return model.structured(_build_messages(question), RouteResponse) def resolve_route( input_mode: str, question: str, sectors: list[str], lens: list[str], type_: list[str], model: ChatModel | None, ) -> Step1Result: """Orchestrate Step 1: dispatch by path, partition sectors. Raises ValueError only for empty-question validation; out-of-scope sectors are passed through to Step 2 for reconciliation.""" if input_mode == "message": if not question or not question.strip(): raise ValueError("Please type a question before routing.") route = route_from_message(question, model) else: route = route_from_chips(sectors, lens, type_) in_scope, out_of_scope = partition_sectors(route.sectors) return Step1Result(route=route, in_scope=in_scope, out_of_scope=out_of_scope) def step1_trace_line(res: Step1Result) -> str: """Skill-style trace line for the route (extends state['trace']).""" all_sectors = res.in_scope + res.out_of_scope line = "Step 1 — Understood request → " + (", ".join(all_sectors) or "no sectors") if res.route.lens: line += f" (lens: {', '.join(res.route.lens)})" return line