DevPilot

app.py

@@ -0,0 +1,7 @@
+import argparse
+import subprocess
+import sys
+from src.dev_pilot.ui.streamlit_ui.streamlit_app import load_app
+
+if __name__ == "__main__":
+     load_app()

app_api.py

@@ -0,0 +1,4 @@
+from src.dev_pilot.api.fastapi_app import load_app
+
+if __name__ == "__main__":
+     load_app()

requirements.txt

@@ -0,0 +1,14 @@
+langchain
+langgraph
+langchain_community
+langchain_core
+langchain_groq
+langchain_openai
+faiss_cpu
+streamlit
+langchain-google-genai
+redis
+upstash_redis
+fastapi
+uvicorn
+loguru

src/dev_pilot/LLMS/geminillm.py

@@ -0,0 +1,24 @@
+import os
+import streamlit as st
+from langchain_google_genai import ChatGoogleGenerativeAI
+
+
+class GeminiLLM:
+    def __init__(self, user_controls_input=None, model=None, api_key=None):
+        self.user_controls_input = user_controls_input
+        self.model = model
+        self.api_key = api_key
+        
+    def get_llm_model(self):
+        try:
+            if  self.user_controls_input:
+                gemini_api_key = self.user_controls_input['GEMINI_API_KEY']
+                selected_gemini_model = self.user_controls_input['selected_gemini_model']
+                llm = ChatGoogleGenerativeAI(api_key=gemini_api_key, model= selected_gemini_model)
+            else:
+                llm = ChatGoogleGenerativeAI(api_key=self.api_key,model=self.model)
+        
+        except Exception as e:
+            raise ValueError(f"Error occured with Exception : {e}")
+        
+        return llm

src/dev_pilot/LLMS/groqllm.py

@@ -0,0 +1,25 @@
+import os
+import streamlit as st
+from langchain_groq import ChatGroq
+
+
+class GroqLLM:
+    def __init__(self, user_controls_input=None, model=None, api_key=None):
+        self.user_controls_input = user_controls_input
+        self.model = model
+        self.api_key = api_key
+        
+        
+    def get_llm_model(self):
+        try:
+            if  self.user_controls_input:
+                groq_api_key = self.user_controls_input['GROQ_API_KEY']
+                selected_groq_model = self.user_controls_input['selected_groq_model']
+                llm = ChatGroq(api_key=groq_api_key, model= selected_groq_model)
+            else:
+                llm = ChatGroq(api_key=self.api_key,model=self.model)
+        
+        except Exception as e:
+            raise ValueError(f"Error occured with Exception : {e}")
+        
+        return llm

src/dev_pilot/LLMS/openai_llm.py

@@ -0,0 +1,25 @@
+import os
+import streamlit as st
+from langchain_openai import ChatOpenAI
+
+
+class OpenAILLM:
+    def __init__(self, user_controls_input=None, model=None, api_key=None):
+        self.user_controls_input = user_controls_input
+        self.model = model
+        self.api_key = api_key
+        
+        
+    def get_llm_model(self):
+        try:
+            if  self.user_controls_input:
+                openai_api_key = self.user_controls_input['OPENAI_API_KEY']
+                selected_openai_model = self.user_controls_input['selected_openai_model']
+                llm = ChatOpenAI(api_key=openai_api_key, model= selected_openai_model)
+            else:
+                llm = ChatOpenAI(api_key=openai_api_key, model= self.model)
+
+        except Exception as e:
+            raise ValueError(f"Error occured with Exception : {e}")
+        
+        return llm

src/dev_pilot/api/fastapi_app.py

