File: babyagi.py

#!/usr/bin/env python3
from dotenv import load_dotenv

# Load default environment variables (.env)
load_dotenv()

import os
import time
import logging
from collections import deque
from typing import Dict, List
import importlib
import openai
import chromadb
import tiktoken as tiktoken
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction
from chromadb.api.types import Documents, EmbeddingFunction, Embeddings
import re

# default opt out of chromadb telemetry.
from chromadb.config import Settings

client = chromadb.Client(Settings(anonymized_telemetry=False))

# Engine configuration

# Model: GPT, LLAMA, HUMAN, etc.
LLM_MODEL = os.getenv("LLM_MODEL", os.getenv("OPENAI_API_MODEL", "gpt-3.5-turbo")).lower()

# API Keys
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
if not (LLM_MODEL.startswith("llama") or LLM_MODEL.startswith("human")):
    assert OPENAI_API_KEY, "\033[91m\033[1m" + "OPENAI_API_KEY environment variable is missing from .env" + "\033[0m\033[0m"

# Table config
RESULTS_STORE_NAME = os.getenv("RESULTS_STORE_NAME", os.getenv("TABLE_NAME", ""))
assert RESULTS_STORE_NAME, "\033[91m\033[1m" + "RESULTS_STORE_NAME environment variable is missing from .env" + "\033[0m\033[0m"

# Run configuration
INSTANCE_NAME = os.getenv("INSTANCE_NAME", os.getenv("BABY_NAME", "BabyAGI"))
COOPERATIVE_MODE = "none"
JOIN_EXISTING_OBJECTIVE = False

# Goal configuration
OBJECTIVE = os.getenv("OBJECTIVE", "")
INITIAL_TASK = os.getenv("INITIAL_TASK", os.getenv("FIRST_TASK", ""))

# Model configuration
OPENAI_TEMPERATURE = float(os.getenv("OPENAI_TEMPERATURE", 0.0))


# Extensions support begin

def can_import(module_name):
    try:
        importlib.import_module(module_name)
        return True
    except ImportError:
        return False


DOTENV_EXTENSIONS = os.getenv("DOTENV_EXTENSIONS", "").split(" ")

# Command line arguments extension
# Can override any of the above environment variables
ENABLE_COMMAND_LINE_ARGS = (
        os.getenv("ENABLE_COMMAND_LINE_ARGS", "false").lower() == "true"
)
if ENABLE_COMMAND_LINE_ARGS:
    if can_import("extensions.argparseext"):
        from extensions.argparseext import parse_arguments

        OBJECTIVE, INITIAL_TASK, LLM_MODEL, DOTENV_EXTENSIONS, INSTANCE_NAME, COOPERATIVE_MODE, JOIN_EXISTING_OBJECTIVE = parse_arguments()

# Human mode extension
# Gives human input to babyagi
if LLM_MODEL.startswith("human"):
    if can_import("extensions.human_mode"):
        from extensions.human_mode import user_input_await

# Load additional environment variables for enabled extensions
# TODO: This might override the following command line arguments as well:
#    OBJECTIVE, INITIAL_TASK, LLM_MODEL, INSTANCE_NAME, COOPERATIVE_MODE, JOIN_EXISTING_OBJECTIVE
if DOTENV_EXTENSIONS:
    if can_import("extensions.dotenvext"):
        from extensions.dotenvext import load_dotenv_extensions

        load_dotenv_extensions(DOTENV_EXTENSIONS)

# TODO: There's still work to be done here to enable people to get
# defaults from dotenv extensions, but also provide command line
# arguments to override them

# Extensions support end

print("\033[95m\033[1m" + "\n*****CONFIGURATION*****\n" + "\033[0m\033[0m")
print(f"Name  : {INSTANCE_NAME}")
print(f"Mode  : {'alone' if COOPERATIVE_MODE in ['n', 'none'] else 'local' if COOPERATIVE_MODE in ['l', 'local'] else 'distributed' if COOPERATIVE_MODE in ['d', 'distributed'] else 'undefined'}")
print(f"LLM   : {LLM_MODEL}")


# Check if we know what we are doing
assert OBJECTIVE, "\033[91m\033[1m" + "OBJECTIVE environment variable is missing from .env" + "\033[0m\033[0m"
assert INITIAL_TASK, "\033[91m\033[1m" + "INITIAL_TASK environment variable is missing from .env" + "\033[0m\033[0m"

LLAMA_MODEL_PATH = os.getenv("LLAMA_MODEL_PATH", "models/llama-13B/ggml-model.bin")
if LLM_MODEL.startswith("llama"):
    if can_import("llama_cpp"):
        from llama_cpp import Llama

        print(f"LLAMA : {LLAMA_MODEL_PATH}" + "\n")
        assert os.path.exists(LLAMA_MODEL_PATH), "\033[91m\033[1m" + f"Model can't be found." + "\033[0m\033[0m"

        CTX_MAX = 1024
        LLAMA_THREADS_NUM = int(os.getenv("LLAMA_THREADS_NUM", 8))

        print('Initialize model for evaluation')
        llm = Llama(
            model_path=LLAMA_MODEL_PATH,
            n_ctx=CTX_MAX,
            n_threads=LLAMA_THREADS_NUM,
            n_batch=512,
            use_mlock=False,
        )

        print('\nInitialize model for embedding')
        llm_embed = Llama(
            model_path=LLAMA_MODEL_PATH,
            n_ctx=CTX_MAX,
            n_threads=LLAMA_THREADS_NUM,
            n_batch=512,
            embedding=True,
            use_mlock=False,
        )

        print(
            "\033[91m\033[1m"
            + "\n*****USING LLAMA.CPP. POTENTIALLY SLOW.*****"
            + "\033[0m\033[0m"
        )
    else:
        print(
            "\033[91m\033[1m"
            + "\nLlama LLM requires package llama-cpp. Falling back to GPT-3.5-turbo."
            + "\033[0m\033[0m"
        )
        LLM_MODEL = "gpt-3.5-turbo"

if LLM_MODEL.startswith("gpt-4"):
    print(
        "\033[91m\033[1m"
        + "\n*****USING GPT-4. POTENTIALLY EXPENSIVE. MONITOR YOUR COSTS*****"
        + "\033[0m\033[0m"
    )

if LLM_MODEL.startswith("human"):
    print(
        "\033[91m\033[1m"
        + "\n*****USING HUMAN INPUT*****"
        + "\033[0m\033[0m"
    )

print("\033[94m\033[1m" + "\n*****OBJECTIVE*****\n" + "\033[0m\033[0m")
print(f"{OBJECTIVE}")

if not JOIN_EXISTING_OBJECTIVE:
    print("\033[93m\033[1m" + "\nInitial task:" + "\033[0m\033[0m" + f" {INITIAL_TASK}")
else:
    print("\033[93m\033[1m" + f"\nJoining to help the objective" + "\033[0m\033[0m")

# Configure OpenAI
openai.api_key = OPENAI_API_KEY


# Llama embedding function
class LlamaEmbeddingFunction(EmbeddingFunction):
    def __init__(self):
        return


    def __call__(self, texts: Documents) -> Embeddings:
        embeddings = []
        for t in texts:
            e = llm_embed.embed(t)
            embeddings.append(e)
        return embeddings


# Results storage using local ChromaDB
class DefaultResultsStorage:
    def __init__(self):
        logging.getLogger('chromadb').setLevel(logging.ERROR)
        # Create Chroma collection
        chroma_persist_dir = "chroma"
        chroma_client = chromadb.PersistentClient(
            settings=chromadb.config.Settings(
                persist_directory=chroma_persist_dir,
            )
        )

        metric = "cosine"
        if LLM_MODEL.startswith("llama"):
            embedding_function = LlamaEmbeddingFunction()
        else:
            embedding_function = OpenAIEmbeddingFunction(api_key=OPENAI_API_KEY)
        self.collection = chroma_client.get_or_create_collection(
            name=RESULTS_STORE_NAME,
            metadata={"hnsw:space": metric},
            embedding_function=embedding_function,
        )

    def add(self, task: Dict, result: str, result_id: str):

        # Break the function if LLM_MODEL starts with "human" (case-insensitive)
        if LLM_MODEL.startswith("human"):
            return
        # Continue with the rest of the function

        embeddings = llm_embed.embed(result) if LLM_MODEL.startswith("llama") else None
        if (
                len(self.collection.get(ids=[result_id], include=[])["ids"]) > 0
        ):  # Check if the result already exists
            self.collection.update(
                ids=result_id,
                embeddings=embeddings,
                documents=result,
                metadatas={"task": task["task_name"], "result": result},
            )
        else:
            self.collection.add(
                ids=result_id,
                embeddings=embeddings,
                documents=result,
                metadatas={"task": task["task_name"], "result": result},
            )

    def query(self, query: str, top_results_num: int) -> List[dict]:
        count: int = self.collection.count()
        if count == 0:
            return []
        results = self.collection.query(
            query_texts=query,
            n_results=min(top_results_num, count),
            include=["metadatas"]
        )
        return [item["task"] for item in results["metadatas"][0]]


# Initialize results storage
def try_weaviate():
    WEAVIATE_URL = os.getenv("WEAVIATE_URL", "")
    WEAVIATE_USE_EMBEDDED = os.getenv("WEAVIATE_USE_EMBEDDED", "False").lower() == "true"
    if (WEAVIATE_URL or WEAVIATE_USE_EMBEDDED) and can_import("extensions.weaviate_storage"):
        WEAVIATE_API_KEY = os.getenv("WEAVIATE_API_KEY", "")
        from extensions.weaviate_storage import WeaviateResultsStorage
        print("\nUsing results storage: " + "\033[93m\033[1m" + "Weaviate" + "\033[0m\033[0m")
        return WeaviateResultsStorage(OPENAI_API_KEY, WEAVIATE_URL, WEAVIATE_API_KEY, WEAVIATE_USE_EMBEDDED, LLM_MODEL, LLAMA_MODEL_PATH, RESULTS_STORE_NAME, OBJECTIVE)
    return None

def try_pinecone():
    PINECONE_API_KEY = os.getenv("PINECONE_API_KEY", "")
    if PINECONE_API_KEY and can_import("extensions.pinecone_storage"):
        PINECONE_ENVIRONMENT = os.getenv("PINECONE_ENVIRONMENT", "")
        assert (
            PINECONE_ENVIRONMENT
        ), "\033[91m\033[1m" + "PINECONE_ENVIRONMENT environment variable is missing from .env" + "\033[0m\033[0m"
        from extensions.pinecone_storage import PineconeResultsStorage
        print("\nUsing results storage: " + "\033[93m\033[1m" + "Pinecone" + "\033[0m\033[0m")
        return PineconeResultsStorage(OPENAI_API_KEY, PINECONE_API_KEY, PINECONE_ENVIRONMENT, LLM_MODEL, LLAMA_MODEL_PATH, RESULTS_STORE_NAME, OBJECTIVE)
    return None

def use_chroma():
    print("\nUsing results storage: " + "\033[93m\033[1m" + "Chroma (Default)" + "\033[0m\033[0m")
    return DefaultResultsStorage()

results_storage = try_weaviate() or try_pinecone() or use_chroma()

# Task storage supporting only a single instance of BabyAGI
class SingleTaskListStorage:
    def __init__(self):
        self.tasks = deque([])
        self.task_id_counter = 0

    def append(self, task: Dict):
        self.tasks.append(task)

    def replace(self, tasks: List[Dict]):
        self.tasks = deque(tasks)

    def popleft(self):
        return self.tasks.popleft()

    def is_empty(self):
        return False if self.tasks else True

    def next_task_id(self):
        self.task_id_counter += 1
        return self.task_id_counter

    def get_task_names(self):
        return [t["task_name"] for t in self.tasks]


# Initialize tasks storage
tasks_storage = SingleTaskListStorage()
if COOPERATIVE_MODE in ['l', 'local']:
    if can_import("extensions.ray_tasks"):
        import sys
        from pathlib import Path

        sys.path.append(str(Path(__file__).resolve().parent))
        from extensions.ray_tasks import CooperativeTaskListStorage

        tasks_storage = CooperativeTaskListStorage(OBJECTIVE)
        print("\nReplacing tasks storage: " + "\033[93m\033[1m" + "Ray" + "\033[0m\033[0m")
elif COOPERATIVE_MODE in ['d', 'distributed']:
    pass


def limit_tokens_from_string(string: str, model: str, limit: int) -> str:
    """Limits the string to a number of tokens (estimated)."""

    try:
        encoding = tiktoken.encoding_for_model(model)
    except:
        encoding = tiktoken.encoding_for_model('gpt2')  # Fallback for others.

    encoded = encoding.encode(string)

    return encoding.decode(encoded[:limit])


def openai_call(
    prompt: str,
    model: str = LLM_MODEL,
    temperature: float = OPENAI_TEMPERATURE,
    max_tokens: int = 100,
):
    while True:
        try:
            if model.lower().startswith("llama"):
                result = llm(prompt[:CTX_MAX],
                             stop=["### Human"],
                             echo=False,
                             temperature=0.2,
                             top_k=40,
                             top_p=0.95,
                             repeat_penalty=1.05,
                             max_tokens=200)
                # print('\n*****RESULT JSON DUMP*****\n')
                # print(json.dumps(result))
                # print('\n')
                return result['choices'][0]['text'].strip()
            elif model.lower().startswith("human"):
                return user_input_await(prompt)
            elif not model.lower().startswith("gpt-"):
                # Use completion API
                response = openai.Completion.create(
                    engine=model,
                    prompt=prompt,
                    temperature=temperature,
                    max_tokens=max_tokens,
                    top_p=1,
                    frequency_penalty=0,
                    presence_penalty=0,
                )
                return response.choices[0].text.strip()
            else:
                # Use 4000 instead of the real limit (4097) to give a bit of wiggle room for the encoding of roles.
                # TODO: different limits for different models.

                trimmed_prompt = limit_tokens_from_string(prompt, model, 4000 - max_tokens)

                # Use chat completion API
                messages = [{"role": "system", "content": trimmed_prompt}]
                response = openai.ChatCompletion.create(
                    model=model,
                    messages=messages,
                    temperature=temperature,
                    max_tokens=max_tokens,
                    n=1,
                    stop=None,
                )
                return response.choices[0].message.content.strip()
        except openai.error.RateLimitError:
            print(
                "   *** The OpenAI API rate limit has been exceeded. Waiting 10 seconds and trying again. ***"
            )
            time.sleep(10)  # Wait 10 seconds and try again
        except openai.error.Timeout:
            print(
                "   *** OpenAI API timeout occurred. Waiting 10 seconds and trying again. ***"
            )
            time.sleep(10)  # Wait 10 seconds and try again
        except openai.error.APIError:
            print(
                "   *** OpenAI API error occurred. Waiting 10 seconds and trying again. ***"
            )
            time.sleep(10)  # Wait 10 seconds and try again
        except openai.error.APIConnectionError:
            print(
                "   *** OpenAI API connection error occurred. Check your network settings, proxy configuration, SSL certificates, or firewall rules. Waiting 10 seconds and trying again. ***"
            )
            time.sleep(10)  # Wait 10 seconds and try again
        except openai.error.InvalidRequestError:
            print(
                "   *** OpenAI API invalid request. Check the documentation for the specific API method you are calling and make sure you are sending valid and complete parameters. Waiting 10 seconds and trying again. ***"
            )
            time.sleep(10)  # Wait 10 seconds and try again
        except openai.error.ServiceUnavailableError:
            print(
                "   *** OpenAI API service unavailable. Waiting 10 seconds and trying again. ***"
            )
            time.sleep(10)  # Wait 10 seconds and try again
        else:
            break


def task_creation_agent(
        objective: str, result: Dict, task_description: str, task_list: List[str]
):
    prompt = f"""
You are to use the result from an execution agent to create new tasks with the following objective: {objective}.
The last completed task has the result: \n{result["data"]}
This result was based on this task description: {task_description}.\n"""

    if task_list:
        prompt += f"These are incomplete tasks: {', '.join(task_list)}\n"
    prompt += "Based on the result, return a list of tasks to be completed in order to meet the objective. "
    if task_list:
        prompt += "These new tasks must not overlap with incomplete tasks. "

    prompt += """
Return one task per line in your response. The result must be a numbered list in the format:

#. First task
#. Second task

The number of each entry must be followed by a period. If your list is empty, write "There are no tasks to add at this time."
Unless your list is empty, do not include any headers before your numbered list or follow your numbered list with any other output."""

    print(f'\n*****TASK CREATION AGENT PROMPT****\n{prompt}\n')
    response = openai_call(prompt, max_tokens=2000)
    print(f'\n****TASK CREATION AGENT RESPONSE****\n{response}\n')
    new_tasks = response.split('\n')
    new_tasks_list = []
    for task_string in new_tasks:
        task_parts = task_string.strip().split(".", 1)
        if len(task_parts) == 2:
            task_id = ''.join(s for s in task_parts[0] if s.isnumeric())
            task_name = re.sub(r'[^\w\s_]+', '', task_parts[1]).strip()
            if task_name.strip() and task_id.isnumeric():
                new_tasks_list.append(task_name)
            # print('New task created: ' + task_name)

    out = [{"task_name": task_name} for task_name in new_tasks_list]
    return out


def prioritization_agent():
    task_names = tasks_storage.get_task_names()
    bullet_string = '\n'

    prompt = f"""
You are tasked with prioritizing the following tasks: {bullet_string + bullet_string.join(task_names)}
Consider the ultimate objective of your team: {OBJECTIVE}.
Tasks should be sorted from highest to lowest priority, where higher-priority tasks are those that act as pre-requisites or are more essential for meeting the objective.
Do not remove any tasks. Return the ranked tasks as a numbered list in the format:

#. First task
#. Second task

The entries must be consecutively numbered, starting with 1. The number of each entry must be followed by a period.
Do not include any headers before your ranked list or follow your list with any other output."""

    print(f'\n****TASK PRIORITIZATION AGENT PROMPT****\n{prompt}\n')
    response = openai_call(prompt, max_tokens=2000)
    print(f'\n****TASK PRIORITIZATION AGENT RESPONSE****\n{response}\n')
    if not response:
        print('Received empty response from priotritization agent. Keeping task list unchanged.')
        return
    new_tasks = response.split("\n") if "\n" in response else [response]
    new_tasks_list = []
    for task_string in new_tasks:
        task_parts = task_string.strip().split(".", 1)
        if len(task_parts) == 2:
            task_id = ''.join(s for s in task_parts[0] if s.isnumeric())
            task_name = re.sub(r'[^\w\s_]+', '', task_parts[1]).strip()
            if task_name.strip():
                new_tasks_list.append({"task_id": task_id, "task_name": task_name})

    return new_tasks_list


# Execute a task based on the objective and five previous tasks
def execution_agent(objective: str, task: str) -> str:
    """
    Executes a task based on the given objective and previous context.

    Args:
        objective (str): The objective or goal for the AI to perform the task.
        task (str): The task to be executed by the AI.

    Returns:
        str: The response generated by the AI for the given task.

    """

    context = context_agent(query=objective, top_results_num=5)
    # print("\n****RELEVANT CONTEXT****\n")
    # print(context)
    # print('')
    prompt = f'Perform one task based on the following objective: {objective}.\n'
    if context:
        prompt += 'Take into account these previously completed tasks:' + '\n'.join(context)
    prompt += f'\nYour task: {task}\nResponse:'
    return openai_call(prompt, max_tokens=2000)


# Get the top n completed tasks for the objective
def context_agent(query: str, top_results_num: int):
    """
    Retrieves context for a given query from an index of tasks.

    Args:
        query (str): The query or objective for retrieving context.
        top_results_num (int): The number of top results to retrieve.

    Returns:
        list: A list of tasks as context for the given query, sorted by relevance.

    """
    results = results_storage.query(query=query, top_results_num=top_results_num)
    # print("****RESULTS****")
    # print(results)
    return results


# Add the initial task if starting new objective
if not JOIN_EXISTING_OBJECTIVE:
    initial_task = {
        "task_id": tasks_storage.next_task_id(),
        "task_name": INITIAL_TASK
    }
    tasks_storage.append(initial_task)


def main():
    loop = True
    while loop:
        # As long as there are tasks in the storage...
        if not tasks_storage.is_empty():
            # Print the task list
            print("\033[95m\033[1m" + "\n*****TASK LIST*****\n" + "\033[0m\033[0m")
            for t in tasks_storage.get_task_names():
                print(" • " + str(t))

            # Step 1: Pull the first incomplete task
            task = tasks_storage.popleft()
            print("\033[92m\033[1m" + "\n*****NEXT TASK*****\n" + "\033[0m\033[0m")
            print(str(task["task_name"]))

            # Send to execution function to complete the task based on the context
            result = execution_agent(OBJECTIVE, str(task["task_name"]))
            print("\033[93m\033[1m" + "\n*****TASK RESULT*****\n" + "\033[0m\033[0m")
            print(result)

            # Step 2: Enrich result and store in the results storage
            # This is where you should enrich the result if needed
            enriched_result = {
                "data": result
            }
            # extract the actual result from the dictionary
            # since we don't do enrichment currently
            # vector = enriched_result["data"]

            result_id = f"result_{task['task_id']}"

            results_storage.add(task, result, result_id)

            # Step 3: Create new tasks and re-prioritize task list
            # only the main instance in cooperative mode does that
            new_tasks = task_creation_agent(
                OBJECTIVE,
                enriched_result,
                task["task_name"],
                tasks_storage.get_task_names(),
            )

            print('Adding new tasks to task_storage')
            for new_task in new_tasks:
                new_task.update({"task_id": tasks_storage.next_task_id()})
                print(str(new_task))
                tasks_storage.append(new_task)

            if not JOIN_EXISTING_OBJECTIVE:
                prioritized_tasks = prioritization_agent()
                if prioritized_tasks:
                    tasks_storage.replace(prioritized_tasks)

            # Sleep a bit before checking the task list again
            time.sleep(5)
        else:
            print('Done.')
            loop = False


if __name__ == "__main__":
    main()



File: tools/results_browser.py

#!/usr/bin/env python3
import os
import curses
import argparse
import openai
import pinecone
from dotenv import load_dotenv
import textwrap

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
assert OPENAI_API_KEY, "OPENAI_API_KEY environment variable is missing from .env"

PINECONE_API_KEY = os.getenv("PINECONE_API_KEY", "")
assert PINECONE_API_KEY, "PINECONE_API_KEY environment variable is missing from .env"

PINECONE_ENVIRONMENT = os.getenv("PINECONE_ENVIRONMENT", "us-east1-gcp")
assert PINECONE_ENVIRONMENT, "PINECONE_ENVIRONMENT environment variable is missing from .env"

# Table config
PINECONE_TABLE_NAME = os.getenv("TABLE_NAME", "")
assert PINECONE_TABLE_NAME, "TABLE_NAME environment variable is missing from .env"

# Function to query records from the Pinecone index
def query_records(index, query, top_k=1000):
    results = index.query(query, top_k=top_k, include_metadata=True)
    return [{"name": f"{task.metadata['task']}", "result": f"{task.metadata['result']}"} for task in results.matches]

# Get embedding for the text
def get_ada_embedding(text):
    return openai.Embedding.create(input=[text], model="text-embedding-ada-002")["data"][0]["embedding"]

def draw_tasks(stdscr, tasks, scroll_pos, selected):
    y = 0
    h, w = stdscr.getmaxyx()
    for idx, task in enumerate(tasks[scroll_pos:], start=scroll_pos):
        if y >= h:
            break
        task_name = f'{task["name"]}'
        truncated_str = task_name[:w-1]
        if idx == selected:
            stdscr.addstr(y, 0, truncated_str, curses.A_REVERSE)
        else:
            stdscr.addstr(y, 0, truncated_str)
        y += 1

def draw_result(stdscr, task):
    task_name = f'Task: {task["name"]}'
    task_result = f'Result: {task["result"]}'

    _, w = stdscr.getmaxyx()
    task_name_wrapped = textwrap.wrap(task_name, width=w)

    for i, line in enumerate(task_name_wrapped):
        stdscr.addstr(i, 0, line)
    
    y, _ = stdscr.getyx()
    stdscr.addstr(y+1, 0, '------------------')
    stdscr.addstr(y+2, 0, task_result)

def draw_summary(stdscr, objective, tasks, start, num):
    stdscr.box()
    summary_text = f'{len(tasks)} tasks ({start}-{num}) | {objective}'
    stdscr.addstr(1, 1, summary_text[:stdscr.getmaxyx()[1] - 2])

def main(stdscr):
    # Configure OpenAI
    openai.api_key = OPENAI_API_KEY

    # Initialize Pinecone
    pinecone.init(api_key=PINECONE_API_KEY)

    # Connect to the objective index
    index = pinecone.Index(PINECONE_TABLE_NAME)

    curses.curs_set(0)
    stdscr.timeout(1000)

    h, w = stdscr.getmaxyx()
    left_w = w // 2
    visible_lines = h - 3

    scroll_pos = 0
    selected = 0

    # Parse command-line arguments
    parser = argparse.ArgumentParser(description="Query Pinecone index using a string.")
    parser.add_argument('objective', nargs='*', metavar='<objective>', help='''
    main objective description. Doesn\'t need to be quoted.
    if not specified, get objective from environment.
    ''', default=[os.getenv("OBJECTIVE", "")])
    args = parser.parse_args()

    # Query records from the index
    objective = ' '.join(args.objective).strip().replace("\n", " ")
    retrieved_tasks = query_records(index, get_ada_embedding(objective))

    while True:
        stdscr.clear()
        draw_tasks(stdscr.subwin(h-3, left_w, 0, 0), retrieved_tasks, scroll_pos, selected)
        draw_result(stdscr.subwin(h, w - left_w, 0, left_w), retrieved_tasks[selected])
        draw_summary(stdscr.subwin(3, left_w, h - 3, 0), objective, retrieved_tasks, scroll_pos+1, scroll_pos+h-3)

        stdscr.refresh()
        key = stdscr.getch()

        if key == ord('q') or key == 27:
            break
        elif key == curses.KEY_UP and selected > 0:
            selected -= 1
            if selected < scroll_pos:
                scroll_pos -= 1
        elif key == curses.KEY_DOWN and selected < len(retrieved_tasks) - 1:
            selected += 1
            if selected - scroll_pos >= visible_lines:
                scroll_pos += 1

curses.wrapper(main)


File: tools/monitor.py

#!/usr/bin/env python3
import sys
import time
import curses

from pathlib import Path
sys.path.append(str(Path(__file__).resolve().parent.parent))
from extensions.ray_objectives import CooperativeObjectivesListStorage
from extensions.ray_tasks import CooperativeTaskListStorage

def print_buffer(stdscr, lines):
    stdscr.clear()
    y = 0
    x = 0
    for line in lines:
        stdscr.addstr(y, x, line)
        y += 1
    stdscr.refresh()

def main(stdscr):
    objectives = CooperativeObjectivesListStorage()
    while True:
        objectives_list = objectives.get_objective_names()
        buffer = []
        if not objectives_list:
            buffer.append("No objectives")
        for objective in objectives_list:
            buffer.append("-----------------")
            buffer.append(f"Objective: {objective}")
            buffer.append("-----------------")
            tasks = CooperativeTaskListStorage(objective)
            tasks_list = tasks.get_task_names()
            buffer.append(f"Tasks:")
            for t in tasks_list:
                buffer.append(f" * {t}")
            buffer.append("-----------------")
        print_buffer(stdscr, buffer)
        time.sleep(30)

curses.wrapper(main)



File: tools/results.py

#!/usr/bin/env python3
import os
import argparse
import openai
import pinecone
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
assert OPENAI_API_KEY, "OPENAI_API_KEY environment variable is missing from .env"

PINECONE_API_KEY = os.getenv("PINECONE_API_KEY", "")
assert PINECONE_API_KEY, "PINECONE_API_KEY environment variable is missing from .env"

PINECONE_ENVIRONMENT = os.getenv("PINECONE_ENVIRONMENT", "us-east1-gcp")
assert PINECONE_ENVIRONMENT, "PINECONE_ENVIRONMENT environment variable is missing from .env"

# Table config
PINECONE_TABLE_NAME = os.getenv("TABLE_NAME", "")
assert PINECONE_TABLE_NAME, "TABLE_NAME environment variable is missing from .env"

# Function to query records from the Pinecone index
def query_records(index, query, top_k=1000):
    results = index.query(query, top_k=top_k, include_metadata=True)
    return [f"{task.metadata['task']}:\n{task.metadata['result']}\n------------------" for task in results.matches]

# Get embedding for the text
def get_ada_embedding(text):
    text = text.replace("\n", " ")
    return openai.Embedding.create(input=[text], model="text-embedding-ada-002")["data"][0]["embedding"]

def main():
    # Parse command-line arguments
    parser = argparse.ArgumentParser(description="Query Pinecone index using a string.")
    parser.add_argument('objective', nargs='*', metavar='<objective>', help='''
    main objective description. Doesn\'t need to be quoted.
    if not specified, get objective from environment.
    ''', default=[os.getenv("OBJECTIVE", "")])
    args = parser.parse_args()

    # Configure OpenAI
    openai.api_key = OPENAI_API_KEY

    # Initialize Pinecone
    pinecone.init(api_key=PINECONE_API_KEY)

    # Connect to the objective index
    index = pinecone.Index(PINECONE_TABLE_NAME)

    # Query records from the index
    query = get_ada_embedding(' '.join(args.objective).strip())
    retrieved_tasks = query_records(index, query)
    for r in retrieved_tasks:
        print(r)

if __name__ == "__main__":
    main()



File: tools/__init__.py




File: classic/BabyCatAGI.py

###### This is a modified version of OG BabyAGI, called BabyCatAGI (future modifications will follow the pattern "Baby<animal>AGI"). This version requires GPT-4, it's very slow, and often errors out.######
######IMPORTANT NOTE: I'm sharing this as a framework to build on top of (with lots of errors for improvement), to facilitate discussion around how to improve these. This is NOT for people who are looking for a complete solution that's ready to use. ######

