Spaces:
Sleeping
Sleeping
File size: 4,449 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 93 94 95 96 97 98 | """Evaluator Agent -> Idea-Observer.
The model acts as an idea-observer. It evaluates the student's ideas against the
source material based on their questions, quiz answers, and Feynman explanations.
It produces a grounded prose observation rather than numeric scores.
"""
from typing import List, Optional
from pydantic import BaseModel, Field
from app.agents.cerebras_client import CerebrasClient
from app.services.journal_service import JournalService
class NodeAssessment(BaseModel):
node_id: str
observation: str = Field(
description="A short cited prose observation about the student's idea or understanding of this concept, and how it held up against the source material."
)
evidence: List[str] = Field(default_factory=list, description="0-3 specific signals/quotes from the journal to ground the observation")
source_evidence_ids: List[str] = Field(
default_factory=list,
description="0-4 exact ev_ IDs from SOURCE EVIDENCE that support the observation",
)
class _EvaluatorOutput(BaseModel):
assessments: List[NodeAssessment]
session_summary: str = Field(description="One-paragraph plain-text summary of the session")
class EvaluatorAgent:
def __init__(
self,
journal_service: Optional[JournalService] = None,
client: Optional[CerebrasClient] = None,
) -> None:
self._client = client or CerebrasClient()
self._journal = journal_service or JournalService()
def evaluate_session(
self,
project_id: str,
prior_context: str = "",
source_context: str = "",
) -> tuple[List[NodeAssessment], str]:
"""Returns (assessments, summary_text).
prior_context is this student's cross-session Cognee memory (from
query_prior_knowledge). The idea-observer reads it so each observation is
judged against the TRAJECTORY -> "continues to conflate X as in an earlier
session" / "has moved from what-is to why questions here" -> not just the
current session's journal in isolation. This is what makes the Evaluator a
memory READER, one of Tier 1A's load-bearing tenets, not a writer alone.
"""
journal = self._journal.get_session(project_id)
if not journal:
return [], "No activity recorded in this session."
journal_text = "\n".join(
f"[{e.event_type}] node={e.node_id} data={e.data}" for e in journal
)
trajectory_note = ""
if prior_context:
trajectory_note = (
"\n\nPRIOR-SESSION MEMORY OF THIS STUDENT (judge against this trajectory, don't "
"restate it): note where an idea has PERSISTED, DEEPENED, or been CORRECTED since "
"earlier sessions, and call out recurring gaps explicitly.\n"
f"{prior_context}"
)
messages = [
{
"role": "system",
"content": (
"You are an idea-observer. For each concept node in the journal, "
"evaluate the student's ideas and understanding against the source material based "
"on their questions, quiz answers, and Feynman explanations. "
"Write a short cited prose observation about their idea and how it held up. "
"Do NOT grade the student or output numeric scores. "
"Focus on the ideas themselves (e.g., 'Proposed X analogy; holds up well but breaks on Y'). "
"For every paper-grounded judgment, include only exact ev_ identifiers from the "
"provided SOURCE EVIDENCE in source_evidence_ids; never invent an ID. Journal evidence "
"and source evidence are separate. Also write a one-paragraph session_summary capturing "
"their overall learning trajectory."
),
},
{
"role": "user",
"content": (
f"Session journal:\n{journal_text}{trajectory_note}\n\n"
f"SOURCE EVIDENCE FROM UPLOADED PAPERS:\n{source_context or 'No source evidence available.'}\n\n"
"Observe each node's ideas."
),
},
]
output = self._client.structured_complete(messages, _EvaluatorOutput)
return output.assessments, output.session_summary
|