agentic_workflow.py

# agentic_workflow.py

# TODO: 1 - Import the following agents: ActionPlanningAgent, KnowledgeAugmentedPromptAgent, EvaluationAgent, RoutingAgent from the workflow_agents.base_agents module
from workflow_agents.base_agents import ActionPlanningAgent
from workflow_agents.base_agents import KnowledgeAugmentedPromptAgent
from workflow_agents.base_agents import EvaluationAgent
from workflow_agents.base_agents import RoutingAgent
import os
from dotenv import load_dotenv
from openai import OpenAI

# TODO: 2 - Load the OpenAI key into a variable called openai_api_key
load_dotenv()

# Retrieve OpenAI API key from environment variables
openai_api_key = os.getenv("OPENAI_API_KEY")

# load the product spec
# TODO: 3 - Load the product spec document Product-Spec-Email-Router.txt into a variable called product_spec
try:
    with open('Product-Spec-Email-Router.txt', 'r') as file:
        product_spec = file.read()
    print("Product Spec Email loaded successfully..")
except FileNotFoundError:
    print("Error: The file 'Product-Spec-Email-Router.txt' was not found.")
except Exception as e:
    print(f"An error occurred: {e}")

# Instantiate all the agents

# Action Planning Agent
knowledge_action_planning = (
    "Stories are defined from a product spec by identifying a "
    "persona, an action, and a desired outcome for each story. "
    "Each story represents a specific functionality of the product "
    "described in the specification. \n"
    "Features are defined by grouping related user stories. \n"
    "Tasks are defined for each story and represent the engineering "
    "work required to develop the product. \n"
    "A development Plan for a product contains all these components"
)
# TODO: 4 - Instantiate an action_planning_agent using the 'knowledge_action_planning'
action_planning_agent = ActionPlanningAgent(openai_api_key, knowledge_action_planning)

# Product Manager - Knowledge Augmented Prompt Agent
persona_product_manager = "You are a Product Manager, you are responsible for defining the user stories for a product."
knowledge_product_manager = (
    "Stories are defined by writing sentences with a persona, an action, and a desired outcome. "
    "The sentences always start with: As a "
    "Write several stories for the product spec below, where the personas are the different users of the product. "
    " New user stories should start on a new line."
    f"The prdouct spec is {product_spec}."
    # TODO: 5 - Complete this knowledge string by appending the product_spec loaded in TODO 3
)
# TODO: 6 - Instantiate a product_manager_knowledge_agent using 'persona_product_manager' and the completed 'knowledge_product_manager'
product_manager_knowledge_agent = KnowledgeAugmentedPromptAgent(openai_api_key, persona_product_manager, knowledge_product_manager)

# Product Manager - Evaluation Agent
# TODO: 7 - Define the persona and evaluation criteria for a Product Manager evaluation agent and instantiate it as product_manager_evaluation_agent. This agent will evaluate the product_manager_knowledge_agent.
# The evaluation_criteria should specify the expected structure for user stories (e.g., "As a [type of user], I want [an action or feature] so that [benefit/value].").
persona = "You are an product manager evaluation agent that checks the answers of other worker agents"
evaluation_criteria = "All user stories should follow the structure: 'As a [type of user], I want [an action or feature] so that [benefit/value].' "

product_manager_evaluation_agent = EvaluationAgent(openai_api_key, persona, evaluation_criteria, product_manager_knowledge_agent, 25)

# Program Manager - Knowledge Augmented Prompt Agent
persona_program_manager = "You are a Program Manager, you are responsible for defining the features for a product."
knowledge_program_manager = "Features of a product are defined by organizing similar user stories into cohesive groups."

# Instantiate a program_manager_knowledge_agent using 'persona_program_manager' and 'knowledge_program_manager'
# (This is a necessary step before TODO 8. Students should add the instantiation code here.)
program_manager_knowledge_agent= KnowledgeAugmentedPromptAgent(openai_api_key, persona_program_manager, knowledge_program_manager)