import openai
import time
import requests
from bs4 import BeautifulSoup
from collections import deque
from typing import Dict, List
import re
import ast
import json
from serpapi import GoogleSearch

### SET THESE 4 VARIABLES ##############################

# Add your API keys here
OPENAI_API_KEY = ""
SERPAPI_API_KEY = "" #If you include SERPAPI KEY, this will enable web-search. If you don't, it will autoatically remove web-search capability.

# Set variables
OBJECTIVE = "Research experts at scaling NextJS and their Twitter accounts."
YOUR_FIRST_TASK = "Develop a task list." #you can provide additional instructions here regarding the task list.

### UP TO HERE ##############################

# Configure OpenAI and SerpAPI client
openai.api_key = OPENAI_API_KEY
if SERPAPI_API_KEY:
  serpapi_client = GoogleSearch({"api_key": SERPAPI_API_KEY})
  websearch_var = "[web-search] "
else:
  websearch_var = ""

# Initialize task list
task_list = []

# Initialize session_summary
session_summary = ""

### Task list functions ##############################
def add_task(task: Dict):
    task_list.append(task)

def get_task_by_id(task_id: int):
    for task in task_list:
        if task["id"] == task_id:
            return task
    return None

def get_completed_tasks():
    return [task for task in task_list if task["status"] == "complete"]


# Print task list and session summary
def print_tasklist():
  print("\033[95m\033[1m" + "\n*****TASK LIST*****\n" + "\033[0m")
  for t in task_list:
      dependent_task = ""
      if t['dependent_task_ids']:
          dependent_task = f"\033[31m<dependencies: {', '.join([f'#{dep_id}' for dep_id in t['dependent_task_ids']])}>\033[0m"
      status_color = "\033[32m" if t['status'] == "complete" else "\033[31m"
      print(f"\033[1m{t['id']}\033[0m: {t['task']} {status_color}[{t['status']}]\033[0m \033[93m[{t['tool']}] {dependent_task}\033[0m")

### Tool functions ##############################
def text_completion_tool(prompt: str):
    messages = [
        {"role": "user", "content": prompt}
    ]

    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages,
        temperature=0.2,
        max_tokens=1500,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )

    return response.choices[0].message['content'].strip()


def web_search_tool(query: str):
    search_params = {
        "engine": "google",
        "q": query,
        "api_key": SERPAPI_API_KEY,
        "num":5 #edit this up or down for more results, though higher often results in OpenAI rate limits
    }
    search_results = GoogleSearch(search_params)
    search_results = search_results.get_dict()
    try:
      search_results = search_results["organic_results"]
    except:
      search_results = {}
    search_results = simplify_search_results(search_results)
    print("\033[90m\033[3m" + "Completed search. Now scraping results.\n" + "\033[0m")
    results = "";
    # Loop through the search results
    for result in search_results:
        # Extract the URL from the result
        url = result.get('link')
        # Call the web_scrape_tool function with the URL
        print("\033[90m\033[3m" + "Scraping: "+url+"" + "...\033[0m")
        content = web_scrape_tool(url, task)
        print("\033[90m\033[3m" +str(content[0:100])[0:100]+"...\n" + "\033[0m")
        results += str(content)+". "
    

    return results


def simplify_search_results(search_results):
    simplified_results = []
    for result in search_results:
        simplified_result = {
            "position": result.get("position"),
            "title": result.get("title"),
            "link": result.get("link"),
            "snippet": result.get("snippet")
        }
        simplified_results.append(simplified_result)
    return simplified_results


def web_scrape_tool(url: str, task:str):
    content = fetch_url_content(url)
    if content is None:
        return None

    text = extract_text(content)
    print("\033[90m\033[3m"+"Scrape completed. Length:" +str(len(text))+".Now extracting relevant info..."+"...\033[0m")
    info = extract_relevant_info(OBJECTIVE, text[0:5000], task)
    links = extract_links(content)

    #result = f"{info} URLs: {', '.join(links)}"
    result = info
    
    return result

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36"
}

def fetch_url_content(url: str):
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        return response.content
    except requests.exceptions.RequestException as e:
        print(f"Error while fetching the URL: {e}")
        return ""

def extract_links(content: str):
    soup = BeautifulSoup(content, "html.parser")
    links = [link.get('href') for link in soup.findAll('a', attrs={'href': re.compile("^https?://")})]
    return links

def extract_text(content: str):
    soup = BeautifulSoup(content, "html.parser")
    text = soup.get_text(strip=True)
    return text



def extract_relevant_info(objective, large_string, task):
    chunk_size = 3000
    overlap = 500
    notes = ""
    
    for i in range(0, len(large_string), chunk_size - overlap):
        chunk = large_string[i:i + chunk_size]
        
        messages = [
            {"role": "system", "content": f"Objective: {objective}\nCurrent Task:{task}"},
            {"role": "user", "content": f"Analyze the following text and extract information relevant to our objective and current task, and only information relevant to our objective and current task. If there is no relevant information do not say that there is no relevant informaiton related to our objective. ### Then, update or start our notes provided here (keep blank if currently blank): {notes}.### Text to analyze: {chunk}.### Updated Notes:"}
        ]

        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=800,
            n=1,
            stop="###",
            temperature=0.7,
        )

        notes += response.choices[0].message['content'].strip()+". ";
    
    return notes

### Agent functions ##############################


def execute_task(task, task_list, OBJECTIVE):
    global task_id_counter
    # Check if dependent_task_ids is not empty
    if task["dependent_task_ids"]:
      all_dependent_tasks_complete = True
      for dep_id in task["dependent_task_ids"]:
          dependent_task = get_task_by_id(dep_id)
          if not dependent_task or dependent_task["status"] != "complete":
              all_dependent_tasks_complete = False
              break
  
        
    # Execute task
    print("\033[92m\033[1m"+"\n*****NEXT TASK*****\n"+"\033[0m\033[0m")
    print(str(task['id'])+": "+str(task['task'])+" ["+str(task['tool']+"]"))
    task_prompt = f"Complete your assigned task based on the objective and only based on information provided in the dependent task output, if provided. Your objective: {OBJECTIVE}. Your task: {task['task']}"
    if task["dependent_task_ids"]:
      dependent_tasks_output = ""
      for dep_id in task["dependent_task_ids"]:
          dependent_task_output = get_task_by_id(dep_id)["output"]
          dependent_task_output = dependent_task_output[0:2000]
          dependent_tasks_output += f" {dependent_task_output}"
      task_prompt += f" Your dependent tasks output: {dependent_tasks_output}\n OUTPUT:"

    # Use tool to complete the task
    if task["tool"] == "text-completion":
        task_output = text_completion_tool(task_prompt)
    elif task["tool"] == "web-search":
        task_output = web_search_tool(str(task['task']))
    elif task["tool"] == "web-scrape":
        task_output = web_scrape_tool(str(task['task']))

    # Find task index in the task_list
    task_index = next((i for i, t in enumerate(task_list) if t["id"] == task["id"]), None)

    # Mark task as complete and save output
    task_list[task_index]["status"] = "complete"
    task_list[task_index]["output"] = task_output

    # Print task output
    print("\033[93m\033[1m"+"\nTask Output:"+"\033[0m\033[0m")
    print(task_output)

    # Add task output to session_summary
    global session_summary
    session_summary += f"\n\nTask {task['id']} - {task['task']}:\n{task_output}"



task_list = []

def task_creation_agent(objective: str) -> List[Dict]:
    global task_list
    minified_task_list = [{k: v for k, v in task.items() if k != "result"} for task in task_list]

    prompt = (
        f"You are a task creation AI tasked with creating a list of tasks as a JSON array, considering the ultimate objective of your team: {OBJECTIVE}. "
        f"Create new tasks based on the objective. Limit tasks types to those that can be completed with the available tools listed below. Task description should be detailed."
        f"Current tool option is [text-completion] {websearch_var} and only." # web-search is added automatically if SERPAPI exists
        f"For tasks using [web-search], provide the search query, and only the search query to use (eg. not 'research waterproof shoes, but 'waterproof shoes')"
        f"dependent_task_ids should always be an empty array, or an array of numbers representing the task ID it should pull results from."
        f"Make sure all task IDs are in chronological order.\n"
        f"The last step is always to provide a final summary report including tasks executed and summary of knowledge acquired.\n"
        f"Do not create any summarizing steps outside of the last step..\n"
        f"An example of the desired output format is: "
        "[{\"id\": 1, \"task\": \"https://untapped.vc\", \"tool\": \"web-scrape\", \"dependent_task_ids\": [], \"status\": \"incomplete\", \"result\": null, \"result_summary\": null}, {\"id\": 2, \"task\": \"Consider additional insights that can be reasoned from the results of...\", \"tool\": \"text-completion\", \"dependent_task_ids\": [1], \"status\": \"incomplete\", \"result\": null, \"result_summary\": null}, {\"id\": 3, \"task\": \"Untapped Capital\", \"tool\": \"web-search\", \"dependent_task_ids\": [], \"status\": \"incomplete\", \"result\": null, \"result_summary\": null}].\n"
        f"JSON TASK LIST="
    )

    print("\033[90m\033[3m" + "\nInitializing...\n" + "\033[0m")
    print("\033[90m\033[3m" + "Analyzing objective...\n" + "\033[0m")
    print("\033[90m\033[3m" + "Running task creation agent...\n" + "\033[0m")
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": "You are a task creation AI."
            },
            {
                "role": "user",
                "content": prompt
            }
        ],
        temperature=0,
        max_tokens=1500,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )

    # Extract the content of the assistant's response and parse it as JSON
    result = response["choices"][0]["message"]["content"]
    print("\033[90m\033[3m" + "\nDone!\n" + "\033[0m")
    try:
        task_list = json.loads(result)
    except Exception as error:
        print(error)

    return task_list

##### START MAIN LOOP########

#Print OBJECTIVE
print("\033[96m\033[1m"+"\n*****OBJECTIVE*****\n"+"\033[0m\033[0m")
print(OBJECTIVE)

# Initialize task_id_counter
task_id_counter = 1

# Run the task_creation_agent to create initial tasks
task_list = task_creation_agent(OBJECTIVE)
print_tasklist()

# Execute tasks in order
while len(task_list) > 0:
    for task in task_list:
        if task["status"] == "incomplete":
            execute_task(task, task_list, OBJECTIVE)
            print_tasklist()
            break

# Print session summary
print("\033[96m\033[1m"+"\n*****SESSION SUMMARY*****\n"+"\033[0m\033[0m")
print(session_summary)



File: classic/BabyDeerAGI.py

###### This is a modified version of OG BabyAGI, called BabyDeerAGI (modifications will follow the pattern "Baby<animal>AGI").######
######IMPORTANT NOTE: I'm sharing this as a framework to build on top of (with lots of room for improvement), to facilitate discussion around how to improve these. This is NOT for people who are looking for a complete solution that's ready to use. ######

import openai
import time
from datetime import datetime
import requests
from bs4 import BeautifulSoup
from collections import deque
from typing import Dict, List
import re
import ast
import json
from serpapi import GoogleSearch
from concurrent.futures import ThreadPoolExecutor
import time

### SET THESE 4 VARIABLES ##############################

# Add your API keys here
OPENAI_API_KEY = ""
SERPAPI_API_KEY = "" #[optional] web-search becomes available automatically when serpapi api key is provided

# Set variables
OBJECTIVE = "Research recent AI news and write a poem about your findings in the style of shakespeare."

#turn on user input (change to "True" to turn on user input tool)
user_input=False

### UP TO HERE ##############################

# Configure OpenAI and SerpAPI client
openai.api_key = OPENAI_API_KEY
if SERPAPI_API_KEY:
  serpapi_client = GoogleSearch({"api_key": SERPAPI_API_KEY})
  websearch_var = "[web-search] "
else:
  websearch_var = ""

if user_input == True:
  user_input_var = "[user-input]"
else:
  user_input_var = ""


# Initialize task list
task_list = []

# Initialize session_summary
session_summary = "OBJECTIVE: "+OBJECTIVE+"\n\n"

### Task list functions ##############################
def get_task_by_id(task_id: int):
    for task in task_list:
        if task["id"] == task_id:
            return task
    return None

# Print task list and session summary
def print_tasklist():
  p_tasklist="\033[95m\033[1m" + "\n*****TASK LIST*****\n" + "\033[0m"
  for t in task_list:
      dependent_task = ""
      if t['dependent_task_ids']:
          dependent_task = f"\033[31m<dependencies: {', '.join([f'#{dep_id}' for dep_id in t['dependent_task_ids']])}>\033[0m"
      status_color = "\033[32m" if t['status'] == "complete" else "\033[31m"
      p_tasklist+= f"\033[1m{t['id']}\033[0m: {t['task']} {status_color}[{t['status']}]\033[0m \033[93m[{t['tool']}] {dependent_task}\033[0m\n"
  print(p_tasklist)

### Tool functions ##############################
def text_completion_tool(prompt: str):
    messages = [
        {"role": "user", "content": prompt}
    ]
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages,
        temperature=0.2,
        max_tokens=1500,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )

    return response.choices[0].message['content'].strip()


def user_input_tool(prompt: str):
    val = input(f"\n{prompt}\nYour response: ")
    return str(val)


def web_search_tool(query: str , dependent_tasks_output : str):
    
    if dependent_tasks_output != "":
      dependent_task = f"Use the dependent task output below as reference to help craft the correct search query for the provided task above. Dependent task output:{dependent_tasks_output}."
    else:
      dependent_task = "."
    query = text_completion_tool("You are an AI assistant tasked with generating a Google search query based on the following task: "+query+". If the task looks like a search query, return the identical search query as your response. " + dependent_task + "\nSearch Query:")
    print("\033[90m\033[3m"+"Search query: " +str(query)+"\033[0m")
    search_params = {
        "engine": "google",
        "q": query,
        "api_key": SERPAPI_API_KEY,
        "num":3 #edit this up or down for more results, though higher often results in OpenAI rate limits
    }
    search_results = GoogleSearch(search_params)
    search_results = search_results.get_dict()
    try:
      search_results = search_results["organic_results"]
    except:
      search_results = {}
    search_results = simplify_search_results(search_results)
    print("\033[90m\033[3m" + "Completed search. Now scraping results.\n" + "\033[0m")
    results = "";
    # Loop through the search results
    for result in search_results:
        # Extract the URL from the result
        url = result.get('link')
        # Call the web_scrape_tool function with the URL
        print("\033[90m\033[3m" + "Scraping: "+url+"" + "...\033[0m")
        content = web_scrape_tool(url, task)
        print("\033[90m\033[3m" +str(content[0:100])[0:100]+"...\n" + "\033[0m")
        results += str(content)+". "
    
    results = text_completion_tool(f"You are an expert analyst. Rewrite the following information as one report without removing any facts.\n###INFORMATION:{results}.\n###REPORT:")
    return results


def simplify_search_results(search_results):
    simplified_results = []
    for result in search_results:
        simplified_result = {
            "position": result.get("position"),
            "title": result.get("title"),
            "link": result.get("link"),
            "snippet": result.get("snippet")
        }
        simplified_results.append(simplified_result)
    return simplified_results


def web_scrape_tool(url: str, task:str):
    content = fetch_url_content(url)
    if content is None:
        return None

    text = extract_text(content)
    print("\033[90m\033[3m"+"Scrape completed. Length:" +str(len(text))+".Now extracting relevant info..."+"...\033[0m")
    info = extract_relevant_info(OBJECTIVE, text[0:5000], task)
    links = extract_links(content)

    #result = f"{info} URLs: {', '.join(links)}"
    result = info
    
    return result

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36"
}

def fetch_url_content(url: str):
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        return response.content
    except requests.exceptions.RequestException as e:
        print(f"Error while fetching the URL: {e}")
        return ""

def extract_links(content: str):
    soup = BeautifulSoup(content, "html.parser")
    links = [link.get('href') for link in soup.findAll('a', attrs={'href': re.compile("^https?://")})]
    return links

def extract_text(content: str):
    soup = BeautifulSoup(content, "html.parser")
    text = soup.get_text(strip=True)
    return text



def extract_relevant_info(objective, large_string, task):
    chunk_size = 3000
    overlap = 500
    notes = ""
    
    for i in range(0, len(large_string), chunk_size - overlap):
        chunk = large_string[i:i + chunk_size]
        
        messages = [
            {"role": "system", "content": f"You are an AI assistant."},
            {"role": "user", "content": f"You are an expert AI research assistant tasked with creating or updating the current notes. If the current note is empty, start a current-notes section by exracting relevant data to the task and objective from the chunk of text to analyze. If there is a current note, add new relevant info frol the chunk of text to analyze. Make sure the new or combined notes is comprehensive and well written. Here's the current chunk of text to analyze: {chunk}. ### Here is the current task: {task}.### For context, here is the objective: {OBJECTIVE}.### Here is the data we've extraced so far that you need to update: {notes}.### new-or-updated-note:"}
        ]

        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=800,
            n=1,
            stop="###",
            temperature=0.7,
        )

        notes += response.choices[0].message['content'].strip()+". ";
    
    return notes

### Agent functions ##############################


def execute_task(task, task_list, OBJECTIVE):
    
    global session_summary
    global task_id_counter
    # Check if dependent_task_ids is not empty
    if task["dependent_task_ids"]:
      all_dependent_tasks_complete = True
      for dep_id in task["dependent_task_ids"]:
          dependent_task = get_task_by_id(dep_id)
          if not dependent_task or dependent_task["status"] != "complete":
              all_dependent_tasks_complete = False
              break
  
        
    # Execute task
    p_nexttask="\033[92m\033[1m"+"\n*****NEXT TASK ID:"+str(task['id'])+"*****\n"+"\033[0m\033[0m"
    p_nexttask += str(task['id'])+": "+str(task['task'])+" ["+str(task['tool']+"]")
    print(p_nexttask)
    task_prompt = f"Complete your assigned task based on the objective and only based on information provided in the dependent task output, if provided. \n###\nYour objective: {OBJECTIVE}. \n###\nYour task: {task['task']}"
    if task["dependent_task_ids"]:
      dependent_tasks_output = ""
      for dep_id in task["dependent_task_ids"]:
          dependent_task_output = get_task_by_id(dep_id)["output"]
          dependent_task_output = dependent_task_output[0:2000]
          dependent_tasks_output += f" {dependent_task_output}"
      task_prompt += f" \n###\ndependent tasks output: {dependent_tasks_output}  \n###\nYour task: {task['task']}\n###\nRESPONSE:"
    else:
      dependent_tasks_output="."

    # Use tool to complete the task
    if task["tool"] == "text-completion":
        task_output = text_completion_tool(task_prompt)
    elif task["tool"] == "web-search":
        task_output = web_search_tool(str(task['task']),str(dependent_tasks_output))
    elif task["tool"] == "web-scrape":
        task_output = web_scrape_tool(str(task['task']))
    elif task["tool"] == "user-input":
        task_output = user_input_tool(str(task['task']))

    

    # Find task index in the task_list
    task_index = next((i for i, t in enumerate(task_list) if t["id"] == task["id"]), None)

    # Mark task as complete and save output
    task_list[task_index]["status"] = "complete"
    task_list[task_index]["output"] = task_output

    # Print task output
    print("\033[93m\033[1m"+"\nTask Output (ID:"+str(task['id'])+"):"+"\033[0m\033[0m")
    print(task_output)
    # Add task output to session_summary
    session_summary += f"\n\nTask {task['id']} - {task['task']}:\n{task_output}"

def task_ready_to_run(task, task_list):
    return all([get_task_by_id(dep_id)["status"] == "complete" for dep_id in task["dependent_task_ids"]])


task_list = []

def task_creation_agent(objective: str) -> List[Dict]:
    global task_list
    minified_task_list = [{k: v for k, v in task.items() if k != "result"} for task in task_list]

    prompt = (
        f"You are an expert task creation AI tasked with creating a  list of tasks as a JSON array, considering the ultimate objective of your team: {OBJECTIVE}. "
        f"Create new tasks based on the objective. Limit tasks types to those that can be completed with the available tools listed below. Task description should be detailed."
        f"Current tool options are [text-completion] {websearch_var} {user_input_var}." # web-search is added automatically if SERPAPI exists
        f"For tasks using [web-search], provide the search query, and only the search query to use (eg. not 'research waterproof shoes, but 'waterproof shoes'). Result will be a summary of relevant information from the first few articles."
        f"When requiring multiple searches, use the [web-search] multiple times. This tool will use the dependent task result to generate the search query if necessary."
        f"Use [user-input] sparingly and only if you need to ask a question to the user who set up the objective. The task description should be the question you want to ask the user.')"
        f"dependent_task_ids should always be an empty array, or an array of numbers representing the task ID it should pull results from."
        f"Make sure all task IDs are in chronological order.\n"
        f"EXAMPLE OBJECTIVE=Look up AI news from today (May 27, 2023) and write a poem."
        "TASK LIST=[{\"id\":1,\"task\":\"AI news today\",\"tool\":\"web-search\",\"dependent_task_ids\":[],\"status\":\"incomplete\",\"result\":null,\"result_summary\":null},{\"id\":2,\"task\":\"Extract key points from AI news articles\",\"tool\":\"text-completion\",\"dependent_task_ids\":[1],\"status\":\"incomplete\",\"result\":null,\"result_summary\":null},{\"id\":3,\"task\":\"Generate a list of AI-related words and phrases\",\"tool\":\"text-completion\",\"dependent_task_ids\":[2],\"status\":\"incomplete\",\"result\":null,\"result_summary\":null},{\"id\":4,\"task\":\"Write a poem using AI-related words and phrases\",\"tool\":\"text-completion\",\"dependent_task_ids\":[3],\"status\":\"incomplete\",\"result\":null,\"result_summary\":null},{\"id\":5,\"task\":\"Final summary report\",\"tool\":\"text-completion\",\"dependent_task_ids\":[1,2,3,4],\"status\":\"incomplete\",\"result\":null,\"result_summary\":null}]"
        f"OBJECTIVE={OBJECTIVE}"
        f"TASK LIST="
    )

    print("\033[90m\033[3m" + "\nInitializing...\n" + "\033[0m")
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {
                "role": "system",
                "content": "You are a task creation AI."
            },
            {
                "role": "user",
                "content": prompt
            }
        ],
        temperature=0,
        max_tokens=1500,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )

    # Extract the content of the assistant's response and parse it as JSON
    result = response["choices"][0]["message"]["content"]
    try:
        task_list = json.loads(result)
    except Exception as error:
        print(error)

    return task_list

##### START MAIN LOOP########

#Print OBJECTIVE
print("\033[96m\033[1m"+"\n*****OBJECTIVE*****\n"+"\033[0m\033[0m")
print(OBJECTIVE)

# Initialize task_id_counter
task_id_counter = 1

# Run the task_creation_agent to create initial tasks
task_list = task_creation_agent(OBJECTIVE)
print_tasklist()

# Create a ThreadPoolExecutor
with ThreadPoolExecutor() as executor:
    while True:
        tasks_submitted = False
        for task in task_list:
            if task["status"] == "incomplete" and task_ready_to_run(task, task_list):
                future = executor.submit(execute_task, task, task_list, OBJECTIVE)
                task["status"] = "running"
                tasks_submitted = True

        if not tasks_submitted and all(task["status"] == "complete" for task in task_list):
            break

        time.sleep(5)

# Print session summary
print("\033[96m\033[1m"+"\n*****SAVING FILE...*****\n"+"\033[0m\033[0m")
file = open(f'output/output_{datetime.now().strftime("%d_%m_%Y_%H_%M_%S")}.txt', 'w')
file.write(session_summary)
file.close()
print("...file saved.")
print("END")



File: classic/BabyBeeAGI.py

###### This is a modified version of OG BabyAGI, called BabyBeeAGI (future modifications will follow the pattern "Baby<animal>AGI"). This version requires GPT-4, it's very slow, and often errors out.######
######IMPORTANT NOTE: I'm sharing this as a framework to build on top of (with lots of errors for improvement), to facilitate discussion around how to improve these. This is NOT for people who are looking for a complete solution that's ready to use. ######

import openai
import pinecone
import time
import requests
from bs4 import BeautifulSoup
from collections import deque
from typing import Dict, List
import re
import ast
import json
from serpapi import GoogleSearch

### SET THESE 4 VARIABLES ##############################

# Add your API keys here
OPENAI_API_KEY = ""
SERPAPI_API_KEY = "" #If you include SERPAPI KEY, this will enable web-search. If you don't, it will automatically remove web-search capability.

# Set variables
OBJECTIVE = "You are an AI. Make the world a better place."
YOUR_FIRST_TASK = "Develop a task list."

### UP TO HERE ##############################

# Configure OpenAI and SerpAPI client
openai.api_key = OPENAI_API_KEY
if SERPAPI_API_KEY:
  serpapi_client = GoogleSearch({"api_key": SERPAPI_API_KEY})
  websearch_var = "[web-search] "
else:
  websearch_var = ""

# Initialize task list
task_list = []

# Initialize session_summary
session_summary = ""

### Task list functions ##############################
def add_task(task: Dict):
    task_list.append(task)

def get_task_by_id(task_id: int):
    for task in task_list:
        if task["id"] == task_id:
            return task
    return None

def get_completed_tasks():
    return [task for task in task_list if task["status"] == "complete"]

### Tool functions ##############################
def text_completion_tool(prompt: str):
    response = openai.Completion.create(
        engine="text-davinci-003",
        prompt=prompt,
        temperature=0.5,
        max_tokens=1500,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )
    return response.choices[0].text.strip()

def web_search_tool(query: str):
    search_params = {
        "engine": "google",
        "q": query,
        "api_key": SERPAPI_API_KEY,
        "num":3
    }
    search_results = GoogleSearch(search_params)
    results = search_results.get_dict()

    return str(results["organic_results"])

def web_scrape_tool(url: str):
    response = requests.get(url)
    print(response)
    soup = BeautifulSoup(response.content, "html.parser")
    result = soup.get_text(strip=True)+"URLs: "
    for link in soup.findAll('a', attrs={'href': re.compile("^https://")}):
      result+= link.get('href')+", "
    return result

### Agent functions ##############################
def execute_task(task, task_list, OBJECTIVE):
    global task_id_counter
    # Check if dependent_task_id is complete
    if task["dependent_task_id"]:
        dependent_task = get_task_by_id(task["dependent_task_id"])
        if not dependent_task or dependent_task["status"] != "complete":
            return

    # Execute task
    
    print("\033[92m\033[1m"+"\n*****NEXT TASK*****\n"+"\033[0m\033[0m")
    print(str(task['id'])+": "+str(task['task'])+" ["+str(task['tool']+"]"))
    task_prompt = f"Complete your assigned task based on the objective: {OBJECTIVE}. Your task: {task['task']}"
    if task["dependent_task_id"]:
        dependent_task_result = dependent_task["result"]
        task_prompt += f"\nThe previous task ({dependent_task['id']}. {dependent_task['task']}) result: {dependent_task_result}"

    task_prompt += "\nResponse:"
    ##print("###task_prompt: "+task_prompt)
    if task["tool"] == "text-completion":
        result = text_completion_tool(task_prompt)
    elif task["tool"] == "web-search":
        result = web_search_tool(task_prompt)
    elif task["tool"] == "web-scrape":
        result = web_scrape_tool(str(task['task']))
    else:
        result = "Unknown tool"

    
    print("\033[93m\033[1m"+"\n*****TASK RESULT*****\n"+"\033[0m\033[0m")
    print_result = result[0:2000]
    if result != result[0:2000]:
      print(print_result+"...")
    else:
      print(result)
    # Update task status and result
    task["status"] = "complete"
    task["result"] = result
    task["result_summary"] = summarizer_agent(result)

    # Update session_summary
    session_summary = overview_agent(task["id"])

    # Increment task_id_counter
    task_id_counter += 1

    # Update task_manager_agent of tasks
    task_manager_agent(
        OBJECTIVE,
        result,
        task["task"],
        [t["task"] for t in task_list if t["status"] == "incomplete"],
        task["id"]  
    )


def task_manager_agent(objective: str, result: str, task_description: str, incomplete_tasks: List[str], current_task_id : int) -> List[Dict]:
    global task_list
    original_task_list = task_list.copy()
    minified_task_list = [{k: v for k, v in task.items() if k != "result"} for task in task_list]
    result = result[0:4000] #come up with better solution later.

    prompt = (
        f"You are a task management AI tasked with cleaning the formatting of and reprioritizing the following tasks: {minified_task_list}. "
        f"Consider the ultimate objective of your team: {OBJECTIVE}. "
        f"Do not remove any tasks. Return the result as a JSON-formatted list of dictionaries.\n"
        f"Create new tasks based on the result of last task if necessary for the objective. Limit tasks types to those that can be completed with the available tools listed below. Task description should be detailed."
        f"The maximum task list length is 7. Do not add an 8th task."
        f"The last completed task has the following result: {result}. "
        f"Current tool option is [text-completion] {websearch_var} and [web-scrape] only."# web-search is added automatically if SERPAPI exists
        f"For tasks using [web-scrape], provide only the URL to scrape as the task description. Do not provide placeholder URLs, but use ones provided by a search step or the initial objective."
        #f"If the objective is research related, use at least one [web-search] with the query as the task description, and after, add up to three URLs from the search result as a task with [web-scrape], then use [text-completion] to write a comprehensive summary of each site thas has been scraped.'"
        f"For tasks using [web-search], provide the search query, and only the search query to use (eg. not 'research waterproof shoes, but 'waterproof shoes')"
        f"dependent_task_id should always be null or a number."
        f"Do not reorder completed tasks. Only reorder and dedupe incomplete tasks.\n"
        f"Make sure all task IDs are in chronological order.\n"
        f"Do not provide example URLs for [web-scrape].\n"
        f"Do not include the result from the last task in the JSON, that will be added after..\n"
        f"The last step is always to provide a final summary report of all tasks.\n"
        f"An example of the desired output format is: "
        "[{\"id\": 1, \"task\": \"https://untapped.vc\", \"tool\": \"web-scrape\", \"dependent_task_id\": null, \"status\": \"incomplete\", \"result\": null, \"result_summary\": null}, {\"id\": 2, \"task\": \"Analyze the contents of...\", \"tool\": \"text-completion\", \"dependent_task_id\": 1, \"status\": \"incomplete\", \"result\": null, \"result_summary\": null}, {\"id\": 3, \"task\": \"Untapped Capital\", \"tool\": \"web-search\", \"dependent_task_id\": null, \"status\": \"incomplete\", \"result\": null, \"result_summary\": null}]."
    )
    print("\033[90m\033[3m" + "\nRunning task manager agent...\n" + "\033[0m")
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": "You are a task manager AI."
            },
            {
                "role": "user",
                "content": prompt
            }
        ],
        temperature=0.2,
        max_tokens=1500,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )

    # Extract the content of the assistant's response and parse it as JSON
    result = response["choices"][0]["message"]["content"]
    print("\033[90m\033[3m" + "\nDone!\n" + "\033[0m")
    try:
      task_list = json.loads(result)
    except Exception as error:
      print(error)
    # Add the 'result' field back in
    for updated_task, original_task in zip(task_list, original_task_list):
        if "result" in original_task:
            updated_task["result"] = original_task["result"]
    task_list[current_task_id]["result"]=result
    #print(task_list)
    return task_list



