gistfile1.txt

import os
from typing import List

import dspy
from dotenv import load_dotenv
from dspy.functional import TypedPredictor
from pydantic import BaseModel, Field

load_dotenv()

llm = dspy.OpenAI(
    # model="gpt-4-turbo-2024-04-09",
    model="gpt-3.5-turbo",
    api_key=os.environ["OPENAI_API_KEY"],
    max_tokens=3000,
)

dspy.settings.configure(lm=llm)


class Task(BaseModel):
    """Class for defining a task to be performed."""

    name: str = Field(..., description="The name of the task to be performed.")
    done: bool = Field(
        False,
        description="The status of the task. True if the task is done, False otherwise.",
    )
    result: str = Field("", description="The result of the task.")


class TaskList(BaseModel):
    """Class for defining a list of tasks."""

    list: List[Task] = Field([], description="A list of tasks to be performed.")


# given the objective, generate a list of tasks to accomplish the objective
class InitiatorAgentSignature(dspy.Signature):
    """Given an objective, generate a list of tasks in order as mentioned (tasks can be comma separated in the objective string)."""

    objective: str = dspy.InputField(desc="The overall objective to accomplish.")
    tasks_list: TaskList = dspy.OutputField(
        desc="A list of tasks (separated by a delimiter like comma, etc) to be performed in order as mentioned in the objective."
    )


# given an objective and a list of tasks, determine if a new task needs to be added to accomplish the objective
class TaskAgentSignature(dspy.Signature):
    """Given an objective and a list of tasks, determine if a new task needs to be added to accomplish the objective."""

    objective: str = dspy.InputField(desc="The overall objective to accomplish.")
    tasks_list: TaskList = dspy.InputField(
        desc="A list of tasks (separated by a delimiter like comma, etc) to be performed in order as mentioned in the objective."
    )
    add: bool = dspy.OutputField(
        desc="True if we need to add a new task to satisfy the objective."
    )
    new_task: Task = dspy.OutputField(desc="The new task to be added to the list.")


# given an objective and a task, execute the task and provide the result of the task execution. If there is a response to the overall objective and we can stop the process, stop is set to True.
class ExecutionAgentSignature(dspy.Signature):
    """Given an objective and a task, execute the task and provide the result."""

    objective: str = dspy.InputField(desc="The overall objective to accomplish.")
    task: Task = dspy.InputField(desc="The task to be executed.")
    result: str = dspy.OutputField(
        desc="The result of the task execution as instructed."
    )
    stop: bool = dspy.OutputField(
        desc="True if there is a response to the overall objective and we can stop the process."
    )


if __name__ == "__main__":
    initiator_agent = TypedPredictor(InitiatorAgentSignature)
    task_agent = TypedPredictor(TaskAgentSignature)
    execution_agent = TypedPredictor(ExecutionAgentSignature)

    OBJECTIVE = input(
        "\033[96m\033[1m"
        + "\n*****Enter the objective of your Baby AGI:*****\n"
        + "\033[0m\033[0m"
    )

    tasks_list = initiator_agent(objective=OBJECTIVE).tasks_list

    while True:
        print(
            "\033[96m\033[1m" + "\n*****Current Tasks List:*****\n" + "\033[0m\033[0m"
        )
        for task in tasks_list.list:
            print(
                f"Starting: Task: {task.name}, Done: {task.done}, Result: {task.result}"
            )

        # Find the first undone task in the list
        undone_task = next((task for task in tasks_list.list if not task.done), None)

        if undone_task:
            print(f"\033[93m\033[1m\nExecuting Task: {undone_task.name}\033[0m\033[0m")
            execution = execution_agent(objective=OBJECTIVE, task=undone_task)
            undone_task.result = execution.result
            undone_task.done = True
            print(
                f"\033[92m\033[1mTask Completed: {undone_task.name}, Result: {undone_task.result}\033[0m\033[0m"
            )

            if execution.stop:
                print(
                    "\033[96m\033[1m"
                    + "\n*****Objective Accomplished!*****\n"
                    + "\033[0m\033[0m"
                )
                print(f"Final Result: {undone_task.result}")
                break
        else:
            print(
                "\033[93m\033[1m\nChecking if additional tasks are needed...\033[0m\033[0m"
            )
            response = task_agent(objective=OBJECTIVE, tasks_list=tasks_list)

            if response.add:
                print(
                    f"\033[93m\033[1mAdding New Task: {response.new_task.name}\033[0m\033[0m"
                )
                tasks_list.list.append(response.new_task)
            else:
                print(
                    "\033[96m\033[1m"
                    + "\n*****No more tasks to add. Objective Accomplished!*****\n"
                    + "\033[0m\033[0m"
                )
                print(f"Final Result: {response}")
                break