Spaces:
Build error
Build error
kamaleswar Mohanta commited on
Commit ·
13f57ba
1
Parent(s): 4fc4e2e
note book updated
Browse files- src/langgraphagenticai/notebook/sdlc.py +110 -149
src/langgraphagenticai/notebook/sdlc.py
CHANGED
|
@@ -243,117 +243,51 @@ class SdlcNode:
|
|
| 243 |
|
| 244 |
return {"user_input": "captured"}
|
| 245 |
|
| 246 |
-
# def generate_requirements(self, state: SDLCState) -> dict:
|
| 247 |
-
# """Generate requirements based on user input."""
|
| 248 |
-
# logger.debug(f"Generating requirements with state: {state}")
|
| 249 |
-
# try:
|
| 250 |
-
# requirements_input = {
|
| 251 |
-
# "project_name": state.project_name if state.project_name is not None else "No project name provided",
|
| 252 |
-
# "project_description": state.project_description if state.project_description is not None else "No project description provided",
|
| 253 |
-
# "project_goals": state.project_goals if state.project_goals is not None else "No project goals provided",
|
| 254 |
-
# "project_scope": state.project_scope if state.project_scope is not None else "No project scope provided",
|
| 255 |
-
# "project_objectives": state.project_objectives if state.project_objectives is not None else "No project objectives provided",
|
| 256 |
-
# }
|
| 257 |
-
# prompt_string = f"""Based on the following project details, generate a comprehensive list of detailed software requirements.
|
| 258 |
-
# Ensure the requirements are clear, unambiguous, verifiable, and complete based on the provided description, goals, scope, and objectives.
|
| 259 |
-
|
| 260 |
-
# Project Details:
|
| 261 |
-
# {json.dumps(requirements_input, indent=2)}
|
| 262 |
-
|
| 263 |
-
# Detailed Requirements:
|
| 264 |
-
# """""
|
| 265 |
-
# # Construct a list of messages for the LLM
|
| 266 |
-
# messages = [SystemMessage(content="You are an expert project requirements generator."),HumanMessage(content=prompt_string)]
|
| 267 |
-
# response = self.llm.invoke(messages)
|
| 268 |
-
# state.generated_requirements = response.content if hasattr(response, 'content') else str(response)
|
| 269 |
-
# return {"generated_requirements": state.generated_requirements}
|
| 270 |
-
# except Exception as e:
|
| 271 |
-
# logger.error(f"Error generating requirements: {e}")
|
| 272 |
-
# state.generated_requirements = f"Error generating requirements: {str(e)}"
|
| 273 |
-
# return {"generated_requirements": state.generated_requirements}
|
| 274 |
-
|
| 275 |
def generate_requirements(self, state: SDLCState) -> dict:
|
| 276 |
-
"""
|
| 277 |
-
logger.debug("
|
| 278 |
-
|
| 279 |
-
|
| 280 |
-
|
| 281 |
-
|
| 282 |
-
|
| 283 |
-
|
| 284 |
-
|
| 285 |
-
|
| 286 |
-
|
| 287 |
-
|
| 288 |
-
|
| 289 |
-
|
| 290 |
-
|
| 291 |
-
|
| 292 |
-
|
| 293 |
-
|
| 294 |
-
|
| 295 |
-
|
| 296 |
-
# Each user story should follow the format: 'As a [type of user], I want [some goal] so that [some reason/benefit].'
|
| 297 |
-
# Ensure the user stories cover the key functionalities outlined in the requirements and are actionable from a development perspective.
|
| 298 |
-
|
| 299 |
-
# Requirements:
|
| 300 |
-
# {state.generated_requirements}
|
| 301 |
-
|
| 302 |
-
# Previous Feedback to Address:
|
| 303 |
-
# {feedback}
|
| 304 |
-
|
| 305 |
-
# User Stories:
|
| 306 |
-
# """""
|
| 307 |
-
# else:
|
| 308 |
-
|
| 309 |
-
# try:
|
| 310 |
-
# if state.generated_requirements:
|
| 311 |
-
# prompt_string = f"""Based on the following software requirements, generate a list of user stories.
|
| 312 |
-
# Each user story should follow the format: 'As a [type of user], I want [some goal] so that [some reason/benefit].'
|
| 313 |
-
# Ensure the user stories cover the key functionalities outlined in the requirements and are actionable from a development perspective.
|
| 314 |
-
|
| 315 |
-
# Requirements:
|
| 316 |
-
# {state.generated_requirements}
|
| 317 |
-
|
| 318 |
-
# User Stories:
|
| 319 |
-
# """""
|
| 320 |
-
# sys_prompt= f"""
|
| 321 |
-
# You are a Senior Software Analyst expert in Agile SDLC and user story creation. Your task is to generate detailed user stories based on the provided requirements.
|
| 322 |
-
|
| 323 |
-
# Project Name: {state.project_name or 'N/A'}
|
| 324 |
-
|
| 325 |
-
# Guidelines:
|
| 326 |
-
|
| 327 |
-
# One Requirement = One User Story: Create a distinct user story for each functional requirement identified.
|
| 328 |
-
# Unique Identifier: Assign each user story a unique ID: [PROJECT_CODE]-US-[XXX] (e.g., BN-US-001 for 'The Book Nook'). Use a short uppercase code for the project.
|
| 329 |
-
# Structure (for each story):
|
| 330 |
-
# Unique Identifier: [PROJECT_CODE]-US-XXX
|
| 331 |
-
# Title: Clear summary of the functionality.
|
| 332 |
-
# Description: As a [user role], I want [goal/feature] so that [reason/benefit].
|
| 333 |
-
# Acceptance Criteria: Bulleted list of testable conditions (- [Criterion]).
|
| 334 |
-
# Clarity: Use domain-specific terms. Ensure stories are specific, testable, achievable, and Agile-aligned.
|
| 335 |
-
# {f'5. Incorporate Feedback: The previous version was rejected. Address the following feedback while refining the user stories: "{feedback}"' if feedback else ''} """
|
| 336 |
-
|
| 337 |
-
# messages = [
|
| 338 |
-
# SystemMessage(content=sys_prompt),
|
| 339 |
-
# HumanMessage(content=prompt_string)
|
| 340 |
-
# ]
|
| 341 |
-
# response = self.llm.invoke(messages)
|
| 342 |
-
# state.user_stories = response.content if hasattr(response, 'content') else str(response)
|
| 343 |
-
# return {"user_stories": state.user_stories}
|
| 344 |
-
# else:
|
| 345 |
-
# state.user_stories = "No requirements generated yet."
|
| 346 |
-
# return {"user_stories": state.user_stories}
|
| 347 |
-
# except Exception as e:
|
| 348 |
-
# logger.error(f"Error generating user stories: {e}")
|
| 349 |
-
# state.user_stories = f"Error generating user stories: {str(e)}"
|
| 350 |
-
# return {"user_stories": state.user_stories}
|
| 351 |
|
| 352 |
def generate_user_stories(self, state: SDLCState) -> dict:
|
| 353 |
-
"""
|
| 354 |
-
logger.debug("
|
| 355 |
-
|
| 356 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 357 |
|
| 358 |
|
| 359 |
def process_feedback(self, state: SDLCState) -> dict:
|
|
@@ -361,63 +295,77 @@ class SdlcNode:
|
|
| 361 |
Process user feedback passed via state and update state with decision.
|
| 362 |
Only { "current_stage": ["accept"] } ends the process.
|
| 363 |
"""
|
| 364 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 365 |
logger.debug(f"Processing feedback for stage: {current_stage_value}")
|
| 366 |
|
| 367 |
raw_feedback = state.feedback
|
| 368 |
-
logger.debug(f"Raw feedback data: {raw_feedback}")
|
| 369 |
|
| 370 |
# Only accept if feedback is {current_stage: ["accept"]}
|
|
|
|
| 371 |
if (
|
| 372 |
isinstance(raw_feedback, dict)
|
| 373 |
and current_stage_value in raw_feedback
|
| 374 |
and isinstance(raw_feedback[current_stage_value], list)
|
| 375 |
-
and raw_feedback[current_stage_value]
|
| 376 |
and raw_feedback[current_stage_value][-1].strip().lower() == "accept"
|
| 377 |
):
|
|
|
|
| 378 |
state.feedback_decision = "accept"
|
| 379 |
-
return {
|
| 380 |
-
"feedback_decision": "accept",
|
| 381 |
-
"feedback": state.feedback
|
| 382 |
-
}
|
| 383 |
else:
|
|
|
|
| 384 |
state.feedback_decision = "reject"
|
| 385 |
-
|
| 386 |
-
|
| 387 |
-
|
| 388 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
| 389 |
|
| 390 |
def feedback_route(self, state: SDLCState) -> str:
|
| 391 |
-
|
| 392 |
-
logger.debug(f"
|
| 393 |
-
|
| 394 |
-
|
| 395 |
-
|
| 396 |
-
|
| 397 |
-
|
| 398 |
-
|
| 399 |
-
|
| 400 |
-
|
| 401 |
-
|
| 402 |
-
|
| 403 |
-
|
| 404 |
-
|
| 405 |
-
|
| 406 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 407 |
else:
|
| 408 |
-
|
| 409 |
-
|
| 410 |
-
|
| 411 |
-
|
| 412 |
-
logger.debug(f"ProcessFeedback node output: {node_output}")
|
| 413 |
-
feedback_decision = node_output.get("feedback_decision")
|
| 414 |
-
logger.debug(f"Found feedback_decision in node output: {feedback_decision}")
|
| 415 |
-
except Exception as e:
|
| 416 |
-
logger.error(f"Error extracting feedback_decision: {e}")
|
| 417 |
-
|
| 418 |
-
route = "accept" if feedback_decision == "accept" else "reject"
|
| 419 |
-
logger.debug(f"Final routing decision: {route}")
|
| 420 |
-
return route
|
| 421 |
|
| 422 |
|
| 423 |
##################################################################################################
|
|
@@ -487,4 +435,17 @@ class SdlcGraphBuilder:
|
|
| 487 |
|
| 488 |
# Assuming 'model' is your initialized LLM
|
| 489 |
sdlc_builder_instance = SdlcGraphBuilder(model)
|
| 490 |
-
agent = sdlc_builder_instance.build_graph()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 243 |
|
| 244 |
return {"user_input": "captured"}
|
| 245 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 246 |
def generate_requirements(self, state: SDLCState) -> dict:
|
| 247 |
+
"""Generate requirements based on user input."""
|
| 248 |
+
logger.debug(f"Generating requirements with state: {state}")
|
| 249 |
+
try:
|
| 250 |
+
requirements_input = {
|
| 251 |
+
"project_name": state.project_name if state.project_name is not None else "No project name provided",
|
| 252 |
+
"project_description": state.project_description if state.project_description is not None else "No project description provided",
|
| 253 |
+
"project_goals": state.project_goals if state.project_goals is not None else "No project goals provided",
|
| 254 |
+
"project_scope": state.project_scope if state.project_scope is not None else "No project scope provided",
|
| 255 |
+
"project_objectives": state.project_objectives if state.project_objectives is not None else "No project objectives provided",
|
| 256 |
+
}
|
| 257 |
+
prompt_string = f"""Based on the following project details, generate a comprehensive list of detailed software requirements.\nEnsure the requirements are clear, unambiguous, verifiable, and complete based on the provided description, goals, scope, and objectives.\n\nProject Details:\n{json.dumps(requirements_input, indent=2)}\n\nDetailed Requirements:\n"""
|
| 258 |
+
messages = [SystemMessage(content="You are an expert project requirements generator."), HumanMessage(content=prompt_string)]
|
| 259 |
+
response = self.llm.invoke(messages)
|
| 260 |
+
state.generated_requirements = response.content if hasattr(response, 'content') else str(response)
|
| 261 |
+
return {"generated_requirements": state.generated_requirements}
|
| 262 |
+
except Exception as e:
|
| 263 |
+
logger.error(f"Error generating requirements: {e}")
|
| 264 |
+
state.generated_requirements = f"Error generating requirements: {str(e)}"
|
| 265 |
+
return {"generated_requirements": state.generated_requirements}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 266 |
|
| 267 |
def generate_user_stories(self, state: SDLCState) -> dict:
|
| 268 |
+
"""Generate user stories based on the requirements, incorporating feedback if available."""
|
| 269 |
+
logger.debug("Generating user stories")
|
| 270 |
+
if not state.generated_requirements:
|
| 271 |
+
state.user_stories = "No requirements generated yet."
|
| 272 |
+
logger.warning("Cannot generate user stories without requirements.")
|
| 273 |
+
return {"user_stories": state.user_stories}
|
| 274 |
+
feedback = state.get_last_feedback_for_stage(SDLCStages.PLANNING)
|
| 275 |
+
logger.debug(f"Feedback for user stories: {feedback}")
|
| 276 |
+
if feedback:
|
| 277 |
+
prompt_string = f"""Based on the following software requirements AND feedback, generate a list of user stories.\nThe previous version was rejected for the following reason: \"{feedback}\"\nPlease make sure to address this feedback in your new user stories.\n\nEach user story should follow the format: 'As a [type of user], I want [some goal] so that [some reason/benefit].'\nEnsure the user stories cover the key functionalities outlined in the requirements and are actionable from a development perspective.\n\nRequirements:\n{state.generated_requirements}\n\nPrevious Feedback to Address:\n{feedback}\n\nUser Stories:\n"""
|
| 278 |
+
else:
|
| 279 |
+
prompt_string = f"""Based on the following software requirements, generate a list of user stories.\nEach user story should follow the format: 'As a [type of user], I want [some goal] so that [some reason/benefit].'\nEnsure the user stories cover the key functionalities outlined in the requirements and are actionable from a development perspective.\n\nRequirements:\n{state.generated_requirements}\n\nUser Stories:\n"""
|
| 280 |
+
sys_prompt = f"""
|
| 281 |
+
You are a Senior Software Analyst expert in Agile SDLC and user story creation. Your task is to generate detailed user stories based on the provided requirements.\n\nProject Name: {state.project_name or 'N/A'}\n\nGuidelines:\n\nOne Requirement = One User Story: Create a distinct user story for each functional requirement identified.\nUnique Identifier: Assign each user story a unique ID: [PROJECT_CODE]-US-[XXX] (e.g., BN-US-001 for 'The Book Nook'). Use a short uppercase code for the project.\nStructure (for each story):\nUnique Identifier: [PROJECT_CODE]-US-XXX\nTitle: Clear summary of the functionality.\nDescription: As a [user role], I want [goal/feature] so that [reason/benefit].\nAcceptance Criteria: Bulleted list of testable conditions (- [Criterion]).\nClarity: Use domain-specific terms. Ensure stories are specific, testable, achievable, and Agile-aligned.\n{f'5. Incorporate Feedback: The previous version was rejected. Address the following feedback while refining the user stories: "{feedback}"' if feedback else ''}\n"""
|
| 282 |
+
try:
|
| 283 |
+
messages = [SystemMessage(content=sys_prompt), HumanMessage(content=prompt_string)]
|
| 284 |
+
response = self.llm.invoke(messages)
|
| 285 |
+
state.user_stories = response.content if hasattr(response, 'content') else str(response)
|
| 286 |
+
return {"user_stories": state.user_stories}
|
| 287 |
+
except Exception as e:
|
| 288 |
+
logger.error(f"Error generating user stories: {e}")
|
| 289 |
+
state.user_stories = f"Error generating user stories: {str(e)}"
|
| 290 |
+
return {"user_stories": state.user_stories}
|
| 291 |
|
| 292 |
|
| 293 |
def process_feedback(self, state: SDLCState) -> dict:
|
|
|
|
| 295 |
Process user feedback passed via state and update state with decision.
|
| 296 |
Only { "current_stage": ["accept"] } ends the process.
|
| 297 |
"""
|
| 298 |
+
logger.debug(f"--- Entering process_feedback ---")
|
| 299 |
+
logger.debug(f"Input state: {state.to_dict() if isinstance(state, SDLCState) else state}")
|
| 300 |
+
|
| 301 |
+
# Normalize current_stage to enum
|
| 302 |
+
if isinstance(state.current_stage, str):
|
| 303 |
+
try:
|
| 304 |
+
current_stage_enum = SDLCStages(state.current_stage)
|
| 305 |
+
except ValueError:
|
| 306 |
+
logger.warning(f"Unknown current_stage string: {state.current_stage}")
|
| 307 |
+
current_stage_enum = None
|
| 308 |
+
else:
|
| 309 |
+
current_stage_enum = state.current_stage
|
| 310 |
+
|
| 311 |
+
current_stage_value = current_stage_enum.value if current_stage_enum else state.current_stage
|
| 312 |
+
|
| 313 |
logger.debug(f"Processing feedback for stage: {current_stage_value}")
|
| 314 |
|
| 315 |
raw_feedback = state.feedback
|
| 316 |
+
logger.debug(f"Raw feedback data received in state: {raw_feedback}")
|
| 317 |
|
| 318 |
# Only accept if feedback is {current_stage: ["accept"]}
|
| 319 |
+
logger.debug(f"Checking acceptance: current_stage_value={current_stage_value}, raw_feedback={raw_feedback}")
|
| 320 |
if (
|
| 321 |
isinstance(raw_feedback, dict)
|
| 322 |
and current_stage_value in raw_feedback
|
| 323 |
and isinstance(raw_feedback[current_stage_value], list)
|
| 324 |
+
and raw_feedback[current_stage_value] # Check if list is not empty
|
| 325 |
and raw_feedback[current_stage_value][-1].strip().lower() == "accept"
|
| 326 |
):
|
| 327 |
+
logger.info(f"Feedback for stage '{current_stage_value}' is ACCEPT. Ending flow.")
|
| 328 |
state.feedback_decision = "accept"
|
|
|
|
|
|
|
|
|
|
|
|
|
| 329 |
else:
|
| 330 |
+
logger.info(f"Feedback for stage '{current_stage_value}' is not accept. Looping back.")
|
| 331 |
state.feedback_decision = "reject"
|
| 332 |
+
|
| 333 |
+
return_value = {
|
| 334 |
+
"feedback_decision": state.feedback_decision,
|
| 335 |
+
"feedback": state.feedback # Pass the original feedback dict back
|
| 336 |
+
}
|
| 337 |
+
logger.debug(f"Returning from process_feedback: {return_value}")
|
| 338 |
+
logger.debug(f"--- Exiting process_feedback ---")
|
| 339 |
+
return return_value
|
| 340 |
|
| 341 |
def feedback_route(self, state: SDLCState) -> str:
|
| 342 |
+
"""Routes based on the feedback decision stored in the state."""
|
| 343 |
+
logger.debug(f"--- Entering feedback_route ---")
|
| 344 |
+
logger.debug(f"Routing feedback. Current state includes feedback_decision: {hasattr(state, 'feedback_decision')}")
|
| 345 |
+
# Log the full state dictionary received by the conditional edge function
|
| 346 |
+
logger.debug(f"Full state content received by feedback_route: {state.to_dict() if isinstance(state, SDLCState) else state}")
|
| 347 |
+
|
| 348 |
+
# The state object passed to a conditional edge function is the updated state
|
| 349 |
+
# after the preceding node (ProcessFeedback) has run.
|
| 350 |
+
if not isinstance(state, SDLCState):
|
| 351 |
+
logger.error(f"Incorrect state type passed to feedback_route: {type(state)}. Defaulting to reject.")
|
| 352 |
+
logger.debug(f"--- Exiting feedback_route (routing: reject due to type error) ---")
|
| 353 |
+
# Default to reject if state type is unexpected
|
| 354 |
+
return "reject"
|
| 355 |
+
|
| 356 |
+
# Access feedback_decision directly from the state object
|
| 357 |
+
feedback_decision = state.feedback_decision
|
| 358 |
+
logger.debug(f"Feedback decision read from state object: {feedback_decision}")
|
| 359 |
+
|
| 360 |
+
if feedback_decision == "accept":
|
| 361 |
+
logger.info("Feedback accepted. Routing to END.")
|
| 362 |
+
logger.debug(f"--- Exiting feedback_route (routing: accept) ---")
|
| 363 |
+
return "accept"
|
| 364 |
else:
|
| 365 |
+
# This branch handles None (if feedback_decision wasn't set) or "reject"
|
| 366 |
+
logger.info(f"Feedback decision is '{feedback_decision}'. Routing back for revision.")
|
| 367 |
+
logger.debug(f"--- Exiting feedback_route (routing: reject) ---")
|
| 368 |
+
return "reject"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 369 |
|
| 370 |
|
| 371 |
##################################################################################################
|
|
|
|
| 435 |
|
| 436 |
# Assuming 'model' is your initialized LLM
|
| 437 |
sdlc_builder_instance = SdlcGraphBuilder(model)
|
| 438 |
+
agent = sdlc_builder_instance.build_graph()
|
| 439 |
+
|
| 440 |
+
if __name__ == "__main__":
|
| 441 |
+
# Example usage
|
| 442 |
+
state = SDLCState(session_id="12345")
|
| 443 |
+
state.project_name = "Sample Project"
|
| 444 |
+
state.project_description = "This is a sample project."
|
| 445 |
+
state.project_goals = "To demonstrate the SDLC process."
|
| 446 |
+
state.project_scope = "Web application development."
|
| 447 |
+
state.project_objectives = "Complete the project in 3 months."
|
| 448 |
+
|
| 449 |
+
# Run the agent with the initial state
|
| 450 |
+
result = agent.run(state)
|
| 451 |
+
print(result) # Output the result of the agent run
|