@@ -0,0 +1,202 @@
+from fastapi import FastAPI, HTTPException, Depends, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+import os
+from dotenv import load_dotenv
+from functools import lru_cache
+from src.dev_pilot.LLMS.groqllm import GroqLLM
+from src.dev_pilot.LLMS.geminillm import GeminiLLM
+from src.dev_pilot.graph.graph_builder import GraphBuilder
+from src.dev_pilot.graph.graph_executor import GraphExecutor
+from src.dev_pilot.dto.sdlc_request import SDLCRequest
+from src.dev_pilot.dto.sdlc_response import SDLCResponse
+import uvicorn
+from contextlib import asynccontextmanager
+from src.dev_pilot.utils.logging_config import setup_logging
+from loguru import logger
+
+## Setup logging level
+setup_logging(log_level="DEBUG")
+
+gemini_models = [
+    "gemini-2.0-flash",
+    "gemini-2.0-flash-lite", 
+    "gemini-2.5-pro-exp-03-25"
+]
+
+groq_models = [
+    "gemma2-9b-it",
+    "llama3-8b-8192",
+    "llama3-70b-8192"
+]
+
+def load_app():
+     uvicorn.run(app, host="0.0.0.0", port=8000)
+
+class Settings:
+    def __init__(self):
+        self.GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
+        self.GROQ_API_KEY = os.getenv("GROQ_API_KEY")        
+
+@lru_cache()
+def get_settings():
+    return Settings()
+
+def validate_api_keys(settings: Settings = Depends(get_settings)):
+    required_keys = {
+        'GEMINI_API_KEY': settings.GEMINI_API_KEY,
+        'GROQ_API_KEY': settings.GROQ_API_KEY
+    }
+    
+    missing_keys = [key for key, value in required_keys.items() if not value]
+    if missing_keys:
+        raise HTTPException(
+            status_code=500,
+            detail=f"Missing required API keys: {', '.join(missing_keys)}"
+        )
+    return settings
+
+
+# Initialize the LLM and GraphBuilder instances once and store them in the app state
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    settings = get_settings()
+    llm = GeminiLLM(model=gemini_models[0], api_key=settings.GEMINI_API_KEY).get_llm_model()
+    graph_builder = GraphBuilder(llm=llm)
+    graph = graph_builder.setup_graph()
+    graph_executor = GraphExecutor(graph)
+    app.state.llm = llm
+    app.state.graph = graph
+    app.state.graph_executor = graph_executor
+    yield
+    # Clean up resources if needed
+    app.state.llm = None
+    app.state.graph = None
+    app.state.graph_executor = None
+
+app = FastAPI(
+    title="DevPilot API",
+    description="AI-powered SDLC API using Langgraph",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+logger.info("Application starting up...")
+
+# Configure CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # In production, replace with specific origins
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/")
+async def root():
+    return {
+        "message": "Welcome to DevPilot API",
+        "docs_url": "/docs",
+        "redoc_url": "/redoc"
+    }
+
+@app.post("/api/v1/sdlc/start", response_model=SDLCResponse)
+async def start_sdlc(
+    sdlc_request: SDLCRequest,
+    settings: Settings = Depends(validate_api_keys)
+    ):
+
+    try:
+        graph_executor = app.state.graph_executor
+        
+        if isinstance (graph_executor, GraphExecutor) == False:
+            raise Exception("Graph Executor not initialized")
+        
+        graph_response = graph_executor.start_workflow(sdlc_request.project_name)
+        
+        logger.debug(f"Start Workflow Response: {graph_response}")
+        
+        return SDLCResponse(
+            status="success",
+            message="SDLC process started successfully",
+            task_id=graph_response["task_id"],
+            state=graph_response["state"]
+        )
+    
+    except Exception as e:
+        error_response = SDLCResponse(
+            status="error",
+            message="Failed to start the process",
+            error=str(e)
+        )
+        return JSONResponse(status_code=500, content=error_response.model_dump())
+    
+    
+@app.post("/api/v1/sdlc/user_stories", response_model=SDLCResponse)
+async def start_sdlc(
+    sdlc_request: SDLCRequest,
+    settings: Settings = Depends(validate_api_keys)
+    ):
+
+    try:
+        graph_executor = app.state.graph_executor
+        
+        if isinstance (graph_executor, GraphExecutor) == False:
+            raise Exception("Graph Executor not initialized")
+        
+        graph_response = graph_executor.generate_stories(sdlc_request.task_id, sdlc_request.requirements)
+        
+        logger.debug(f"Generate Stories Response: {graph_response}")
+        
+        return SDLCResponse(
+            status="success",
+            message="User Stories generated successfully",
+            task_id=graph_response["task_id"],
+            state=graph_response["state"]
+        )
+    
+    except Exception as e:
+        error_response = SDLCResponse(
+            status="error",
+            message="Failed to generate user stories",
+            error=str(e)
+        )
+        return JSONResponse(status_code=500, content=error_response.model_dump())
+    
+
+@app.post("/api/v1/sdlc/progress_flow", response_model=SDLCResponse)
+async def progress_sdlc(
+    sdlc_request: SDLCRequest,
+    settings: Settings = Depends(validate_api_keys)
+    ):
+
+    try:
+
+        graph_executor = app.state.graph_executor
+        
+        if isinstance (graph_executor, GraphExecutor) == False:
+            raise Exception("Graph Executor not initialized")
+        
+        graph_response = graph_executor.graph_review_flow(
+            sdlc_request.task_id, 
+            sdlc_request.status, 
+            sdlc_request.feedback,
+            sdlc_request.next_node)
+        
+        logger.debug(f"Flow Node: {sdlc_request.next_node}")
+        logger.debug(f"Progress Flow Response: {graph_response}")
+        
+        return SDLCResponse(
+            status="success",
+            message="Flow progressed successfully to next step",
+            task_id=graph_response["task_id"],
+            state=graph_response["state"]
+        )
+    
+    except Exception as e:
+        error_response = SDLCResponse(
+            status="error",
+            message="Failed to progress the flow",
+            error=str(e)
+        )
+        return JSONResponse(status_code=500, content=error_response.model_dump())

src/dev_pilot/cache/redis_cache.py

@@ -0,0 +1,54 @@
+import redis
+import json
+from typing import Optional
+from src.dev_pilot.state.sdlc_state import CustomEncoder, SDLCState
+from upstash_redis import Redis
+import os
+from dotenv import load_dotenv
+from loguru import logger
+
+load_dotenv()
+
+
+# Initialize Redis client
+
+## Upstash Redis Client Configuraion
+REDIS_URL = os.getenv("REDIS_URL")
+REDIS_TOKEN = os.getenv("REDIS_TOKEN")
+redis_client = redis = Redis(url=REDIS_URL, token=REDIS_TOKEN)
+
+## For testing locally with docker
+# redis_client = redis.Redis(
+#     host='localhost',  # Replace with your Redis host
+#     port=6379,         # Replace with your Redis port
+#     db=0               # Replace with your Redis database number
+# )
+
+def save_state_to_redis(task_id: str, state: SDLCState):
+    """Save the state to Redis."""
+    state = json.dumps(state, cls=CustomEncoder)
+    redis_client.set(task_id, state)
+
+    # Set expiration for 24 hours
+    redis_client.expire(task_id, 86400)
+
+def get_state_from_redis(task_id: str) -> Optional[SDLCState]:
+    """ Retrieves the state from redis """
+    state_json = redis_client.get(task_id)
+    if not state_json:
+        return None
+    
+    state_dict = json.loads(state_json)[0]
+    return SDLCState(**state_dict)
+
+def delete_from_redis(task_id: str):
+    """ Delete from redis """
+    redis_client.delete(task_id)
+
+def flush_redis_cache():
+    """ Flushes the whole cache"""
+
+    # Clear all keys in all databases
+    redis_client.flushall()
+
+    logger.info("--- Redis cache cleared ---")

src/dev_pilot/dto/sdlc_request.py

@@ -0,0 +1,29 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+
+class SDLCRequest(BaseModel):
+    project_name: str = Field(..., 
+                              example="Ecommerce Platform",
+                              description="The name of the project")
+    
+    requirements: Optional[list[str]] = Field(None, 
+                                                example=["Users can browser the products", 
+                                                         "Users should be able to add the product in the cart",
+                                                         "Users should be able to do the payment",
+                                                         "Users should be able to see their order history"],
+                                                description="The list of requirements for the project")
+    task_id: Optional[str] = Field(None, 
+                                    example="sdlc-session-5551defc",
+                                    description="The task id of the workflow session")
+    
+    next_node: Optional[str] = Field(None, 
+                                example="review_user_stories",
+                                description="The node to be executed in the workflow. Pass the node information returned from previous API")
+    
+    status: Optional[str] = Field(None, 
+                                example="approved or feedback",
+                                description="The status of the review")
+    
+    feedback: Optional[str] = Field(None, 
+                                example="The user stories are good but need to be more specific",
+                                description="The feedback for the review")

src/dev_pilot/dto/sdlc_response.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel
+from typing import Optional
+from src.dev_pilot.state.sdlc_state import SDLCState
+from typing import Dict, Any
+
+class SDLCResponse(BaseModel):
+    status: str
+    message: str
+    task_id: Optional[str] = None
+    state: Optional[Dict[str, Any]] = None
+    error: Optional[str] = None

src/dev_pilot/graph/graph_builder.py

@@ -0,0 +1,174 @@
+from langgraph.graph import StateGraph,START, END
+from src.dev_pilot.state.sdlc_state import SDLCState
+from src.dev_pilot.nodes.project_requirement_node import ProjectRequirementNode
+from src.dev_pilot.nodes.design_document_node import DesingDocumentNode
+from src.dev_pilot.nodes.coding_node import CodingNode
+from src.dev_pilot.nodes.markdown_node import MarkdownArtifactsNode
+from langgraph.checkpoint.memory import MemorySaver
+from langchain_core.runnables.graph import MermaidDrawMethod
+
+class GraphBuilder:
+    
+    def __init__(self, llm):
+        self.llm = llm
+        self.graph_builder = StateGraph(SDLCState)
+        self.memory = MemorySaver()
+                
+    
+    def build_sdlc_graph(self):
+        """
+            Configure the graph by adding nodes, edges
+        """
+        
+        self.project_requirement_node = ProjectRequirementNode(self.llm)
+        self.design_document_node = DesingDocumentNode(self.llm)
+        self.coding_node = CodingNode(self.llm)
+        self.markdown_node = MarkdownArtifactsNode()
+        
+        ## Nodes
+        self.graph_builder.add_node("initialize_project", self.project_requirement_node.initialize_project)
+        self.graph_builder.add_node("get_user_requirements", self.project_requirement_node.get_user_requirements)
+        
+        self.graph_builder.add_node("generate_user_stories", self.project_requirement_node.generate_user_stories)
+        self.graph_builder.add_node("review_user_stories", self.project_requirement_node.review_user_stories)
+        self.graph_builder.add_node("revise_user_stories", self.project_requirement_node.revise_user_stories)
+        
+        self.graph_builder.add_node("create_design_documents", self.design_document_node.create_design_documents)
+        self.graph_builder.add_node("review_design_documents", self.design_document_node.review_design_documents)
+        self.graph_builder.add_node("revise_design_documents", self.design_document_node.revise_design_documents)
+        
+        self.graph_builder.add_node("generate_code", self.coding_node.generate_code)
+        self.graph_builder.add_node("code_review", self.coding_node.code_review)
+        self.graph_builder.add_node("fix_code", self.coding_node.fix_code)
+        
+        self.graph_builder.add_node("security_review_recommendations", self.coding_node.security_review_recommendations)
+        self.graph_builder.add_node("security_review", self.coding_node.security_review)
+        self.graph_builder.add_node("fix_code_after_security_review", self.coding_node.fix_code_after_security_review)
+       
+        self.graph_builder.add_node("write_test_cases", self.coding_node.write_test_cases)
+        self.graph_builder.add_node("review_test_cases", self.coding_node.review_test_cases)
+        self.graph_builder.add_node("revise_test_cases", self.coding_node.revise_test_cases)
+        
+        self.graph_builder.add_node("qa_testing", self.coding_node.qa_testing)
+        self.graph_builder.add_node("qa_review", self.coding_node.qa_review)
+        self.graph_builder.add_node("deployment", self.coding_node.deployment)
+        self.graph_builder.add_node("donwload_artifacts", self.markdown_node.generate_markdown_artifacts)
+        
+        
+        ## Edges
+        self.graph_builder.add_edge(START,"initialize_project")
+        self.graph_builder.add_edge("initialize_project","get_user_requirements")
+        self.graph_builder.add_edge("get_user_requirements","generate_user_stories")
+        self.graph_builder.add_edge("generate_user_stories","review_user_stories") 
+        self.graph_builder.add_conditional_edges(
+            "review_user_stories",
+            self.project_requirement_node.review_user_stories_router,
+            {
+                "approved": "create_design_documents",
+                "feedback": "revise_user_stories"
+            }
+        )
+        self.graph_builder.add_edge("revise_user_stories","generate_user_stories")
+        self.graph_builder.add_edge("create_design_documents","review_design_documents")
+        self.graph_builder.add_conditional_edges(
+            "review_design_documents",
+            self.design_document_node.review_design_documents_router,
+            {
+                "approved": "generate_code",
+                "feedback": "revise_design_documents"
+            }
+        )
+        self.graph_builder.add_edge("revise_design_documents","create_design_documents")
+        self.graph_builder.add_edge("generate_code","code_review")
+        self.graph_builder.add_conditional_edges(
+            "code_review",
+            self.coding_node.code_review_router,
+            {
+                "approved": "security_review_recommendations",
+                "feedback": "fix_code"
+            }
+        )
+        self.graph_builder.add_edge("fix_code","generate_code")
+        self.graph_builder.add_edge("security_review_recommendations","security_review")
+        self.graph_builder.add_conditional_edges(
+            "security_review",
+            self.coding_node.security_review_router,
+            {
+                "approved": "write_test_cases",
+                "feedback": "fix_code_after_security_review"
+            }
+        )
+        self.graph_builder.add_edge("fix_code_after_security_review","generate_code")
+        self.graph_builder.add_edge("write_test_cases", "review_test_cases")
+        self.graph_builder.add_conditional_edges(
+            "review_test_cases",
+            self.coding_node.review_test_cases_router,
+            {
+                "approved": "qa_testing",
+                "feedback": "revise_test_cases"
+            }
+        )
+        self.graph_builder.add_edge("revise_test_cases", "write_test_cases")
+        self.graph_builder.add_edge("qa_testing", "qa_review")
+        self.graph_builder.add_conditional_edges(
+            "qa_review",
+            self.coding_node.review_test_cases_router,
+            {
+                "approved": "deployment",
+                "feedback": "generate_code"
+            }
+        )
+        self.graph_builder.add_edge("deployment", "donwload_artifacts")
+        self.graph_builder.add_edge("donwload_artifacts", END)
+         
+        
+    def setup_graph(self):
+        """
+        Sets up the graph
+        """
+        self.build_sdlc_graph()
+        return self.graph_builder.compile(
+            interrupt_before=[
+                'get_user_requirements',
+                'review_user_stories',
+                'review_design_documents',
+                'code_review',
+                'security_review',
+                'review_test_cases',
+                'qa_review'
+            ],checkpointer=self.memory
+        )
+        
+             
+    # def setup_graph(self):
+    #     """
+    #     Sets up the graph
+    #     """
+    #     self.build_sdlc_graph()
+    #     graph =self.graph_builder.compile(
+    #         interrupt_before=[
+    #             'get_user_requirements',
+    #             'review_user_stories',
+    #             'review_design_documents',
+    #             'code_review',
+    #             'security_review',
+    #             'review_test_cases',
+    #             'qa_review'
+    #         ],checkpointer=self.memory
+    #     )
+    #     self.save_graph_image(graph)         
+    #     return graph
+    
+    
+    def save_graph_image(self,graph):
+        # Generate the PNG image
+        img_data = graph.get_graph().draw_mermaid_png(
+            draw_method=MermaidDrawMethod.API
+            )
+
+        # Save the image to a file
+        graph_path = "workflow_graph.png"
+        with open(graph_path, "wb") as f:
+            f.write(img_data)        
+            
+        

src/dev_pilot/graph/graph_executor.py

@@ -0,0 +1,114 @@
+from src.dev_pilot.state.sdlc_state import SDLCState
+from src.dev_pilot.cache.redis_cache import flush_redis_cache, save_state_to_redis, get_state_from_redis
+import uuid
+import src.dev_pilot.utils.constants as const
+from loguru import logger
+
+class GraphExecutor:
+    def __init__(self, graph):
+        self.graph = graph
+
+    def get_thread(self, task_id):
+        return {"configurable": {"thread_id": task_id}}
+    
+    ## ------- Start the Workflow ------- ##
+    def start_workflow(self, project_name: str):
+        
+        graph = self.graph
+        
+        flush_redis_cache()
+        
+        # Generate a unique task id
+        task_id = f"sdlc-session-{uuid.uuid4().hex[:8]}"
+        
+        thread = self.get_thread(task_id)
+        
+        state = None
+        for event in graph.stream({"project_name": project_name},thread, stream_mode="values"):
+           state = event
+        
+        current_state = graph.get_state(thread)
+        save_state_to_redis(task_id, current_state)
+        
+        return {"task_id" : task_id, "state": state}
+    
+    ## ------- User Story Generation ------- ##
+    def generate_stories(self, task_id:str, requirements: list[str]):
+        saved_state = get_state_from_redis(task_id)
+        if saved_state:
+            saved_state['requirements'] = requirements
+            saved_state['next_node'] = const.REVIEW_USER_STORIES
+        
+        return self.update_and_resume_graph(saved_state,task_id,"get_user_requirements")
+
+
+    ## ------- Generic Review Flow for all the feedback stages  ------- ##
+    def graph_review_flow(self, task_id, status, feedback, review_type):
+        saved_state = get_state_from_redis(task_id)
+        
+        if saved_state:
+            if review_type == const.REVIEW_USER_STORIES:
+                saved_state['user_stories_review_status'] = status
+                saved_state['user_stories_feedback'] = feedback
+                node_name = "review_user_stories"
+                saved_state['next_node'] = const.REVIEW_USER_STORIES if status == "feedback" else const.REVIEW_DESIGN_DOCUMENTS
+                
+            elif review_type == const.REVIEW_DESIGN_DOCUMENTS:
+                saved_state['design_documents_review_status'] = status
+                saved_state['design_documents_feedback'] = feedback
+                node_name = "review_design_documents"
+                saved_state['next_node'] = const.REVIEW_DESIGN_DOCUMENTS if status == "feedback" else const.REVIEW_CODE
+                
+            elif review_type == const.REVIEW_CODE:
+                saved_state['code_review_status'] = status
+                saved_state['code_review_feedback'] = feedback
+                node_name = "code_review"
+                saved_state['next_node'] = const.REVIEW_CODE if status == "feedback" else const.REVIEW_SECURITY_RECOMMENDATIONS
+                
+            elif review_type == const.REVIEW_SECURITY_RECOMMENDATIONS:
+                saved_state['security_review_status'] = status
+                saved_state['security_review_comments'] = feedback
+                node_name = "security_review"   
+                saved_state['next_node'] = const.REVIEW_SECURITY_RECOMMENDATIONS if status == "feedback" else const.REVIEW_TEST_CASES
+                
+            elif review_type == const.REVIEW_TEST_CASES:
+                saved_state['test_case_review_status'] = status
+                saved_state['test_case_review_feedback'] = feedback
+                node_name = "review_test_cases" 
+                saved_state['next_node'] = const.REVIEW_TEST_CASES if status == "feedback" else const.REVIEW_QA_TESTING
+                 
+            elif review_type == const.REVIEW_QA_TESTING:
+                saved_state['qa_testing_status'] = status
+                saved_state['qa_testing_feedback'] = feedback
+                node_name = "qa_review"  
+                saved_state['next_node'] = const.REVIEW_QA_TESTING if status == "feedback" else const.END_NODE
+                 
+            else:
+                raise ValueError(f"Unsupported review type: {review_type}")
+            
+        return self.update_and_resume_graph(saved_state,task_id,node_name)
+        
+    ## -------- Helper Method to handle the graph resume state ------- ##
+    def update_and_resume_graph(self, saved_state,task_id, as_node):
+        graph = self.graph
+        thread = self.get_thread(task_id)
+        
+        graph.update_state(thread, saved_state, as_node=as_node)
+        
+         # Resume the graph
+        state = None
+        for event in graph.stream(None, thread, stream_mode="values"):
+            logger.debug(f"Event Received: {event}")
+            state = event
+        
+        # saving the state before asking the product owner for review
+        current_state = graph.get_state(thread)
+        save_state_to_redis(task_id, current_state)
+        
+        return {"task_id" : task_id, "state": state}
+
+
+    def get_updated_state(self, task_id):
+        saved_state = get_state_from_redis(task_id)
+        return {"task_id" : task_id, "state": saved_state}
+    

src/dev_pilot/nodes/coding_node.py

@@ -0,0 +1,280 @@
+from src.dev_pilot.state.sdlc_state import SDLCState, UserStoryList
+from src.dev_pilot.utils.Utility import Utility
+from loguru import logger
+
+class CodingNode:
+    """
+    Graph Node for the Coding
+    
+    """
+    
+    def __init__(self, model):
+        self.llm = model
+        self.utility = Utility()
+    
+    ## ---- Code Generation ----- ##
+    def generate_code(self, state: SDLCState):
+        """
+            Generates the code for the given SDLC state as multiple Python files.
+        """
+        logger.info("----- Generating the code ----")
+        
+        requirements = state.get('requirements', '')
+        user_stories = state.get('user_stories', '')
+        code_feedback = state.get('code_review_feedback', '') if 'code_generated' in state else ""
+        security_feedback = state.get('security_recommendations', '') if 'security_recommendations' in state else ""
+        
+        prompt = f"""
+        Generate a complete Python project organized as multiple code files. 
+        Based on the following SDLC state, generate only the Python code files with their complete implementations. 
+        Do NOT include any explanations, requirements text, or design document details in the output—only code files with proper names and code content.
+
+        SDLC State:
+        ---------------
+        Project Name: {state['project_name']}
+
+        Requirements:
+        {self.utility.format_list(requirements)}
+
+        User Stories:
+        {self.utility.format_user_stories(user_stories)}
+
+        Functional Design Document:
+        {state['design_documents']['functional']}
+
+        Technical Design Document:
+        {state['design_documents']['technical']}
+
+        {"Note: Incorporate the following code review feedback: " + code_feedback if code_feedback else ""}
+        {"Note: Apply the following security recommendations: " + security_feedback if security_feedback else ""}
+
+        Instructions:
+        - Structure the output as multiple code files (for example, "main.py", "module1.py", etc.), each separated clearly.
+        - Each file should contain only the code necessary for a modular, fully-functional project based on the input state.
+        - Do not output any additional text, explanations, or commentary outside the code files.
+        - Ensure the code follows Python best practices, is syntactically correct, and is ready for development.
+        """
+        response = self.llm.invoke(prompt)
+        code_review_comments = self.get_code_review_comments(code=response.content)
+        return {
+                'code_generated': response.content, 
+                'code_review_comments': code_review_comments
+            }
+    
+    ## This code review comments will be used while generating test cases
+    def get_code_review_comments(self, code: str):
+        """
+        Generate code review comments for the provided code
+        """
+        logger.info("----- Generating code review comments ----")
+        
+        # Create a prompt for the LLM to review the code
+        prompt = f"""
+            You are a coding expert. Please review the following code and provide detailed feedback:
+            ```
+            {code}
+            ```
+            Focus on:
+            1. Code quality and best practices
+            2. Potential bugs or edge cases
+            3. Performance considerations
+            4. Security concerns
+            
+            End your review with an explicit APPROVED or NEEDS_FEEDBACK status.
+        """
+        
+        # Get the review from the LLM
+        response = self.llm.invoke(prompt)
+        review_comments = response.content
+        return review_comments
+    
+    def code_review(self, state: SDLCState):
+        return state
+    
+    def fix_code(self, state: SDLCState):
+        pass
+    
+    def code_review_router(self, state: SDLCState):
+        """
+            Evaluates Code review is required or not.
+        """
+        return state.get("code_review_status", "approved")  # default to "approved" if not present
+    
+    ## ---- Security Review ----- ##
+    def security_review_recommendations(self, state: SDLCState):
+        """
+            Performs security review of the code generated
+        """
+        logger.info("----- Generating security recommendations ----")
+          
+         # Get the generated code from the state
+        code_generated = state.get('code_generated', '')
+
+         # Create a prompt for the LLM to review the code for security concerns
+        prompt = f"""
+            You are a security expert. Please review the following Python code for potential security vulnerabilities:
+            ```
+            {code_generated}
+            ```
+            Focus on:
+            1. Identifying potential security risks (e.g., SQL injection, XSS, insecure data handling).
+            2. Providing recommendations to mitigate these risks.
+            3. Highlighting any best practices that are missing.
+
+            End your review with an explicit APPROVED or NEEDS_FEEDBACK status.
+        """
+
+         # Invoke the LLM to perform the security review
+        response = self.llm.invoke(prompt)
+        state["security_recommendations"] =  response.content
+        return state
+    
+    def security_review(self, state: SDLCState):
+        return state
+    
+    def fix_code_after_security_review(self, state: SDLCState):
+        pass
+    
+    def security_review_router(self, state: SDLCState):
+        """
+            Security Code review is required or not.
+        """
+        return state.get("security_review_status", "approved")  # default to "approved" if not present
+    
+    ## ---- Test Cases ----- ##
+    def write_test_cases(self, state: SDLCState):
+        """
+            Generates the test cases based on the generated code and code review comments
+        """
+        logger.info("----- Generating Test Cases ----")
+    
+        # Get the generated code and code review comments from the state
+        code_generated = state.get('code_generated', '')
+        code_review_comments = state.get('code_review_comments', '')
+
+         # Create a prompt for the LLM to generate test cases
+        prompt = f"""
+            You are a software testing expert. Based on the following Python code and its review comments, generate comprehensive test cases:
+            
+            ### Code:
+            ```
+                {code_generated}
+                ```
+
+                ### Code Review Comments:
+                {code_review_comments}
+
+                Focus on:
+                1. Covering all edge cases and boundary conditions.
+                2. Ensuring functional correctness of the code.
+                3. Including both positive and negative test cases.
+                4. Writing test cases in Python's `unittest` framework format.
+
+                Provide the test cases in Python code format, ready to be executed.
+        """
+
+        response = self.llm.invoke(prompt)
+        state["test_cases"] = response.content
+
+        return state
+    
+    def review_test_cases(self, state: SDLCState):
+        return state
+    
+    def revise_test_cases(self, state: SDLCState):
+        pass
+    
+    def review_test_cases_router(self, state: SDLCState):
+        """
+            Evaluates Test Cases review is required or not.
+        """
+        return state.get("test_case_review_status", "approved")  # default to "approved" if not present
+    
+    ## ---- QA Testing ----- ##
+    def qa_testing(self, state: SDLCState):
+        """
+            Performs QA testing based on the generated code and test cases
+        """
+        logger.info("----- Performing QA Testing ----")
+        # Get the generated code and test cases from the state
+        code_generated = state.get('code_generated', '')
+        test_cases = state.get('test_cases', '')
+
+        # Create a prompt for the LLM to simulate running the test cases
+        prompt = f"""
+            You are a QA testing expert. Based on the following Python code and test cases, simulate running the test cases and provide feedback:
+            
+            ### Code:
+            ```
+            {code_generated}
+            ```
+
+            ### Test Cases:
+            ```
+            {test_cases}
+            ```
+
+            Focus on:
+            1. Identifying which test cases pass and which fail.
+            2. Providing detailed feedback for any failed test cases, including the reason for failure.
+            3. Suggesting improvements to the code or test cases if necessary.
+
+            Provide the results in the following format:
+            - Test Case ID: [ID]
+            Status: [Pass/Fail]
+            Feedback: [Detailed feedback if failed]
+        """
+
+        # Invoke the LLM to simulate QA testing
+        response = self.llm.invoke(prompt)
+        qa_testing_comments = response.content
+
+        state["qa_testing_comments"]= qa_testing_comments
+        return state
+    
+    def qa_review(self, state: SDLCState):
+        pass
+    
+    def deployment(self, state: SDLCState):
+        """
+            Performs the deployment
+        """
+        logger.info("----- Generating Deployment Simulation----")
+
+        code_generated = state.get('code_generated', '')
+
+        # Create a prompt for the LLM to simulate deployment
+        prompt = f"""
+            You are a DevOps expert. Based on the following Python code, simulate the deployment process and provide feedback:
+            
+            ### Code:
+            ```
+            {code_generated}
+            ```
+
+            Focus on:
+            1. Identifying potential deployment issues (e.g., missing dependencies, configuration errors).
+            2. Providing recommendations to resolve any issues.
+            3. Confirming whether the deployment is successful or needs further action.
+
+            Provide the results in the following format:
+            - Deployment Status: [Success/Failed]
+            - Feedback: [Detailed feedback on the deployment process]
+        """
+
+        # Invoke the LLM to simulate deployment
+        response = self.llm.invoke(prompt)
+        deployment_feedback = response.content
+
+         # Determine the deployment status based on the feedback
+        if "SUCCESS" in deployment_feedback.upper():
+            deployment_status = "success"
+        else:
+            deployment_status = "failed"
+
+        # Update the state with the deployment results
+        return {
+            **state,
+            "deployment_status": deployment_status,
+            "deployment_feedback": deployment_feedback
+        }

src/dev_pilot/nodes/design_document_node.py

@@ -0,0 +1,145 @@
+from src.dev_pilot.state.sdlc_state import SDLCState, DesignDocument
+from src.dev_pilot.utils.Utility import Utility
+from loguru import logger
+
+class DesingDocumentNode:
+    """
+    Graph Node for the Desing Documents
+    
+    """
+    
+    def __init__(self, model):
+        self.llm = model
+        self.utility = Utility()
+    
+    def create_design_documents(self, state: SDLCState):
+        """
+        Generates the Design document functional and technical
+        """
+        logger.info("----- Creating Design Document ----")
+        requirements = state.get('requirements', '')
+        user_stories = state.get('user_stories', '')
+        project_name = state.get('project_name', '')
+        design_feedback = None
+        
+        if 'design_documents' in state:
+            design_feedback = state.get('design_documents_feedback','')
+
+        functional_documents = self.generate_functional_design(
+            project_name=project_name,
+            requirements=requirements,
+            user_stories=user_stories,
+            design_feedback=design_feedback
+        )
+
+        technical_documents = self.generate_technical_design(
+            project_name=project_name,
+            requirements=requirements,
+            user_stories=user_stories,
+            design_feedback=design_feedback
+        )
+
+        design_documents = DesignDocument(
+            functional=functional_documents,
+            technical = technical_documents
+        )
+
+        return {
+            **state,
+            "design_documents": design_documents,
+            "technical_documents": technical_documents
+        }
+    
+    def generate_functional_design(self, project_name, requirements, user_stories, design_feedback):
+        """
+        Helper method to generate functional design document
+        """
+        logger.info("----- Creating Functional Design Document ----")
+        prompt = f"""
+            Create a comprehensive functional design document for {project_name} in Markdown format.
+    
+            The document should use proper Markdown syntax with headers (# for main titles, ## for sections, etc.), 
+            bullet points, tables, and code blocks where appropriate.
+            
+            Requirements:
+            {self.utility.format_list(requirements)}
+            
+            User Stories:
+            {self.utility.format_user_stories(user_stories)}
+
+             {f"When creating this functional design document, please incorporate the following feedback about the requirements: {design_feedback}" if design_feedback else ""}
+            
+            The functional design document should include the following sections, each with proper Markdown formatting:
+            
+            # Functional Design Document: {project_name}
+            
+            ## 1. Introduction and Purpose
+            ## 2. Project Scope
+            ## 3. User Roles and Permissions
+            ## 4. Functional Requirements Breakdown
+            ## 5. User Interface Design Guidelines
+            ## 6. Business Process Flows
+            ## 7. Data Entities and Relationships
+            ## 8. Validation Rules
+            ## 9. Reporting Requirements
+            ## 10. Integration Points
+            
+            Make sure to maintain proper Markdown formatting throughout the document.
+        """
+        # invoke the llm
+        response = self.llm.invoke(prompt)
+        return response.content    
+    
+    def generate_technical_design(self, project_name, requirements, user_stories, design_feedback):
+            """
+                Helper method to generate technical design document in Markdown format
+            """
+            logger.info("----- Creating Technical Design Document ----")
+            prompt = f"""
+                Create a comprehensive technical design document for {project_name} in Markdown format.
+                
+                The document should use proper Markdown syntax with headers (# for main titles, ## for sections, etc.), 
+                bullet points, tables, code blocks, and diagrams described in text form where appropriate.
+                
+                Requirements:
+                {self.utility.format_list(requirements)}
+            
+                User Stories:
+                {self.utility.format_user_stories(user_stories)}
+
+                {f"When creating this technical design document, please incorporate the following feedback about the requirements: {design_feedback}" if design_feedback else ""}
+                
+                The technical design document should include the following sections, each with proper Markdown formatting:
+                
+                # Technical Design Document: {project_name}
+
+                ## 1. System Architecture
+                ## 2. Technology Stack and Justification
+                ## 3. Database Schema
+                ## 4. API Specifications
+                ## 5. Security Considerations
+                ## 6. Performance Considerations
+                ## 7. Scalability Approach
+                ## 8. Deployment Strategy
+                ## 9. Third-party Integrations
+                ## 10. Development, Testing, and Deployment Environments
+                
+                For any code examples, use ```language-name to specify the programming language.
+                For database schemas, represent tables and relationships using Markdown tables.
+                Make sure to maintain proper Markdown formatting throughout the document.
+            """
+            response = self.llm.invoke(prompt)
+            return response.content
+    
+    def review_design_documents(self, state: SDLCState):
+        return state
+    
+    def revise_design_documents(self, state: SDLCState):
+        pass
+    
+    def review_design_documents_router(self, state: SDLCState):
+        """
+            Evaluates design review is required or not.
+        """
+        return state.get("design_documents_review_status", "approved")  # default to "approved" if not present
+    

src/dev_pilot/nodes/markdown_node.py

@@ -0,0 +1,124 @@
+import os
+from src.dev_pilot.state.sdlc_state import SDLCState
+from src.dev_pilot.utils.Utility import Utility
+from loguru import logger
+
+class MarkdownArtifactsNode:
+    """
+    Graph Node for generating Markdown artifacts for the SDLC process.
+    This node generates Markdown files for:
+    - Project Requirements
+    - User Stories
+    - Design Documents
+    - Generated Code
+    and saves them to the "artifacts" folder.
+    """
+    
+    def __init__(self):
+        self.utility = Utility()
+    
+    def generate_markdown_artifacts(self, state: SDLCState):
+        """
+        Generate Markdown files for each step in the SDLC state and save them to the artifacts folder.
+        Returns the updated state with a new key 'artifacts' that maps to a dictionary of file paths.
+        """
+        artifacts_dir = "artifacts"
+        os.makedirs(artifacts_dir, exist_ok=True)
+        
+        project_name = state.get("project_name", "Project")
+        
+        # -- Project Requirements Markdown --
+        requirements = state.get("requirements", [])
+        md_project = f"# Project Requirement for {project_name}\n\n"
+        md_project += "## Requirements\n"
+        md_project += self.utility.format_list(requirements)
+        file_project = os.path.join(artifacts_dir, "Project_Requirement.md")
+        with open(file_project, "w") as f:
+            f.write(md_project)
+        
+        # -- User Stories Markdown --
+        user_stories = state.get("user_stories", None)
+        file_stories = None
+        if user_stories:
+            md_stories = f"# User Stories for {project_name}\n\n"
+            md_stories += self.utility.format_user_stories(user_stories)
+            file_stories = os.path.join(artifacts_dir, "User_Stories.md")
+            with open(file_stories, "w") as f:
+                f.write(md_stories)
+        
+        # -- Design Documents Markdown --
+        design_docs = state.get("design_documents", None)
+        file_design = None
+        if design_docs:
+            md_design = f"# Design Documents for {project_name}\n\n"
+            md_design += "## Functional Design Document\n"
+            md_design += design_docs.get("functional", "No Functional Design Document available.")
+            md_design += "\n\n## Technical Design Document\n"
+            md_design += design_docs.get("technical", "No Technical Design Document available.")
+            file_design = os.path.join(artifacts_dir, "Design_Documents.md")
+            with open(file_design, "w") as f:
+                f.write(md_design)
+        
+        # -- Generated Code Markdown --
+        code_generated = state.get("code_generated", None)
+        file_code = None
+        if code_generated:
+            md_code = f"# Generated Code for {project_name}\n\n"
+            md_code += "\n" + code_generated
+            file_code = os.path.join(artifacts_dir, "Generated_Code.md")
+            with open(file_code, "w") as f:
+                f.write(md_code)
+        
+       # -- Security Recommendations Markdown --
+        security_recommendations = state.get("security_recommendations", None)
+        file_security = None
+        if security_recommendations:
+            md_security = f"# Security Recommendations for {project_name}\n\n"
+            md_security += security_recommendations
+            file_security = os.path.join(artifacts_dir, "Security_Recommendations.md")
+            with open(file_security, "w") as f:
+                f.write(md_security)
+        
+        # -- Test Cases Markdown --
+        test_cases = state.get("test_cases", None)
+        file_tests = None
+        if test_cases:
+            md_tests = f"# Test Cases for {project_name}\n\n"
+            md_tests += "\n" + test_cases
+            file_tests = os.path.join(artifacts_dir, "Test_Cases.md")
+            with open(file_tests, "w") as f:
+                f.write(md_tests)
+        
+        # -- QA Testing Comments Markdown --
+        qa_testing_comments = state.get("qa_testing_comments", None)
+        file_qa = None
+        if qa_testing_comments:
+            md_qa = f"# QA Testing Comments for {project_name}\n\n"
+            md_qa += qa_testing_comments
+            file_qa = os.path.join(artifacts_dir, "QA_Testing_Comments.md")
+            with open(file_qa, "w") as f:
+                f.write(md_qa)
+        
+        # -- Deployment Feedback Markdown --
+        deployment_feedback = state.get("deployment_feedback", None)
+        file_deployment = None
+        if deployment_feedback:
+            md_deployment = f"# Deployment Feedback for {project_name}\n\n"
+            md_deployment += deployment_feedback
+            file_deployment = os.path.join(artifacts_dir, "Deployment_Feedback.md")
+            with open(file_deployment, "w") as f:
+                f.write(md_deployment)
+        
+        # Update the state with the paths to the generated artifact files.
+        state["artifacts"] = {
+            "Project_Requirements": file_project,
+            "User_Stories": file_stories,
+            "Design_Documents": file_design,
+            "Generated_Code": file_code,
+            "Security_Recommendations": file_security,
+            "Test_Cases": file_tests,
+            "QA_Testing_Comments": file_qa,
+            "Deployment_Feedback": file_deployment
+        }
+        logger.info("Markdown artifacts generated in folder:", artifacts_dir)
+        return state

src/dev_pilot/nodes/project_requirement_node.py

@@ -0,0 +1,81 @@
+from src.dev_pilot.state.sdlc_state import SDLCState, UserStoryList
+from langchain_core.messages import SystemMessage
+
+class ProjectRequirementNode:
+    """
+    Graph Node for the project requirements
+    
+    """
+    
+    def __init__(self, model):
+        self.llm = model
+      
+    def initialize_project(self, state: SDLCState):
+        """
+            Performs the project initilazation
+        """
+        return state
+    
+    def get_user_requirements(self, state: SDLCState):
+        """
+            Gets the requirements from the user
+        """
+        pass
+    
+    def generate_user_stories(self, state: SDLCState):
+        """
+        Auto-generate highly detailed and accurate user stories for each requirement.
+        """
+        project_name = state["project_name"]
+        requirements = state["requirements"]
+        feedback_reason = state.get("user_stories_feedback", None)
+
+        prompt = f"""
+        You are a senior software analyst specializing in Agile SDLC and user story generation. 
+        Your task is to generate **a separate and detailed user story for EACH requirement** from the project details below.
+
+        ---
+        **Project Name:** "{project_name}"
+
+        **Requirements:** "{requirements}
+
+        ---
+        **Instructions for User Story Generation:**
+        - Create **one user story per requirement**.
+        - Assign a **unique identifier** (e.g., US-001, US-002, etc.).
+        - Provide a **clear and concise title** summarizing the user story.
+        - Write a **detailed description** using the "As a [user role], I want [goal] so that [benefit]" format.
+        - Assign a **priority level** (1 = Critical, 2 = High, 3 = Medium, 4 = Low).
+        - Define **acceptance criteria** with bullet points to ensure testability.
+        - Use **domain-specific terminology** for clarity.
+        
+        {f"Additionally, consider the following feedback while refining the user stories: {feedback_reason}" if feedback_reason else ""}
+
+        ---
+        **Expected Output Format (for each user story):**
+        - Unique Identifier: US-XXX
+        - Title: [User Story Title]
+        - Description:  
+        - As a [user role], I want [feature] so that [benefit].
+        - Priority: [1-4]
+        - Acceptance Criteria:
+        - [Criteria 1]
+        - [Criteria 2]
+        - [Criteria 3]
+
+        Ensure that the user stories are **specific, testable, and aligned with Agile principles**.
+        """
+
+        llm_with_structured = self.llm.with_structured_output(UserStoryList)
+        response = llm_with_structured.invoke(prompt)
+        state["user_stories"] = response
+        return state
+    
+    def review_user_stories(self, state: SDLCState):
+        return state
+    
+    def revise_user_stories(self, state: SDLCState):
+        pass
+    
+    def review_user_stories_router(self, state: SDLCState):
+        return state.get("user_stories_review_status", "approved")  # default to "approved" if not present