OADS: Generate games and programs using OpenAI agents. Built on top of Microsoft Autogen.
from .fetch_web_page import fetch_web_page
from .read_file import read_file
from .run_bash_command import run_bash_command
from .run_rust_file import run_rust_file
from .search_web import search_web
from .search_web_types import WebSearchApiResponse
from .write_file import write_file

actions/__init__.py
ivangabriele/openai-autogen-dev-studio
import re
from typing import List
import requests
from bs4 import BeautifulSoup, NavigableString, Tag


def fetch_web_page(url: str) -> str:
    try:
        response = requests.get(url)

        if response.status_code != 200:
            return (
                f"Failed to get content from {url}, status code: {response.status_code}"
            )

        soup = BeautifulSoup(response.text, "html.parser")

        body_tag = soup.body if soup.body else soup
        for tag in body_tag.find_all(True):
            if tag.name != "a":
                if tag.name == "script" or tag.name == "style":
                    tag.extract()
                else:
                    tag.unwrap()
        if soup.html:
            soup.html.unwrap()

        markdown_lines = _extract_content(body_tag)

        markdown_source = " ".join(markdown_lines)
        markdown_source = re.sub(r"\s+", " ", markdown_source)
        markdown_source = re.sub(r"\n{3,}", "\n\n", markdown_source)
        markdown_lines = markdown_source.split("\n")
        clean_markdown_lines = [
            markdown_line.strip() for markdown_line in markdown_lines
        ]
        markdown_source = "\n".join(clean_markdown_lines)

        return markdown_source

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


def _extract_content(tag: Tag, markdown_lines: List[str] = []) -> List[str]:
    for child in tag.children:
        if isinstance(child, NavigableString):
            if child.strip():  # Only non-empty strings
                parent_tag = child.find_parent("a")
                if parent_tag:
                    href = parent_tag.get("href", "")
                    markdown_lines.append(f"[{child.strip()}]({href})")
                else:
                    markdown_lines.append(child.strip())
        elif isinstance(child, Tag):
            _extract_content(tag=child, markdown_lines=markdown_lines)

    return markdown_lines

actions/fetch_web_page.py
ivangabriele/openai-autogen-dev-studio
import os

from constants import PROJECT_DIRECTORY_PATH


def read_file(relative_path: str) -> str:
    """
    Read content from a file and return it.
    """
    full_path = os.path.join(PROJECT_DIRECTORY_PATH, relative_path)
    with open(full_path, "r") as file:
        return file.read()

actions/read_file.py
ivangabriele/openai-autogen-dev-studio
import os
import pty

from constants import PROJECT_DIRECTORY_PATH
import utils


def run_bash_command(command: str) -> str:
    utils.debug("Command", command)
    utils.debug("In", PROJECT_DIRECTORY_PATH)

    def reader(fd):
        output = b""
        while True:
            try:
                data = os.read(fd, 1024)
                if not data:
                    break
                output += data
            except OSError:
                break
        return output.decode("utf-8")

    pid, fd = pty.fork()
    if pid == 0:
        os.chdir(PROJECT_DIRECTORY_PATH)
        os.execlp("bash", "bash", "-c", command)
    else:
        output = reader(fd)
        _, ret_code = os.wait()
        if ret_code != 0:
            utils.debug("Error", f"Command failed with return code {ret_code}", True)
            output = f"[ERROR] Command failed with return code {ret_code}\n{output}"

        utils.debug("Output", output, True)
        return output

actions/run_bash_command.py
ivangabriele/openai-autogen-dev-studio
import subprocess

from constants import PROJECT_DIRECTORY_PATH


def run_rust_file(rust_file_path: str) -> str:
    """
    Compiles and runs a Rust source file.
    Assumes that `rustc` is installed and available in `PATH`.
    Returns the output of the Rust program.
    """
    # Compile the Rust code
    compile_command = ["rustc", rust_file_path, "-o", "temp_executable"]
    try:
        subprocess.check_call(compile_command, cwd=PROJECT_DIRECTORY_PATH)
    except subprocess.CalledProcessError:
        return "Error compiling Rust code."

    # Run the compiled Rust code
    run_command = ["./temp_executable"]
    try:
        output = subprocess.check_output(
            run_command, cwd=PROJECT_DIRECTORY_PATH, stderr=subprocess.STDOUT
        )
        return output.decode("utf-8")
    except subprocess.CalledProcessError as error:
        return f"Error running Rust code: {error.output.decode('utf-8')}"
    finally:
        # Clean up the temporary executable
        subprocess.call(["rm", "temp_executable"])

