Spaces:
Running
Running
| from __future__ import annotations | |
| from typing import List, Optional | |
| from pydantic import BaseModel | |
| from app.agents.cerebras_client import CerebrasClient | |
| from app.schemas.graph import HTML5VisualPayload, TextRefContent | |
| _MAX_CHUNK_CHARS = 3000 # mirrors ModalityRouter._MAX_CHUNK_CHARS / BrainAgent.extract_curriculum cap | |
| _DEFAULT_DECLINE_REASON = "This concept doesn't have a narrative worth a dedicated explainer — better explored in chat." | |
| class TextRefGrounding(BaseModel): | |
| renderable: bool | |
| anchor: str = "" | |
| body_markdown: str = "" | |
| decline_reason: str = "" | |
| class TextRefEngine: | |
| """Two-stage grounded markdown-explainer extraction: extract (with a self-checked | |
| verbatim anchor) then render into an HTML5VisualPayload. Mirrors FormulaEngine's | |
| extract/generate split and anchor self-check discipline.""" | |
| def __init__(self, client: Optional[CerebrasClient] = None) -> None: | |
| self._client = client or CerebrasClient() | |
| def generate( | |
| self, | |
| concept: str, | |
| chunks: List[dict], | |
| familiarity: str, | |
| web_results: Optional[List[dict]] = None, | |
| ) -> HTML5VisualPayload: | |
| from app.agents.tutor_agent import TutorAgent | |
| grounding = self._extract(concept, chunks, familiarity) | |
| if not grounding.renderable: | |
| reason = grounding.decline_reason or _DEFAULT_DECLINE_REASON | |
| return HTML5VisualPayload( | |
| html_code=TutorAgent._decline_html(concept, reason), | |
| animation_type="2d_text", | |
| explanation=reason, | |
| ) | |
| source_label = "" | |
| source_url = "" | |
| if web_results: | |
| source_label = web_results[0].get("title", "") | |
| source_url = web_results[0].get("url", "") | |
| return HTML5VisualPayload( | |
| html_code="", | |
| animation_type="2d_text", | |
| text_ref=TextRefContent( | |
| body_markdown=grounding.body_markdown, | |
| source_label=source_label, | |
| source_url=source_url, | |
| ), | |
| explanation=(f"From the source: {grounding.anchor}" if grounding.anchor else ""), | |
| source="paper" if grounding.anchor else ("web" if web_results else "model_knowledge"), | |
| ) | |
| def _extract(self, concept: str, chunks: List[dict], familiarity: str) -> TextRefGrounding: | |
| chunk_text = "\n\n".join(f"[Source: {c.get('source', '?')}]: {c['text']}" for c in chunks)[:_MAX_CHUNK_CHARS] | |
| messages = [ | |
| {"role": "system", "content": ( | |
| "You write a grounded markdown explainer for a concept from the SOURCE MATERIAL only. " | |
| "Write `body_markdown` using ONLY facts, claims, and phrasing actually present in the " | |
| "source — never invent or estimate anything. Reference at most the single most relevant " | |
| "source in prose, woven naturally into the markdown (e.g. \"as covered in...\") — do NOT " | |
| "produce a per-chunk citation list or any per-claim citation array. Also copy a VERBATIM " | |
| "30-80 character substring of the source text as `anchor` — it must match the source text " | |
| "literally, not a paraphrase. If the concept is too abstract or there is nothing in the " | |
| "source worth a dedicated markdown explainer, set renderable=false and give a short, calm, " | |
| "student-facing `decline_reason` (e.g. " | |
| f"\"'{concept}' doesn't have a narrative worth a dedicated explainer — better explored in " | |
| "chat.\").\n\n" | |
| f"SOURCE MATERIAL:\n{chunk_text}" | |
| )}, | |
| {"role": "user", "content": f"Concept: '{concept}' (level: {familiarity}). Write the explainer or decline."}, | |
| ] | |
| grounding = self._client.structured_complete(messages, TextRefGrounding, reasoning_effort="medium") | |
| if grounding.anchor and grounding.anchor.strip() not in chunk_text: | |
| grounding = grounding.model_copy(update={"anchor": ""}) | |
| return grounding | |