-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdebug_workflow.py
More file actions
61 lines (49 loc) · 2.28 KB
/
debug_workflow.py
File metadata and controls
61 lines (49 loc) · 2.28 KB
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
#!/usr/bin/env python3
import os
from src.app.flows_planner import build_planner_flow
from src.core.workflow_manager import WorkflowManager
# Build flow
graph, agents, node_policies = build_planner_flow(
executor_agent_name="Executor",
executor_model_config={"fail_once": True},
retry_limit=2,
planner_model_config={
"executor_agent": "Executor",
"model": os.getenv("OLLAMA_MODEL", "gemma3:latest"),
"options": {"temperature": 0.1}
}
)
print("=== FLOW DEBUG ===")
print(f"Graph: {graph}")
print(f"Agents: {list(agents.keys())}")
print(f"Node policies: {node_policies}")
# Test just the Planner first
print("\n=== TESTING PLANNER ALONE ===")
from src.core.types import Message
planner = agents["Planner"]
planner_msg = Message(data={"request": "Build a simple React app"})
planner_result = planner.run(planner_msg)
print(f"Planner result success: {planner_result.success}")
print(f"Planner result control: {planner_result.control}")
print(f"Planner output keys: {list(planner_result.output.keys()) if isinstance(planner_result.output, dict) else 'not dict'}")
if isinstance(planner_result.output, dict) and "plan_meta" in planner_result.output:
print("✅ Planner output has plan_meta - should work with Updater")
# Test Updater manually
print("\n=== TESTING UPDATER WITH PLANNER OUTPUT ===")
updater = agents["Updater"]
updater_msg = Message(data=planner_result.output)
updater_result = updater.run(updater_msg)
print(f"Updater result success: {updater_result.success}")
print(f"Updater result control: {updater_result.control}")
print(f"Updater output keys: {list(updater_result.output.keys()) if isinstance(updater_result.output, dict) else 'not dict'}")
else:
print("❌ Planner output missing plan_meta")
# Now test the full workflow
print("\n=== TESTING FULL WORKFLOW ===")
wm = WorkflowManager(graph=graph, agents=agents, node_policies=node_policies)
results = wm.run_workflow("Planner", {"request": "Build a simple React app"})
print(f"\nWorkflow results count: {len(results)}")
for i, result in enumerate(results):
print(f"Result {i+1}: success={result.success}, display='{result.display_output}'")
if hasattr(result, 'metrics') and 'agent' in result.metrics:
print(f" Agent: {result.metrics['agent']}")