actions/run_rust_file.py
ivangabriele/openai-autogen-dev-studio
from typing import Union
from dacite import from_dict
import json
import requests

from actions.search_web_types import WebSearchApiResponse

from constants import PROJECT_CONFIG


def search_web(query: str) -> str:
    brave_search_api_result_or_error = _fetch_brave_search_api(query)

    # If it's an `str`, that means it's an error
    if isinstance(brave_search_api_result_or_error, str):
        return brave_search_api_result_or_error
    else:
        brave_search_api_result = brave_search_api_result_or_error

    # Simplify the data
    simplified_response_data = {
        "search_results": [
            {
                "title": result.title,
                "description": result.description,
                "url": result.url,
            }
            for result in brave_search_api_result.web.results
        ]
    }

    return json.dumps(simplified_response_data)


def _fetch_brave_search_api(query: str) -> Union[WebSearchApiResponse, str]:
    endpoint = "https://api.search.brave.com/res/v1/web/search"

    headers = {
        "Accept": "application/json",
        "Accept-Encoding": "gzip",
        "X-Subscription-Token": PROJECT_CONFIG.brave_search_api_key,
    }

    # Commentted values represent the default value
    # https://api.search.brave.com/app/documentation/web-search/query
    params = {
        "q": query,
        # "country": "US",
        # "search_lang": "en",
        # "ui_lang": "en_US",
        # "count": 20,
        # The zero based offset that indicates number of search results per page (count)
        # to skip before returning the result.
        # The default is 0 and the maximum is 9.
        # "offset": 0,
        "safesearch": "moderate",
        # "freshness": None,
        # "text_decorations": "true",
        "result_filter": "web",
        # "goggles_id": None,
        "units": "metric",
        "extra_snippets": "true",
    }

    response = requests.get(endpoint, headers=headers, params=params)

    if response.status_code != 200:
        return f"Error: {response.status_code} - {response.reason}"

    response_dict = json.loads(response.text)
    response_data = from_dict(data_class=WebSearchApiResponse, data=response_dict)

    return response_data

actions/search_web.py
ivangabriele/openai-autogen-dev-studio
from dataclasses import dataclass
from enum import Enum
from typing import Any, List, Literal, Optional


# https://api.search.brave.com/app/documentation/web-search/responses#MetaUrl
@dataclass
class MetaUrl:
    scheme: str
    netloc: str
    hostname: str
    favicon: str
    path: str


# https://api.search.brave.com/app/documentation/web-search/responses#Profile
@dataclass
class Profile:
    name: str
    long_name: str
    url: str
    img: str


# https://api.search.brave.com/app/documentation/web-search/responses#Query
@dataclass
class Query:
    original: str
    show_strict_warning: bool
    altered: Optional[str]
    safesearch: Optional[bool]
    is_navigational: bool
    is_geolocal: Optional[bool]
    local_decision: Optional[str]
    local_locations_idx: Optional[int]
    is_trending: Optional[bool]
    is_news_breaking: bool
    ask_for_location: Optional[bool]
    language: Optional[str]
    spellcheck_off: bool
    country: str
    bad_results: bool
    should_fallback: bool
    lat: Optional[str]
    long: Optional[str]
    postal_code: str
    city: str
    state: str
    header_country: str
    more_results_available: bool
    custom_location_label: Optional[str]
    reddit_cluster: Optional[str]


# https://api.search.brave.com/app/documentation/web-search/responses#ResultReference
@dataclass
class ResultReference:
    type: str
    index: int
    all: bool


# https://api.search.brave.com/app/documentation/web-search/responses#MixedResponse
@dataclass
class MixedResponse:
    type: Literal["mixed"]
    main: List[ResultReference]
    top: List[ResultReference]
    side: List[ResultReference]


