File size: 3,659 Bytes
38b4eff | 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 | import asyncio
import aiohttp
import os
import json
# Configuration
MODEL_NAME = "byok:97f38ab9-cce0-48fb-bfb0-f956ec13f723"
API_URL = "https://api.zo.computer/zo/ask"
TOKEN = os.environ.get("ZO_CLIENT_IDENTITY_TOKEN")
async def spawn_worker(session, task_id, prompt):
"""
Spawns a child Zo agent (Worker) to handle a specific sub-task.
"""
print(f"[Manager] Dispatching Worker {task_id}: {prompt[:50]}...")
payload = {
"input": f"TASK_ID: {task_id}\n\n{prompt}",
"model_name": MODEL_NAME,
}
headers = {
"authorization": TOKEN,
"content-type": "application/json",
"Accept": "application/json",
}
try:
async with session.post(API_URL, headers=headers, json=payload) as resp:
if resp.status == 200:
data = await resp.json()
print(f"[Worker {task_id}] Completed successfully.")
return {"id": task_id, "output": data.get("output"), "status": "success"}
else:
print(f"[Worker {task_id}] Failed with status {resp.status}")
return {"id": task_id, "output": None, "status": "error"}
except Exception as e:
print(f"[Worker {task_id}] Exception: {str(e)}")
return {"id": task_id, "output": None, "status": "exception"}
async def manager_orchestrate(main_goal):
"""
The Manager implementation of the Orchestrator logic.
"""
print(f"🚀 Starting Orchestration for Goal: {main_goal}\n")
# 1. Goal Decomposition (In a real scenario, another LLM call would generate this list)
# For the prototype, we manually decompose a sample goal: "Research the transition from
# Ancient Civilizations to Classical Antiquity and identify key technological shifts."
tasks = [
{"id": "worker_1", "prompt": "Research the key characteristics of the Greek Archaic period and the rise of the City-State (Polis)."},
{"id": "worker_2", "prompt": "Analyze the technological shifts in metallurgy and warfare during the transition to Classical Antiquity."},
{"id": "worker_3", "prompt": "Summarize the influence of Phoenician trade on the spread of the alphabet and early Mediterranean diplomacy."}
]
async with aiohttp.ClientSession() as session:
# 2. Parallel Worker Dispatch
print(f"[Manager] Spawning {len(tasks)} Workers in parallel...")
worker_futures = [spawn_worker(session, t["id"], t["prompt"]) for t in tasks]
results = await asyncio.gather(*worker_futures)
# 3. Synthesis
print("\n[Manager] All Workers returned. Synthesizing results...")
final_report = "## 🏛️ Orchestrated Synthesis: Transition to Classical Antiquity\n\n"
for res in results:
if res["status"] == "success":
final_report += f"### Result from {res['id']}\n{res['output']}\n\n"
else:
final_report += f"### Result from {res['id']}\n[Worker failed to provide data]\n\n"
return final_report
if __name__ == "__main__":
if not TOKEN:
print("Error: ZO_CLIENT_IDENTITY_TOKEN not found in environment.")
else:
goal = "Research the transition from Ancient Civilizations to Classical Antiquity."
final_output = asyncio.run(manager_orchestrate(goal))
# Save the synthesized result to the workspace
output_path = "/home/workspace/Projects/NeuralAI/orchestrator_prototype_result.md"
with open(output_path, "w") as f:
f.write(final_output)
print(f"\n✅ Orchestration complete. Synthesis saved to: {output_path}")
|