Spaces:
Sleeping
Sleeping
File size: 5,120 Bytes
e5e882e |
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 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 |
from typing import Dict, Any, TypedDict, Annotated, Sequence
from langgraph.graph import Graph, StateGraph
from langgraph.prebuilt import ToolExecutor
from agents.business_analyst import BusinessAnalystAgent
from agents.designer import DesignerAgent
from agents.developer import DeveloperAgent
from agents.tester import TesterAgent
from core.database import db
class AgentState(TypedDict):
"""State for the agent workflow"""
requirements: str
user_stories: str
design: str
code: str
test_cases: str
test_results: str
current_step: str
status: str
messages: Sequence[str]
class DevelopmentWorkflow:
def __init__(self):
self.business_analyst = BusinessAnalystAgent()
self.designer = DesignerAgent()
self.developer = DeveloperAgent()
self.tester = TesterAgent()
# Define the workflow
self.workflow = self._create_workflow()
def _create_workflow(self) -> Graph:
"""Create the LangGraph workflow"""
workflow = StateGraph(AgentState)
# Add nodes for each step
workflow.add_node("create_user_stories", self._create_user_stories)
workflow.add_node("create_design", self._create_design)
workflow.add_node("generate_code", self._generate_code)
workflow.add_node("create_tests", self._create_tests)
workflow.add_node("execute_tests", self._execute_tests)
# Define the edges
workflow.add_edge("create_user_stories", "create_design")
workflow.add_edge("create_design", "generate_code")
workflow.add_edge("generate_code", "create_tests")
workflow.add_edge("create_tests", "execute_tests")
# Set entry point
workflow.set_entry_point("create_user_stories")
# Set exit point
workflow.set_finish_point("execute_tests")
return workflow.compile()
async def _create_user_stories(self, state: AgentState) -> AgentState:
"""Create user stories from requirements"""
result = await self.business_analyst.create_user_stories(state["requirements"])
return {
**state,
"user_stories": result["user_stories"],
"current_step": "create_user_stories",
"status": "in_progress",
"messages": [*state["messages"], "User stories created successfully"]
}
async def _create_design(self, state: AgentState) -> AgentState:
"""Create design from user stories"""
result = await self.designer.create_design(state["user_stories"])
return {
**state,
"design": result["design"],
"current_step": "create_design",
"status": "in_progress",
"messages": [*state["messages"], "Design created successfully"]
}
async def _generate_code(self, state: AgentState) -> AgentState:
"""Generate code from design"""
result = await self.developer.generate_code(state["design"])
return {
**state,
"code": result["code"],
"current_step": "generate_code",
"status": "in_progress",
"messages": [*state["messages"], "Code generated successfully"]
}
async def _create_tests(self, state: AgentState) -> AgentState:
"""Create test cases from code"""
result = await self.tester.create_test_cases(state["code"])
return {
**state,
"test_cases": result["test_cases"],
"current_step": "create_tests",
"status": "in_progress",
"messages": [*state["messages"], "Test cases created successfully"]
}
async def _execute_tests(self, state: AgentState) -> AgentState:
"""Execute tests on the code"""
result = await self.tester.execute_tests(state["code"], state["test_cases"])
return {
**state,
"test_results": result["test_results"],
"current_step": "execute_tests",
"status": "completed",
"messages": [*state["messages"], "Tests executed successfully"]
}
async def run(self, requirements: str) -> Dict[str, Any]:
"""Run the complete workflow"""
initial_state = AgentState(
requirements=requirements,
user_stories="",
design="",
code="",
test_cases="",
test_results="",
current_step="start",
status="initialized",
messages=[]
)
final_state = await self.workflow.ainvoke(initial_state)
return {
"status": "success",
"user_stories": final_state["user_stories"],
"design": final_state["design"],
"code": final_state["code"],
"test_results": final_state["test_results"],
"messages": final_state["messages"]
} |