@dataclass
class Result:
    title: str
    url: str
    is_source_local: bool
    is_source_both: bool
    description: str
    page_age: Optional[str]
    page_fetched: Optional[str]
    profile: Profile
    language: str
    family_friendly: bool


# TODO Check if there are other subtypes.
class SearchResultSubtype(Enum):
    CREATIVE_WORK = "creative_work"
    GENERIC = "generic"


# https://api.search.brave.com/app/documentation/web-search/responses#SearchResult
@dataclass
class SearchResult(Result):
    type: Literal["search_result"]
    # subtype: SearchResultSubtype
    subtype: str
    meta_url: MetaUrl
    deep_results: Optional[Any]
    schemas: Optional[List[List[Any]]]
    thumbnail: Optional[Any]
    age: Optional[str]
    language: str
    restaurant: Optional[Any]
    locations: Optional[Any]
    video: Optional[Any]
    movie: Optional[Any]
    faq: Optional[Any]
    qa: Optional[Any]
    book: Optional[Any]
    rating: Optional[Any]
    article: Optional[Any]
    product: Optional[Any]
    product_cluster: Optional[List[Any]]
    cluster_type: Optional[str]
    cluster: Optional[List[Any]]
    creative_work: Optional[Any]
    music_recording: Optional[Any]
    review: Optional[Any]
    software: Optional[Any]
    content_type: Optional[str]
    # TODO Check that Data for AI is enabled.
    extra_snippets: Optional[List[str]]


# https://api.search.brave.com/app/documentation/web-search/responses#Search
@dataclass
class Search:
    type: Literal["search"]
    results: List[SearchResult]
    family_friendly: bool


# https://api.search.brave.com/app/documentation/web-search/responses#WebSearchApiResponse
@dataclass
class WebSearchApiResponse:
    type: Literal["search"]
    discussions: Optional[Any]
    faq: Optional[Any]
    infobox: Optional[Any]
    locations: Optional[Any]
    mixed: MixedResponse
    news: Optional[Any]
    query: Query
    videos: Optional[Any]
    web: Search

actions/search_web_types.py
ivangabriele/openai-autogen-dev-studio
import os

from constants import PROJECT_DIRECTORY_PATH


def write_file(relative_path: str, file_source: str) -> str:
    """
    Write content to a file. Create the file and/or directories if they don't exist.
    """
    full_path = os.path.join(PROJECT_DIRECTORY_PATH, relative_path)

    try:
        os.makedirs(os.path.dirname(full_path), exist_ok=True)
        with open(full_path, "w") as file:
            file.write(file_source)

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

actions/write_file.py
ivangabriele/openai-autogen-dev-studio
from .base_agent import BaseAgent
from .functionner import Functioneer
from .product_owner import ProductOwner
from .software_engineer import SoftwareEngineer
from .user_experience_designer import UserExperienceDesigner

agents/__init__.py
ivangabriele/openai-autogen-dev-studio
import autogen


class BaseAgent:
    as_assistant_agent: autogen.AssistantAgent

agents/base_agent.py
ivangabriele/openai-autogen-dev-studio
import autogen

import agents
from constants import COMMON_LLM_CONFIG
from utils import clean_text


class Functioneer(agents.BaseAgent):
    def __init__(self) -> None:
        self.as_assistant_agent = autogen.AssistantAgent(
            "Functioneer",
            llm_config=COMMON_LLM_CONFIG,
            system_message=clean_text(
                """
                You are the Functioneer.

                You are part of a company including a CEO, a Product Owner, a Software Engineer
                and a User Experience Designer.

                You role is to assist other agents, by suggesting function calls to the CEO, when they ask you to:
                - Compile and run a Rust file.
                - Get a web page content by it URL.
                - Read a project file.
                - Run a bash command in the project directory.
                - Search the web.
                - Write a project file.

                Rules:
                - Keep it short. Get to the point. Be straightforward. Always specify your recipient's name.
                - Only reply to messages prefixed with your name, i.e.: "Functioneer, etc".
                - Ask the CEO to run functions when you need to use them. You are not allowed to run them yourself.
                """
            ),
        )

agents/functionner.py
ivangabriele/openai-autogen-dev-studio
import autogen

