Spaces:
Sleeping
Sleeping
| """ | |
| echo/core/orchestrator.py | |
| ------------------------- | |
| The conductor. Wires Curator + Screenwriter + Verifier + tools into the loop | |
| that grows the tree of alternate lives. | |
| Branch-growth loop (one user choice -> one new node): | |
| 1. Curator reads parent + branch history (+ research grounding) -> child | |
| 2. Verifier audits child vs branch -> regenerate up to N times if needed | |
| 3. Voice tool speaks the child's line | |
| 4. Screenwriter plans the child's next two forks | |
| 5. node added to the tree | |
| The orchestrator is UI-agnostic: the Gradio app calls seed() once, then | |
| choose_fork() each time the user clicks a branch. Everything it returns is | |
| plain data the front-end renders. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from typing import Optional | |
| from .world_state import WorldState, root_state | |
| from .tree import LifeTree | |
| from ..agents.curator import Curator | |
| from ..agents.screenwriter import Screenwriter | |
| from ..agents.verifier import Verifier | |
| from ..tools.research import ResearchTool | |
| from ..tools.voice import VoiceTool | |
| from ..llm.client import LLMClient | |
| class GrowthStats: | |
| """Telemetry — powers the Tiny Titan experiment (regeneration rate).""" | |
| nodes_grown: int = 0 | |
| regenerations: int = 0 | |
| def regen_rate(self) -> float: | |
| denom = self.nodes_grown + self.regenerations | |
| return self.regenerations / denom if denom else 0.0 | |
| class Orchestrator: | |
| def __init__(self, llm: LLMClient, research: ResearchTool, voice: VoiceTool, | |
| out_dir: str = "echo_out", max_regen: int = 2, | |
| base_year: int = 2025): | |
| self.curator = Curator(llm) | |
| self.screenwriter = Screenwriter(llm) | |
| self.verifier = Verifier(llm) | |
| self.research = research | |
| self.voice = voice | |
| self.out_dir = out_dir | |
| self.max_regen = max_regen | |
| self.base_year = base_year | |
| self.tree: Optional[LifeTree] = None | |
| self.stats = GrowthStats() | |
| # ------------------------------------------------------------------ seed | |
| def seed(self, seed_choice: str, base_age: int = 30) -> WorldState: | |
| """Plant the tree: create the root life and its first two forks.""" | |
| self.tree = LifeTree(seed_choice=seed_choice) | |
| root = root_state(seed_choice, base_age) | |
| # The root itself is curated from the seed (depth 0, 0 years elapsed). | |
| grounding = self.research.ground(root.facts.location, self.base_year) | |
| curated = self.curator.grow( | |
| parent=root, | |
| branch_narrative=f"Origin: {seed_choice}", | |
| chosen_fork=seed_choice, | |
| years=0, | |
| grounding=grounding, | |
| ) | |
| curated.parent_id = None | |
| curated.depth = 0 | |
| curated.divergence = seed_choice | |
| self._voice(curated) | |
| curated.pending_forks = self.screenwriter.plan( | |
| curated, self.tree.branch_narrative(curated.node_id) | |
| if curated.node_id in self.tree.nodes else f"Origin: {seed_choice}" | |
| ) | |
| self.tree.add(curated) | |
| # plan forks again now that it's in the tree (branch narrative valid) | |
| curated.pending_forks = self.screenwriter.plan( | |
| curated, self.tree.branch_narrative(curated.node_id)) | |
| self.stats.nodes_grown += 1 | |
| return curated | |
| # ----------------------------------------------------------- grow branch | |
| def choose_fork(self, parent_id: str, fork_index: int, | |
| years: int = 5) -> WorldState: | |
| """User picks one of a node's pending forks -> grow the child node.""" | |
| assert self.tree is not None, "call seed() first" | |
| parent = self.tree.nodes[parent_id] | |
| fork = parent.pending_forks[fork_index] | |
| narrative = self.tree.branch_narrative(parent_id) | |
| child = self._grow_verified(parent, narrative, fork, years) | |
| self._voice(child) | |
| child.pending_forks = self.screenwriter.plan( | |
| child, self.tree.branch_narrative(parent_id) + f"\n-> {fork}") | |
| self.tree.add(child) | |
| self.stats.nodes_grown += 1 | |
| return child | |
| # ----------------------------------------------------- internal helpers | |
| def _grow_verified(self, parent: WorldState, narrative: str, | |
| fork: str, years: int) -> WorldState: | |
| """Curator + Verifier loop with bounded regeneration.""" | |
| location_year = self.base_year + (parent.depth + 1) * years | |
| grounding = self.research.ground(parent.facts.location, location_year) | |
| child = self.curator.grow(parent, narrative, fork, years, grounding) | |
| # Verify EVERY attempt, including the last regeneration. (do-while) | |
| # Breaking out with an unverified child is how a flagged node used to | |
| # slip into the tree silently. | |
| for attempt in range(self.max_regen + 1): | |
| verdict = self.verifier.check(child, parent, narrative) | |
| if verdict.consistent or attempt == self.max_regen: | |
| break | |
| self.stats.regenerations += 1 | |
| # regenerate with the contradiction noted in the grounding hint | |
| child = self.curator.grow( | |
| parent, narrative, fork, years, | |
| grounding=(grounding or "") + | |
| f"\nFIX: previous attempt was inconsistent ({verdict.reason}).") | |
| return child | |
| def _voice(self, state: WorldState) -> None: | |
| if state.voice_line: | |
| state.voice_audio_path = self.voice.speak(state, self.out_dir) | |
| # ----------------------------------------------------------------- views | |
| def graph(self) -> dict: | |
| return self.tree.to_graph() if self.tree else {"nodes": [], "edges": []} | |
| def final_map_summary(self) -> str: | |
| """The closing artifact: a gentle celebration of the lives explored.""" | |
| if not self.tree: | |
| return "" | |
| n = self.tree.size | |
| flourishing = sum(1 for x in self.tree.nodes.values() | |
| if x.tone.is_flourishing) | |
| return (f"You explored {n} lives that all lived inside one choice. " | |
| f"{flourishing} of them were flourishing. " | |
| f"None of them are more real than the one you're in.") | |