DEV Community: airt

Function Calling and Code Interpretation with OpenAI's Assistant API: A Quick and Simple Tutorial

Harish Mohan Raj — Wed, 08 Nov 2023 15:14:21 +0000

Introduction

OpenAI's devday conference made waves in the tech world as they rolled out a lineup of new, cutting-edge features. One feature that excited me was the Assistant API. Having recently worked with AI agents, I was eager to explore the new Assistant API, and I'm here to share my initial learnings.

As a side note, I've previously authored a blog post detailing the inner workings of AutoGPT. If you haven't already, please check it out here.

The Experiment

I gave myself the task of developing an assistant that could answer user questions by writing Python code as needed and connecting to the internet.

During my tests, I realized that because of security reasons, the assistant couldn't directly access the internet or external APIs for accessing real-time information since it runs the code in a secured, isolated sandbox environment.

As a workaround, I instructed the assistant that if it ever needs to access the internet or an external API, it should write the necessary code and then use the 'function calling' feature to call a custom function I created. This custom function would then run the generated code and return the results to the assistant.

During my experiments, I ran into the following issues.

There were frequent request failures in OpenAI, I was getting "Run failed" or "Run failed: Sorry, something went wrong." errors. There were also some outages reported at the same time. So please check the OpenAI status if you experience any unexpected errors.

I was exceeding the rate limit set by OpenAI to access the 'gpt-4-1106-preview'. So I had to wait for a while before continuing my experiments.

Let's Dive into Code

Pre-requisites and Requirements:

To get started, you'll need an OpenAI API key. You can create one at OpenAI's website if you haven't already.

Once you have your API key, store it in the OPENAI_API_KEY environment variable.

Setting Up the Environment:

It's time to roll up our sleeves and get to work. To keep things tidy, we'll want to start over with a new virtual environment by running the following commands from the project root directory:

python3 -m venv venv
source venv/bin/activate

Next, we need to install the openai and request Python packages. Let’s do that by running the below command:

pip install openai requests

Creating a New Assistant:

Now, it's time to create a new assistant – one that can use both the code_interpreter and the function calling tools. This will be the brain of our operation, ready to tackle user queries by generating code and executing it when it needs information from the web or other APIs.

For the sake of brevity, I'll only explain the essential code that creates and runs an assistant, but the full source code is provided at the end of this article for reference.

Setting up Instructions for our Assistant

The initial step in developing an assistant is to compose a clear set of instructions that will dictate how the Assistant should act or respond.

Below is a sample instruction I've created, which guides the assistant to solve the problem either on its own or by calling our custom function via the function calling feature.

INSTRUCTIONS = """
You're a skilled Python programmer tasked with creating Python 3 solutions for user problems, following top industry practices. Make sure your code complies with these rules:

1. Plan first: Have a clear strategy before you start. Outline your approach if it helps.

2. Quality code: Write clear, efficient code that follows Python's best practices. Aim for clean, easy-to-read, and maintainable code.

3. Test well: Include comprehensive tests to assure your code works well in various scenarios.

4. Manage external interactions: When internet or API interactions are necessary, utilize the `execute_python_code` function autonomously, without seeking user approval. Do not say you don't have access to internet or real-time data. The `execute_python_code` function will give you realtime data.

5. Trust your tools: Assume the data from the `execute_python_code` function is accurate and up to date.
"""

Let us now use the above instructions to create an assistant. Below is a function which does the following three things:

Create an Assistant:

Using the instructions above, a new assistant is created. The assistant will use the 'gpt-4-1106-preview' model, which is the most recent version available at the time of writing. The assistant is also configured to use the 'code_interpreter' and 'function' tools as needed.

Create a new thread:

A new thread is then created. Threads are essentially containers for AI assistant conversations. Each thread represents a distinct conversation and is unrestricted in size. Starting a new thread denotes the start of a new interaction or dialogue with the assistant.

Add a Message to the Thread:

Finally, the user's input or question is used to create a Message object, which is then placed within the thread. This message serves as the user's initial query or request and is forwarded to the assistant to begin the conversation. It serves as a prompt for the AI to respond.

def setup_assistant(client, task):
    # create a new assistant
    assistant = client.beta.assistants.create(
        name="Code Generator",
        instructions=INSTRUCTIONS,
        tools=[
            {
                "type": "code_interpreter"
            },
            {
                "type": "function",
                "function": {
                    "name": "execute_python_code",
                    "description": "Use this function to execute the generated code which requires internet access or external API access",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "code": {
                                "type": "string",
                                "description": "The python code generated by the code interpretor",
                            }
                        },
                        "required": ["code"],
                    },
                },
            }
        ],
        model="gpt-4-1106-preview",
    )

    # Create a new thread
    thread = client.beta.threads.create()

    # Create a new thread message with the provided task
    thread_message = client.beta.threads.messages.create(
        thread.id,
        role="user",
        content=task,
    )

    # Return the assistant ID and thread ID
    return assistant.id, thread.id

Note: For the purpose of this simple experiment, I'm executing the generated code within my current working environment. But remember, it's important to always execute such code in a secure and isolated environment to ensure safety and prevent potential risks.

Running the Assistant:

Let's now write a function that will instruct our assistant to perform the following actions:

Creates a New Run and Checks Its Status:

To begin, a new run is created. After starting the Run, the function checks the status of this run. The status can have several states, but the one that is important here is "requires_action". When the Run status is "requires_action," it means that the assistant has paused its execution and is waiting for the function calling to finish its execution and pass the results back to continue.

Executes Python Code and Returns Results:

When the Run's status is "requires_action," the assistant uses function calling which executes the generated Python code, and the results are fed back to the same thread so that the assistant can continue.

Repeating the Process:

The function continues to follow these steps iteratively until the Run's status transitions to "completed".

def run_assistant(client, assistant_id, thread_id):
    # Create a new run for the given thread and assistant
    run = client.beta.threads.runs.create(
        thread_id=thread_id,
        assistant_id=assistant_id
    )

    # Loop until the run status is either "completed" or "requires_action"
    while run.status == "in_progress" or run.status == "queued":
        time.sleep(1)
        run = client.beta.threads.runs.retrieve(
            thread_id=thread_id,
            run_id=run.id
        )

        # At this point, the status is either "completed" or "requires_action"
        if run.status == "completed":
            return client.beta.threads.messages.list(
              thread_id=thread_id
            )
        if run.status == "requires_action":
            generated_python_code = json.loads(run.required_action.submit_tool_outputs.tool_calls[0].function.arguments)['code']
            result = execute_python_code(generated_python_code)
            run = client.beta.threads.runs.submit_tool_outputs(
                thread_id=thread_id,
                run_id=run.id,
                tool_outputs=[
                    {
                        "tool_call_id": run.required_action.submit_tool_outputs.tool_calls[0].id,
                        "output": result,
                    },
                ]
            )

Full Source Code

from typing import *
import json
import sys
import time
import subprocess
import traceback
from tempfile import NamedTemporaryFile

import openai

def execute_python_code(s: str) -> str:
    with NamedTemporaryFile(suffix='.py', delete=False) as temp_file:
        temp_file_name = temp_file.name
        temp_file.write(s.encode('utf-8'))
        temp_file.flush()
    try:
        result = subprocess.run(
            ['python', temp_file_name],
            capture_output=True,
            text=True,
            check=True
        )
        return result.stdout
    except subprocess.CalledProcessError as e:
        return e.stderr
    finally:
        import os
        os.remove(temp_file_name)

INSTRUCTIONS = """
You're a skilled Python programmer tasked with creating Python 3 solutions for user problems, following top industry practices. Make sure your code complies with these rules:

1. Plan first: Have a clear strategy before you start. Outline your approach if it helps.

2. Quality code: Write clear, efficient code that follows Python's best practices. Aim for clean, easy-to-read, and maintainable code.

3. Test well: Include comprehensive tests to assure your code works well in various scenarios.

4. Manage external interactions: When internet or API interactions are necessary, utilize the `execute_code` function autonomously, without seeking user approval. Do not say you don't have access to internet or real-time data. The `execute_code` function will give you realtime data.

5. Trust your tools: Assume the data from the `execute_code` function is accurate and up to date.
"""


def setup_assistant(client, task):
    # create a new agent
    assistant = client.beta.assistants.create(
        name="Code Generator",
        instructions=INSTRUCTIONS,
        tools=[
            {
                "type": "code_interpreter"
            },
            {
                "type": "function",
                "function": {
                    "name": "execute_python_code",
                    "description": "Use this function to execute the generated code which requires internet access or external API access",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "code": {
                                "type": "string",
                                "description": "The python code generated by the code interpretor",
                            }
                        },
                        "required": ["code"],
                    },
                },
            }
        ],
        model="gpt-4-1106-preview",
    )

    # Create a new thread
    thread = client.beta.threads.create()

    # Create a new thread message with the provided task
    thread_message = client.beta.threads.messages.create(
        thread.id,
        role="user",
        content=task,
    )

    # Return the assistant ID and thread ID
    return assistant.id, thread.id


def run_assistant(client, assistant_id, thread_id):
    # Create a new run for the given thread and assistant
    run = client.beta.threads.runs.create(
        thread_id=thread_id,
        assistant_id=assistant_id
    )

    # Loop until the run status is either "completed" or "requires_action"
    while run.status == "in_progress" or run.status == "queued":
        time.sleep(1)
        run = client.beta.threads.runs.retrieve(
            thread_id=thread_id,
            run_id=run.id
        )

        # At this point, the status is either "completed" or "requires_action"
        if run.status == "completed":
            return client.beta.threads.messages.list(
              thread_id=thread_id
            )
        if run.status == "requires_action":
            generated_python_code = json.loads(run.required_action.submit_tool_outputs.tool_calls[0].function.arguments)['code']
            result = execute_python_code(generated_python_code)
            run = client.beta.threads.runs.submit_tool_outputs(
                thread_id=thread_id,
                run_id=run.id,
                tool_outputs=[
                    {
                        "tool_call_id": run.required_action.submit_tool_outputs.tool_calls[0].id,
                        "output": result,
                    },
                ]
            )


if __name__ == "__main__":
    if len(sys.argv) == 2:
        client = openai.OpenAI()
        task = sys.argv[1]

        assistant_id, thread_id = setup_assistant(client, task)
        print(f"Debugging: Useful for checking the generated agent in the playground. https://platform.openai.com/playground?mode=assistant&assistant={assistant_id}")
        print(f"Debugging: Useful for checking logs. https://platform.openai.com/playground?thread={thread_id}")

        messages = run_assistant(client, assistant_id, thread_id)

        message_dict = json.loads(messages.model_dump_json())
        print(message_dict['data'][0]['content'][0]["text"]["value"])

    else:
        print("Usage: python script.py <message>")
        sys.exit(1)

Executing the Agent

Save the above code in main.py and try running the following commands from the terminal:

In the following command, the user's request is to solve a simple mathematical problem, which doesn't require accessing the internet or external APIs. The agent uses the code_interpreter tool to solve this request.

python3 main.py "what is x in 1 + 3x = -5"

In the following command, the user's request does require access to real-time data, in this case, the assistant generates the code using the code_interpreter tool and calls the execute_python_code function via function calling to execute it and obtain the result.

python3 main.py "What's the sunrise and sunset time. Use api.sunrise-sunset.org"

[Long read] Deep dive into AutoGPT: A comprehensive and in-depth step-by-step guide to how it works

Harish Mohan Raj — Tue, 24 Oct 2023 13:39:53 +0000