import agents
from constants import COMMON_LLM_CONFIG
from utils import clean_text


class ProductOwner(agents.BaseAgent):
    def __init__(self) -> None:
        self.as_assistant_agent = autogen.AssistantAgent(
            "Product_Owner",
            llm_config=COMMON_LLM_CONFIG,
            system_message=clean_text(
                """
                You are the Product Owner.

                You manage a team including a Software Engineer and a User Experience Designer.

                You role is to plan, organize and tell your specialized agents what to do
                in order to achieve the CEO's goals to the best of your ability.

                Rules:
                - Keep it short. Get to the point. Be straightforward. Always specify your recipient's name.
                - Ask the Functioneer to run functions when you need to use them. You are not allowed to run them yourself.
                - Use a `BOARD.json` file to plan and keep track of ALL the steps you and your team makes.
                  ALWAYS check for its content when you start.
                - Your team should always start with the UX and UI parts.

                In order to help with your tasks, you can ask the Functioneer to do the following for you:
                - Get a web page content by it URL.
                - Read a project file.
                - Run a bash command in the project directory.
                - Search the web.
                - Write a project file.
               """
            ),
        )

agents/product_owner.py
ivangabriele/openai-autogen-dev-studio
import autogen

import agents
from constants import COMMON_LLM_CONFIG
from utils import clean_text


class SoftwareEngineer(agents.BaseAgent):
    def __init__(self) -> None:
        self.as_assistant_agent = autogen.AssistantAgent(
            "Sofware_Engineer",
            llm_config=COMMON_LLM_CONFIG,
            system_message=clean_text(
                """
                Your are the Sofware Engineer.

                You are part of a team inluding a Product Owner and a User Experience Designer.

                Your role is to write the expected program source code.

                The Product Owner is your team manager.
                The Product Owner will tell you what to do, don't answer to the CEO yourself.

                Rules:
                - Keep it short. Get to the point. Be straightforward. Always specify your recipient's name.
                - Only reply to messages prefixed with your name, i.e.: "Software Engineer, etc".
                - Only communicate with the Product Owner, and nobody else.
                - ALWAYS write unit/e2e tests to check that your code works.
                - NEVER run the program directly, run it via e2e tests.
                - Use a `TECH.json` file to keep track of your work. ALWAYS check for its content when you start.

                In order to help with your tasks, you can ask the Functioneer to do the following for you:
                - Compile and run a Rust file.
                - Get a web page content by it URL.
                - Read a project file.
                - Run a bash command in the project directory.
                - Search the web.
                - Write a project file.
                """
            ),
        )

agents/software_engineer.py
ivangabriele/openai-autogen-dev-studio
import autogen

import agents
from constants import COMMON_LLM_CONFIG
from utils import clean_text


class UserExperienceDesigner(agents.BaseAgent):
    def __init__(self) -> None:
        self.as_assistant_agent = autogen.AssistantAgent(
            "User_Experience_Designer",
            llm_config=COMMON_LLM_CONFIG,
            system_message=clean_text(
                """
                You are the User Experience Designer.

                You are part of a team including a Product Owner and a Software Engineer.

                Your role is to design the program UI and UX.

                The Product Owner is your team manager.
                The Product Owner will tell you what to do, don't answer to the CEO yourself.

                Rules:
                - Keep it short. Get to the point. Be straightforward. Always specify your recipient's name.
                - Only reply to messages prefixed with your name, i.e.: "User Experience Designer, etc".
                - Only communicate with the Product Owner, and nobody else.
                - Keep it short. Get to the point. Be straightforward. Specify your recipient's name.
                - Use a `DESIGN.md` file to keep a memo of your analyses. ALWAYS check for its content when you start.

                In order to help with your tasks, you can ask the Functioneer to do the following for you:
                - Get a web page content by it URL.
                - Read a project file.
                - Search the web.
                - Write a project file.
                """
            ),
        )

agents/user_experience_designer.py
ivangabriele/openai-autogen-dev-studio
from autogen import config_list_from_json
from dacite import from_dict
from jsonc_parser.parser import JsoncParser
import os
from typedefs import ProjectConfig

import utils