def summarizer_agent(text: str) -> str:
    text = text[0:4000]
    prompt = f"Please summarize the following text:\n{text}\nSummary:"
    response = openai.Completion.create(
        engine="text-davinci-003",
        prompt=prompt,
        temperature=0.5,
        max_tokens=100,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )
    return response.choices[0].text.strip()


def overview_agent(last_task_id: int) -> str:
    global session_summary

    completed_tasks = get_completed_tasks()
    completed_tasks_text = "\n".join(
        [f"{task['id']}. {task['task']} - {task['result_summary']}" for task in completed_tasks]
    )

    prompt = f"Here is the current session summary:\n{session_summary}\nThe last completed task is task {last_task_id}. Please update the session summary with the information of the last task:\n{completed_tasks_text}\nUpdated session summary, which should describe all tasks in chronological order:"
    response = openai.Completion.create(
        engine="text-davinci-003",
        prompt=prompt,
        temperature=0.5,
        max_tokens=200,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )
    session_summary = response.choices[0].text.strip()
    return session_summary


### Main Loop ##############################

# Add the first task
first_task = {
    "id": 1,
    "task": YOUR_FIRST_TASK,
    "tool": "text-completion",
    "dependent_task_id": None,
    "status": "incomplete",
    "result": "",
    "result_summary": ""
}
add_task(first_task)

task_id_counter = 0
#Print OBJECTIVE
print("\033[96m\033[1m"+"\n*****OBJECTIVE*****\n"+"\033[0m\033[0m")
print(OBJECTIVE)

# Continue the loop while there are incomplete tasks
while any(task["status"] == "incomplete" for task in task_list):

    # Filter out incomplete tasks
    incomplete_tasks = [task for task in task_list if task["status"] == "incomplete"]

    if incomplete_tasks:
        # Sort tasks by ID
        incomplete_tasks.sort(key=lambda x: x["id"])

        # Pull the first task
        task = incomplete_tasks[0]

        # Execute task & call task manager from function
        execute_task(task, task_list, OBJECTIVE)

        # Print task list and session summary
        print("\033[95m\033[1m" + "\n*****TASK LIST*****\n" + "\033[0m")
        for t in task_list:
            dependent_task = ""
            if t['dependent_task_id'] is not None:
                dependent_task = f"\033[31m<dependency: #{t['dependent_task_id']}>\033[0m"
            status_color = "\033[32m" if t['status'] == "complete" else "\033[31m"
            print(f"\033[1m{t['id']}\033[0m: {t['task']} {status_color}[{t['status']}]\033[0m \033[93m[{t['tool']}] {dependent_task}\033[0m")
        print("\033[93m\033[1m" + "\n*****SESSION SUMMARY*****\n" + "\033[0m\033[0m")
        print(session_summary)

    time.sleep(1)  # Sleep before checking the task list again

### Objective complete ##############################

# Print the full task list if there are no incomplete tasks
if all(task["status"] != "incomplete" for task in task_list):
    print("\033[92m\033[1m" + "\n*****ALL TASKS COMPLETED*****\n" + "\033[0m\033[0m")
    for task in task_list:
        print(f"ID: {task['id']}, Task: {task['task']}, Result: {task['result']}")



File: classic/babyagi.py

import openai
import pinecone
import time
from collections import deque
from typing import Dict, List

#Set API Keys
OPENAI_API_KEY = ""
PINECONE_API_KEY = ""
PINECONE_ENVIRONMENT = "us-east1-gcp" #Pinecone Environment (eg. "us-east1-gcp")

#Set Variables
YOUR_TABLE_NAME = "test-table"
OBJECTIVE = "Solve world hunger."
YOUR_FIRST_TASK = "Develop a task list."

#Print OBJECTIVE
print("\033[96m\033[1m"+"\n*****OBJECTIVE*****\n"+"\033[0m\033[0m")
print(OBJECTIVE)

# Configure OpenAI and Pinecone
openai.api_key = OPENAI_API_KEY
pinecone.init(api_key=PINECONE_API_KEY, environment=PINECONE_ENVIRONMENT)

# Create Pinecone index
table_name = YOUR_TABLE_NAME
dimension = 1536
metric = "cosine"
pod_type = "p1"
if table_name not in pinecone.list_indexes():
    pinecone.create_index(table_name, dimension=dimension, metric=metric, pod_type=pod_type)

# Connect to the index
index = pinecone.Index(table_name)

# Task list
task_list = deque([])

def add_task(task: Dict):
    task_list.append(task)

def get_ada_embedding(text):
    text = text.replace("\n", " ")
    return openai.Embedding.create(input=[text], model="text-embedding-ada-002")["data"][0]["embedding"]

def task_creation_agent(objective: str, result: Dict, task_description: str, task_list: List[str]):
    prompt = f"You are an task creation AI that uses the result of an execution agent to create new tasks with the following objective: {objective}, The last completed task has the result: {result}. This result was based on this task description: {task_description}. These are incomplete tasks: {', '.join(task_list)}. Based on the result, create new tasks to be completed by the AI system that do not overlap with incomplete tasks. Return the tasks as an array."
    response = openai.Completion.create(engine="text-davinci-003",prompt=prompt,temperature=0.5,max_tokens=100,top_p=1,frequency_penalty=0,presence_penalty=0)
    new_tasks = response.choices[0].text.strip().split('\n')
    return [{"task_name": task_name} for task_name in new_tasks]

def prioritization_agent(this_task_id:int):
    global task_list
    task_names = [t["task_name"] for t in task_list]
    next_task_id = int(this_task_id)+1
    prompt = f"""You are an task prioritization AI tasked with cleaning the formatting of and reprioritizing the following tasks: {task_names}. Consider the ultimate objective of your team:{OBJECTIVE}. Do not remove any tasks. Return the result as a numbered list, like:
    #. First task
    #. Second task
    Start the task list with number {next_task_id}."""
    response = openai.Completion.create(engine="text-davinci-003",prompt=prompt,temperature=0.5,max_tokens=1000,top_p=1,frequency_penalty=0,presence_penalty=0)
    new_tasks = response.choices[0].text.strip().split('\n')
    task_list = deque()
    for task_string in new_tasks:
        task_parts = task_string.strip().split(".", 1)
        if len(task_parts) == 2:
            task_id = task_parts[0].strip()
            task_name = task_parts[1].strip()
            task_list.append({"task_id": task_id, "task_name": task_name})

def execution_agent(objective:str,task: str) -> str:
    #context = context_agent(index="quickstart", query="my_search_query", n=5)
    context=context_agent(index=YOUR_TABLE_NAME, query=objective, n=5)
    #print("\n*******RELEVANT CONTEXT******\n")
    #print(context)
    response = openai.Completion.create(
        engine="text-davinci-003",
        prompt=f"You are an AI who performs one task based on the following objective: {objective}. Your task: {task}\nResponse:",
        temperature=0.7,
        max_tokens=2000,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )
    return response.choices[0].text.strip()

def context_agent(query: str, index: str, n: int):
    query_embedding = get_ada_embedding(query)
    index = pinecone.Index(index_name=index)
    results = index.query(query_embedding, top_k=n,
    include_metadata=True)
    #print("***** RESULTS *****")
    #print(results)
    sorted_results = sorted(results.matches, key=lambda x: x.score, reverse=True)    
    return [(str(item.metadata['task'])) for item in sorted_results]

# Add the first task
first_task = {
    "task_id": 1,
    "task_name": YOUR_FIRST_TASK
}

add_task(first_task)
# Main loop
task_id_counter = 1
while True:
    if task_list:
        # Print the task list
        print("\033[95m\033[1m"+"\n*****TASK LIST*****\n"+"\033[0m\033[0m")
        for t in task_list:
            print(str(t['task_id'])+": "+t['task_name'])

        # Step 1: Pull the first task
        task = task_list.popleft()
        print("\033[92m\033[1m"+"\n*****NEXT TASK*****\n"+"\033[0m\033[0m")
        print(str(task['task_id'])+": "+task['task_name'])

        # Send to execution function to complete the task based on the context
        result = execution_agent(OBJECTIVE,task["task_name"])
        this_task_id = int(task["task_id"])
        print("\033[93m\033[1m"+"\n*****TASK RESULT*****\n"+"\033[0m\033[0m")
        print(result)

        # Step 2: Enrich result and store in Pinecone
        enriched_result = {'data': result}  # This is where you should enrich the result if needed
        result_id = f"result_{task['task_id']}"
        vector = enriched_result['data']  # extract the actual result from the dictionary
        index.upsert([(result_id, get_ada_embedding(vector),{"task":task['task_name'],"result":result})])

    # Step 3: Create new tasks and reprioritize task list
    new_tasks = task_creation_agent(OBJECTIVE,enriched_result, task["task_name"], [t["task_name"] for t in task_list])

    for new_task in new_tasks:
        task_id_counter += 1
        new_task.update({"task_id": task_id_counter})
        add_task(new_task)
    prioritization_agent(this_task_id)

time.sleep(1)  # Sleep before checking the task list again



File: classic/BabyElfAGI/main.py

import os
from dotenv import load_dotenv
import importlib.util
import json
import openai
import concurrent.futures
import time
from datetime import datetime
from skills.skill import Skill
from skills.skill_registry import SkillRegistry
from tasks.task_registry import TaskRegistry


load_dotenv()  # Load environment variables from .env file

# Retrieve all API keys
api_keys = {
    'openai': os.environ['OPENAI_API_KEY'],
    'serpapi': os.environ['SERPAPI_API_KEY']
    # Add more keys here as needed
}

# Set OBJECTIVE
OBJECTIVE = "Create an example objective and tasklist for 'write a poem', which only uses text_completion in the tasks. Do this by usign code_reader to read example1.json, then writing the JSON objective tasklist pair using text_completion, and saving it using objective_saver."
LOAD_SKILLS = ['text_completion','code_reader','objective_saver']
REFLECTION = False

##### START MAIN LOOP########

# Print OBJECTIVE
print("\033[96m\033[1m"+"\n*****OBJECTIVE*****\n"+"\033[0m\033[0m")
print(OBJECTIVE)

if __name__ == "__main__":
    session_summary = ""
  
    # Initialize the SkillRegistry and TaskRegistry
    skill_registry = SkillRegistry(api_keys=api_keys, skill_names=LOAD_SKILLS)
    skill_descriptions = ",".join(f"[{skill.name}: {skill.description}]" for skill in skill_registry.skills.values())
    task_registry = TaskRegistry()

    # Create the initial task list based on an objective
    task_registry.create_tasklist(OBJECTIVE, skill_descriptions)
  
    # Initialize task outputs
    task_outputs = {i: {"completed": False, "output": None} for i, _ in enumerate(task_registry.get_tasks())}

    # Create a thread pool for parallel execution
    with concurrent.futures.ThreadPoolExecutor() as executor:
        # Loop until all tasks are completed
        while not all(task["completed"] for task in task_outputs.values()):
        
            # Get the tasks that are ready to be executed (i.e., all their dependencies have been completed)
            tasks = task_registry.get_tasks()
            # Print the updated task list
            task_registry.print_tasklist(tasks) 
            
            # Update task_outputs to include new tasks
            for task in tasks:
                if task["id"] not in task_outputs:
                    task_outputs[task["id"]] = {"completed": False, "output": None}

          
            ready_tasks = [(task["id"], task) for task in tasks
               if all((dep in task_outputs and task_outputs[dep]["completed"]) 
               for dep in task.get('dependent_task_ids', [])) 
               and not task_outputs[task["id"]]["completed"]]

            session_summary += str(task)+"\n"
            futures = [executor.submit(task_registry.execute_task, task_id, task, skill_registry, task_outputs, OBJECTIVE) 
                       for task_id, task in ready_tasks if not task_outputs[task_id]["completed"]]
            
            # Wait for the tasks to complete
            for future in futures:
                i, output = future.result()
                task_outputs[i]["output"] = output
                task_outputs[i]["completed"] = True
                
                # Update the task in the TaskRegistry
                task_registry.update_tasks({"id": i, "status": "completed", "result": output})
                
                completed_task = task_registry.get_task(i)
                print(f"\033[92mTask #{i}: {completed_task.get('task')} \033[0m\033[92m[COMPLETED]\033[0m\033[92m[{completed_task.get('skill')}]\033[0m")

                # Reflect on the output
                if output:
                    session_summary += str(output)+"\n"

                  
                    if REFLECTION == True:
                      new_tasks, insert_after_ids, tasks_to_update = task_registry.reflect_on_output(output, skill_descriptions)
                      # Insert new tasks
                      for new_task, after_id in zip(new_tasks, insert_after_ids):
                          task_registry.add_task(new_task, after_id)
      
                      # Update existing tasks
                      for task_to_update in tasks_to_update:
                        task_registry.update_tasks(task_to_update)
                    


            #print(task_outputs.values())
            if all(task["status"] == "completed" for task in task_registry.tasks):
              print("All tasks completed!")
              break
                  
            # Short delay to prevent busy looping
            time.sleep(0.1)


        # Print session summary
        print("\033[96m\033[1m"+"\n*****SAVING FILE...*****\n"+"\033[0m\033[0m")
        file = open(f'output/output_{datetime.now().strftime("%d_%m_%Y_%H_%M_%S")}.txt', 'w')
        file.write(session_summary)
        file.close()
        print("...file saved.")
        print("END")
        executor.shutdown()


File: classic/BabyElfAGI/tasks/task_registry.py

import openai
import json
import threading
import os
import numpy as np

class TaskRegistry:
    def __init__(self):
        self.tasks = []
        # Initialize the lock
        self.lock = threading.Lock()
        objectives_file_path = "tasks/example_objectives"
        self.example_loader = ExampleObjectivesLoader(objectives_file_path)

    def load_example_objectives(self, user_objective):
        return self.example_loader.load_example_objectives(user_objective)

      
    def create_tasklist(self, objective, skill_descriptions):
        #load most relevant object and tasklist from objectives_examples.json
        example_objective, example_tasklist = self.load_example_objectives(objective)

        prompt = (
            f"You are an expert task list creation AI tasked with creating a  list of tasks as a JSON array, considering the ultimate objective of your team: {objective}. "
            f"Create a very short task list based on the objective, the final output of the last task will be provided back to the user. Limit tasks types to those that can be completed with the available skills listed below. Task description should be detailed.###"
            f"AVAILABLE SKILLS: {skill_descriptions}.###"
            f"RULES:"
            f"Do not use skills that are not listed."
            f"Always include one skill."
            f"dependent_task_ids should always be an empty array, or an array of numbers representing the task ID it should pull results from."
            f"Make sure all task IDs are in chronological order.###\n"
            f"EXAMPLE OBJECTIVE={json.dumps(example_objective)}"
            f"TASK LIST={json.dumps(example_tasklist)}"
            f"OBJECTIVE={objective}"
            f"TASK LIST="
        )

        print("\033[90m\033[3m" + "\nInitializing...\n" + "\033[0m")
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-0613",
            messages=[
                {
                    "role": "system",
                    "content": "You are a task creation AI."
                },
                {
                    "role": "user",
                    "content": prompt
                }
            ],
            temperature=0,
            max_tokens=1500,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )

        # Extract the content of the assistant's response and parse it as JSON
        result = response["choices"][0]["message"]["content"]
        try:
            task_list = json.loads(result)
            self.tasks = task_list
        except Exception as error:
            print(error)


    def execute_task(self, i, task, skill_registry, task_outputs, objective):
        p_nexttask="\033[92m\033[1m"+"\n*****NEXT TASK ID:"+str(task['id'])+"*****\n"+"\033[0m\033[0m"
        p_nexttask += f"\033[ EExecuting task {task.get('id')}: {task.get('task')}) [{task.get('skill')}]\033[)"
        print(p_nexttask)
        # Retrieve the skill from the registry
        skill = skill_registry.get_skill(task['skill'])
        # Get the outputs of the dependent tasks
        dependent_task_outputs = {dep: task_outputs[dep]["output"] for dep in task['dependent_task_ids']} if 'dependent_task_ids' in task else {}
        # Execute the skill
        # print("execute:"+str([task['task'], dependent_task_outputs, objective]))
        task_output = skill.execute(task['task'], dependent_task_outputs, objective)
        print("\033[93m\033[1m"+"\nTask Output (ID:"+str(task['id'])+"):"+"\033[0m\033[0m")
        print("TASK: "+str(task["task"]))
        print("OUTPUT: "+str(task_output))
        return i, task_output

  
    def reorder_tasks(self):
        self.tasks = sorted(self.tasks, key=lambda task: task['id'])

  
    def add_task(self, task, after_task_id):
        # Get the task ids
        task_ids = [t["id"] for t in self.tasks]

        # Get the index of the task id to add the new task after
        insert_index = task_ids.index(after_task_id) + 1 if after_task_id in task_ids else len(task_ids)

        # Insert the new task
        self.tasks.insert(insert_index, task)
        self.reorder_tasks()


    def update_tasks(self, task_update):
        for task in self.tasks:
            if task['id'] == task_update['id']:
                # This merges the original task dictionary with the update, overwriting only the fields present in the update.
                task.update(task_update)
                self.reorder_tasks()

    def reflect_on_output(self, task_output, skill_descriptions):
        with self.lock:
            example = [
                [
                    {"id": 3, "task": "New task 1 description", "skill": "text_completion_skill",
                     "dependent_task_ids": [], "status": "complete"},
                    {"id": 4, "task": "New task 2 description", "skill": "text_completion_skill",
                     "dependent_task_ids": [], "status": "incomplete"}
                ],
                [2, 3],
                {"id": 5, "task": "Complete the objective and provide a final report",
                 "skill": "text_completion_skill", "dependent_task_ids": [1, 2, 3, 4], "status": "incomplete"}
            ]

            prompt = (
                f"You are an expert task manager, review the task output to decide at least one new task to add."
                f"As you add a new task, see if there are any tasks that need to be updated (such as updating dependencies)."
                f"Use the current task list as reference."
                f"Do not add duplicate tasks to those in the current task list."
                f"Only provide JSON as your response without further comments."
                f"Every new and updated task must include all variables, even they are empty array."
                f"Dependent IDs must be smaller than the ID of the task."
                f"New tasks IDs should be no larger than the last task ID."
                f"Always select at least one skill."
                f"Task IDs should be unique and in chronological order."                f"Do not change the status of complete tasks."
                f"Only add skills from the AVAILABLE SKILLS, using the exact same spelling."
                f"Provide your array as a JSON array with double quotes. The first object is new tasks to add as a JSON array, the second array lists the ID numbers where the new tasks should be added after (number of ID numbers matches array), and the third object provides the tasks that need to be updated."
                f"Make sure to keep dependent_task_ids key, even if an empty array."
                f"AVAILABLE SKILLS: {skill_descriptions}.###"
                f"\n###Here is the last task output: {task_output}"
                f"\n###Here is the current task list: {self.tasks}"
                f"\n###EXAMPLE OUTPUT FORMAT = {json.dumps(example)}"
                f"\n###OUTPUT = "
            )
            print("\033[90m\033[3m" + "\nReflecting on task output to generate new tasks if necessary...\n" + "\033[0m")
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo-16k-0613",
                messages=[
                    {
                        "role": "system",
                        "content": "You are a task creation AI."
                    },
                    {
                        "role": "user",
                        "content": prompt
                    }
                ],
                temperature=0.7,
                max_tokens=1500,
                top_p=1,
                frequency_penalty=0,
                presence_penalty=0
            )

            # Extract the content of the assistant's response and parse it as JSON
            result = response["choices"][0]["message"]["content"]
            print("\n#" + str(result))

            # Check if the returned result has the expected structure
            if isinstance(result, str):
                try:
                    task_list = json.loads(result)
                    # print("RESULT:")

                    print(task_list)
                    # return [],[],[]
                    return task_list[0], task_list[1], task_list[2]
                except Exception as error:
                    print(error)

            else:
                raise ValueError("Invalid task list structure in the output")

    def get_tasks(self):
        """
        Returns the current list of tasks.

        Returns:
        list: the list of tasks.
        """
        return self.tasks

    def get_task(self, task_id):
        """
        Returns a task given its task_id.

        Parameters:
        task_id : int
            The unique ID of the task.

        Returns:
        dict
            The task that matches the task_id.
        """
        matching_tasks = [task for task in self.tasks if task["id"] == task_id]

        if matching_tasks:
            return matching_tasks[0]
        else:
            print(f"No task found with id {task_id}")
            return None

    def print_tasklist(self, task_list):
        p_tasklist="\033[95m\033[1m" + "\n*****TASK LIST*****\n" + "\033[0m"
        for t in task_list:
            dependent_task_ids = t.get('dependent_task_ids', [])
            dependent_task = ""
            if dependent_task_ids:
                dependent_task = f"\033[31m<dependencies: {', '.join([f'#{dep_id}' for dep_id in dependent_task_ids])}>\033[0m"
            status_color = "\033[32m" if t.get('status') == "completed" else "\033[31m"
            p_tasklist+= f"\033[1m{t.get('id')}\033[0m: {t.get('task')} {status_color}[{t.get('status')}]\033[0m \033[93m[{t.get('skill')}] {dependent_task}\033[0m\n"
        print(p_tasklist)



class ExampleObjectivesLoader:
    def __init__(self, objectives_folder_path):
        self.objectives_folder_path = objectives_folder_path
        self.objectives_examples = []  # Initialize as an empty list

    def load_objectives_examples(self):
        self.objectives_examples = []
        for filename in os.listdir(self.objectives_folder_path):
            file_path = os.path.join(self.objectives_folder_path, filename)
            with open(file_path, 'r') as file:
                objectives = json.load(file)
                self.objectives_examples.extend(objectives)
    

    def find_most_relevant_objective(self, user_input):
        user_input_embedding = self.get_embedding(user_input, model='text-embedding-ada-002')
        most_relevant_objective = max(
            self.objectives_examples,
            key=lambda pair: self.cosine_similarity(pair['objective'], user_input_embedding)
        )
        return most_relevant_objective['objective'], most_relevant_objective['examples']


    def get_embedding(self, text, model='text-embedding-ada-002'):
        response = openai.Embedding.create(input=[text], model=model)
        embedding = response['data'][0]['embedding']
        return embedding

    def cosine_similarity(self, objective, embedding):
        max_similarity = float('-inf')
        objective_embedding = self.get_embedding(objective, model='text-embedding-ada-002')
        similarity = self.calculate_similarity(objective_embedding, embedding)
        max_similarity = max(max_similarity, similarity)
        return max_similarity

    def calculate_similarity(self, embedding1, embedding2):
        embedding1 = np.array(embedding1, dtype=np.float32)
        embedding2 = np.array(embedding2, dtype=np.float32)
        similarity = np.dot(embedding1, embedding2) / (np.linalg.norm(embedding1) * np.linalg.norm(embedding2))
        return similarity

    def load_example_objectives(self, user_objective):
        self.load_objectives_examples()
        most_relevant_objective, most_relevant_tasklist = self.find_most_relevant_objective(user_objective)
        example_objective = most_relevant_objective
        example_tasklist = most_relevant_tasklist
        return example_objective, example_tasklist
    


File: classic/BabyElfAGI/skills/skill.py

class Skill:
    name = 'base skill'
    description = 'This is the base skill.'
    api_keys_required = []

    def __init__(self, api_keys):
        self.api_keys = api_keys
        missing_keys = self.check_required_keys(api_keys)
        if missing_keys:
            print(f"Missing API keys for {self.name}: {missing_keys}")
            self.valid = False
        else:
            self.valid = True
        for key in self.api_keys_required:
            if isinstance(key, list):
                for subkey in key:
                    if subkey in api_keys:
                        setattr(self, f"{subkey}_api_key", api_keys.get(subkey))
            elif key in api_keys:
                setattr(self, f"{key}_api_key", api_keys.get(key))

    def check_required_keys(self, api_keys):
        missing_keys = []
        for key in self.api_keys_required:
            if isinstance(key, list):  # If the key is actually a list of alternatives
                if not any(k in api_keys for k in key):  # If none of the alternatives are present
                    missing_keys.append(key)  # Add the list of alternatives to the missing keys
            elif key not in api_keys:  # If the key is a single key and it's not present
                missing_keys.append(key)  # Add the key to the missing keys
        return missing_keys

    def execute(self, params, dependent_task_outputs, objective):
        raise NotImplementedError('Execute method must be implemented in subclass.')


File: classic/BabyElfAGI/skills/skill_saver.py

from skills.skill import Skill
import os
import openai

class SkillSaver(Skill):
    name = 'skill_saver'
    description = "A skill that saves code written in a previous step into a file within the skills folder. Not for writing code."
    api_keys_required = []

    def __init__(self, api_keys):
        super().__init__(api_keys)

    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return

        task_prompt = f"Extract the code and only the code from the dependent task output here: {dependent_task_outputs}  \n###\nCODE:"
      
        messages = [
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.4,
            max_tokens=3000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        ) 
    
        code =  response.choices[0].message['content'].strip()
        task_prompt = f"Come up with a file name (eg. 'get_weather.py') for the following skill:{code}\n###\nFILE_NAME:"
      
        messages = [
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.4,
            max_tokens=3000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        ) 
    
        file_name =  response.choices[0].message['content'].strip()
        file_path = os.path.join('skills',file_name)

        try:
            with open(file_path, 'w') as file:
                file.write(code)
                print(f"Code saved successfully: {file_name}")
        except:
            print("Error saving code.")

        return None


File: classic/BabyElfAGI/skills/directory_structure.py

from skills.skill import Skill
import os

class DirectoryStructure(Skill):
    name = 'directory_structure'
    description = "A tool that outputs the file and folder structure of its top parent folder."

    def __init__(self, api_keys=None):
        super().__init__(api_keys)

    def execute(self, params, dependent_task_outputs, objective):
        # Get the current script path
        current_script_path = os.path.realpath(__file__)

        # Get the top parent directory of current script
        top_parent_path = self.get_top_parent_path(current_script_path)
        # Get the directory structure from the top parent directory
        dir_structure = self.get_directory_structure(top_parent_path)

        return dir_structure

    def get_directory_structure(self, start_path):
        dir_structure = {}
        ignore_dirs = ['.','__init__.py', '__pycache__', 'pydevd', 'poetry','venv']  # add any other directories to ignore here
    
        for root, dirs, files in os.walk(start_path):
            dirs[:] = [d for d in dirs if not any(d.startswith(i) for i in ignore_dirs)]  # exclude specified directories
            files = [f for f in files if not f[0] == '.' and f.endswith('.py')]  # exclude hidden files and non-Python files

            current_dict = dir_structure
            path_parts = os.path.relpath(root, start_path).split(os.sep)
            for part in path_parts:
                if part:  # skip empty parts
                    if part not in current_dict:
                        current_dict[part] = {}
                    current_dict = current_dict[part]
            for f in files:
                current_dict[f] = None
        #print("#############################")
        #print(str(current_dict)[0:100])
        return dir_structure
    
    

    def get_top_parent_path(self, current_path):
        relative_path = ""
        while True:
            new_path = os.path.dirname(current_path)
            print(new_path)
            if new_path == '/home/runner/BabyElfAGI/skills':  # reached the top
            #if new_path == current_path:  # reached the top
                #return relative_path
                return '/home/runner/BabyElfAGI'
            current_path = new_path
            relative_path = os.path.join("..", relative_path)
            print(relative_path)


File: classic/BabyElfAGI/skills/objective_saver.py

from skills.skill import Skill
import os
import openai

class ObjectiveSaver(Skill):
    name = 'objective_saver'
    description = "A skill that saves a new example_objective based on the concepts from skill_saver.py"
    api_keys_required = []

    def __init__(self, api_keys):
        super().__init__(api_keys)

    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return
        #print(dependent_task_outputs[2])
        code =  dependent_task_outputs[2]
        task_prompt = f"Come up with a file name (eg. 'research_shoes.json') for the following objective:{code}\n###\nFILE_NAME:"
      
        messages = [
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.4,
            max_tokens=3000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        ) 
    
        file_name =  response.choices[0].message['content'].strip()
        file_path = os.path.join('tasks/example_objectives',file_name)

        try:
            with open(file_path, 'w') as file:
                file.write("["+code+"]")
                print(f"Code saved successfully: {file_name}")
        except:
            print("Error saving code.")

        return None


File: classic/BabyElfAGI/skills/text_completion.py

from skills.skill import Skill
import openai