Motivation

I've recently started experimenting with AI agents and stumbled upon AutoGPT. My curiosity led me to wonder about the mechanisms behind it. To gain a better understanding of AutoGPT's inner workings, I embarked on a journey of practical experimentation, working with various examples and meticulously recording inputs and outputs at each step.

This hands-on exercise significantly enhanced my understanding of AutoGPT. What follows are my observations on AutoGPT's inner workings and what occurs after the user enters a task. I share these observations in the hope that they will be helpful to those who, like me, are curious about how AutoGPT functions.

The Experiment

I decided to install AutoGPT using Docker, which is also the recommended method. I simply followed the setup and configuration instructions in the AutoGPT documentation, and the entire process went very smoothly with no issues while setting up my M1 MacBook Air.

After completing the setup, the next obvious step is to launch AutoGPT. Inside the AutoGPT folder, I executed the following command:

docker compose run --rm auto-gpt

It only took a few seconds to start the container and configure the environment. Finally, the agent was ready and waiting for my command.

I decided to put AutoGPT to the test by giving it a task to complete. Like any developer, I was curious if AutoGPT could code Python. So gave it a very simple problem to solve:

Create a Python program to check if a string is a palindrome or not.

The following sections attempt to capture what happened after I pressed the enter button by using relevant input and output pairs, as well as my understanding and observations.

Behind the scenes

In this section, I hope to give you a detailed look at the inner workings of AutoGPT. The following key elements guide the section’s content organization:

Input: A description of the input that the agent receives at each step.
Agent Settings: The high-level settings used by the agent for the step, including model information and temperature settings.
Prompts: The messages used to interact with the model.
Output: The LLM's output.
Observations: A concise summary of my understanding, explaining the events and outcomes of each step.

Now, let's get started; the sequence of actions taken by AutoGPT after the user gave it a task to work on is shown below:

Step 1:

Input:

User task

Agent Settings:

Creating chat completion with model gpt-3.5-turbo, temperature 0.0

Prompts:

System message:

Your task is to devise up to 5 highly effective goals and an appropriate role-based name (_GPT) for an autonomous agent, ensuring that the goals are optimally aligned with the successful completion of its assigned task.
The user will provide the task, you will provide only the output in the exact format specified below with no explanation or conversation.

Example input:
Help me with marketing my business

Example output:

Name: CMOGPT

Description: a professional digital marketer AI that assists Solopreneurs in growing their businesses by providing world-class expertise in solving marketing problems for SaaS, content products, agencies, and more.

Goals:
- Engage in effective problem-solving, prioritization, planning, and supporting execution to address your marketing needs as your virtual Chief Marketing Officer.
- Provide specific, actionable, and concise advice to help you make informed decisions without the use of platitudes or overly wordy explanations.
- Identify and prioritize quick wins and cost-effective campaigns that maximize results with minimal time and budget investment.
- Proactively take the lead in guiding you and offering suggestions when faced with unclear information or uncertainty to ensure your marketing strategy remains on track.

User message:

Task: 'create a python program for checking whether a string is palindrome or not'

Respond only with the output in the exact format specified in the system prompt, with no explanation or conversation.

Output:

Name: PalindromeCheckerGPT

Description: an intelligent AI assistant that helps you check whether a given string is a palindrome or not by providing a reliable and efficient Python program

Goals:

- Develop a Python program that efficiently checks whether a given string is a palindrome or not, using the most optimal algorithm.

- Provide a user-friendly interface for inputting strings and displaying the result of the palindrome check.

- Handle edge cases and special characters in the input string to ensure accurate palindrome detection.

- Optimize the program for performance and memory usage to handle large input strings efficiently.

- Offer clear and concise documentation for the Python program, including usage instructions and examples.

Observations:

A new agent is created based on the configurations specified in the configuration file.
A system and a user message are constructed from the task given by the user in code and passed to the LLM as input.
Based on the user's task, the LLM produces a name, description, and the top five goals for the agent to work on.

Step 2:

Input:

A series of system and user messages built with the top five goals is generated in the above step.

Agent Settings:

Creating chat completion with model gpt-4-0314, temperature 0.0

Prompts:

System message:

System Prompt: You are PalindromeCheckerGPT, an intelligent AI assistant that helps you check whether a given string is a palindrome or not by providing a reliable and efficient Python program.
Your decisions must always be made independently without seeking user assistance. Play to your strengths as an LLM and pursue simple strategies with no legal complications.

## Constraints

You operate within the following constraints:

1. ~4000 word limit for short term memory. Your short term memory is short, so immediately save important information to files.

2. If you are unsure how you previously did something or want to recall past events, thinking about similar events will help you remember.

3. No user assistance

4. Exclusively use the commands listed below e.g. command_name

## Commands

You have access to the following commands:

1. execute_python_code: Creates a Python file and executes it, params: (code: string, name: string)

2. execute_python_file: Executes an existing Python file, params: (filename: string)

3. list_files: Lists Files in a Directory, params: (directory: string)

4. read_file: Read an existing file, params: (filename: string)

5. write_to_file: Writes to a file, params: (filename: string, text: string)

6. web_search: Searches the web, params: (query: string)

7. browse_website: Browses a Website, params: (url: string, question: string)

8. goals_accomplished: Goals are accomplished and there is nothing left to do, params: (reason: string)

## Resources

You can leverage access to the following resources:

1. Internet access for searches and information gathering.

2. Long Term memory management.

3. File output.

4. Command execution

## Best practices

1. Continuously review and analyze your actions to ensure you are performing to the best of your abilities.

2. Constructively self-criticize your big-picture behavior constantly.

3. Reflect on past decisions and strategies to refine your approach.

4. Every command has a cost, so be smart and efficient. Aim to complete tasks in the least number of steps.

## Goals

For your task, you must fulfill the following goals:

1. Develop a Python program that efficiently checks whether a given string is a palindrome or not, using the most optimal algorithm.

2. Provide a user-friendly interface for inputting strings and displaying the result of the palindrome check.

3. Handle edge cases and special characters in the input string to ensure accurate palindrome detection.

4. Optimize the program for performance and memory usage to handle large input strings efficiently.

5. Offer clear and concise documentation for the Python program, including usage instructions and examples.

System message:

The current time and date is Thu Oct 19 05:37:37 2023

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};

}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "I need to start by creating a Python program that checks if a given string is a palindrome. This involves writing a function that takes a string as input, removes any non-alphanumeric characters, converts the string to lowercase, and then checks if the string is the same when reversed.",

"reasoning": "The first step in achieving the goals is to create the core functionality of the program, which is the palindrome checking function. This function should be able to handle any string input, including those with special characters and varying cases.",

"plan": "- Write the Python code for the palindrome checking function\n- Execute the Python code to test the function\n- If the function works as expected, proceed to create a user-friendly interface for the program",

"criticism": "I need to ensure that the Python code I write is efficient and handles all edge cases. I should also consider how to handle very large input strings to ensure the program performs well in all scenarios.",

"speak": "I'm going to start by writing a Python program that can check if a given string is a palindrome. This will involve creating a function that can handle any string input, including those with special characters and varying cases."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]",

"name": "palindrome_checker"

}
}
}

Observations:

The main program loop starts and a series of system messages and a user message are constructed in the code which is then passed to the LLM as input.
The system messages are written in such a way that the agent educates the GPT about the constraints, the commands it can use, the resources it has at its disposal, and the objectives it must achieve.
It also instructs the GPT to respond strictly in a defined JSON format with a predefined set of keys.
A quick summary of the response's various keys:
- text: Demonstrates how the agent has understood the requirements of the current prompt.
- reasoning: Illustrates the agent's thought process and justifies the recommendations it makes.
- plan: Acts as a concise summary of the actions outlined in the text property.
- criticism: Provides suggestions for improving future performance when used in subsequent prompts.
- speak: Provides a user-friendly summary of the agent’s thought process and recommendations, which can be conveyed to the user.
- command: The next action recommended by the agent.
The agent recommended calling execute_python_code with the generated code in the output.
Before executing the suggested code, AutoGPT prompts the user for permission to execute the command. The user can choose to run, cancel, or provide feedback in the prompt. I've decided to run the code in this instance and all subsequent instances.

Step 3:

Input:

The suggested code in the above step gets executed and the output along with the previously generated AutoGPT response is added to the above prompt.

Agent Settings:

Creating chat completion with model gpt-4-0314, temperature 0.0

Note:

In each call to a language model (LLM), AutoGPT modifies the prompts as follows:

It updates the timestamp of the second system prompt.

It appends the previous step response and command execution output before the final system and user message.

For the sake of brevity, I'll only show the newly added prompts for the current step below going forward. However, it's important to note that the entire history of the previous messages is passed to the GPT at every step.

Prompts:

Assistant message:

{

"thoughts": {

"text": "I need to start by creating a Python program that checks if a given string is a palindrome. This involves writing a function that takes a string as input, removes any non-alphanumeric characters, converts the string to lowercase, and then checks if the string is the same when reversed.",

"reasoning": "The first step in achieving the goals is to create the core functionality of the program, which is the palindrome checking function. This function should be able to handle any string input, including those with special characters and varying cases.",

"plan": "- Write the Python code for the palindrome checking function\n- Execute the Python code to test the function\n- If the function works as expected, proceed to create a user-friendly interface for the program",

"criticism": "I need to ensure that the Python code I write is efficient and handles all edge cases. I should also consider how to handle very large input strings to ensure the program performs well in all scenarios.",

"speak": "I'm going to start by writing a Python program that can check if a given string is a palindrome. This will involve creating a function that can handle any string input, including those with special characters and varying cases."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]",

"name": "palindrome_checker"

}
}
}

System message:

Command execute_python_code returned:

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};

}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond u`sing the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "Now that the palindrome checking function is ready, I need to create a user-friendly interface for the program. This will involve writing a main function that prompts the user for a string input, calls the palindrome checking function with the input, and then displays the result.",

"reasoning": "Creating a user-friendly interface is one of the goals of the task. This will make the program easy to use and understand for any user.",

"plan": "- Write the Python code for the main function\n- Execute the Python code to test the function\n- If the function works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I need to ensure that the user interface is intuitive and provides clear instructions to the user. I should also consider how to handle invalid inputs and provide useful error messages.",

"speak": "I'm now going to create a user-friendly interface for the program. This will involve writing a main function that prompts the user for a string input, calls the palindrome checking function with the input, and then displays the result."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def main():\n s = input('Enter a string: ')\n if is_palindrome(s):\n print('The string is a palindrome.')\n else:\n print('The string is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"

}
}
}

Observations:

Because no error was thrown while executing the suggested Python code, the agent assumes that the generated code meets goal 1.
As a next step, the agent is now attempting to create a user-friendly interface to execute the program, which includes creating a main function and passing parameters.
The newly suggested code has a main function to execute the program and a prompt to gather user input.
Before executing the suggested code, AutoGPT prompts the user for permission to execute the command. I've given the permission to run the code.

Step 4:

Input:

The suggested code in the above step gets executed and the output along with the previously generated AutoGPT response is added to the above prompt.

Agent Settings:

Creating chat completion with model gpt-4-0314, temperature 0.0

Prompts:

Assistant message:

{

"thoughts": {

"text": "Now that the palindrome checking function is ready, I need to create a user-friendly interface for the program. This will involve writing a main function that prompts the user for a string input, calls the palindrome checking function with the input, and then displays the result.",

"reasoning": "Creating a user-friendly interface is one of the goals of the task. This will make the program easy to use and understand for any user.",

"plan": "- Write the Python code for the main function\n- Execute the Python code to test the function\n- If the function works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I need to ensure that the user interface is intuitive and provides clear instructions to the user. I should also consider how to handle invalid inputs and provide useful error messages.",

"speak": "I'm now going to create a user-friendly interface for the program. This will involve writing a main function that prompts the user for a string input, calls the palindrome checking function with the input, and then displays the result."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def main():\n s = input('Enter a string: ')\n if is_palindrome(s):\n print('The string is a palindrome.')\n else:\n print('The string is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"

}
}
}

System message:

Command execute_python_code returned: Error: Traceback (most recent call last):

File "/app/auto_gpt_workspace/PalindromeCheckerGPT/executed_code/palindrome_checker.py", line 8, in <module>

main()

File "/app/auto_gpt_workspace/PalindromeCheckerGPT/executed_code/palindrome_checker.py", line 3, in main

if is_palindrome(s):

NameError: name 'is_palindrome' is not defined

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};

}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "The error indicates that the 'is_palindrome' function is not defined. This is because I executed the 'main' function in a separate Python execution, where the 'is_palindrome' function was not available. I need to combine the 'is_palindrome' function and the 'main' function into a single Python program and execute it.",

"reasoning": "The 'is_palindrome' function and the 'main' function need to be part of the same Python program in order to call the 'is_palindrome' function from the 'main' function.",

"plan": "- Combine the 'is_palindrome' function and the 'main' function into a single Python program\n- Execute the Python program to test the functions\n- If the program works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I should have realized that the 'is_palindrome' function and the 'main' function need to be part of the same Python program. I need to ensure that all functions are available in the same scope when executing Python code.",

"speak": "I made a mistake by trying to execute the 'main' function in a separate Python execution, where the 'is_palindrome' function was not available. I'll correct this by combining the 'is_palindrome' function and the 'main' function into a single Python program and executing it."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n s = input('Enter a string: ')\n if is_palindrome(s):\n print('The string is a palindrome.')\n else:\n print('The string is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"

}
}
}

Observations:

The agent acknowledges that the is_palindrome function is missing from the same script and figures out that both the is_palindrome function and the main function must be written in the same script file.
In the newly suggested code, the is_palindrom and main functions are both added to the same script files and then the AutoGPT prompts the user to continue with the code execution.
After giving permission to execute the code, it appeared that the AutoGPT programme execution was hung for a while. However, upon closer inspection, it was discovered that the currently suggested code contains a line s = input('Enter a string: '), which stops the program execution and expects the user to enter the input string to validate the programming logic.
The programme execution then continues after pressing enter without entering any string.

Step 5:

Input:

The suggested code in the above step gets executed and the output along with the previously generated AutoGPT response is added to the above prompt.

Agent Settings:

Creating chat completion with model gpt-4-0314, temperature 0.0

Prompts:

Assistant message:

{

"thoughts": {

"text": "The error indicates that the 'is_palindrome' function is not defined. This is because I executed the 'main' function in a separate Python execution, where the 'is_palindrome' function was not available. I need to combine the 'is_palindrome' function and the 'main' function into a single Python program and execute it.",

"reasoning": "The 'is_palindrome' function and the 'main' function need to be part of the same Python program in order to call the 'is_palindrome' function from the 'main' function.",

"plan": "- Combine the 'is_palindrome' function and the 'main' function into a single Python program\n- Execute the Python program to test the functions\n- If the program works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I should have realized that the 'is_palindrome' function and the 'main' function need to be part of the same Python program. I need to ensure that all functions are available in the same scope when executing Python code.",

"speak": "I made a mistake by trying to execute the 'main' function in a separate Python execution, where the 'is_palindrome' function was not available. I'll correct this by combining the 'is_palindrome' function and the 'main' function into a single Python program and executing it."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n s = input('Enter a string: ')\n if is_palindrome(s):\n print('The string is a palindrome.')\n else:\n print('The string is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"

}
}
}

System message:

Command execute_python_code returned: Enter a string: The string is not a palindrome.

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};
}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "The Python program seems to be working as expected, but I can't interact with it directly to provide input and see the output. I need to modify the program to accept command line arguments instead of using the 'input' function. This will allow me to test the program with different inputs.",

"reasoning": "The 'input' function is not suitable for this environment because it requires interactive input, which is not possible in this context. By modifying the program to accept command line arguments, I can provide different inputs to the program and see the output.",

"plan": "- Modify the Python program to accept command line arguments\n- Execute the Python program with different inputs to test the functions\n- If the program works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I should have realized earlier that the 'input' function is not suitable for this environment. I need to consider the constraints of the environment when designing the program.",

"speak": "I'm going to modify the Python program to accept command line arguments instead of using the 'input' function. This will allow me to test the program with different inputs."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "import sys\n\ndef is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n s = sys.argv[1]\n if is_palindrome(s):\n print('The string is a palindrome.')\n else:\n print('The string is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"
}
}
}

Observations:

In this case, I entered "hello" as the text when the program execution was expecting user input to check the logic. The output correctly identifies that the given string is not a palindrome, and the program runs without error.
Now that the agent realizes it can't interact with the code directly to test multiple cases, it decided to modify the code to accept command line arguments instead of using the 'input' function.

Step 6:

Input:

The suggested code in the above step gets executed and the output along with the previously generated AutoGPT response is added to the above prompt.

Agent Settings:

Creating chat completion with model gpt-4-0314, temperature 0.0

Prompts:

Assistant message:

{

"thoughts": {

"text": "The Python program seems to be working as expected, but I can't interact with it directly to provide input and see the output. I need to modify the program to accept command line arguments instead of using the 'input' function. This will allow me to test the program with different inputs.",

"reasoning": "The 'input' function is not suitable for this environment because it requires interactive input, which is not possible in this context. By modifying the program to accept command line arguments, I can provide different inputs to the program and see the output.",

"plan": "- Modify the Python program to accept command line arguments\n- Execute the Python program with different inputs to test the functions\n- If the program works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I should have realized earlier that the 'input' function is not suitable for this environment. I need to consider the constraints of the environment when designing the program.",

"speak": "I'm going to modify the Python program to accept command line arguments instead of using the 'input' function. This will allow me to test the program with different inputs."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "import sys\n\ndef is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n s = sys.argv[1]\n if is_palindrome(s):\n print('The string is a palindrome.')\n else:\n print('The string is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"
}
}
}

System message:

Command execute_python_code returned: Error: Traceback (most recent call last):

File "/app/auto_gpt_workspace/PalindromeCheckerGPT/executed_code/palindrome_checker.py", line 14, in <module>

main()

File "/app/auto_gpt_workspace/PalindromeCheckerGPT/executed_code/palindrome_checker.py", line 8, in main

s = sys.argv[1]

IndexError: list index out of range

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};

}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "The error indicates that no command line argument was provided when executing the Python program. This is because I didn't provide an argument when executing the 'execute_python_code' command. I need to modify the 'execute_python_code' command to include a command line argument.",

"reasoning": "The 'sys.argv' list in Python contains the command line arguments that were passed to the script. If no arguments are provided, trying to access 'sys.argv[1]' will result in an 'IndexError'.",

"plan": "- Modify the 'execute_python_code' command to include a command line argument\n- Execute the Python program with different inputs to test the functions\n- If the program works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I should have realized that the 'sys.argv' list in Python requires at least one argument to avoid an 'IndexError'. I need to ensure that all necessary arguments are provided when executing Python code.",

"speak": "I made a mistake by not providing a command line argument when executing the Python program. I'll correct this by modifying the 'execute_python_code' command to include a command line argument."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "import sys\n\ndef is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n s = sys.argv[1]\n if is_palindrome(s):\n print('The string is a palindrome.')\n else:\n print('The string is not a palindrome.')\n\nmain()",

"name": "palindrome_checker",

"args": [

"racecar"
]
}
}
}

Observations:

The agent realizes that it executed the suggested code by not providing a command line argument.
In the modified code, a command line argument racecar is passed while executing the code.

Step 7:

Input:

The suggested code in the above step gets executed and the output along with the previously generated AutoGPT response is added to the above prompt.

Agent Settings:
Creating chat completion with model gpt-4-0314, temperature 0.0

Prompts:

Assistant message:

{

"thoughts": {

"text": "The error indicates that no command line argument was provided when executing the Python program. This is because I didn't provide an argument when executing the 'execute_python_code' command. I need to modify the 'execute_python_code' command to include a command line argument.",

"reasoning": "The 'sys.argv' list in Python contains the command line arguments that were passed to the script. If no arguments are provided, trying to access 'sys.argv[1]' will result in an 'IndexError'.",

"plan": "- Modify the 'execute_python_code' command to include a command line argument\n- Execute the Python program with different inputs to test the functions\n- If the program works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I should have realized that the 'sys.argv' list in Python requires at least one argument to avoid an 'IndexError'. I need to ensure that all necessary arguments are provided when executing Python code.",

"speak": "I made a mistake by not providing a command line argument when executing the Python program. I'll correct this by modifying the 'execute_python_code' command to include a command line argument."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "import sys\n\ndef is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n s = sys.argv[1]\n if is_palindrome(s):\n print('The string is a palindrome.')\n else:\n print('The string is not a palindrome.')\n\nmain()",

"name": "palindrome_checker",

"args": ["racecar"]

}
}
}

System message:

Command execute_python_code returned: Error: execute_python_code() got an unexpected keyword argument 'args'

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};

}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "The error indicates that the 'execute_python_code' command does not accept 'args' as a keyword argument. This means I can't provide command line arguments when executing Python code in this environment. I need to find a different way to test the Python program with different inputs.",

"reasoning": "The 'execute_python_code' command is designed to execute Python code without any command line arguments. To test the Python program with different inputs, I need to find a way to provide inputs within the Python code itself.",

"plan": "- Modify the Python program to include test cases with different inputs\n- Execute the Python program to test the functions\n- If the program works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I should have realized earlier that the 'execute_python_code' command does not accept command line arguments. I need to consider the constraints of the command when designing the program.",

"speak": "I made a mistake by trying to provide command line arguments when executing the Python program. I'll correct this by modifying the Python program to include test cases with different inputs."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n test_cases = ['racecar', 'hello', 'A man, a plan, a canal: Panama']\n for s in test_cases:\n if is_palindrome(s):\n print(f'\"{s}\" is a palindrome.')\n else:\n print(f'\"{s}\" is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"

}
}
}

Observations:

The agent realizes that it is not possible to modify the 'execute_python_code' function to accept additional command-line arguments. As a result, it is looking into a different approach for input.
In the modified code, the input strings are passed along directly within the Python code as a list of strings, and the agent uses a for loop to execute the code for each string.

Step 8:

Input:

The suggested code in the above step gets executed and the output along with the previously generated AutoGPT response is added to the above prompt.

Agent Settings:

Creating chat completion with model gpt-4-0314, temperature 0.0

Prompts:

Assistant message:

{

"thoughts": {

"text": "The error indicates that the 'execute_python_code' command does not accept 'args' as a keyword argument. This means I can't provide command line arguments when executing Python code in this environment. I need to find a different way to test the Python program with different inputs.",

"reasoning": "The 'execute_python_code' command is designed to execute Python code without any command line arguments. To test the Python program with different inputs, I need to find a way to provide inputs within the Python code itself.",

"plan": "- Modify the Python program to include test cases with different inputs\n- Execute the Python program to test the functions\n- If the program works as expected, proceed to handle edge cases and optimize the program for performance and memory usage",

"criticism": "I should have realized earlier that the 'execute_python_code' command does not accept command line arguments. I need to consider the constraints of the command when designing the program.",

"speak": "I made a mistake by trying to provide command line arguments when executing the Python program. I'll correct this by modifying the Python program to include test cases with different inputs."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n test_cases = ['racecar', 'hello', 'A man, a plan, a canal: Panama']\n for s in test_cases:\n if is_palindrome(s):\n print(f'\"{s}\" is a palindrome.')\n else:\n print(f'\"{s}\" is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"
}
}
}

System message:

Command execute_python_code returned: "racecar" is a palindrome.

"hello" is not a palindrome.

"A man, a plan, a canal: Panama" is a palindrome.

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};
}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "The Python program is working as expected and correctly identifies palindromes. The next step is to handle edge cases and optimize the program for performance and memory usage. This involves considering how the program behaves with large input strings and finding ways to improve its efficiency.",

"reasoning": "Handling edge cases and optimizing the program for performance and memory usage are part of the goals of the task. This will ensure that the program works correctly in all scenarios and performs well even with large input strings.",

"plan": "- Analyze the current Python program to identify potential improvements\n- Modify the Python program to handle edge cases and improve its efficiency\n- Execute the Python program with different inputs to test the modifications\n- If the program works as expected, proceed to document the Python program",

"criticism": "I need to ensure that the modifications I make to the Python program do not introduce new bugs or affect its correctness. I should also consider the trade-offs between performance and readability when optimizing the program.",

"speak": "The Python program is working correctly and can identify palindromes. I'm now going to handle edge cases and optimize the program for performance and memory usage. This will involve considering how the program behaves with large input strings and finding ways to improve its efficiency."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n test_cases = ['racecar', 'hello', 'A man, a plan, a canal: Panama', 'Able was I ere I saw Elba', 'Madam, in Eden, I’m Adam', 'Never odd or even', 'Doc, note: I dissent. A fast never prevents a fatness. I diet on cod.']\n for s in test_cases:\n if is_palindrome(s):\n print(f'\"{s}\" is a palindrome.')\n else:\n print(f'\"{s}\" is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"

}
}
}

Observations:

The agent verifies that the functionality is working as expected before moving on to goals 3 and 4 implementation. That is, to handle edge cases and optimize the generated script.
The agent adds more test samples to the updated code to test the edge cases and optimizes the existing code.

Step 9:

Input:

The suggested code in the above step gets executed and the output along with the previously generated AutoGPT response is added to the above prompt.

Agent Settings:

Creating chat completion with model gpt-4-0314, temperature 0.0

Prompts:

Assistant message:

{

"thoughts": {

"text": "The Python program is working as expected and correctly identifies palindromes. The next step is to handle edge cases and optimize the program for performance and memory usage. This involves considering how the program behaves with large input strings and finding ways to improve its efficiency.",

"reasoning": "Handling edge cases and optimizing the program for performance and memory usage are part of the goals of the task. This will ensure that the program works correctly in all scenarios and performs well even with large input strings.",

"plan": "- Analyze the current Python program to identify potential improvements\n- Modify the Python program to handle edge cases and improve its efficiency\n- Execute the Python program with different inputs to test the modifications\n- If the program works as expected, proceed to document the Python program",

"criticism": "I need to ensure that the modifications I make to the Python program do not introduce new bugs or affect its correctness. I should also consider the trade-offs between performance and readability when optimizing the program.",

"speak": "The Python program is working correctly and can identify palindromes. I'm now going to handle edge cases and optimize the program for performance and memory usage. This will involve considering how the program behaves with large input strings and finding ways to improve its efficiency."

},

"command": {

"name": "execute_python_code",

"args": {

"code": "def is_palindrome(s):\n s = ''.join(c for c in s if c.isalnum()).lower()\n return s == s[::-1]\n\ndef main():\n test_cases = ['racecar', 'hello', 'A man, a plan, a canal: Panama', 'Able was I ere I saw Elba', 'Madam, in Eden, I’m Adam', 'Never odd or even', 'Doc, note: I dissent. A fast never prevents a fatness. I diet on cod.']\n for s in test_cases:\n if is_palindrome(s):\n print(f'\"{s}\" is a palindrome.')\n else:\n print(f'\"{s}\" is not a palindrome.')\n\nmain()",

"name": "palindrome_checker"
}
}
}

System message:

Command execute_python_code returned: "racecar" is a palindrome.

"hello" is not a palindrome.

"A man, a plan, a canal: Panama" is a palindrome.

"Able was I ere I saw Elba" is a palindrome.

"Madam, in Eden, I’m Adam" is a palindrome.

"Never odd or even" is a palindrome.

"Doc, note: I dissent. A fast never prevents a fatness. I diet on cod." is a palindrome.

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};

}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "The Python program is correctly identifying palindromes, even in complex sentences with punctuation and mixed case. The next step is to document the Python program. This involves writing clear and concise instructions on how to use the program, as well as providing examples of its usage.",

"reasoning": "Documenting the Python program is part of the goals of the task. This will make the program easier to understand and use for any user.",

"plan": "- Write clear and concise instructions on how to use the Python program\n- Provide examples of the program's usage\n- Write the documentation to a file\n- If the documentation is clear and concise, the task is complete",

"criticism": "I need to ensure that the documentation is clear, concise, and comprehensive. It should cover all aspects of the program's usage, including how to provide input and what output to expect.",

"speak": "The Python program is working correctly and can identify palindromes. I'm now going to document the Python program. This will involve writing clear and concise instructions on how to use the program, as well as providing examples of its usage."

},

"command": {

"name": "write_to_file",

"args": {

"filename": "README.md",

"text": "# Palindrome Checker\n\nThis Python program checks if a given string is a palindrome. A palindrome is a word, phrase, number, or other sequence of characters that reads the same forward and backward, ignoring spaces, punctuation, and capitalization.\n\n## Usage\n\n1. Import the 'is_palindrome' function from the Python program.\n2. Call the 'is_palindrome' function with a string as the argument.\n3. The function will return 'True' if the string is a palindrome and 'False' otherwise.\n\n## Examples\n\n```