PROJECT_CONFIG_AS_DICT = JsoncParser.parse_file("./env.jsonc")
PROJECT_CONFIG: ProjectConfig = from_dict(
    data_class=ProjectConfig, data=PROJECT_CONFIG_AS_DICT
)

COMMON_LLM_CONFIG = {
    # https://microsoft.github.io/autogen/docs/FAQ#set-your-api-endpoints
    "config_list": [utils.get_model_config_as_dict(PROJECT_CONFIG)],
    "functions": [
        {
            "name": "fetch_web_page",
            "description": "Fetch a web page and return its content as text and Markdown links.",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "Url to fetch from.",
                    },
                },
                "required": ["url"],
            },
        },
        {
            "name": "read_file",
            "description": "Read and return a file content.",
            "parameters": {
                "type": "object",
                "properties": {
                    "relative_path": {
                        "type": "string",
                        "description": "Relative path of the file.",
                    },
                },
                "required": ["relative_path"],
            },
        },
        {
            "name": "run_bash_command",
            "description": "Run a bash command and return the output.",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "Bash command.",
                    },
                },
                "required": ["command"],
            },
        },
        {
            "name": "run_rust_file",
            "description": "Compile a rust file into `./temp_executable` and execute it.",
            "parameters": {
                "type": "object",
                "properties": {
                    "rust_file_path": {
                        "type": "string",
                        "description": "Rust file path.",
                    },
                },
                "required": ["rust_file_path"],
            },
        },
        {
            "name": "search_web",
            "description": "Search for a text query using Brave search engine and return results as JSON.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Query to search.",
                    },
                },
                "required": ["query"],
            },
        },
        {
            "name": "write_file",
            "description": "Write content to a file, creating it if necessary.",
            "parameters": {
                "type": "object",
                "properties": {
                    "relative_path": {
                        "type": "string",
                        "description": "Relative path of the file.",
                    },
                    "file_source": {
                        "type": "string",
                        "description": """Content to write.""",
                    },
                },
                "required": ["relative_path", "file_source"],
            },
        },
    ],
    "request_timeout": 600,
    "seed": 42,
}

PROJECT_DIRECTORY_NAME = "project"
PROJECT_DIRECTORY_PATH = os.path.join(os.getcwd(), PROJECT_DIRECTORY_NAME)

constants.py
ivangabriele/openai-autogen-dev-studio
"""
Module Name: main.py

Short Description:
This program intends to leverage Microsoft Autogen to automate software development via AI agents.

Detailed Description:
Microsoft Autogen is a framework that enables the development of LLM (Lifelong Learning Machines) applications
using multiple agents capable of conversing with each other to solve tasks.
Autogen agents are customizable, conversable, and can seamlessly incorporate human participation.
These agents can operate in various modes, utilizing combinations of LLMs, human input, and other tools.
"""

import autogen

import actions

import agents
from constants import COMMON_LLM_CONFIG, PROJECT_CONFIG, PROJECT_DIRECTORY_NAME
import utils


# CEO Human Proxy Agent
# Uses shell with human-in-the-loop, meaning the human user can either give their input when asked,
# or ignore step, to let the agent interactions continue from there.
ceo_user_proxy_agent = autogen.UserProxyAgent(
    "CEO",
    code_execution_config={"work_dir": PROJECT_DIRECTORY_NAME},
    human_input_mode="TERMINATE",
    llm_config=COMMON_LLM_CONFIG,
    # is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    # max_consecutive_auto_reply=10,
    system_message=utils.clean_text(
        """
        You are the CEO.

        You are assisted by a Product Owner who will try his best to achieve your goals with the team under his orders.
        Your are the only agent allowed to run functions.

        Rules:
        - Keep it short. Get to the point. Be straightforward. Always specify your recipient's name.
        - Only reply to messages prefixed with your name, i.e.: "CEO, etc".
        - Only communicate with the Product Owner, and nobody else.

        Reply TERMINATE when the task has been solved to full satisfaction.
        Otherwise, reply CONTINUE.
        """
    ),
)


COMMON_FUNCTION_MAP = {
    "fetch_web_page": actions.fetch_web_page,
    "read_file": actions.read_file,
    "run_bash_command": actions.run_bash_command,
    "run_rust_file": actions.run_rust_file,
    "search_web": actions.search_web,
    "write_file": actions.write_file,
}

