Spaces:
Running
Running
File size: 6,227 Bytes
897d5bd 5607db3 897d5bd 5607db3 897d5bd | 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 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """
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
@dataclass
class GrowthStats:
"""Telemetry — powers the Tiny Titan experiment (regeneration rate)."""
nodes_grown: int = 0
regenerations: int = 0
@property
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.")
|