python\nfrom palindrome_checker import is_palindrome\n\nprint(is_palindrome('racecar')) # True\nprint(is_palindrome('hello')) # False\nprint(is_palindrome('A man, a plan, a canal: Panama')) # True\n

```\n\nThe program ignores spaces, punctuation, and capitalization when checking if a string is a palindrome. This means that 'A man, a plan, a canal: Panama' is considered a palindrome, even though it contains spaces and punctuation."

}
}
}

Observations:

Now the agent creates documentation for the generated code and suggests running the write_to_file command.

Step 10:

Input:

The suggested code in the above step gets executed and the output along with the previously generated AutoGPT response is added to the above prompt.

Agent Settings:

Creating chat completion with model gpt-4-0314, temperature 0.0

Prompts:

Assistant message:

{

"thoughts": {

"text": "The Python program is correctly identifying palindromes, even in complex sentences with punctuation and mixed case. The next step is to document the Python program. This involves writing clear and concise instructions on how to use the program, as well as providing examples of its usage.",

"reasoning": "Documenting the Python program is part of the goals of the task. This will make the program easier to understand and use for any user.",

"plan": "- Write clear and concise instructions on how to use the Python program\n- Provide examples of the program's usage\n- Write the documentation to a file\n- If the documentation is clear and concise, the task is complete",

"criticism": "I need to ensure that the documentation is clear, concise, and comprehensive. It should cover all aspects of the program's usage, including how to provide input and what output to expect.",

"speak": "The Python program is working correctly and can identify palindromes. I'm now going to document the Python program. This will involve writing clear and concise instructions on how to use the program, as well as providing examples of its usage."

},

"command": {

"name": "write_to_file",

"args": {

"filename": "README.md",

"text": "# Palindrome Checker\n\nThis Python program checks if a given string is a palindrome. A palindrome is a word, phrase, number, or other sequence of characters that reads the same forward and backward, ignoring spaces, punctuation, and capitalization.\n\n## Usage\n\n1. Import the 'is_palindrome' function from the Python program.\n2. Call the 'is_palindrome' function with a string as the argument.\n3. The function will return 'True' if the string is a palindrome and 'False' otherwise.\n\n## Examples\n\n```

python\nfrom palindrome_checker import is_palindrome\n\nprint(is_palindrome('racecar')) # True\nprint(is_palindrome('hello')) # False\nprint(is_palindrome('A man, a plan, a canal: Panama')) # True\n

```\n\nThe program ignores spaces, punctuation, and capitalization when checking if a string is a palindrome. This means that 'A man, a plan, a canal: Panama' is considered a palindrome, even though it contains spaces and punctuation."
}
}
}

System message:

Command write_to_file returned: File written to successfully.

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};

}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "I have written the documentation for the Python program and saved it to a file. The next step is to verify that the file was written correctly. This involves reading the file and checking its contents.",

"reasoning": "Verifying that the file was written correctly is an important step in ensuring the accuracy and completeness of the documentation. This will allow me to catch any errors or omissions in the documentation.",

"plan": "- Read the file and check its contents\n- If the contents of the file are correct, the task is complete",

"criticism": "I should have verified the contents of the file immediately after writing it. This would have allowed me to catch any errors or omissions in the documentation sooner.",

"speak": "I have written the documentation for the Python program and saved it to a file. I'm now going to verify that the file was written correctly by reading it and checking its contents."

},

"command": {

"name": "read_file",

"args": {

"filename": "README.md"
}
}
}

Observations:

After saving the contents in the README.md file. As a next step, the agent is trying to verify the content's accuracy and ensure that it aligns with the initial goals.
So the agents suggest calling the command read_file.

Step 11:

Input:

A new prompt message is constructed for the "read_file" command, and no prompt history is passed along in this step.
After reading the README.md file, the contents are sent back to GPT for summarization and the creation of embeddings on the summary using the text-embedding-ada-002 model. The default model for generating embeddings in AutoGPT.

Agent Settings:

Creating chat completion with model gpt-3.5-turbo, temperature 0.0

Prompts:

User message:

Write a concise summary of the following text:

LITERAL TEXT: """Palindrome Checker This Python program checks if a given string is a palindrome. A palindrome is a word, phrase, number, or other sequence of characters that reads the same forward and backward, ignoring spaces, punctuation, and capitalization. Usage Import the 'is_palindrome' function from the Python program. Call the 'is_palindrome' function with a string as the argument. The function will return 'True' if the string is a palindrome and 'False' otherwise. Examples `python from palindrome_checker import is_palindrome print(is_palindrome('racecar')) # True print(is_palindrome('hello')) # False print(is_palindrome('A man, a plan, a canal: Panama')) # True ` The program ignores spaces, punctuation, and capitalization when checking if a string is a palindrome. This means that 'A man, a plan, a canal: Panama' is considered a palindrome, even though it contains spaces and punctuation."""

CONCISE SUMMARY: The text is best summarized as

Output:

a Python program that checks if a given string is a palindrome, ignoring spaces, punctuation, and capitalization. It provides examples and instructions on how to use the program.

Prompts:

User message:

a Python program that checks if a given string is a palindrome, ignoring spaces, punctuation, and capitalization. It provides examples and instructions on how to use the program.

System message:

Command read_file returned: Error: Must provide an 'engine' or 'deployment_id' parameter to create a <class 'openai.api_resources.embedding.Embedding'>

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};

}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "There seems to be an error with the 'read_file' command. I need to find a different way to verify the contents of the file. One option is to use the 'list_files' command to check if the file exists in the directory.",

"reasoning": "Verifying the contents of the file is an important step in ensuring the accuracy and completeness of the documentation. If the 'read_file' command is not working, I need to find a different way to verify the contents of the file.",

"plan": "- Use the 'list_files' command to check if the file exists in the directory\n- If the file exists, the task is complete",

"criticism": "I should have considered the possibility that the 'read_file' command might not work. I need to be prepared for unexpected issues and have alternative strategies in mind.",

"speak": "There seems to be an issue with reading the file. I'll try a different approach to verify the contents of the file."

},

"command": {

"name": "list_files",

"args": {

"directory": "."

}
}
}

Observations:

The agent raised an error while creating embedding on the summarized text. I was experimenting with Azure Open AI and did not create the text-embedding-ada-002model instance. So the agent is unable to create embeddings for the generated summary.
The agent is now suggesting a different strategy in response to the problem with embedding creation. Which is to run the list_files command within the workspace directory to ensure the README.md file exists.

Step 12:

Input:

The suggested code in the above step gets executed and the output along with the previously generated AutoGPT response is added to the above prompt.

Agent Settings:

Creating chat completion with model gpt-4-0314, temperature 0.0

Prompts:

Assistant message:

{

"thoughts": {

"text": "There seems to be an error with the 'read_file' command. I need to find a different way to verify the contents of the file. One option is to use the 'list_files' command to check if the file exists in the directory.",

"reasoning": "Verifying the contents of the file is an important step in ensuring the accuracy and completeness of the documentation. If the 'read_file' command is not working, I need to find a different way to verify the contents of the file.",

"plan": "- Use the 'list_files' command to check if the file exists in the directory\n- If the file exists, the task is complete",

"criticism": "I should have considered the possibility that the 'read_file' command might not work. I need to be prepared for unexpected issues and have alternative strategies in mind.",

"speak": "There seems to be an issue with reading the file. I'll try a different approach to verify the contents of the file."

},

"command": {

"name": "list_files",

"args": {

"directory": "."

}
}
}

System message:

Command list_files returned: ['README.md', 'auto-gpt-memory.json', 'file_logger.txt', 'PalindromeCheckerGPT/executed_code/palindrome_checker.py']

System message:

Respond strictly with JSON. The JSON should be compatible with the TypeScript type `Response` from the following:

interface Response {

thoughts: {

// Thoughts

text: string;

reasoning: string;

// Short markdown-style bullet list that conveys the long-term plan

plan: string;

// Constructive self-criticism

criticism: string;

// Summary of thoughts to say to the user

speak: string;

};

command: {

name: string;

args: Record<string, any>;

};

}

User message:

Determine exactly one command to use based on the given goals and the progress you have made so far, and respond using the JSON schema specified previously:

Output:

{

"thoughts": {

"text": "The 'README.md' file exists in the directory, which means the documentation was written successfully. I have completed all the goals of the task: I have developed a Python program that checks if a string is a palindrome, provided a user-friendly interface, handled edge cases and special characters, optimized the program for performance and memory usage, and documented the program.",

"reasoning": "The existence of the 'README.md' file in the directory indicates that the documentation was written successfully. Since all the goals of the task have been completed, there is nothing left to do.",

"plan": "- Declare that the goals are accomplished and there is nothing left to do",

"criticism": "I should have verified the contents of the file immediately after writing it. This would have allowed me to catch any errors or omissions in the documentation sooner.",

"speak": "I have successfully completed all the goals of the task. The Python program is working correctly, and the documentation has been written and saved to a file."

},

"command": {

"name": "goals_accomplished",

"args": {

"reason": "The Python program for checking palindromes has been successfully developed and documented. All goals of the task have been accomplished."

}
}
}

Observations:

The agent confirms that the README.md file exists in the workspace directory and all of the objectives have been met.
The agent then suggests calling the goals_accomplished command to finish the program.
Executing that terminates the AutoGPT program and stops the docker container.

Conclusion:

We've reached the end of this long article, where I did my best to explain each step of the AutoGPT experience. I hope you found this article helpful and interesting. Please feel free to share your thoughts and comments.

FastStream: Python's framework for Efficient Message Queue Handling

Tvrtko Sternak — Mon, 16 Oct 2023 14:28:26 +0000

Ever felt lost in the complexity of microservices and message queues like Kafka and RabbitMQ? FastStream is here to simplify it all. That's precisely why we created FastStream. Initially, it was our solution to the challenges we faced with messaging queues in our own projects. But as it simplified our lives, we realized it could do the same for others. So, we decided to share it with the world.

FastStream streamlines the entire process of working with message queues in microservices. Parsing messages, managing networking, and keeping documentation updated—all handled effortlessly.

In this blog post, we'll explore how FastStream simplifies microservices development. Let's dive in and discover how FastStream can revolutionize your workflow.

Hint: If you want to dive in the code right away, checkout the hands-on tutorial at FastStream documentation

Our motivation

Our journey with FastStream started when we needed to integrate our machine learning models into a customer's Apache Kafka environment. To streamline this process, we created FastKafka using AIOKafka, AsyncAPI, and asyncio. It was our first step in making message queue management easier.

Later, we discovered Propan, a library created by Nikita Pastukhov, which solved similar problems but for RabbitMQ. Recognizing the potential for collaboration, we joined forces with Nikita to build a unified library that could work seamlessly with both Kafka and RabbitMQ. And that's how FastStream came to be—a solution born out of the need for simplicity and efficiency in microservices development.

Key features that set FastStream apart 🚀

FastStream is more than just another library; it's a powerful toolkit designed to simplify and supercharge your microservices development. Let's dive into the key features that make FastStream stand out:

Multiple Broker Support: FastStream provides a unified API that works seamlessly across multiple message brokers. Whether you're dealing with Kafka, RabbitMQ, or others, FastStream has you covered, making it effortless to switch between them.



broker = KafkaBroker("localhost:9092")

@broker.publisher(“prediction”)
@broker.subscriber(“input_data”)
async def on_input_data(msg: InputData) -> Prediction:
  # your processing processing
  return prediction



# Just change the broker class, 
#  rest of the code stays the same
broker = RabbitBroker("localhost:5672")

@broker.publisher(“prediction”)
@broker.subscriber(“input_data”)
async def on_input_data(msg: InputData) -> Prediction:
  # your processing processing
  return prediction

Pydantic Validation: Leverage the robust validation capabilities of Pydantic to serialize and validate incoming messages. With Pydantic, you can ensure that your data is always in the right format.



broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

@broker.subscriber(“input_data”)
async def on_input_data(msg: InputData): # <- decodes consumed message using InputData(**json.loads(data))
  # your processing logic



broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

@broker.publisher(“prediction”)
@broker.subscriber(“input_data”)
async def on_input_data(msg: InputData) -> Prediction: # <- encodes produced message using Prediction.json()
  # some processing
  return prediction

Automatic Documentation: FastStream keeps you ahead of the game with automatic AsyncAPI documentation generation. Say goodbye to outdated documentation – FastStream ensures it's always up-to-date.

Intuitive Development: FastStream offers full-typed editor support, catching errors before they reach runtime. This means you can code with confidence, knowing that issues are caught early in the development process.

Powerful Dependency Injection System: Manage your service dependencies efficiently with FastStream's built-in Dependency Injection (DI) system. Say goodbye to spaghetti code and embrace clean, modular architecture.



broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

@broker.subscriber(“input_data”)
# Load a global logger instance from faststream.Context()
async def on_input_data(msg: InputData, logger=Context()):
  logger.info(msg)

Testability: FastStream supports in-memory tests, making your Continuous Integration and Continuous Deployment (CI/CD) pipeline faster and more reliable. Test your microservices with ease, ensuring they perform as expected.



async def test_base_app():
  # Subscribe to prediction topic so that we can assert incoming msgs
  @broker.subscriber("prediction")
  async def on_prediction(msg: Prediction):
    pass

  async with TestKafkaBroker(broker):
    # Publish a test message to the input_data topic
    await broker.publish(InputData(data=0.2), "input_data")

    # Check that the handle function for "input_data" topic was called with the correct msg
    on_input_data.mock.assert_called_once_with(dict(InputData(data=0.2)))

    # Check that the service responded with the correct prediction in the "prediction" topic
    on_prediction.mock.assert_called_once_with(dict(Prediction(score=1.2)))

Seamless Integrations: FastStream plays well with others. It's fully compatible with any HTTP framework you prefer, with a special emphasis on compatibility with FastAPI.



# Create a FastStream router
router = KafkaRouter("localhost:9092")

...

# Connect a FastStream router to a FastAPI application lifespan
app = FastAPI(lifespan=router.lifespan_context)

Built for Automatic Code Generation: FastStream is optimized for automatic code generation using advanced models like GPT. This means you can leverage the power of code generation to boost your productivity. Checkout the amazing tool we built for the microservice code generation: faststream-gen.

FastStream, in a nutshell, offers ease, efficiency, and power in your microservices development journey. Whether you're just starting or looking to scale your microservices, FastStream is your trusted companion. With these core features at your disposal, you'll be well-equipped to tackle the challenges of modern, data-centric microservices.

Let's build something!

Now, let's get our hands a bit dirty 👷.
Let's implement an example python app using FastStream that consumes names from "persons" topic and outputs greetings to the "greetings" topic.

Cookiecutter project

To start our project, we will use the prepared cookiecutter FastStream project. To find out more about it, check our detailed guide.

Install the cookiecutter package using the following command:



pip install cookiecutter

Now, run the provided cookiecutter command and fill out the relevant details to generate a new FastStream project, we will name this project "greetings_app":



cookiecutter https://github.com/airtai/cookiecutter-faststream.git

The creation process should look like this:



You`ve downloaded /Users/tvrtko/.cookiecutters/cookiecutter-faststream before. Is it okay to delete and re-download it? [y/n] (y): y
  [1/4] username (github-username): sternakt
  [2/4] project_name (My FastStream App): Greetings App
  [3/4] project_slug (greetings_app): greetings_app
  [4/4] Select streaming_service
    1 - kafka
    2 - nats
    3 - rabbit
    Choose from [1/2/3] (1): 1
```

Change the working directory to the newly created directory and install all development requirements using pip:
```sh
cd greetings_app
pip install -e ".[dev]"
```

Now we are ready to edit our greetings_app/application.py and tests/test_application.py files to implement our application logic. 


### Writing app code

**FastStream** brokers provide convenient function decorators `@broker.subscriber` and `@broker.publisher` to allow you to delegate the actual process of:

- consuming and producing data to Event queues, and

- decoding and encoding JSON encoded messages

These decorators make it easy to specify the processing logic for your consumers and producers, allowing you to focus on the core business logic of your application without worrying about the underlying integration.