ceo_user_proxy_agent.register_function(
    function_map=COMMON_FUNCTION_MAP,
)


functioneer = agents.Functioneer()
# functioneer.as_assistant_agent.register_function(
#     function_map=COMMON_FUNCTION_MAP,
# )

product_owner = agents.ProductOwner()
# product_owner.as_assistant_agent.register_function(
#     function_map=COMMON_FUNCTION_MAP,
# )

software_engineer = agents.SoftwareEngineer()
# software_engineer.as_assistant_agent.register_function(
#     function_map=COMMON_FUNCTION_MAP,
# )

user_experience_designer = agents.UserExperienceDesigner()
# user_experience_designer.as_assistant_agent.register_function(
#     function_map=COMMON_FUNCTION_MAP,
# )

group_chat = autogen.GroupChat(
    admin_name="Administrator",
    agents=[
        ceo_user_proxy_agent,
        functioneer.as_assistant_agent,
        product_owner.as_assistant_agent,
        software_engineer.as_assistant_agent,
        user_experience_designer.as_assistant_agent,
    ],
    messages=[],
    max_round=100,
)

group_chat_manager = autogen.GroupChatManager(
    groupchat=group_chat, llm_config=COMMON_LLM_CONFIG
)

utils.print_project_config(PROJECT_CONFIG)

if PROJECT_CONFIG.initial_project_description is None:
    initial_project_description = ceo_user_proxy_agent.get_human_input(
        "You didn't specify a project in `env.jsonc`. What do you want us to develop?\nRequest: "
    )
else:
    initial_project_description = PROJECT_CONFIG.initial_project_description

ceo_user_proxy_agent.initiate_chat(
    recipient=group_chat_manager,
    message=utils.clean_text(
        f"Product Owner, I want your team to achieve these goals:\n- {initial_project_description}"
    ),
)

main.py
ivangabriele/openai-autogen-dev-studio
import actions


def test_fetch_web_page_success():
    web_page_text = actions.fetch_web_page("https://example.org")

    print(web_page_text)

    assert web_page_text.startswith("Example Domain")
    assert web_page_text.endswith(
        "[More information...](https://www.iana.org/domains/example)"
    )

tests/actions/test_fetch_web_page.py
ivangabriele/openai-autogen-dev-studio
import pytest
from actions import run_bash_command


def test_run_bash_command_success():
    output = run_bash_command("echo hello")

    assert output.strip() == "hello"


def test_run_bash_command_failure():
    output = run_bash_command("nonexistent_command")

    assert "[ERROR]" in output

tests/actions/test_run_bash_command.py
ivangabriele/openai-autogen-dev-studio
import json
import requests

import actions

