build-small-hackathon/hackathon-advisor
16
1"""The atlas chat engine: a two-pass, tool-grounded conversation over the idea map.2 3Flow per turn (the native MiniCPM5 tool protocol, run on the BASE model):4 51. *Pass 1* — the model sees the chat history plus the tool schemas (injected by the6 chat template via ``tools=``) and either calls one tool or answers plain prose.72. The call is validated and degraded through the chat-specific ladder8 (``resolve_chat_tool_call``), then executed against a fresh9 :class:`~hackathon_advisor.dashboard_repository.DashboardRepository` snapshot.103. The full verified result streams to the UI *first* (``tool_result`` + optional11 ``map_action``) so cards and the map always carry the real numbers.124. *Pass 2* — a compact digest (urls/ids/scores stripped, so the model cannot13 misquote what it never saw) goes back as a ``role:"tool"`` message and the model14 writes a short grounded answer at temperature 0 with NO tools injected.15 Empty results skip pass 2 entirely: a 1B narrating absent data is where it16 hallucinates, so those turns get a deterministic templated sentence instead.17 18The engine is UI- and app-agnostic: it depends only on a ChatRunner and a19repository factory, and every yielded event is a JSON-serializable dict.20"""21 22from __future__ import annotations23 24from collections.abc import Callable, Iterator25import re26from typing import Any27 28from hackathon_advisor._text import clean29from hackathon_advisor.aliases import normalize_text30from hackathon_advisor.dashboard_chat_contracts import (31 ChatToolResolution,32 chat_tool_schemas,33 data_intent_call,34 resolve_chat_tool_call,35 smalltalk_intent,36 strip_function_blocks,37)38from hackathon_advisor.dashboard_repository import DashboardRepository39from hackathon_advisor.model_runtime import ChatRunner40from hackathon_advisor.tool_contracts import ToolCall41 42# One generous budget for every chat generation: with thinking enabled the model43# reasons inside <think>...</think> before the tool call / answer, and the trace44# alone can run long. The model stops at EOS well before the cap on normal turns.45MAX_CHAT_GENERATION_TOKENS = 409646MAX_HISTORY_MESSAGES = 12 # six user/assistant turns47MAX_ANSWER_HISTORY_MESSAGES = 4 # two turns of context for the prose passes48MAX_HISTORY_MESSAGE_CHARS = 60049 50CHAT_PLANNING_PROMPT = (51 "You are the Atlas Guide for the Build Small hackathon idea map. "52 "You cannot see the atlas directly: the ONLY way to answer a question about projects, "53 "clusters, quests, teams, or recent activity is to call one of the provided tools, which "54 "read the live atlas. For any such question respond with exactly one tool call and no "55 'other text, for example: <function name="search_projects"><param name="query">voice'56 "</param></function>. Reply in plain prose only for greetings or questions about yourself."57)58 59CHAT_ANSWER_PROMPT = (60 "You are the Atlas Guide for the Build Small hackathon idea map. "61 "Write a short conversational answer to the user's question using ONLY the facts in the "62 "tool response. Quote counts and names exactly as given. Do not invent projects, numbers, "63 "or links. Do not enumerate every item: summarize, naming at most three examples. "64 "Two to four sentences, no lists, no markdown."65)66 67CHAT_SMALLTALK_PROMPT = (68 "You are the Atlas Guide for the Build Small hackathon idea map. "69 "Reply briefly and warmly. You have NO project data in this conversation: never state "70 "project names, counts, likes, or rankings, and do not defend earlier numbers — if asked "71 "about data, say you should look it up and suggest asking what everyone is building, "72 "which projects completed the most quests, or what clusters exist. One or two sentences."73)74 75# Keys stripped from the model-facing digest. The UI renders links and ids from the76# verified payload; the model only needs labels, titles, and counts.77_DIGEST_DROPPED_KEYS = frozenset({"url", "id", "score", "host", "quest_ids"})78 79# A trailing fragment of "<function" left by a max_new_tokens cut mid-marker.80_PARTIAL_TAG_RE = re.compile(r"<[a-z]{0,8}$")81 82THINK_END_MARKER = "</think>"83 84 85class _ThinkSplitter:86 """Incrementally split a thinking-mode stream into (kind, text) chunks.87 88 With ``enable_thinking`` the chat template ends the prompt with ``<think>\\n``,89 so the generation is reasoning text up to ``</think>`` followed by the real90 content. The marker can arrive split across stream pieces, so a small tail91 buffer is kept until it can no longer be a marker prefix. When the runner does92 not think (rules backend) every piece passes straight through as answer text."""93 94 def __init__(self, active: bool) -> None:95 self._thinking = bool(active)96 self._buffer = ""97 98 def feed(self, piece: str) -> list[tuple[str, str]]:99 if not self._thinking:100 return [("answer", piece)] if piece else []101 self._buffer += piece102 marker = self._buffer.find(THINK_END_MARKER)103 if marker >= 0:104 thought = self._buffer[:marker]105 rest = self._buffer[marker + len(THINK_END_MARKER) :].lstrip("\n")106 self._buffer = ""107 self._thinking = False108 chunks: list[tuple[str, str]] = []109 if thought:110 chunks.append(("thinking", thought))111 if rest:112 chunks.append(("answer", rest))113 return chunks114 keep = _marker_prefix_length(self._buffer)115 flush, self._buffer = (116 self._buffer[: len(self._buffer) - keep],117 self._buffer[len(self._buffer) - keep :],118 )119 return [("thinking", flush)] if flush else []120 121 def finish(self) -> list[tuple[str, str]]:122 """Flush the tail when the stream ends mid-thought (max_new_tokens cut)."""123 if self._thinking and self._buffer:124 tail, self._buffer = self._buffer, ""125 return [("thinking", tail)]126 return []127 128 129def _marker_prefix_length(text: str) -> int:130 for length in range(min(len(text), len(THINK_END_MARKER) - 1), 0, -1):131 if THINK_END_MARKER.startswith(text[-length:]):132 return length133 return 0134 135 136class DashboardChatEngine:137 def __init__(138 self,139 runner: ChatRunner,140 repository_factory: Callable[[], DashboardRepository],141 ) -> None:142 self.runner = runner143 self.repository_factory = repository_factory144 145 def turn_stream(146 self,147 message: str,148 history: list[dict[str, Any]] | None = None,149 ) -> Iterator[dict[str, Any]]:150 history = _normalize_history(history)151 normalized, corrections = normalize_text(message)152 yield {153 "type": "start",154 "normalized_text": normalized,155 "corrections": [correction.to_dict() for correction in corrections],156 }157 repository = self.repository_factory()158 159 yield {"type": "stage", "stage": "planning", "label": "Reading the atlas"}160 resolution, raw_output = yield from self._pick_tool(normalized, history)161 if resolution.status == "none":162 # Accuracy backstop: when the model answers in prose (declining its tools),163 # route any substantive question to a tool — a matched intent first, BM25164 # search otherwise. Only greetings/meta/short follow-ups may stay on the165 # ungrounded small-talk path; this is a data surface, and letting a question166 # like "how many voice apps" through is how invented facts reach the user.167 intent = data_intent_call(normalized)168 if intent is None and not smalltalk_intent(normalized):169 intent = ToolCall("search_projects", {"query": normalized})170 if intent is not None:171 resolution = ChatToolResolution(172 status="defaulted",173 call=intent,174 errors=("model answered without a tool; routed by intent",),175 )176 yield {177 "type": "tool_call",178 "name": resolution.call.name if resolution.call else "",179 "arguments": resolution.call.arguments if resolution.call else {},180 "status": resolution.status,181 "errors": list(resolution.errors),182 }183 184 if resolution.status == "none":185 response = yield from self._smalltalk(normalized, history, raw_output)186 yield self._done(normalized, history, response, tool="", data={}, map_action=None)187 return188 189 call = resolution.call190 assert call is not None191 yield {192 "type": "stage",193 "stage": "running_tool",194 "tool": call.name,195 "label": f"Calling {call.name}",196 }197 # _execute may swap the tool (show_project falls back to search when no198 # project matches), so the executed name drives rendering from here on.199 tool_name, data, map_action, empty_reason = self._execute(call, repository)200 yield {"type": "tool_result", "tool": tool_name, "data": data, "map_action": map_action}201 202 if empty_reason:203 response = _templated_sentence(call, data, empty_reason)204 yield {"type": "answer_skipped", "reason": empty_reason, "text": response}205 else:206 yield {"type": "stage", "stage": "writing", "label": "Writing the answer"}207 executed = ToolCall(tool_name, call.arguments)208 response = yield from self._grounded_answer(normalized, history, executed, data)209 if not response:210 response = _templated_sentence(call, data, "empty_answer")211 yield {"type": "answer_skipped", "reason": "empty_answer", "text": response}212 213 yield self._done(214 normalized, history, response, tool=tool_name, data=data, map_action=map_action215 )216 217 def _pick_tool(218 self,219 message: str,220 history: list[dict[str, Any]],221 ) -> Iterator[dict[str, Any]]:222 messages = [223 {"role": "system", "content": CHAT_PLANNING_PROMPT},224 *history,225 {"role": "user", "content": message},226 ]227 splitter = _ThinkSplitter(getattr(self.runner, "supports_thinking", False))228 answer_pieces: list[str] = []229 for count, piece in self.runner.stream(230 messages,231 tools=chat_tool_schemas(),232 max_new_tokens=MAX_CHAT_GENERATION_TOKENS,233 enable_thinking=True,234 ):235 for kind, text in splitter.feed(piece):236 if kind == "thinking":237 yield {"type": "thinking", "pass": 1, "text": text}238 else:239 answer_pieces.append(text)240 yield {241 "type": "model_progress",242 "pass": 1,243 "tokens": count,244 "max_tokens": MAX_CHAT_GENERATION_TOKENS,245 }246 for _kind, text in splitter.finish():247 yield {"type": "thinking", "pass": 1, "text": text}248 # Only the post-thinking text may be parsed: the reasoning trace legitimately249 # talks about <function ...> syntax without being a call.250 raw_output = "".join(answer_pieces).strip()251 return resolve_chat_tool_call(raw_output, fallback_query=message), raw_output252 253 def _smalltalk(254 self,255 message: str,256 history: list[dict[str, Any]],257 raw_output: str,258 ) -> Iterator[dict[str, Any]]:259 """Dedicated no-tools generation: the pass-1 output is tuned for tool260 selection, not for a satisfying greeting, so chit-chat gets its own pass."""261 yield {"type": "stage", "stage": "writing", "label": "Writing the answer"}262 messages = [263 {"role": "system", "content": CHAT_SMALLTALK_PROMPT},264 *_answer_history(history),265 {"role": "user", "content": message},266 ]267 response = yield from self._stream_prose(messages, MAX_CHAT_GENERATION_TOKENS)268 if not response:269 response = strip_function_blocks(raw_output) or (270 "Hello! Ask me what everyone is building, which projects completed the most "271 "quests, or what clusters exist."272 )273 yield {"type": "answer_skipped", "reason": "empty_answer", "text": response}274 return response275 276 def _grounded_answer(277 self,278 message: str,279 history: list[dict[str, Any]],280 call: ToolCall,281 data: dict[str, Any],282 ) -> Iterator[dict[str, Any]]:283 digest = render_digest(_digest_for_model(call.name, data))284 # NO history here: every fact the answer needs is in the digest, and a greedy285 # 1B echoes similar-sounding lines from prior turns over the digest in front286 # of it. Conversation context only matters for pass-1's tool choice.287 messages = [288 {"role": "system", "content": CHAT_ANSWER_PROMPT},289 {"role": "user", "content": message},290 {291 "role": "assistant",292 "content": "",293 "tool_calls": [{"name": call.name, "arguments": call.arguments}],294 },295 {"role": "tool", "content": digest},296 ]297 return (yield from self._stream_prose(messages, MAX_CHAT_GENERATION_TOKENS))298 299 def _stream_prose(300 self,301 messages: list[dict[str, Any]],302 max_new_tokens: int,303 ) -> Iterator[dict[str, Any]]:304 """Stream a no-tools generation as thinking + token events; returns the prose.305 306 The reasoning trace streams as ``thinking`` events; only the post-think text307 becomes the answer. If a stray ``<function`` shows up in the answer the stream308 stops early; the ``done`` response carries the stripped text, which the UI309 treats as authoritative."""310 splitter = _ThinkSplitter(getattr(self.runner, "supports_thinking", False))311 pieces: list[str] = []312 stream = self.runner.stream(messages, max_new_tokens=max_new_tokens, enable_thinking=True)313 stray_function = False314 try:315 for count, piece in stream:316 for kind, text in splitter.feed(piece):317 if kind == "thinking":318 yield {"type": "thinking", "pass": 2, "text": text}319 continue320 pieces.append(text)321 if "<function" in "".join(pieces[-4:]):322 stray_function = True323 break324 yield {"type": "token", "text": text}325 if stray_function:326 break327 yield {328 "type": "model_progress",329 "pass": 2,330 "tokens": count,331 "max_tokens": max_new_tokens,332 }333 finally:334 close = getattr(stream, "close", None)335 if close is not None:336 close()337 for _kind, text in splitter.finish():338 yield {"type": "thinking", "pass": 2, "text": text}339 text = "".join(pieces)340 marker = text.find("<function")341 if marker >= 0:342 text = text[:marker]343 # A generation cut at max_new_tokens can end mid-marker ("<fun"); drop any344 # trailing partial tag so it never reaches the authoritative response.345 text = _PARTIAL_TAG_RE.sub("", text)346 return clean(strip_function_blocks(text))347 348 def _execute(349 self,350 call: ToolCall,351 repository: DashboardRepository,352 ) -> tuple[str, dict[str, Any], dict[str, Any] | None, str]:353 """Run one validated tool; returns (executed tool, data, map action, empty reason)."""354 name = call.name355 if name == "atlas_overview":356 data = repository.overview()357 return name, data, {"type": "clear_filters"}, ""358 if name == "list_clusters":359 data = repository.list_clusters()360 return name, data, None, "" if data["clusters"] else "no_clusters"361 if name == "show_cluster":362 label = clean(call.arguments.get("label"))363 detail = repository.cluster_detail(label)364 if detail is None:365 return name, {"requested_label": label}, None, "unknown_cluster"366 return name, detail, {"type": "filter_cluster", "label": detail["label"]}, ""367 if name == "list_quests":368 data = repository.list_quests()369 if data["status"] != "analyzed":370 return name, data, None, "quests_not_analyzed"371 return name, data, None, ""372 if name == "show_quest":373 quest = clean(call.arguments.get("quest"))374 detail = repository.quest_detail(quest)375 if detail is None:376 return name, {"requested_quest": quest}, None, "unknown_quest"377 if detail["status"] != "analyzed":378 return name, detail, None, "quests_not_analyzed"379 map_action = {"type": "filter_quest", "quest": detail["id"]}380 if detail["project_count"] == 0:381 return name, detail, map_action, "quest_no_projects"382 return name, detail, map_action, ""383 if name == "show_project":384 requested = clean(call.arguments.get("project"))385 detail = repository.project_detail(requested)386 if detail is None:387 # Half-remembered names still get useful cards: fall back to search.388 return self._search(repository, requested)389 return name, detail, {"type": "highlight_projects", "ids": [detail["id"]]}, ""390 if name == "top_projects_by_quests":391 data = repository.top_by_quests()392 if data["status"] != "analyzed":393 return name, data, None, "quests_not_analyzed"394 if not data["rows"]:395 return name, data, None, "no_leaderboard_rows"396 ids = [row["id"] for row in data["rows"]]397 return name, data, {"type": "highlight_projects", "ids": ids}, ""398 if name == "search_projects":399 return self._search(repository, clean(call.arguments.get("query")))400 if name == "recent_activity":401 data = repository.recent_activity()402 if not data["projects"]:403 return name, data, None, "no_projects"404 ids = [project["id"] for project in data["projects"]]405 return name, data, {"type": "highlight_projects", "ids": ids}, ""406 # Unreachable for validated calls; degrade to a safe overview.407 return "atlas_overview", repository.overview(), None, ""408 409 def _search(410 self,411 repository: DashboardRepository,412 query: str,413 ) -> tuple[str, dict[str, Any], dict[str, Any] | None, str]:414 data = repository.search(query)415 if not data["results"]:416 return "search_projects", data, None, "no_search_results"417 ids = [result["id"] for result in data["results"]]418 return (419 "search_projects",420 data,421 {"type": "highlight_projects", "ids": ids, "query": query},422 "",423 )424 425 def _done(426 self,427 message: str,428 history: list[dict[str, Any]],429 response: str,430 *,431 tool: str,432 data: dict[str, Any],433 map_action: dict[str, Any] | None,434 ) -> dict[str, Any]:435 new_history = [436 *history,437 {"role": "user", "content": message},438 {"role": "assistant", "content": response},439 ]440 return {441 "type": "done",442 "response": response,443 "tool": tool,444 "data": data,445 "map_action": map_action,446 "history": _normalize_history(new_history),447 }448 449 450def _normalize_history(history: Any) -> list[dict[str, Any]]:451 """Keep only well-formed prior prose turns, clipped, deduplicated, and capped.452 453 Tool digests are deliberately dropped from history: stale counts must never454 leak into a later answer — every turn re-reads a fresh repository snapshot.455 Repeated assistant sentences are collapsed too: a greedy 1B that sees the456 same line twice in history will echo it a third time regardless of the457 digest in front of it."""458 if not isinstance(history, list):459 return []460 cleaned: list[dict[str, Any]] = []461 for item in history:462 if not isinstance(item, dict):463 continue464 role = str(item.get("role") or "")465 content = clean(item.get("content"))466 if role not in ("user", "assistant") or not content:467 continue468 cleaned.append({"role": role, "content": content[:MAX_HISTORY_MESSAGE_CHARS]})469 return _dedupe_assistant_echoes(cleaned)[-MAX_HISTORY_MESSAGES:]470 471 472def _dedupe_assistant_echoes(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:473 """Collapse consecutive identical assistant answers, keeping the NEWEST turn.474 475 Walks backwards so the latest user/assistant pair always survives; the older476 repeats (and the user turns that elicited them) are dropped."""477 deduped_reversed: list[dict[str, Any]] = []478 previous_assistant = None479 skip_next_user = False480 for item in reversed(messages):481 if item["role"] == "assistant":482 if item["content"] == previous_assistant:483 skip_next_user = True484 continue485 previous_assistant = item["content"]486 deduped_reversed.append(item)487 else:488 if skip_next_user:489 skip_next_user = False490 continue491 deduped_reversed.append(item)492 return list(reversed(deduped_reversed))493 494 495def _answer_history(history: list[dict[str, Any]]) -> list[dict[str, Any]]:496 """The short tail of history given to answer generations.497 498 Facts come from the digest, not from history; the prose passes only need499 enough context for follow-ups, and a longer tail mostly adds echo bait."""500 return history[-MAX_ANSWER_HISTORY_MESSAGES:]501 502 503def _digest_for_model(tool: str, data: dict[str, Any]) -> Any:504 """Compact the verified payload into what the model may safely restate.505 506 Beyond stripping urls/ids/scores, long listings are trimmed per tool: a 1B asked507 to repeat ten labels starts blending them, so it only sees the few it may name.508 The UI renders the FULL verified payload independently."""509 trimmed: dict[str, Any] = dict(data)510 if tool == "atlas_overview":511 # Self-describing keys, most-liked first: with three lists in one digest a 1B512 # answering "what's the coolest project" otherwise grabs the wrong column.513 trimmed = {514 "most_liked_projects": data.get("most_liked"),515 "project_count": data.get("project_count"),516 "cluster_count": data.get("cluster_count"),517 "largest_clusters": data.get("top_clusters"),518 "most_completed_quests": data.get("top_quests"),519 "quest_status": data.get("quest_status"),520 }521 if tool == "list_clusters":522 # Ten compound labels is past what a 1B can restate without blending them;523 # it gets the count and the largest cluster, the cards carry the full list.524 clusters = data.get("clusters") or []525 trimmed = {526 "cluster_count": data.get("cluster_count"),527 "largest_cluster": clusters[0] if clusters else None,528 "note": "the full cluster list is already shown to the user as cards",529 }530 if tool == "list_quests":531 quests = data.get("quests") or []532 trimmed = {533 "status": data.get("status"),534 "quest_count": len(quests),535 "most_completed_quest": quests[0] if quests else None,536 "note": "the full quest list is already shown to the user as cards",537 }538 if tool == "show_cluster":539 trimmed["examples"] = (data.get("examples") or [])[:3]540 if tool == "show_quest":541 trimmed["examples"] = (data.get("examples") or [])[:3]542 if tool == "search_projects":543 # BM25 "total" counts any term overlap; quoting it as "N projects about X"544 # would mislead, so the model only sees the close matches themselves.545 trimmed.pop("total", None)546 return _strip_digest_keys(trimmed)547 548 549def _strip_digest_keys(data: Any) -> Any:550 if isinstance(data, dict):551 return {552 key: _strip_digest_keys(value)553 for key, value in data.items()554 if key not in _DIGEST_DROPPED_KEYS555 }556 if isinstance(data, list):557 return [_strip_digest_keys(item) for item in data]558 return data559 560 561def render_digest(data: Any, indent: int = 0) -> str:562 """Render the digest as plain ``key: value`` lines instead of JSON.563 564 A 1B model copying labels out of nested JSON starts blending adjacent strings;565 one fact per line keeps its quotes literal."""566 return "\n".join(_digest_lines(data, indent))567 568 569def _digest_lines(value: Any, indent: int) -> list[str]:570 pad = " " * indent571 if isinstance(value, dict):572 lines: list[str] = []573 for key, item in value.items():574 if isinstance(item, (dict, list)):575 lines.append(f"{pad}{key}:")576 lines.extend(_digest_lines(item, indent + 1))577 else:578 lines.append(f"{pad}{key}: {_digest_value(item)}")579 return lines580 if isinstance(value, list):581 lines = []582 for item in value:583 if isinstance(item, dict):584 flat = ", ".join(585 f"{key}: {_digest_value(entry)}"586 for key, entry in item.items()587 if not isinstance(entry, (dict, list))588 )589 lines.append(f"{pad}- {flat}")590 else:591 lines.append(f"{pad}- {_digest_value(item)}")592 return lines593 return [f"{pad}{_digest_value(value)}"]594 595 596def _digest_value(value: Any) -> Any:597 # Quote strings so compound labels like "Dream / Oracle" keep hard copy598 # boundaries — a greedy 1B blends adjacent unquoted multi-word labels.599 if isinstance(value, str):600 return f'"{value}"'601 return value602 603 604def _templated_sentence(call: ToolCall, data: dict[str, Any], reason: str) -> str:605 """Deterministic sentences for the turns where the model must not improvise."""606 if reason == "quests_not_analyzed":607 return (608 "Quest analysis has not run for this snapshot yet, so quest coverage is empty. "609 "Refresh the map to classify the field, or ask about clusters and projects instead."610 )611 if reason == "unknown_cluster":612 requested = clean(data.get("requested_label")) or "that name"613 return (614 f"I could not find a cluster matching {requested} in the current snapshot. "615 "Ask me to list the clusters to see the live labels."616 )617 if reason == "unknown_quest":618 requested = clean(data.get("requested_quest")) or "that name"619 return (620 f"I could not match {requested} to a hackathon quest. "621 "Ask me to list the quests to see the official names."622 )623 if reason == "quest_no_projects":624 label = clean(data.get("label"))625 if label:626 return f"No project in the current snapshot has completed {label} yet."627 return "No project in the current snapshot has completed that quest yet."628 if reason == "no_leaderboard_rows":629 return (630 "Quest analysis ran, but no project in the current snapshot has completed a "631 "quest yet — the leaderboard is empty."632 )633 if reason == "no_search_results":634 query = clean(data.get("query")) or "that"635 return (636 f"The atlas has no match for {query}. "637 "That can be good news for originality — try a broader term to double-check."638 )639 if reason == "no_clusters" or reason == "no_projects":640 return "The current snapshot has no data for that yet. Try refreshing the map."641 if reason == "empty_answer":642 return "The verified results are on the cards below."643 return "The verified results are on the cards below."644