Spaces:
Sleeping
Sleeping
| """Graph Curator Agent. | |
| A loop-engineered agent that wakes on COMMIT_PROJECT (Push). | |
| It reads the uncommitted journal (or recently explored topics), evaluates if | |
| new topics were genuinely explored that are missing from the graph, and spawns | |
| distinct new nodes to represent them. | |
| """ | |
| from typing import List, Optional, Any, Dict | |
| from pydantic import BaseModel, Field | |
| from app.agents.cerebras_client import CerebrasClient | |
| from app.services.journal_service import JournalService | |
| class SpawndNode(BaseModel): | |
| node_id: str = Field(description="A unique id for the new node, e.g., 'n_curator_1'") | |
| label: str = Field(description="The short name of the new topic (<=6 words)") | |
| description: str = Field(description="A brief description of the topic") | |
| parent_id: str = Field(description="The ID of the existing node this should attach to") | |
| class _CuratorOutput(BaseModel): | |
| new_nodes: List[SpawndNode] = Field(description="List of new nodes to add to the graph") | |
| class GraphCuratorAgent: | |
| 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 curate( | |
| self, | |
| project_id: str, | |
| current_nodes: List[Dict[str, Any]], | |
| prior_context: str = "", | |
| ) -> List[SpawndNode]: | |
| """Reads the journal and current graph, and returns SpawndNode specs for | |
| genuinely-explored topics missing from the graph. The caller (COMMIT_PROJECT | |
| handler) is responsible for turning each into a real NodeData and streaming | |
| GRAPH_NODE_ADDED/GRAPH_EDGE_ADDED so it renders live. | |
| prior_context is the student's Cognee memory -> the Curator judges GENUINE | |
| engagement (an idea actually developed) against it, not just "an event fired", | |
| so a passing mention doesn't spawn a node. | |
| """ | |
| journal = self._journal.get_session(project_id) | |
| if not journal: | |
| return [] | |
| # Find nodes the user actually discussed/studied | |
| journal_text = "\n".join( | |
| f"[{e.event_type}] node={e.node_id} data={e.data}" for e in journal | |
| ) | |
| nodes_desc = "\n".join( | |
| f"- {n.get('id')}: \"{n.get('label')}\"" for n in current_nodes | |
| ) | |
| messages = [ | |
| { | |
| "role": "system", | |
| "content": ( | |
| "You are the Graph Curator. Your job is to read the student's recent activity journal, " | |
| "and determine if they have genuinely explored any NEW topics that are NOT currently in the graph.\n" | |
| "If they have, spawn new nodes for those topics. " | |
| "Each new node MUST specify an existing parent_id from the current graph to attach to.\n" | |
| "Only spawn nodes for genuinely engaged topics, not passing mentions." | |
| ), | |
| }, | |
| { | |
| "role": "user", | |
| "content": ( | |
| f"CURRENT GRAPH NODES:\n{nodes_desc}\n\n" | |
| f"RECENT JOURNAL ACTIVITY:\n{journal_text}\n\n" | |
| + (f"STUDENT MEMORY (use to judge genuine engagement):\n{prior_context}\n\n" if prior_context else "") | |
| + "Spawn new nodes ONLY for topics they genuinely developed an idea about, " | |
| "not passing mentions." | |
| ), | |
| }, | |
| ] | |
| try: | |
| output = self._client.structured_complete(messages, _CuratorOutput) | |
| except Exception as exc: | |
| import logging | |
| logging.getLogger(__name__).warning("GraphCurator failed: %s", exc) | |
| return [] | |
| # Defensively drop any spawn whose parent_id doesn't actually exist in the | |
| # current graph -> the model can hallucinate one, which would otherwise | |
| # create an orphaned node with no valid attachment point. | |
| valid_parent_ids = {n.get("id") for n in current_nodes} | |
| return [n for n in output.new_nodes if n.parent_id in valid_parent_ids] | |