Spaces:
Sleeping
Sleeping
File size: 1,627 Bytes
b85de73 | 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 | from __future__ import annotations
from typing import Literal
from langgraph.graph import END, START, StateGraph
from .agents import orchestrator_node, requirements_analyst_node, reviewer_node, test_designer_node
from .state import QAState
def route_from_orchestrator(state: QAState) -> Literal["finish", "requirements_analyst", "test_designer", "reviewer", "fail"]:
decision = state.get("orchestration_decision", {})
if state.get("errors") and decision.get("decision") != "backtrack":
return "fail"
if decision.get("decision") == "finish":
return "finish"
if decision.get("decision") == "backtrack":
target = decision.get("target_agent")
if target in {"requirements_analyst", "test_designer", "reviewer"}:
return target
return "fail"
def build_graph():
graph = StateGraph(QAState)
graph.add_node("requirements_analyst", requirements_analyst_node)
graph.add_node("test_designer", test_designer_node)
graph.add_node("reviewer", reviewer_node)
graph.add_node("orchestrator", orchestrator_node)
graph.add_edge(START, "requirements_analyst")
graph.add_edge("requirements_analyst", "test_designer")
graph.add_edge("test_designer", "reviewer")
graph.add_edge("reviewer", "orchestrator")
graph.add_conditional_edges(
"orchestrator",
route_from_orchestrator,
{
"finish": END,
"requirements_analyst": "requirements_analyst",
"test_designer": "test_designer",
"reviewer": "reviewer",
"fail": END,
},
)
return graph.compile()
|