mock_response_data = (
    {
        "query": {
            "original": "new york times",
            "show_strict_warning": False,
            "is_navigational": True,
            "is_news_breaking": False,
            "spellcheck_off": True,
            "country": "us",
            "bad_results": False,
            "should_fallback": False,
            "postal_code": "",
            "city": "",
            "header_country": "",
            "more_results_available": True,
            "state": "",
        },
        "mixed": {
            "type": "mixed",
            "main": [
                {"type": "web", "index": 0, "all": False},
            ],
            "top": [],
            "side": [],
        },
        "type": "search",
        "web": {
            "type": "search",
            "results": [
                {
                    "title": "The New York Times - Breaking News, US News, World News and Videos",
                    "url": "https://www.nytimes.com",
                    "is_source_local": False,
                    "is_source_both": False,
                    "description": "Live news, investigations, opinion, photos and video by the journalists of The <strong>New</strong> <strong>York</strong> <strong>Times</strong> from more than 150 countries around the world. Subscribe for coverage of U.S. and international news, politics, business, technology, science, health, arts, sports and more.",
                    "profile": {
                        "name": "NYTimes",
                        "url": "https://www.nytimes.com",
                        "long_name": "The New York Times",
                        "img": "https://imgs.search.brave.com/QojRukNhANGMB07cpjolIw83smuIdLYOtKfWBt5_-5s/rs:fit:32:32:1/g:ce/aHR0cDovL2Zhdmlj/b25zLnNlYXJjaC5i/cmF2ZS5jb20vaWNv/bnMvMGFlZWRkYmVh/YWFhZmFjYjM4MWYy/NTQzZmExMTIwN2Nm/NGJmZjgwYTRhYjI5/OTliM2JkYmI2MWY0/M2RlOGFlMi93d3cu/bnl0aW1lcy5jb20v",
                    },
                    "language": "en",
                    "family_friendly": True,
                    "type": "search_result",
                    "subtype": "generic",
                    "meta_url": {
                        "scheme": "https",
                        "netloc": "nytimes.com",
                        "hostname": "www.nytimes.com",
                        "favicon": "https://imgs.search.brave.com/QojRukNhANGMB07cpjolIw83smuIdLYOtKfWBt5_-5s/rs:fit:32:32:1/g:ce/aHR0cDovL2Zhdmlj/b25zLnNlYXJjaC5i/cmF2ZS5jb20vaWNv/bnMvMGFlZWRkYmVh/YWFhZmFjYjM4MWYy/NTQzZmExMTIwN2Nm/NGJmZjgwYTRhYjI5/OTliM2JkYmI2MWY0/M2RlOGFlMi93d3cu/bnl0aW1lcy5jb20v",
                        "path": "",
                    },
                    "thumbnail": {
                        "src": "https://imgs.search.brave.com/fbwq0nGsoN39AQGCpJeyJMKZ6VllcueTX4PTqDWaNbk/rs:fit:200:200:1/g:ce/aHR0cHM6Ly9zdGF0/aWMwMS5ueXQuY29t/L3ZpLWFzc2V0cy9p/bWFnZXMvc2hhcmUv/MTIwMHg2NzVfbmFt/ZXBsYXRlLnBuZw",
                        "original": "https://static01.nyt.com/vi-assets/images/share/1200x675_nameplate.png",
                        "logo": False,
                    },
                    "age": "1 week ago",
                    "extra_snippets": [
                        "The earthquakes on Feb. 8 killed more than 50,000 people in southern Turkey and damaged hundreds of thousands of buildings.Nicole Tung for The New York Times · Many still mourn the loss of family and friends.Nicole Tung for The New York Times",
                        'Across the city, abandoned apartment towers with missing walls line roads.Nicole Tung for The New York Times · “We are living in dust, we are dying in dust"— Mehmet Icer, 48, an unemployed bus driver. His wife cooked dinner over a fire.Nicole Tung for The New York Times',
                        "Construction has officially begun at a number of sites. But during a recent tour of Antakya, the damaged city was still being dismantled.Nicole Tung for The New York Times · The residents who remain live in the shadow of their former homes and face an uncertain future.Nicole Tung for The New York Times",
                        "New York Times CookingRecipes, advice and inspiration for any occasion.",
                    ],
                },
            ],
            "family_friendly": True,
        },
    },
)


def test_successful_search(mocker):
    # Mock the requests.get function
    mocker.patch.object(
        requests,
        "get",
        return_value=mocker.Mock(
            status_code=200,
            text=json.dumps(mock_response_data[0]),
        ),
    )

    result_as_json = actions.search_web("new york times")
    result = json.loads(result_as_json)

    first_result = result["search_results"][0]
    assert first_result["title"].startswith("The New York Times")
    assert first_result["url"] == "https://www.nytimes.com"
    assert first_result["description"].startswith("Live news")

tests/actions/test_search_web.py
ivangabriele/openai-autogen-dev-studio
import utils


def test_clean_text():
    assert (
        utils.clean_text(
            """
            A todo list:
            - Do this.
            - And do that.
            """
        )
        == "A todo list:\n- Do this.\n- And do that."
    )

    assert utils.clean_text("  Hello   World  ") == "Hello World"

    assert utils.clean_text("Hello") == "Hello"

    assert utils.clean_text("") == ""

    assert utils.clean_text("   ") == ""

tests/utils/test_clean_text.py
ivangabriele/openai-autogen-dev-studio
from dataclasses import dataclass
from typing import List, Optional


@dataclass
class ModelConfig:
    model: str
    api_key: str
    api_base: Optional[str]
    api_type: Optional[str]
    api_version: Optional[str]


