Spaces:
Sleeping
Sleeping
File size: 4,080 Bytes
2e818da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | 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
|