Spaces:
Running
Running
File size: 2,735 Bytes
1922dbd |
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 |
#!/usr/bin/env python3
"""
Demo: Magentic-One Orchestrator for DeepCritical.
This script demonstrates Phase 5 functionality:
- Multi-Agent Coordination (Searcher + Judge + Manager)
- Magentic-One Workflow
Usage:
export OPENAI_API_KEY=...
uv run python examples/orchestrator_demo/run_magentic.py "metformin cancer"
"""
import argparse
import asyncio
import os
import sys
from src.agent_factory.judges import JudgeHandler
from src.orchestrator_factory import create_orchestrator
from src.tools.pubmed import PubMedTool
from src.tools.search_handler import SearchHandler
from src.tools.websearch import WebTool
from src.utils.models import OrchestratorConfig
async def main() -> None:
"""Run the magentic agent demo."""
parser = argparse.ArgumentParser(description="Run DeepCritical Magentic Agent")
parser.add_argument("query", help="Research query (e.g., 'metformin cancer')")
parser.add_argument("--iterations", type=int, default=10, help="Max rounds")
args = parser.parse_args()
# Check for keys
if not (os.getenv("OPENAI_API_KEY") or os.getenv("ANTHROPIC_API_KEY")):
print("Error: No API key found. Magentic requires real LLM.")
sys.exit(1)
print(f"\n{'='*60}")
print("DeepCritical Magentic Agent Demo")
print(f"Query: {args.query}")
print("Mode: MAGENTIC (Multi-Agent)")
print(f"{ '='*60}\n")
# 1. Setup Search Tools
search_handler = SearchHandler(tools=[PubMedTool(), WebTool()], timeout=30.0)
# 2. Setup Judge
judge_handler = JudgeHandler()
# 3. Setup Orchestrator via Factory
config = OrchestratorConfig(max_iterations=args.iterations)
orchestrator = create_orchestrator(
search_handler=search_handler,
judge_handler=judge_handler,
config=config,
mode="magentic",
)
if not orchestrator:
print("Failed to create Magentic orchestrator. Is agent-framework installed?")
sys.exit(1)
# 4. Run Loop
try:
async for event in orchestrator.run(args.query):
# Print event with icon
# Clean up markdown for CLI
msg_obj = event.message
msg_text = ""
if hasattr(msg_obj, "text"):
msg_text = msg_obj.text
else:
msg_text = str(msg_obj)
msg = msg_text.replace("\n", " ").replace("**", "")[:150]
print(f"[{event.type.upper()}] {msg}...")
if event.type == "complete":
print("\n--- FINAL OUTPUT ---\n")
print(msg_text)
except Exception as e:
print(f"\n❌ Error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(main())
|