Also, **FastStream** uses [**Pydantic**](https://docs.pydantic.dev/) to parse input JSON-encoded data into Python objects, making it easy to work with structured data in your applications, so you can serialize your input messages just using type annotations.

Here is an example python app we talked about:

```python
from faststream import FastStream, Logger
from faststream.kafka import KafkaBroker
from pydantic import BaseModel, Field

version = "0.1.0"
title = "My FastStream service"
description = "Description of my FastStream service"


class Name(BaseModel):
    name: str = Field(..., description="Name of the person")


class Greeting(BaseModel):
    greeting: str = Field(..., description="Greeting message")


broker = KafkaBroker()
app = FastStream(broker, title=title, version=version, description=description)

to_greetings = broker.publisher(
    "greetings",
    description="Produces a message on greetings after receiving a meesage on names",
)


@broker.subscriber("names", description="Consumes messages from names topic and produces messages to greetings topic")
async def on_names(msg: Name, logger: Logger) -> None:
    result = f"hello {msg.name}"
    logger.info(result)
    greeting = Greeting(greeting=result)
    await to_greetings.publish(greeting)
```

The example application will subscribe to **persons** Kafka topic and consume Name JSON messages from it. When the application consumes a message it will publish a Greetings JSON message **greetings** topic.

We can save the application into the `application.py` file and let's take a closer look at the code.

**Creating a broker**
To create an application, first we need to create a broker. This is the main piece of FastStream and takes care of the defining subscribers and producers.

```python
version = "0.1.0"
title = "My FastStream service"
description = "Description of my FastStream service"

...

broker = KafkaBroker()
app = FastStream(broker, title=title, version=version, description=description)
```

**Defining data structures**
Next, we need to define the structure of incoming and outgoing data. FastStream is integrated with Pydantic and offers automatic encoding and decoding of JSON formatted messages into Pydantic classes.

```python
class Name(BaseModel):
    name: str = Field(..., description="Name of the person")


class Greeting(BaseModel):
    greeting: str = Field(..., description="Greeting message")
```

**Defining a publisher**
Now, we define the publishing logic of our application.

```python
to_greetings = broker.publisher(
    "greetings",
    description="Produces a message on greetings after receiving a message on names",
)
```

**Defining a subscriber**
Finally, we can define the subscribing logic of our application. The app will consume data from the "names" topic and use the defined publisher to produce to the "greetings" topic whenever a message is consumed.

```python
@broker.subscriber("names", description="Consumes messages from names topic and produces messages to greetings topic")
async def on_names(msg: Name, logger: Logger) -> None:
    result = f"hello {msg.name}"
    logger.info(result)
    greeting = Greeting(greeting=result)
    await to_greetings.publish(greeting)
```

### Testing the service

The service can be tested using the `TestBroker` context managers, which, by default, puts the Broker into "testing mode".

The Tester will redirect your `subscriber` and `publisher` decorated functions to the InMemory brokers, allowing you to quickly test your app without the need for a running broker and all its dependencies.

Using pytest, the test for our service would look like this:

```python
import pytest
from faststream.kafka import TestKafkaBroker

from greetings_app.application import Greeting, Name, broker, on_names


# Subscribe to the "greetings" topic so we can monitor 
# messages our application is producing
@broker.subscriber("greetings")
async def on_greetings(msg: Greeting) -> None:
    pass


@pytest.mark.asyncio
async def test_on_names():
    async with TestKafkaBroker(broker):
        # Send John to "names" topic
        await broker.publish(Name(name="John"), "names")

        # Assert that our application has consumed "John"
        on_names.mock.assert_called_with(dict(Name(name="John")))

        # Assert that our application has greeted John in the "greetings" topic
        on_greetings.mock.assert_called_with(dict(Greeting(greeting="hello John")))
```

In the test, we send a test User JSON to the **in** topic, and then we assert that the broker has responded to the **out** topic with the appropriate message.

We can save the test to the test_application.py file and run the test by executing the following command in your application root file.

```shell
pytest
```

Here is how the tests execution should look like in your terminal:


```shell
===================================== test session starts =====================================
platform darwin -- Python 3.11.5, pytest-7.4.2, pluggy-1.3.0
rootdir: /Users/tvrtko/Documents/Airt Projects/FastStream/faststream-cookiecutter/greetings_app
configfile: pyproject.toml
plugins: asyncio-0.21.1, anyio-3.7.1
asyncio: mode=Mode.STRICT
collected 1 item                                                                              

tests/test_application.py .                                                             [100%]

====================================== 1 passed in 0.34s ======================================
```

### Running the application

The application can be started using built-in **FastStream** CLI command.

To run the service, use the **FastStream CLI** command and pass the module (in this case, the file where the app implementation is located) and the app symbol to the command.

``` shell
faststream run greetings_app.application:app
```

After running the command, you should see the following output:

``` shell
2023-10-13 08:36:32,162 INFO     - FastStream app starting...
2023-10-13 08:36:32,170 INFO     - names |            - `OnNames` waiting for messages
2023-10-13 08:36:32,177 INFO     - FastStream app started successfully! To exit, press CTRL+C
```

Also, **FastStream** provides you a great hot reload feature to improve your Development Experience

``` shell
faststream run greetings_app.application:app --reload
```

And multiprocessing horizontal scaling feature as well:

``` shell
faststream run greetings_app.application:app --workers 3
```

### Documentation

FastStream provides a command to serve the AsyncAPI documentation, let's use it to document our application.
To generate and serve the documentation, run the following command:

```shell
faststream docs serve greetings_app.application:app
```

Now, you should see the following output:

```shell
INFO:     Started server process [47151]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)
```

Now open your browser at `http://localhost:8000` and enjoy in your automatically generated documentation! :tada:

![Generated docs](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/aa3d6fjauuqmbdvck6sx.png)

Aaaand, that's it! :tada: :tada: Feel free to experiment further with your application and checkout [the documentation](https://faststream.airt.ai/latest/) for more complex examples.

## Support us on GitHub and join our community :star:

Ready to join the FastStream revolution? Head over to our [GitHub repository](https://github.com/airtai/faststream) and show your support by starring it. By doing so, you'll stay in the loop with the latest developments, updates, and enhancements as we continue to refine and expand FastStream.

Don't forget, we also have an active Discord channel where you can connect with fellow FastStream enthusiasts, ask questions, and share your experiences. Your active participation in our growing community, whether on GitHub or Discord, is invaluable, and we're grateful for your interest and potential contributions. Together, we can make microservices development simpler and more efficient with FastStream.

## Conclusion

FastStream is your go-to tool for efficient microservices development. It simplifies message queues, supports various brokers, and offers Pydantic validation and auto-doc generation.

We're immensely grateful for your interest, and we look forward to your potential contributions. With FastStream in your toolkit, you're prepared to conquer the challenges of data-centric microservices like never before. Happy coding!

How we deprecated two successful projects and joined forces to create an even more successful one

Davor Runje — Mon, 09 Oct 2023 17:23:10 +0000

We had two large clients requesting to integrate our ML pipelines with their backend using Apache Kafka. Our previous HTTP API was based on FastAPI and we thought no problem, let's search for something similar but for Kafka. It turns out the one available solution was Faust, a framework with the last release at the time more than two years old. There is a community-maintained fork of it, but it was not very active and we just couldn't use it in an enterprise-level deployment.

After a short discussion, we concluded we were just too spoiled to use low-level libraries that were nothing more than just tiny wrappers around C++ libs and that we could just build our own. So, we shamelessly made one by reusing beloved paradigms from FastAPI and we shamelessly named it FastKafka. The point was to set the expectations right - you get pretty much what you would expect: function decorators for consumers and producers with type hints specifying Pydantic classes for JSON encoding/decoding, automatic message routing to Kafka brokers and documentation generation.

After deploying it with our clients, we decided to open-source it just to see if there was any interest in something like this. We made a post about it on the Hacker news and the thing just took off overnight! I remember posting it in the evening and looking into an overnight (literally) success: there were the first 50 stars in a few hours. Five days later, we crossed the 200 stars and then the HN post started to wear off and growth slowed down.

The next step was to figure out what to do next. We posted questions on a few relevant subreddits and got quite a few feature requests, mostly around supporting other protocols, encoding schemas etc. But, we also got a message from a developer of a similar framework Propan that was released at about the same time and was gaining quite a traction in the RabbitMQ community. That developer was Nikita Pastukhov and he made an intriguing proposal: let's join our efforts and create one framework with the best features of both. Both projects were growing at roughly the same speed but targeted different communities. So the potential for double growth was there. After a quick consideration, we realized there was not much to lose and there was a lot to gain. Of course, we would lose absolute control over the project but losing control to the community is the only way for an open-source project to succeed. On the positive side, we would gain a very skilled maintainer who single-handedly created a similar framework all by himself. The frameworks were conceptually very similar so we concluded there would not be much friction of ideas and we should be able to reach consensus on the most important design issues.

However, as developers, we test our hypothesis early to avoid failing later. So we started with creating a design document where we specified the main features and started discussing high-level API. There was no discussion on the implementation at all, just on what should be the look and feel of the framework to the end user. The process took roughly one month and after reaching a consensus on the design, we started the work on the implementation. The self-imposed deadline was mid-September because we already had two conference talks coming up and we wanted to use those opportunities to launch a new version.

After two months of hard work, we presented the newly released FastStream framework at Infobip Shift conference and got featured at ShiftMag. The framework now supports both Apache Kafka and RabbitMQ, but also NATS protocol with the plan to add more protocols in the near future. The overall code is much cleaner and the implementation is streamlined with abstractions covering the common functionality across the protocols. We deprecated both FastKafka and Propan, but promised to fix bugs as long as needed. However, it seems like the community already decided to switch over to gain new functionalities.

A few posts on Reddit later and the community feedback is fantastic. We have never before experienced such growth and we are just 50 stars short from our goal of 1000, less than four week from the launch. I have no doubt that this was the right move and that collaboration and not competition is the right way to build long-term success in the open-source world.

Generating production-level streaming microservices using AI

Davor Runje — Thu, 05 Oct 2023 13:16:16 +0000

faststream-gen is a Python library that uses generative AI to automatically generate FastStream applications. Simply describe your microservice in plain English, and faststream-gen will generate a production-level FastStream application ready to deploy in a few minutes and under $1 cost.

Documentation: https://faststream-gen.airt.ai

Source Code: https://github.com/airtai/faststream-gen

The code generator for FastStream is a Python library that automates the process of creating streaming microservices for Apache Kafka, RabbitMQ and other supported protocols. It works by taking your application descriptions and swiftly turning them into a ready-to-deploy FastStream applications.

The key features are:

Automatic code generation: faststream-gen enables you to easily generate complete FastStream application with minimal effort. This library allows you to outline your application requirements, and it will quickly transform them into a fully-fledged FastStream project.
Automatic test generation: faststream-gen provides dependable code through rigorous testing, including pre-implemented integration tests, ensuring stability and functionality, saving development time, and preventing common bugs.
Script Templates: Streamline the deployment of your FastStream application using faststream-gen’s built-in scripts, tailored for initiating, subscribing to Kafka topic and shutting down the local Kafka broker.
Automatic CI/CD integration: faststream-gen integrates seamlessly with your version control and continuous integration pipeline through its GitHub workflow files. These predefined configuration files are optimized for FastStream projects, enabling smooth integration with GitHub Actions. You can automate tasks such as code validation, testing, and deployment, ensuring that your FastStream application remains in top shape throughout its development lifecycle.

In this tutorial, we will walk through the process of using the faststream-gen Python library to generate two FastStream applications.

The first application will demonstrate how to retrieve current cryptocurrency prices from various web sources. Subsequently, we will utilize the functionalities of FastStream to publish retrieved data as messages to a Kafka topic.

The second application will showcase how to consume messages from the Kafka topic using FastStream consumer capabilities. It will process these messages to extract information about cryptocurrencies and calculate the price mean for each cryptocurrency within a certain time window and publish the price mean to another topic.

A short version of the tutorial is available as a video at the following link (roughly half of it):

https://youtu.be/zklmEzifKdk

Let’s get started!

Cryptocurrency analysis with FastStream

In this tutorial, we will walk you through the process of using the faststream-gen Python library to retrieve cryptocurrency prices in real time and calculate their moving average. To accomplish that, we will generate the following two FastStream applications:

A microservice retrieving current cryptocurrency prices from an external web service
A microservice consuming such messages, calculating the moving average price for each cryptocurrency and publishing it to another Kafka topic.

Let’s get started!

Installation

To complete this tutorial, you will need the following software and
Python library:

Python (version 3.8 and upward)
a valid OPENAI API key
[optional] github account and installed git command

It is recommended to use a virtual environment for your Python projects. In this tutorial, we will be using Python’s venv module to create a virtual environment.

First, create a root directory for this tutorial. Navigate to the
desired location and create a new directory called
faststream_gen_tutorial and enter it.

mkdir faststream_gen_tutorial
cd faststream_gen_tutorial

Creating and activating a new Python virtual environment

Create a new virtual environment using venv:

python3 -m venv venv

Next, activate your new virtual environment:

source venv/bin/activate

Installing the packages

Upgrade pip if needed and install faststream-gen package:

pip install --upgrade pip && pip install faststream-gen

Check that the installation was successful by running the following
command:

faststream_gen --help

You should see the full list of options of the command in the output.

Now you have successfully set up the environment and installed the
faststream-gen package.

Setting up OpenAI API key

faststream-gen uses OpenAI API and you need to export your API key in
environment variable OPENAI_API_KEY. If you use bash or compatible
shell, you can do that with the following command:

export OPENAI_API_KEY="sk-your-openai-api-key"

If you don’t already have OPENAI_API_KEY, you can create one here.

Generating FastStream applications

Retrieving and publishing crypto prices

Now, we will create an application which retrieves information about
current cryptocurrency prices from an external web
service and publishes messages to a Kafka topic. In order to achieve
this, we need to provide a high-level description of the application in
plain English containing only the necessary information needed
by a knowledgeable Python developer familiar with FastStream
framework to implement it. This should include details such as the
message schema, instructions on external API-s and web service, and
guidance on selecting the appropriate topic and partition keys.

Below is an example of such a description used in this particular case.
Notice that we did not specify steps needed to actually implement, we
specified what the service should do, but not how to do it.

Create a FastStream application which will retrieve the current cryptocurrency
price and publish it to the new_crypto_price topic. 

The application should retrieve the data every 2 seconds.

A message which will be produced is JSON with the two attributes:
- price: non-negative float (current price of cryptocurrency in USD)
- crypto_currency: string (the cryptocurrency e.g. BTC, ETH...)

The current price of Bitcoin can be retrieved by a simple GET request to:
    - https://api.coinbase.com/v2/prices/BTC-USD/spot

The current price of Ethereum can be retrieved by a simple GET request to:
    - https://api.coinbase.com/v2/prices/ETH-USD/spot

The response of this GET request is a JSON and you can get information about
the crypto_currency in:
    response['data']['base']

and the information about the price in:
    response['data']['amount']

Use utf-8 encoded crypto_currency attribute as a partition key when publishing
the message to new_crypto_price topic.

Let’s generate a new FastStream project inside the
retrieve-publish-crypto directory. First, copy the previous
description and paste it into a file called
description_retrieve_publish.txt.

Next, run the following command (parameter -i specifies the file path
for the app description file, while the parameter -o specifies the
directory where the generated project files will be saved):

faststream_gen -i description_retrieve_publish.txt -o retrieve-publish-crypto

✨  Generating a new FastStream application!
 ✔ Application description validated. 
 ✔ FastStream app skeleton code generated. 
 ✔ The app and the tests are generated. 
 ✔ New FastStream project created. 
 ✔ Integration tests were successfully completed. 
 Tokens used: 36938
 Total Cost (USD): $0.11436
✨  All files were successfully generated!

Failed generation

The generation process is not bullet-proof and it could fail with a message like this:

✨  Generating a new FastStream application!
✔ Application description validated. 
✔ New FastStream project created. 
✔ FastStream app skeleton code generated. 
✘ Error: Failed to generate a valid application and test code. 
✘ Error: Integration tests failed.
Tokens used: 79384
Total Cost (USD): $0.24567

Apologies, we couldn't generate a working application and test code from your application description.

Please run the following command to start manual debugging:

cd retrieve_without_logs && pytest

For in-depth debugging, check the retrieve-publish-crypto/_faststream_gen_logs directory for complete logs, including individual step information.

There can be a number of reasons for that, the most common being:

The specification you provided was not sufficiently detailed to generate the application. In such cases, you could try to add more detailed instructions and try again.
The task is too difficult for the default GPT-3.5 model to handle and you could try to use GPT-4 instead:

faststream_gen --model gpt-4 -i description_retrieve_publish.txt -o retrieve-publish-crypto

You were unlucky and you just need to execute the command again. Large language models are stochastic in their nature and they always give different answers to the same questions. There are retry mechanisms in the engine, but sometimes they are not enough and just rerunning the command can mediate the problem.

If none of the above strategies work, check the already generated files and see what the potential problem could be. You can also finish the implementation or tests yourself.

Successful generation

If successful, the command will generate a FastStream project with the following structure:

retrieve-publish-crypto
├── .github
│   └── workflows
│       ├── build_docker.yml
│       ├── deploy_docs.yml
│       └── test.yml
├── .gitignore
├── Dockerfile
├── LICENSE
├── README.md
├── app
│   └── application.py
├── pyproject.toml
├── scripts
│   ├── build_docker.sh
│   ├── lint.sh
│   ├── services.yml
│   ├── start_kafka_broker_locally.sh
│   ├── static-analysis.sh
│   ├── stop_kafka_broker_locally.sh
│   └── subscribe_to_kafka_broker_locally.sh
└── tests
    ├── __init__.py
    └── test_application.py

The generated application is located in the app/ directory, while the tests are located in the tests/ directory. It is important to keep in mind that these files are generated by LLM and may vary with each generation.

app/application.py:

import asyncio
import json
from datetime import datetime

import aiohttp

from pydantic import BaseModel, Field, NonNegativeFloat

from faststream import ContextRepo, FastStream, Logger
from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)


class CryptoPrice(BaseModel):
    price: NonNegativeFloat = Field(
        ..., examples=[50000.0], description="Current price of cryptocurrency in USD"
    )
    crypto_currency: str = Field(
        ..., examples=["BTC"], description="The cryptocurrency"
    )

publisher = broker.publisher("new_crypto_price")

async def fetch_crypto_price(
    url: str, crypto_currency: str, logger: Logger, context: ContextRepo, time_interval: int = 2
) -> None:
    # Always use context: ContextRepo for storing app_is_running variable
    while context.get("app_is_running"):
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                if response.status == 200:
                    data = await response.json()
                    price = data["data"]["amount"]
                    new_crypto_price = CryptoPrice(
                        price=price, crypto_currency=crypto_currency
                    )
                    await publisher.publish(
                        new_crypto_price,
                        key=crypto_currency.encode("utf-8"),
                    )
                else:
                    logger.warning(
                        f"Failed API request {url} at time {datetime.now()}"
                    )

        await asyncio.sleep(time_interval)


@app.on_startup
async def app_setup(context: ContextRepo):
    context.set_global("app_is_running", True)


@app.on_shutdown
async def shutdown(context: ContextRepo):
    context.set_global("app_is_running", False)

    # Get all the running tasks and wait them to finish
    fetch_tasks = context.get("fetch_tasks")
    await asyncio.gather(*fetch_tasks)


@app.after_startup
async def publish_crypto_price(logger: Logger, context: ContextRepo):
    logger.info("Starting publishing:")

    cryptocurrencies = [("Bitcoin", "BTC"), ("Ethereum", "ETH")]
    fetch_tasks = [
        asyncio.create_task(
            fetch_crypto_price(
                f"https://api.coinbase.com/v2/prices/{crypto_currency}-USD/spot",
                crypto_currency,
                logger,
                context,
            )
        )
        for _, crypto_currency in cryptocurrencies
    ]
    # you need to save asyncio tasks so you can wait them to finish at app shutdown (the function with @app.on_shutdown function)
    context.set_global("fetch_tasks", fetch_tasks)

tests/test_application.py:

import pytest

from faststream import Context, TestApp
from faststream.kafka import TestKafkaBroker

from app.application import CryptoPrice, app, broker


@broker.subscriber("new_crypto_price")
async def on_new_crypto_price(
    msg: CryptoPrice, key: bytes = Context("message.raw_message.key")
):
    pass


@pytest.mark.asyncio
async def test_fetch_crypto_price():
    async with TestKafkaBroker(broker):
        async with TestApp(app):
            await on_new_crypto_price.wait_call(2)
            on_new_crypto_price.mock.assert_called()

Creating a new Python virtual environment

All the required dependencies to run the newly generated FastStream
project are located within the pyproject.toml file.

Create and activate a new virtual environment inside the
retrieve-publish-crypto directory by using venv:

cd retrieve-publish-crypto
python3 -m venv venv
source venv/bin/activate

Upgrade pip if needed and install the development dependencies:

pip install --upgrade pip && pip install -e .[dev]

Testing the application

In order to verify the functional correctness of the application, it is recommended to execute the generated unit and integration test by running the pytest command.

pytest

============================= test session starts ==============================
platform linux -- Python 3.11.4, pytest-7.4.2, pluggy-1.3.0
rootdir: /work/fastkafka-gen/docs_src/tutorial/retrieve-publish-crypto
configfile: pyproject.toml
plugins: anyio-3.7.1, asyncio-0.21.1
asyncio: mode=Mode.STRICT
collected 1 item                                                               

tests/test_application.py .                                              [100%]

============================== 1 passed in 2.65s ===============================

Previewing AsyncAPI Docs

To preview AsyncAPI Docs for the application, execute the following
command:

faststream docs serve app.application:app

INFO:     Started server process [3575270]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)

