Stateful Agent

import logging
import os
import openai
import itertools

from typing import List

from actionweaver.llms.openai.chat import OpenAIChatCompletion
from actionweaver.llms.openai.tokens import TokenUsageTracker
from actionweaver import action, SelectOne, RequireNext

from actionweaver.mixins.examples import LangChainTools, Folium, OpenAIAPI
openai.api_key = os.getenv("OPENAI_API_KEY")

logging.basicConfig(
    filename='agent.log',
    filemode='a',
    format='%(asctime)s.%(msecs)04d %(levelname)s {%(module)s} [%(funcName)s] %(message)s',
    level=logging.INFO,
    datefmt='%Y-%m-%d %H:%M:%S'
)

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

class AgentV0:
    def __init__(self, logger):
        self.logger = logger
        self.token_tracker = TokenUsageTracker(budget=None, logger=logger)
        self.llm = OpenAIChatCompletion("gpt-3.5-turbo", token_usage_tracker = self.token_tracker, logger=logger)

        self.messages = [{"role": "system", "content": "You are a resourceful assistant."}]
        self.times = []

    def __call__(self, text):
        self.messages += [{"role": "user", "content":text}]
        response = self.llm.create(messages=self.messages, actions = [self.get_current_time, self.sleep], stream=True)

        return response

    @action(name="GetCurrentTime")
    def get_current_time(self) -> str:
        """
        Use this for getting the current time in the specified time zone.

        :return: A string representing the current time in the specified time zone.
        """
        import datetime
        current_time = datetime.datetime.now()

        self.times += [str(current_time)]

        return f"The current time is {current_time}"


    @action(name="Sleep")
    def sleep(self, seconds: int) -> str:
        """
        Introduces a sleep delay of the specified seconds and returns a message.

        Args:
            seconds (int): The duration to sleep in seconds.

        Returns:
            str: A message indicating the completion of the sleep.
        """
        import time
        time.sleep(seconds)
        return f"Now I wake up after sleep {seconds} seconds."


    @action(name="Ask")
    def ask(self, question: str) -> str:
        """
        Invoke this if you want to ask question, or there is anything need clarification.

        Args:
            question (str): The question to ask the user.
        """
        ans = input(f"Question: {question}\n")
        return ans

def print_output(output):
    if type(output) == itertools._tee:
        for chunk in output:
            content = chunk.choices[0].delta.content
            if content is not None:
                print(content, end='')
    else:
        print (output)


agent = AgentV0(logger)

print_output(agent("what time is it"))

The current time is 11:03 AM.

class AgentV1(AgentV0, LangChainTools, Folium, OpenAIAPI):
    def __call__(self, text):
        self.messages += [{"role": "user", "content":text}]
        response = self.llm.create(messages=self.messages, actions = [self.search, self.show_map, self.get_current_time, self.sleep], stream=True)

        return response

agent = AgentV1(logger)

print_output(agent("search what is Langchain"))

Langchain is a framework for developing applications powered by language models. It enables the creation of context-aware, reasoning applications using language models. Langchain provides a flexible set of abstractions and an extensive toolkit for developers to build applications that leverage the power of language models. It was launched as an open-source project in October 2022 by Harrison Chase, while working at machine learning startup Robust Intelligence. Langchain offers both Python and JavaScript versions for developers to use.

class FileUtility(AgentV0):
    @action(name="FileHandler", orch_expr=SelectOne(['FileHandler', 'ListFiles', 'ReadFile']))
    def handle_file(self, instruction: str) -> str:
        """
        Handles user instructions related to file operations. Put every context in the instruction only!

        Args:
            instruction (str): The user's instruction about file handling.

        Returns:
            str: The response to the user's question.
        """
        return instruction


    @action(name="ListFiles", scope="file")
    def list_all_files_in_repo(self, repo_path: str ='.') -> List:
        """
        Lists all the files in the given repository.

        :param repo_path: Path to the repository. Defaults to the current directory.
        :return: List of file paths.
        """

        logger.info(f"list_all_files_in_repo: {repo_path}")

        file_list = []
        for root, _, files in os.walk(repo_path):
            for file in files:
                file_list.append(os.path.join(root, file))
            break
        return file_list

    @action(name="ReadFile", scope="file")
    def read_from_file(self, file_path: str) -> str:
        """
        Reads the content of a file and returns it as a string.

        :param file_path: The path to the file that needs to be read.
        :return: A string containing the content of the file.
        """
        logger.info(f"read_from_file: {file_path}")

        with open(file_path, 'r') as file:
            content = file.read()
        return f"The file content: \n{content}"

    @action(name="WriteFile", scope="file")
    def write_to_file(self, file_path: str, content: str) -> str:
        """
        Writes the given content to a file.

        :param file_path: The path to the file where the content should be written.
        :param content: The content to be written to the file.
        """
        try:
            with open(file_path, 'a') as file:
                file.write(content)
            return "Content successfully appended to the file."
        except Exception as e:
            return f"An error occurred while appending to the file: {str(e)}"



class AgentV2(FileUtility):
    def __call__(self, text):
        self.messages += [{"role": "user", "content":text}]
        response = self.llm.create(messages=self.messages, actions = [self.search, self.show_map, self.get_current_time, self.sleep], stream=True)

        return response