# Program Manager - Evaluation Agent
persona_program_manager_eval = "You are an evaluation agent that checks the answers of other worker agents."

# TODO: 8 - Instantiate a program_manager_evaluation_agent using 'persona_program_manager_eval' and the evaluation criteria below.
#                      "The answer should be product features that follow the following structure: " \
#                      "Feature Name: A clear, concise title that identifies the capability\n" \
#                      "Description: A brief explanation of what the feature does and its purpose\n" \
#                      "Key Functionality: The specific capabilities or actions the feature provides\n" \
#                      "User Benefit: How this feature creates value for the user"
# For the 'agent_to_evaluate' parameter, refer to the provided solution code's pattern.
program_evaluation_criteria = "The answer should be product features that follow the following structure: " \
                              "Feature Name: A clear, concise title that identifies the capability\n" \
                              "Description: A brief explanation of what the feature does and its purpose\n" \
                              "Key Functionality: The specific capabilities or actions the feature provides\n" \
                              "User Benefit: How this feature creates value for the user"

program_manager_evaluation_agent = EvaluationAgent(openai_api_key, persona_program_manager_eval, program_evaluation_criteria, program_manager_knowledge_agent, 25)

# Development Engineer - Knowledge Augmented Prompt Agent
persona_dev_engineer = "You are a Development Engineer, you are responsible for defining the development tasks for a product."
knowledge_dev_engineer = "Development tasks are defined by identifying what needs to be built to implement each user story. " \
                         "Include acceptance criteria, effort estimates, and dependencies. " \
                         "Make sure to include tasks for all features in your response, not just for a subset of the features."

# Instantiate a development_engineer_knowledge_agent using 'persona_dev_engineer' and 'knowledge_dev_engineer'
# (This is a necessary step before TODO 9. Students should add the instantiation code here.)
development_engineer_knowledge_agent= KnowledgeAugmentedPromptAgent(openai_api_key, persona_dev_engineer, knowledge_dev_engineer)

# Development Engineer - Evaluation Agent
persona_dev_engineer_eval = "You are an evaluation agent that checks the answers of other worker agents."
# TODO: 9 - Instantiate a development_engineer_evaluation_agent using 'persona_dev_engineer_eval' and the evaluation criteria below.
#                      "The answer should be tasks following this exact structure: " \
#                      "Task ID: A unique identifier for tracking purposes\n" \
#                      "Task Title: Brief description of the specific development work\n" \
#                      "Related User Story: Reference to the parent user story\n" \
#                      "Description: Detailed explanation of the technical work required\n" \
#                      "Acceptance Criteria: Specific requirements that must be met for completion\n" \
#                      "Estimated Effort: Time or complexity estimation\n" \
#                      "Dependencies: Any tasks that must be completed first"
# For the 'agent_to_evaluate' parameter, refer to the provided solution code's pattern.
development_engineer_evaluation_criteria = "The answer should be tasks following this exact structure: " \
                                           "Task ID: A unique identifier for tracking purposes\n" \
                                           "Task Title: Brief description of the specific development work\n" \
                                           "Related User Story: Reference to the parent user story\n" \
                                           "Description: Detailed explanation of the technical work required\n" \
                                           "Acceptance Criteria: Specific requirements that must be met for completion\n" \
                                           "Estimated Effort: Time or complexity estimation\n" \
                                           "Dependencies: Any tasks that must be completed first"

development_engineer_evaluation_agent = EvaluationAgent(openai_api_key, persona_dev_engineer_eval, development_engineer_evaluation_criteria, development_engineer_knowledge_agent, 25)

# Routing Agent
# TODO: 10 - Instantiate a routing_agent. You will need to define a list of agent dictionaries (routes) for Product Manager, Program Manager, and Development Engineer. Each dictionary should contain 'name', 'description', and 'func' (linking to a support function). Assign this list to the routing_agent's 'agents' attribute.
routing_agent = RoutingAgent(openai_api_key, {})