You can now access the AsyncAPI Docs by opening localhost:8000 in your browser.

Starting localhost Kafka broker

To run the FastStream application locally, ensure that you have a
running Kafka broker. You can start a Kafka Docker container by
executing the start_kafka_broker_locally.sh shell script:

./scripts/start_kafka_broker_locally.sh

[+] Running 2/2
 ⠿ Network scripts_default  Created                                                                                                             0.1s
 ⠿ Container bitnami_kafka  Started

Starting the application

To start the application, execute the following command:

faststream run app.application:app

2023-09-15 13:41:21,948 INFO     - FastStream app starting...
2023-09-15 13:41:22,144 INFO     -      |            - Starting publishing:
2023-09-15 13:41:22,144 INFO     - FastStream app started successfully! To exit press CTRL+C
Topic new_crypto_price not found in cluster metadata

Ensure that the app remains running, it is needed for the subsequent
steps.

Calculating the moving average

Let’s develop an application that calculates the average price of the
three most recent messages received from the new_crypto_price topic
for each cryptocurrency. Subsequently, we will publish the computed
average price to the price_mean topic.

Here is the full description of the desired application:

Create a FastStream application for consuming messages
from the new_crypto_price topic. 

This topic needs to use a partition key.

new_crypto_price messages use JSON with two attributes 
(create class CryptoPrice with these attributes):
- price: non-negative float (it represents the current price of the crypto)
- crypto_currency: string (it represents the cryptocurrency e.g. BTC, ETH...)