class TextCompletion(Skill):
    name = 'text_completion'
    description = "A tool that uses OpenAI's text completion API to generate, summarize, and/or analyze text and code."
    api_keys_required = ['openai']

    def __init__(self, api_keys):
        super().__init__(api_keys)

    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return
        
        task_prompt = f"Complete your assigned task based on the objective and only based on information provided in the dependent task output, if provided. \n###\nYour objective: {objective}. \n###\nYour task: {params} \n###\nDependent tasks output: {dependent_task_outputs}  \n###\nYour task: {params}\n###\nRESPONSE:"
      
        messages = [
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.4,
            max_tokens=2000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        return "\n\n"+response.choices[0].message['content'].strip()



File: classic/BabyElfAGI/skills/code_reader.py

from skills.skill import Skill
import openai
import os

class CodeReader(Skill):
    name = 'code_reader'
    description = "A skill that finds a file's location in it's own program's directory and returns its contents."
    api_keys_required = ['openai']

    def __init__(self, api_keys):
        super().__init__(api_keys)

    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return

        dir_structure = self.get_directory_structure(self.get_top_parent_path(os.path.realpath(__file__)))
        print(f"Directory structure: {dir_structure}")
        example_dir_structure = {'.': {'main.py': None}, 'skills': {'__init__.py': None, 'web_scrape.py': None, 'skill.py': None, 'test_skill.py': None, 'text_completion.py': None, 'web_search.py': None, 'skill_registry.py': None, 'directory_structure.py': None, 'code_reader.py': None}, 'tasks': {'task_registry.py': None}, 'output': {}}
        example_params = "Analyze main.py"
        example_response = "main.py"
      
        task_prompt = f"Find a specific file in a directory and return only the file path, based on the task description below. Always return a directory.###The directory structure is as follows: \n{example_dir_structure}\nYour task: {example_params}\n###\nRESPONSE:{example_response} ###The directory structure is as follows: \n{dir_structure}\nYour task: {params}\n###\nRESPONSE:"
      
        messages = [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.2,
            max_tokens=1500,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
        file_path = response.choices[0].message['content'].strip()
        print(f"AI suggested file path: {file_path}")

        try:
            with open(file_path, 'r') as file:
                file_content = file.read()
                print(f"File content:\n{file_content}")
                return file_content
        except FileNotFoundError:
            print("File not found. Please check the AI's suggested file path.")
            return None

    def get_directory_structure(self, start_path):
        dir_structure = {}
        ignore_dirs = ['.','__init__.py', '__pycache__', 'pydevd', 'poetry','venv']  # add any other directories to ignore here
    
        for root, dirs, files in os.walk(start_path):
            dirs[:] = [d for d in dirs if not any(d.startswith(i) for i in ignore_dirs)]  # exclude specified directories
            files = [f for f in files if not f[0] == '.' and f.endswith('.py')]  # exclude hidden files and non-Python files

            current_dict = dir_structure
            path_parts = os.path.relpath(root, start_path).split(os.sep)
            for part in path_parts:
                if part:  # skip empty parts
                    if part not in current_dict:
                        current_dict[part] = {}
                    current_dict = current_dict[part]
            for f in files:
                current_dict[f] = None

        return dir_structure
    
    def get_top_parent_path(self, current_path):
        relative_path = ""
        while True:
            new_path = os.path.dirname(current_path)
            if new_path == '/home/runner/BabyElfAGI/skills':  # reached the top
                return '/home/runner/BabyElfAGI'
            current_path = new_path
            relative_path = os.path.join("..", relative_path)

        return relative_path



File: classic/BabyElfAGI/skills/skill_registry.py

import os
import importlib.util
import inspect
from .skill import Skill

class SkillRegistry:
    def __init__(self, api_keys, skill_names=None):
        self.skills = {}
        skill_files = [f for f in os.listdir('skills') if f.endswith('.py') and f != 'skill.py']
        for skill_file in skill_files:
            module_name = skill_file[:-3]
            if skill_names and module_name not in skill_names:
                continue
            module = importlib.import_module(f'skills.{module_name}')
            for attr_name in dir(module):
                attr_value = getattr(module, attr_name)
                if inspect.isclass(attr_value) and issubclass(attr_value, Skill) and attr_value is not Skill:
                    skill = attr_value(api_keys)
                    if skill.valid:
                        self.skills[skill.name] = skill
        # Print the names and descriptions of all loaded skills
        skill_info = "\n".join([f"{skill_name}: {skill.description}" for skill_name, skill in self.skills.items()])
        # print(skill_info)

    def load_all_skills(self):
        skills_dir = os.path.dirname(__file__)
        for filename in os.listdir(skills_dir):
            if filename.endswith(".py") and filename not in ["__init__.py", "skill.py", "skill_registry.py"]:
                skill_name = filename[:-3]  # Remove .py extension
                self.load_skill(skill_name)

    def load_specific_skills(self, skill_names):
        for skill_name in skill_names:
            self.load_skill(skill_name)

    def load_skill(self, skill_name):
        skills_dir = os.path.dirname(__file__)
        filename = f"{skill_name}.py"
        if os.path.isfile(os.path.join(skills_dir, filename)):
            spec = importlib.util.spec_from_file_location(skill_name, os.path.join(skills_dir, filename))
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
            for item_name in dir(module):
                item = getattr(module, item_name)
                if isinstance(item, type) and issubclass(item, Skill) and item is not Skill:
                    skill_instance = item(self.api_keys)
                    self.skills[skill_instance.name] = skill_instance

    def get_skill(self, skill_name):
        skill = self.skills.get(skill_name)
        if skill is None:
            raise Exception(
                f"Skill '{skill_name}' not found. Please make sure the skill is loaded and all required API keys are set.")
        return skill

    def get_all_skills(self):
        return self.skills


File: classic/BabyElfAGI/skills/web_search.py


from skills.skill import Skill
from serpapi import GoogleSearch
import openai
from bs4 import BeautifulSoup
import requests
import re

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36"
}

class WebSearch(Skill):
    name = 'web_search'
    description = 'A tool that performs web searches.'
    api_keys_required = [['openai'],['serpapi']]

    def __init__(self, api_keys):
        super().__init__(api_keys)

    def execute(self, params, dependent_task_outputs, objective):
        # Your function goes here

        
        # Modify the query based on the dependent task output
        if dependent_task_outputs != "":
            dependent_task = f"Use the dependent task output below as reference to help craft the correct search query for the provided task above. Dependent task output:{dependent_task_outputs}."
        else:
            dependent_task = "."
        query = self.text_completion_tool("You are an AI assistant tasked with generating a Google search query based on the following task: "+params+". If the task looks like a search query, return the identical search query as your response. " + dependent_task + "\nSearch Query:")
        print("\033[90m\033[3m"+"Search query: " +str(query)+"\033[0m")
        # Set the search parameters
        search_params = {
            "engine": "google",
            "q": query,
            "api_key": self.serpapi_api_key,
            "num": 3
        }
        # Perform the web search
        search_results = GoogleSearch(search_params).get_dict()
        
        # Simplify the search results
        search_results = self.simplify_search_results(search_results.get('organic_results', []))
        print("\033[90m\033[3mCompleted search. Now scraping results.\n\033[0m")

        # Store the results from web scraping
        results = ""
        for result in search_results:
            url = result.get('link')
            print("\033[90m\033[3m" + "Scraping: "+url+"" + "...\033[0m")
            content = self.web_scrape_tool({"url": url, "task": params,"objective":objective})
            results += str(content) + ". "
        print("\033[90m\033[3m"+str(results[0:100])[0:100]+"...\033[0m")
        # Process the results and generate a report
        results = self.text_completion_tool(f"You are an expert analyst combining the results of multiple web scrapes. Rewrite the following information as one cohesive report without removing any facts. Ignore any reports of not having info, unless all reports say so - in which case explain that the search did not work and suggest other web search queries to try. \n###INFORMATION:{results}.\n###REPORT:")
        return results

    def simplify_search_results(self, search_results):
        simplified_results = []
        for result in search_results:
            simplified_result = {
                "position": result.get("position"),
                "title": result.get("title"),
                "link": result.get("link"),
                "snippet": result.get("snippet")
            }
            simplified_results.append(simplified_result)
        return simplified_results

  
    def text_completion_tool(self, prompt: str):
        messages = [
            {"role": "user", "content": prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k-0613",
            messages=messages,
            temperature=0.2,
            max_tokens=1500,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        return response.choices[0].message['content'].strip()


    def web_scrape_tool(self, params):
        content = self.fetch_url_content(params['url'])
        if content is None:
            return None
    
        text = self.extract_text(content)
        print("\033[90m\033[3m"+"Scrape completed. Length:" +str(len(text))+".Now extracting relevant info..."+"...\033[0m")
        info = self.extract_relevant_info(params['objective'], text[0:11000], params['task'])
        links = self.extract_links(content)
        #result = f"{info} URLs: {', '.join(links)}"
        result = info
        
        return result
    
    def fetch_url_content(self,url: str):
        try:
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            return response.content
        except requests.exceptions.RequestException as e:
            print(f"Error while fetching the URL: {e}")
            return ""
    
    def extract_links(self,content: str):
        soup = BeautifulSoup(content, "html.parser")
        links = [link.get('href') for link in soup.findAll('a', attrs={'href': re.compile("^https?://")})]
        return links
    
    def extract_text(self,content: str):
        soup = BeautifulSoup(content, "html.parser")
        text = soup.get_text(strip=True)
        return text
    
    def extract_relevant_info(self, objective, large_string, task):
        chunk_size = 12000
        overlap = 500
        notes = ""
      
        if len(large_string) == 0:
          print("error scraping")
          return "Error scraping."
        
        for i in range(0, len(large_string), chunk_size - overlap):
            
            print("\033[90m\033[3m"+"Reading chunk..."+"\033[0m")  
            chunk = large_string[i:i + chunk_size]
            
            messages = [
                {"role": "system", "content": f"You are an AI assistant."},
                {"role": "user", "content": f"You are an expert AI research assistant tasked with creating or updating the current notes. If the current note is empty, start a current-notes section by exracting relevant data to the task and objective from the chunk of text to analyze. If there is a current note, add new relevant info frol the chunk of text to analyze. Make sure the new or combined notes is comprehensive and well written. Here's the current chunk of text to analyze: {chunk}. ### Here is the current task: {task}.### For context, here is the objective: {objective}.### Here is the data we've extraced so far that you need to update: {notes}.### new-or-updated-note:"}
            ]
    
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo-16k-0613",
                messages=messages,
                max_tokens=800,
                n=1,
                stop="###",
                temperature=0.7,
            )
    
            notes += response.choices[0].message['content'].strip()+". ";
        
        return notes


File: classic/babyfoxagi/ongoing_tasks.py

ongoing_tasks = {}



File: classic/babyfoxagi/babyagi.py

import os
from dotenv import load_dotenv
import time
from datetime import datetime
from skills.skill_registry import SkillRegistry
from tasks.task_registry import TaskRegistry
from ongoing_tasks import ongoing_tasks

load_dotenv()  # Load environment variables from .env file

api_keys = {
    'openai': os.getenv('OPENAI_API_KEY'),
    'serpapi': os.getenv('SERPAPI_API_KEY')
    #'airtable': os.getenv('AIRTABLE_API_KEY')
}

OBJECTIVE = "Research Yohei Nakajima and write a poem about him."
LOAD_SKILLS = ['web_search', 'text_completion', 'code_reader','google_jobs_api_search','image_generation','startup_analysis','play_music','game_generation']
#add web_search and documentation_search after you add SERPAPI_API_KEY in your secrets. airtable_search once you've added your AIRTABLE_API_KEY, and add base/table/column data to airtable_search.py, etc...
REFLECTION = False #Experimental reflection step between each task run (when running tasklist)

def run_single_task(task_id, task, skill_registry, task_outputs, OBJECTIVE, task_registry):
    """Execute a single task and update its status"""
    task_output = task_registry.execute_task(task_id, task, skill_registry, task_outputs, OBJECTIVE)
    
    task_outputs[task_id]["output"] = task_output
    task_outputs[task_id]["completed"] = True
    task_outputs[task_id]["description"] = task.get('description', 'No description available')
    task_outputs[task_id]["skill"] = task.get('skill', 'No skill information available')
    
    if task_output:
        task_registry.update_tasks({"id": task_id, "status": "completed", "result": task_output})

        completed_task = task_registry.get_task(task_id)
        print(f"Task #{task_id}: {completed_task.get('task')} [COMPLETED][{completed_task.get('skill')}]")
        
        if REFLECTION:
            new_tasks, insert_after_ids, tasks_to_update = task_registry.reflect_on_output(task_output, skill_descriptions)
            for new_task, after_id in zip(new_tasks, insert_after_ids):
                task_registry.add_task(new_task, after_id)

            if isinstance(tasks_to_update, dict) and tasks_to_update:
                tasks_to_update = [tasks_to_update]
            
            for task_to_update in tasks_to_update:
                task_registry.update_tasks(task_to_update)




def run_main_loop(OBJECTIVE, LOAD_SKILLS, api_keys, REFLECTION=False):
    """Main execution loop"""
    try:
        skill_descriptions = ",".join(f"[{skill.name}: {skill.description}]" for skill in global_skill_registry.skills.values())
        task_registry = TaskRegistry()
        task_registry.create_tasklist(OBJECTIVE, skill_descriptions)

        skill_names = [skill.name for skill in global_skill_registry.skills.values()]
        session_summary = f"OBJECTIVE:{OBJECTIVE}.#SKILLS:{','.join(skill_names)}.#"

        task_outputs = {task["id"]: {"completed": False, "output": None} for task in task_registry.get_tasks()}

        task_output = None  # Initialize task_output to None

        while not all(task["completed"] for task in task_outputs.values()):
            tasks = task_registry.get_tasks()
            task_registry.print_tasklist(tasks)

            for task in tasks:
                if task["id"] not in task_outputs:
                    task_outputs[task["id"]] = {"completed": False, "output": None}

                ready_tasks = [(task["id"], task) for task in tasks if all((dep in task_outputs and task_outputs[dep]["completed"]) for dep in task.get('dependent_task_ids', [])) and not task_outputs[task["id"]]["completed"]]

                for task_id, task in ready_tasks:
                    run_single_task(task_id, task, global_skill_registry, task_outputs, OBJECTIVE, task_registry)

                time.sleep(0.1)

            # Assuming the last task in tasks has the latest output. Adjust if your use case is different.
            last_task_id = tasks[-1]["id"] if tasks else None
            task_output = task_outputs[last_task_id]["output"] if last_task_id else None

            task_registry.reflect_on_final(OBJECTIVE, task_registry.get_tasks(), task_output, skill_descriptions)
            global_skill_registry.reflect_skills(OBJECTIVE, task_registry.get_tasks(), task_output, skill_descriptions)

            with open(f'output/output_{datetime.now().strftime("%d_%m_%Y_%H_%M_%S")}.txt', 'w') as file:
                file.write(session_summary)
            print("...file saved.")
            print("END")

        return task_output  # Return the last task output

    except Exception as e:
        return f"An error occurred: {e}"



# Removed repeated logic for initiating skill registry
global_skill_registry = SkillRegistry(api_keys=api_keys, main_loop_function=run_main_loop, skill_names=LOAD_SKILLS)


def execute_skill(skill_name, objective, task_id):
    """Execute a single skill"""
    skill = global_skill_registry.get_skill(skill_name)
    if skill:
        try:
            result = skill.execute(objective, "", objective)
            ongoing_tasks[task_id].update({"status": "completed", "output": result})
        except Exception as e:
            ongoing_tasks[task_id].update({"status": "error", "error": str(e)})
        return task_id
    return "Skill not found :("

def execute_task_list(objective, api_keys, task_id):
    """Execute a list of tasks"""
    try:
        task_registry = TaskRegistry()
        result = run_main_loop(objective, get_skills(), api_keys)
        ongoing_tasks[task_id].update({"status": "completed", "output": result})
        return task_registry.get_tasks(), task_id
    except Exception as e:
        ongoing_tasks[task_id].update({"status": "error", "error": str(e)})
        print(f"Error in execute_task_list: {e}")
    return task_id



def get_skills():
    """Return the global skill registry"""
    # Removed repeated logic for initiating skill registry
    global global_skill_registry
    print("Returning GLOBAL SKILL REGISTRY")
    return global_skill_registry

# Removed repeated logic for initiating skill registry
global_skill_registry = SkillRegistry(api_keys=api_keys, main_loop_function=run_main_loop, skill_names=LOAD_SKILLS)

if __name__ == "__main__":
    run_main_loop(OBJECTIVE, LOAD_SKILLS, api_keys, REFLECTION)


File: classic/babyfoxagi/main.py

from flask import Flask, render_template, request, jsonify, send_from_directory
import openai
import os
import json
import threading
from babyagi import get_skills, execute_skill, execute_task_list, api_keys, LOAD_SKILLS
from ongoing_tasks import ongoing_tasks

app = Flask(__name__, static_folder='public/static')
openai.api_key = os.getenv('OPENAI_API_KEY')


@app.route('/')
def hello_world():

  return render_template('index.html')


FOREVER_CACHE_FILE = "forever_cache.ndjson"
OVERALL_SUMMARY_FILE = "overall_summary.ndjson"


@app.route('/get-all-messages', methods=["GET"])
def get_all_messages():
  try:
    messages = []
    with open("forever_cache.ndjson", "r") as file:
      for line in file:
        messages.append(json.loads(line))

    return jsonify(messages)

  except Exception as e:
    return jsonify({"error": str(e)}), 500


def get_latest_summary():
  with open(OVERALL_SUMMARY_FILE, 'r') as file:
    lines = file.readlines()
  if lines:
    return json.loads(lines[-1])["summary"]  # Return the latest summary
  return ""


def summarize_text(text):
  system_message = (
    "Your task is to generate a concise summary for the provided conversation, this will be fed to a separate AI call as context to generate responses for future user queries."
    " The conversation contains various messages that have been exchanged between participants."
    " Please ensure your summary captures the main points and context of the conversation without being too verbose."
    " The summary should be limited to a maximum of 500 tokens."
    " Here's the conversation you need to summarize:")

  completion = openai.ChatCompletion.create(model="gpt-3.5-turbo-16k",
                                            messages=[{
                                              "role": "system",
                                              "content": system_message
                                            }, {
                                              "role": "user",
                                              "content": text
                                            }])

  # Extracting the content from the assistant's message
  return completion.choices[0]['message']['content'].strip()


def combine_summaries(overall, latest):
  system_message = (
    "Your task is to generate a concise summary for the provided conversation, this will be fed to a separate AI call as context to generate responses for future user queries."
    "You will do this by combining two given summaries into one cohesive summary."
    " Make sure to retain the key points from both summaries and create a concise, unified summary."
    " The combined summary should not exceed 500 tokens."
    " Here are the summaries you need to combine:")

  completion = openai.ChatCompletion.create(
    model="gpt-3.5-turbo-16k",
    messages=[{
      "role": "system",
      "content": system_message
    }, {
      "role":
      "user",
      "content":
      f"Overall summary: {overall}\nLatest summary: {latest}"
    }])

  # Extracting the content from the assistant's message
  return completion.choices[0]['message']['content'].strip()



def openai_function_call(user_message):
  global ongoing_tasks
  global global_skill_registry
  global_skill_registry = LOAD_SKILLS
  print("Returning GLOBAL SKILL REGISTRY")
  print(global_skill_registry)
  
  # Append the new user message to the forever_cache file
  user_entry = {"role": "user", "content": user_message}
  append_to_ndjson(FOREVER_CACHE_FILE, user_entry)

  # Retrieve the last 20 stored messages
  with open(FOREVER_CACHE_FILE, "r") as file:
    lines = file.readlines()

  last_20_messages = [json.loads(line) for line in lines][-20:]

  # Always update the summary in a separate thread
  threading.Thread(target=update_summary, args=(last_20_messages, )).start()

  overall_summary = get_latest_summary()
  print("LOAD_SKILLS")
  print(global_skill_registry)
  system_message = (
    f"You are a fun happy and quirky AI chat assistant that uses Gen Z language and lots of emojis named BabyAGI with capabilities beyond chat. For every user message, you quickly analyze whether this is a request that you can simply respond via ChatCompletion, whether you need to use one of the skills provided, or whether you should create a task list and chain multiple skills together. You will always provide a message_to_user. If path is Skill or TaskList, always generate an objective. If path is Skill, ALWAYS include skill_used from one of the available skills. ###Here are your available skills: {global_skill_registry}.###For context, here is the overall summary of the chat: {overall_summary}."
  )

  completion = openai.ChatCompletion.create(
    model="gpt-3.5-turbo-16k",
    messages=[
      {"role": "system","content": system_message},
      *last_20_messages,
      {"role": "user","content": user_message},
    ],
    functions=[{
      "name": "determine_response_type",
      "description":
      "Determine whether to respond via ChatCompletion, use a skill, or create a task list. Always provide a message_to_user.",
      "parameters": {
        "type": "object",
        "properties": {
          "message_to_user": {
            "type":
            "string",
            "description":
            "A message for the user, indicating the AI's action or providing the direct chat response. ALWAYS REQUIRED. Do not use line breaks."
          },
          "path": {
            "type":
            "string",
            "enum": ["ChatCompletion", "Skill", "TaskList"],  # Restrict the values to these three options
            "description":
            "The type of response – either 'ChatCompletion', 'Skill', or 'TaskList'"
          },
          "skill_used": {
            "type":
            "string",
            "description":
            f"If path is 'Skill', indicates which skill to use. If path is 'Skill', ALWAYS use skill_used. Must be one of these: {global_skill_registry}"
          },
          "objective": {
            "type":
            "string",
            "description":
            "If path is 'Skill' or 'TaskList', describes the main task or objective. Always include if path is 'Skill' or 'TaskList'."
          }
        },
        "required": ["path", "message_to_user", "objective", "skill_used"]
      }
    }],
    function_call={"name": "determine_response_type"})

  # Extract AI's structured response from function call
  response_data = completion.choices[0]['message']['function_call'][
    'arguments']
  if isinstance(response_data, str):
    response_data = json.loads(response_data)
  print("RESPONSE DATA:")
  print(response_data)
  path = response_data.get("path")
  skill_used = response_data.get("skill_used")
  objective = response_data.get("objective")
  task_id = generate_task_id()
  response_data["taskId"] = task_id
  if path == "Skill":
      ongoing_tasks[task_id] = {
          'status': 'ongoing',
          'description': objective,
          'skill_used': skill_used
      }
      threading.Thread(target=execute_skill, args=(skill_used, objective, task_id)).start()
      update_ongoing_tasks_file()
  elif path == "TaskList":
      ongoing_tasks[task_id] = {
          'status': 'ongoing',
          'description': objective,
          'skill_used': 'Multiple'
      }
      threading.Thread(target=execute_task_list, args=(objective, api_keys, task_id)).start()
      update_ongoing_tasks_file()

  return response_data


def generate_task_id():
  """Generates a unique task ID"""
  return f"{str(len(ongoing_tasks) + 1)}"


def update_summary(messages):
  # Combine messages to form text and summarize
  messages_text = " ".join([msg['content'] for msg in messages])
  latest_summary = summarize_text(messages_text)

  # Update overall summary
  overall = get_latest_summary()
  combined_summary = combine_summaries(overall, latest_summary)
  append_to_ndjson(OVERALL_SUMMARY_FILE, {"summary": combined_summary})


@app.route('/determine-response', methods=["POST"])
def determine_response():
  try:
    # Ensure that the request contains JSON data
    if not request.is_json:
      return jsonify({"error": "Expected JSON data"}), 400

    user_message = request.json.get("user_message")

    # Check if user_message is provided
    if not user_message:
      return jsonify({"error": "user_message field is required"}), 400

    response_data = openai_function_call(user_message)

    data = {
      "message": response_data['message_to_user'],
      "skill_used": response_data.get('skill_used', None),
      "objective": response_data.get('objective', None),
      "task_list": response_data.get('task_list', []),
      "path": response_data.get('path', []),
      "task_id": response_data.get('taskId')
    }

    # Storing AI's response to NDJSON file
    ai_entry = {
      "role": "assistant",
      "content": response_data['message_to_user']
    }
    append_to_ndjson("forever_cache.ndjson", ai_entry)

    print("END OF DETERMINE-RESPONSE. PRINTING 'DATA'")
    print(data)
    return jsonify(data)

  except Exception as e:
    print(f"Exception occurred: {str(e)}")
    return jsonify({"error": str(e)}), 500


def append_to_ndjson(filename, data):
  try:
    print(f"Appending to {filename} with data: {data}")
    with open(filename, 'a') as file:
      file.write(json.dumps(data) + '\n')
  except Exception as e:
    print(f"Error in append_to_ndjson: {str(e)}")



@app.route('/check-task-status/<task_id>', methods=["GET"])
def check_task_status(task_id):
    global ongoing_tasks
    update_ongoing_tasks_file()
    print("CHECK_TASK_STATUS")
    print(task_id)
    task = ongoing_tasks.get(task_id)

    # First check if task is None
    if not task:
        return jsonify({"error": f"No task with ID {task_id} found."}), 404

    # Now, it's safe to access attributes of task
    print(task.get("status"))
    return jsonify({"status": task.get("status")})


@app.route('/fetch-task-output/<task_id>', methods=["GET"])
def fetch_task_output(task_id):
  print("FETCH_TASK_STATUS")
  print(task_id)
  task = ongoing_tasks.get(task_id)
  if not task:
    return jsonify({"error": f"No task with ID {task_id} found."}), 404
  return jsonify({"output": task.get("output")})


def update_ongoing_tasks_file():
    with open("ongoing_tasks.py", "w") as file:
        file.write(f"ongoing_tasks = {ongoing_tasks}\n")

@app.route('/get-all-tasks', methods=['GET'])
def get_all_tasks():
    tasks = []
    for task_id, task_data in ongoing_tasks.items():
        task = {
            'task_id': task_id,
            'status': task_data.get('status', 'unknown'),
            'description': task_data.get('description', 'N/A'),
            'skill_used': task_data.get('skill_used', 'N/A'),
            'output': task_data.get('output', 'N/A')
        }
        tasks.append(task)
    return jsonify(tasks)



if __name__ == "__main__":
  app.run(host='0.0.0.0', port=8080)



File: classic/babyfoxagi/tasks/task_registry.py

import openai
import json
import threading
import os
import numpy as np
from datetime import datetime
from collections import defaultdict

class TaskRegistry:
    def __init__(self):
        self.tasks = []
        # Initialize the lock
        self.lock = threading.Lock()
        objectives_file_path = "tasks/example_objectives"
        self.example_loader = ExampleObjectivesLoader(objectives_file_path)

    def load_example_objectives(self, user_objective):
        return self.example_loader.load_example_objectives(user_objective)

      
    def create_tasklist(self, objective, skill_descriptions):
        #reflect on objective
        notes = self.reflect_on_objective(objective,skill_descriptions)
        #load most relevant object and tasklist from objectives_examples.json
        example_objective, example_tasklist, example_reflection = self.load_example_objectives(objective)

        prompt = (
            f"You are an expert task list creation AI tasked with creating a  list of tasks as a JSON array, considering the ultimate objective of your team: {objective}. "
            f"Create a very short task list based on the objective, the final output of the last task will be provided back to the user. Limit tasks types to those that can be completed with the available skills listed below. Task description should be detailed.###"
            f"AVAILABLE SKILLS: {skill_descriptions}.###"
            f"RULES:"
            f"Do not use skills that are not listed."
            f"Always provide an ID to each task."
            f"Always include one skill."
            f"The final task should always output the final result of the overall objective."
            f"dependent_task_ids should always be an empty array, or an array of numbers representing the task ID it should pull results from."
            f"Make sure all task IDs are in chronological order.###\n"
            f"Helpful Notes as guidance:{notes}###\n"
            f"EXAMPLE OBJECTIVE={json.dumps(example_objective)}"
            f"TASK LIST={json.dumps(example_tasklist)}"
            f"OBJECTIVE={objective}"
            f"TASK LIST="
        )
        #print(prompt)
        print("\033[90m\033[3m" + "\nInitializing...\n" + "\033[0m")
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=[
                {
                    "role": "system",
                    "content": "You are a task creation AI."
                },
                {
                    "role": "user",
                    "content": prompt
                }
            ],
            temperature=0,
            max_tokens=2500,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )

        # Extract the content of the assistant's response and parse it as JSON
        result = response["choices"][0]["message"]["content"]
        try:
            task_list = json.loads(result)
            #print(task_list)
            self.tasks = task_list
        except Exception as error:
            print(error)


    def reflect_on_objective(self, objective, skill_descriptions):
        #load most relevant object and tasklist from objectives_examples.json
        example_objective, example_tasklist, example_reflection = self.load_example_objectives(objective)

        prompt = (
            f"You are an Ai specializing in generating helpful thoughts and ideas on tackling an objective, and your task is to think about how to tackle this objective: {objective}. "
            f"These are the skills available to you: {skill_descriptions}.###"
            f"Think about what tools and information you need to handle this objective, and which of the available skills would be most helpful to you and writea descriptive note to pass onto a task creation AI."
            f"Consider the following example objective, tasklist, and reflection as a sample."
            f"###EXAMPLE OBJECTIVE:{example_objective}."
            f"###EXAMPLE TASKLIST:{example_tasklist}."
            f"###REFLECTION FROM EXAMPLE:{example_reflection}."
            f"###THE AI AGENT'S OBJECTIVE:{example_reflection}."
            f"###INSTRUCTION: please provide helpful notes for the task creation agent specific to this objective."
        )
        #print(prompt)
        print("\033[90m\033[3m" + "\nInitializing...\n" + "\033[0m")
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=[
                {
                    "role": "system",
                    "content": f"You are an Ai specializing in generating helpful thoughts and ideas on tackling an objective, and your task is to think about how to tackle this objective: {objective}. "
                },
                {
                    "role": "user",
                    "content": prompt
                }
            ],
            temperature=0,
            max_tokens=250,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )

        # Extract the content of the assistant's response and parse it as JSON
        result = response["choices"][0]["message"]["content"]
        print(result)
        return result


    def execute_task(self, i, task, skill_registry, task_outputs, objective):
        p_nexttask="\033[92m\033[1m"+"\n*****NEXT TASK ID:"+str(task['id'])+"*****\n"+"\033[0m\033[0m"
        p_nexttask += f"\033[ EExecuting task {task.get('id')}: {task.get('task')}) [{task.get('skill')}]\033[)"
        print(p_nexttask)
        # Retrieve the skill from the registry
        skill = skill_registry.get_skill(task['skill'])
        # Get the outputs of the dependent tasks
        dependent_task_outputs = {dep: task_outputs[dep]["output"] for dep in task['dependent_task_ids']} if 'dependent_task_ids' in task else {}
        # Execute the skill
        # print("execute:"+str([task['task'], dependent_task_outputs, objective]))
        task_output = skill.execute(task['task'], dependent_task_outputs, objective)
        print("\033[93m\033[1m"+"\nTask Output (ID:"+str(task['id'])+"):"+"\033[0m\033[0m")
        print("TASK: "+str(task["task"]))
        print("OUTPUT: "+str(task_output))
        return i, task_output

  
    def reorder_tasks(self):
        self.tasks= sorted(self.tasks, key=lambda task: task['id'])


  
    def add_task(self, task, after_task_id):
        # Get the task ids
        task_ids = [t["id"] for t in self.tasks]

        # Get the index of the task id to add the new task after
        insert_index = task_ids.index(after_task_id) + 1 if after_task_id in task_ids else len(task_ids)

        # Insert the new task
        self.tasks.insert(insert_index, task)
        self.reorder_tasks()


    def get_tasks(self):
        return self.tasks

    def update_tasks(self, task_update):
        for task in self.tasks:
            if task['id'] == task_update['id']:
                task.update(task_update)
                self.reorder_tasks()


    def reflect_on_output(self, task_output, skill_descriptions):
        with self.lock:
            example = [
                [
                    {"id": 3, "task": "New task 1 description", "skill": "text_completion_skill",
                     "dependent_task_ids": [], "status": "complete"},
                    {"id": 4, "task": "New task 2 description", "skill": "text_completion_skill",
                     "dependent_task_ids": [], "status": "incomplete"}
                ],
                [2, 3],
                {"id": 5, "task": "Complete the objective and provide a final report",
                 "skill": "text_completion_skill", "dependent_task_ids": [1, 2, 3, 4], "status": "incomplete"}
            ]

            prompt = (
                f"You are an expert task manager, review the task output to decide whether any new tasks need to be added, or whether any tasks need to be updated."
                f"As you add a new task, see if there are any tasks that need to be updated (such as updating dependencies)."
                f"Use the current task list as reference."
                f"Do not add duplicate tasks to those in the current task list."
                f"Only provide JSON as your response without further comments."
                f"Every new and updated task must include all variables, even they are empty array."
                f"Dependent IDs must be smaller than the ID of the task."
                f"New tasks IDs should be no larger than the last task ID."
                f"Always select at least one skill."
                f"Task IDs should be unique and in chronological order."                f"Do not change the status of complete tasks."
                f"Only add skills from the AVAILABLE SKILLS, using the exact same spelling."
                f"Provide your array as a JSON array with double quotes. The first object is new tasks to add as a JSON array, the second array lists the ID numbers where the new tasks should be added after (number of ID numbers matches array), and the third object provides the tasks that need to be updated."
                f"Make sure to keep dependent_task_ids key, even if an empty array."
                f"AVAILABLE SKILLS: {skill_descriptions}.###"
                f"\n###Here is the last task output: {task_output}"
                f"\n###Here is the current task list: {self.tasks}"
                f"\n###EXAMPLE OUTPUT FORMAT = {json.dumps(example)}"
                f"\n###OUTPUT = "
            )
            print("\033[90m\033[3m" + "\nReflecting on task output to generate new tasks if necessary...\n" + "\033[0m")
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo-16k-0613",
                messages=[
                    {
                        "role": "system",
                        "content": "You are a task creation AI."
                    },
                    {
                        "role": "user",
                        "content": prompt
                    }
                ],
                temperature=0.7,
                max_tokens=1500,
                top_p=1,
                frequency_penalty=0,
                presence_penalty=0
            )

            # Extract the content of the assistant's response and parse it as JSON
            result = response["choices"][0]["message"]["content"]
            print("\n#" + str(result))

            # Check if the returned result has the expected structure
            if isinstance(result, str):
                try:
                    task_list = json.loads(result)
                    print("####task_list in function")
                    
                    print(task_list)
                    print("####task_list split in function")
                    print(task_list[0], task_list[1], task_list[2])
                    return task_list[0], task_list[1], task_list[2]
                except Exception as error:
                    print(error)

            else:
                raise ValueError("Invalid task list structure in the output")

    def get_tasks(self):
        """
        Returns the current list of tasks.

        Returns:
        list: the list of tasks.
        """
        return self.tasks

    def get_task(self, task_id):
        """
        Returns a task given its task_id.

        Parameters:
        task_id : int
            The unique ID of the task.

        Returns:
        dict
            The task that matches the task_id.
        """
        matching_tasks = [task for task in self.tasks if task["id"] == task_id]

        if matching_tasks:
            return matching_tasks[0]
        else:
            print(f"No task found with id {task_id}")
            return None

    def print_tasklist(self, tasks):
        p_tasklist="\033[95m\033[1m" + "\n*****TASK LIST*****\n" + "\033[0m"
        for t in tasks:
            dependent_task_ids = t.get('dependent_task_ids', [])
            dependent_task = ""
            if dependent_task_ids:
                dependent_task = f"\033[31m<dependencies: {', '.join([f'#{dep_id}' for dep_id in dependent_task_ids])}>\033[0m"
            status_color = "\033[32m" if t.get('status') == "completed" else "\033[31m"
            p_tasklist+= f"\033[1m{t.get('id')}\033[0m: {t.get('task')} {status_color}[{t.get('status')}]\033[0m \033[93m[{t.get('skill')}] {dependent_task}\033[0m\n"
        print(p_tasklist)



    def reflect_tasklist(self, objective, task_list, task_outputs, skill_descriptions):
        prompt = (
            f"You are an expert task manager. Reflect on the objective, entire task list, and the corresponding outputs to generate a better task list for the objective."
            f"Do not included 'results', and change every status to 'incomplete'."
            f"Only provide JSON as your response without further comments. "
            f"Use the current task list as reference. "
            f"Always make at least one change to the current task list "
            f"OBJECTIVE: {objective}."
            f"AVAILABLE SKILLS: {skill_descriptions}."
            f"\n###Here is the current task list: {json.dumps(task_list)}"
            f"\n###Here is the task outputs: {json.dumps(task_outputs)}"
            f"\n###IMPROVED TASKLIST = "
        )
        print("\033[90m\033[3m" + "\nReflecting on entire task list...\n" + "\033[0m")
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=[
                {
                    "role": "system",
                    "content": "You are an AI specializing in reflecting on task lists and improving them. You will never simply return the provided task list, but always improve on it."
                },
                {
                    "role": "user",
                    "content": prompt
                }
            ],
            temperature=0,
            max_tokens=4000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        # Extract the content of the assistant's response and parse it as JSON
        result = response["choices"][0]["message"]["content"]
        try:
            improved_task_list = json.loads(result)
            # Formatting improved_task_list to your desired format
            formatted_improved_task_list = [{
                "objective": objective,
                "examples": improved_task_list,
                "date": datetime.now().strftime("%Y-%m-%d")
            }]
            with open(f'tasks/example_objectives/improved_{datetime.now().strftime("%Y%m%d%H%M%S")}.json', 'w') as f:
                json.dump(formatted_improved_task_list, f)
            print(f"IMPROVED TASK LIST:{formatted_improved_task_list}")
        except Exception as error:
            print(error)
          
    def reflect_on_result(self, objective, task_list, task_outputs, skill_descriptions):
        prompt = (
            f"You are an expert AI specializing in analyzing yourself, an autonomous agent that combines multiple LLM calls. Reflect on the objective, entire task list, and the corresponding outputs and provide an analysis of the performance of yourself and how you could have performed better."
            f"\n###OBJECTIVE: {objective}."
            f"\n###AVAILABLE SKILLS: {skill_descriptions}."
            f"\n###TASK LIST: {json.dumps(task_list)}"
            f"\n###TASK OUTPUTS: {json.dumps(task_outputs)}"
            f"\n###ANALYSIS:"
        )
        print("\033[90m\033[3m" + "\nReflecting on result...\n" + "\033[0m")
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=[
                {
                    "role": "system",
                    "content": "You are an expert AI specializing in analyzing yourself, an autonomous agent that combines multiple LLM calls. Reflect on the objective, entire task list, and the corresponding outputs and provide an analysis of the performance of yourself and how you could have performed better."
                },
                {
                    "role": "user",
                    "content": prompt
                }
            ],
            temperature=0,
            max_tokens=2000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        # Extract the content of the assistant's response and parse it as JSON
        result = response["choices"][0]["message"]["content"]
        try:
            print(result)
            return result
        except Exception as error:
            print(error)

  
    def reflect_on_final(self, objective, task_list, task_outputs, skill_descriptions):
        print("here!")
        system_content_result = "You are an expert AI specializing in analyzing yourself, an autonomous agent that combines multiple LLM calls. Reflect on the objective, entire task list, and the corresponding outputs and provide an analysis of the performance of yourself and how you could have performed better."
        role_content_result = (
            f"You are an expert AI specializing in analyzing yourself, an autonomous agent that combines multiple LLM calls. Reflect on the objective, entire task list, and the corresponding outputs and provide an analysis of the performance of yourself and how you could have performed better."
            f"\n###OBJECTIVE: {objective}."
            f"\n###AVAILABLE SKILLS: {skill_descriptions}."
            f"\n###TASK LIST: {json.dumps(task_list)}"
            f"\n###TASK OUTPUTS: {json.dumps(task_outputs)}"
            f"\n###ANALYSIS:"
        )
        print("\033[90m\033[3m" + "\nReflecting on result...\n" + "\033[0m")
        response = self.chatcompletion(role_content_result, system_content_result,500)
        # Extract the content of the assistant's response and parse it as JSON
        simple_reflection = response["choices"][0]["message"]["content"]
        try:
            print(simple_reflection)
        except Exception as error:
            print(error)
          
        system_content_task = "You are an AI specializing in reflecting on task lists and improving them. You will never simply return the provided task list, but always improve on it."
        role_content_task = (
            f"You are an expert task manager. Reflect on the objective, entire task list, and the corresponding outputs to generate a better task list for the objective."
            f"Do not included 'results', and change every status to 'incomplete'."
            f"Only provide JSON as your response without further comments. "
            f"Use the current task list as reference. "
            f"Always make at least one change to the current task list "
            f"OBJECTIVE: {objective}."
            f"AVAILABLE SKILLS: {skill_descriptions}."
            f"SIMPLE REFLECTION: {simple_reflection}."
            f"\n###Here is the current task list: {json.dumps(task_list)}"
            f"\n###Here is the task outputs: {json.dumps(task_outputs)}"
            f"\n###IMPROVED TASKLIST = "
        )
        print("\033[90m\033[3m" + "\nReflecting on entire task list...\n" + "\033[0m")
        response = self.chatcompletion(role_content_task, system_content_task,4000)
    
        # Extract the content of the assistant's response and parse it as JSON
        result = response["choices"][0]["message"]["content"]
        print(result)
        try:
            improved_task_list = json.loads(result)
            # Formatting improved_task_list to your desired format
            formatted_improved_task_list = [{
                "objective": objective,
                "examples": improved_task_list,
                "date": datetime.now().strftime("%Y-%m-%d"),
                "reflection":simple_reflection
            }]
            with open(f'tasks/example_objectives/improved_{datetime.now().strftime("%Y%m%d%H%M%S")}.json', 'w') as f:
                json.dump(formatted_improved_task_list, f)
            print(f"IMPROVED TASK LIST:{formatted_improved_task_list}")
        except Exception as error:
            print(error)



    def chatcompletion(self, role_content, system_content, max_tokens):
      return openai.ChatCompletion.create(
          model="gpt-3.5-turbo-16k",
          messages=[
              {
                  "role": "system",
                  "content": system_content
              },
              {
                  "role": "user",
                  "content": role_content
              }
          ],
          temperature=0,
          max_tokens=max_tokens,
          top_p=1,
          frequency_penalty=0,
          presence_penalty=0
      )