agents = [ #Product Manager, Program Manager, and Development Engineer.
    {
        "name": "Product Manager",
        "description": "Answer a question about the product",
        # Call the product manager to respond to product prompts
        "func": lambda x: f"Routing data: {product_manager_support_function(x)}"
    },
    {
        "name": "Program Manager",
        "description": "Answer a question about the program",
        # Call the program manager to respond to program prompts
        "func": lambda x: f"Routing data: {program_manager_support_function(x)}"
    },
    {
        "name": "Development Engineer",
        "description": "When a product requires development answers",
        # Call the development engineer regarding product development questions
        "func": lambda x: f"Routing data: {development_engineer_support_function(x)}"
    }
]
routing_agent.agents = agents

# Job function persona support functions
# TODO: 11 - Define the support functions for the routes of the routing agent (e.g., product_manager_support_function, program_manager_support_function, development_engineer_support_function).
# Each support function should:
#   1. Take the input query (e.g., a step from the action plan).
#   2. Get a response from the respective Knowledge Augmented Prompt Agent.
#   3. Have the response evaluated by the corresponding Evaluation Agent.
#   4. Return the final validated response.
def product_manager_support_function (prompt):
    """ Product Manager Support function.
    Sends routed prompt to the product manager knowledge agent for initial response.
    Then sends that response for evaluation and correction if needed.
    Parameters:
    propt (str): query routed to the product manager.
    Returns:
    - the final evaluation.
    """
    product_manager_evaluation = product_manager_evaluation_agent.evaluate(prompt)
    return product_manager_evaluation["final_response"]

def program_manager_support_function (prompt):
    """ Program Manager Support function.
    Sends routed prompt to the program manager knowledge agent for initial response.
    Then sends that response for evaluation and correction if needed.
    Parameters:
    propt (str): query routed to the program manager.
    Returns:
    - the final evaluation.
    """
    program_manager_evaluation = program_manager_evaluation_agent.evaluate(prompt)
    return program_manager_evaluation["final_response"]

def development_engineer_support_function (prompt):
    """ Development Engineer Support function.
    Sends routed prompt to the development engineer knowledge agent for initial response.
    Then sends that response for evaluation and correction if needed.
    Parameters:
    propt (str): query routed to the development engineer.
    Returns:
    - the final evaluation.
    """
    development_engineer_evaluation = development_engineer_evaluation_agent.evaluate(prompt)
    return development_engineer_evaluation["final_response"]

def workflow():
    # Run the workflow

    print("\n*** Workflow execution started ***\n")
    # Workflow Prompt
    # ****
    workflow_prompt = "What would the development tasks for this product be?"

    # ****
    print(f"Task to complete in this workflow, workflow prompt = {workflow_prompt}")

    print("\nDefining workflow steps from the workflow prompt")
    # TODO: 12 - Implement the workflow.
    #   1. Use the 'action_planning_agent' to extract steps from the 'workflow_prompt'.
    #   2. Initialize an empty list to store 'completed_steps'.
    #   3. Loop through the extracted workflow steps:
    #      a. For each step, use the 'routing_agent' to route the step to the appropriate support function.
    #      b. Append the result to 'completed_steps'.
    #      c. Print information about the step being executed and its result.
    #   4. After the loop, print the final output of the workflow (the last completed step).

    action_agent_response = action_planning_agent.extract_steps_from_prompt(workflow_prompt)
    completed_steps = []

    for step in action_agent_response:
        print(f"Executing step: {step}")
        step_result = routing_agent.route(step)
        completed_steps.append(step_result)

    print(completed_steps[-1])

    file_path = "completed_steps.txt"
    try:
        with open(file_path, 'w') as file:
            for item in completed_steps:
                file.write(f"{item}\\n")
        print("completed_steps.txt saved loaded successfully..")
    except FileNotFoundError:
        print("Error: The file 'completed_steps.txt' was not found.")
    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    workflow()