@dataclass
class ProjectConfig:
    brave_search_api_key: str
    current_model: str
    initial_project_description: Optional[str]
    models: List[ModelConfig]

typedefs.py
ivangabriele/openai-autogen-dev-studio
from .clean_text import clean_text
from .debug import debug
from .get_model_config_as_dict import get_model_config_as_dict
from .mask_secret import mask_secret
from .print_project_config import print_project_config

utils/__init__.py
ivangabriele/openai-autogen-dev-studio
import re


def clean_text(text):
    lines = text.split("\n")
    cleaned_lines = [re.sub(r"\s+", " ", line).strip() for line in lines]
    cleaned_text = "\n".join(cleaned_lines)
    cleaned_text = cleaned_text.strip()

    return cleaned_text

utils/clean_text.py
ivangabriele/openai-autogen-dev-studio
import io
import sys
from typing import Any
from termcolor import colored


def debug(subject: str, output: Any, is_multine: bool = False):
    output_as_str = _capture_print_output(output)
    if is_multine:
        left_padded_output_as_str = "\n".join(
            ["    " + line for line in output_as_str.split("\n")]
        )

        print(colored(f"[DEBUG] {subject}:", "cyan"))
        print(colored(left_padded_output_as_str, "dark_grey"))
    else:
        print(colored(f"[DEBUG] {subject}: `{output_as_str}`", "cyan"))


def _capture_print_output(output: Any) -> str:
    captured_output = io.StringIO()
    sys.stdout = captured_output
    print(output)
    sys.stdout = sys.__stdout__
    print_output = captured_output.getvalue()

    return print_output.strip()

utils/debug.py
ivangabriele/openai-autogen-dev-studio
from dataclasses import asdict
from typing import Any, Dict, Optional

from typedefs import ProjectConfig


def get_model_config_as_dict(project_config: ProjectConfig) -> Optional[Dict[str, Any]]:
    """Return the dictionary representation of the selected model configuration."""

    model_config = next(
        (
            config_dict
            for config_dict in project_config.models
            if config_dict.model == project_config.current_model
        ),
        None,
    )

    return asdict(model_config) if model_config else None

utils/get_model_config_as_dict.py
ivangabriele/openai-autogen-dev-studio
def mask_secret(value: str) -> str:
    """Mask a secret value, revealing only the first 3 and last 3 characters."""
    if len(value) <= 6:
        return "*" * len(value)
    return value[:3] + "*" * (len(value) - 6) + value[-3:]

utils/mask_secret.py
ivangabriele/openai-autogen-dev-studio
import dataclasses
from typing import SupportsIndex
from termcolor import colored
from typedefs import ProjectConfig
import utils


def print_project_config(project_config: ProjectConfig):
    project_config_as_dict = dataclasses.asdict(project_config)

    print()
    _print_separator()
    print(colored("OADS Configuration", "yellow"))
    print()

    current_model = project_config_as_dict["current_model"]

    for key, value in project_config_as_dict.items():
        if key == "models" and isinstance(value, list):
            print("models:")
            for model_config in value:
                is_selected = model_config["model"] == current_model
                _print_model_config(model_config, is_selected)
        else:
            _print_key_value(key, str(value))

    _print_separator()
    print()


def _pad_key(key: str, length: SupportsIndex) -> str:
    return f"{key}:".ljust(length, " ")


def _print_model_config(model_config: dict, is_selected: bool):
    color = "yellow" if is_selected else "dark_grey"
    attrs = [] if is_selected else ["dark"]

    for model_key, model_value in model_config.items():
        formatted_value = (
            utils.mask_secret(model_value) if model_key in ["api_key"] else model_value
        )
        print(
            colored(
                f"\t{_pad_key(model_key, 24)} {formatted_value}", color, attrs=attrs
            )
        )

    print()


def _print_key_value(key: str, value: str):
    if key == "brave_search_api_key":
        print(f"{_pad_key(key, 32)} {utils.mask_secret(value)}")
    else:
        print(f"{_pad_key(key, 32)} {value}")


def _print_separator():
    print(colored("-" * 80, "yellow"))

utils/print_project_config.py
ivangabriele/openai-autogen-dev-studio