from datetime import datetime

class ExampleObjectivesLoader:
    def __init__(self, objectives_folder_path, decay_factor=0.01):
        self.objectives_folder_path = objectives_folder_path
        self.decay_factor = decay_factor
        self.objectives_examples = []  # Initialize as an empty list

    def load_objectives_examples(self):
        objectives_dict = defaultdict(dict)

        for filename in os.listdir(self.objectives_folder_path):
            file_path = os.path.join(self.objectives_folder_path, filename)
            with open(file_path, 'r') as file:
                objectives = json.load(file)

                for objective in objectives:
                    key = objective['objective']
                    date = objective.get('date', None)

                    if date is not None:
                        date = datetime.strptime(date, '%Y-%m-%d')

                    if key not in objectives_dict or (date and datetime.strptime(objectives_dict[key]['date'], "%Y-%m-%d") < date):
                        objectives_dict[key] = objective

        self.objectives_examples = list(objectives_dict.values())

    def find_most_relevant_objective(self, user_input):
        user_input_embedding = self.get_embedding(user_input, model='text-embedding-ada-002')
        most_relevant_objective = max(
            self.objectives_examples,
            key=lambda pair: self.cosine_similarity(pair['objective'], user_input_embedding) * self.get_decay(pair)
        )
        return most_relevant_objective['objective'], most_relevant_objective['examples'], most_relevant_objective.get('reflection', '')

    def get_decay(self, objective):
        date = objective.get('date', None)
        if date is not None:
            date = datetime.strptime(date, '%Y-%m-%d')
            days_passed = (datetime.now() - date).days
        else:
            # if there's no date, assume a large number of days passed
            days_passed = 365 * 10  # 10 years

        decay = np.exp(-self.decay_factor * days_passed)
        return decay

    def get_embedding(self, text, model='text-embedding-ada-002'):
        response = openai.Embedding.create(input=[text], model=model)
        embedding = response['data'][0]['embedding']
        return embedding

    def cosine_similarity(self, objective, embedding):
        max_similarity = float('-inf')
        objective_embedding = self.get_embedding(objective, model='text-embedding-ada-002')
        similarity = self.calculate_similarity(objective_embedding, embedding)
        max_similarity = max(max_similarity, similarity)
        return max_similarity

    def calculate_similarity(self, embedding1, embedding2):
        embedding1 = np.array(embedding1, dtype=np.float32)
        embedding2 = np.array(embedding2, dtype=np.float32)
        similarity = np.dot(embedding1, embedding2) / (np.linalg.norm(embedding1) * np.linalg.norm(embedding2))
        return similarity

    def load_example_objectives(self, user_objective):
        self.load_objectives_examples()
        most_relevant_objective, most_relevant_tasklist, most_relevant_reflection = self.find_most_relevant_objective(user_objective)
        example_objective = most_relevant_objective
        example_tasklist = most_relevant_tasklist
        example_reflection = most_relevant_reflection
        return example_objective, example_tasklist, example_reflection



File: classic/babyfoxagi/tasks/__init__.py




File: classic/babyfoxagi/skills/skill.py

class Skill:
    name = 'base skill'
    description = 'This is the base skill.'
    api_keys_required = []

    def __init__(self, api_keys, main_loop_function):
        print(f"Initializing {self.name}")
        self.api_keys = api_keys
        self.main_loop_function = main_loop_function
        missing_keys = self.check_required_keys(api_keys)
        if missing_keys:
            print(f"Missing API keys for {self.name}: {missing_keys}")
            self.valid = False
        else:
            self.valid = True
        for key in self.api_keys_required:
            if isinstance(key, list):
                for subkey in key:
                    if subkey in api_keys:
                        setattr(self, f"{subkey}_api_key", api_keys.get(subkey))
            elif key in api_keys:
                setattr(self, f"{key}_api_key", api_keys.get(key))
        print(f"{self.name} initialized successfully")

    def check_required_keys(self, api_keys):
        missing_keys = []
        for key in self.api_keys_required:
            if isinstance(key, list):  # If the key is actually a list of alternatives
                if not any(k in api_keys for k in key):  # If none of the alternatives are present
                    missing_keys.append(key)  # Add the list of alternatives to the missing keys
            elif key not in api_keys:  # If the key is a single key and it's not present
                missing_keys.append(key)  # Add the key to the missing keys
        return missing_keys

    def execute(self, params, dependent_task_outputs, objective):
        raise NotImplementedError('Execute method must be implemented in subclass.')


File: classic/babyfoxagi/skills/image_generation.py

from skills.skill import Skill
import openai
import requests
import os
from urllib.parse import urlparse

class ImageGeneration(Skill):
    name = 'image_generation'
    description = "A drawing tool that uses OpenAI's DALL-E API to generate images."
    api_keys_required = ['openai']

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)
    
    def download_and_save_image(self, url, folder_path="public/static/images"):
        if not os.path.exists(folder_path):
            os.makedirs(folder_path)
        
        response = requests.get(url)
        
        # Extract file name from URL
        parsed_url = urlparse(url)
        file_name = os.path.basename(parsed_url.path)
        
        # Save image to folder
        file_path = os.path.join(folder_path, file_name)
        with open(file_path, 'wb') as f:
            f.write(response.content)
        
        return file_path

    def generate_prompt(self, objective):
        prompt_generation_messages = [
            {"role": "system", "content": "You are a helpful assistant specialized in generating creative and descriptive prompts for image generation. Your response is always one sentence with no line breaks. If the input is already detailed, do not change it."},
            {"role": "user", "content": "Generate a prompt for image creation based on the following objective: Draw a dystopian industrial city"},
            {"role": "assistant", "content": "Realistic painting of a dystopian industrial city with towering factories, pollution-filled air, and a gloomy sky."},
            {"role": "user", "content": "Generate a prompt for image creation based on the following objective: Draw interior photography of a cozy corner in a coffee shop, using natural light."},
            {"role": "assistant", "content": "Interior photography of a cozy corner in a coffee shop, using natural light."},
            {"role": "user", "content": "Generate a prompt for image creation based on the following objective: Draw an enchanted forest"},
            {"role": "assistant", "content": "Whimsical painting of an enchanted forest with mythical creatures, vibrant colors, and intricate details."},
            {"role": "user", "content": "Generate a prompt for image creation based on the following objective: " + objective}
        ]
        
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=prompt_generation_messages
        )
        
        return response.choices[0].message['content'].strip()
    
    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return
        print(f"Objective: {objective}")
    
        # Generate a creative prompt for DALL-E based on the objective
        generated_prompt = self.generate_prompt(objective)
        print(f"generated_prompt: {generated_prompt}")
    
        # Make API request to generate image based on the generated prompt
        response = openai.Image.create(
            prompt=generated_prompt,
            n=1,
            size="512x512",
            response_format="url"
        )
    
        # Extract URL of the generated image
        image_url = response.data[0]['url']
        print(f"image_url: {image_url}")
    
        # Download and save the image to your specified folder
        saved_image_path = self.download_and_save_image(image_url)
        print(f"saved_image_path: {saved_image_path}")

        # Generate the HTML code with a hard-coded relative path to display the image
        relative_image_path = os.path.relpath(saved_image_path, start="public")
        print(f"relative_image_path: {relative_image_path}")
        
        # The path is hardcoded here based on what you said works
        html_code = f'<img src="static/images/{os.path.basename(saved_image_path)}" alt="Generated Image">'
        print(f"html_code: {html_code}")

        return html_code



File: classic/babyfoxagi/skills/play_music.py

from skills.skill import Skill
import openai
import requests
import os
import json
from urllib.parse import urlparse

class PlayMusic(Skill):
    name = 'play_music'
    description = "A skill that uses the Deezer API to search for music and play it with an html player."
    api_keys_required = ['openai']

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)
    
    def download_and_save_image(self, url, folder_path="public/static/images"):
        if not os.path.exists(folder_path):
            os.makedirs(folder_path)
        
        response = requests.get(url)
        
        # Extract file name from URL
        parsed_url = urlparse(url)
        file_name = os.path.basename(parsed_url.path)
        
        # Save image to folder
        file_path = os.path.join(folder_path, file_name)
        with open(file_path, 'wb') as f:
            f.write(response.content)
        
        return file_path

    def generate_prompt(self, objective):
        prompt_generation_messages = [
            {"role": "system", "content": "You are a helpful assistant specialized in generating a query for Deezer's music API which is a query and search type. Search types are artist, album name, genre, and track name. Both query and search type should have single quotes, and separated by a comma"},
            {"role": "user", "content": "Play some Eminem"},
            {"role": "assistant", "content": "'Eminem','artist'"},
            {"role": "user", "content": "Play tropical house."},
            {"role": "assistant", "content": "'tropical house','genre'"},
            {"role": "user", "content": "Play 99 problems"},
            {"role": "assistant", "content": "'99 problems','track'"},
            {"role": "user", "content": "Play abbey road"},
            {"role": "assistant", "content": "'abbey road','album'"},
            {"role": "user", "content": objective}
        ]
        
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=prompt_generation_messages
        )
        
        return response.choices[0].message['content'].strip()

  
    def deezer_search(self, query, search_type):
        base_url = "https://api.deezer.com/search"
        
        params = {"q": f"{search_type}:'{query}'"}
        
        response = requests.get(base_url, params=params)
        
        if response.status_code != 200:
            print("Failed to get data:", response.status_code)
            return None
        
        results = json.loads(response.text)
        
        if not results['data']:
            print("No results found")
            return None
        
        first_result = results['data'][0]
        
        if search_type == "artist":
            artist_id = first_result['artist']['id']
            return f'<iframe title="deezer-widget" src="https://widget.deezer.com/widget/dark/artist/{artist_id}/top_tracks" width="100%" height="300" frameborder="0" allowtransparency="true" allow="encrypted-media; clipboard-write"></iframe>'
        elif search_type == "album":
            album_id = first_result['album']['id']
            return f'<iframe title="deezer-widget" src="https://widget.deezer.com/widget/dark/album/{album_id}" width="100%" height="300" frameborder="0" allowtransparency="true" allow="encrypted-media; clipboard-write"></iframe>'
        elif search_type == "track":
            track_id = first_result['id']
            return f'<iframe title="deezer-widget" src="https://widget.deezer.com/widget/dark/track/{track_id}" width="100%" height="300" frameborder="0" allowtransparency="true" allow="encrypted-media; clipboard-write"></iframe>'
        elif search_type == "genre":
            genre_id = first_result['id']
            return f'<iframe title="deezer-widget" src="https://widget.deezer.com/widget/dark/genre/{genre_id}" width="100%" height="300" frameborder="0" allowtransparency="true" allow="encrypted-media; clipboard-write"></iframe>'
        else:
            print("Unsupported search type")
            return None
    
    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return
        print(f"Objective: {objective}")

        generated_prompt = self.generate_prompt(objective)
        print(f"generated_prompt: {generated_prompt}")

        query, search_type = [x.strip().strip("'") for x in generated_prompt.split(",")]

        response = self.deezer_search(query, search_type)  # Modified this line

        return response



File: classic/babyfoxagi/skills/startup_analysis.py

from skills.skill import Skill
import openai
import json

class StartupAnalysis(Skill):
  name = 'startup_analysis'
  description = "A tool specialized in providing a detailed analysis of startups using OpenAI's text completion API. Analyzes parameters like customers, ICP, pain points, competitors, etc."
  api_keys_required = ['openai']

  def __init__(self, api_keys, main_loop_function):
    super().__init__(api_keys, main_loop_function)

  def execute(self, params, dependent_task_outputs, objective):
    if not self.valid:
      return

    completion = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": "You are a great AI analyst specialized in writing VC investment memos. Write one paragraph each in the third person on customers, ICP, pain points, alternate solutions, competitors, potential partners, key incumbents, history of innovation in the industry, key risks, legal considerations, important figures or influencers, and other recent trends in the space. The user will provide the startup description."
            },
            {
                "role": "user",
                "content": params
                # User provides a detailed description of their startup
            }
        ],
        functions=[
            {
                "name": "analyze_startup",
                "description": "Analyze the startup based on various parameters",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "customers": {
                            "type": "string",
                            "description": "Who are the startup's primary customers. Describe in detail in the third person."
                        },
                        "ICP": {
                            "type": "string",
                            "description": "What is the startup's Ideal Customer Profile. Describe in detail in the third person."
                        },
                        "pain_points": {
                            "type": "string",
                            "description": "What are the pain points the startup is trying to solve. Describe in detail in the third person."
                        },
                        "alternate_solutions": {
                            "type": "string",
                            "description": "What are the alternate solutions available in the market. Describe in detail in the third person."
                        },
                        "competitors": {
                            "type": "string",
                            "description": "Who are the startup's main competitors. Describe in detail in the third person."
                        },
                        "potential_partners": {
                            "type": "string",
                            "description": "Who could be the startup's potential partners. Describe in detail in the third person."
                        },
                        "key_incumbents": {
                            "type": "string",
                            "description": "Key incumbents in the industry. Describe in detail."
                        },
                        "history_of_innovation": {
                            "type": "string",
                            "description": "History of innovation in the startup's industry. Describe in detail in the third person."
                        },
                        "key_risks": {
                            "type": "string",
                            "description": "What are the key risks involved in the startup. Describe in detail in the third person."
                        },
                        "legal_considerations": {
                            "type": "string",
                            "description": "Any legal considerations that the startup should be aware of"
                        },
                        "important_figures": {
                            "type": "string",
                            "description": "Important figures or influencers in the industry. Describe in detail in the third person."
                        },
                        "recent_trends": {
                            "type": "string",
                            "description": "What are the recent trends in the startup's industry. Describe in detail."
                        }
                    },
                    "required": [
                        "customers", "ICP", "pain_points", "alternate_solutions",
                        "competitors", "potential_partners", "key_incumbents",
                        "history_of_innovation", "key_risks", "legal_considerations",
                        "important_figures", "recent_trends"
                    ],
                },
            },
        ],
        function_call={"name": "analyze_startup"}
    )
    
    response_data = completion.choices[0]['message']['function_call']['arguments']
    
    html_content = ""
    for key, value in json.loads(response_data).items():
        html_content += f"<strong>{key.capitalize()}</strong>: {value}</br>\n"
    

    return html_content



File: classic/babyfoxagi/skills/skill_saver.py

from skills.skill import Skill
import os
import openai

class SkillSaver(Skill):
    name = 'skill_saver'
    description = "A skill that saves code written in a previous step into a file within the skills folder. Not for writing code."
    api_keys_required = []

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return

        task_prompt = f"Extract the code and only the code from the dependent task output here: {dependent_task_outputs}  \n###\nCODE:"
      
        messages = [
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=messages,
            temperature=0.4,
            max_tokens=3000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        ) 
    
        code =  response.choices[0].message['content'].strip()
        task_prompt = f"Come up with a file name (eg. 'get_weather.py') for the following skill:{code}\n###\nFILE_NAME:"
      
        messages = [
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.4,
            max_tokens=3000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        ) 
    
        file_name =  response.choices[0].message['content'].strip()
        file_path = os.path.join('skills',file_name)

        try:
            with open(file_path, 'w') as file:
                file.write(code)
                print(f"Code saved successfully: {file_name}")
        except:
            print("Error saving code.")

        return None


File: classic/babyfoxagi/skills/call_babyagi.py

from skills.skill import Skill
import openai

#This is an experimental skill to have BabyAGI use BabyAGI as a skill, which really isnt necessary, and also buggy.
class CallBabyagi(Skill):
    name = 'call_babyagi'
    description = 'A skill that rewrites a task description into an objective, and sends it to itself (an autonomous agent) to complete. Helpful for research.'
    api_keys_required = [['openai']]

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        # Generate the new objective by rewriting the task description
        new_objective = self.generate_new_objective(params)

        LOAD_SKILLS = ['web_search','text_completion']
        # Send the new objective to the main loop
        result = self.main_loop_function(new_objective, LOAD_SKILLS, self.api_keys, False)

        # Return the main loop's output as the result of the skill
        return result


    def generate_new_objective(self, task_description):
        prompt = f"You are an AI assistant tasked with rewriting the following task description into an objective, and remove any mention of call_babyagi: {task_description}\nObjective:"

        messages = [
            {"role": "user", "content": prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=messages,
            temperature=0.2,
            max_tokens=1500,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )

        return response.choices[0].message['content'].strip()



File: classic/babyfoxagi/skills/documentation_search.py


from skills.skill import Skill
from serpapi import GoogleSearch
import openai
from bs4 import BeautifulSoup
import requests
import re

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36"
}