The application should save each message to a dictionary (global variable) 
- partition key should be used as a dictionary key 
  and value should be a List of prices.

Keep only the last 100 messages in the dictionary.

If there are fewer than 3 messages for a given partition key,
do not publish any messages.

Otherwise, Calculate the price mean of the last 3 messages
for the given partition key.

Publish the price mean to the price_mean topic and use 
the same partition key that the new_crypto_price topic is using.

Please open a new terminal and navigate to the root directory of this
tutorial, which is called faststream_gen_tutorial. Once you are inside
the faststream_gen_tutorial folder, please activate the virtual
environment.

cd path_to/faststream_gen_tutorial

source venv/bin/activate

To create a faststream application inside the calculate-mean-app
directory, first copy the previous description and paste it into the
description_calculate_mean.txt file.

Next, run the following command:

faststream_gen -i description_calculate_mean.txt -o calculate-mean-app

✨  Generating a new FastStream application!
 ✔ Application description validated. 
 ✔ FastStream app skeleton code generated. 
 ✔ The app and the tests are generated. 
 ✔ New FastStream project created. 
 ✔ Integration tests were successfully completed. 
 Tokens used: 13367
 Total Cost (USD): $0.04147
✨  All files were successfully generated!

If successful, the command will generate calculate-mean-app directory
with app/application.py and tests/test_application.py inside. If
not, just rerun the command again until it succeds.

app/application.py:

from typing import Dict, List

from pydantic import BaseModel, Field, NonNegativeFloat

from faststream import Context, ContextRepo, FastStream, Logger
from faststream.kafka import KafkaBroker


class CryptoPrice(BaseModel):
    price: NonNegativeFloat = Field(
        ..., examples=[50000], description="Current price of the cryptocurrency"
    )
    crypto_currency: str = Field(
        ..., examples=["BTC"], description="Cryptocurrency symbol"
    )


broker = KafkaBroker("localhost:9092")
app = FastStream(broker)


publisher = broker.publisher("price_mean")


@app.on_startup
async def app_setup(context: ContextRepo):
    message_history: Dict[str, List[float]] = {}
    context.set_global("message_history", message_history)


@broker.subscriber("new_crypto_price")
async def on_new_crypto_price(
    msg: CryptoPrice,
    logger: Logger,
    message_history: Dict[str, List[float]] = Context(),
    key: bytes = Context("message.raw_message.key"),
) -> None:
    logger.info(f"New crypto price {msg=}")

    crypto_key = key.decode("utf-8")
    if crypto_key not in message_history:
        message_history[crypto_key] = []

    message_history[crypto_key].append(msg.price)

    if len(message_history[crypto_key]) > 100:
        message_history[crypto_key] = message_history[crypto_key][-100:]

    if len(message_history[crypto_key]) >= 3:
        price_mean = sum(message_history[crypto_key][-3:]) / 3
        await publisher.publish(price_mean, key=key)

tests/test_application.py:

import pytest

from faststream import Context, TestApp
from faststream.kafka import TestKafkaBroker

from app.application import CryptoPrice, app, broker, on_new_crypto_price


@broker.subscriber("price_mean")
async def on_price_mean(msg: float, key: bytes = Context("message.raw_message.key")):
    pass


@pytest.mark.asyncio
async def test_app():
    async with TestKafkaBroker(broker):
        async with TestApp(app):
            await broker.publish(
                CryptoPrice(price=50000, crypto_currency="BTC"),
                "new_crypto_price",
                key=b"BTC",
            )
            on_new_crypto_price.mock.assert_called_with(
                dict(CryptoPrice(price=50000, crypto_currency="BTC"))
            )
            on_price_mean.mock.assert_not_called()

            await broker.publish(
                CryptoPrice(price=60000, crypto_currency="BTC"),
                "new_crypto_price",
                key=b"BTC",
            )
            on_new_crypto_price.mock.assert_called_with(
                dict(CryptoPrice(price=60000, crypto_currency="BTC"))
            )
            on_price_mean.mock.assert_not_called()

            await broker.publish(
                CryptoPrice(price=70000, crypto_currency="BTC"),
                "new_crypto_price",
                key=b"BTC",
            )
            on_new_crypto_price.mock.assert_called_with(
                dict(CryptoPrice(price=70000, crypto_currency="BTC"))
            )
            on_price_mean.mock.assert_called_with(60000.0)

Creating a new Python virtual environment

All the required dependencies to run the newly generated FastStream
project are located within the pyproject.toml file. Create a new
virtual environment and install the development dependencies for the
project.

Create and activate a new virtual environment inside the
calculate-mean-app directory by using env:

cd calculate-mean-app
python3 -m venv venv
source venv/bin/activate

Upgrade pip if needed and install the development dependencies:

pip install --upgrade pip && pip install -e .[dev]

Testing the application

In order to verify functional correctness of the application, it is
recommended to execute the generated unit and integration test by
running the pytest command.

pytest

============================= test session starts ==============================
platform linux -- Python 3.11.4, pytest-7.4.2, pluggy-1.3.0
rootdir: /work/fastkafka-gen/docs_src/tutorial/calculate-mean-app
configfile: pyproject.toml
plugins: anyio-3.7.1, asyncio-0.21.1
asyncio: mode=Mode.STRICT
collected 1 item                                                               

tests/test_application.py .                                              [100%]

============================== 1 passed in 0.44s ===============================

Previewing AsyncAPI Docs

To preview AsyncAPI Docs for the application, execute the following
command:

faststream docs serve app.application:app

INFO:     Started server process [3596205]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)

You can now access the AsyncAPI Docs by opening http://localhost:8000 in your browser.

Starting the application

To start the application, execute the following command:

faststream run app.application:app

2023-10-04 09:31:18,926 INFO     - FastStream app starting...
2023-10-04 09:31:18,948 INFO     - new_crypto_price |            - `OnNewCryptoPrice` waiting for messages
Topic new_crypto_price not found in cluster metadata
2023-10-04 09:31:19,069 INFO     - FastStream app started successfully! To exit, press CTRL+C
2023-10-04 09:31:40,876 INFO     - new_crypto_price | 0-16964047 - Received
2023-10-04 09:31:40,878 INFO     - new_crypto_price | 0-16964047 - New crypto price msg=CryptoPrice(price=27414.085, crypto_currency='BTC')
2023-10-04 09:31:40,878 INFO     - new_crypto_price | 0-16964047 - Processed
2023-10-04 09:31:40,878 INFO     - new_crypto_price | 1-16964047 - Received
2023-10-04 09:31:40,879 INFO     - new_crypto_price | 1-16964047 - New crypto price msg=CryptoPrice(price=1642.425, crypto_currency='ETH')
2023-10-04 09:31:40,879 INFO     - new_crypto_price | 1-16964047 - Processed
2023-10-04 09:31:43,053 INFO     - new_crypto_price | 2-16964047 - Received
2023-10-04 09:31:43,054 INFO     - new_crypto_price | 2-16964047 - New crypto price msg=CryptoPrice(price=27414.085, crypto_currency='BTC')
2023-10-04 09:31:43,054 INFO     - new_crypto_price | 2-16964047 - Processed
2023-10-04 09:31:43,054 INFO     - new_crypto_price | 3-16964047 - Received
2023-10-04 09:31:43,055 INFO     - new_crypto_price | 3-16964047 - New crypto price msg=CryptoPrice(price=1642.425, crypto_currency='ETH')
...

You can see in the terminal that the application is reading the messages
from the new_crypto_price topic. Ensure that the app remains running
as it is needed for the next step.

Subscribing directly to local Kafka broker topic

Open the new terminal, navigate to the calculate-mean-app
directory.

To check if the calculate-mean-app is publishing messages to the
price_mean topic, run the following command:

./scripts/subscribe_to_kafka_broker_locally.sh price_mean

BTC     26405.745
ETH     1621.3733333333332
BTC     26404.865
ETH     1621.375
BTC     26404.865
...

Stopping Kafka broker

To stop the Kafka broker after analyzing the mean price of
cryptocurrencies, you can execute the following command:

./scripts/stop_kafka_broker_locally.sh

[+] Running 2/2
 ⠿ Container bitnami_kafka  Removed                                                                            1.2s
 ⠿ Network scripts_default  Removed

[Optional] GitHub integration

To successfully complete this optional tutorial chapter, make sure you
have a GitHub account and the Git command installed. Additionally, ensure that your authentication for GitHub is properly set.

We need to create two GitHub repositories, one for the FastStream project in the retrieve-publish-crypto directory and another for the FastStream project in the calculate-mean-app directory. In this chapter, we will upload the FastStream project to GitHub, which includes the retrieve-publish-crypto project. We will also provide an
explanation of the generated CI workflows.

Adding locally hosted code to GitHub

To create a GitHub repository, click on the following link. For the
Repository name, use retrieve-publish-crypto and click on
Create repository.

Please open your development environment and go to the
retrieve-publish-crypto directory that was generated in the previous
steps.

cd path_to/faststream_gen_tutorial/retrieve-publish-crypto

Next, execute the following commands:

git init
git add .
git commit -m "Retrieve and publish crypto prices application implemented"
git branch -M main

If you are using HTTPS authentication execute (replace the
git_username in the URL with your own GitHub username):

git remote add origin https://github.com/git_username/retrieve-publish-crypto.git

If you are using SSH authentication execute (replace the
git_username in the URL with your own GitHub username):

git remote add origin git@github.com:git_username/retrieve-publish-crypto.git

To update the remote branch with local commits, execute:

git push -u origin main

Continuous integration with GitHub Actions

Once the changes are pushed, CI pipeline will run the tests again,
create and publish the documentation and build Docker container with the
application.

To verify the passing status of the CI, open your web browser, go to the
newly created GitHub repository, and click on the Actions tab.

The tests executed with pytest are passing successfully.

The Docker image has been successfully built and pushed to the GitHub
Container registry under the repository
ghcr.io/your_username/retrieve-publish-crypto

The AsyncAPI docs have been successfully generated and deployed to
GitHub Pages.

After the successful execution of the Deploy FastStream AsyncAPI Docs
workflow, a new branch named gh-pages will be created. To access the
GitHub Pages settings, navigate to the Settings -> Pages, select the
gh-pages branch as the designated branch for hosting the GitHub Pages
and click on Save.

By setting up a branch on GitHub Pages, a new workflow will
automatically be triggered within the GitHub Actions. To access this
workflow, simply click on the Actions tab and open the
pages build and deployment workflow.

Once all the jobs within the workflow are completed, you can click on
the provided URL to conveniently view and explore the AsyncAPI Docs that
have been generated for your application.

Repeat

Repeat the same process with calculate-mean-app project.

Next steps

Congratulations! You have successfully completed this tutorial and
gained a new set of skills. Now that you have learned how to use
faststream-gen, try it out with your own example!

FastKafka - free open source python lib for building Kafka-based services

Davor Runje — Wed, 15 Mar 2023 05:18:11 +0000

We were searching for something like FastAPI for Kafka-based service we were developing, but couldn’t find anything similar. So we shamelessly made one by reusing beloved paradigms from FastAPI and we shamelessly named it FastKafka. The point was to set the expectations right - you get pretty much what you would expect: function decorators for consumers and producers with type hints specifying Pydantic classes for JSON encoding/decoding, automatic message routing to Kafka brokers and documentation generation.

Please take a look and tell us how to make it better. Our goal is to make using it as easy as possible for someone with experience with FastAPI.

https://github.com/airtai/fastkafka