Spaces:
Running
Running
File size: 2,720 Bytes
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 | """
echo/smoke_test.py
------------------
Runs the FULL agentic pipeline with MockLLM + mock tools — no GPU, no ML deps.
Proves Curator/Screenwriter/Verifier/orchestrator/tree/tools all interoperate.
Run: python -m echo.smoke_test
"""
from __future__ import annotations
from .llm.client import MockLLM
from .tools.research import MockResearch
from .tools.voice import MockVoice
from .core.orchestrator import Orchestrator
def main():
print("=" * 60)
print("The Echo — agentic pipeline smoke test (no ML deps)")
print("=" * 60)
llm = MockLLM(seed=7)
orch = Orchestrator(llm, MockResearch(), MockVoice(), out_dir="/tmp/echo_out")
# 1. seed the tree
seed = "I left Brazil to study abroad instead of staying"
root = orch.seed(seed, base_age=24)
print(f"\n[1] Seed: {seed!r}")
print(f" Root life: {root.facts.constraints_text()}")
print(f" Voice line: {root.voice_line!r}")
print(f" Audio: {root.voice_audio_path}")
print(f" Planned forks: {root.pending_forks}")
# 2. explore several branches (simulate the user navigating the tree)
print("\n[2] Exploring branches...")
cur = root
for step in range(4):
if not cur.pending_forks:
break
child = orch.choose_fork(cur.node_id, fork_index=step % 2, years=5)
print(f" depth {child.depth}: chose {child.divergence!r}")
print(f" -> {child.facts.constraints_text()}")
print(f" tone: {child.tone.dominant_feeling} "
f"(valence {child.tone.valence}, "
f"{'gold' if child.tone.is_flourishing else 'dark'})")
cur = child
# 3. branch into a second direction from root
print("\n[3] Branching the OTHER way from root...")
other = orch.choose_fork(root.node_id, fork_index=1, years=8)
print(f" depth {other.depth}: {other.divergence!r}")
print(f" -> {other.facts.constraints_text()}")
# 4. tree + telemetry
g = orch.graph()
print(f"\n[4] Tree: {len(g['nodes'])} nodes, {len(g['edges'])} edges, "
f"max depth {orch.tree.max_depth}")
print(f" Regeneration rate: {orch.stats.regen_rate:.0%} "
f"({orch.stats.regenerations} regens / "
f"{orch.stats.nodes_grown} grown)")
# 5. closing artifact
print(f"\n[5] Final map:\n {orch.final_map_summary()}")
# 6. verify graph serialization (what Gradio will consume)
assert all("valence" in n for n in g["nodes"]), "graph nodes well-formed"
print(f"\n[6] Graph serialization OK — ready for the front-end.")
print("\n" + "=" * 60)
print("ALL STAGES PASSED ✓ (agentic loop verified)")
print("=" * 60)
if __name__ == "__main__":
main()
|