class DocumentationSearch(Skill):
    name = 'documentation_search'
    description = 'A tool that performs a web search for finding documentation.'
    api_keys_required = [['openai'],['serpapi']]

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        # Your function goes here

        
        # Modify the query based on the dependent task output
        if dependent_task_outputs != "":
            dependent_task = f"Use the dependent task output below as reference to help craft the correct search query for the provided task above. Dependent task output:{dependent_task_outputs}."
        else:
            dependent_task = "."
        query = self.text_completion_tool("You are an AI assistant tasked with generating a Google search query based on the following task: "+params+". If the task looks like a search query, return the identical search query as your response. " + dependent_task + "\nSearch Query:")
        print("\033[90m\033[3m"+"Search query: " +str(query)+"\033[0m")
        # Set the search parameters
        search_params = {
            "engine": "google",
            "q": query,
            "api_key": self.serpapi_api_key,
            "num": 3
        }
        # Perform the web search
        search_results = GoogleSearch(search_params).get_dict()
        
        # Simplify the search results
        search_results = self.simplify_search_results(search_results.get('organic_results', []))
        print("\033[90m\033[3mCompleted search. Now scraping results.\n\033[0m")

        # Store the results from web scraping
        results = ""
        for result in search_results:
            url = result.get('link')
            print("\033[90m\033[3m" + "Scraping: "+url+"" + "...\033[0m")
            content = self.web_scrape_tool({"url": url, "task": params,"objective":objective})
            results += "URL:"+url+"CONTENT:"+str(content) + ". "
        print("\033[90m\033[3m"+str(results[0:100])[0:100]+"...\033[0m")
        # Process the results and generate a report
        results = self.text_completion_tool(f"You are an expert analyst combining the results of multiple web scrapes. Rewrite the following information as one cohesive report without removing any facts, variables, or code snippets. Any chance you get, make sure to capture working or good example code examples. Ignore any reports of not having info, unless all reports say so - in which case explain that the search did not work and suggest other web search queries to try. \n###INFORMATION:{results}.\n###REPORT:")
        return results

    def simplify_search_results(self, search_results):
        simplified_results = []
        for result in search_results:
            simplified_result = {
                "position": result.get("position"),
                "title": result.get("title"),
                "link": result.get("link"),
                "snippet": result.get("snippet")
            }
            simplified_results.append(simplified_result)
        return simplified_results

  
    def text_completion_tool(self, prompt: str):
        messages = [
            {"role": "user", "content": prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=messages,
            temperature=0.2,
            max_tokens=1500,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        return response.choices[0].message['content'].strip()


    def web_scrape_tool(self, params):
        content = self.fetch_url_content(params['url'])
        if content is None:
            return None
    
        text = self.extract_text(content)
        print("\033[90m\033[3m"+"Scrape completed. Length:" +str(len(text))+".Now extracting relevant info..."+"...\033[0m")
        info = self.extract_relevant_info(params['objective'], text[0:11000], params['task'])
        links = self.extract_links(content)
        #result = f"{info} URLs: {', '.join(links)}"
        result = info
        
        return result
    
    def fetch_url_content(self,url: str):
        try:
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            return response.content
        except requests.exceptions.RequestException as e:
            print(f"Error while fetching the URL: {e}")
            return ""
    
    def extract_links(self,content: str):
        soup = BeautifulSoup(content, "html.parser")
        links = [link.get('href') for link in soup.findAll('a', attrs={'href': re.compile("^https?://")})]
        return links
    
    def extract_text(self,content: str):
        soup = BeautifulSoup(content, "html.parser")
        text = soup.get_text(strip=True)
        return text
    
    def extract_relevant_info(self, objective, large_string, task):
        chunk_size = 12000
        overlap = 500
        notes = ""
      
        if len(large_string) == 0:
          print("error scraping")
          return "Error scraping."
        
        for i in range(0, len(large_string), chunk_size - overlap):
            
            print("\033[90m\033[3m"+"Reading chunk..."+"\033[0m")  
            chunk = large_string[i:i + chunk_size]
            
            messages = [
                {"role": "system", "content": f"You are an AI assistant."},
                {"role": "user", "content": f"You are an expert AI research assistant tasked with creating or updating the current notes. If the current note is empty, start a current-notes section by exracting relevant data to the task and objective from the chunk of text to analyze. Any chance you get, make sure to capture working or good example code examples. If there is a current note, add new relevant info frol the chunk of text to analyze. Make sure the new or combined notes is comprehensive and well written. Here's the current chunk of text to analyze: {chunk}. ### Here is the current task: {task}.### For context, here is the objective: {objective}.### Here is the data we've extraced so far that you need to update: {notes}.### new-or-updated-note:"}
            ]
    
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo-16k-0613",
                messages=messages,
                max_tokens=1500,
                n=1,
                stop="###",
                temperature=0.7,
            )
    
            notes += response.choices[0].message['content'].strip()+". ";
        
        return notes


File: classic/babyfoxagi/skills/directory_structure.py

from skills.skill import Skill
import os

class DirectoryStructure(Skill):
    name = 'directory_structure'
    description = "A tool that outputs the file and folder structure of its top parent folder."

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        # Get the current script path
        current_script_path = os.path.realpath(__file__)

        # Get the top parent directory of current script
        top_parent_path = self.get_top_parent_path(current_script_path)
        # Get the directory structure from the top parent directory
        dir_structure = self.get_directory_structure(top_parent_path)

        return dir_structure

    def get_directory_structure(self, start_path):
        dir_structure = {}
        ignore_dirs = ['.','__init__.py', '__pycache__', 'pydevd', 'poetry','venv']  # add any other directories to ignore here
    
        for root, dirs, files in os.walk(start_path):
            dirs[:] = [d for d in dirs if not any(d.startswith(i) for i in ignore_dirs)]  # exclude specified directories
            files = [f for f in files if not f[0] == '.' and f.endswith('.py')]  # exclude hidden files and non-Python files

            current_dict = dir_structure
            path_parts = os.path.relpath(root, start_path).split(os.sep)
            for part in path_parts:
                if part:  # skip empty parts
                    if part not in current_dict:
                        current_dict[part] = {}
                    current_dict = current_dict[part]
            for f in files:
                current_dict[f] = None
        #print("#############################")
        #print(str(current_dict)[0:100])
        return dir_structure
    
    

    def get_top_parent_path(self, current_path):
        relative_path = ""
        while True:
            new_path = os.path.dirname(current_path)
            print(new_path)
            if new_path == '/home/runner/BabyElfAGI/skills':  # reached the top
            #if new_path == current_path:  # reached the top
                #return relative_path
                return '/home/runner/BabyElfAGI'
            current_path = new_path
            relative_path = os.path.join("..", relative_path)
            print(relative_path)


File: classic/babyfoxagi/skills/drawing.py

from skills.skill import Skill
import openai

class Drawing(Skill):
    name = 'drawing'
    description = "A tool that draws on an HTML canvas using OpenAI."
    api_keys_required = ['openai']

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return
        
        task_prompt = f"Do a detailed HTML canvas drawing of the user request: {params}. Start with <canvas> and end with </canvas>. Only provide code.:"
      
        messages = [
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=messages,
            temperature=0.4,
            max_tokens=4000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        return "\n\n"+response.choices[0].message['content'].strip()



File: classic/babyfoxagi/skills/__init__.py




File: classic/babyfoxagi/skills/google_jobs_api_search.py

from skills.skill import Skill
from serpapi import GoogleSearch
import openai
from bs4 import BeautifulSoup
import requests
import re
import time

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36"
}

class GoogleJobsAPISearch(Skill):
    name = 'google_jobs_api_search'
    description = 'A skill for searching for job listings using the Google Jobs API.'
    api_keys_required = [['serpapi']]

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        # Modify the query based on the dependent task output
        if dependent_task_outputs != "":
            dependent_task = f"Use the dependent task output below as reference to help craft the correct search query for the provided task above. Dependent task output:{dependent_task_outputs}."
        else:
            dependent_task = ""

        # Generate the search query
        query = self.text_completion_tool(f"You are an AI assistant tasked with generating a Google Jobs API search query based on the following task: {params}. If the task looks like a search query, return the identical search query as your response. {dependent_task}\nSearch Query:")
        print("\033[90m\033[3m" + "Search query: " + str(query) + "\033[0m")

        # Set the search parameters
        search_params = {
            "engine": "google_jobs",
            "q": query,
            "api_key": self.serpapi_api_key,
            "num": 3
        }

        # Perform the job search
        job_results = GoogleSearch(search_params).get_dict()

        # Simplify the job results
        simplified_results = self.simplify_job_results(job_results.get('jobs_results', []))

        # Generate a report
        report = self.text_completion_tool(f"You are an expert analyst combining the results of multiple job searches. Rewrite the following information as one cohesive report without removing any facts. Keep job URL for each. Ignore any reports of not having info, unless all reports say so - in which case explain that the search did not work and suggest other job search queries to try.\n###INFORMATION:{simplified_results}.\n###REPORT:")

        time.sleep(1)

        return report

    def simplify_job_results(self, job_results):
        simplified_results = []
        for result in job_results:
            simplified_result = {
                "title": result.get("title"),
                "company_name": result.get("company_name"),
                "location": result.get("location"),
                "via": result.get("via"),
                "description": result.get("description"),
                "related_links": result.get("related_links"),
                "extensions": result.get("extensions"),
                "detected_extensions": result.get("detected_extensions")
            }
            simplified_results.append(simplified_result)
        return simplified_results

    def text_completion_tool(self, prompt: str):
        messages = [
            {"role": "user", "content": prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=messages,
            temperature=0.2,
            max_tokens=500,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )

        return response.choices[0].message['content'].strip()


File: classic/babyfoxagi/skills/text_completion.py

from skills.skill import Skill
import openai

class TextCompletion(Skill):
    name = 'text_completion'
    description = "A tool that uses OpenAI's text completion API to generate, summarize, and/or analyze text and code."
    api_keys_required = ['openai']

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return
        
        task_prompt = f"Complete your assigned task based on the objective and only based on information provided in the dependent task output, if provided. \n###\nYour objective: {objective}. \n###\nYour task: {params} \n###\nDependent tasks output: {dependent_task_outputs}  \n###\nYour task: {params}\n###\nRESPONSE:"
      
        messages = [
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=messages,
            temperature=0.4,
            max_tokens=4000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        return "\n\n"+response.choices[0].message['content'].strip()



File: classic/babyfoxagi/skills/pretty_me.py

import openai
import logging
from skills.skill import Skill
import os

# A little sloppy and experimental, but it's fun. Here's a demo: https://twitter.com/yoheinakajima/status/1699689836746358866

class PrettyMe(Skill):
    name = 'pretty_me'
    description = "A tool to update the current website UI by generating JavaScript that will execute upon load to change the front-end interface of the website it's on. The code response will be directly wrapped in a window.onload function and executed on the front-end. Use for any requests related to updating the UI (background, theme, etc.)"
    api_keys_required = ['openai']

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        try:
            if not self.valid:
                return "API keys not configured properly."

            # Read the contents of your CSS and HTML files
            css_path = os.path.join("public", "static", "style.css")
            html_path = os.path.join("templates", "index.html")

            with open(css_path, 'r') as css_file, open(html_path, 'r') as html_file:
                css_content = css_file.read()
                html_content = html_file.read()

            # Prepare the prompt with CSS and HTML content
            task_prompt = f"Generate JavaScript code to execute based on the following user input: {objective} {params}\nCSS Content: {css_content}\nHTML Content: {html_content}\nInstructions: only provide Javascript code that would go between the script tags, but do not provide the script tags.\n### Code:"
            example_input2 = "make me nature themed."

            example_output2 = '''
            // Remove body background image and adjust body background color
            document.body.style.backgroundImage = "none";
            document.body.style.backgroundColor = "#4CAF50"; // Green color
            
            // Adjust chat box background color
            document.querySelector(".chat-box").style.backgroundColor = "#4CAF50"; // Green color
            
            // Adjust chat messages' background color and bubble tail styles
            document.querySelectorAll(".bg-blue-200").forEach(function(element) {
                element.style.backgroundColor = "#357E68"; // Darker green
            });
            document.querySelectorAll(".bg-gray-300").forEach(function(element) {
                element.style.backgroundColor = "#295E4C"; // Darker green
                element.style.borderLeftColor = "#357E68"; // Border color matching user's bubble
            });
            
            // Adjust objectives box background color
            document.querySelector(".objectives-box").style.backgroundColor = "#4CAF50"; // Green color
            
            // Adjust task item background color
            document.querySelectorAll(".task-item").forEach(function(element) {
                element.style.backgroundColor = "#295E4C"; // Darker green
            });
            
            // Adjust task output background color
            document.querySelectorAll(".task-output").forEach(function(element) {
                element.style.backgroundColor = "#fffdfd"; // Light gray
            });
            '''

# Use the example2 variable in your execute function


            messages = [
                {"role": "user", "content": example_input2},
                {"role": "assistant", "content": example_output2},
                {"role": "user", "content": "make my background red."},
                {"role": "assistant", "content": "document.body.style.backgroundColor = \"red\";\ndocument.body.style.backgroundImage = \"none\";"},
                {"role": "user", "content": task_prompt}
            ]
            print(messages)
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo-16k",
                messages=messages,
                temperature=0,
                max_tokens=4000,
                top_p=1,
                frequency_penalty=0,
                presence_penalty=0
            )
            
            generated_code = response.choices[0].message['content'].strip()

            # Wrap the generated code with an onload event handler
            wrapped_code = f'''
            <script>
                var script = document.createElement("script");
                script.type = "text/javascript";
                script.innerHTML = `{generated_code}`;
                document.head.appendChild(script);
            </script>
            '''
            return "\n\n" + wrapped_code
        except Exception as exc:
            # Log the exception for debugging
            logging.error(f"Error in PrettyMe skill: {exc}")
            # Return a user-friendly error message
            return "An error occurred while processing your request."



File: classic/babyfoxagi/skills/code_reader.py

from skills.skill import Skill
import openai
import os

class CodeReader(Skill):
    name = 'code_reader'
    description = "A skill that finds a file's location in it's own program's directory and returns its contents."
    api_keys_required = ['openai']

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return

        dir_structure = self.get_directory_structure(self.get_top_parent_path(os.path.realpath(__file__)))
        print(f"Directory structure: {dir_structure}")
        example_dir_structure = {'.': {'main.py': None}, 'skills': {'__init__.py': None, 'web_scrape.py': None, 'skill.py': None, 'test_skill.py': None, 'text_completion.py': None, 'web_search.py': None, 'skill_registry.py': None, 'directory_structure.py': None, 'code_reader.py': None}, 'tasks': {'task_registry.py': None}, 'output': {}}
        example_params = "Analyze main.py"
        example_response = "main.py"
      
        task_prompt = f"Find a specific file in a directory and return only the file path, based on the task description below. Always return a directory.###The directory structure is as follows: \n{example_dir_structure}\nYour task: {example_params}\n###\nRESPONSE:{example_response} ###The directory structure is as follows: \n{dir_structure}\nYour task: {params}\n###\nRESPONSE:"
      
        messages = [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": task_prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.2,
            max_tokens=1500,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
        file_path = response.choices[0].message['content'].strip()
        print(f"AI suggested file path: {file_path}")

        try:
            with open(file_path, 'r') as file:
                file_content = file.read()
                #print(f"File content:\n{file_content}")
                return file_content
        except FileNotFoundError:
            print("File not found. Please check the AI's suggested file path.")
            return None

    def get_directory_structure(self, start_path):
        dir_structure = {}
        ignore_dirs = ['.','__init__.py', '__pycache__', 'pydevd', 'poetry','venv']  # add any other directories to ignore here
    
        for root, dirs, files in os.walk(start_path):
            dirs[:] = [d for d in dirs if not any(d.startswith(i) for i in ignore_dirs)]  # exclude specified directories
            files = [f for f in files if not f[0] == '.' and f.endswith('.py')]  # exclude hidden files and non-Python files

            current_dict = dir_structure
            path_parts = os.path.relpath(root, start_path).split(os.sep)
            for part in path_parts:
                if part:  # skip empty parts
                    if part not in current_dict:
                        current_dict[part] = {}
                    current_dict = current_dict[part]
            for f in files:
                current_dict[f] = None

        return dir_structure
    
    def get_top_parent_path(self, current_path):
        relative_path = ""
        while True:
            new_path = os.path.dirname(current_path)
            if new_path == '/home/runner/BabyfoxAGIUI/skills':  # reached the top
                return '/home/runner/BabyfoxAGIUI'
            current_path = new_path
            relative_path = os.path.join("..", relative_path)

        return relative_path



File: classic/babyfoxagi/skills/skill_registry.py

import os
import json
import openai
import importlib.util
import inspect
from .skill import Skill

class SkillRegistry:
    def __init__(self, api_keys, main_loop_function, skill_names=None):
        self.main_loop_function = main_loop_function
        self.skills = {}
        skill_files = [f for f in os.listdir('skills') if f.endswith('.py') and f != 'skill.py']
        for skill_file in skill_files:
            module_name = skill_file[:-3]
            if skill_names and module_name not in skill_names:
                continue
            module = importlib.import_module(f'skills.{module_name}')
            for attr_name in dir(module):
                attr_value = getattr(module, attr_name)
                if inspect.isclass(attr_value) and issubclass(attr_value, Skill) and attr_value is not Skill:
                    print(f"Attempting to instantiate skill: {attr_name}")
                  
                    try:
                        skill = attr_value(api_keys, self.main_loop_function)
                        if skill.valid:
                            self.skills[skill.name] = skill
                    except Exception as e:
                        print(f"Error while instantiating skill '{attr_name}': {e}")
                    skill = attr_value(api_keys, self.main_loop_function)
        # Print the names and descriptions of all loaded skills
        skill_info = "\n".join([f"{skill_name}: {skill.description}" for skill_name, skill in self.skills.items()])
        print(skill_info)

    def load_all_skills(self):
        skills_dir = os.path.dirname(__file__)
        for filename in os.listdir(skills_dir):
            if filename.endswith(".py") and filename not in ["__init__.py", "skill.py", "skill_registry.py"]:
                skill_name = filename[:-3]  # Remove .py extension
                self.load_skill(skill_name)

    def load_specific_skills(self, skill_names):
        for skill_name in skill_names:
            self.load_skill(skill_name)

    def load_skill(self, skill_name):
        skills_dir = os.path.dirname(__file__)
        filename = f"{skill_name}.py"
        if os.path.isfile(os.path.join(skills_dir, filename)):
            spec = importlib.util.spec_from_file_location(skill_name, os.path.join(skills_dir, filename))
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
            for item_name in dir(module):
                item = getattr(module, item_name)
                if isinstance(item, type) and issubclass(item, Skill) and item is not Skill:
                    skill_instance = item(self.api_keys)
                    self.skills[skill_instance.name] = skill_instance

    def get_skill(self, skill_name):
        skill = self.skills.get(skill_name)
        if skill is None:
            raise Exception(
                f"Skill '{skill_name}' not found. Please make sure the skill is loaded and all required API keys are set.")
        return skill

    def get_all_skills(self):
        return self.skills


    def reflect_skills(self, objective, task_list, task_outputs, skill_descriptions):
        prompt = (
            f"You are an expert task manager. Reflect on the objective, entire task list, and the corresponding outputs. "
            f"Determine whether the available skills were sufficient, and if not, identify and describe new skills that are needed. "
            f"Provide your response as a JSON array. "
            f"OBJECTIVE: {objective}."
            f"AVAILABLE SKILLS: {skill_descriptions}."
            f"\n###Here is the current task list: {json.dumps(task_list)}"
            f"\n###Here is the task outputs: {json.dumps(task_outputs)}."
            f"Missing skills:"
        )
        print("\033[90m\033[3m" + "\nReflecting on skills used in task list...\n" + "\033[0m")
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=[
                {
                    "role": "system",
                    "content": "You are an AI specializing in reflecting on skills used in tasks and identifying missing skills. You will provide a JSON array as your response."
                },
                {
                    "role": "user",
                    "content": prompt
                }
            ],
            temperature=0,
            max_tokens=4000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        # Extract the content of the assistant's response and parse it as JSON
        result = response["choices"][0]["message"]["content"]
        try:
            skills_analysis = json.loads(result)
            print(skills_analysis)
        except Exception as error:
            print(error)
    


File: classic/babyfoxagi/skills/web_search.py

from skills.skill import Skill
from serpapi import GoogleSearch
import openai
from bs4 import BeautifulSoup
import requests
import re
import time

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36"
}

class WebSearch(Skill):
    name = 'web_search'
    description = 'A tool that performs web searches.'
    api_keys_required = [['openai'],['serpapi']]

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        # Your function goes here

        
        # Modify the query based on the dependent task output
        if dependent_task_outputs != "":
            dependent_task = f"Use the dependent task output below as reference to help craft the correct search query for the provided task above. Dependent task output:{dependent_task_outputs}."
        else:
            dependent_task = "."
        query = self.text_completion_tool("You are an AI assistant tasked with generating a short and simple Google search query based on the following task: "+params+"."+dependent_task + "\nExample Task: Perform a web search to find reputable sources of news in the field of AI for today's date.\nExample Query:AI News\nExample Task:Perform a search for Yohei Nakajima at Untapped Capital\nExample Query:Yohei Nakajima Untapped Capital\nTask:"+params+"\nQuery:")
        print("\033[90m\033[3m"+"Search query: " +str(query)+"\033[0m")
        # Set the search parameters
        
        search_params = {
            "engine": "google",
            "q": query,
            "api_key": self.serpapi_api_key,
            "num": 3
        }
        # Perform the web search
        search_results = GoogleSearch(search_params).get_dict()
        
        # Simplify the search results
        search_results = self.simplify_search_results(search_results.get('organic_results', []))
        print("\033[90m\033[3mCompleted search. Now scraping results.\n\033[0m")

        # Store the results from web scraping
        results = ""
        for result in search_results:
            url = result.get('link')
            print("\033[90m\033[3m" + "Scraping: "+url+"" + "...\033[0m")
            content = self.web_scrape_tool({"url": url, "task": params,"objective":objective})
            results += str(content) + ". "
        print("\033[90m\033[3m"+str(results[0:100])[0:100]+"...\033[0m")
        # Process the results and generate a report
        results = self.text_completion_tool(f"You are an expert analyst combining the results of multiple web scrapes. Rewrite the following information as one cohesive report without removing any facts. Ignore any reports of not having info, unless all reports say so - in which case explain that the search did not work and suggest other web search queries to try. \n###INFORMATION:{results}.\n###REPORT:")
        time.sleep(1)
        print("DONE!!!")
        return results

    def simplify_search_results(self, search_results):
        simplified_results = []
        for result in search_results:
            simplified_result = {
                "position": result.get("position"),
                "title": result.get("title"),
                "link": result.get("link"),
                "snippet": result.get("snippet")
            }
            simplified_results.append(simplified_result)
        return simplified_results

  
    def text_completion_tool(self, prompt: str):
        messages = [
            {"role": "user", "content": prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=messages,
            temperature=0,
            max_tokens=2000,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        return response.choices[0].message['content'].strip()


    def web_scrape_tool(self, params):
        content = self.fetch_url_content(params['url'])
        if content is None:
            return None
    
        text = self.extract_text(content)
        print("\033[90m\033[3m"+"Scrape completed. Length:" +str(len(text))+".Now extracting relevant info..."+"...\033[0m")
        info = self.extract_relevant_info(params['objective'], text[0:11000], params['task'])
        links = self.extract_links(content)
        #result = f"{info} URLs: {', '.join(links)}"
        result = info
        
        return result
    
    def fetch_url_content(self,url: str):
        try:
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            return response.content
        except requests.exceptions.RequestException as e:
            print(f"Error while fetching the URL: {e}")
            return ""
    
    def extract_links(self,content: str):
        soup = BeautifulSoup(content, "html.parser")
        links = [link.get('href') for link in soup.findAll('a', attrs={'href': re.compile("^https?://")})]
        return links
    
    def extract_text(self,content: str):
        soup = BeautifulSoup(content, "html.parser")
        text = soup.get_text(strip=True)
        return text
    
    def extract_relevant_info(self, objective, large_string, task):
        chunk_size = 12000
        overlap = 500
        notes = ""
      
        if len(large_string) == 0:
          print("error scraping")
          return "Error scraping."
        
        for i in range(0, len(large_string), chunk_size - overlap):
            
            print("\033[90m\033[3m"+"Reading chunk..."+"\033[0m")  
            chunk = large_string[i:i + chunk_size]
            
            messages = [
                {"role": "system", "content": f"You are an AI assistant."},
                {"role": "user", "content": f"You are an expert AI research assistant tasked with creating or updating the current notes. If the current note is empty, start a current-notes section by exracting relevant data to the task and objective from the chunk of text to analyze. If there is a current note, add new relevant info frol the chunk of text to analyze. Make sure the new or combined notes is comprehensive and well written. Here's the current chunk of text to analyze: {chunk}. ### Here is the current task: {task}.### For context, here is the objective: {objective}.### Here is the data we've extraced so far that you need to update: {notes}.### new-or-updated-note:"}
            ]
    
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo-16k",
                messages=messages,
                max_tokens=2000,
                n=1,
                stop="###",
                temperature=0.7,
            )
    
            notes += response.choices[0].message['content'].strip()+". ";
        time.sleep(1)
      
        return notes


File: classic/babyfoxagi/skills/airtable_search.py

from skills.skill import Skill
import openai
import os
import requests
from urllib.parse import quote  # Python built-in URL encoding function

#This Airtable Skill is tuned to work with our Airtable set up, where we are seaeching a specific column in a specific table, in a specific base. ll three variables need to be set below (Lines 55-59)
class AirtableSearch(Skill):
    name = 'airtable_search'
    description = "A skill that retrieves data from our Airtable notes using a search. Useful for remembering who we talked to about a certain topic."
    api_keys_required = ['code_reader','skill_saver','documentation_search','text_completion']

    def __init__(self, api_keys, main_loop_function):
        super().__init__(api_keys, main_loop_function)

    def execute(self, params, dependent_task_outputs, objective):
        if not self.valid:
            return

        # Initialize a list to keep track of queries tried
        tried_queries = []

        # Modify the query based on the dependent task output
        if dependent_task_outputs != "":
            dependent_task = f"Use the dependent task output below as reference to help craft the correct search query for the provided task above. Dependent task output:{dependent_task_outputs}."
        else:
            dependent_task = "."
        
        # Initialize output
        output = ''
        
        while not output: 
            # If there are tried queries, add them to the prompt
            tried_queries_prompt = f" Do not include search queries we have tried, and instead try synonyms or misspellings. Search queries we have tried: {', '.join(tried_queries)}." if tried_queries else ""
            print(tried_queries_prompt)
            query = self.text_completion_tool(f"You are an AI assistant tasked with generating a one word search query based on the following task: {params}. Provide only the search query as a response. {tried_queries_prompt} Take into account output from the previous task:{dependent_task}.\nExample Task: Retrieve data from Airtable notes using the skill 'airtable_search' to find people we have talked to about TED AI.\nExample Query:TED AI\nExample Task:Conduct a search in our Airtable notes to identify investors who have expressed interest in climate.\nExample Query:climate\nTask:{params}\nQuery:")

            # Add the query to the list of tried queries
            tried_queries.append(query)

            print("\033[90m\033[3m"+"Search query: " +str(query)+"\033[0m")

            # Retrieve the Airtable API key
            airtable_api_key = self.api_keys['airtable']
          
            # Set the headers for the API request
            headers = {
                "Authorization": f"Bearer {airtable_api_key}",
                "Content-Type": "application/json"
            }
          

            # Set base id
            base_id = '<base_id>'
            table_name = '<table_name'


            # The field to filter on and the value to search for
            filter_field = '<filter_field>' 
            filter_value = query

            # URL encode the filter_field
            encoded_filter_field = quote(filter_field)

            # If filter_field contains a space, wrap it in curly brackets
            formula_field = f'{{{filter_field}}}' if ' ' in filter_field else f'{filter_field}'

            # Construct the Airtable formula and URL encode it
            formula = f'"{filter_value}",{formula_field}'
            encoded_formula = quote(formula)

            # Construct the Airtable API URL
            url = f"https://api.airtable.com/v0/{base_id}/{table_name}?fields%5B%5D={encoded_filter_field}&filterByFormula=FIND({encoded_formula})"
            print(url)
        

            # Make the API request to retrieve the data from Airtable
            response = requests.get(url, headers=headers)

            # Check if the API request was successful
            if response.status_code == 200:
                data = response.json()
                
                # Iterate through the records and process each one
                for record in data['records']:
                    # Combine objective, task, and record into a single string
                    input_str = f"Your objective:{objective}\nYour task:{params}\nThe Airtable record:{record['fields']}"
                    #print(record['fields'])
                  
                    if output == "":
                      instructions = ""
                    else:
                      instructions = f"Update the existing summary by adding information from this new record. ###Current summary:{output}."
                    
                    # Send the combined string to the OpenAI ChatCompletion API
                    response = self.text_completion_tool(f"You are an AI assistant that will review content from an Airtable record provided by the user, and extract only relevant information to the task at hand with context on why that information is relevant, taking into account the objective. If there is no relevant info, simply respond with '###'. Note that the Airtable note may use shorthand, and do your best to understand it first.{instructions} #####AIRTABLE DATA: {input_str}.") 
                    print("\033[90m\033[3m" +str(response)+"\033[0m")
                    output += response + "\n####"
            else:
                print(f"Failed to retrieve data from Airtable. Status code: {response.status_code}")
                return None

        # Return the combined output
      
        output = "Tried Queries: "+str(tried_queries)+"###Result:"+output
        return output


    def text_completion_tool(self, prompt: str):
        messages = [
            {"role": "user", "content": prompt}
        ]
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo-16k",
            messages=messages,
            temperature=0.2,
            max_tokens=350,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
    
        return response.choices[0].message['content'].strip()



File: extensions/ray_tasks.py

import sys
import logging
import ray
from collections import deque
from typing import Dict, List

from pathlib import Path
sys.path.append(str(Path(__file__).resolve().parent.parent))
from extensions.ray_objectives import CooperativeObjectivesListStorage

try:
    ray.init(address="auto", namespace="babyagi", logging_level=logging.FATAL, ignore_reinit_error=True)
except:
    ray.init(namespace="babyagi", logging_level=logging.FATAL, ignore_reinit_error=True)

@ray.remote
class CooperativeTaskListStorageActor:
    def __init__(self):
        self.tasks = deque([])
        self.task_id_counter = 0

    def append(self, task: Dict):
        self.tasks.append(task)

    def replace(self, tasks: List[Dict]):
        self.tasks = deque(tasks)

    def popleft(self):
        return self.tasks.popleft()

    def is_empty(self):
        return False if self.tasks else True

    def next_task_id(self):
        self.task_id_counter += 1
        return self.task_id_counter

    def get_task_names(self):
        return [t["task_name"] for t in self.tasks]

class CooperativeTaskListStorage:
    def __init__(self, name: str):
        self.name = name

        try:
            self.actor = ray.get_actor(name=self.name, namespace="babyagi")
        except ValueError:
            self.actor = CooperativeTaskListStorageActor.options(name=self.name, namespace="babyagi", lifetime="detached").remote()

        objectives = CooperativeObjectivesListStorage()
        objectives.append(self.name)

    def append(self, task: Dict):
        self.actor.append.remote(task)

    def replace(self, tasks: List[Dict]):
        self.actor.replace.remote(tasks)

    def popleft(self):
        return ray.get(self.actor.popleft.remote())

    def is_empty(self):
        return ray.get(self.actor.is_empty.remote())

    def next_task_id(self):
        return ray.get(self.actor.next_task_id.remote())

    def get_task_names(self):
        return ray.get(self.actor.get_task_names.remote())



File: extensions/dotenvext.py

from dotenv import load_dotenv

def load_dotenv_extensions(dotenv_files):
    for dotenv_file in dotenv_files:
        load_dotenv(dotenv_file)



File: extensions/weaviate_storage.py

import importlib
import logging
import re
from typing import Dict, List

import openai
import weaviate
from weaviate.embedded import EmbeddedOptions


def can_import(module_name):
    try:
        importlib.import_module(module_name)
        return True
    except ImportError:
        return False


assert can_import("weaviate"), (
    "\033[91m\033[1m"
    + "Weaviate storage requires package weaviate-client.\nInstall:  pip install -r extensions/requirements.txt"
)


def create_client(
    weaviate_url: str, weaviate_api_key: str, weaviate_use_embedded: bool
):
    if weaviate_use_embedded:
        client = weaviate.Client(embedded_options=EmbeddedOptions())
    else:
        auth_config = (
            weaviate.auth.AuthApiKey(api_key=weaviate_api_key)
            if weaviate_api_key
            else None
        )
        client = weaviate.Client(weaviate_url, auth_client_secret=auth_config)

    return client


class WeaviateResultsStorage:
    schema = {
        "properties": [
            {"name": "result_id", "dataType": ["string"]},
            {"name": "task", "dataType": ["string"]},
            {"name": "result", "dataType": ["text"]},
        ]
    }

    def __init__(
        self,
        openai_api_key: str,
        weaviate_url: str,
        weaviate_api_key: str,
        weaviate_use_embedded: bool,
        llm_model: str,
        llama_model_path: str,
        results_store_name: str,
        objective: str,
    ):
        openai.api_key = openai_api_key
        self.client = create_client(
            weaviate_url, weaviate_api_key, weaviate_use_embedded
        )
        self.index_name = None
        self.create_schema(results_store_name)

        self.llm_model = llm_model
        self.llama_model_path = llama_model_path

    def create_schema(self, results_store_name: str):
        valid_class_name = re.compile(r"^[A-Z][a-zA-Z0-9_]*$")
        if not re.match(valid_class_name, results_store_name):
            raise ValueError(
                f"Invalid index name: {results_store_name}. "
                "Index names must start with a capital letter and "
                "contain only alphanumeric characters and underscores."
            )

        self.schema["class"] = results_store_name
        if self.client.schema.contains(self.schema):
            logging.info(
                f"Index named {results_store_name} already exists. Reusing it."
            )
        else:
            logging.info(f"Creating index named {results_store_name}")
            self.client.schema.create_class(self.schema)

        self.index_name = results_store_name

    def add(self, task: Dict, result: Dict, result_id: int, vector: List):
        enriched_result = {"data": result}
        vector = self.get_embedding(enriched_result["data"])

        with self.client.batch as batch:
            data_object = {
                "result_id": result_id,
                "task": task["task_name"],
                "result": result,
            }
            batch.add_data_object(
                data_object=data_object, class_name=self.index_name, vector=vector
            )

    def query(self, query: str, top_results_num: int) -> List[dict]:
        query_embedding = self.get_embedding(query)

        results = (
            self.client.query.get(self.index_name, ["task"])
            .with_hybrid(query=query, alpha=0.5, vector=query_embedding)
            .with_limit(top_results_num)
            .do()
        )

        return self._extract_tasks(results)

    def _extract_tasks(self, data):
        task_data = data.get("data", {}).get("Get", {}).get(self.index_name, [])
        return [item["task"] for item in task_data]

    # Get embedding for the text
    def get_embedding(self, text: str) -> list:
        text = text.replace("\n", " ")

        if self.llm_model.startswith("llama"):
            from llama_cpp import Llama

            llm_embed = Llama(
                model_path=self.llama_model_path,
                n_ctx=2048,
                n_threads=4,
                embedding=True,
                use_mlock=True,
            )
            return llm_embed.embed(text)

        return openai.Embedding.create(input=[text], model="text-embedding-ada-002")[
            "data"
        ][0]["embedding"]



File: extensions/human_mode.py

import sys

def user_input_await(prompt: str) -> str:
    print("\033[94m\033[1m" + "\n> COPY FOLLOWING TEXT TO CHATBOT\n" + "\033[0m\033[0m")
    print(prompt)
    print("\033[91m\033[1m" + "\n AFTER PASTING, PRESS: (ENTER), (CTRL+Z), (ENTER) TO FINISH\n" + "\033[0m\033[0m")
    print("\033[96m\033[1m" + "\n> PASTE YOUR RESPONSE:\n" + "\033[0m\033[0m")
    input_text = sys.stdin.read()
    return input_text.strip()


File: extensions/argparseext.py

import os
import sys
import importlib
import argparse

def can_import(module_name):
    try:
        importlib.import_module(module_name)
        return True
    except ImportError:
        return False

# Extract the env filenames in the -e flag only
# Ignore any other arguments
def parse_dotenv_extensions(argv):
    env_argv = []
    if '-e' in argv:
        tmp_argv = argv[argv.index('-e') + 1:]
        parsed_args = []
        for arg in tmp_argv:
            if arg.startswith('-'):
                break
            parsed_args.append(arg)
        env_argv = ['-e'] + parsed_args

    parser = argparse.ArgumentParser()
    parser.add_argument('-e', '--env', nargs='+', help='''
    filenames for additional env variables to load
    ''', default=os.getenv("DOTENV_EXTENSIONS", "").split(' '))

    return parser.parse_args(env_argv).env

def parse_arguments():
    dotenv_extensions = parse_dotenv_extensions(sys.argv)
    # Check if we need to load any additional env files
    # This allows us to override the default .env file
    # and update the default values for any command line arguments
    if dotenv_extensions:
        from extensions.dotenvext import load_dotenv_extensions
        load_dotenv_extensions(parse_dotenv_extensions(sys.argv))

    # Now parse the full command line arguments
    parser = argparse.ArgumentParser(
        add_help=False,
    )
    parser.add_argument('objective', nargs='*', metavar='<objective>', help='''
    main objective description. Doesn\'t need to be quoted.
    if not specified, get objective from environment.
    ''', default=[os.getenv("OBJECTIVE", "")])
    parser.add_argument('-n', '--name', required=False, help='''
    instance name.
    if not specified, get the instance name from environment.
    ''', default=os.getenv("INSTANCE_NAME", os.getenv("BABY_NAME", "BabyAGI")))
    parser.add_argument('-m', '--mode', choices=['n', 'none', 'l', 'local', 'd', 'distributed'], help='''
    cooperative mode type
    ''', default='none')
    group = parser.add_mutually_exclusive_group()
    group.add_argument('-t', '--task', metavar='<initial task>', help='''
    initial task description. must be quoted.
    if not specified, get initial_task from environment.
    ''', default=os.getenv("INITIAL_TASK", os.getenv("FIRST_TASK", "")))
    group.add_argument('-j', '--join', action='store_true', help='''
    join an existing objective.
    install cooperative requirements.
    ''')
    group2 = parser.add_mutually_exclusive_group()
    group2.add_argument('-4', '--gpt-4', dest='llm_model', action='store_const', const="gpt-4", help='''
    use GPT-4 instead of the default model.
    ''')
    group2.add_argument('-l', '--llama', dest='llm_model', action='store_const', const="llama", help='''
    use LLaMa instead of the default model. Requires llama.cpp.
    ''')
    # This will parse -e again, which we want, because we need
    # to load those in the main file later as well
    parser.add_argument('-e', '--env', nargs='+', help='''
    filenames for additional env variables to load
    ''', default=os.getenv("DOTENV_EXTENSIONS", "").split(' '))
    parser.add_argument('-h', '-?', '--help', action='help', help='''
    show this help message and exit
    ''')

    args = parser.parse_args()

    llm_model = args.llm_model if args.llm_model else os.getenv("LLM_MODEL", os.getenv("OPENAI_API_MODEL", "gpt-3.5-turbo")).lower()

    dotenv_extensions = args.env

    instance_name = args.name
    if not instance_name:
        print("\033[91m\033[1m" + "BabyAGI instance name missing\n" + "\033[0m\033[0m")
        parser.print_help()
        parser.exit()

    module_name = "ray"
    cooperative_mode = args.mode
    if cooperative_mode in ['l', 'local'] and not can_import(module_name):
        print("\033[91m\033[1m"+f"Local cooperative mode requires package {module_name}\nInstall:  pip install -r extensions/requirements.txt\n" + "\033[0m\033[0m")
        parser.print_help()
        parser.exit()
    elif cooperative_mode in ['d', 'distributed']:
        print("\033[91m\033[1m" + "Distributed cooperative mode is not implemented yet\n" + "\033[0m\033[0m")
        parser.print_help()
        parser.exit()

    join_existing_objective = args.join
    if join_existing_objective and cooperative_mode in ['n', 'none']:
        print("\033[91m\033[1m"+f"Joining existing objective requires local or distributed cooperative mode\n" + "\033[0m\033[0m")
        parser.print_help()
        parser.exit()

    objective = ' '.join(args.objective).strip()
    if not objective:
        print("\033[91m\033[1m" + "No objective specified or found in environment.\n" + "\033[0m\033[0m")
        parser.print_help()
        parser.exit()

    initial_task = args.task
    if not initial_task and not join_existing_objective:
        print("\033[91m\033[1m" + "No initial task specified or found in environment.\n" + "\033[0m\033[0m")
        parser.print_help()
        parser.exit()

    return objective, initial_task, llm_model, dotenv_extensions, instance_name, cooperative_mode, join_existing_objective


File: extensions/__init__.py




File: extensions/pinecone_storage.py

from typing import Dict, List
import importlib
import openai
import pinecone
import re

def can_import(module_name):
    try:
        importlib.import_module(module_name)
        return True
    except ImportError:
        return False

assert (
    can_import("pinecone")
), "\033[91m\033[1m"+"Pinecone storage requires package pinecone-client.\nInstall:  pip install -r extensions/requirements.txt"

class PineconeResultsStorage:
    def __init__(self, openai_api_key: str, pinecone_api_key: str, pinecone_environment: str, llm_model: str, llama_model_path: str, results_store_name: str, objective: str):
        openai.api_key = openai_api_key
        pinecone.init(api_key=pinecone_api_key, environment=pinecone_environment)

        # Pinecone namespaces are only compatible with ascii characters (used in query and upsert)
        self.namespace = re.sub(re.compile('[^\x00-\x7F]+'), '', objective)

        self.llm_model = llm_model
        self.llama_model_path = llama_model_path

        results_store_name = results_store_name
        dimension = 1536 if not self.llm_model.startswith("llama") else 5120
        metric = "cosine"
        pod_type = "p1"
        if results_store_name not in pinecone.list_indexes():
            pinecone.create_index(
                results_store_name, dimension=dimension, metric=metric, pod_type=pod_type
            )

        self.index = pinecone.Index(results_store_name)
        index_stats_response = self.index.describe_index_stats()
        assert dimension == index_stats_response['dimension'], "Dimension of the index does not match the dimension of the LLM embedding"

    def add(self, task: Dict, result: str, result_id: int):
        vector = self.get_embedding(
            result
        )
        self.index.upsert(
            [(result_id, vector, {"task": task["task_name"], "result": result})], namespace=self.namespace
        )

    def query(self, query: str, top_results_num: int) -> List[dict]:
        query_embedding = self.get_embedding(query)
        results = self.index.query(query_embedding, top_k=top_results_num, include_metadata=True, namespace=self.namespace)
        sorted_results = sorted(results.matches, key=lambda x: x.score, reverse=True)
        return [(str(item.metadata["task"])) for item in sorted_results]

    # Get embedding for the text
    def get_embedding(self, text: str) -> list:
        text = text.replace("\n", " ")

        if self.llm_model.startswith("llama"):
            from llama_cpp import Llama

            llm_embed = Llama(
                model_path=self.llama_model_path,
                n_ctx=2048, n_threads=4,
                embedding=True, use_mlock=True,
            )
            return llm_embed.embed(text)

        return openai.Embedding.create(input=[text], model="text-embedding-ada-002")["data"][0]["embedding"]



File: extensions/ray_objectives.py

import logging
import ray
from collections import deque

ACTOR_NAME="BabyAGI Objectives"

try:
    ray.init(address="auto", namespace="babyagi", logging_level=logging.FATAL, ignore_reinit_error=True)
except:
    ray.init(namespace="babyagi", logging_level=logging.FATAL, ignore_reinit_error=True)

@ray.remote
class CooperativeObjectivesListStorageActor:
    def __init__(self):
        self.objectives = deque([])

    def append(self, objective: str):
        if not objective in self.objectives:
            self.objectives.append(objective)

    def is_empty(self):
        return False if self.objectives else True

    def get_objective_names(self):
        return [t for t in self.objectives]

class CooperativeObjectivesListStorage:
    def __init__(self):
        try:
            self.actor = ray.get_actor(name=ACTOR_NAME, namespace="babyagi")
        except ValueError:
            self.actor = CooperativeObjectivesListStorageActor.options(name=ACTOR_NAME, namespace="babyagi", lifetime="detached").remote()

    def append(self, objective: str):
        self.actor.append.remote(objective)

    def is_empty(self):
        return ray.get(self.actor.is_empty.remote())

    def get_objective_names(self):
        return ray.get(self.actor.get_objective_names.remote())



File: babycoder/babycoder.py

import os
import openai
import time
import sys
from typing import List, Dict, Union
from dotenv import load_dotenv
import json
import subprocess
import platform

from embeddings import Embeddings

# Set Variables
load_dotenv()
current_directory = os.getcwd()
os_version = platform.release()

openai_calls_retried = 0
max_openai_calls_retries = 3

# Set API Keys
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
assert OPENAI_API_KEY, "OPENAI_API_KEY environment variable is missing from .env"
openai.api_key = OPENAI_API_KEY

OPENAI_API_MODEL = os.getenv("OPENAI_API_MODEL", "gpt-3.5-turbo")
assert OPENAI_API_MODEL, "OPENAI_API_MODEL environment variable is missing from .env"

if "gpt-4" in OPENAI_API_MODEL.lower():
    print(
        f"\033[91m\033[1m"
        + "\n*****USING GPT-4. POTENTIALLY EXPENSIVE. MONITOR YOUR COSTS*****"
        + "\033[0m\033[0m"
    )

if len(sys.argv) > 1:
    OBJECTIVE = sys.argv[1]
elif os.path.exists(os.path.join(current_directory, "objective.txt")):
    with open(os.path.join(current_directory, "objective.txt")) as f:
        OBJECTIVE = f.read()

assert OBJECTIVE, "OBJECTIVE missing"

## Start of Helper/Utility functions ##

def print_colored_text(text, color):
    color_mapping = {
        'blue': '\033[34m',
        'red': '\033[31m',
        'yellow': '\033[33m',
        'green': '\033[32m',
    }
    color_code = color_mapping.get(color.lower(), '')
    reset_code = '\033[0m'
    print(color_code + text + reset_code)

def print_char_by_char(text, delay=0.00001, chars_at_once=3):
    for i in range(0, len(text), chars_at_once):
        chunk = text[i:i + chars_at_once]
        print(chunk, end='', flush=True) 
        time.sleep(delay) 
    print()

def openai_call(
    prompt: str,
    model: str = OPENAI_API_MODEL,
    temperature: float = 0.5,
    max_tokens: int = 100,
):
    global openai_calls_retried
    if not model.startswith("gpt-"):
        # Use completion API
        response = openai.Completion.create(
            engine=model,
            prompt=prompt,
            temperature=temperature,
            max_tokens=max_tokens,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
        return response.choices[0].text.strip()
    else:
        # Use chat completion API
        messages=[{"role": "user", "content": prompt}]
        try:
            response = openai.ChatCompletion.create(
                model=model,
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens,
                n=1,
                stop=None,
            )
            openai_calls_retried = 0
            return response.choices[0].message.content.strip()
        except Exception as e:
            # try again
            if openai_calls_retried < max_openai_calls_retries:
                openai_calls_retried += 1
                print(f"Error calling OpenAI. Retrying {openai_calls_retried} of {max_openai_calls_retries}...")
                return openai_call(prompt, model, temperature, max_tokens)

def execute_command_json(json_string):
    try:
        command_data = json.loads(json_string)
        full_command = command_data.get('command')
        
        process = subprocess.Popen(full_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, shell=True, cwd='playground')
        stdout, stderr = process.communicate(timeout=60)

        return_code = process.returncode

        if return_code == 0:
            return stdout
        else:
            return stderr

    except json.JSONDecodeError as e:
        return f"Error: Unable to decode JSON string: {str(e)}"
    except subprocess.TimeoutExpired:
        process.terminate()
        return "Error: Timeout reached (60 seconds)"
    except Exception as e:
        return f"Error: {str(e)}"

def execute_command_string(command_string):
    try:
        result = subprocess.run(command_string, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, shell=True, cwd='playground')
        output = result.stdout or result.stderr or "No output"
        return output

    except Exception as e:
        return f"Error: {str(e)}"

def save_code_to_file(code: str, file_path: str):
    full_path = os.path.join(current_directory, "playground", file_path)
    try:
        mode = 'a' if os.path.exists(full_path) else 'w'
        with open(full_path, mode, encoding='utf-8') as f:
            f.write(code + '\n\n')
    except:
        pass

def refactor_code(modified_code: List[Dict[str, Union[int, str]]], file_path: str):
    full_path = os.path.join(current_directory, "playground", file_path)

    with open(full_path, "r", encoding="utf-8") as f:
        lines = f.readlines()

    for modification in modified_code:
        start_line = modification["start_line"]
        end_line = modification["end_line"]
        modified_chunk = modification["modified_code"].splitlines()

        # Remove original lines within the range
        del lines[start_line - 1:end_line]

        # Insert the new modified_chunk lines
        for i, line in enumerate(modified_chunk):
            lines.insert(start_line - 1 + i, line + "\n")

    with open(full_path, "w", encoding="utf-8") as f:
        f.writelines(lines)

def split_code_into_chunks(file_path: str, chunk_size: int = 50) -> List[Dict[str, Union[int, str]]]:
    full_path = os.path.join(current_directory, "playground", file_path)

    with open(full_path, "r", encoding="utf-8") as f:
        lines = f.readlines()

    chunks = []
    for i in range(0, len(lines), chunk_size):
        start_line = i + 1
        end_line = min(i + chunk_size, len(lines))
        chunk = {"start_line": start_line, "end_line": end_line, "code": "".join(lines[i:end_line])}
        chunks.append(chunk)
    return chunks

## End of Helper/Utility functions ##

## TASKS AGENTS ##

def code_tasks_initializer_agent(objective: str):
    prompt = f"""You are an AGI agent responsible for creating a detailed JSON checklist of tasks that will guide other AGI agents to complete a given programming objective. Your task is to analyze the provided objective and generate a well-structured checklist with a clear starting point and end point, as well as tasks broken down to be very specific, clear, and executable by other agents without the context of other tasks.

    The current agents work as follows:
    - code_writer_agent: Writes code snippets or functions and saves them to the appropriate files. This agent can also append code to existing files if required.
    - code_refactor_agent: Responsible for modifying and refactoring existing code to meet the requirements of the task.
    - command_executor_agent: Executes terminal commands for tasks such as creating directories, installing dependencies, etc.

    Keep in mind that the agents cannot open files in text editors, and tasks should be designed to work within these agent capabilities.

    Here is the programming objective you need to create a checklist for: {objective}.

    To generate the checklist, follow these steps:

    1. Analyze the objective to identify the high-level requirements and goals of the project. This will help you understand the scope and create a comprehensive checklist.

    2. Break down the objective into smaller, highly specific tasks that can be worked on independently by other agents. Ensure that the tasks are designed to be executed by the available agents (code_writer_agent, code_refactor and command_executor_agent) without requiring opening files in text editors.

    3. Assign a unique ID to each task for easy tracking and organization. This will help the agents to identify and refer to specific tasks in the checklist.

    4. Organize the tasks in a logical order, with a clear starting point and end point. The starting point should represent the initial setup or groundwork necessary for the project, while the end point should signify the completion of the objective and any finalization steps.

    5. Provide the current context for each task, which should be sufficient for the agents to understand and execute the task without referring to other tasks in the checklist. This will help agents avoid task duplication.

    6. Pay close attention to the objective and make sure the tasks implement all necessary pieces needed to make the program work.
    
    7. Compile the tasks into a well-structured JSON format, ensuring that it is easy to read and parse by other AGI agents. The JSON should include fields such as task ID, description and file_path.

    IMPORTANT: BE VERY CAREFUL WITH IMPORTS AND MANAGING MULTIPLE FILES. REMEMBER EACH AGENT WILL ONLY SEE A SINGLE TASK. ASK YOURSELF WHAT INFORMATION YOU NEED TO INCLUDE IN THE CONTEXT OF EACH TASK TO MAKE SURE THE AGENT CAN EXECUTE THE TASK WITHOUT SEEING THE OTHER TASKS OR WHAT WAS ACCOMPLISHED IN OTHER TASKS.

    Pay attention to the way files are passed in the tasks, always use full paths. For example 'project/main.py'.

    Make sure tasks are not duplicated.

    Do not take long and complex routes, minimize tasks and steps as much as possible.

    Here is a sample JSON output for a checklist:

            {{
                "tasks": [
                    {{
                    "id": 1,
                    "description": "Run a command to create the project directory named 'project'",
                    "file_path": "./project",
                    }},
                    {{
                    "id": 2,
                    "description": "Run a command to Install the following dependencies: 'numpy', 'pandas', 'scikit-learn', 'matplotlib'",
                    "file_path": "null",
                    }},
                    {{
                    "id": 3,
                    "description": "Write code to create a function named 'parser' that takes an input named 'input' of type str, [perform a specific task on it], and returns a specific output",
                    "file_path": "./project/main.py",
                    }},
                    ...
                    {{
                    "id": N,
                    "description": "...",
                    }}
                ],
            }}

    The tasks will be executed by either of the three agents: command_executor, code_writer or code_refactor. They can't interact with programs. They can either run terminal commands or write code snippets. Their output is controlled by other functions to run the commands or save their output to code files. Make sure the tasks are compatible with the current agents. ALL tasks MUST start either with the following phrases: 'Run a command to...', 'Write code to...', 'Edit existing code to...' depending on the agent that will execute the task. RETURN JSON ONLY:"""

    return openai_call(prompt, temperature=0.8, max_tokens=2000)

def code_tasks_refactor_agent(objective: str, task_list_json):
    prompt = f"""You are an AGI tasks_refactor_agent responsible for adapting a task list generated by another agent to ensure the tasks are compatible with the current AGI agents. Your goal is to analyze the task list and make necessary modifications so that the tasks can be executed by the agents listed below

    YOU SHOULD OUTPUT THE MODIFIED TASK LIST IN THE SAME JSON FORMAT AS THE INITIAL TASK LIST. DO NOT CHANGE THE FORMAT OF THE JSON OUTPUT. DO NOT WRITE ANYTHING OTHER THAN THE MODIFIED TASK LIST IN THE JSON FORMAT.
    
    The current agents work as follows:
    - code_writer_agent: Writes code snippets or functions and saves them to the appropriate files. This agent can also append code to existing files if required.
    - code_refactor_agent: Responsible for editing current existing code/files.
    - command_executor_agent: Executes terminal commands for tasks such as creating directories, installing dependencies, etc.

    Here is the overall objective you need to refactor the tasks for: {objective}.
    Here is the JSON task list you need to refactor for compatibility with the current agents: {task_list_json}.

    To refactor the task list, follow these steps:
    1. Modify the task descriptions to make them compatible with the current agents, ensuring that the tasks are self-contained, clear, and executable by the agents without additional context. You don't need to mention the agents in the task descriptions, but the tasks should be compatible with the current agents.
    2. If necessary, add new tasks or remove irrelevant tasks to make the task list more suitable for the current agents.
    3. Keep the JSON structure of the task list intact, maintaining the "id", "description" and "file_path" fields for each task.
    4. Pay close attention to the objective and make sure the tasks implement all necessary pieces needed to make the program work.

    Always specify file paths to files. Make sure tasks are not duplicated. Never write code to create files. If needed, use commands to create files and folders.
    Return the updated JSON task list with the following format:

            {{
                "tasks": [
                    {{
                    "id": 1,
                    "description": "Run a commmand to create a folder named 'project' in the current directory",
                    "file_path": "./project",
                    }},
                    {{
                    "id": 2,
                    "description": "Write code to print 'Hello World!' with Python",
                    "file_path": "./project/main.py",
                    }},
                    {{
                    "id": 3,
                    "description": "Write code to create a function named 'parser' that takes an input named 'input' of type str, [perform a specific task on it], and returns a specific output",
                    "file_path": "./project/main.py",
                    }}
                    {{
                    "id": 3,
                    "description": "Run a command calling the script in ./project/main.py",
                    "file_path": "./project/main.py",
                    }}
                    ...
                ],
            }}

    IMPORTANT: All tasks should start either with the following phrases: 'Run a command to...', 'Write a code to...', 'Edit the code to...' depending on the agent that will execute the task:
            
    ALWAYS ENSURE ALL TASKS HAVE RELEVANT CONTEXT ABOUT THE CODE TO BE WRITTEN, INCLUDE DETAILS ON HOW TO CALL FUNCTIONS, CLASSES, IMPORTS, ETC. AGENTS HAVE NO VIEW OF OTHER TASKS, SO THEY NEED TO BE SELF-CONTAINED. RETURN THE JSON:"""

    return openai_call(prompt, temperature=0, max_tokens=2000)

def code_tasks_details_agent(objective: str, task_list_json):
    prompt = f"""You are an AGI agent responsible for improving a list of tasks in JSON format and adding ALL the necessary details to each task. These tasks will be executed individually by agents that have no idea about other tasks or what code exists in the codebase. It is FUNDAMENTAL that each task has enough details so that an individual isolated agent can execute. The metadata of the task is the only information the agents will have.

    Each task should contain the details necessary to execute it. For example, if it creates a function, it needs to contain the details about the arguments to be used in that function and this needs to be consistent across all tasks.

    Look at all tasks at once, and update the task description adding details to it for each task so that it can be executed by an agent without seeing the other tasks and to ensure consistency across all tasks. DETAILS ARE CRUCIAL. For example, if one task creates a class, it should have all the details about the class, including the arguments to be used in the constructor. If another task creates a function that uses the class, it should have the details about the class and the arguments to be used in the constructor.

    RETURN JSON OUTPUTS ONLY.
    
    Here is the overall objective you need to refactor the tasks for: {objective}.
    Here is the task list you need to improve: {task_list_json}
    
    RETURN THE SAME TASK LIST but with the description improved to contain the details you is adding for each task in the list. DO NOT MAKE OTHER MODIFICATIONS TO THE LIST. Your input should go in the 'description' field of each task.
    
    RETURN JSON ONLY:"""
    return openai_call(prompt, temperature=0.7, max_tokens=2000)

def code_tasks_context_agent(objective: str, task_list_json):
    prompt = f"""You are an AGI agent responsible for improving a list of tasks in JSON format and adding ALL the necessary context to it. These tasks will be executed individually by agents that have no idea about other tasks or what code exists in the codebase. It is FUNDAMENTAL that each task has enough context so that an individual isolated agent can execute. The metadata of the task is the only information the agents will have.

    Look at all tasks at once, and add the necessary context to each task so that it can be executed by an agent without seeing the other tasks. Remember, one agent can only see one task and has no idea about what happened in other tasks. CONTEXT IS CRUCIAL. For example, if one task creates one folder and the other tasks creates a file in that folder. The second tasks should contain the name of the folder that already exists and the information that it already exists.

    This is even more important for tasks that require importing functions, classes, etc. If a task needs to call a function or initialize a Class, it needs to have the detailed arguments, etc.

    Note that you should identify when imports need to happen and specify this in the context. Also, you should identify when functions/classes/etc already exist and specify this very clearly because the agents sometimes duplicate things not knowing.

    Always use imports with the file name. For example, 'from my_script import MyScript'. 
    
    RETURN JSON OUTPUTS ONLY.
    
    Here is the overall objective you need to refactor the tasks for: {objective}.
    Here is the task list you need to improve: {task_list_json}
    
    RETURN THE SAME TASK LIST but with a new field called 'isolated_context' for each task in the list. This field should be a string with the context you are adding. DO NOT MAKE OTHER MODIFICATIONS TO THE LIST.
    
    RETURN JSON ONLY:"""
    return openai_call(prompt, temperature=0.7, max_tokens=2000)

def task_assigner_recommendation_agent(objective: str, task: str):
    prompt = f"""You are an AGI agent responsible for providing recommendations on which agent should be used to handle a specific task. Analyze the provided major objective of the project and a single task from the JSON checklist generated by the previous agent, and suggest the most appropriate agent to work on the task.

    The overall objective is: {objective}
    The current task is: {task}
    
    The available agents are:
    1. code_writer_agent: Responsible for writing code based on the task description.
    2. code_refactor_agent: Responsible for editing existing code.
    3. command_executor_agent: Responsible for executing commands and handling file operations, such as creating, moving, or deleting files.

    When analyzing the task, consider the following tips:
    - Pay attention to keywords in the task description that indicate the type of action required, such as "write", "edit", "run", "create", "move", or "delete".
    - Keep the overall objective in mind, as it can help you understand the context of the task and guide your choice of agent.
    - If the task involves writing new code or adding new functionality, consider using the code_writer_agent.
    - If the task involves modifying or optimizing existing code, consider using the code_refactor_agent.
    - If the task involves file operations, command execution, or running a script, consider using the command_executor_agent.

    Based on the task and overall objective, suggest the most appropriate agent to work on the task."""
    return openai_call(prompt, temperature=0.5, max_tokens=2000)

def task_assigner_agent(objective: str, task: str, recommendation: str):
    prompt = f"""You are an AGI agent responsible for choosing the best agent to work on a given task. Your goal is to analyze the provided major objective of the project and a single task from the JSON checklist generated by the previous agent, and choose the best agent to work on the task.

    The overall objective is: {objective}
    The current task is: {task}

    Use this recommendation to guide you: {recommendation}
        
    The available agents are:
    1. code_writer_agent: Responsible for writing code based on the task description.
    2. code_refactor_agent: Responsible for editing existing code.
    2. command_executor_agent: Responsible for executing commands and handling file operations, such as creating, moving, or deleting files.

    Please consider the task description and the overall objective when choosing the most appropriate agent. Keep in mind that creating a file and writing code are different tasks. If the task involves creating a file, like "calculator.py" but does not mention writing any code inside it, the command_executor_agent should be used for this purpose. The code_writer_agent should only be used when the task requires writing or adding code to a file. The code_refactor_agent should only be used when the task requires modifying existing code.
    
    TLDR: To create files, use command_executor_agent, to write text/code to files, use code_writer_agent, to modify existing code, use code_refactor_agent.

    Choose the most appropriate agent to work on the task and return a JSON output with the following format: {{"agent": "agent_name"}}. ONLY return JSON output:"""
    return openai_call(prompt, temperature=0, max_tokens=2000)

def command_executor_agent(task: str, file_path: str):
    prompt = f"""You are an AGI agent responsible for executing a given command on the {os_version} OS. Your goal is to analyze the provided major objective of the project and a single task from the JSON checklist generated by the previous agent, and execute the command on the {os_version} OS. 

    The current task is: {task}
    File or folder name referenced in the task (relative file path): {file_path} 
    
    Based on the task, write the appropriate command to execute on the {os_version} OS. Make sure the command is relevant to the task and objective. For example, if the task is to create a new folder, the command should be 'mkdir new_folder_name'. Return the command as a JSON output with the following format: {{"command": "command_to_execute"}}. ONLY return JSON output:"""
    return openai_call(prompt, temperature=0, max_tokens=2000)

def code_writer_agent(task: str, isolated_context: str, context_code_chunks):
    prompt = f"""You are an AGI agent responsible for writing code to accomplish a given task. Your goal is to analyze the provided major objective of the project and a single task from the JSON checklist generated by the previous agent, and write the necessary code to complete the task.

    The current task is: {task}

    To help you make the code useful in this codebase, use this context as reference of the other pieces of the codebase that are relevant to your task. PAY ATTENTION TO THIS: {isolated_context}
    
    The following code chunks were found to be relevant to the task. You can use them as reference to write the code if they are useful. PAY CLOSE ATTENTION TO THIS: 
    {context_code_chunks}

    Note: Always use 'encoding='utf-8'' when opening files with open().
    
    Based on the task and objective, write the appropriate code to achieve the task. Make sure the code is relevant to the task and objective, and follows best practices. Return the code as a plain text output and NOTHING ELSE. Use identation and line breaks in the in the code. Make sure to only write the code and nothing else as your output will be saved directly to the file by other agent. IMPORTANT" If the task is asking you to write code to write files, this is a mistake! Interpret it and either do nothing or return  the plain code, not a code to write file, not a code to write code, etc."""
    return openai_call(prompt, temperature=0, max_tokens=2000)

def code_refactor_agent(task_description: str, existing_code_snippet: str, context_chunks, isolated_context: str):

    prompt = f"""You are an AGI agent responsible for refactoring code to accomplish a given task. Your goal is to analyze the provided major objective of the project, the task descriptionm and refactor the code accordingly.

    The current task description is: {task_description}
    To help you make the code useful in this codebase, use this context as reference of the other pieces of the codebase that are relevant to your task: {isolated_context}

    Here are some context chunks that might be relevant to the task:
    {context_chunks}
    
    Existing code you should refactor: 
    {existing_code_snippet}
    
    Based on the task description, objective, refactor the existing code to achieve the task. Make sure the refactored code is relevant to the task and objective, follows best practices, etc.

    Return a plain text code snippet with your refactored code. IMPORTANT: JUST RETURN CODE, YOUR OUTPUT WILL BE ADDED DIRECTLY TO THE FILE BY OTHER AGENT. BE MINDFUL OF THIS:"""

    return openai_call(prompt, temperature=0, max_tokens=2000)

def file_management_agent(objective: str, task: str, current_directory_files: str, file_path: str):
    prompt = f"""You are an AGI agent responsible for managing files in a software project. Your goal is to analyze the provided major objective of the project and a single task from the JSON checklist generated by the previous agent, and determine the appropriate file path and name for the generated code.

    The overall objective is: {objective}
    The current task is: {task}
    Specified file path (relative path from the current dir): {file_path}

    Make the file path adapted for the current directory files. The current directory files are: {current_directory_files}. Assume this file_path will be interpreted from the root path of the directory.

    Do not use '.' or './' in the file path.

    BE VERY SPECIFIC WITH THE FILES, AVOID FILE DUPLICATION, AVOID SPECIFYING THE SAME FILE NAME UNDER DIFFERENT FOLDERS, ETC.

    Based on the task, determine the file path and name for the generated code. Return the file path and name as a JSON output with the following format: {{"file_path": "file_path_and_name"}}. ONLY return JSON output:"""
    return openai_call(prompt, temperature=0, max_tokens=2000)

def code_relevance_agent(objective: str, task_description: str, code_chunk: str):
    prompt = f"""You are an AGI agent responsible for evaluating the relevance of a code chunk in relation to a given task. Your goal is to analyze the provided major objective of the project, the task description, and the code chunk, and assign a relevance score from 0 to 10, where 0 is completely irrelevant and 10 is highly relevant.

    The overall objective is: {objective}
    The current task description is: {task_description}
    The code chunk is as follows (line numbers included):
    {code_chunk}

    Based on the task description, objective, and code chunk, assign a relevance score between 0 and 10 (inclusive) for the code chunk. DO NOT OUTPUT ANYTHING OTHER THAN THE RELEVANCE SCORE AS A NUMBER."""

    relevance_score = openai_call(prompt, temperature=0.5, max_tokens=50)

    return json.dumps({"relevance_score": relevance_score.strip()})

def task_human_input_agent(task: str, human_feedback: str):
    prompt = f"""You are an AGI agent responsible for getting human input to improve the quality of tasks in a software project. Your goal is to analyze the provided task and adapt it based on the human's suggestions. The tasks should  start with either 'Run a command to...', 'Write code to...', or 'Edit existing code to...' depending on the agent that will execute the task.

    For context, this task will be executed by other AGI agents with the following characteristics:
    - code_writer_agent: Writes code snippets or functions and saves them to the appropriate files. This agent can also append code to existing files if required.
    - code_refactor_agent: Responsible for modifying and refactoring existing code to meet the requirements of the task.
    - command_executor_agent: Executes terminal commands for tasks such as creating directories, installing dependencies, etc.

    The current task is:
    {task}

    The human feedback is:
    {human_feedback}

    If the human feedback is empty, return the task as is. If the human feedback is saying to ignore the task, return the following string: <IGNORE_TASK>

    Note that your output will replace the existing task, so make sure that your output is a valid task that starts with one of the required phrases ('Run a command to...', 'Write code to...', 'Edit existing code to...').
    
    Please adjust the task based on the human feedback while ensuring it starts with one of the required phrases ('Run a command to...', 'Write code to...', 'Edit existing code to...'). Return the improved task as a plain text output and nothing else. Write only the new task."""

    return openai_call(prompt, temperature=0.3, max_tokens=200)

## END OF AGENTS ##

print_colored_text(f"****Objective****", color='green')
print_char_by_char(OBJECTIVE, 0.00001, 10)

# Create the tasks
print_colored_text("*****Working on tasks*****", "red")
print_colored_text(" - Creating initial tasks", "yellow")
task_agent_output = code_tasks_initializer_agent(OBJECTIVE)
print_colored_text(" - Reviewing and refactoring tasks to fit agents", "yellow")
task_agent_output = code_tasks_refactor_agent(OBJECTIVE, task_agent_output)
print_colored_text(" - Adding relevant technical details to the tasks", "yellow")
task_agent_output = code_tasks_details_agent(OBJECTIVE, task_agent_output)
print_colored_text(" - Adding necessary context to the tasks", "yellow")
task_agent_output = code_tasks_context_agent(OBJECTIVE, task_agent_output)
print()

print_colored_text("*****TASKS*****", "green")
print_char_by_char(task_agent_output, 0.00000001, 10)

# Task list
task_json = json.loads(task_agent_output)

embeddings = Embeddings(current_directory)

for task in task_json["tasks"]:
    task_description = task["description"]
    task_isolated_context = task["isolated_context"]

    print_colored_text("*****TASK*****", "yellow")
    print_char_by_char(task_description)
    print_colored_text("*****TASK CONTEXT*****", "yellow")
    print_char_by_char(task_isolated_context)

    # HUMAN FEEDBACK
    # Uncomment below to enable human feedback before each task. This can be used to improve the quality of the tasks,
    # skip tasks, etc. I believe it may be very relevant in future versions that may have more complex tasks and could
    # allow a ton of automation when working on large projects.
    #
    # Get user input as a feedback to the task_description
    # print_colored_text("*****TASK FEEDBACK*****", "yellow")
    # user_input = input("\n>:")
    # task_description = task_human_input_agent(task_description, user_input)
    # if task_description == "<IGNORE_TASK>":
    #     continue
    # print_colored_text("*****IMPROVED TASK*****", "green")
    # print_char_by_char(task_description)
    
    # Assign the task to an agent
    task_assigner_recommendation = task_assigner_recommendation_agent(OBJECTIVE, task_description)
    task_agent_output = task_assigner_agent(OBJECTIVE, task_description, task_assigner_recommendation)

    print_colored_text("*****ASSIGN*****", "yellow")
    print_char_by_char(task_agent_output)

    chosen_agent = json.loads(task_agent_output)["agent"]

    if chosen_agent == "command_executor_agent":
        command_executor_output = command_executor_agent(task_description, task["file_path"])
        print_colored_text("*****COMMAND*****", "green")
        print_char_by_char(command_executor_output)
        
        command_execution_output = execute_command_json(command_executor_output)
    else:
        # CODE AGENTS
        if chosen_agent == "code_writer_agent":
            # Compute embeddings for the codebase
            # This will recompute embeddings for all files in the 'playground' directory
            print_colored_text("*****RETRIEVING RELEVANT CODE CONTEXT*****", "yellow")
            embeddings.compute_repository_embeddings()
            relevant_chunks = embeddings.get_relevant_code_chunks(task_description, task_isolated_context)

            current_directory_files = execute_command_string("ls")
            file_management_output = file_management_agent(OBJECTIVE, task_description, current_directory_files, task["file_path"])
            print_colored_text("*****FILE MANAGEMENT*****", "yellow")
            print_char_by_char(file_management_output)
            file_path = json.loads(file_management_output)["file_path"]

            code_writer_output = code_writer_agent(task_description, task_isolated_context, relevant_chunks)
            
            print_colored_text("*****CODE*****", "green")
            print_char_by_char(code_writer_output)

            # Save the generated code to the file the agent selected
            save_code_to_file(code_writer_output, file_path)

        elif chosen_agent == "code_refactor_agent":
            # The code refactor agent works with multiple agents:
            # For each task, the file_management_agent is used to select the file to edit.Then, the 
            # code_relevance_agent is used to select the relevant code chunks from that filewith the 
            # goal of finding the code chunk that is most relevant to the task description. This is 
            # the code chunk that will be edited. Finally, the code_refactor_agent is used to edit 
            # the code chunk.

            current_directory_files = execute_command_string("ls")
            file_management_output = file_management_agent(OBJECTIVE, task_description, current_directory_files, task["file_path"])
            file_path = json.loads(file_management_output)["file_path"]

            print_colored_text("*****FILE MANAGEMENT*****", "yellow")
            print_char_by_char(file_management_output)
            
            # Split the code into chunks and get the relevance scores for each chunk
            code_chunks = split_code_into_chunks(file_path, 80)
            print_colored_text("*****ANALYZING EXISTING CODE*****", "yellow")
            relevance_scores = []
            for chunk in code_chunks:
                score = code_relevance_agent(OBJECTIVE, task_description, chunk["code"])
                relevance_scores.append(score)

            # Select the most relevant chunk
            selected_chunk = sorted(zip(relevance_scores, code_chunks), key=lambda x: x[0], reverse=True)[0][1]

            # Refactor the code
            modified_code_output = code_refactor_agent(task_description, selected_chunk, context_chunks=[selected_chunk], isolated_context=task_isolated_context)

            # Extract the start_line and end_line of the selected chunk. This will be used to replace the code in the original file
            start_line = selected_chunk["start_line"]
            end_line = selected_chunk["end_line"]

            # Count the number of lines in the modified_code_output
            modified_code_lines = modified_code_output.count("\n") + 1
            # Create a dictionary with the necessary information for the refactor_code function
            modified_code_info = {
                "start_line": start_line,
                "end_line": start_line + modified_code_lines - 1,
                "modified_code": modified_code_output
            }
            print_colored_text("*****REFACTORED CODE*****", "green")
            print_char_by_char(modified_code_output)

            # Save the refactored code to the file
            refactor_code([modified_code_info], file_path)



File: babycoder/embeddings.py

import os
import csv 
import shutil
import openai
import pandas as pd
import numpy as np
from transformers import GPT2TokenizerFast
from dotenv import load_dotenv
import time

# Heavily derived from OpenAi's cookbook example

load_dotenv()

# the dir is the ./playground directory
REPOSITORY_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "playground")

class Embeddings:
    def __init__(self, workspace_path: str):
        self.workspace_path = workspace_path
        openai.api_key = os.getenv("OPENAI_API_KEY", "")

        self.DOC_EMBEDDINGS_MODEL = f"text-embedding-ada-002"
        self.QUERY_EMBEDDINGS_MODEL = f"text-embedding-ada-002"

        self.SEPARATOR = "\n* "

        self.tokenizer = GPT2TokenizerFast.from_pretrained("gpt2")
        self.separator_len = len(self.tokenizer.tokenize(self.SEPARATOR))

    def compute_repository_embeddings(self):
        try:
            playground_data_path = os.path.join(self.workspace_path, 'playground_data')

            # Delete the contents of the playground_data directory but not the directory itself
            # This is to ensure that we don't have any old data lying around
            for filename in os.listdir(playground_data_path):
                file_path = os.path.join(playground_data_path, filename)

                try:
                    if os.path.isfile(file_path) or os.path.islink(file_path):
                        os.unlink(file_path)
                    elif os.path.isdir(file_path):
                        shutil.rmtree(file_path)
                except Exception as e:
                    print(f"Failed to delete {file_path}. Reason: {str(e)}")
        except Exception as e:
            print(f"Error: {str(e)}")

        # extract and save info to csv
        info = self.extract_info(REPOSITORY_PATH)
        self.save_info_to_csv(info)

        df = pd.read_csv(os.path.join(self.workspace_path, 'playground_data\\repository_info.csv'))
        df = df.set_index(["filePath", "lineCoverage"])
        self.df = df
        context_embeddings = self.compute_doc_embeddings(df)
        self.save_doc_embeddings_to_csv(context_embeddings, df, os.path.join(self.workspace_path, 'playground_data\\doc_embeddings.csv'))

        try:
            self.document_embeddings = self.load_embeddings(os.path.join(self.workspace_path, 'playground_data\\doc_embeddings.csv'))
        except:
            pass

    # Extract information from files in the repository in chunks
    # Return a list of [filePath, lineCoverage, chunkContent]
    def extract_info(self, REPOSITORY_PATH):
        # Initialize an empty list to store the information
        info = []
        
        LINES_PER_CHUNK = 60

        # Iterate through the files in the repository
        for root, dirs, files in os.walk(REPOSITORY_PATH):
            for file in files:
                file_path = os.path.join(root, file)

                # Read the contents of the file
                with open(file_path, "r", encoding="utf-8") as f:
                    try:
                        contents = f.read()
                    except:
                        continue
                
                # Split the contents into lines
                lines = contents.split("\n")
                # Ignore empty lines
                lines = [line for line in lines if line.strip()]
                # Split the lines into chunks of LINES_PER_CHUNK lines
                chunks = [
                        lines[i:i+LINES_PER_CHUNK]
                        for i in range(0, len(lines), LINES_PER_CHUNK)
                    ]
                # Iterate through the chunks
                for i, chunk in enumerate(chunks):
                    # Join the lines in the chunk back into a single string
                    chunk = "\n".join(chunk)
                    # Get the first and last line numbers
                    first_line = i * LINES_PER_CHUNK + 1
                    last_line = first_line + len(chunk.split("\n")) - 1
                    line_coverage = (first_line, last_line)
                    # Add the file path, line coverage, and content to the list
                    info.append((os.path.join(root, file), line_coverage, chunk))
            
        # Return the list of information
        return info

    def save_info_to_csv(self, info):
        # Open a CSV file for writing
        os.makedirs(os.path.join(self.workspace_path, "playground_data"), exist_ok=True)
        with open(os.path.join(self.workspace_path, 'playground_data\\repository_info.csv'), "w", newline="") as csvfile:
            # Create a CSV writer
            writer = csv.writer(csvfile)
            # Write the header row
            writer.writerow(["filePath", "lineCoverage", "content"])
            # Iterate through the info
            for file_path, line_coverage, content in info:
                # Write a row for each chunk of data
                writer.writerow([file_path, line_coverage, content])

    def get_relevant_code_chunks(self, task_description: str, task_context: str):
        query = task_description + "\n" + task_context
        most_relevant_document_sections = self.order_document_sections_by_query_similarity(query, self.document_embeddings)
        selected_chunks = []
        for _, section_index in most_relevant_document_sections:
            try:
                document_section = self.df.loc[section_index]
                selected_chunks.append(self.SEPARATOR + document_section['content'].replace("\n", " "))
                if len(selected_chunks) >= 2:
                    break
            except:
                pass

        return selected_chunks

    def get_embedding(self, text: str, model: str) -> list[float]:
        result = openai.Embedding.create(
        model=model,
        input=text
        )
        return result["data"][0]["embedding"]

    def get_doc_embedding(self, text: str) -> list[float]:
        return self.get_embedding(text, self.DOC_EMBEDDINGS_MODEL)

    def get_query_embedding(self, text: str) -> list[float]:
        return self.get_embedding(text, self.QUERY_EMBEDDINGS_MODEL)

    def compute_doc_embeddings(self, df: pd.DataFrame) -> dict[tuple[str, str], list[float]]:
        """
        Create an embedding for each row in the dataframe using the OpenAI Embeddings API.

        Return a dictionary that maps between each embedding vector and the index of the row that it corresponds to.
        """
        embeddings = {}
        for idx, r in df.iterrows():
            # Wait one second before making the next call to the OpenAI Embeddings API
            # print("Waiting one second before embedding next row\n")
            time.sleep(1)
            embeddings[idx] = self.get_doc_embedding(r.content.replace("\n", " "))
        return embeddings

    def save_doc_embeddings_to_csv(self, doc_embeddings: dict, df: pd.DataFrame, csv_filepath: str):
        # Get the dimensionality of the embedding vectors from the first element in the doc_embeddings dictionary
        if len(doc_embeddings) == 0:
            return

        EMBEDDING_DIM = len(list(doc_embeddings.values())[0])

        # Create a new dataframe with the filePath, lineCoverage, and embedding vector columns
        embeddings_df = pd.DataFrame(columns=["filePath", "lineCoverage"] + [f"{i}" for i in range(EMBEDDING_DIM)])

        # Iterate over the rows in the original dataframe
        for idx, _ in df.iterrows():
            # Get the embedding vector for the current row
            embedding = doc_embeddings[idx]
            # Create a new row in the embeddings dataframe with the filePath, lineCoverage, and embedding vector values
            row = [idx[0], idx[1]] + embedding
            embeddings_df.loc[len(embeddings_df)] = row

        # Save the embeddings dataframe to a CSV file
        embeddings_df.to_csv(csv_filepath, index=False)

    def vector_similarity(self, x: list[float], y: list[float]) -> float:
        return np.dot(np.array(x), np.array(y))

    def order_document_sections_by_query_similarity(self, query: str, contexts: dict[(str, str), np.array]) -> list[(float, (str, str))]:
        """
        Find the query embedding for the supplied query, and compare it against all of the pre-calculated document embeddings
        to find the most relevant sections. 
        
        Return the list of document sections, sorted by relevance in descending order.
        """
        query_embedding = self.get_query_embedding(query)
        
        document_similarities = sorted([
            (self.vector_similarity(query_embedding, doc_embedding), doc_index) for doc_index, doc_embedding in contexts.items()
        ], reverse=True)
        
        return document_similarities
    
    def load_embeddings(self, fname: str) -> dict[tuple[str, str], list[float]]:       
        df = pd.read_csv(fname, header=0)
        max_dim = max([int(c) for c in df.columns if c != "filePath" and c != "lineCoverage"])
        return {
            (r.filePath, r.lineCoverage): [r[str(i)] for i in range(max_dim + 1)] for _, r in df.iterrows()
        }


# Translations:

[<img title="عربي" alt="عربي" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/sa.svg" width="30">](docs/README-ar.md)
[<img title="Français" alt="Français" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/fr.svg" width="30">](docs/README-fr.md)
[<img title="Polski" alt="Polski" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/pl.svg" width="30">](docs/README-pl.md)
[<img title="Portuguese" alt="Portuguese" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/br.svg" width="30">](docs/README-pt-br.md)
[<img title="Romanian" alt="Romanian" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/ro.svg" width="30">](docs/README-ro.md)
[<img title="Russian" alt="Russian" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/ru.svg" width="30">](docs/README-ru.md)
[<img title="Slovenian" alt="Slovenian" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/si.svg" width="30">](docs/README-si.md)
[<img title="Spanish" alt="Spanish" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/es.svg" width="30">](docs/README-es.md)
[<img title="Turkish" alt="Turkish" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/tr.svg" width="30">](docs/README-tr.md)
[<img title="Ukrainian" alt="Ukrainian" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/ua.svg" width="30">](docs/README-ua.md)
[<img title="简体中文" alt="Simplified Chinese" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/cn.svg" width="30">](docs/README-cn.md)
[<img title="繁體中文 (Traditional Chinese)" alt="繁體中文 (Traditional Chinese)" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/tw.svg" width="30">](docs/README-zh-tw.md)
[<img title="日本語" alt="日本語" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/jp.svg" width="30">](docs/README-ja.md)
[<img title="한국어" alt="한국어" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/kr.svg" width="30">](docs/README-ko.md)
[<img title="Magyar" alt="Magyar" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/hu.svg" width="30">](docs/README-hu.md)
[<img title="فارسی" alt="فارسی" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/ir.svg" width="30">](docs/README-fa.md)
[<img title="German" alt="German" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/de.svg" width="30">](docs/README-de.md)
[<img title="Indian" alt="Indian" src="https://cdn.staticaly.com/gh/hjnilsson/country-flags/master/svg/in.svg" width="30">](docs/README-in.md)
[<img title="Vietnamese" alt="Vietnamese" src="https://raw.githubusercontent.com/hampusborgos/country-flags/main/svg/vn.svg" width="30">](docs/README-vn.md)


# Objective

This Python script is an example of an AI-powered task management system. The system uses OpenAI and vector databases such as Chroma or Weaviate to create, prioritize, and execute tasks. The main idea behind this system is that it creates tasks based on the result of previous tasks and a predefined objective. The script then uses OpenAI's natural language processing (NLP) capabilities to create new tasks based on the objective, and Chroma/Weaviate to store and retrieve task results for context. This is a pared-down version of the original [Task-Driven Autonomous Agent](https://twitter.com/yoheinakajima/status/1640934493489070080?s=20) (Mar 28, 2023).

This README will cover the following:

- [How the script works](#how-it-works)

- [How to use the script](#how-to-use)

- [Supported Models](#supported-models)

- [Warning about running the script continuously](#continuous-script-warning)

# How It Works<a name="how-it-works"></a>

The script works by running an infinite loop that does the following steps:

1. Pulls the first task from the task list.
2. Sends the task to the execution agent, which uses OpenAI's API to complete the task based on the context.
3. Enriches the result and stores it in [Chroma](https://docs.trychroma.com)/[Weaviate](https://weaviate.io/).
4. Creates new tasks and reprioritizes the task list based on the objective and the result of the previous task.
   </br>

![image](https://user-images.githubusercontent.com/21254008/235015461-543a897f-70cc-4b63-941a-2ae3c9172b11.png)

The `execution_agent()` function is where the OpenAI API is used. It takes two parameters: the objective and the task. It then sends a prompt to OpenAI's API, which returns the result of the task. The prompt consists of a description of the AI system's task, the objective, and the task itself. The result is then returned as a string.

The `task_creation_agent()` function is where OpenAI's API is used to create new tasks based on the objective and the result of the previous task. The function takes four parameters: the objective, the result of the previous task, the task description, and the current task list. It then sends a prompt to OpenAI's API, which returns a list of new tasks as strings. The function then returns the new tasks as a list of dictionaries, where each dictionary contains the name of the task.

The `prioritization_agent()` function is where OpenAI's API is used to reprioritize the task list. The function takes one parameter, the ID of the current task. It sends a prompt to OpenAI's API, which returns the reprioritized task list as a numbered list.

Finally, the script uses Chroma/Weaviate to store and retrieve task results for context. The script creates a Chroma/Weaviate collection based on the table name specified in the TABLE_NAME variable. Chroma/Weaviate is then used to store the results of the task in the collection, along with the task name and any additional metadata.

# How to Use<a name="how-to-use"></a>

To use the script, you will need to follow these steps:

1. Clone the repository via `git clone https://github.com/yoheinakajima/babyagi.git` and `cd` into the cloned repository.
2. Install the required packages: `pip install -r requirements.txt`
3. Copy the .env.example file to .env: `cp .env.example .env`. This is where you will set the following variables.
4. Set your OpenAI API key in the OPENAI_API_KEY and OPENAI_API_MODEL variables. In order to use with Weaviate you will also need to setup additional variables detailed [here](docs/weaviate.md).
5. Set the name of the table where the task results will be stored in the TABLE_NAME variable.
6. (Optional) Set the name of the BabyAGI instance in the BABY_NAME variable.
7. (Optional) Set the objective of the task management system in the OBJECTIVE variable.
8. (Optional) Set the first task of the system in the INITIAL_TASK variable.
9. Run the script: `python babyagi.py`

All optional values above can also be specified on the command line.

# Running inside a docker container

As a prerequisite, you will need docker and docker-compose installed. Docker desktop is the simplest option https://www.docker.com/products/docker-desktop/

To run the system inside a docker container, setup your .env file as per steps above and then run the following:

```
docker-compose up
```

# Supported Models<a name="supported-models"></a>

This script works with all OpenAI models, as well as Llama and its variations through Llama.cpp. Default model is **gpt-3.5-turbo**. To use a different model, specify it through LLM_MODEL or use the command line.

## Llama

Llama integration requires llama-cpp package. You will also need the Llama model weights. 

- **Under no circumstances share IPFS, magnet links, or any other links to model downloads anywhere in this repository, including in issues, discussions or pull requests. They will be immediately deleted.**

Once you have them, set LLAMA_MODEL_PATH to the path of the specific model to use. For convenience, you can link `models` in BabyAGI repo to the folder where you have the Llama model weights. Then run the script with `LLM_MODEL=llama` or `-l` argument.

# Warning<a name="continuous-script-warning"></a>

This script is designed to be run continuously as part of a task management system. Running this script continuously can result in high API usage, so please use it responsibly. Additionally, the script requires the OpenAI API to be set up correctly, so make sure you have set up the API before running the script.

# Contribution

Needless to say, BabyAGI is still in its infancy and thus we are still determining its direction and the steps to get there. Currently, a key design goal for BabyAGI is to be _simple_ such that it's easy to understand and build upon. To maintain this simplicity, we kindly request that you adhere to the following guidelines when submitting PRs:

- Focus on small, modular modifications rather than extensive refactoring.
- When introducing new features, provide a detailed description of the specific use case you are addressing.

A note from @yoheinakajima (Apr 5th, 2023):

> I know there are a growing number of PRs, appreciate your patience - as I am both new to GitHub/OpenSource, and did not plan my time availability accordingly this week. Re:direction, I've been torn on keeping it simple vs expanding - currently leaning towards keeping a core Baby AGI simple, and using this as a platform to support and promote different approaches to expanding this (eg. BabyAGIxLangchain as one direction). I believe there are various opinionated approaches that are worth exploring, and I see value in having a central place to compare and discuss. More updates coming shortly.

I am new to GitHub and open source, so please be patient as I learn to manage this project properly. I run a VC firm by day, so I will generally be checking PRs and issues at night after I get my kids down - which may not be every night. Open to the idea of bringing in support, will be updating this section soon (expectations, visions, etc). Talking to lots of people and learning - hang tight for updates!

# BabyAGI Activity Report

To help the BabyAGI community stay informed about the project's progress, Blueprint AI has developed a Github activity summarizer for BabyAGI. This concise report displays a summary of all contributions to the BabyAGI repository over the past 7 days (continuously updated), making it easy for you to keep track of the latest developments.

To view the BabyAGI 7-day activity report, go here: [https://app.blueprint.ai/github/yoheinakajima/babyagi](https://app.blueprint.ai/github/yoheinakajima/babyagi)

[<img width="293" alt="image" src="https://user-images.githubusercontent.com/334530/235789974-f49d3cbe-f4df-4c3d-89e9-bfb60eea6308.png">](https://app.blueprint.ai/github/yoheinakajima/babyagi)


# Inspired projects

In the short time since it was release, BabyAGI inspired many projects. You can see them all [here](docs/inspired-projects.md).

# Backstory

BabyAGI is a pared-down version of the original [Task-Driven Autonomous Agent](https://twitter.com/yoheinakajima/status/1640934493489070080?s=20) (Mar 28, 2023) shared on Twitter. This version is down to 140 lines: 13 comments, 22 blanks, and 105 code. The name of the repo came up in the reaction to the original autonomous agent - the author does not mean to imply that this is AGI.

Made with love by [@yoheinakajima](https://twitter.com/yoheinakajima), who happens to be a VC (would love to see what you're building!)
