File: docs/create_api_rst.py

"""Script for auto-generating api_reference.rst"""
import glob
import re

from pathlib import Path

ROOT_DIR = Path(__file__).parents[1].absolute()
print(ROOT_DIR)
PKG_DIR = ROOT_DIR / "gpt_engineer"
WRITE_FILE = Path(__file__).parent / "api_reference.rst"


def load_members() -> dict:
    members: dict = {}
    for py in glob.glob(str(PKG_DIR) + "/**/*.py", recursive=True):
        module = py[len(str(PKG_DIR)) + 1 :].replace(".py", "").replace("/", ".")
        top_level = module.split(".")[0]
        if top_level not in members:
            members[top_level] = {"classes": [], "functions": []}
        with open(py, "r") as f:
            for line in f.readlines():
                cls = re.findall(r"^class ([^_].*)\(", line)
                members[top_level]["classes"].extend([module + "." + c for c in cls])
                func = re.findall(r"^def ([^_].*)\(", line)
                afunc = re.findall(r"^async def ([^_].*)\(", line)
                func_strings = [module + "." + f for f in func + afunc]
                members[top_level]["functions"].extend(func_strings)
    return members


def construct_doc(members: dict) -> str:
    full_doc = """\
.. _api_reference:

=============
API Reference
=============

"""
    for module, _members in sorted(members.items(), key=lambda kv: kv[0]):
        classes = _members["classes"]
        functions = _members["functions"]
        if not (classes or functions):
            continue

        module_title = module.replace("_", " ").title()
        if module_title == "Llms":
            module_title = "LLMs"
        section = f":mod:`gpt_engineer.{module}`: {module_title}"
        full_doc += f"""\
{section}
{'=' * (len(section) + 1)}

.. automodule:: gpt_engineer.{module}
    :no-members:
    :no-inherited-members:

"""

        if classes:
            cstring = "\n    ".join(sorted(classes))
            full_doc += f"""\
Classes
--------------
.. currentmodule:: gpt_engineer

.. autosummary::
    :toctree: {module}
    :template: class.rst

    {cstring}

"""
        if functions:
            fstring = "\n    ".join(sorted(functions))
            full_doc += f"""\
Functions
--------------
.. currentmodule:: gpt_engineer

.. autosummary::
    :toctree: {module}

    {fstring}

"""
    return full_doc


def main() -> None:
    members = load_members()
    full_doc = construct_doc(members)
    with open(WRITE_FILE, "w") as f:
        f.write(full_doc)


if __name__ == "__main__":
    main()



File: docs/conf.py

#!/usr/bin/env python
#
# file_processor documentation build configuration file, created by
# sphinx-quickstart on Fri Jun  9 13:47:02 2017.
#
# This file is execfile()d with the current directory set to its
# containing dir.
#
# Note that not all possible configuration values are present in this
# autogenerated file.
#
# All configuration values have a default; values that are commented out
# serve to show the default.

# If extensions (or modules to document with autodoc) are in another
# directory, add these directories to sys.path here. If the directory is
# relative to the documentation root, use os.path.abspath to make it
# absolute, like shown here.
#
import os
import sys

from pathlib import Path

import toml

sys.path.insert(0, os.path.abspath(".."))

ROOT_DIR = Path(__file__).parents[1].absolute()

with open("../pyproject.toml") as f:
    data = toml.load(f)


# The master toctree document.
master_doc = "index"

# General information about the project.
project = data["tool"]["poetry"]["name"]
copyright = "2023 Anton Osika"
author = " Anton Osika & Contributors"

# The version info for the project you're documenting, acts as replacement
# for |version| and |release|, also used in various other places throughout
# the built documents.
#
# The short X.Y version.
version = data["tool"]["poetry"]["version"]
# The full version, including alpha/beta/rc tags.
release = data["tool"]["poetry"]["version"]


# -- General configuration ---------------------------------------------

# If your documentation needs a minimal Sphinx version, state it here.
#
# needs_sphinx = '1.0'

# Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
extensions = [
    "sphinx.ext.autodoc",
    "sphinx.ext.autodoc.typehints",
    "sphinx.ext.autosummary",
    "sphinx.ext.napoleon",
    "sphinx.ext.viewcode",
    "sphinx_copybutton",
    "myst_parser",
    "IPython.sphinxext.ipython_console_highlighting",
]

# The suffix(es) of source filenames.
# You can specify multiple suffix as a list of string:

source_suffix = [".rst", ".md"]

autodoc_pydantic_model_show_json = False
autodoc_pydantic_field_list_validators = False
autodoc_pydantic_config_members = False
autodoc_pydantic_model_show_config_summary = False
autodoc_pydantic_model_show_validator_members = False
autodoc_pydantic_model_show_validator_summary = False
autodoc_pydantic_model_signature_prefix = "class"
autodoc_pydantic_field_signature_prefix = "param"
autodoc_member_order = "groupwise"
autoclass_content = "both"
autodoc_typehints_format = "short"

autodoc_default_options = {
    "members": True,
    "show-inheritance": True,
    "inherited-members": "BaseModel",
    "undoc-members": False,
}

# Add any paths that contain templates here, relative to this directory.
templates_path = ["_templates"]


# source_suffix = '.rst'


# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
#
# This is also used if you do content translation via gettext catalogs.
# Usually you set "language" from the command line for these cases.
language = "en"

# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
# This patterns also effect to html_static_path and html_extra_path
exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]

# The name of the Pygments (syntax highlighting) style to use.
pygments_style = "sphinx"

# If true, `todo` and `todoList` produce output, else they produce nothing.
todo_include_todos = False


# -- Options for HTML output -------------------------------------------

# The theme to use for HTML and HTML Help pages.  See the documentation for
# a list of builtin themes.
#
# html_theme = 'alabaster'
html_theme = "sphinx_rtd_theme"

# Theme options are theme-specific and customize the look and feel of a
# theme further.  For a list of options available for each theme, see the
# documentation.
#
# html_theme_options = {}

# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "default.css" will overwrite the builtin "default.css".
# html_static_path = ["_static"]


# -- Options for HTMLHelp output ---------------------------------------

# Output file base name for HTML help builder.
htmlhelp_basename = "gpt_engineerdoc"


# -- Options for LaTeX output ------------------------------------------

latex_elements = {
    # The paper size ('letterpaper' or 'a4paper').
    #
    # 'papersize': 'letterpaper',
    # The font size ('10pt', '11pt' or '12pt').
    #
    # 'pointsize': '10pt',
    # Additional stuff for the LaTeX preamble.
    #
    # 'preamble': '',
    # Latex figure (float) alignment
    #
    # 'figure_align': 'htbp',
}

# Grouping the document tree into LaTeX files. List of tuples
# (source start file, target name, title, author, documentclass
# [howto, manual, or own class]).
latex_documents = [
    (master_doc, "gpt_engineer.tex", "GPT-ENgineer Documentation", "manual"),
]


# -- Options for manual page output ------------------------------------

# One entry per manual page. List of tuples
# (source start file, name, description, authors, manual section).
man_pages = [(master_doc, "gpt_engineer", "GPT-Engineer Documentation", [author], 1)]


# -- Options for Texinfo output ----------------------------------------

# Grouping the document tree into Texinfo files. List of tuples
# (source start file, target name, title, author,
#  dir menu entry, description, category)
texinfo_documents = [
    (
        master_doc,
        "gpt_engineer",
        "GPT-Engineer Documentation",
        author,
        "gpt_engineer",
        "One line description of project.",
        "Miscellaneous",
    ),
]

# generate autosummary even if no references
autosummary_generate = True

myst_enable_extensions = [
    "colon_fence",
]

myst_all_links_external = True



File: scripts/print_chat.py

"""
This module provides functionality to print a conversation with messages
colored according to the role of the speaker.
"""

import json

import typer

from termcolor import colored

app = typer.Typer()


def pretty_print_conversation(messages):
    """
    Prints a conversation with messages formatted and colored by role.

    Parameters
    ----------
    messages : list
        A list of message dictionaries, each containing 'role', 'name', and 'content' keys.

    """

    role_to_color = {
        "system": "red",
        "user": "green",
        "assistant": "blue",
        "function": "magenta",
    }
    formatted_messages = []
    for message in messages:
        if message["role"] == "function":
            formatted_messages.append(
                f"function ({message['name']}): {message['content']}\n"
            )
        else:
            assistant_content = (
                message["function_call"]
                if message.get("function_call")
                else message["content"]
            )
            role_to_message = {
                "system": f"system: {message['content']}\n",
                "user": f"user: {message['content']}\n",
                "assistant": f"assistant: {assistant_content}\n",
            }
            formatted_messages.append(role_to_message[message["role"]])

    for formatted_message in formatted_messages:
        role = messages[formatted_messages.index(formatted_message)]["role"]
        color = role_to_color[role]
        print(colored(formatted_message, color))


@app.command()
def main(
    messages_path: str,
):
    """
    Main function that loads messages from a JSON file and prints them using pretty formatting.

    Parameters
    ----------
    messages_path : str
        The file path to the JSON file containing the messages.

    """
    with open(messages_path) as f:
        messages = json.load(f)

    pretty_print_conversation(messages)


if __name__ == "__main__":
    app()



File: scripts/clean_benchmarks.py

"""
This module provides functionality to clean up benchmark directories by removing
all files and folders except for 'prompt' and 'main_prompt'.
"""

# list all folders in benchmark folder
# for each folder, run the benchmark

import os
import shutil

from pathlib import Path

from typer import run


def main():
    """
    Main function that iterates through all directories in the 'benchmark' folder
    and cleans them by removing all files and directories except for 'prompt' and
    'main_prompt'.
    """

    benchmarks = Path("benchmark")

    for benchmark in benchmarks.iterdir():
        if benchmark.is_dir():
            print(f"Cleaning {benchmark}")
            for path in benchmark.iterdir():
                if path.name in ["prompt", "main_prompt"]:
                    continue

                # Get filename of Path object
                if path.is_dir():
                    # delete the entire directory
                    shutil.rmtree(path)
                else:
                    # delete the file
                    os.remove(path)


if __name__ == "__main__":
    run(main)



File: scripts/legacy_benchmark.py

"""
This module provides functionality to run benchmarks on different folders within
the 'benchmark' directory, wait for their completion, and generate a report.
"""

# list all folders in benchmark folder
# for each folder, run the benchmark
import contextlib
import json
import os
import subprocess

from datetime import datetime
from itertools import islice
from pathlib import Path
from typing import Iterable, Union

from tabulate import tabulate
from typer import run


def main(
    n_benchmarks: Union[int, None] = None,
):
    """
    Main function that runs benchmarks on folders within the 'benchmark' directory.

    Parameters
    ----------
    n_benchmarks : Union[int, None], optional
        The number of benchmarks to run. If None, all benchmarks are run.

    """

    path = Path("benchmark")

    folders: Iterable[Path] = path.iterdir()

    if n_benchmarks:
        folders = islice(folders, n_benchmarks)

    benchmarks = []
    results = []
    for bench_folder in folders:
        if os.path.isdir(bench_folder):
            print(f"Running benchmark for {bench_folder}")

            log_path = bench_folder / "log.txt"
            log_file = open(log_path, "w")
            process = subprocess.Popen(
                [
                    "python",
                    "-u",  # Unbuffered output
                    "-m",
                    "gpt_engineer.cli.main",
                    bench_folder,
                    "--steps",
                    "benchmark",
                ],
                stdout=log_file,
                stderr=log_file,
                bufsize=0,
            )
            benchmarks.append(bench_folder)
            results.append((process, log_file))

            print("You can stream the log file by running:")
            print(f"tail -f {log_path}")
            print()

    for bench_folder, (process, file) in zip(benchmarks, results):
        process.wait()
        file.close()

        print("process", bench_folder.name, "finished with code", process.returncode)
        print("Running it. Original benchmark prompt:")
        print()
        with open(bench_folder / "prompt") as f:
            print(f.read())
        print()

        with contextlib.suppress(KeyboardInterrupt):
            subprocess.run(
                [
                    "python",
                    "-m",
                    "gpt_engineer.cli.main",
                    bench_folder,
                    "--steps",
                    "evaluate",
                ],
            )

    generate_report(benchmarks, path)


def generate_report(benchmarks, benchmark_path):
    """
    Generates a report of the benchmark results and optionally appends it to a markdown file.

    Parameters
    ----------
    benchmarks : list
        A list of benchmark folder paths that have been processed.
    benchmark_path : Path
        The path to the benchmark directory.

    """

    headers = ["Benchmark", "Ran", "Works", "Perfect", "Notes"]
    rows = []
    for bench_folder in benchmarks:
        memory = bench_folder / ".gpteng" / "memory"
        with open(memory / "review") as f:
            review = json.loads(f.read())
            rows.append(
                [
                    bench_folder.name,
                    to_emoji(review.get("ran", None)),
                    to_emoji(review.get("works", None)),
                    to_emoji(review.get("perfect", None)),
                    review.get("comments", None),
                ]
            )
    table: str = tabulate(rows, headers, tablefmt="pipe")
    print("\nBenchmark report:\n")
    print(table)
    print()
    append_to_results = ask_yes_no("Append report to the results file?")
    if append_to_results:
        results_path = benchmark_path / "RESULTS.md"
        current_date = datetime.now().strftime("%Y-%m-%d")
        insert_markdown_section(results_path, current_date, table, 2)


def to_emoji(value: bool) -> str:
    """
    Converts a boolean value to its corresponding emoji representation.

    Parameters
    ----------
    value : bool
        The boolean value to convert.

    Returns
    -------
    str
        An emoji string representing the boolean value.

    """

    return "\U00002705" if value else "\U0000274C"


def insert_markdown_section(file_path, section_title, section_text, level):
    """
    Inserts a new section into a markdown file at the specified level.

    Parameters
    ----------
    file_path : Path
        The path to the markdown file.
    section_title : str
        The title of the section to insert.
    section_text : str
        The text content of the section to insert.
    level : int
        The header level of the section.

    """

    with open(file_path, "r") as file:
        lines = file.readlines()

    header_prefix = "#" * level
    new_section = f"{header_prefix} {section_title}\n\n{section_text}\n\n"

    # Find the first section with the specified level
    line_number = -1
    for i, line in enumerate(lines):
        if line.startswith(header_prefix):
            line_number = i
            break

    if line_number != -1:
        lines.insert(line_number, new_section)
    else:
        print(
            f"Markdown file was of unexpected format. No section of level {level} found. "
            "Did not write results."
        )
        return

    # Write the file
    with open(file_path, "w") as file:
        file.writelines(lines)


def ask_yes_no(question: str) -> bool:
    """
    Asks a yes/no question and returns the response as a boolean value.

    Parameters
    ----------
    question : str
        The yes/no question to ask.

    Returns
    -------
    bool
        True if the answer is 'yes', False if 'no'.

    """

    while True:
        response = input(question + " (y/n): ").lower().strip()
        if response == "y":
            return True
        elif response == "n":
            return False
        else:
            print("Please enter either 'y' or 'n'.")


if __name__ == "__main__":
    run(main)



File: scripts/test_api.py

"""This is just a demo to test api.py."""

from time import sleep

import requests


def post_data(url, extra_arguments):
    """
    Make an HTTP POST request with extra_arguments as data.

    Parameters
    ----------
     url : str
        The URL to which the POST request should be sent.
    extra_arguments : dict
        A dictionary of data that needs to be sent in the POST request.

    Returns
    -------
    response
        The response from the server.
    """

    response = requests.post(url, json=extra_arguments)
    return response


if __name__ == "__main__":
    URL_BASE = "http://127.0.0.1:8000"

    arguments = {
        "input": "We are writing snake in python. MVC components split \
        in separate files. Keyboard control.",  # our prompt
        "additional_input": {"improve_option": False},
    }

    # create a task
    response = post_data(f"{URL_BASE}/agent/tasks", arguments)
    print(response.json())
    task_id = response.json()["task_id"]

    sleep(1)  # this is not needed

    # execute the step for our task
    response = post_data(f"{URL_BASE}/agent/tasks/{task_id}/steps", {})
    print(response.json())



File: gpt_engineer/__init__.py

# Adding convenience imports to the package

# from gpt_engineer.tools import code_vector_repository
# from gpt_engineer.core.default import on_disk_repository



File: gpt_engineer/benchmark/run.py

"""
Module for running benchmarks.

This module defines functions to run benchmarks using a given agent and to print
the results of the benchmark tasks.

Functions
---------
run : function
    Runs the benchmark tasks using the provided agent and returns a list of TaskResult objects.

print_results : function
    Prints the results of the benchmark tasks to the console.
"""
import time

from typing import List

import yaml

from gpt_engineer.benchmark.types import Assertable, Benchmark, TaskResult
from gpt_engineer.core.base_agent import BaseAgent
from gpt_engineer.core.default.disk_execution_env import DiskExecutionEnv


def run(
    agent: BaseAgent,
    benchmark: Benchmark,
    verbose=False,
) -> List[TaskResult]:
    """
    Runs the benchmark tasks using the provided agent and returns a list of TaskResult objects.

    Parameters
    ----------
    agent : BaseAgent
        The agent to use for running the benchmark tasks.
    benchmark : Benchmark
        The benchmark containing the tasks to run.
    verbose : bool, default=False
        A flag to indicate whether to print verbose output during the benchmark.

    Returns
    -------
    List[TaskResult]
        A list of TaskResult objects representing the results of the benchmark tasks.
    """
    task_results = []
    for task in benchmark.tasks:
        print(f"--> Running task: {task.name}\n")

        t0 = time.time()
        files_dict = agent.improve(task.initial_code, task.prompt)
        t1 = time.time()

        env = DiskExecutionEnv()
        env.upload(files_dict)

        if task.command:
            p = env.popen(task.command)
            stdout, stderr = p.communicate(benchmark.timeout)
            stdout, stderr = stdout.decode("utf-8"), stderr.decode("utf-8")
        else:
            p, stdout, stderr = None, None, None

        exec_result = Assertable(
            files=files_dict,
            env=env,
            process=p,
            stdout=stdout,
            stderr=stderr,
        )

        task_results.append(
            TaskResult(
                task_name=task.name,
                assertion_results={
                    assertion_name: assertion(exec_result)
                    for assertion_name, assertion in task.assertions.items()
                },
                duration=t1 - t0,
            )
        )

        if verbose:
            print_results(task_results)
    return task_results


def print_results(results: list[TaskResult]):
    """
    Prints the results of the benchmark tasks to the console.

    Parameters
    ----------
    results : list[TaskResult]
        A list of TaskResult objects representing the results of the benchmark tasks.

    Returns
    -------
    None
    """
    for task_result in results:
        print(f"\n--- Results for {task_result.task_name} ---")
        print(f"{task_result.task_name} ({task_result.duration:.2f}s)")
        for assertion_name, assertion_result in task_result.assertion_results.items():
            checkmark = "✅" if assertion_result else "❌"
            print(f"  {checkmark} {assertion_name}")
        print()

    success_rates = [task_result.success_rate for task_result in results]
    avg_success_rate = sum(success_rates) / len(results)

    total_time = sum(task_result.duration for task_result in results)

    correct_assertions = sum(
        sum(
            assertion_result
            for assertion_result in task_result.assertion_results.values()
        )
        for task_result in results
    )
    total_assertions = sum(
        len(task_result.assertion_results) for task_result in results
    )
    correct_tasks = [
        task_result for task_result in results if task_result.success_rate == 1
    ]

    print("--- Results ---")
    print(f"Total time: {total_time:.2f}s")
    print(f"Completely correct tasks: {len(correct_tasks)}/{len(results)}")
    print(f"Total correct assertions: {correct_assertions}/{total_assertions}")
    print(f"Average success rate: {avg_success_rate * 100}% on {len(results)} tasks")
    print("--- Results ---")
    print()


def export_yaml_results(yaml_path, complete_results, config):
    for results in complete_results.values():
        correct_tasks = [
            task_result
            for task_result in results["detailed"]
            if task_result["solved"] == 1.0
        ]
        fraction_correct = len(correct_tasks) / len(results["detailed"])
        results["fully_solved"] = fraction_correct
    complete_results["config"] = config
    with open(yaml_path, "w") as f:
        yaml.dump(complete_results, f, indent=4)



File: gpt_engineer/benchmark/__init__.py




File: gpt_engineer/benchmark/types.py

"""
Module defining types used in benchmarking.

This module contains dataclass definitions for various types used throughout the
benchmarking process, such as Assertable, Task, Benchmark, and TaskResult.

Classes:
    Assertable:
        Represents an object that can be asserted against in a benchmark task.

    Assertion:
        Type alias for a callable that takes an Assertable and returns a boolean.

    Task:
        Represents a single task within a benchmark, including its assertions.

    Benchmark:
        Represents a collection of tasks used to evaluate a model's performance.

    TaskResult:
        Represents the result of running a single task within a benchmark.
"""
from dataclasses import dataclass
from subprocess import Popen
from typing import Callable, Dict, Optional

from gpt_engineer.core.base_execution_env import BaseExecutionEnv
from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.prompt import Prompt


@dataclass
class Assertable:
    """
    A class representing an object which can be asserted against.

    Attributes:
        files (FilesDict): The code files involved in the assertion.
        env (BaseExecutionEnv): The execution environment in which the code is run.
        process (Popen): The subprocess in which the code is run.
        stdout (str): The standard output from the code execution.
        stderr (str): The standard error from the code execution.
    """

    files: FilesDict
    env: BaseExecutionEnv
    process: Optional[Popen]
    stdout: Optional[str]
    stderr: Optional[str]


Assertion = Callable[[Assertable], bool]


@dataclass
class Task:
    name: str
    initial_code: Optional[FilesDict]
    command: Optional[str]
    prompt: Prompt
    assertions: Optional[Dict[str, Assertion]]


@dataclass
class Benchmark:
    """A benchmark is a collection of tasks that evaluate a model's performance."""

    name: str
    tasks: list[Task]
    timeout: Optional[int] = None


@dataclass
class TaskResult:
    task_name: str
    assertion_results: dict[str, bool]
    duration: float

    # Returns success rate from 0.00 up to 1.00
    @property
    def success_rate(self) -> float:
        if not self.assertion_results:
            return 0.0

        succeeded = len(
            [result for result in self.assertion_results.values() if result is True]
        )

        return succeeded / len(self.assertion_results)

    def to_dict(self) -> dict:
        out_dict = {key: value for key, value in self.__dict__.items()}
        out_dict["solved"] = self.success_rate
        return out_dict



File: gpt_engineer/benchmark/bench_config.py

from dataclasses import dataclass, field
from pathlib import Path

from tomlkit.items import Integer

from gpt_engineer.core.project_config import read_config


@dataclass
class AppsConfig:
    active: bool | None = True
    test_start_index: int | None = 0
    test_end_index: int | None = 1
    train_start_index: int | None = 0
    train_end_index: int | None = 0
    examples_per_problem: int | None = 10


@dataclass
class MbppConfig:
    active: bool | None = True
    test_len: int | None = 1
    train_len: int | None = 0


@dataclass
class GptmeConfig:
    active: bool | None = True


@dataclass
class BenchConfig:
    """Configuration for the GPT Engineer CLI and gptengineer.app via `gpt-engineer.toml`."""

    apps: AppsConfig = field(default_factory=AppsConfig)
    mbpp: MbppConfig = field(default_factory=MbppConfig)
    gptme: GptmeConfig = field(default_factory=GptmeConfig)

    @classmethod
    def from_toml(cls, config_file: Path | str):
        if isinstance(config_file, str):
            config_file = Path(config_file)
        config_dict = read_config(config_file)
        return cls.from_dict(config_dict)

    @classmethod
    def from_dict(cls, config_dict: dict):
        return cls(
            apps=AppsConfig(**config_dict.get("apps", {})),
            mbpp=MbppConfig(**config_dict.get("mbpp", {})),
            gptme=GptmeConfig(**config_dict.get("gptme", {})),
        )

    @staticmethod
    def recursive_resolve(data_dict):
        for key, value in data_dict.items():
            if isinstance(value, Integer):
                data_dict[key] = int(value)
            elif isinstance(value, dict):
                BenchConfig.recursive_resolve(value)

    def to_dict(self):
        dict_config = {
            benchmark_name: {key: val for key, val in spec_config.__dict__.items()}
            for benchmark_name, spec_config in self.__dict__.items()
        }
        BenchConfig.recursive_resolve(dict_config)

        return dict_config



File: gpt_engineer/benchmark/__main__.py

"""
Main entry point for the benchmarking tool.

This module provides a command-line interface for running benchmarks using Typer.
It allows users to specify the path to an agent, the benchmark(s) to run, and other
options such as verbosity.

Functions
---------
get_agent : function
    Dynamically imports and returns the default configuration agent from the given path.

main : function
    The main function that runs the specified benchmarks with the given agent.
    Outputs the results to the console.

Attributes
----------
__name__ : str
    The standard boilerplate for invoking the main function when the script is executed.
"""
import importlib
import os.path
import sys

from typing import Annotated, Optional

import typer

from langchain.globals import set_llm_cache
from langchain_community.cache import SQLiteCache

from gpt_engineer.applications.cli.main import load_env_if_needed
from gpt_engineer.benchmark.bench_config import BenchConfig
from gpt_engineer.benchmark.benchmarks.load import get_benchmark
from gpt_engineer.benchmark.run import export_yaml_results, print_results, run

app = typer.Typer(
    context_settings={"help_option_names": ["-h", "--help"]}
)  # creates a CLI app


def get_agent(path):
    """
    Dynamically imports and returns the default configuration agent from the given path.

    Parameters
    ----------
    path : str
        The file path to the module containing the default configuration agent.

    Returns
    -------
    BaseAgent
        An instance of the imported default configuration agent.
    """
    # Dynamically import the python module at path
    sys.path.append(os.path.dirname(path))
    agent_module = importlib.import_module(path.replace("/", ".").replace(".py", ""))
    return agent_module.default_config_agent()


@app.command(
    help="""
        Run any benchmark(s) against the specified agent.

        \b
        Currently available benchmarks are: apps and mbpp
    """
)
def main(
    path_to_agent: Annotated[
        str,
        typer.Argument(
            help="python file that contains a function called 'default_config_agent'"
        ),
    ],
    bench_config: Annotated[
        str, typer.Argument(help="optional task name in benchmark")
    ] = os.path.join(os.path.dirname(__file__), "default_bench_config.toml"),
    yaml_output: Annotated[
        Optional[str],
        typer.Option(help="print results for each task", show_default=False),
    ] = None,
    verbose: Annotated[
        Optional[bool],
        typer.Option(help="print results for each task", show_default=False),
    ] = False,
    use_cache: Annotated[
        Optional[bool],
        typer.Option(
            help="Speeds up computations and saves tokens when running the same prompt multiple times by caching the LLM response.",
            show_default=False,
        ),
    ] = True,
):
    """
    The main function that runs the specified benchmarks with the given agent and outputs the results to the console.

    Parameters
    ----------
    path_to_agent : str
        The file path to the Python module that contains a function called 'default_config_agent'.
    bench_config : str, default=default_bench_config.toml
        Configuration file for choosing which benchmark problems to run. See default config for more details.
    yaml_output: Optional[str], default=None
        Pass a path to a yaml file to have results written to file.
    verbose : Optional[bool], default=False
        A flag to indicate whether to print results for each task.
    use_cache : Optional[bool], default=True
        Speeds up computations and saves tokens when running the same prompt multiple times by caching the LLM response.
    Returns
    -------
    None
    """
    if use_cache:
        set_llm_cache(SQLiteCache(database_path=".langchain.db"))
    load_env_if_needed()
    config = BenchConfig.from_toml(bench_config)
    print("using config file: " + bench_config)
    benchmarks = list()
    benchmark_results = dict()
    for specific_config_name in vars(config):
        specific_config = getattr(config, specific_config_name)
        if hasattr(specific_config, "active"):
            if specific_config.active:
                benchmarks.append(specific_config_name)

    for benchmark_name in benchmarks:
        benchmark = get_benchmark(benchmark_name, config)
        if len(benchmark.tasks) == 0:
            print(
                benchmark_name
                + " was skipped, since no tasks are specified. Increase the number of tasks in the config file at: "
                + bench_config
            )
            continue
        agent = get_agent(path_to_agent)

        results = run(agent, benchmark, verbose=verbose)
        print(
            f"\n--- Results for agent {path_to_agent}, benchmark: {benchmark_name} ---"
        )
        print_results(results)
        print()
        benchmark_results[benchmark_name] = {
            "detailed": [result.to_dict() for result in results]
        }
    if yaml_output is not None:
        export_yaml_results(yaml_output, benchmark_results, config.to_dict())


if __name__ == "__main__":
    typer.run(main)



File: gpt_engineer/benchmark/benchmarks/load.py

"""
Module for loading benchmarks.

This module provides a central point to access different benchmarks by name.
It maps benchmark names to their respective loading functions.

Functions
---------
get_benchmark : function
    Retrieves a Benchmark object by name. Raises ValueError if the benchmark is unknown.
"""
from gpt_engineer.benchmark.bench_config import BenchConfig
from gpt_engineer.benchmark.benchmarks.apps.load import load_apps
from gpt_engineer.benchmark.benchmarks.gptme.load import load_gptme
from gpt_engineer.benchmark.benchmarks.mbpp.load import load_mbpp
from gpt_engineer.benchmark.types import Benchmark

BENCHMARKS = {
    "gptme": load_gptme,
    "apps": load_apps,
    "mbpp": load_mbpp,
}


def get_benchmark(name: str, config: BenchConfig) -> Benchmark:
    """
    Retrieves a Benchmark object by name. Raises ValueError if the benchmark is unknown.

    Parameters
    ----------
    name : str
        The name of the benchmark to retrieve.
    config : BenchConfig
        Configuration object for the benchmarks.

    Returns
    -------
    Benchmark
        The Benchmark object corresponding to the given name.

    Raises
    ------
    ValueError
        If the benchmark name is not found in the BENCHMARKS mapping.
    """
    if name not in BENCHMARKS:
        raise ValueError(f"Unknown benchmark {name}.")
    return BENCHMARKS[name](config.__getattribute__(name))



File: gpt_engineer/benchmark/benchmarks/mbpp/problems.py

# TODO: Pick problems
# Temporary testing against these problems
PROBLEM_IDS = range(0, 100)



File: gpt_engineer/benchmark/benchmarks/mbpp/problem.py

from dataclasses import dataclass
from typing import List


@dataclass(frozen=True)
class Problem:
    source_file: int
    task_id: str
    prompt: str
    code: str
    test_imports: str
    test_list: List[str]

    @property
    def starting_code(self) -> str:
        lines: List[str] = []

        for line in self.code.split("\n"):
            lines.append(line)

            if line.startswith("def "):
                lines.append("pass #  TODO: Implement method\n")
                break

        return "\n".join(lines)



File: gpt_engineer/benchmark/benchmarks/mbpp/load.py

"""
Module for loading MBPP evaluation tasks.

This module provides functionality to load tasks for evaluating GPT-based models
on smaller, more focused tasks. It defines a set of tasks with predefined prompts
and assertions to benchmark the performance of AI models.

Functions
---------
load_mbpp : function
    Loads the MBPP benchmark, which consists of a series coding problems.
"""
from pathlib import Path
from subprocess import TimeoutExpired
from typing import Union

from datasets import Dataset, DatasetDict, load_dataset, load_from_disk

from gpt_engineer.benchmark.bench_config import MbppConfig
from gpt_engineer.benchmark.benchmarks.mbpp.problem import Problem
from gpt_engineer.benchmark.types import Assertable, Benchmark, Task
from gpt_engineer.core.default.disk_execution_env import DiskExecutionEnv
from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.prompt import Prompt

DATASET_PATH = Path(__file__).parent / "dataset"


class MbppAssertion:
    def __init__(self, assertion: str):
        self.assertion = assertion

    def evaluate(self, assertable: Assertable) -> bool:
        generated_code = assertable.files["main.py"]
        code_with_assertion = f"{generated_code}\n{self.assertion}"

        # Create new execution environment for every run to avoid side effects
        env = DiskExecutionEnv()
        env.upload(FilesDict({"main.py": code_with_assertion}))
        pro = env.popen("python main.py")

        try:
            stdout, stderr = pro.communicate(timeout=2)
            stdout, stderr = stdout.decode("utf-8"), stderr.decode("utf-8")
        except TimeoutExpired:
            print("Execution Timeout")
            return False

        return not stderr


def _get_dataset() -> Union[Dataset, DatasetDict]:
    try:
        return load_from_disk(str(DATASET_PATH))
    except FileNotFoundError:
        print("Dataset not found locally, downloading...")

    dataset = load_dataset("mbpp", "sanitized", trust_remote_code=True)
    dataset.save_to_disk(str(DATASET_PATH))

    return dataset


def load_mbpp(config: MbppConfig) -> Benchmark:
    """
    Loads the MBPP benchmark, which consists of a series coding problems.

    Returns
    -------
    Benchmark
        A Benchmark object containing a list of Task objects for the MBPP evaluation.
    """
    dataset = _get_dataset()
    tasks = []
    problems = []
    for dataset_type in ["test", "train"]:
        problems += [
            Problem(
                source_file=problem["source_file"],
                task_id=problem["task_id"],
                prompt=problem["prompt"],
                code=problem["code"],
                test_imports=problem["test_imports"],
                test_list=problem["test_list"],
            )
            for index, problem in enumerate(dataset[dataset_type])
            if index < config.__getattribute__(dataset_type + "_len")
        ]

    for problem in problems:
        prompt = Prompt(
            problem.prompt
            + "Please extend given function without changing it's declaration including arguments."
        )

        tasks.append(
            Task(
                name=str(problem.task_id),
                initial_code=FilesDict({"main.py": problem.starting_code}),
                command=None,  # Explicitly setting `None` because each assertion runs code
                prompt=prompt,
                assertions={
                    f"correct assertion {i}": MbppAssertion(
                        assertion=assertion
                    ).evaluate
                    for i, assertion in enumerate(problem.test_list)
                },
            )
        )

    return Benchmark(
        name="mbpp",
        tasks=tasks,
    )



File: gpt_engineer/benchmark/benchmarks/gptme/load.py

"""
Module for loading GPT-Me evaluation tasks.

This module provides functionality to load tasks for evaluating GPT-based models
on smaller, more focused tasks. It defines a set of tasks with predefined prompts
and assertions to benchmark the performance of AI models.

Functions
---------
load_gptme : function
    Loads the GPT-Me benchmark, which consists of a series of tasks for evaluation.
"""
from gpt_engineer.benchmark.bench_config import GptmeConfig
from gpt_engineer.benchmark.types import Benchmark, Task
from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.prompt import Prompt


def load_gptme(config: GptmeConfig) -> Benchmark:
    """
    Loads the GPT-Me benchmark, which consists of a series of tasks for evaluation.

    Returns
    -------
    Benchmark
        A Benchmark object containing a list of Task objects for the GPT-Me evaluation.
    """
    return Benchmark(
        name="gptme",
        tasks=[
            Task(
                name="hello",
                initial_code=FilesDict({"hello.py": "print('Hello, world!')"}),
                command="python hello.py",
                prompt=Prompt("Change the code in hello.py to print 'Hello, human!'"),
                assertions={
                    "correct output": lambda assertable: assertable.stdout
                    == "Hello, human!\n",
                    "correct file": lambda assertable: assertable.files[
                        "hello.py"
                    ].strip()
                    == "print('Hello, human!')",
                },
            ),
            Task(
                name="hello-patch",
                initial_code=FilesDict({"hello.py": "print('Hello, world!')"}),
                command="python hello.py",
                prompt=Prompt("Patch the code in hello.py to print 'Hello, human!'"),
                assertions={
                    "correct output": lambda assertable: assertable.stdout
                    == "Hello, human!\n",
                    "correct file": lambda assertable: assertable.files[
                        "hello.py"
                    ].strip()
                    == "print('Hello, human!')",
                },
            ),
            Task(
                name="hello-ask",
                initial_code=FilesDict({"hello.py": "print('Hello, world!')"}),
                command="echo 'Erik' | python hello.py",
                prompt=Prompt(
                    "modify hello.py to ask the user for their name and print 'Hello, <name>!'. don't try to execute it"
                ),
                assertions={
                    "correct output": lambda assertable: "Hello, Erik!"
                    in assertable.stdout,
                },
            ),
            Task(
                name="prime100",
                initial_code=FilesDict(
                    {}
                ),  # Empty dictionary since no initial code is provided
                command="python prime.py",
                prompt=Prompt(
                    "write a script prime.py that computes and prints the 100th prime number"
                ),
                assertions={
                    "correct output": lambda assertable: "541"
                    in assertable.stdout.split(),
                },
            ),
            Task(
                name="init-git",
                initial_code=FilesDict(
                    {}
                ),  # Empty dictionary since no initial code is provided
                command="git status",
                prompt=Prompt(
                    "initialize a git repository, write a main.py file, and commit it"
                ),
                assertions={
                    "clean exit": lambda assertable: assertable.process.returncode == 0,
                    "clean working tree": lambda assertable: "nothing to commit, working tree clean"
                    in assertable.stdout,
                    "main.py exists": lambda assertable: "main.py" in assertable.files,
                    "we have a commit": lambda assertable: "No commits yet"
                    not in assertable.stdout,
                },
            ),
        ],
    )



File: gpt_engineer/benchmark/benchmarks/apps/problems.py

# TODO: Pick problems
# Temporary testing against these problems
PROBLEM_IDS = list(range(0, 50))



File: gpt_engineer/benchmark/benchmarks/apps/problem.py

import json

from dataclasses import dataclass
from functools import cached_property
from typing import List


@dataclass(frozen=True)
class Problem:
    id: int
    question: str
    input_output: str
    starter_code: str

    @property
    def inputs(self) -> List[str]:
        return self._parsed_inputs_outputs["inputs"]

    @property
    def outputs(self) -> List[str]:
        return self._parsed_inputs_outputs["outputs"]

    @cached_property
    def _parsed_inputs_outputs(self):
        return json.loads(self.input_output.replace("\n", ""))



File: gpt_engineer/benchmark/benchmarks/apps/load.py

"""
Module for loading APPS evaluation tasks.

This module provides functionality to load tasks for evaluating GPT-based models
on smaller, more focused tasks. It defines a set of tasks with predefined prompts
and assertions to benchmark the performance of AI models.

Functions
---------
load_apps : function
    Loads the APPS benchmark, which consists of a series coding problems.
"""
from pathlib import Path
from subprocess import TimeoutExpired
from typing import Union

from datasets import Dataset, DatasetDict, load_dataset, load_from_disk

from gpt_engineer.benchmark.bench_config import AppsConfig
from gpt_engineer.benchmark.benchmarks.apps.problem import Problem
from gpt_engineer.benchmark.types import Assertable, Benchmark, Task
from gpt_engineer.core.default.disk_execution_env import DiskExecutionEnv
from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.prompt import Prompt

DATASET_PATH = Path(__file__).parent / "dataset"


class AppsAssertion:
    def __init__(self, expected: str, command: str):
        self.expected_output = self._format(expected)
        self.command = command

    def evaluate(self, assertable: Assertable) -> bool:
        # Create new execution environment for every run to avoid side effects
        env = DiskExecutionEnv()
        env.upload(assertable.files)
        pro = env.popen(self.command)
        try:
            stdout, stderr = pro.communicate(timeout=2)
            stdout, stderr = stdout.decode("utf-8"), stderr.decode("utf-8")
        except TimeoutExpired:
            print("Execution Timeout")
            return False

        return self.expected_output in self._format(stdout)

    def _format(self, string: str) -> str:
        return string.replace(" ", "").replace("\n", "")


def _get_dataset() -> Union[Dataset, DatasetDict]:
    try:
        return load_from_disk(str(DATASET_PATH))
    except FileNotFoundError:
        print("Dataset not found locally, downloading...")

    dataset = load_dataset("codeparrot/apps", trust_remote_code=True)
    dataset.save_to_disk(str(DATASET_PATH))

    return dataset


def load_apps(config: AppsConfig) -> Benchmark:
    """
    Loads the APPS benchmark, which consists of a series coding problems.

    Returns
    -------
    Benchmark
        A Benchmark object containing a list of Task objects for the APPS evaluation.
    """
    dataset = _get_dataset()
    tasks = []
    problems = list()
    for dataset_type in ["test", "train"]:
        problems += [
            Problem(
                id=problem["problem_id"],
                question=problem["question"],
                input_output=problem["input_output"],
                starter_code=problem["starter_code"],
            )
            for index, problem in enumerate(dataset[dataset_type])
            if (index < config.__getattribute__(dataset_type + "_end_index"))
            and (index >= config.__getattribute__(dataset_type + "_start_index"))
        ]

    for problem in problems:
        prompt = Prompt(
            problem.question
            + "\nThe program, including its inputs, should be run from the command "
            "line like 'python main \"input1 input2 etc \"', with all inputs inside "
            "the quotation marks. The program should not read inputs from stdin."
        )

        tasks.append(
            Task(
                name=str(problem.id),
                initial_code=FilesDict({"main.py": problem.starter_code}),
                command=None,  # Explicitly setting `None` because each assertion specifies its command
                prompt=prompt,
                assertions={
                    f"correct output {i}": AppsAssertion(
                        expected=problem.outputs[i],
                        command="python main.py" + ' "' + problem.inputs[i] + '"',
                    ).evaluate
                    for i in range(
                        min(len(problem.outputs), config.examples_per_problem)
                    )
                },
            )
        )

    return Benchmark(
        name="apps",
        tasks=tasks,
    )



File: gpt_engineer/tools/__init__.py




File: gpt_engineer/tools/supported_languages.py

"""
This module defines the supported programming languages for document chunking.

Variables:
    SUPPORTED_LANGUAGES (list): A list of dictionaries defining supported languages.
"""

SUPPORTED_LANGUAGES = [
    {"name": "Python", "extensions": [".py"], "tree_sitter_name": "python"},
    {
        "name": "JavaScript",
        "extensions": [".js", ".mjs"],
        "tree_sitter_name": "javascript",
    },
    {"name": "HTML", "extensions": [".html", ".htm"], "tree_sitter_name": "html"},
    {"name": "CSS", "extensions": [".css"], "tree_sitter_name": "css"},
    {"name": "Java", "extensions": [".java"], "tree_sitter_name": "java"},
    {"name": "C#", "extensions": [".cs"], "tree_sitter_name": "c_sharp"},
    {
        "name": "TypeScript",
        "extensions": [".ts", ".tsx"],
        "tree_sitter_name": "typescript",
    },
    {"name": "Ruby", "extensions": [".rb", ".erb"], "tree_sitter_name": "ruby"},
    {
        "name": "PHP",
        "extensions": [
            ".php",
            ".phtml",
            ".php3",
            ".php4",
            ".php5",
            ".php7",
            ".phps",
            ".php-s",
            ".pht",
            ".phar",
        ],
        "tree_sitter_name": "php",
    },
    {"name": "Go", "extensions": [".go"], "tree_sitter_name": "go"},
    {"name": "Kotlin", "extensions": [".kt", ".kts"], "tree_sitter_name": "kotlin"},
    {"name": "Rust", "extensions": [".rs"], "tree_sitter_name": "rust"},
    {
        "name": "C++",
        "extensions": [".cpp", ".cc", ".cxx", ".h", ".hpp", ".hxx"],
        "tree_sitter_name": "cpp",
    },
    {"name": "C", "extensions": [".c", ".h"], "tree_sitter_name": "c"},
    {"name": "Markdown", "extensions": [".md"], "tree_sitter_name": "md"},
    {"name": "Arduino C", "extensions": [".ino"], "tree_sitter_name": "ino"}
    # ---- the following are not supported by the current code chunker implementation ----
    # {
    #     "name": "Swift",
    #     "extensions": [".swift"],
    #     "tree_sitter_name": "swift"
    # },
]



File: gpt_engineer/tools/custom_steps.py

from platform import platform
from sys import version_info
from typing import List, Union

from langchain.schema import AIMessage, HumanMessage, SystemMessage

from gpt_engineer.core.ai import AI
from gpt_engineer.core.base_execution_env import BaseExecutionEnv
from gpt_engineer.core.base_memory import BaseMemory
from gpt_engineer.core.chat_to_files import chat_to_files_dict
from gpt_engineer.core.default.paths import CODE_GEN_LOG_FILE, ENTRYPOINT_FILE
from gpt_engineer.core.default.steps import curr_fn, improve_fn, setup_sys_prompt
from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.preprompts_holder import PrepromptsHolder
from gpt_engineer.core.prompt import Prompt

# Type hint for chat messages
Message = Union[AIMessage, HumanMessage, SystemMessage]
MAX_SELF_HEAL_ATTEMPTS = 10


def get_platform_info() -> str:
    """
    Returns a string containing the OS and Python version information.

    This function is used for self-healing by providing information about the current
    operating system and Python version. It assumes that the Python version in the
    virtual environment is the one being used.

    Returns:
        str: A string containing the OS and Python version information.
    """

    v = version_info
    a = f"Python Version: {v.major}.{v.minor}.{v.micro}"
    b = f"\nOS: {platform()}\n"
    return a + b


def self_heal(
    ai: AI,
    execution_env: BaseExecutionEnv,
    files_dict: FilesDict,
    prompt: Prompt = None,
    preprompts_holder: PrepromptsHolder = None,
    memory: BaseMemory = None,
    diff_timeout=3,
) -> FilesDict:
    """
    Attempts to execute the code from the entrypoint and if it fails, sends the error output back to the AI with instructions to fix.

    Parameters
    ----------
    ai : AI
        An instance of the AI model.
    execution_env : BaseExecutionEnv
        The execution environment where the code is run.
    files_dict : FilesDict
        A dictionary of file names to their contents.
    preprompts_holder : PrepromptsHolder, optional
        A holder for preprompt messages.

    Returns
    -------
    FilesDict
        The updated files dictionary after self-healing attempts.

    Raises
    ------
    FileNotFoundError
        If the required entrypoint file does not exist in the code.
    AssertionError
        If the preprompts_holder is None.

    Notes
    -----
    This code will make `MAX_SELF_HEAL_ATTEMPTS` to try and fix the code
    before giving up.
    This makes the assuption that the previous step was `gen_entrypoint`,
    this code could work with `simple_gen`, or `gen_clarified_code` as well.
    """

    # step 1. execute the entrypoint
    # log_path = dbs.workspace.path / "log.txt"
    if ENTRYPOINT_FILE not in files_dict:
        raise FileNotFoundError(
            "The required entrypoint "
            + ENTRYPOINT_FILE
            + " does not exist in the code."
        )

    attempts = 0
    if preprompts_holder is None:
        raise AssertionError("Prepromptsholder required for self-heal")
    while attempts < MAX_SELF_HEAL_ATTEMPTS:
        attempts += 1
        timed_out = False

        # Start the process
        execution_env.upload(files_dict)
        p = execution_env.popen(files_dict[ENTRYPOINT_FILE])

        # Wait for the process to complete and get output
        stdout_full, stderr_full = p.communicate()

        if (p.returncode != 0 and p.returncode != 2) and not timed_out:
            print("run.sh failed.  The log is:")
            print(stdout_full.decode("utf-8"))
            print(stderr_full.decode("utf-8"))

            new_prompt = Prompt(
                f"A program with this specification was requested:\n{prompt}\n, but running it produced the following output:\n{stdout_full}\n and the following errors:\n{stderr_full}. Please change it so that it fulfills the requirements."
            )
            files_dict = improve_fn(
                ai, new_prompt, files_dict, memory, preprompts_holder, diff_timeout
            )
        else:
            break
    return files_dict


def clarified_gen(
    ai: AI, prompt: Prompt, memory: BaseMemory, preprompts_holder: PrepromptsHolder
) -> FilesDict:
    """
    Generates code based on clarifications obtained from the user and saves it to a specified workspace.

    Parameters
    ----------
    ai : AI
        An instance of the AI model, responsible for processing and generating the code.
    prompt : str
        The user's clarification prompt.
    memory : BaseMemory
        The memory instance where the generated code log is saved.
    preprompts_holder : PrepromptsHolder
        A holder for preprompt messages.

    Returns
    -------
    FilesDict
        A dictionary of file names to their contents generated by the AI.
    """

    preprompts = preprompts_holder.get_preprompts()
    messages: List[Message] = [SystemMessage(content=preprompts["clarify"])]
    user_input = prompt.text  # clarify does not work with vision right now
    while True:
        messages = ai.next(messages, user_input, step_name=curr_fn())
        msg = messages[-1].content.strip()

        if "nothing to clarify" in msg.lower():
            break

        if msg.lower().startswith("no"):
            print("Nothing to clarify.")
            break

        print('(answer in text, or "c" to move on)\n')
        user_input = input("")
        print()

        if not user_input or user_input == "c":
            print("(letting gpt-engineer make its own assumptions)")
            print()
            messages = ai.next(
                messages,
                "Make your own assumptions and state them explicitly before starting",
                step_name=curr_fn(),
            )
            print()

        user_input += """
            \n\n
            Is anything else unclear? If yes, ask another question.\n
            Otherwise state: "Nothing to clarify"
            """

    print()

    messages = [
        SystemMessage(content=setup_sys_prompt(preprompts)),
    ] + messages[
        1:
    ]  # skip the first clarify message, which was the original clarify priming prompt
    messages = ai.next(
        messages,
        preprompts["generate"].replace("FILE_FORMAT", preprompts["file_format"]),
        step_name=curr_fn(),
    )
    print()
    chat = messages[-1].content.strip()
    memory.log(CODE_GEN_LOG_FILE, "\n\n".join(x.pretty_repr() for x in messages))
    files_dict = chat_to_files_dict(chat)
    return files_dict


def lite_gen(
    ai: AI, prompt: Prompt, memory: BaseMemory, preprompts_holder: PrepromptsHolder
) -> FilesDict:
    """
    Executes the AI model using the main prompt and saves the generated results to the specified workspace.

    Parameters
    ----------
    ai : AI
        An instance of the AI model.
    prompt : str
        The main prompt to feed to the AI model.
    memory : BaseMemory
        The memory instance where the generated code log is saved.
    preprompts_holder : PrepromptsHolder
        A holder for preprompt messages.

    Returns
    -------
    FilesDict
        A dictionary of file names to their contents generated by the AI.

    Notes
    -----
    The function assumes the `ai.start` method and the `to_files` utility to be correctly
    set up and functional. Ensure these prerequisites before invoking `lite_gen`.
    """

    preprompts = preprompts_holder.get_preprompts()
    messages = ai.start(
        prompt.to_langchain_content(), preprompts["file_format"], step_name=curr_fn()
    )
    chat = messages[-1].content.strip()
    memory.log(CODE_GEN_LOG_FILE, "\n\n".join(x.pretty_repr() for x in messages))
    files_dict = chat_to_files_dict(chat)
    return files_dict



File: gpt_engineer/core/files_dict.py

"""
FilesDict Module

This module provides a FilesDict class which is a dictionary-based container for managing code files.
It extends the standard dictionary to enforce string keys and values, representing filenames and their
corresponding code content. It also provides methods to format its contents for chat-based interaction
with an AI agent and to enforce type checks on keys and values.

Classes:
    FilesDict: A dictionary-based container for managing code files.
"""
from collections import OrderedDict
from pathlib import Path
from typing import Union


# class Code(MutableMapping[str | Path, str]):
# ToDo: implement as mutable mapping, potentially holding a dict instead of being a dict.
class FilesDict(dict):
    """
    A dictionary-based container for managing code files.

    This class extends the standard dictionary to enforce string keys and values,
    representing filenames and their corresponding code content. It provides methods
    to format its contents for chat-based interaction with an AI agent and to enforce
    type checks on keys and values.
    """

    def __setitem__(self, key: Union[str, Path], value: str):
        """
        Set the code content for the given filename, enforcing type checks on the key and value.

        Overrides the dictionary's __setitem__ to enforce type checks on the key and value.
        The key must be a string or a Path object, and the value must be a string representing
        the code content.

        Parameters
        ----------
        key : Union[str, Path]
            The filename as a key for the code content.
        value : str
            The code content to associate with the filename.

        Raises
        ------
        TypeError
            If the key is not a string or Path, or if the value is not a string.
        """
        if not isinstance(key, (str, Path)):
            raise TypeError("Keys must be strings or Path's")
        if not isinstance(value, str):
            raise TypeError("Values must be strings")
        super().__setitem__(key, value)

    def to_chat(self):
        """
        Formats the items of the object (assuming file name and content pairs)
        into a string suitable for chat display.

        Returns
        -------
        str
            A string representation of the files.
        """
        chat_str = ""
        for file_name, file_content in self.items():
            lines_dict = file_to_lines_dict(file_content)
            chat_str += f"File: {file_name}\n"
            for line_number, line_content in lines_dict.items():
                chat_str += f"{line_number} {line_content}\n"
            chat_str += "\n"
        return f"```\n{chat_str}```"

    def to_log(self):
        """
        Formats the items of the object (assuming file name and content pairs)
        into a string suitable for log display.

        Returns
        -------
        str
            A string representation of the files.
        """
        log_str = ""
        for file_name, file_content in self.items():
            log_str += f"File: {file_name}\n"
            log_str += file_content
            log_str += "\n"
        return log_str


def file_to_lines_dict(file_content: str) -> dict:
    """
    Converts file content into a dictionary where each line number is a key
    and the corresponding line content is the value.

    Parameters
    ----------
    file_name : str
        The name of the file.
    file_content : str
        The content of the file.

    Returns
    -------
    dict
        A dictionary with file names as keys and dictionaries (line numbers as keys and line contents as values) as values.
    """
    lines_dict = OrderedDict(
        {
            line_number: line_content
            for line_number, line_content in enumerate(file_content.split("\n"), 1)
        }
    )
    return lines_dict



File: gpt_engineer/core/git.py

import shutil
import subprocess

from pathlib import Path
from typing import List

from gpt_engineer.core.files_dict import FilesDict


def is_git_installed():
    return shutil.which("git") is not None


def is_git_repo(path: Path):
    return (
        subprocess.run(
            ["git", "rev-parse", "--is-inside-work-tree"],
            cwd=path,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        ).returncode
        == 0
    )


def init_git_repo(path: Path):
    subprocess.run(["git", "init"], cwd=path)


def has_uncommitted_changes(path: Path):
    return bool(
        subprocess.run(
            ["git", "diff", "--exit-code"],
            cwd=path,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        ).returncode
    )


def filter_files_with_uncommitted_changes(
    basepath: Path, files_dict: FilesDict
) -> List[Path]:
    files_with_diff = (
        subprocess.run(
            ["git", "diff", "--name-only"], cwd=basepath, stdout=subprocess.PIPE
        )
        .stdout.decode()
        .splitlines()
    )
    return [f for f in files_dict.keys() if f in files_with_diff]


def stage_files(path: Path, files: List[str]):
    subprocess.run(["git", "add", *files], cwd=path)


def filter_by_gitignore(path: Path, file_list: List[str]) -> List[str]:
    out = subprocess.run(
        ["git", "-C", ".", "check-ignore", "--no-index", "--stdin"],
        cwd=path,
        input="\n".join(file_list).encode(),
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    paths = out.stdout.decode().splitlines()
    # return file_list but filter out the results from git check-ignore
    return [f for f in file_list if f not in paths]


def stage_uncommitted_to_git(path, files_dict, improve_mode):
    # Check if there's a git repo and verify that there aren't any uncommitted changes
    if is_git_installed() and not improve_mode:
        if not is_git_repo(path):
            print("\nInitializing an empty git repository")
            init_git_repo(path)

    if is_git_repo(path):
        modified_files = filter_files_with_uncommitted_changes(path, files_dict)
        if modified_files:
            print(
                "Staging the following uncommitted files before overwriting: ",
                ", ".join(modified_files),
            )
            stage_files(path, modified_files)



File: gpt_engineer/core/chat_to_files.py

"""
This Python script provides functionalities for parsing chat transcripts that contain file paths and code blocks,
applying diffs to these files, and parsing unified git diff format strings. The script is designed to work within
a larger system that involves processing and manipulating code files based on chat inputs and diff information.

Key Components:
- chat_to_files_dict: Parses a chat transcript, extracting file paths and associated code blocks, and organizes
  them into a FilesDict object, which is a custom dictionary format designed to hold file contents keyed by their paths.

- apply_diffs: Takes a dictionary of Diff objects (which represent changes to be made to files) and a FilesDict
  object containing the current state of files. It applies the changes described by the Diff objects to the
  corresponding files in the FilesDict, updating the file contents as specified by the diffs.

- parse_diffs: Parses a string containing diffs in the unified git diff format, extracting the changes described
  in the diffs and organizing them into a dictionary of Diff objects, keyed by the filename to which each diff applies.

- parse_diff_block: Parses a single block of text from a diff string, translating it into a Diff object that
  represents the changes described in that block of text.

This script is intended for use in environments where code collaboration or review is conducted through chat interfaces,
allowing for the dynamic application of changes to code bases and the efficient handling of file and diff information in chat transcripts.
"""

import logging
import re

from typing import Dict, Tuple

from regex import regex

from gpt_engineer.core.diff import ADD, REMOVE, RETAIN, Diff, Hunk
from gpt_engineer.core.files_dict import FilesDict, file_to_lines_dict

# Initialize a logger for this module
logger = logging.getLogger(__name__)


def chat_to_files_dict(chat: str) -> FilesDict:
    """
    Converts a chat string containing file paths and code blocks into a FilesDict object.

    Args:
    - chat (str): The chat string containing file paths and code blocks.

    Returns:
    - FilesDict: A dictionary with file paths as keys and code blocks as values.
    """
    # Regex to match file paths and associated code blocks
    regex = r"(\S+)\n\s*```[^\n]*\n(.+?)```"
    matches = re.finditer(regex, chat, re.DOTALL)

    files_dict = FilesDict()
    for match in matches:
        # Clean and standardize the file path
        path = re.sub(r'[\:<>"|?*]', "", match.group(1))
        path = re.sub(r"^\[(.*)\]$", r"\1", path)
        path = re.sub(r"^`(.*)`$", r"\1", path)
        path = re.sub(r"[\]\:]$", "", path)

        # Extract and clean the code content
        content = match.group(2)

        # Add the cleaned path and content to the FilesDict
        files_dict[path.strip()] = content.strip()

    return files_dict


def apply_diffs(diffs: Dict[str, Diff], files: FilesDict) -> FilesDict:
    """
    Applies diffs to the provided files.

    Args:
    - diffs (Dict[str, Diff]): A dictionary of diffs to apply, keyed by filename.
    - files (FilesDict): The original files to which diffs will be applied.

    Returns:
    - FilesDict: The updated files after applying diffs.
    """
    files = FilesDict(files.copy())
    REMOVE_FLAG = "<REMOVE_LINE>"  # Placeholder to mark lines for removal
    for diff in diffs.values():
        if diff.is_new_file():
            # If it's a new file, create it with the content from the diff
            files[diff.filename_post] = "\n".join(
                line[1] for hunk in diff.hunks for line in hunk.lines
            )
        else:
            # Convert the file content to a dictionary of lines
            line_dict = file_to_lines_dict(files[diff.filename_pre])
            for hunk in diff.hunks:
                current_line = hunk.start_line_pre_edit
                for line in hunk.lines:
                    if line[0] == RETAIN:
                        current_line += 1
                    elif line[0] == ADD:
                        # Handle added lines
                        current_line -= 1
                        if (
                            current_line in line_dict.keys()
                            and line_dict[current_line] != REMOVE_FLAG
                        ):
                            line_dict[current_line] += "\n" + line[1]
                        else:
                            line_dict[current_line] = line[1]
                        current_line += 1
                    elif line[0] == REMOVE:
                        # Mark removed lines with REMOVE_FLAG
                        line_dict[current_line] = REMOVE_FLAG
                        current_line += 1

            # Remove lines marked for removal
            line_dict = {
                key: line_content
                for key, line_content in line_dict.items()
                if REMOVE_FLAG not in line_content
            }
            # Reassemble the file content
            files[diff.filename_post] = "\n".join(line_dict.values())
    return files


def parse_diffs(diff_string: str, diff_timeout=3) -> dict:
    """
    Parses a diff string in the unified git diff format.

    Args:
    - diff_string (str): The diff string to parse.

    Returns:
    - dict: A dictionary of Diff objects keyed by filename.
    """
    # Regex to match individual diff blocks
    diff_block_pattern = regex.compile(
        r"```.*?\n\s*?--- .*?\n\s*?\+\+\+ .*?\n(?:@@ .*? @@\n(?:[-+ ].*?\n)*?)*?```",
        re.DOTALL,
    )

    diffs = {}
    try:
        for block in diff_block_pattern.finditer(diff_string, timeout=diff_timeout):
            diff_block = block.group()

            # Parse individual diff blocks and update the diffs dictionary
            diff = parse_diff_block(diff_block)
            for filename, diff_obj in diff.items():
                if filename not in diffs:
                    diffs[filename] = diff_obj
                else:
                    print(
                        f"\nMultiple diffs found for {filename}. Only the first one is kept."
                    )
    except TimeoutError:
        print("gpt-engineer timed out while parsing git diff")

    if not diffs:
        print(
            "GPT did not provide any proposed changes. Please try to reselect the files for uploading and edit your prompt file."
        )

    return diffs


def parse_diff_block(diff_block: str) -> dict:
    """
    Parses a block of diff text into a Diff object.

    Args:
    - diff_block (str): A single block of diff text.

    Returns:
    - dict: A dictionary containing a single Diff object keyed by the post-edit filename.
    """
    lines = diff_block.strip().split("\n")[1:-1]  # Exclude the opening and closing ```
    diffs = {}
    current_diff = None
    hunk_lines = []
    filename_pre = None
    filename_post = None
    hunk_header = None

    for line in lines:
        if line.startswith("--- "):
            # Pre-edit filename
            filename_pre = line[4:]
        elif line.startswith("+++ "):
            # Post-edit filename and initiation of a new Diff object
            if (
                filename_post is not None
                and current_diff is not None
                and hunk_header is not None
            ):
                current_diff.hunks.append(Hunk(*hunk_header, hunk_lines))
                hunk_lines = []
            filename_post = line[4:]
            current_diff = Diff(filename_pre, filename_post)
            diffs[filename_post] = current_diff
        elif line.startswith("@@ "):
            # Start of a new hunk in the diff
            if hunk_lines and current_diff is not None and hunk_header is not None:
                current_diff.hunks.append(Hunk(*hunk_header, hunk_lines))
                hunk_lines = []
            hunk_header = parse_hunk_header(line)
        elif line.startswith("+"):
            # Added line
            hunk_lines.append((ADD, line[1:]))
        elif line.startswith("-"):
            # Removed line
            hunk_lines.append((REMOVE, line[1:]))
        else:
            # Retained line
            hunk_lines.append((RETAIN, line[1:]))

    # Append the last hunk if any
    if current_diff is not None and hunk_lines and hunk_header is not None:
        current_diff.hunks.append(Hunk(*hunk_header, hunk_lines))

    return diffs


def parse_hunk_header(header_line) -> Tuple[int, int, int, int]:
    """
    Parses the header of a hunk from a diff.

    Args:
    - header_line (str): The header line of a hunk.

    Returns:
    - tuple: A tuple containing start and length information for pre- and post-edit.
    """
    pattern = re.compile(r"^@@ -\d{1,},\d{1,} \+\d{1,},\d{1,} @@$")

    if not pattern.match(header_line):
        # Return a default value if the header does not match the expected format
        return 0, 0, 0, 0

    pre, post = header_line.split(" ")[1:3]
    start_line_pre_edit, hunk_len_pre_edit = map(int, pre[1:].split(","))
    start_line_post_edit, hunk_len_post_edit = map(int, post[1:].split(","))
    return (
        start_line_pre_edit,
        hunk_len_pre_edit,
        start_line_post_edit,
        hunk_len_post_edit,
    )



File: gpt_engineer/core/version_manager.py

"""
Version Manager Module

This module provides an abstract base class for a version manager that handles the creation of snapshots
for code. Implementations of this class are expected to provide methods to create a snapshot of the given
code and return a reference to it.
"""
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Union

from gpt_engineer.core.files_dict import FilesDict


class BaseVersionManager(ABC):
    """
    Abstract base class for a version manager.

    Defines the interface for version managers that handle the creation of snapshots for code.
    Implementations of this class are expected to provide methods to create a snapshot of the given
    code and return a reference to it.
    """

    @abstractmethod
    def __init__(self, path: Union[str, Path]):
        pass

    @abstractmethod
    def snapshot(self, files_dict: FilesDict) -> str:
        pass



File: gpt_engineer/core/base_memory.py

"""
Base Memory Module

This module provides a type alias for a mutable mapping that represents the base memory structure
used in the GPT Engineer project. The base memory is a mapping from file names (as strings or Path objects)
to their corresponding code content (as strings).

Type Aliases:
    BaseMemory: A mutable mapping from file names to code content.
"""

from pathlib import Path
from typing import MutableMapping, Union

BaseMemory = MutableMapping[Union[str, Path], str]



File: gpt_engineer/core/base_agent.py

"""
Base Agent Module

This module provides an abstract base class for an agent that interacts with code. It defines the interface
for agents capable of initializing and improving code based on a given prompt. Implementations of this class
are expected to provide concrete methods for these actions.

Classes:
    BaseAgent: Abstract base class for an agent that interacts with code.
"""
from abc import ABC, abstractmethod

from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.prompt import Prompt


class BaseAgent(ABC):
    """
    Abstract base class for an agent that interacts with code.

    Defines the interface for agents capable of initializing and improving code based on a given prompt.
    Implementations of this class are expected to provide concrete methods for these actions.
    """

    @abstractmethod
    def init(self, prompt: Prompt) -> FilesDict:
        pass

    @abstractmethod
    def improve(self, files_dict: FilesDict, prompt: Prompt) -> FilesDict:
        pass



File: gpt_engineer/core/ai.py

"""
AI Module

This module provides an AI class that interfaces with language models to perform various tasks such as
starting a conversation, advancing the conversation, and handling message serialization. It also includes
backoff strategies for handling rate limit errors from the OpenAI API.

Classes:
    AI: A class that interfaces with language models for conversation management and message serialization.

Functions:
    serialize_messages(messages: List[Message]) -> str
        Serialize a list of messages to a JSON string.
"""

from __future__ import annotations

import json
import logging
import os

from pathlib import Path
from typing import Any, List, Optional, Union

import backoff
import openai
import pyperclip

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain.chat_models.base import BaseChatModel
from langchain.schema import (
    AIMessage,
    HumanMessage,
    SystemMessage,
    messages_from_dict,
    messages_to_dict,
)
from langchain_anthropic import ChatAnthropic
from langchain_openai import AzureChatOpenAI, ChatOpenAI

from gpt_engineer.core.token_usage import TokenUsageLog

# Type hint for a chat message
Message = Union[AIMessage, HumanMessage, SystemMessage]

# Set up logging
logger = logging.getLogger(__name__)


class AI:
    """
    A class that interfaces with language models for conversation management and message serialization.

    This class provides methods to start and advance conversations, handle message serialization,
    and implement backoff strategies for rate limit errors when interacting with the OpenAI API.

    Attributes
    ----------
    temperature : float
        The temperature setting for the language model.
    azure_endpoint : str
        The endpoint URL for the Azure-hosted language model.
    model_name : str
        The name of the language model to use.
    streaming : bool
        A flag indicating whether to use streaming for the language model.
    llm : BaseChatModel
        The language model instance for conversation management.
    token_usage_log : TokenUsageLog
        A log for tracking token usage during conversations.

    Methods
    -------
    start(system: str, user: str, step_name: str) -> List[Message]
        Start the conversation with a system message and a user message.
    next(messages: List[Message], prompt: Optional[str], step_name: str) -> List[Message]
        Advances the conversation by sending message history to LLM and updating with the response.
    backoff_inference(messages: List[Message]) -> Any
        Perform inference using the language model with an exponential backoff strategy.
    serialize_messages(messages: List[Message]) -> str
        Serialize a list of messages to a JSON string.
    deserialize_messages(jsondictstr: str) -> List[Message]
        Deserialize a JSON string to a list of messages.
    _create_chat_model() -> BaseChatModel
        Create a chat model with the specified model name and temperature.
    """

    def __init__(
        self,
        model_name="gpt-4-turbo",
        temperature=0.1,
        azure_endpoint=None,
        streaming=True,
        vision=False,
    ):
        """
        Initialize the AI class.

        Parameters
        ----------
        model_name : str, optional
            The name of the model to use, by default "gpt-4".
        temperature : float, optional
            The temperature to use for the model, by default 0.1.
        """
        self.temperature = temperature
        self.azure_endpoint = azure_endpoint
        self.model_name = model_name
        self.streaming = streaming
        self.vision = (
            ("vision-preview" in model_name)
            or ("gpt-4-turbo" in model_name and "preview" not in model_name)
            or ("claude" in model_name)
        )
        self.llm = self._create_chat_model()
        self.token_usage_log = TokenUsageLog(model_name)

        logger.debug(f"Using model {self.model_name}")

    def start(self, system: str, user: Any, *, step_name: str) -> List[Message]:
        """
        Start the conversation with a system message and a user message.

        Parameters
        ----------
        system : str
            The content of the system message.
        user : str
            The content of the user message.
        step_name : str
            The name of the step.

        Returns
        -------
        List[Message]
            The list of messages in the conversation.
        """

        messages: List[Message] = [
            SystemMessage(content=system),
            HumanMessage(content=user),
        ]
        return self.next(messages, step_name=step_name)

    def _extract_content(self, content):
        """
        Extracts text content from a message, supporting both string and list types.
        Parameters
        ----------
        content : Union[str, List[dict]]
            The content of a message, which could be a string or a list.
        Returns
        -------
        str
            The extracted text content.
        """
        if isinstance(content, str):
            return content
        elif isinstance(content, list) and content and "text" in content[0]:
            # Assuming the structure of list content is [{'type': 'text', 'text': 'Some text'}, ...]
            return content[0]["text"]
        else:
            return ""

    def _collapse_text_messages(self, messages: List[Message]):
        """
        Combine consecutive messages of the same type into a single message, where if the message content
        is a list type, the first text element's content is taken. This method keeps `combined_content` as a string.

        This method iterates through the list of messages, combining consecutive messages of the same type
        by joining their content with a newline character. If the content is a list, it extracts text from the first
        text element's content. This reduces the number of messages and simplifies the conversation for processing.

        Parameters
        ----------
        messages : List[Message]
            The list of messages to collapse.

        Returns
        -------
        List[Message]
            The list of messages after collapsing consecutive messages of the same type.
        """
        collapsed_messages = []
        if not messages:
            return collapsed_messages

        previous_message = messages[0]
        combined_content = self._extract_content(previous_message.content)

        for current_message in messages[1:]:
            if current_message.type == previous_message.type:
                combined_content += "\n\n" + self._extract_content(
                    current_message.content
                )
            else:
                collapsed_messages.append(
                    previous_message.__class__(content=combined_content)
                )
                previous_message = current_message
                combined_content = self._extract_content(current_message.content)

        collapsed_messages.append(previous_message.__class__(content=combined_content))
        return collapsed_messages

    def next(
        self,
        messages: List[Message],
        prompt: Optional[str] = None,
        *,
        step_name: str,
    ) -> List[Message]:
        """
        Advances the conversation by sending message history
        to LLM and updating with the response.

        Parameters
        ----------
        messages : List[Message]
            The list of messages in the conversation.
        prompt : Optional[str], optional
            The prompt to use, by default None.
        step_name : str
            The name of the step.

        Returns
        -------
        List[Message]
            The updated list of messages in the conversation.
        """

        if prompt:
            messages.append(HumanMessage(content=prompt))

        logger.debug(
            "Creating a new chat completion: %s",
            "\n".join([m.pretty_repr() for m in messages]),
        )

        if not self.vision:
            messages = self._collapse_text_messages(messages)

        response = self.backoff_inference(messages)

        self.token_usage_log.update_log(
            messages=messages, answer=response.content, step_name=step_name
        )
        messages.append(response)
        logger.debug(f"Chat completion finished: {messages}")

        return messages

    @backoff.on_exception(backoff.expo, openai.RateLimitError, max_tries=7, max_time=45)
    def backoff_inference(self, messages):
        """
        Perform inference using the language model while implementing an exponential backoff strategy.

        This function will retry the inference in case of a rate limit error from the OpenAI API.
        It uses an exponential backoff strategy, meaning the wait time between retries increases
        exponentially. The function will attempt to retry up to 7 times within a span of 45 seconds.

        Parameters
        ----------
        messages : List[Message]
            A list of chat messages which will be passed to the language model for processing.

        callbacks : List[Callable]
            A list of callback functions that are triggered after each inference. These functions
            can be used for logging, monitoring, or other auxiliary tasks.

        Returns
        -------
        Any
            The output from the language model after processing the provided messages.

        Raises
        ------
        openai.error.RateLimitError
            If the number of retries exceeds the maximum or if the rate limit persists beyond the
            allotted time, the function will ultimately raise a RateLimitError.

        Example
        -------
        >>> messages = [SystemMessage(content="Hello"), HumanMessage(content="How's the weather?")]
        >>> response = backoff_inference(messages)
        """
        return self.llm.invoke(messages)  # type: ignore

    @staticmethod
    def serialize_messages(messages: List[Message]) -> str:
        """
        Serialize a list of messages to a JSON string.

        Parameters
        ----------
        messages : List[Message]
            The list of messages to serialize.

        Returns
        -------
        str
            The serialized messages as a JSON string.
        """
        return json.dumps(messages_to_dict(messages))

    @staticmethod
    def deserialize_messages(jsondictstr: str) -> List[Message]:
        """
        Deserialize a JSON string to a list of messages.

        Parameters
        ----------
        jsondictstr : str
            The JSON string to deserialize.

        Returns
        -------
        List[Message]
            The deserialized list of messages.
        """
        data = json.loads(jsondictstr)
        # Modify implicit is_chunk property to ALWAYS false
        # since Langchain's Message schema is stricter
        prevalidated_data = [
            {**item, "tools": {**item.get("tools", {}), "is_chunk": False}}
            for item in data
        ]
        return list(messages_from_dict(prevalidated_data))  # type: ignore

    def _create_chat_model(self) -> BaseChatModel:
        """
        Create a chat model with the specified model name and temperature.

        Parameters
        ----------
        model : str
            The name of the model to create.
        temperature : float
            The temperature to use for the model.

        Returns
        -------
        BaseChatModel
            The created chat model.
        """
        if self.azure_endpoint:
            return AzureChatOpenAI(
                azure_endpoint=self.azure_endpoint,
                openai_api_version=os.getenv(
                    "OPENAI_API_VERSION", "2024-05-01-preview"
                ),
                deployment_name=self.model_name,
                openai_api_type="azure",
                streaming=self.streaming,
                callbacks=[StreamingStdOutCallbackHandler()],
            )
        elif "claude" in self.model_name:
            return ChatAnthropic(
                model=self.model_name,
                temperature=self.temperature,
                callbacks=[StreamingStdOutCallbackHandler()],
                streaming=self.streaming,
                max_tokens_to_sample=4096,
            )
        elif self.vision:
            return ChatOpenAI(
                model=self.model_name,
                temperature=self.temperature,
                streaming=self.streaming,
                callbacks=[StreamingStdOutCallbackHandler()],
                max_tokens=4096,  # vision models default to low max token limits
            )
        else:
            return ChatOpenAI(
                model=self.model_name,
                temperature=self.temperature,
                streaming=self.streaming,
                callbacks=[StreamingStdOutCallbackHandler()],
            )


def serialize_messages(messages: List[Message]) -> str:
    return AI.serialize_messages(messages)


class ClipboardAI(AI):
    # Ignore not init superclass
    def __init__(self, **_):  # type: ignore
        self.vision = False
        self.token_usage_log = TokenUsageLog("clipboard_llm")

    @staticmethod
    def serialize_messages(messages: List[Message]) -> str:
        return "\n\n".join([f"{m.type}:\n{m.content}" for m in messages])

    @staticmethod
    def multiline_input():
        print("Enter/Paste your content. Ctrl-D or Ctrl-Z ( windows ) to save it.")
        content = []
        while True:
            try:
                line = input()
            except EOFError:
                break
            content.append(line)
        return "\n".join(content)

    def next(
        self,
        messages: List[Message],
        prompt: Optional[str] = None,
        *,
        step_name: str,
    ) -> List[Message]:
        """
        Not yet fully supported
        """
        if prompt:
            messages.append(HumanMessage(content=prompt))

        logger.debug(f"Creating a new chat completion: {messages}")

        msgs = self.serialize_messages(messages)
        pyperclip.copy(msgs)
        Path("clipboard.txt").write_text(msgs)
        print(
            "Messages copied to clipboard and written to clipboard.txt,",
            len(msgs),
            "characters in total",
        )

        response = self.multiline_input()

        messages.append(AIMessage(content=response))
        logger.debug(f"Chat completion finished: {messages}")

        return messages



File: gpt_engineer/core/__init__.py




File: gpt_engineer/core/project_config.py

"""
Functions for reading and writing the `gpt-engineer.toml` configuration file.

The `gpt-engineer.toml` file is a TOML file that contains project-specific configuration used by the GPT Engineer CLI and gptengineer.app.
"""
from dataclasses import asdict, dataclass, field
from pathlib import Path

import tomlkit

default_config_filename = "gpt-engineer.toml"

example_config = """
[run]
build = "npm run build"
test = "npm run test"
lint = "quick-lint-js"

[paths]
base = "./frontend"  # base directory to operate in (for monorepos)
src = "./src"        # source directory (under the base directory) from which context will be retrieved

[gptengineer-app]  # this namespace is used for gptengineer.app, may be used for internal experiments
project_id = "..."

# we support multiple OpenAPI schemas, used as context for the LLM
openapi = [
    { url = "https://api.gptengineer.app/openapi.json" },
    { url = "https://some-color-translating-api/openapi.json" },
]
"""


@dataclass
class _PathsConfig:
    base: str | None = None
    src: str | None = None


@dataclass
class _RunConfig:
    build: str | None = None
    test: str | None = None
    lint: str | None = None
    format: str | None = None


@dataclass
class _OpenApiConfig:
    url: str


@dataclass
class _GptEngineerAppConfig:
    project_id: str
    openapi: list[_OpenApiConfig] | None = None


def filter_none(d: dict) -> dict:
    # Drop None values and empty dictionaries from a dictionary
    return {
        k: v
        for k, v in (
            (k, filter_none(v) if isinstance(v, dict) else v)
            for k, v in d.items()
            if v is not None
        )
        if not (isinstance(v, dict) and not v)  # Check for non-empty after filtering
    }


@dataclass
class Config:
    """Configuration for the GPT Engineer CLI and gptengineer.app via `gpt-engineer.toml`."""

    paths: _PathsConfig = field(default_factory=_PathsConfig)
    run: _RunConfig = field(default_factory=_RunConfig)
    gptengineer_app: _GptEngineerAppConfig | None = None

    @classmethod
    def from_toml(cls, config_file: Path | str):
        if isinstance(config_file, str):
            config_file = Path(config_file)
        config_dict = read_config(config_file)
        return cls.from_dict(config_dict)

    @classmethod
    def from_dict(cls, config_dict: dict):
        run = _RunConfig(**config_dict.get("run", {}))
        paths = _PathsConfig(**config_dict.get("paths", {}))

        # load optional gptengineer-app section
        gptengineer_app_dict = config_dict.get("gptengineer-app", {})
        gptengineer_app = None
        if gptengineer_app_dict:
            assert (
                "project_id" in gptengineer_app_dict
            ), "project_id is required in gptengineer-app section"
            gptengineer_app = _GptEngineerAppConfig(
                # required if gptengineer-app section is present
                project_id=gptengineer_app_dict["project_id"],
                openapi=[
                    _OpenApiConfig(**openapi)
                    for openapi in gptengineer_app_dict.get("openapi", [])
                ]
                or None,
            )

        return cls(paths=paths, run=run, gptengineer_app=gptengineer_app)

    def to_dict(self) -> dict:
        d = asdict(self)
        d["gptengineer-app"] = d.pop("gptengineer_app", None)

        # Drop None values and empty dictionaries
        # Needed because tomlkit.dumps() doesn't handle None values,
        # and we don't want to write empty sections.
        d = filter_none(d)

        return d

    def to_toml(self, config_file: Path | str, save=True) -> str:
        """Write the configuration to a TOML file."""
        if isinstance(config_file, str):
            config_file = Path(config_file)

        # Load the TOMLDocument and overwrite it with the new values
        config = read_config(config_file)
        default_config = Config().to_dict()
        for k, v in self.to_dict().items():
            # only write values that are already explicitly set, or that differ from defaults
            if k in config or v != default_config[k]:
                if isinstance(v, dict):
                    config[k] = {
                        k2: v2
                        for k2, v2 in v.items()
                        if (
                            k2 in config[k]
                            or default_config.get(k) is None
                            or v2 != default_config[k].get(k2)
                        )
                    }
                else:
                    config[k] = v

        toml_str = tomlkit.dumps(config)
        if save:
            with open(config_file, "w") as f:
                f.write(toml_str)

        return toml_str


def read_config(config_file: Path) -> tomlkit.TOMLDocument:
    """Read the configuration file"""
    assert config_file.exists(), f"Config file {config_file} does not exist"
    with open(config_file, "r") as f:
        return tomlkit.load(f)



File: gpt_engineer/core/linting.py

import black

from gpt_engineer.core.files_dict import FilesDict


class Linting:
    def __init__(self):
        # Dictionary to hold linting methods for different file types
        self.linters = {".py": self.lint_python}

    import black

    def lint_python(self, content, config):
        """Lint Python files using the `black` library, handling all exceptions silently and logging them.
        This function attempts to format the code and returns the formatted code if successful.
        If any error occurs during formatting, it logs the error and returns the original content.
        """
        try:
            # Try to format the content using black
            linted_content = black.format_str(content, mode=black.FileMode(**config))
        except black.NothingChanged:
            # If nothing changed, log the info and return the original content
            print("\nInfo: No changes were made during formatting.\n")
            linted_content = content
        except Exception as error:
            # If any other exception occurs, log the error and return the original content
            print(f"\nError: Could not format due to {error}\n")
            linted_content = content
        return linted_content

    def lint_files(self, files_dict: FilesDict, config: dict = None) -> FilesDict:
        """
        Lints files based on their extension using registered linting functions.

        Parameters
        ----------
        files_dict : FilesDict
            The dictionary of file names to their respective source code content.
        config : dict, optional
            A dictionary of configuration options for the linting tools.

        Returns
        -------
        FilesDict
            The dictionary of file names to their respective source code content after linting.
        """
        if config is None:
            config = {}

        for filename, content in files_dict.items():
            extension = filename[
                filename.rfind(".") :
            ].lower()  # Ensure case insensitivity
            if extension in self.linters:
                original_content = content
                linted_content = self.linters[extension](content, config)
                if linted_content != original_content:
                    print(f"Linted {filename}.")
                else:
                    print(f"No changes made for {filename}.")
                files_dict[filename] = linted_content
            else:
                print(f"No linter registered for {filename}.")
        return files_dict



File: gpt_engineer/core/preprompts_holder.py

from pathlib import Path
from typing import Dict

from gpt_engineer.core.default.disk_memory import DiskMemory


class PrepromptsHolder:
    """
    A holder for preprompt texts that are stored on disk.

    This class provides methods to retrieve preprompt texts from a specified directory.

    Attributes
    ----------
    preprompts_path : Path
        The file path to the directory containing preprompt texts.

    Methods
    -------
    get_preprompts() -> Dict[str, str]
        Retrieve all preprompt texts from the directory and return them as a dictionary.
    """

    def __init__(self, preprompts_path: Path):
        self.preprompts_path = preprompts_path

    def get_preprompts(self) -> Dict[str, str]:
        preprompts_repo = DiskMemory(self.preprompts_path)
        return {file_name: preprompts_repo[file_name] for file_name in preprompts_repo}



File: gpt_engineer/core/token_usage.py

import base64
import io
import logging
import math

from dataclasses import dataclass
from typing import List, Union

import tiktoken

from langchain.schema import AIMessage, HumanMessage, SystemMessage
from PIL import Image

# workaround for function moved in:
# https://github.com/langchain-ai/langchain/blob/535db72607c4ae308566ede4af65295967bb33a8/libs/community/langchain_community/callbacks/openai_info.py
try:
    from langchain.callbacks.openai_info import (
        get_openai_token_cost_for_model,  # fmt: skip
    )
except ImportError:
    from langchain_community.callbacks.openai_info import (
        get_openai_token_cost_for_model,  # fmt: skip
    )


Message = Union[AIMessage, HumanMessage, SystemMessage]

logger = logging.getLogger(__name__)


@dataclass
class TokenUsage:
    """
    Dataclass representing token usage statistics for a conversation step.

    Attributes
    ----------
    step_name : str
        The name of the conversation step.
    in_step_prompt_tokens : int
        The number of prompt tokens used in the step.
    in_step_completion_tokens : int
        The number of completion tokens used in the step.
    in_step_total_tokens : int
        The total number of tokens used in the step.
    total_prompt_tokens : int
        The cumulative number of prompt tokens used up to this step.
    total_completion_tokens : int
        The cumulative number of completion tokens used up to this step.
    total_tokens : int
        The cumulative total number of tokens used up to this step.
    """

    """
    Represents token usage statistics for a conversation step.
    """

    step_name: str
    in_step_prompt_tokens: int
    in_step_completion_tokens: int
    in_step_total_tokens: int
    total_prompt_tokens: int
    total_completion_tokens: int
    total_tokens: int


class Tokenizer:
    """
    Tokenizer for counting tokens in text.
    """

    def __init__(self, model_name):
        self.model_name = model_name
        self._tiktoken_tokenizer = (
            tiktoken.encoding_for_model(model_name)
            if "gpt-4" in model_name or "gpt-3.5" in model_name
            else tiktoken.get_encoding("cl100k_base")
        )

    def num_tokens(self, txt: str) -> int:
        """
        Get the number of tokens in a text.

        Parameters
        ----------
        txt : str
            The text to count the tokens in.

        Returns
        -------
        int
            The number of tokens in the text.
        """
        return len(self._tiktoken_tokenizer.encode(txt))

    def num_tokens_for_base64_image(
        self, image_base64: str, detail: str = "high"
    ) -> int:
        """
        Calculate the token size for a base64 encoded image based on OpenAI's token calculation rules.

        Parameters:
        - image_base64 (str): The base64 encoded string of the image.
        - detail (str): The detail level of the image, 'low' or 'high'.

        Returns:
        - int: The token size of the image.
        """

        if detail == "low":
            return 85  # Fixed cost for low detail images

        # Decode image from base64
        image_data = base64.b64decode(image_base64)

        # Convert byte data to image for size extraction
        image = Image.open(io.BytesIO(image_data))

        # Calculate the initial scale to fit within 2048 square while maintaining aspect ratio
        max_dimension = max(image.size)
        scale_factor = min(2048 / max_dimension, 1)  # Ensure we don't scale up
        new_width = int(image.size[0] * scale_factor)
        new_height = int(image.size[1] * scale_factor)

        # Scale such that the shortest side is 768px
        shortest_side = min(new_width, new_height)
        if shortest_side > 768:
            resize_factor = 768 / shortest_side
            new_width = int(new_width * resize_factor)
            new_height = int(new_height * resize_factor)

        # Calculate the number of 512px tiles needed
        width_tiles = math.ceil(new_width / 512)
        height_tiles = math.ceil(new_height / 512)
        total_tiles = width_tiles * height_tiles

        # Each tile costs 170 tokens, plus a base cost of 85 tokens for high detail
        token_cost = total_tiles * 170 + 85

        return token_cost

    def num_tokens_from_messages(self, messages: List[Message]) -> int:
        """
        Get the total number of tokens used by a list of messages, accounting for text and base64 encoded images.

        Parameters
        ----------
        messages : List[Message]
            The list of messages to count the tokens in.

        Returns
        -------
        int
            The total number of tokens used by the messages.
        """
        n_tokens = 0
        for message in messages:
            n_tokens += 4  # Account for message framing tokens

            if isinstance(message.content, str):
                # Content is a simple string
                n_tokens += self.num_tokens(message.content)
            elif isinstance(message.content, list):
                # Content is a list, potentially mixed with text and images
                for item in message.content:
                    if item.get("type") == "text":
                        n_tokens += self.num_tokens(item["text"])
                    elif item.get("type") == "image_url":
                        image_detail = item["image_url"].get("detail", "high")
                        image_base64 = item["image_url"].get("url")
                        n_tokens += self.num_tokens_for_base64_image(
                            image_base64, detail=image_detail
                        )

            n_tokens += 2  # Account for assistant's reply framing tokens

        return n_tokens


class TokenUsageLog:
    """
    Represents a log of token usage statistics for a conversation.
    """

    def __init__(self, model_name):
        self.model_name = model_name
        self._cumulative_prompt_tokens = 0
        self._cumulative_completion_tokens = 0
        self._cumulative_total_tokens = 0
        self._log = []
        self._tokenizer = Tokenizer(model_name)

    def update_log(self, messages: List[Message], answer: str, step_name: str) -> None:
        """
        Update the token usage log with the number of tokens used in the current step.

        Parameters
        ----------
        messages : List[Message]
            The list of messages in the conversation.
        answer : str
            The answer from the AI.
        step_name : str
            The name of the step.
        """
        prompt_tokens = self._tokenizer.num_tokens_from_messages(messages)
        completion_tokens = self._tokenizer.num_tokens(answer)
        total_tokens = prompt_tokens + completion_tokens

        self._cumulative_prompt_tokens += prompt_tokens
        self._cumulative_completion_tokens += completion_tokens
        self._cumulative_total_tokens += total_tokens

        self._log.append(
            TokenUsage(
                step_name=step_name,
                in_step_prompt_tokens=prompt_tokens,
                in_step_completion_tokens=completion_tokens,
                in_step_total_tokens=total_tokens,
                total_prompt_tokens=self._cumulative_prompt_tokens,
                total_completion_tokens=self._cumulative_completion_tokens,
                total_tokens=self._cumulative_total_tokens,
            )
        )

    def log(self) -> List[TokenUsage]:
        """
        Get the token usage log.

        Returns
        -------
        List[TokenUsage]
            A log of token usage details per step in the conversation.
        """
        return self._log

    def format_log(self) -> str:
        """
        Format the token usage log as a CSV string.

        Returns
        -------
        str
            The token usage log formatted as a CSV string.
        """
        result = "step_name,prompt_tokens_in_step,completion_tokens_in_step,total_tokens_in_step,total_prompt_tokens,total_completion_tokens,total_tokens\n"
        for log in self._log:
            result += f"{log.step_name},{log.in_step_prompt_tokens},{log.in_step_completion_tokens},{log.in_step_total_tokens},{log.total_prompt_tokens},{log.total_completion_tokens},{log.total_tokens}\n"
        return result

    def is_openai_model(self) -> bool:
        """
        Check if the model is an OpenAI model.

        Returns
        -------
        bool
            True if the model is an OpenAI model, False otherwise.
        """
        return "gpt" in self.model_name.lower()

    def total_tokens(self) -> int:
        """
        Return the total number of tokens used in the conversation.

        Returns
        -------
        int
            The total number of tokens used in the conversation.
        """
        return self._cumulative_total_tokens

    def usage_cost(self) -> float | None:
        """
        Return the total cost in USD of the API usage.

        Returns
        -------
        float
            Cost in USD.
        """
        if not self.is_openai_model():
            return None

        try:
            result = 0
            for log in self.log():
                result += get_openai_token_cost_for_model(
                    self.model_name, log.total_prompt_tokens, is_completion=False
                )
                result += get_openai_token_cost_for_model(
                    self.model_name, log.total_completion_tokens, is_completion=True
                )
            return result
        except Exception as e:
            print(f"Error calculating usage cost: {e}")
            return None



File: gpt_engineer/core/base_execution_env.py

from abc import ABC, abstractmethod
from subprocess import Popen
from typing import Optional, Tuple

from gpt_engineer.core.files_dict import FilesDict


class BaseExecutionEnv(ABC):
    """
    Abstract base class for an execution environment capable of running code.

    This class defines the interface for execution environments that can execute commands,
    handle processes, and manage file uploads and downloads.
    """

    @abstractmethod
    def run(self, command: str, timeout: Optional[int] = None) -> Tuple[str, str, int]:
        """
        Runs a command in the execution environment.
        """
        raise NotImplementedError

    @abstractmethod
    def popen(self, command: str) -> Popen:
        """
        Runs a command in the execution environment.
        """
        raise NotImplementedError

    @abstractmethod
    def upload(self, files: FilesDict) -> "BaseExecutionEnv":
        """
        Uploads files to the execution environment.
        """
        raise NotImplementedError

    @abstractmethod
    def download(self) -> FilesDict:
        """
        Downloads files from the execution environment.
        """
        raise NotImplementedError



File: gpt_engineer/core/diff.py

"""
File Overview:

This Python module is designed for processing and analyzing diffs in source code files. Diffs represent the changes between two versions of a file, which are crucial in version control systems for tracking file modifications. The module focuses on the detailed examination of these diffs, enabling users to understand, validate, and correct changes between file versions.

Key Features:

1. The `Hunk` class encapsulates a contiguous block of changes within a file. It includes detailed information such as start lines before and after edits, lengths of change blocks, and specific line changes categorized as additions, deletions, or unchanged.

2. The `Diff` class represents a complete set of changes across a file and may contain multiple `Hunk` objects. It facilitates operations like generating string representations of diffs, and validating and correcting hunks based on the original file content.

3. Functions within the module allow for the validation of hunks against original files, identifying mismatches, and making necessary corrections. This feature ensures that diffs are accurate and reflect true changes.

4. Utility functions `is_similar` and `count_ratio` offer the capability to compare strings for similarity, accounting for variations in spacing and case. This aids in the validation process by allowing a flexible comparison of code lines.

Dependencies:

- `logging`: Utilized for logging warnings and errors encountered during the validation and correction process.
- `collections.Counter`: Used for counting occurrences of characters in strings, supporting the string similarity assessment functions.

Functions and Classes:

1. `Hunk`: Class representing a block of changes within a file, with methods for managing and validating these changes.

2. `Diff`: Class representing the entire set of changes in a file, containing multiple `Hunk` instances and methods for overall diff management.

3. `is_similar(str1, str2, similarity_threshold)`: Function to compare two strings for similarity, useful in validating line changes in hunks.

4. `count_ratio(str1, str2)`: Function that computes the ratio of common characters to the length of the longer string, aiding in the assessment of line similarity.

This module is essential for developers and teams utilizing version control systems, providing tools for a deeper analysis and correction of diffs, ensuring the integrity and accuracy of code changes.

"""
import logging

from collections import Counter
from typing import List

RETAIN = "retain"
ADD = "add"
REMOVE = "remove"


class Hunk:
    """
    Represents a section of a file diff, containing changes made to that section.

    Attributes:
        start_line_pre_edit (int): The starting line number in the original file.
        hunk_len_pre_edit (int): The length of the hunk in the original file.
        start_line_post_edit (int): The starting line number in the edited file.
        hunk_len_post_edit (int): The length of the hunk in the edited file.
        lines (list): A list of tuples representing the lines in the hunk and their types (RETAIN, ADD, REMOVE).
        category_counts (dict): A count of lines by their type.
        is_new_file (bool): Flag indicating if the hunk represents a new file.
    """

    def __init__(
        self,
        start_line_pre_edit,
        hunk_len_pre_edit,
        start_line_post_edit,
        hunk_len_post_edit,
        lines,
    ) -> None:
        self.start_line_pre_edit = start_line_pre_edit
        self.hunk_len_pre_edit = hunk_len_pre_edit
        self.start_line_post_edit = start_line_post_edit
        self.hunk_len_post_edit = hunk_len_post_edit
        self.category_counts = {RETAIN: 0, ADD: 0, REMOVE: 0}
        self.lines = list()
        self.add_lines(lines)
        self.forward_block_len = 10
        # Note that this assumption should not be done on hunk level, however, if the below is true, no validation is possible anyway.
        if self.category_counts[RETAIN] == 0 and self.category_counts[REMOVE] == 0:
            self.is_new_file = True
        else:
            self.is_new_file = False

    def add_retained_line(self, line, index) -> None:
        """Adds a retained line to the hunk at the specified index."""
        self.lines.insert(index, (RETAIN, line))
        self.category_counts[RETAIN] += 1

    def relabel_line(self, index, new_label) -> None:
        """Changes the label of a line at the specified index."""
        old_label = self.lines[index][0]
        self.lines[index] = (new_label, self.lines[index][1])
        self.category_counts[old_label] -= 1
        self.category_counts[new_label] += 1

    def pop_line(self, line, index) -> None:
        """Removes a line from the hunk at the specified index."""
        self.lines.pop(index)
        assert self.category_counts[line[0]] > 0
        self.category_counts[line[0]] -= 1

    def add_lines(self, new_lines) -> None:
        """Adds multiple lines to the hunk."""
        for line in new_lines:
            self.lines.append(line)
            self.category_counts[line[0]] += 1

    def hunk_to_string(self) -> str:
        """Converts the hunk to a string representation."""
        string = f"@@ -{self.start_line_pre_edit},{self.hunk_len_pre_edit} +{self.start_line_post_edit},{self.hunk_len_post_edit} @@\n"
        for line_type, line_content in self.lines:
            line_prefix = (
                " " if line_type == RETAIN else "+" if line_type == ADD else "-"
            )
            string += f"{line_prefix}{line_content}\n"
        return string

    def make_forward_block(self, hunk_ind: int, forward_block_len) -> str:
        """Creates a block of lines for forward comparison."""
        forward_lines = [
            line[1] for line in self.lines[hunk_ind:] if not line[0] == ADD
        ]
        forward_block = "\n".join(forward_lines[0:forward_block_len])
        return forward_block

    def check_start_line(self, lines_dict: dict) -> bool:
        """Check if the starting line of a hunk is present in the original code and returns a boolean value accordingly."""
        if self.is_new_file:
            # this hunk cannot be falsified and is by definition true
            return True
        if self.start_line_pre_edit in lines_dict:
            # check the location of the actual starting line:
            is_similar(self.lines[0][1], lines_dict[self.start_line_pre_edit])
        else:
            pass

    def find_start_line(self, lines_dict: dict, problems: list) -> bool:
        """Finds the starting line of the hunk in the original code and returns a boolean value accordingly. If the starting line is not found, it appends a problem message to the problems list."""

        # ToDo handle the case where the start line is 0 or 1 characters separately
        if self.lines[0][0] == ADD:
            # handle the case where the start line is an add
            start_line = None
            # find the first line that is not an add
            for index, line in enumerate(self.lines):
                if line[0] != ADD:
                    for line_number, line_content in lines_dict.items():
                        # if the line is similar to a non-blank line in line_dict, we can pick the line prior to it
                        if is_similar(line[1], line_content) and line[1] != "":
                            start_line = line_number - 1
                            break
                    # if the start line is not found, append a problem message
                    if start_line is None:
                        problems.append(
                            f"In {self.hunk_to_string()}:can not find the starting line of the diff"
                        )
                        return False

                    else:
                        # the line prior to the start line is found now we insert it to the first place as the start line
                        self.start_line_pre_edit = start_line
                        retain_line = lines_dict.get(start_line, "")
                        if retain_line:
                            self.add_retained_line(lines_dict[start_line], 0)
                            return self.validate_and_correct(lines_dict, problems)
                        else:
                            problems.append(
                                f"In {self.hunk_to_string()}:The starting line of the diff {self.hunk_to_string()} does not exist in the code"
                            )
                            return False
        pot_start_lines = {
            key: is_similar(self.lines[0][1], line) for key, line in lines_dict.items()
        }
        sum_of_matches = sum(pot_start_lines.values())
        if sum_of_matches == 0:
            # before we go any further, we should check if it's a comment from LLM
            if self.lines[0][1].count("#") > 0:
                # if it is, we can mark it as an ADD lines
                self.relabel_line(0, ADD)
                # and restart the validation at the next line
                return self.validate_and_correct(lines_dict, problems)

            else:
                problems.append(
                    f"In {self.hunk_to_string()}:The starting line of the diff {self.hunk_to_string()} does not exist in the code"
                )
                return False
        elif sum_of_matches == 1:
            start_ind = list(pot_start_lines.keys())[
                list(pot_start_lines.values()).index(True)
            ]  # lines are one indexed
        else:
            logging.warning("multiple candidates for starting index")
            # ToDo handle all the cases better again here. Smartest choice is that, for each candidate check match to the next line etc (recursively)
            start_ind = list(pot_start_lines.keys())[
                list(pot_start_lines.values()).index(True)
            ]
        self.start_line_pre_edit = start_ind

        # This should now be fulfilled by default
        assert is_similar(self.lines[0][1], lines_dict[self.start_line_pre_edit])
        return True

    def validate_lines(self, lines_dict: dict, problems: list) -> bool:
        """Validates the lines of the hunk against the original file and returns a boolean value accordingly. If the lines do not match, it appends a problem message to the problems list."""
        hunk_ind = 0
        file_ind = self.start_line_pre_edit
        # make an orig hunk lines for logging
        # orig_hunk_lines = deepcopy(self.lines)
        while hunk_ind < len(self.lines) and file_ind <= max(lines_dict):
            if self.lines[hunk_ind][0] == ADD:
                # this cannot be validated, jump one index
                hunk_ind += 1
            elif not is_similar(self.lines[hunk_ind][1], lines_dict[file_ind]):
                # before we go any further, we should relabel the comment from LLM
                if self.lines[hunk_ind][1].count("#") > 0:
                    self.relabel_line(hunk_ind, ADD)
                    continue

                # make a forward block from the code for comparisons
                forward_code = "\n".join(
                    [
                        lines_dict[ind]
                        for ind in range(
                            file_ind,
                            min(
                                file_ind + self.forward_block_len,
                                max(lines_dict.keys()),
                            ),
                        )
                    ]
                )
                # make the original forward block for quantitative comparison
                forward_block = self.make_forward_block(
                    hunk_ind, self.forward_block_len
                )
                orig_count_ratio = count_ratio(forward_block, forward_code)
                # Here we have 2 cases
                # 1) some lines were simply skipped in the diff and we should add them to the diff
                # If this is the case, adding the line to the diff, should give an improved forward diff
                forward_block_missing_line = self.make_forward_block(
                    hunk_ind, self.forward_block_len - 1
                )
                # insert the missing line in front of the block
                forward_block_missing_line = "\n".join(
                    [lines_dict[file_ind], forward_block_missing_line]
                )
                missing_line_count_ratio = count_ratio(
                    forward_block_missing_line, forward_code
                )
                # 2) Additional lines, not belonging to the code were added to the diff
                forward_block_false_line = self.make_forward_block(
                    hunk_ind + 1, self.forward_block_len
                )
                false_line_count_ratio = count_ratio(
                    forward_block_false_line, forward_code
                )
                if (
                    orig_count_ratio >= missing_line_count_ratio
                    and orig_count_ratio >= false_line_count_ratio
                ):
                    problems.append(
                        f"In Hunk:{self.hunk_to_string()}, there was at least one mismatch."
                    )
                    return False

                elif missing_line_count_ratio > false_line_count_ratio:
                    self.add_retained_line(lines_dict[file_ind], hunk_ind)
                    hunk_ind += 1
                    file_ind += 1
                    # NOTE: IF THE LLM SKIPS SOME LINES AND HAS ADDs ADJACENT TO THE SKIPPED BLOCK,
                    # WE CANNOT KNOW WHETHER THE ADDs SHOULD BE BEFORE OR AFTER THE BLOCK. WE OPT FOR PUTTING IT BEFORE.
                    # IF IT MATTERED, WE ASSUME THE LLM WOULD NOT SKIP THE BLOCK
                else:
                    self.pop_line(self.lines[hunk_ind], hunk_ind)

            else:
                hunk_ind += 1
                file_ind += 1
        # if we have not validated all lines, we have a problem
        if hunk_ind < len(self.lines) - 1:
            remaining_lines = "\n".join(
                f"{line_type}: {line_content}"
                for line_type, line_content in self.lines[file_ind + 1 :]
            )
            problems.append(
                f"In {self.hunk_to_string()}:Hunk validation stopped before the lines {remaining_lines} were validated. The diff is incorrect"
            )
            return False
        return True

    def validate_and_correct(
        self,
        lines_dict: dict,
        problems: list,
    ) -> bool:
        """
        Validates and corrects the hunk based on the original lines.

        This function attempts to validate the hunk by comparing its lines to the original file and making corrections
        where necessary. It also identifies problems such as non-matching lines or incorrect line types.
        """
        start_true = self.check_start_line(lines_dict)

        if not start_true:
            if not self.find_start_line(lines_dict, problems):
                return False

        # Now we should be able to validate the hunk line by line and add missing line
        if not self.validate_lines(lines_dict, problems):
            return False
        # Pass the validation
        return True


class Diff:
    """
    Represents a file diff, containing multiple hunks of changes.

    Attributes:
        filename_pre (str): The name of the original file.
        filename_post (str): The name of the edited file.
        hunks (list): A list of Hunk objects representing the changes in the diff.
    """

    def __init__(self, filename_pre, filename_post) -> None:
        self.filename_pre = filename_pre
        self.filename_post = filename_post
        self.hunks = []

    def is_new_file(self) -> bool:
        """Determines if the diff represents a new file."""
        if self.filename_pre == "/dev/null":
            return True
        return any(hunk.is_new_file for hunk in self.hunks)

    def diff_to_string(self) -> str:
        """Converts the diff to a string representation."""
        string = f"--- {self.filename_pre}\n+++ {self.filename_post}\n"
        for hunk in self.hunks:
            string += hunk.hunk_to_string()
        return string.strip()

    def validate_and_correct(self, lines_dict: dict) -> List[str]:
        """Validates and corrects each hunk in the diff."""
        problems = []
        past_hunk = None
        cut_lines_dict = lines_dict.copy()
        for hunk in self.hunks:
            if past_hunk is not None:
                # make sure to not cut so much that the start_line gets out of range
                cut_ind = min(
                    past_hunk.start_line_pre_edit + past_hunk.hunk_len_pre_edit,
                    hunk.start_line_pre_edit,
                )
                cut_lines_dict = {
                    key: val for key, val in cut_lines_dict.items() if key >= (cut_ind)
                }
            is_valid = hunk.validate_and_correct(cut_lines_dict, problems)
            if not is_valid and len(problems) > 0:
                for idx, val in enumerate(problems):
                    print(f"\nInvalid Hunk NO.{idx}---\n{val}\n---")
                self.hunks.remove(hunk)
            # now correct the numbers, assuming the start line pre-edit has been fixed
            hunk.hunk_len_pre_edit = (
                hunk.category_counts[RETAIN] + hunk.category_counts[REMOVE]
            )
            hunk.hunk_len_post_edit = (
                hunk.category_counts[RETAIN] + hunk.category_counts[ADD]
            )
            if past_hunk is not None:
                hunk.start_line_post_edit = (
                    hunk.start_line_pre_edit
                    + past_hunk.hunk_len_post_edit
                    - past_hunk.hunk_len_pre_edit
                    + past_hunk.start_line_post_edit
                    - past_hunk.start_line_pre_edit
                )
            else:
                hunk.start_line_post_edit = hunk.start_line_pre_edit
            past_hunk = hunk
        return problems


def is_similar(str1, str2, similarity_threshold=0.9) -> bool:
    """
    Compares two strings for similarity, ignoring spaces and case.

    Parameters
    ----------
    str1, str2 : str
        The strings to compare.
    similarity_threshold: float
        How similar must the strings be

    Returns
    -------
    bool
        True if the strings are similar, False otherwise.
    """

    return count_ratio(str1, str2) >= similarity_threshold


def count_ratio(str1, str2) -> float:
    """
    Computes the ratio of common characters to the length of the longer string, ignoring spaces and case.

    Parameters:
    - str1, str2 (str): The strings to compare.

    Returns:
    - float: The ratio of common characters to the length of the longer string.
    """
    str1, str2 = str1.replace(" ", "").lower(), str2.replace(" ", "").lower()

    counter1, counter2 = Counter(str1), Counter(str2)
    intersection = sum((counter1 & counter2).values())
    longer_length = max(len(str1), len(str2))
    if longer_length == 0:
        return 1
    else:
        return intersection / longer_length



File: gpt_engineer/core/prompt.py

import json

from typing import Dict, Optional


class Prompt:
    def __init__(
        self,
        text: str,
        image_urls: Optional[Dict[str, str]] = None,
        entrypoint_prompt: str = "",
    ):
        self.text = text
        self.image_urls = image_urls
        self.entrypoint_prompt = entrypoint_prompt

    def __repr__(self):
        return f"Prompt(text={self.text!r}, image_urls={self.image_urls!r})"

    def to_langchain_content(self):
        content = [{"type": "text", "text": f"Request: {self.text}"}]

        if self.image_urls:
            for name, url in self.image_urls.items():
                image_content = {
                    "type": "image_url",
                    "image_url": {
                        "url": url,
                        "detail": "low",
                    },
                }
                content.append(image_content)

        return content

    def to_dict(self):
        return {
            "text": self.text,
            "image_urls": self.image_urls,
            "entrypoint_prompt": self.entrypoint_prompt,
        }

    def to_json(self):
        return json.dumps(self.to_dict())



File: gpt_engineer/core/default/steps.py

"""
Module for defining the steps involved in generating and improving code using AI.

This module provides functions that represent different steps in the process of generating
and improving code using an AI model. These steps include generating code from a prompt,
creating an entrypoint for the codebase, executing the entrypoint, and refining code edits.

Functions
---------
curr_fn : function
    Returns the name of the current function.

setup_sys_prompt : function
    Sets up the system prompt for generating code.

gen_code : function
    Generates code from a prompt using AI and returns the generated files.

gen_entrypoint : function
    Generates an entrypoint for the codebase and returns the entrypoint files.

execute_entrypoint : function
    Executes the entrypoint of the codebase.

setup_sys_prompt_existing_code : function
    Sets up the system prompt for improving existing code.


improve : function
    Improves the code based on user input and returns the updated files.
"""

import inspect
import io
import re
import sys
import traceback

from pathlib import Path
from typing import List, MutableMapping, Union

from langchain.schema import HumanMessage, SystemMessage
from termcolor import colored

from gpt_engineer.core.ai import AI
from gpt_engineer.core.base_execution_env import BaseExecutionEnv
from gpt_engineer.core.base_memory import BaseMemory
from gpt_engineer.core.chat_to_files import apply_diffs, chat_to_files_dict, parse_diffs
from gpt_engineer.core.default.constants import MAX_EDIT_REFINEMENT_STEPS
from gpt_engineer.core.default.paths import (
    CODE_GEN_LOG_FILE,
    DEBUG_LOG_FILE,
    DIFF_LOG_FILE,
    ENTRYPOINT_FILE,
    ENTRYPOINT_LOG_FILE,
    IMPROVE_LOG_FILE,
)
from gpt_engineer.core.files_dict import FilesDict, file_to_lines_dict
from gpt_engineer.core.preprompts_holder import PrepromptsHolder
from gpt_engineer.core.prompt import Prompt


def curr_fn() -> str:
    """
    Returns the name of the current function.

    Returns
    -------
    str
        The name of the function that called this function.
    """
    return inspect.stack()[1].function


def setup_sys_prompt(preprompts: MutableMapping[Union[str, Path], str]) -> str:
    """
    Sets up the system prompt for generating code.

    Parameters
    ----------
    preprompts : MutableMapping[Union[str, Path], str]
        A mapping of preprompt messages to guide the AI model.

    Returns
    -------
    str
        The system prompt message for the AI model.
    """
    return (
        preprompts["roadmap"]
        + preprompts["generate"].replace("FILE_FORMAT", preprompts["file_format"])
        + "\nUseful to know:\n"
        + preprompts["philosophy"]
    )


def setup_sys_prompt_existing_code(
    preprompts: MutableMapping[Union[str, Path], str]
) -> str:
    """
    Sets up the system prompt for improving existing code.

    Parameters
    ----------
    preprompts : MutableMapping[Union[str, Path], str]
        A mapping of preprompt messages to guide the AI model.

    Returns
    -------
    str
        The system prompt message for the AI model to improve existing code.
    """
    return (
        preprompts["roadmap"]
        + preprompts["improve"].replace("FILE_FORMAT", preprompts["file_format_diff"])
        + "\nUseful to know:\n"
        + preprompts["philosophy"]
    )


def gen_code(
    ai: AI, prompt: Prompt, memory: BaseMemory, preprompts_holder: PrepromptsHolder
) -> FilesDict:
    """
    Generates code from a prompt using AI and returns the generated files.

    Parameters
    ----------
    ai : AI
        The AI model used for generating code.
    prompt : str
        The user prompt to generate code from.
    memory : BaseMemory
        The memory interface where the code and related data are stored.
    preprompts_holder : PrepromptsHolder
        The holder for preprompt messages that guide the AI model.

    Returns
    -------
    FilesDict
        A dictionary of file names to their respective source code content.
    """
    preprompts = preprompts_holder.get_preprompts()
    messages = ai.start(
        setup_sys_prompt(preprompts), prompt.to_langchain_content(), step_name=curr_fn()
    )
    chat = messages[-1].content.strip()
    memory.log(CODE_GEN_LOG_FILE, "\n\n".join(x.pretty_repr() for x in messages))
    files_dict = chat_to_files_dict(chat)
    return files_dict


def gen_entrypoint(
    ai: AI,
    prompt: Prompt,
    files_dict: FilesDict,
    memory: BaseMemory,
    preprompts_holder: PrepromptsHolder,
) -> FilesDict:
    """
    Generates an entrypoint for the codebase and returns the entrypoint files.

    Parameters
    ----------
    ai : AI
        The AI model used for generating the entrypoint.
    files_dict : FilesDict
        The dictionary of file names to their respective source code content.
    memory : BaseMemory
        The memory interface where the code and related data are stored.
    preprompts_holder : PrepromptsHolder
        The holder for preprompt messages that guide the AI model.

    Returns
    -------
    FilesDict
        A dictionary containing the entrypoint file.
    """
    user_prompt = prompt.entrypoint_prompt
    if not user_prompt:
        user_prompt = """
        Make a unix script that
        a) installs dependencies
        b) runs all necessary parts of the codebase (in parallel if necessary)
        """
    preprompts = preprompts_holder.get_preprompts()
    messages = ai.start(
        system=(preprompts["entrypoint"]),
        user=user_prompt
        + "\nInformation about the codebase:\n\n"
        + files_dict.to_chat(),
        step_name=curr_fn(),
    )
    print()
    chat = messages[-1].content.strip()
    regex = r"```\S*\n(.+?)```"
    matches = re.finditer(regex, chat, re.DOTALL)
    entrypoint_code = FilesDict(
        {ENTRYPOINT_FILE: "\n".join(match.group(1) for match in matches)}
    )
    memory.log(ENTRYPOINT_LOG_FILE, "\n\n".join(x.pretty_repr() for x in messages))
    return entrypoint_code


def execute_entrypoint(
    ai: AI,
    execution_env: BaseExecutionEnv,
    files_dict: FilesDict,
    prompt: Prompt = None,
    preprompts_holder: PrepromptsHolder = None,
    memory: BaseMemory = None,
) -> FilesDict:
    """
    Executes the entrypoint of the codebase.

    Parameters
    ----------
    ai : AI
        The AI model used for generating the entrypoint.
    execution_env : BaseExecutionEnv
        The execution environment in which the code is executed.
    files_dict : FilesDict
        The dictionary of file names to their respective source code content.
    preprompts_holder : PrepromptsHolder, optional
        The holder for preprompt messages that guide the AI model.

    Returns
    -------
    FilesDict
        The dictionary of file names to their respective source code content after execution.
    """
    if ENTRYPOINT_FILE not in files_dict:
        raise FileNotFoundError(
            "The required entrypoint "
            + ENTRYPOINT_FILE
            + " does not exist in the code."
        )

    command = files_dict[ENTRYPOINT_FILE]

    print()
    print(
        colored(
            "Do you want to execute this code? (Y/n)",
            "red",
        )
    )
    print()
    print(command)
    print()
    if input("").lower() not in ["", "y", "yes"]:
        print("Ok, not executing the code.")
        return files_dict
    print("Executing the code...")
    print()
    print(
        colored(
            "Note: If it does not work as expected, consider running the code"
            + " in another way than above.",
            "green",
        )
    )
    print()
    print("You can press ctrl+c *once* to stop the execution.")
    print()

    execution_env.upload(files_dict).run(f"bash {ENTRYPOINT_FILE}")
    return files_dict


def improve_fn(
    ai: AI,
    prompt: Prompt,
    files_dict: FilesDict,
    memory: BaseMemory,
    preprompts_holder: PrepromptsHolder,
    diff_timeout=3,
) -> FilesDict:
    """
    Improves the code based on user input and returns the updated files.

    Parameters
    ----------
    ai : AI
        The AI model used for improving code.
    prompt :str
        The user prompt to improve the code.
    files_dict : FilesDict
        The dictionary of file names to their respective source code content.
    memory : BaseMemory
        The memory interface where the code and related data are stored.
    preprompts_holder : PrepromptsHolder
        The holder for preprompt messages that guide the AI model.

    Returns
    -------
    FilesDict
        The dictionary of file names to their respective updated source code content.
    """
    preprompts = preprompts_holder.get_preprompts()
    messages = [
        SystemMessage(content=setup_sys_prompt_existing_code(preprompts)),
    ]

    # Add files as input
    messages.append(HumanMessage(content=f"{files_dict.to_chat()}"))
    messages.append(HumanMessage(content=prompt.to_langchain_content()))
    memory.log(
        DEBUG_LOG_FILE,
        "UPLOADED FILES:\n" + files_dict.to_log() + "\nPROMPT:\n" + prompt.text,
    )
    return _improve_loop(ai, files_dict, memory, messages, diff_timeout=diff_timeout)


def _improve_loop(
    ai: AI, files_dict: FilesDict, memory: BaseMemory, messages: List, diff_timeout=3
) -> FilesDict:
    messages = ai.next(messages, step_name=curr_fn())
    files_dict, errors = salvage_correct_hunks(
        messages, files_dict, memory, diff_timeout=diff_timeout
    )

    retries = 0
    while errors and retries < MAX_EDIT_REFINEMENT_STEPS:
        messages.append(
            HumanMessage(
                content="Some previously produced diffs were not on the requested format, or the code part was not found in the code. Details:\n"
                + "\n".join(errors)
                + "\n Only rewrite the problematic diffs, making sure that the failing ones are now on the correct format and can be found in the code. Make sure to not repeat past mistakes. \n"
            )
        )
        messages = ai.next(messages, step_name=curr_fn())
        files_dict, errors = salvage_correct_hunks(
            messages, files_dict, memory, diff_timeout
        )
        retries += 1

    return files_dict


def salvage_correct_hunks(
    messages: List, files_dict: FilesDict, memory: BaseMemory, diff_timeout=3
) -> tuple[FilesDict, List[str]]:
    error_messages = []
    ai_response = messages[-1].content.strip()

    diffs = parse_diffs(ai_response, diff_timeout=diff_timeout)
    # validate and correct diffs

    for _, diff in diffs.items():
        # if diff is a new file, validation and correction is unnecessary
        if not diff.is_new_file():
            problems = diff.validate_and_correct(
                file_to_lines_dict(files_dict[diff.filename_pre])
            )
            error_messages.extend(problems)
    files_dict = apply_diffs(diffs, files_dict)
    memory.log(IMPROVE_LOG_FILE, "\n\n".join(x.pretty_repr() for x in messages))
    memory.log(DIFF_LOG_FILE, "\n\n".join(error_messages))
    return files_dict, error_messages


class Tee(object):
    def __init__(self, *files):
        self.files = files

    def write(self, obj):
        for file in self.files:
            file.write(obj)

    def flush(self):
        for file in self.files:
            file.flush()


def handle_improve_mode(prompt, agent, memory, files_dict, diff_timeout=3):
    captured_output = io.StringIO()
    old_stdout = sys.stdout
    sys.stdout = Tee(sys.stdout, captured_output)

    try:
        files_dict = agent.improve(files_dict, prompt, diff_timeout=diff_timeout)
    except Exception as e:
        print(
            f"Error while improving the project: {e}\nCould you please upload the debug_log_file.txt in {memory.path}/logs folder to github?\nFULL STACK TRACE:\n"
        )
        traceback.print_exc(file=sys.stdout)  # Print the full stack trace
    finally:
        # Reset stdout
        sys.stdout = old_stdout

        # Get the captured output
        captured_string = captured_output.getvalue()
        print(captured_string)
        memory.log(DEBUG_LOG_FILE, "\nCONSOLE OUTPUT:\n" + captured_string)

    return files_dict



File: gpt_engineer/core/default/simple_agent.py

"""
Module for defining a simple agent that uses AI to manage code generation and improvement.

This module provides a class that represents an agent capable of initializing and improving
a codebase using AI. It handles interactions with the AI model, memory, and execution
environment to generate and refine code based on user prompts.

"""

import tempfile

from typing import Optional

from gpt_engineer.core.ai import AI
from gpt_engineer.core.base_agent import BaseAgent
from gpt_engineer.core.base_execution_env import BaseExecutionEnv
from gpt_engineer.core.base_memory import BaseMemory
from gpt_engineer.core.default.disk_execution_env import DiskExecutionEnv
from gpt_engineer.core.default.disk_memory import DiskMemory
from gpt_engineer.core.default.paths import PREPROMPTS_PATH, memory_path
from gpt_engineer.core.default.steps import gen_code, gen_entrypoint, improve_fn
from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.preprompts_holder import PrepromptsHolder
from gpt_engineer.core.prompt import Prompt


class SimpleAgent(BaseAgent):
    """
    An agent that uses AI to generate and improve code based on a given prompt.

    This agent is capable of initializing a codebase from a prompt and improving an existing
    codebase based on user input. It uses an AI model to generate and refine code, and it
    interacts with a repository and an execution environment to manage and execute the code.

    Attributes
    ----------
    memory : BaseMemory
        The memory interface where the code and related data are stored.
    execution_env : BaseExecutionEnv
        The execution environment in which the code is executed.
    ai : AI
        The AI model used for generating and improving code.
    preprompts_holder : PrepromptsHolder
        The holder for preprompt messages that guide the AI model.
    """

    def __init__(
        self,
        memory: BaseMemory,
        execution_env: BaseExecutionEnv,
        ai: AI = None,
        preprompts_holder: PrepromptsHolder = None,
    ):
        self.preprompts_holder = preprompts_holder or PrepromptsHolder(PREPROMPTS_PATH)
        self.memory = memory
        self.execution_env = execution_env
        self.ai = ai or AI()

    @classmethod
    def with_default_config(
        cls, path: str, ai: AI = None, preprompts_holder: PrepromptsHolder = None
    ):
        return cls(
            memory=DiskMemory(memory_path(path)),
            execution_env=DiskExecutionEnv(),
            ai=ai,
            preprompts_holder=preprompts_holder or PrepromptsHolder(PREPROMPTS_PATH),
        )

    def init(self, prompt: Prompt) -> FilesDict:
        files_dict = gen_code(self.ai, prompt, self.memory, self.preprompts_holder)
        entrypoint = gen_entrypoint(
            self.ai, prompt, files_dict, self.memory, self.preprompts_holder
        )
        combined_dict = {**files_dict, **entrypoint}
        files_dict = FilesDict(combined_dict)
        return files_dict

    def improve(
        self,
        files_dict: FilesDict,
        prompt: Prompt,
        execution_command: Optional[str] = None,
    ) -> FilesDict:
        files_dict = improve_fn(
            self.ai, prompt, files_dict, self.memory, self.preprompts_holder
        )
        return files_dict


def default_config_agent():
    """
    Creates an instance of SimpleAgent with default configuration.

    Returns
    -------
    SimpleAgent
        An instance of SimpleAgent with a temporary directory as its base path.
    """
    return SimpleAgent.with_default_config(tempfile.mkdtemp())



File: gpt_engineer/core/default/file_store.py

import tempfile

from pathlib import Path
from typing import Union

from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.linting import Linting


class FileStore:
    """
    Module for managing file storage in a temporary directory.

    This module provides a class that manages the storage of files in a temporary directory.
    It includes methods for uploading files to the directory and downloading them as a
    collection of files.

    Classes
    -------
    FileStore
        Manages file storage in a temporary directory, allowing for upload and download of files.

    Imports
    -------
    - tempfile: For creating temporary directories.
    - Path: For handling file system paths.
    - Union: For type annotations.
    - FilesDict: For handling collections of files.
    """

    def __init__(self, path: Union[str, Path, None] = None):
        if path is None:
            path = Path(tempfile.mkdtemp(prefix="gpt-engineer-"))

        self.working_dir = Path(path)
        self.working_dir.mkdir(parents=True, exist_ok=True)
        self.id = self.working_dir.name.split("-")[-1]

    def push(self, files: FilesDict):
        for name, content in files.items():
            path = self.working_dir / name
            path.parent.mkdir(parents=True, exist_ok=True)
            with open(path, "w") as f:
                f.write(content)
        return self

    def linting(self, files: FilesDict) -> FilesDict:
        # lint the code
        linting = Linting()
        return linting.lint_files(files)

    def pull(self) -> FilesDict:
        files = {}
        for path in self.working_dir.glob("**/*"):
            if path.is_file():
                with open(path, "r") as f:
                    try:
                        content = f.read()
                    except UnicodeDecodeError:
                        content = "binary file"
                    files[str(path.relative_to(self.working_dir))] = content
        return FilesDict(files)



File: gpt_engineer/core/default/paths.py

"""
Module defining file system paths used by the application.

This module contains definitions of file system paths that are used throughout the
application to locate and manage various files and directories, such as logs, memory,
and preprompts.

Constants
---------
META_DATA_REL_PATH : str
    The relative path to the directory where metadata is stored.

MEMORY_REL_PATH : str
    The relative path to the directory where memory-related files are stored.

CODE_GEN_LOG_FILE : str
    The filename for the log file that contains all output from code generation.

DEBUG_LOG_FILE : str
    The filename for the log file that contains debug information.

ENTRYPOINT_FILE : str
    The filename for the entrypoint script that is executed to run the application.

ENTRYPOINT_LOG_FILE : str
    The filename for the log file that contains the chat related to entrypoint generation.

PREPROMPTS_PATH : Path
    The file system path to the directory containing preprompt files.

Functions
---------
memory_path : function
    Constructs the full path to the memory directory based on a given base path.

metadata_path : function
    Constructs the full path to the metadata directory based on a given base path.
"""
import os

from pathlib import Path

META_DATA_REL_PATH = ".gpteng"
MEMORY_REL_PATH = os.path.join(META_DATA_REL_PATH, "memory")
CODE_GEN_LOG_FILE = "all_output.txt"
IMPROVE_LOG_FILE = "improve.txt"
DIFF_LOG_FILE = "diff_errors.txt"
DEBUG_LOG_FILE = "debug_log_file.txt"
ENTRYPOINT_FILE = "run.sh"
ENTRYPOINT_LOG_FILE = "gen_entrypoint_chat.txt"
ENTRYPOINT_FILE = "run.sh"
PREPROMPTS_PATH = Path(__file__).parent.parent.parent / "preprompts"


def memory_path(path):
    """
    Constructs the full path to the memory directory based on a given base path.

    Parameters
    ----------
    path : str
        The base path to append the memory directory to.

    Returns
    -------
    str
        The full path to the memory directory.
    """
    return os.path.join(path, MEMORY_REL_PATH)


def metadata_path(path):
    """
    Constructs the full path to the metadata directory based on a given base path.

    Parameters
    ----------
    path : str
        The base path to append the metadata directory to.

    Returns
    -------
    str
        The full path to the metadata directory.
    """
    return os.path.join(path, META_DATA_REL_PATH)



File: gpt_engineer/core/default/constants.py

"""
Module defining constants used throughout the application.

This module contains definitions of constants that are used across various
components of the application to maintain consistency and ease of configuration.

Constants
---------
MAX_EDIT_REFINEMENT_STEPS : int
    The maximum number of refinement steps allowed when generating edit blocks.
"""
MAX_EDIT_REFINEMENT_STEPS = 2



File: gpt_engineer/core/default/disk_memory.py

"""
Disk Memory Module
==================

This module provides a simple file-based key-value database system, where keys are
represented as filenames and values are the contents of these files. The `DiskMemory` class
is responsible for the CRUD operations on the database.

Attributes
----------
None

Functions
---------
None

Classes
-------
DiskMemory
    A file-based key-value store where keys correspond to filenames and values to file contents.
"""

import base64
import json
import shutil

from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Iterator, Optional, Union

from gpt_engineer.core.base_memory import BaseMemory
from gpt_engineer.tools.supported_languages import SUPPORTED_LANGUAGES


# This class represents a simple database that stores its tools as files in a directory.
class DiskMemory(BaseMemory):
    """
    A file-based key-value store where keys correspond to filenames and values to file contents.

    This class provides an interface to a file-based database, leveraging file operations to
    facilitate CRUD-like interactions. It allows for quick checks on the existence of keys,
    retrieval of values based on keys, and setting new key-value pairs.

    Attributes
    ----------
    path : Path
        The directory path where the database files are stored.
    """

    def __init__(self, path: Union[str, Path]):
        """
        Initialize the DiskMemory class with a specified path.

        Parameters
        ----------
        path : str or Path
            The path to the directory where the database files will be stored.

        """
        self.path: Path = Path(path).absolute()

        self.path.mkdir(parents=True, exist_ok=True)

    def __contains__(self, key: str) -> bool:
        """
        Determine whether the database contains a file with the specified key.

        Parameters
        ----------
        key : str
            The key (filename) to check for existence in the database.

        Returns
        -------
        bool
            Returns True if the file exists, False otherwise.

        """
        return (self.path / key).is_file()

    def __getitem__(self, key: str) -> str:
        """
        Retrieve the content of a file in the database corresponding to the given key.
        If the file is an image with a .png or .jpeg extension, it returns the content
        in Base64-encoded string format.

        Parameters
        ----------
        key : str
            The key (filename) whose content is to be retrieved.

        Returns
        -------
        str
            The content of the file associated with the key, or Base64-encoded string if it's a .png or .jpeg file.

        Raises
        ------
        KeyError
            If the file corresponding to the key does not exist in the database.
        """
        full_path = self.path / key

        if not full_path.is_file():
            raise KeyError(f"File '{key}' could not be found in '{self.path}'")

        if full_path.suffix in [".png", ".jpeg", ".jpg"]:
            with full_path.open("rb") as image_file:
                encoded_string = base64.b64encode(image_file.read()).decode("utf-8")
                mime_type = "image/png" if full_path.suffix == ".png" else "image/jpeg"
                return f"data:{mime_type};base64,{encoded_string}"
        else:
            with full_path.open("r", encoding="utf-8") as f:
                return f.read()

    def get(self, key: str, default: Optional[Any] = None) -> Any:
        """
        Retrieve the content of a file in the database, or return a default value if not found.

        Parameters
        ----------
        key : str
            The key (filename) whose content is to be retrieved.
        default : Any, optional
            The default value to return if the file does not exist. Default is None.

        Returns
        -------
        Any
            The content of the file if it exists, a new DiskMemory instance if the key corresponds to a directory.
        """

        item_path = self.path / key
        try:
            if item_path.is_file():
                return self[key]
            elif item_path.is_dir():
                return DiskMemory(item_path)
            else:
                return default
        except:
            return default

    def __setitem__(self, key: Union[str, Path], val: str) -> None:
        """
        Set or update the content of a file in the database corresponding to the given key.

        Parameters
        ----------
        key : str or Path
            The key (filename) where the content is to be set.
        val : str
            The content to be written to the file.

        Raises
        ------
        ValueError
            If the key attempts to access a parent path.
        TypeError
            If the value is not a string.

        """
        if str(key).startswith("../"):
            raise ValueError(f"File name {key} attempted to access parent path.")

        if not isinstance(val, str):
            raise TypeError("val must be str")

        full_path = self.path / key
        full_path.parent.mkdir(parents=True, exist_ok=True)

        full_path.write_text(val, encoding="utf-8")

    def __delitem__(self, key: Union[str, Path]) -> None:
        """
        Delete a file or directory from the database corresponding to the given key.

        Parameters
        ----------
        key : str or Path
            The key (filename or directory name) to be deleted.

        Raises
        ------
        KeyError
            If the file or directory corresponding to the key does not exist in the database.

        """
        item_path = self.path / key
        if not item_path.exists():
            raise KeyError(f"Item '{key}' could not be found in '{self.path}'")

        if item_path.is_file():
            item_path.unlink()
        elif item_path.is_dir():
            shutil.rmtree(item_path)

    def __iter__(self) -> Iterator[str]:
        """
        Iterate over the keys (filenames) in the database.

        Yields
        ------
        Iterator[str]
            An iterator over the sorted list of keys (filenames) in the database.

        """
        return iter(
            sorted(
                str(item.relative_to(self.path))
                for item in sorted(self.path.rglob("*"))
                if item.is_file()
            )
        )

    def __len__(self) -> int:
        """
        Get the number of files in the database.

        Returns
        -------
        int
            The number of files in the database.

        """
        return len(list(self.__iter__()))

    def _supported_files(self) -> str:
        valid_extensions = {
            ext for lang in SUPPORTED_LANGUAGES for ext in lang["extensions"]
        }
        file_paths = [
            str(item)
            for item in self
            if Path(item).is_file() and Path(item).suffix in valid_extensions
        ]
        return "\n".join(file_paths)

    def _all_files(self) -> str:
        file_paths = [str(item) for item in self if Path(item).is_file()]
        return "\n".join(file_paths)

    def to_path_list_string(self, supported_code_files_only: bool = False) -> str:
        """
        Generate a string representation of the file paths in the database.

        Parameters
        ----------
        supported_code_files_only : bool, optional
            If True, filter the list to include only supported code file extensions.
            Default is False.

        Returns
        -------
        str
            A newline-separated string of file paths.

        """
        if supported_code_files_only:
            return self._supported_files()
        else:
            return self._all_files()

    def to_dict(self) -> Dict[Union[str, Path], str]:
        """
        Convert the database contents to a dictionary.

        Returns
        -------
        Dict[Union[str, Path], str]
            A dictionary with keys as filenames and values as file contents.

        """
        return {file_path: self[file_path] for file_path in self}

    def to_json(self) -> str:
        """
        Serialize the database contents to a JSON string.

        Returns
        -------
        str
            A JSON string representation of the database contents.

        """
        return json.dumps(self.to_dict())

    def log(self, key: Union[str, Path], val: str) -> None:
        """
        Append to a file or create and write to it if it doesn't exist.

        Parameters
        ----------
        key : str or Path
            The key (filename) where the content is to be appended.
        val : str
            The content to be appended to the file.

        """

        if str(key).startswith("../"):
            raise ValueError(f"File name {key} attempted to access parent path.")

        if not isinstance(val, str):
            raise TypeError("val must be str")

        full_path = self.path / "logs" / key
        full_path.parent.mkdir(parents=True, exist_ok=True)

        # Touch if it doesnt exist
        if not full_path.exists():
            full_path.touch()

        with open(full_path, "a", encoding="utf-8") as file:
            file.write(f"\n{datetime.now().isoformat()}\n")
            file.write(val + "\n")

    def archive_logs(self):
        """
        Moves all logs to archive directory based on current timestamp
        """
        if "logs" in self:
            archive_dir = (
                self.path / f"logs_{datetime.now().strftime('%Y-%m-%d-%H-%M-%S')}"
            )
            shutil.move(self.path / "logs", archive_dir)



File: gpt_engineer/core/default/__init__.py




File: gpt_engineer/core/default/disk_execution_env.py

"""
Module for managing the execution environment on the local disk.

This module provides a class that handles the execution of code stored on the local
file system. It includes methods for uploading files to the execution environment,
running commands, and capturing the output.

Classes
-------
DiskExecutionEnv
    An execution environment that runs code on the local file system and captures
    the output of the execution.

Imports
-------
- subprocess: For running shell commands.
- time: For timing the execution of commands.
- Path: For handling file system paths.
- Optional, Tuple, Union: For type annotations.
- BaseExecutionEnv: For inheriting the base execution environment interface.
- FileStore: For managing file storage.
- FilesDict: For handling collections of files.
"""

import subprocess
import time

from pathlib import Path
from typing import Optional, Tuple, Union

from gpt_engineer.core.base_execution_env import BaseExecutionEnv
from gpt_engineer.core.default.file_store import FileStore
from gpt_engineer.core.files_dict import FilesDict


class DiskExecutionEnv(BaseExecutionEnv):
    """
    An execution environment that runs code on the local file system and captures
    the output of the execution.

    This class is responsible for executing code that is stored on disk. It ensures that
    the necessary entrypoint file exists and then runs the code using a subprocess. If the
    execution is interrupted by the user, it handles the interruption gracefully.

    Attributes
    ----------
    store : FileStore
        An instance of FileStore that manages the storage of files in the execution
        environment.
    """

    def __init__(self, path: Union[str, Path, None] = None):
        self.files = FileStore(path)

    def upload(self, files: FilesDict) -> "DiskExecutionEnv":
        self.files.push(files)
        return self

    def download(self) -> FilesDict:
        return self.files.pull()

    def popen(self, command: str) -> subprocess.Popen:
        p = subprocess.Popen(
            command,
            shell=True,
            cwd=self.files.working_dir,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        return p

    def run(self, command: str, timeout: Optional[int] = None) -> Tuple[str, str, int]:
        start = time.time()
        print("\n--- Start of run ---")
        # while running, also print the stdout and stderr
        p = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            cwd=self.files.working_dir,
            text=True,
            shell=True,
        )
        print("$", command)
        stdout_full, stderr_full = "", ""

        try:
            while p.poll() is None:
                assert p.stdout is not None
                assert p.stderr is not None
                stdout = p.stdout.readline()
                stderr = p.stderr.readline()
                if stdout:
                    print(stdout, end="")
                    stdout_full += stdout
                if stderr:
                    print(stderr, end="")
                    stderr_full += stderr
                if timeout and time.time() - start > timeout:
                    print("Timeout!")
                    p.kill()
                    raise TimeoutError()
        except KeyboardInterrupt:
            print()
            print("Stopping execution.")
            print("Execution stopped.")
            p.kill()
            print()
            print("--- Finished run ---\n")

        return stdout_full, stderr_full, p.returncode



File: gpt_engineer/applications/__init__.py




File: gpt_engineer/applications/cli/collect.py

"""
Module `collect` - Data Handling and RudderStack Integration

This module provides functionalities to handle and send learning data to RudderStack
for the purpose of analysis and to improve the gpt-engineer system. The data is sent
only when the user gives consent to share.

Functions:
    send_learning(learning): Sends learning data to RudderStack.
    collect_learnings(prompt, model, temperature, config, memory, review): Processes and sends learning data.
    collect_and_send_human_review(prompt, model, temperature, config, memory): Collects human feedback and sends it.

Dependencies:
    hashlib: For generating SHA-256 hash.
    typing: For type annotations.
    gpt_engineer.core: Core functionalities of gpt-engineer.
    gpt_engineer.cli.learning: Handles the extraction of learning data.

Notes:
    Data sent to RudderStack is not shared with third parties and is used solely to
    improve gpt-engineer and allow it to handle a broader range of use cases.
    Consent logic is in gpt_engineer/learning.py.
"""

from typing import Tuple

from gpt_engineer.applications.cli.learning import (
    Learning,
    Review,
    extract_learning,
    human_review_input,
)
from gpt_engineer.core.default.disk_memory import DiskMemory
from gpt_engineer.core.prompt import Prompt


def send_learning(learning: Learning):
    """
    Send the learning data to RudderStack for analysis.

    Parameters
    ----------
    learning : Learning
        An instance of the Learning class containing the data to be sent.

    Notes
    -----
    This function is only called if consent is given to share data.
    Data is not shared to a third party. It is used with the sole purpose of
    improving gpt-engineer, and letting it handle more use cases.
    Consent logic is in gpt_engineer/learning.py.
    """
    import rudderstack.analytics as rudder_analytics

    rudder_analytics.write_key = "2Re4kqwL61GDp7S8ewe6K5dbogG"
    rudder_analytics.dataPlaneUrl = "https://gptengineerezm.dataplane.rudderstack.com"

    rudder_analytics.track(
        user_id=learning.session,
        event="learning",
        properties=learning.to_dict(),  # type: ignore
    )


def collect_learnings(
    prompt: Prompt,
    model: str,
    temperature: float,
    config: any,
    memory: DiskMemory,
    review: Review,
):
    """
    Collect the learning data and send it to RudderStack for analysis.

    Parameters
    ----------
    prompt : str
        The initial prompt or question that was provided to the model.
    model : str
        The name of the model used for generating the response.
    temperature : float
        The temperature setting used in the model's response generation.
    config : any
        Configuration parameters used for the learning session.
    memory : DiskMemory
        An instance of DiskMemory for storing and retrieving data.
    review : Review
        An instance of Review containing human feedback on the model's response.

    Notes
    -----
    This function attempts to send the learning data to RudderStack. If the data size exceeds
    the maximum allowed size, it trims the data and retries sending it.
    """
    learnings = extract_learning(prompt, model, temperature, config, memory, review)
    try:
        send_learning(learnings)
    except RuntimeError:
        # try to remove some parts of learning that might be too big
        # rudderstack max event size is 32kb
        max_size = 32 << 10  # 32KB in bytes
        current_size = len(learnings.to_json().encode("utf-8"))  # get size in bytes

        overflow = current_size - max_size

        # Add some extra characters for the "[REMOVED...]" string and for safety margin
        remove_length = overflow + len(f"[REMOVED {overflow} CHARACTERS]") + 100

        learnings.logs = (
            learnings.logs[:-remove_length]
            + f"\n\n[REMOVED {remove_length} CHARACTERS]"
        )

        print(
            "WARNING: learning too big, removing some parts. "
            "Please report if this results in a crash."
        )
        try:
            send_learning(learnings)
        except RuntimeError:
            print(
                "Sending learnings crashed despite truncation. Progressing without saving learnings."
            )


# def steps_file_hash():
#     """
#     Compute the SHA-256 hash of the steps file.
#
#     Returns
#     -------
#     str
#         The SHA-256 hash of the steps file.
#     """
#     with open(steps.__file__, "r") as f:
#         content = f.read()
#         return hashlib.sha256(content.encode("utf-8")).hexdigest()


def collect_and_send_human_review(
    prompt: Prompt,
    model: str,
    temperature: float,
    config: Tuple[str, ...],
    memory: DiskMemory,
):
    """
    Collects human feedback on the code and sends it for analysis.

    Parameters
    ----------
    prompt : str
        The initial prompt or question that was provided to the model.
    model : str
        The name of the model used for generating the response.
    temperature : float
        The temperature setting used in the model's response generation.
    config : Tuple[str, ...]
        Configuration parameters used for the learning session.
    memory : DiskMemory
        An instance of DiskMemory for storing and retrieving data.

    Returns
    -------
    None

    Notes
    -----
    This function prompts the user for a review of the generated or improved code using the
    `human_review_input` function. If a valid review is provided, it's serialized to JSON format
    and stored within the database's memory under the "review" key.
    """

    review = human_review_input()
    if review:
        collect_learnings(prompt, model, temperature, config, memory, review)



File: gpt_engineer/applications/cli/cli_agent.py

"""
This module provides the CliAgent class which manages the lifecycle of code generation and improvement
using an AI model. It includes functionalities to initialize code generation, improve existing code,
and process the code through various steps defined in the step bundle.
"""

from typing import Callable, Optional, TypeVar

# from gpt_engineer.core.default.git_version_manager import GitVersionManager
from gpt_engineer.core.ai import AI
from gpt_engineer.core.base_agent import BaseAgent
from gpt_engineer.core.base_execution_env import BaseExecutionEnv
from gpt_engineer.core.base_memory import BaseMemory
from gpt_engineer.core.default.disk_execution_env import DiskExecutionEnv
from gpt_engineer.core.default.disk_memory import DiskMemory
from gpt_engineer.core.default.paths import PREPROMPTS_PATH
from gpt_engineer.core.default.steps import (
    execute_entrypoint,
    gen_code,
    gen_entrypoint,
    improve_fn,
)
from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.preprompts_holder import PrepromptsHolder
from gpt_engineer.core.prompt import Prompt

CodeGenType = TypeVar("CodeGenType", bound=Callable[[AI, str, BaseMemory], FilesDict])
CodeProcessor = TypeVar(
    "CodeProcessor", bound=Callable[[AI, BaseExecutionEnv, FilesDict], FilesDict]
)
ImproveType = TypeVar(
    "ImproveType", bound=Callable[[AI, str, FilesDict, BaseMemory], FilesDict]
)


class CliAgent(BaseAgent):
    """
    The `CliAgent` class is responsible for managing the lifecycle of code generation and improvement
    using an AI model. It orchestrates the generation of new code and the improvement of existing code
    based on given prompts and utilizes a memory system and execution environment for processing.

    Parameters
    ----------
    memory : BaseMemory
        An instance of a class that adheres to the BaseMemory interface, used for storing and retrieving
        information during the code generation process.
    execution_env : BaseExecutionEnv
        An instance of a class that adheres to the BaseExecutionEnv interface, used for executing code
        and managing the execution environment.
    ai : AI, optional
        An instance of the AI class that manages calls to the language model. If not provided, a default
        instance is created.
    code_gen_fn : CodeGenType, optional
        A callable that takes an AI instance, a prompt, and a memory instance to generate code. Defaults
        to the `gen_code` function.
    improve_fn : ImproveType, optional
        A callable that takes an AI instance, a prompt, a FilesDict instance, and a memory instance to
        improve code. Defaults to the `improve` function.
    process_code_fn : CodeProcessor, optional
        A callable that takes an AI instance, an execution environment, and a FilesDict instance to
        process code. Defaults to the `execute_entrypoint` function.
    preprompts_holder : PrepromptsHolder, optional
        An instance of PrepromptsHolder that manages preprompt templates. If not provided, a default
        instance is created using the PREPROMPTS_PATH.

    Attributes
    ----------
    memory : BaseMemory
        The memory instance where the agent stores and retrieves information.
    execution_env : BaseExecutionEnv
        The execution environment instance where the agent executes and manages code.
    ai : AI
        The AI instance used for interacting with the language model.
    code_gen_fn : CodeGenType
        The function used for generating code.
    improve_fn : ImproveType
        The function used for improving code.
    process_code_fn : CodeProcessor
        The function used for processing code.
    preprompts_holder : PrepromptsHolder
        The holder for preprompt templates.
    """

    def __init__(
        self,
        memory: BaseMemory,
        execution_env: BaseExecutionEnv,
        ai: AI = None,
        code_gen_fn: CodeGenType = gen_code,
        improve_fn: ImproveType = improve_fn,
        process_code_fn: CodeProcessor = execute_entrypoint,
        preprompts_holder: PrepromptsHolder = None,
    ):
        self.memory = memory
        self.execution_env = execution_env
        self.ai = ai or AI()
        self.code_gen_fn = code_gen_fn
        self.process_code_fn = process_code_fn
        self.improve_fn = improve_fn
        self.preprompts_holder = preprompts_holder or PrepromptsHolder(PREPROMPTS_PATH)

    @classmethod
    def with_default_config(
        cls,
        memory: DiskMemory,
        execution_env: DiskExecutionEnv,
        ai: AI = None,
        code_gen_fn: CodeGenType = gen_code,
        improve_fn: ImproveType = improve_fn,
        process_code_fn: CodeProcessor = execute_entrypoint,
        preprompts_holder: PrepromptsHolder = None,
        diff_timeout=3,
    ):
        """
        Creates a new instance of CliAgent with default configurations for memory, execution environment,
        AI, and other functional parameters.

        Parameters
        ----------
        memory : DiskMemory
            An instance of DiskMemory for storing and retrieving information.
        execution_env : DiskExecutionEnv
            An instance of DiskExecutionEnv for executing code.
        ai : AI, optional
            An instance of AI for interacting with the language model. Defaults to None, which will create
            a new AI instance.
        code_gen_fn : CodeGenType, optional
            A function for generating code. Defaults to `gen_code`.
        improve_fn : ImproveType, optional
            A function for improving code. Defaults to `improve`.
        process_code_fn : CodeProcessor, optional
            A function for processing code. Defaults to `execute_entrypoint`.
        preprompts_holder : PrepromptsHolder, optional
            An instance of PrepromptsHolder for managing preprompt templates. Defaults to None, which will
            create a new PrepromptsHolder instance using PREPROMPTS_PATH.

        Returns
        -------
        CliAgent
            An instance of CliAgent configured with the provided or default parameters.
        """
        return cls(
            memory=memory,
            execution_env=execution_env,
            ai=ai,
            code_gen_fn=code_gen_fn,
            process_code_fn=process_code_fn,
            improve_fn=improve_fn,
            preprompts_holder=preprompts_holder or PrepromptsHolder(PREPROMPTS_PATH),
        )

    def init(self, prompt: Prompt) -> FilesDict:
        """
        Generates a new piece of code using the AI and step bundle based on the provided prompt.

        Parameters
        ----------
        prompt : str
            A string prompt that guides the code generation process.

        Returns
        -------
        FilesDict
            An instance of the `FilesDict` class containing the generated code.
        """

        files_dict = self.code_gen_fn(
            self.ai, prompt, self.memory, self.preprompts_holder
        )
        entrypoint = gen_entrypoint(
            self.ai, prompt, files_dict, self.memory, self.preprompts_holder
        )
        combined_dict = {**files_dict, **entrypoint}
        files_dict = FilesDict(combined_dict)
        files_dict = self.process_code_fn(
            self.ai,
            self.execution_env,
            files_dict,
            preprompts_holder=self.preprompts_holder,
            prompt=prompt,
            memory=self.memory,
        )
        return files_dict

    def improve(
        self,
        files_dict: FilesDict,
        prompt: Prompt,
        execution_command: Optional[str] = None,
        diff_timeout=3,
    ) -> FilesDict:
        """
        Improves an existing piece of code using the AI and step bundle based on the provided prompt.

        Parameters
        ----------
        files_dict : FilesDict
            An instance of `FilesDict` containing the code to be improved.
        prompt : str
            A string prompt that guides the code improvement process.
        execution_command : str, optional
            An optional command to execute the code. If not provided, the default execution command is used.

        Returns
        -------
        FilesDict
            An instance of the `FilesDict` class containing the improved code.
        """

        files_dict = self.improve_fn(
            self.ai,
            prompt,
            files_dict,
            self.memory,
            self.preprompts_holder,
            diff_timeout=diff_timeout,
        )
        # entrypoint = gen_entrypoint(
        #     self.ai, prompt, files_dict, self.memory, self.preprompts_holder
        # )
        # combined_dict = {**files_dict, **entrypoint}
        # files_dict = FilesDict(combined_dict)
        # files_dict = self.process_code_fn(
        #     self.ai,
        #     self.execution_env,
        #     files_dict,
        #     preprompts_holder=self.preprompts_holder,
        #     prompt=prompt,
        #     memory=self.memory,
        # )

        return files_dict



File: gpt_engineer/applications/cli/__init__.py




File: gpt_engineer/applications/cli/file_selector.py

"""
file_selector.py

This module offers interactive file selection for projects. Leveraging a terminal-based,
tree-structured display, users can navigate and select files for editing or processing.
It integrates with system editors for direct file modification and supports saving
selections for later use. Designed for efficient workflow enhancement in file-intensive
environments, it offers customizable file filtering and seamless editor integration.

Key Components:
- FileSelector: Manages file selection and interaction.
- DisplayablePath: Provides a structured view of file paths.

Usage:
Typically used in project setup or management phases for selecting specific files.
It operates within the GPT-Engineer environment, relying on core functionalities for
file handling and persistence.
"""

import fnmatch
import os
import subprocess

from pathlib import Path
from typing import Any, Dict, Generator, List, Union

import toml

from gpt_engineer.core.default.disk_memory import DiskMemory
from gpt_engineer.core.default.paths import metadata_path
from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.git import filter_by_gitignore, is_git_repo


class FileSelector:
    """
    Manages file selection and interaction within a project directory.

    This class provides methods to interactively select files from the terminal,
    save selections for later use, and integrate with system editors for direct
    file modification.

    Attributes
    ----------
    IGNORE_FOLDERS : set
        A set of directory names to ignore during file selection.
    FILE_LIST_NAME : str
        The name of the file that stores the selected files list.
    COMMENT : str
        The comment string to be added to the top of the file selection list.
    """

    IGNORE_FOLDERS = {"site-packages", "node_modules", "venv", "__pycache__"}
    FILE_LIST_NAME = "file_selection.toml"
    COMMENT = (
        "# Remove '#' to select a file or turn off linting.\n\n"
        "# Linting with BLACK (Python) enhances code suggestions from LLMs. "
        "To disable linting, uncomment the relevant option in the linting settings.\n\n"
        "# gpt-engineer can only read selected files. "
        "Including irrelevant files will degrade performance, "
        "cost additional tokens and potentially overflow token limit.\n\n"
    )
    LINTING_STRING = '[linting]\n# "linting" = "off"\n\n'
    is_linting = True

    def __init__(self, project_path: Union[str, Path]):
        """
        Initializes the FileSelector with a given project path.

        Parameters
        ----------
        project_path : Union[str, Path]
            The path to the project directory where file selection is to be performed.
        """
        self.project_path = project_path
        self.metadata_db = DiskMemory(metadata_path(self.project_path))
        self.toml_path = self.metadata_db.path / self.FILE_LIST_NAME

    def ask_for_files(self, skip_file_selection=False) -> tuple[FilesDict, bool]:
        """
        Prompts the user to select files for context improvement.

        This method supports selection from the terminal or using a previously saved list.
        In test mode, it retrieves files from a predefined TOML configuration.

        Returns
        -------
        FilesDict
            A dictionary with file paths as keys and file contents as values.
        """

        if os.getenv("GPTE_TEST_MODE") or skip_file_selection:
            # In test mode, retrieve files from a predefined TOML configuration
            # also get from toml if skip_file_selector is active
            assert self.FILE_LIST_NAME in self.metadata_db
            selected_files = self.get_files_from_toml(self.project_path, self.toml_path)
        else:
            # Otherwise, use the editor file selector for interactive selection
            if self.FILE_LIST_NAME in self.metadata_db:
                print(
                    f"File list detected at {self.toml_path}. Edit or delete it if you want to select new files."
                )
                selected_files = self.editor_file_selector(self.project_path, False)
            else:
                selected_files = self.editor_file_selector(self.project_path, True)

        content_dict = {}
        for file_path in selected_files:
            # selected files contains paths that are relative to the project path
            try:
                # to open the file we need the path from the cwd
                with open(
                    Path(self.project_path) / file_path, "r", encoding="utf-8"
                ) as content:
                    content_dict[str(file_path)] = content.read()
            except FileNotFoundError:
                print(f"Warning: File not found {file_path}")
            except UnicodeDecodeError:
                print(f"Warning: File not UTF-8 encoded {file_path}, skipping")

        return FilesDict(content_dict), self.is_linting

    def editor_file_selector(
        self, input_path: Union[str, Path], init: bool = True
    ) -> List[str]:
        """
        Provides an interactive file selection interface using a .toml file.

        Parameters
        ----------
        input_path : Union[str, Path]
            The path where file selection is to be performed.
        init : bool, optional
            Indicates whether to initialize the .toml file with the file tree.

        Returns
        -------
        List[str]
            A list of strings representing the paths of selected files.
        """

        root_path = Path(input_path)
        tree_dict = {}
        toml_file = DiskMemory(metadata_path(input_path)).path / "file_selection.toml"
        # Define the toml file path

        # Initialize .toml file with file tree if in initial state
        if init:
            tree_dict = {x: "selected" for x in self.get_current_files(root_path)}

            s = toml.dumps({"files": tree_dict})

            # add comments on all lines that match = "selected"
            s = "\n".join(
                [
                    "# " + line if line.endswith(' = "selected"') else line
                    for line in s.split("\n")
                ]
            )
            # Write to the toml file
            with open(toml_file, "w") as f:
                f.write(self.COMMENT)
                f.write(self.LINTING_STRING)
                f.write(s)

        else:
            # Load existing files from the .toml configuration
            all_files = self.get_current_files(root_path)
            s = toml.dumps({"files": {x: "selected" for x in all_files}})

            # get linting status from the toml file
            with open(toml_file, "r") as file:
                linting_status = toml.load(file)
            if (
                "linting" in linting_status
                and linting_status["linting"].get("linting", "").lower() == "off"
            ):
                self.is_linting = False
                self.LINTING_STRING = '[linting]\n"linting" = "off"\n\n'
                print("\nLinting is disabled")

            with open(toml_file, "r") as file:
                selected_files = toml.load(file)

            lines = s.split("\n")
            s = "\n".join(
                lines[:1]
                + [
                    line
                    if line.split(" = ")[0].strip('"') in selected_files["files"]
                    else "# " + line
                    for line in lines[1:]
                ]
            )

            # Write the merged list back to the .toml for user review and modification
            with open(toml_file, "w") as file:
                file.write(self.COMMENT)  # Ensure to write the comment
                file.write(self.LINTING_STRING)
                file.write(s)

        print(
            "Please select and deselect (add # in front) files, save it, and close it to continue..."
        )
        self.open_with_default_editor(
            toml_file
        )  # Open the .toml file in the default editor for user modification
        return self.get_files_from_toml(
            input_path, toml_file
        )  # Return the list of selected files after user edits

    def open_with_default_editor(self, file_path: Union[str, Path]):
        """
        Opens a file with the system's default text editor.

        Parameters
        ----------
        file_path : Union[str, Path]
            The path to the file to be opened in the text editor.
        """

        editors = [
            "gedit",
            "notepad",
            "nvim",
            "write",
            "nano",
            "vim",
            "emacs",
        ]  # Putting the beginner-friendly text editor forward
        chosen_editor = os.environ.get("EDITOR")

        # Try the preferred editor first, then fallback to common editors
        if chosen_editor:
            try:
                subprocess.run([chosen_editor, file_path])
                return
            except Exception:
                pass

        for editor in editors:
            try:
                subprocess.run([editor, file_path])
                return
            except Exception:
                continue
        print("No suitable text editor found. Please edit the file manually.")

    def is_utf8(self, file_path: Union[str, Path]) -> bool:
        """
        Checks if the file at the given path is UTF-8 encoded.

        Parameters
        ----------
        file_path : Union[str, Path]
            The path to the file to be checked.

        Returns
        -------
        bool
            True if the file is UTF-8 encoded, False otherwise.
        """

        try:
            with open(file_path, "rb") as file:
                file.read().decode("utf-8")
                return True
        except UnicodeDecodeError:
            return False

    def get_files_from_toml(
        self, input_path: Union[str, Path], toml_file: Union[str, Path]
    ) -> List[str]:
        """
        Retrieves a list of selected files from a .toml configuration file.

        Parameters
        ----------
        input_path : Union[str, Path]
            The path where file selection was performed.
        toml_file : Union[str, Path]
            The path to the .toml file containing the file selection.

        Returns
        -------
        List[str]
            A list of strings representing the paths of selected files.

        Raises
        ------
        Exception
            If no files are selected in the .toml file.
        """
        selected_files = []
        edited_tree = toml.load(toml_file)  # Load the edited .toml file

        # check if users have disabled linting or not
        if (
            "linting" in edited_tree
            and edited_tree["linting"].get("linting", "").lower() == "off"
        ):
            self.is_linting = False
            print("\nLinting is disabled")
        else:
            self.is_linting = True

        # Iterate through the files in the .toml and append selected files to the list
        for file, _ in edited_tree["files"].items():
            selected_files.append(file)

        # Ensure that at least one file is selected, or raise an exception
        if not selected_files:
            raise Exception(
                "No files were selected. Please select at least one file to proceed."
            )

        print(f"\nYou have selected the following files:\n{input_path}")

        project_path = Path(input_path).resolve()
        selected_paths = set(
            project_path.joinpath(file).resolve(strict=False) for file in selected_files
        )

        for displayable_path in DisplayablePath.make_tree(project_path):
            if displayable_path.path in selected_paths:
                p = displayable_path
                while p.parent and p.parent.path not in selected_paths:
                    selected_paths.add(p.parent.path)
                    p = p.parent

        try:
            for displayable_path in DisplayablePath.make_tree(project_path):
                if displayable_path.path in selected_paths:
                    print(displayable_path.displayable())

        except FileNotFoundError:
            print("Specified path does not exist: ", project_path)
        except Exception as e:
            print("An error occurred while trying to display the file tree:", e)

        print("\n")
        return selected_files

    def merge_file_lists(
        self, existing_files: Dict[str, Any], new_files: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Merges two lists of files, preserving the selection status.

        Parameters
        ----------
        existing_files : Dict[str, Any]
            The dictionary of existing files with their properties.
        new_files : Dict[str, Any]
            The dictionary of new files with their properties.

        Returns
        -------
        Dict[str, Any]
            The updated dictionary of files after merging.
        """
        # Update the existing files with any new files or changes
        for file, properties in new_files.items():
            if file not in existing_files:
                existing_files[file] = properties  # Add new files as unselected
            # If you want to update other properties of existing files, you can do so here

        return existing_files

    def should_filter_file(self, file_path: Path, filters: List[str]) -> bool:
        """
        Determines if a file should be ignored based on .gitignore rules.
        """
        for f in filters:
            if fnmatch.fnmatchcase(str(file_path), f):
                return True
        return False

    def get_current_files(self, project_path: Union[str, Path]) -> List[str]:
        """
        Generates a list of all files in the project directory. Will use .gitignore files if project_path is a git repository.

        Parameters
        ----------
        project_path : Union[str, Path]
            The path to the project directory.

        Returns
        -------
        List[str]
            A list of strings representing the relative paths of all files in the project directory.
        """
        all_files = []
        project_path = Path(
            project_path
        ).resolve()  # Ensure path is absolute and resolved

        file_list = project_path.glob("**/*")

        for path in file_list:  # Recursively list all files
            if path.is_file():
                relpath = path.relative_to(project_path)
                parts = relpath.parts
                if any(part.startswith(".") for part in parts):
                    continue  # Skip hidden files
                if any(part in self.IGNORE_FOLDERS for part in parts):
                    continue
                if relpath.name == "prompt":
                    continue  # Skip files named 'prompt'

                all_files.append(str(relpath))

        if is_git_repo(project_path) and "projects" not in project_path.parts:
            all_files = filter_by_gitignore(project_path, all_files)

        return sorted(all_files, key=lambda x: Path(x).as_posix())


class DisplayablePath(object):
    """
    Represents and displays a file system path in a tree-like structure.

    This class is used to visually represent the structure of directories and files
    in a way that is similar to a file explorer's tree view.
    """

    display_filename_prefix_middle = "├── "
    display_filename_prefix_last = "└── "
    display_parent_prefix_middle = "    "
    display_parent_prefix_last = "│   "

    def __init__(
        self, path: Union[str, Path], parent_path: "DisplayablePath", is_last: bool
    ):
        """
        Initializes a DisplayablePath object with a given path and parent.

        Parameters
        ----------
        path : Union[str, Path]
            The file system path to be displayed.
        parent_path : DisplayablePath
            The parent path in the tree structure.
        is_last : bool
            Indicates whether this is the last sibling in the tree structure.
        """
        self.depth = 0
        self.path = Path(str(path))
        self.parent = parent_path
        self.is_last = is_last
        if self.parent:
            self.depth = self.parent.depth + 1  # Increment depth if it has a parent

    @property
    def display_name(self) -> str:
        """
        Get the display name of the file or directory.
        """
        if self.path.is_dir():
            return self.path.name + "/"
        return self.path.name

    @classmethod
    def make_tree(
        cls, root: Union[str, Path], parent=None, is_last=False, criteria=None
    ) -> Generator["DisplayablePath", None, None]:
        """
        Creates a tree of DisplayablePath objects from a root directory.

        Parameters
        ----------
        root : Union[str, Path]
            The root directory from which to start creating the tree.
        parent : DisplayablePath, optional
            The parent path in the tree structure.
        is_last : bool, optional
            Indicates whether this is the last sibling in the tree structure.
        criteria : callable, optional
            A function to filter the paths included in the tree.

        Yields
        ------
        DisplayablePath
            The next DisplayablePath object in the tree.
        """
        root = Path(str(root))  # Ensure root is a Path object
        criteria = criteria or cls._default_criteria
        displayable_root = cls(root, parent, is_last)
        yield displayable_root

        if root.is_dir():  # Check if root is a directory before iterating
            children = sorted(
                list(path for path in root.iterdir() if criteria(path)),
                key=lambda s: str(s).lower(),
            )
            count = 1
            for path in children:
                is_last = count == len(children)
                yield from cls.make_tree(
                    path, parent=displayable_root, is_last=is_last, criteria=criteria
                )
                count += 1

    @classmethod
    def _default_criteria(cls, path: Path) -> bool:
        """
        The default criteria function to filter the paths.
        """
        return True

    def displayable(self) -> str:
        """
        Returns a string representation of the path for display in a tree-like structure.

        Returns
        -------
        str
            The displayable string representation of the file or directory.
        """
        if self.parent is None:
            return self.display_name

        _filename_prefix = (
            self.display_filename_prefix_last
            if self.is_last
            else self.display_filename_prefix_middle
        )

        parts = ["{!s} {!s}".format(_filename_prefix, self.display_name)]

        parent = self.parent
        while parent and parent.parent is not None:
            parts.append(
                self.display_parent_prefix_middle
                if parent.is_last
                else self.display_parent_prefix_last
            )
            parent = parent.parent

        return "".join(reversed(parts))  # Assemble the parts into the final string



File: gpt_engineer/applications/cli/learning.py

"""
The `learning` module is designed to facilitate the collection and storage of user feedback on the outputs generated by the GPT Engineer tool. It provides mechanisms for obtaining user consent, capturing user reviews, and storing this information for future analysis and enhancement of the tool's performance.

Classes
-------
Review : dataclass
    Represents a user's review of the generated code, including whether it ran, was perfect, was useful, and any additional comments.
Learning : dataclass
    Encapsulates the metadata and feedback collected during a session of using the GPT Engineer tool, including the prompt, model, temperature, configuration, logs, session identifier, user review, and timestamp.

Functions
---------
human_review_input() -> Optional[Review]
    Interactively gathers feedback from the user regarding the performance of generated code and returns a Review instance.
check_collection_consent() -> bool
    Checks if the user has previously given consent to store their data and, if not, asks for it.
ask_collection_consent() -> bool
    Prompts the user for consent to store their data for the purpose of improving GPT Engineer.
extract_learning(prompt: Prompt, model: str, temperature: float, config: Tuple[str, ...], memory: DiskMemory, review: Review) -> Learning
    Extracts feedback and session details to create a Learning instance based on the provided parameters.
get_session() -> str
    Retrieves a unique identifier for the current user session, creating one if it does not exist.

Constants
---------
TERM_CHOICES : tuple
    Terminal color choices for user interactive prompts, formatted with termcolor for readability.
"""

import json
import random
import tempfile

from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Optional, Tuple

from dataclasses_json import dataclass_json
from termcolor import colored

from gpt_engineer.core.default.disk_memory import DiskMemory
from gpt_engineer.core.prompt import Prompt


@dataclass_json
@dataclass
class Review:
    """
    A dataclass that represents a user's review of the generated code.

    Attributes
    ----------
    ran : Optional[bool]
        Indicates whether the generated code ran without errors.
    perfect : Optional[bool]
        Indicates whether the generated code met all the user's requirements.
    works : Optional[bool]
        Indicates whether the generated code was useful, even if not perfect.
    comments : str
        Any additional comments provided by the user.
    raw : str
        A raw string representation of the user's responses.
    """

    ran: Optional[bool]
    perfect: Optional[bool]
    works: Optional[bool]
    comments: str
    raw: str


@dataclass_json
@dataclass
class Learning:
    """
    A dataclass that encapsulates the learning data collected during a GPT Engineer session.

    Attributes
    ----------
    prompt : str
        A JSON string representing the prompt provided to GPT Engineer.
    model : str
        The name of the model used during the session.
    temperature : float
        The temperature setting used for the model's responses.
    config : str
        A JSON string representing the configuration settings for the session.
    logs : str
        A JSON string representing the logs of the session.
    session : str
        A unique identifier for the user session.
    review : Optional[Review]
        The user's review of the generated code.
    timestamp : str
        The UTC timestamp when the learning data was created.
    version : str
        The version of the learning data schema.
    """

    prompt: str
    model: str
    temperature: float
    config: str
    logs: str
    session: str
    review: Optional[Review]
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    version: str = "0.3"


TERM_CHOICES = (
    colored("y", "green")
    + "/"
    + colored("n", "red")
    + "/"
    + colored("u", "yellow")
    + "(ncertain): "
)


def human_review_input() -> Optional[Review]:
    """
    Interactively prompts the user to review the generated code and returns their feedback encapsulated in a Review object.

    This function will first check if the user has given consent to collect their feedback. If consent is given, it will ask the user a series of questions about the generated code's performance and capture their responses.

    Returns
    -------
    Optional[Review]
        A Review object containing the user's feedback, or None if consent is not given.
    """
    print()
    if not check_collection_consent():
        return None
    print()
    print(
        colored("To help gpt-engineer learn, please answer 3 questions:", "light_green")
    )
    print()

    ran = input("Did the generated code run at all? " + TERM_CHOICES)
    ran = ask_for_valid_input(ran)

    if ran == "y":
        perfect = input(
            "Did the generated code do everything you wanted? " + TERM_CHOICES
        )
        perfect = ask_for_valid_input(perfect)

        if perfect != "y":
            useful = input("Did the generated code do anything useful? " + TERM_CHOICES)
            useful = ask_for_valid_input(useful)
        else:
            useful = ""
    else:
        perfect = ""
        useful = ""

    if perfect != "y":
        comments = input(
            "If you have time, please explain what was not working "
            + colored("(ok to leave blank)\n", "light_green")
        )
    else:
        comments = ""

    return Review(
        raw=", ".join([ran, perfect, useful]),
        ran={"y": True, "n": False, "u": None, "": None}[ran],
        works={"y": True, "n": False, "u": None, "": None}[useful],
        perfect={"y": True, "n": False, "u": None, "": None}[perfect],
        comments=comments,
    )


def ask_for_valid_input(ran):
    while ran not in ("y", "n", "u"):
        ran = input("Invalid input. Please enter y, n, or u: ")
    return ran


def check_collection_consent() -> bool:
    """
    Checks if the user has previously given consent to store their data for feedback collection.

    This function looks for a file that stores the user's consent status. If the file exists and contains 'true', consent is assumed. If the file does not exist or does not contain 'true', the function will prompt the user for consent.

    Returns
    -------
    bool
        True if the user has given consent, False otherwise.
    """
    path = Path(".gpte_consent")
    if path.exists() and path.read_text() == "true":
        return True
    else:
        return ask_collection_consent()


def ask_collection_consent() -> bool:
    """
    Asks the user for their consent to store their data for the purpose of improving the GPT Engineer tool.

    The user's response is recorded in a file for future reference. If the user consents, the function will write 'true' to the file. If the user does not consent, no data will be collected, and the function will not modify the file.

    Returns
    -------
    bool
        True if the user consents, False otherwise.
    """
    answer = input(
        "Is it ok if we store your prompts to help improve GPT Engineer? (y/n)"
    )
    while answer.lower() not in ("y", "n"):
        answer = input("Invalid input. Please enter y or n: ")

    if answer.lower() == "y":
        path = Path(".gpte_consent")
        path.write_text("true")
        print(colored("Thank you️", "light_green"))
        print()
        print(
            "(If you no longer wish to participate in data collection, delete the file .gpte_consent)"
        )
        return True
    else:
        print(
            colored(
                "No worries! GPT Engineer will not collect your prompts. ❤️",
                "light_green",
            )
        )
        return False


def extract_learning(
    prompt: Prompt,
    model: str,
    temperature: float,
    config: Tuple[str, ...],
    memory: DiskMemory,
    review: Review,
) -> Learning:
    """
    Constructs a Learning object containing the session's metadata and user feedback.

    Parameters
    ----------
    prompt : str
        The initial prompt provided to the GPT Engineer.
    model : str
        The name of the model used during the session.
    temperature : float
        The temperature setting used for the model's responses.
    config : Tuple[str, ...]
        A tuple representing the configuration settings for the session.
    memory : DiskMemory
        An object representing the disk memory used during the session.
    review : Review
        The user's review of the generated code.

    Returns
    -------
    Learning
        An instance of Learning containing all the session details and user feedback.
    """
    return Learning(
        prompt=prompt.to_json(),
        model=model,
        temperature=temperature,
        config=json.dumps(config),
        session=get_session(),
        logs=memory.to_json(),
        review=review,
    )


def get_session() -> str:
    """
    Retrieves or generates a unique identifier for the current user session.

    This function attempts to read a unique user ID from a temporary file. If the file does not exist, it generates a new random ID, writes it to the file, and returns it. This ID is used to uniquely identify the user's session.

    Returns
    -------
    str
        A unique identifier for the user session.
    """
    path = Path(tempfile.gettempdir()) / "gpt_engineer_user_id.txt"

    try:
        if path.exists():
            user_id = path.read_text()
        else:
            # random uuid:
            user_id = str(random.randint(0, 2**32))
            path.write_text(user_id)
        return user_id
    except IOError:
        return "ephemeral_" + str(random.randint(0, 2**32))



File: gpt_engineer/applications/cli/main.py

"""
Entrypoint for the CLI tool.

This module serves as the entry point for a command-line interface (CLI) tool.
It is designed to interact with OpenAI's language models.
The module provides functionality to:
- Load necessary environment variables,
- Configure various parameters for the AI interaction,
- Manage the generation or improvement of code projects.

Main Functionality
------------------
- Load environment variables required for OpenAI API interaction.
- Parse user-specified parameters for project configuration and AI behavior.
- Facilitate interaction with AI models, databases, and archival processes.

Parameters
----------
None

Notes
-----
- The `OPENAI_API_KEY` must be set in the environment or provided in a `.env` file within the working directory.
- The default project path is `projects/example`.
- When using the `azure_endpoint` parameter, provide the Azure OpenAI service endpoint URL.
"""

import difflib
import json
import logging
import os
import platform
import subprocess
import sys

from pathlib import Path

import openai
import typer

from dotenv import load_dotenv
from langchain.globals import set_llm_cache
from langchain_community.cache import SQLiteCache
from termcolor import colored

from gpt_engineer.applications.cli.cli_agent import CliAgent
from gpt_engineer.applications.cli.collect import collect_and_send_human_review
from gpt_engineer.applications.cli.file_selector import FileSelector
from gpt_engineer.core.ai import AI, ClipboardAI
from gpt_engineer.core.default.disk_execution_env import DiskExecutionEnv
from gpt_engineer.core.default.disk_memory import DiskMemory
from gpt_engineer.core.default.file_store import FileStore
from gpt_engineer.core.default.paths import PREPROMPTS_PATH, memory_path
from gpt_engineer.core.default.steps import (
    execute_entrypoint,
    gen_code,
    handle_improve_mode,
    improve_fn as improve_fn,
)
from gpt_engineer.core.files_dict import FilesDict
from gpt_engineer.core.git import stage_uncommitted_to_git
from gpt_engineer.core.preprompts_holder import PrepromptsHolder
from gpt_engineer.core.prompt import Prompt
from gpt_engineer.tools.custom_steps import clarified_gen, lite_gen, self_heal

app = typer.Typer(
    context_settings={"help_option_names": ["-h", "--help"]}
)  # creates a CLI app


def load_env_if_needed():
    """
    Load environment variables if the OPENAI_API_KEY is not already set.

    This function checks if the OPENAI_API_KEY environment variable is set,
    and if not, it attempts to load it from a .env file in the current working
    directory. It then sets the openai.api_key for use in the application.
    """
    # We have all these checks for legacy reasons...
    if os.getenv("OPENAI_API_KEY") is None:
        load_dotenv()
    if os.getenv("OPENAI_API_KEY") is None:
        load_dotenv(dotenv_path=os.path.join(os.getcwd(), ".env"))

    openai.api_key = os.getenv("OPENAI_API_KEY")

    if os.getenv("ANTHROPIC_API_KEY") is None:
        load_dotenv()
    if os.getenv("ANTHROPIC_API_KEY") is None:
        load_dotenv(dotenv_path=os.path.join(os.getcwd(), ".env"))


def concatenate_paths(base_path, sub_path):
    # Compute the relative path from base_path to sub_path
    relative_path = os.path.relpath(sub_path, base_path)

    # If the relative path is not in the parent directory, use the original sub_path
    if not relative_path.startswith(".."):
        return sub_path

    # Otherwise, concatenate base_path and sub_path
    return os.path.normpath(os.path.join(base_path, sub_path))


def load_prompt(
    input_repo: DiskMemory,
    improve_mode: bool,
    prompt_file: str,
    image_directory: str,
    entrypoint_prompt_file: str = "",
) -> Prompt:
    """
    Load or request a prompt from the user based on the mode.

    Parameters
    ----------
    input_repo : DiskMemory
        The disk memory object where prompts and other data are stored.
    improve_mode : bool
        Flag indicating whether the application is in improve mode.

    Returns
    -------
    str
        The loaded or inputted prompt.
    """

    if os.path.isdir(prompt_file):
        raise ValueError(
            f"The path to the prompt, {prompt_file}, already exists as a directory. No prompt can be read from it. Please specify a prompt file using --prompt_file"
        )
    prompt_str = input_repo.get(prompt_file)
    if prompt_str:
        print(colored("Using prompt from file:", "green"), prompt_file)
        print(prompt_str)
    else:
        if not improve_mode:
            prompt_str = input(
                "\nWhat application do you want gpt-engineer to generate?\n"
            )
        else:
            prompt_str = input("\nHow do you want to improve the application?\n")

    if entrypoint_prompt_file == "":
        entrypoint_prompt = ""
    else:
        full_entrypoint_prompt_file = concatenate_paths(
            input_repo.path, entrypoint_prompt_file
        )
        if os.path.isfile(full_entrypoint_prompt_file):
            entrypoint_prompt = input_repo.get(full_entrypoint_prompt_file)

        else:
            raise ValueError("The provided file at --entrypoint-prompt does not exist")

    if image_directory == "":
        return Prompt(prompt_str, entrypoint_prompt=entrypoint_prompt)

    full_image_directory = concatenate_paths(input_repo.path, image_directory)
    if os.path.isdir(full_image_directory):
        if len(os.listdir(full_image_directory)) == 0:
            raise ValueError("The provided --image_directory is empty.")
        image_repo = DiskMemory(full_image_directory)
        return Prompt(
            prompt_str,
            image_repo.get(".").to_dict(),
            entrypoint_prompt=entrypoint_prompt,
        )
    else:
        raise ValueError("The provided --image_directory is not a directory.")


def get_preprompts_path(use_custom_preprompts: bool, input_path: Path) -> Path:
    """
    Get the path to the preprompts, using custom ones if specified.

    Parameters
    ----------
    use_custom_preprompts : bool
        Flag indicating whether to use custom preprompts.
    input_path : Path
        The path to the project directory.

    Returns
    -------
    Path
        The path to the directory containing the preprompts.
    """
    original_preprompts_path = PREPROMPTS_PATH
    if not use_custom_preprompts:
        return original_preprompts_path

    custom_preprompts_path = input_path / "preprompts"
    if not custom_preprompts_path.exists():
        custom_preprompts_path.mkdir()

    for file in original_preprompts_path.glob("*"):
        if not (custom_preprompts_path / file.name).exists():
            (custom_preprompts_path / file.name).write_text(file.read_text())
    return custom_preprompts_path


def compare(f1: FilesDict, f2: FilesDict):
    def colored_diff(s1, s2):
        lines1 = s1.splitlines()
        lines2 = s2.splitlines()

        diff = difflib.unified_diff(lines1, lines2, lineterm="")

        RED = "\033[38;5;202m"
        GREEN = "\033[92m"
        RESET = "\033[0m"

        colored_lines = []
        for line in diff:
            if line.startswith("+"):
                colored_lines.append(GREEN + line + RESET)
            elif line.startswith("-"):
                colored_lines.append(RED + line + RESET)
            else:
                colored_lines.append(line)

        return "\n".join(colored_lines)

    for file in sorted(set(f1) | set(f2)):
        diff = colored_diff(f1.get(file, ""), f2.get(file, ""))
        if diff:
            print(f"Changes to {file}:")
            print(diff)


def prompt_yesno() -> bool:
    TERM_CHOICES = colored("y", "green") + "/" + colored("n", "red") + " "
    while True:
        response = input(TERM_CHOICES).strip().lower()
        if response in ["y", "yes"]:
            return True
        if response in ["n", "no"]:
            break
        print("Please respond with 'y' or 'n'")


def get_system_info():
    system_info = {
        "os": platform.system(),
        "os_version": platform.version(),
        "architecture": platform.machine(),
        "python_version": sys.version,
        "packages": format_installed_packages(get_installed_packages()),
    }
    return system_info


def get_installed_packages():
    try:
        result = subprocess.run(
            [sys.executable, "-m", "pip", "list", "--format=json"],
            capture_output=True,
            text=True,
        )
        packages = json.loads(result.stdout)
        return {pkg["name"]: pkg["version"] for pkg in packages}
    except Exception as e:
        return str(e)


def format_installed_packages(packages):
    return "\n".join([f"{name}: {version}" for name, version in packages.items()])


@app.command(
    help="""
        GPT-engineer lets you:

        \b
        - Specify a software in natural language
        - Sit back and watch as an AI writes and executes the code
        - Ask the AI to implement improvements
    """
)
def main(
    project_path: str = typer.Argument(".", help="path"),
    model: str = typer.Option(
        os.environ.get("MODEL_NAME", "gpt-4o"), "--model", "-m", help="model id string"
    ),
    temperature: float = typer.Option(
        0.1,
        "--temperature",
        "-t",
        help="Controls randomness: lower values for more focused, deterministic outputs",
    ),
    improve_mode: bool = typer.Option(
        False,
        "--improve",
        "-i",
        help="Improve an existing project by modifying the files.",
    ),
    lite_mode: bool = typer.Option(
        False,
        "--lite",
        "-l",
        help="Lite mode: run a generation using only the main prompt.",
    ),
    clarify_mode: bool = typer.Option(
        False,
        "--clarify",
        "-c",
        help="Clarify mode - discuss specification with AI before implementation.",
    ),
    self_heal_mode: bool = typer.Option(
        False,
        "--self-heal",
        "-sh",
        help="Self-heal mode - fix the code by itself when it fails.",
    ),
    azure_endpoint: str = typer.Option(
        "",
        "--azure",
        "-a",
        help="""Endpoint for your Azure OpenAI Service (https://xx.openai.azure.com).
            In that case, the given model is the deployment name chosen in the Azure AI Studio.""",
    ),
    use_custom_preprompts: bool = typer.Option(
        False,
        "--use-custom-preprompts",
        help="""Use your project's custom preprompts instead of the default ones.
          Copies all original preprompts to the project's workspace if they don't exist there.""",
    ),
    llm_via_clipboard: bool = typer.Option(
        False,
        "--llm-via-clipboard",
        help="Use the clipboard to communicate with the AI.",
    ),
    verbose: bool = typer.Option(
        False, "--verbose", "-v", help="Enable verbose logging for debugging."
    ),
    debug: bool = typer.Option(
        False, "--debug", "-d", help="Enable debug mode for debugging."
    ),
    prompt_file: str = typer.Option(
        "prompt",
        "--prompt_file",
        help="Relative path to a text file containing a prompt.",
    ),
    entrypoint_prompt_file: str = typer.Option(
        "",
        "--entrypoint_prompt",
        help="Relative path to a text file containing a file that specifies requirements for you entrypoint.",
    ),
    image_directory: str = typer.Option(
        "",
        "--image_directory",
        help="Relative path to a folder containing images.",
    ),
    use_cache: bool = typer.Option(
        False,
        "--use_cache",
        help="Speeds up computations and saves tokens when running the same prompt multiple times by caching the LLM response.",
    ),
    skip_file_selection: bool = typer.Option(
        False,
        "--skip-file-selection",
        "-s",
        help="Skip interactive file selection in improve mode and use the generated TOML file directly.",
    ),
    no_execution: bool = typer.Option(
        False,
        "--no_execution",
        help="Run setup but to not call LLM or write any code. For testing purposes.",
    ),
    sysinfo: bool = typer.Option(
        False,
        "--sysinfo",
        help="Output system information for debugging",
    ),
    diff_timeout: int = typer.Option(
        3,
        "--diff_timeout",
        help="Diff regexp timeout. Default: 3. Increase if regexp search timeouts.",
    ),
):
    """
    The main entry point for the CLI tool that generates or improves a project.

    This function sets up the CLI tool, loads environment variables, initializes
    the AI, and processes the user's request to generate or improve a project
    based on the provided arguments.

    Parameters
    ----------
    project_path : str
        The file path to the project directory.
    model : str
        The model ID string for the AI.
    temperature : float
        The temperature setting for the AI's responses.
    improve_mode : bool
        Flag indicating whether to improve an existing project.
    lite_mode : bool
        Flag indicating whether to run in lite mode.
    clarify_mode : bool
        Flag indicating whether to discuss specifications with AI before implementation.
    self_heal_mode : bool
        Flag indicating whether to enable self-healing mode.
    azure_endpoint : str
        The endpoint for Azure OpenAI services.
    use_custom_preprompts : bool
        Flag indicating whether to use custom preprompts.
    prompt_file : str
        Relative path to a text file containing a prompt.
    entrypoint_prompt_file: str
        Relative path to a text file containing a file that specifies requirements for you entrypoint.
    image_directory: str
        Relative path to a folder containing images.
    use_cache: bool
        Speeds up computations and saves tokens when running the same prompt multiple times by caching the LLM response.
    verbose : bool
        Flag indicating whether to enable verbose logging.
    skip_file_selection: bool
        Skip interactive file selection in improve mode and use the generated TOML file directly
    no_execution: bool
        Run setup but to not call LLM or write any code. For testing purposes.
    sysinfo: bool
        Flag indicating whether to output system information for debugging.

    Returns
    -------
    None
    """

    if debug:
        import pdb

        sys.excepthook = lambda *_: pdb.pm()

    if sysinfo:
        sys_info = get_system_info()
        for key, value in sys_info.items():
            print(f"{key}: {value}")
        raise typer.Exit()

    # Validate arguments
    if improve_mode and (clarify_mode or lite_mode):
        typer.echo("Error: Clarify and lite mode are not compatible with improve mode.")
        raise typer.Exit(code=1)

    # Set up logging
    logging.basicConfig(level=logging.DEBUG if verbose else logging.INFO)
    if use_cache:
        set_llm_cache(SQLiteCache(database_path=".langchain.db"))
    if improve_mode:
        assert not (
            clarify_mode or lite_mode
        ), "Clarify and lite mode are not active for improve mode"

    load_env_if_needed()

    if llm_via_clipboard:
        ai = ClipboardAI()
    else:
        ai = AI(
            model_name=model,
            temperature=temperature,
            azure_endpoint=azure_endpoint,
        )

    path = Path(project_path)
    print("Running gpt-engineer in", path.absolute(), "\n")

    prompt = load_prompt(
        DiskMemory(path),
        improve_mode,
        prompt_file,
        image_directory,
        entrypoint_prompt_file,
    )

    # todo: if ai.vision is false and not llm_via_clipboard - ask if they would like to use gpt-4-vision-preview instead? If so recreate AI
    if not ai.vision:
        prompt.image_urls = None

    # configure generation function
    if clarify_mode:
        code_gen_fn = clarified_gen
    elif lite_mode:
        code_gen_fn = lite_gen
    else:
        code_gen_fn = gen_code

    # configure execution function
    if self_heal_mode:
        execution_fn = self_heal
    else:
        execution_fn = execute_entrypoint

    preprompts_holder = PrepromptsHolder(
        get_preprompts_path(use_custom_preprompts, Path(project_path))
    )

    memory = DiskMemory(memory_path(project_path))
    memory.archive_logs()

    execution_env = DiskExecutionEnv()
    agent = CliAgent.with_default_config(
        memory,
        execution_env,
        ai=ai,
        code_gen_fn=code_gen_fn,
        improve_fn=improve_fn,
        process_code_fn=execution_fn,
        preprompts_holder=preprompts_holder,
    )

    files = FileStore(project_path)
    if not no_execution:
        if improve_mode:
            files_dict_before, is_linting = FileSelector(project_path).ask_for_files(
                skip_file_selection=skip_file_selection
            )

            # lint the code
            if is_linting:
                files_dict_before = files.linting(files_dict_before)

            files_dict = handle_improve_mode(
                prompt, agent, memory, files_dict_before, diff_timeout=diff_timeout
            )
            if not files_dict or files_dict_before == files_dict:
                print(
                    f"No changes applied. Could you please upload the debug_log_file.txt in {memory.path}/logs folder in a github issue?"
                )

            else:
                print("\nChanges to be made:")
                compare(files_dict_before, files_dict)

                print()
                print(colored("Do you want to apply these changes?", "light_green"))
                if not prompt_yesno():
                    files_dict = files_dict_before

        else:
            files_dict = agent.init(prompt)
            # collect user feedback if user consents
            config = (code_gen_fn.__name__, execution_fn.__name__)
            collect_and_send_human_review(prompt, model, temperature, config, memory)

        stage_uncommitted_to_git(path, files_dict, improve_mode)

        files.push(files_dict)

    if ai.token_usage_log.is_openai_model():
        print("Total api cost: $ ", ai.token_usage_log.usage_cost())
    elif os.getenv("LOCAL_MODEL"):
        print("Total api cost: $ 0.0 since we are using local LLM.")
    else:
        print("Total tokens used: ", ai.token_usage_log.total_tokens())


if __name__ == "__main__":
    app()



# gpt-engineer

[![GitHub Repo stars](https://img.shields.io/github/stars/gpt-engineer-org/gpt-engineer?style=social)](https://github.com/gpt-engineer-org/gpt-engineer)
[![Discord Follow](https://dcbadge.vercel.app/api/server/8tcDQ89Ej2?style=flat)](https://discord.gg/8tcDQ89Ej2)
[![License](https://img.shields.io/github/license/gpt-engineer-org/gpt-engineer)](https://github.com/gpt-engineer-org/gpt-engineer/blob/main/LICENSE)
[![GitHub Issues or Pull Requests](https://img.shields.io/github/issues/gpt-engineer-org/gpt-engineer)](https://github.com/gpt-engineer-org/gpt-engineer/issues)
![GitHub Release](https://img.shields.io/github/v/release/gpt-engineer-org/gpt-engineer)
[![Twitter Follow](https://img.shields.io/twitter/follow/antonosika?style=social)](https://twitter.com/antonosika)

gpt-engineer lets you:
- Specify software in natural language
- Sit back and watch as an AI writes and executes the code
- Ask the AI to implement improvements

## Getting Started

### Install gpt-engineer

For **stable** release:

- `python -m pip install gpt-engineer`

For **development**:
- `git clone https://github.com/gpt-engineer-org/gpt-engineer.git`
- `cd gpt-engineer`
- `poetry install`
- `poetry shell` to activate the virtual environment

We actively support Python 3.10 - 3.12. The last version to support Python 3.8 - 3.9 was [0.2.6](https://pypi.org/project/gpt-engineer/0.2.6/).

### Setup API key

Choose **one** of:
- Export env variable (you can add this to .bashrc so that you don't have to do it each time you start the terminal)
    - `export OPENAI_API_KEY=[your api key]`
- .env file:
    - Create a copy of `.env.template` named `.env`
    - Add your OPENAI_API_KEY in .env
- Custom model:
    - See [docs](https://gpt-engineer.readthedocs.io/en/latest/open_models.html), supports local model, azure, etc.

Check the [Windows README](./WINDOWS_README.md) for Windows usage.

**Other ways to run:**
- Use Docker ([instructions](docker/README.md))
- Do everything in your browser:
[![Open in GitHub Codespaces](https://github.com/codespaces/badge.svg)](https://github.com/gpt-engineer-org/gpt-engineer/codespaces)

### Create new code (default usage)
- Create an empty folder for your project anywhere on your computer
- Create a file called `prompt` (no extension) inside your new folder and fill it with instructions
- Run `gpte <project_dir>` with a relative path to your folder
  - For example: `gpte projects/my-new-project` from the gpt-engineer directory root with your new folder in `projects/`

### Improve existing code
- Locate a folder with code which you want to improve anywhere on your computer
- Create a file called `prompt` (no extension) inside your new folder and fill it with instructions for how you want to improve the code
- Run `gpte <project_dir> -i` with a relative path to your folder
  - For example: `gpte projects/my-old-project -i` from the gpt-engineer directory root with your folder in `projects/`

### Benchmark custom agents
- gpt-engineer installs the binary 'bench', which gives you a simple interface for benchmarking your own agent implementations against popular public datasets.
- The easiest way to get started with benchmarking is by checking out the [template](https://github.com/gpt-engineer-org/gpte-bench-template) repo, which contains detailed instructions and an agent template.
- Currently supported benchmark:
  - [APPS](https://github.com/hendrycks/apps)
  - [MBPP](https://github.com/google-research/google-research/tree/master/mbpp)

By running gpt-engineer, you agree to our [terms](https://github.com/gpt-engineer-org/gpt-engineer/blob/main/TERMS_OF_USE.md).


## Relation to gptengineer.app (GPT Engineer)
[gptengineer.app](https://gptengineer.app/) is a commercial project for the automatic generation of web apps.
It features a UI for non-technical users connected to a git-controlled codebase.
The gptengineer.app team is actively supporting the open source community.


## Features

### Pre Prompts
You can specify the "identity" of the AI agent by overriding the `preprompts` folder with your own version of the `preprompts`. You can do so via the `--use-custom-preprompts` argument.

Editing the `preprompts` is how you make the agent remember things between projects.

### Vision

By default, gpt-engineer expects text input via a `prompt` file. It can also accept image inputs for vision-capable models. This can be useful for adding UX or architecture diagrams as additional context for GPT Engineer. You can do this by specifying an image directory with the `—-image_directory` flag and setting a vision-capable model in the second CLI argument.

E.g. `gpte projects/example-vision gpt-4-vision-preview --prompt_file prompt/text --image_directory prompt/images -i`

### Open source, local and alternative models

By default, gpt-engineer supports OpenAI Models via the OpenAI API or Azure OpenAI API, as well as Anthropic models.

With a little extra setup, you can also run with open source models like WizardCoder. See the [documentation](https://gpt-engineer.readthedocs.io/en/latest/open_models.html) for example instructions.

## Mission

The gpt-engineer community mission is to **maintain tools that coding agent builders can use and facilitate collaboration in the open source community**.

If you are interested in contributing to this, we are interested in having you.

If you want to see our broader ambitions, check out the [roadmap](https://github.com/gpt-engineer-org/gpt-engineer/blob/main/ROADMAP.md), and join
[discord](https://discord.gg/8tcDQ89Ej2)
to learn how you can [contribute](.github/CONTRIBUTING.md) to it.

gpt-engineer is [governed](https://github.com/gpt-engineer-org/gpt-engineer/blob/main/GOVERNANCE.md) by a board of long-term contributors. If you contribute routinely and have an interest in shaping the future of gpt-engineer, you will be considered for the board.

## Significant contributors
<ul style="list-style-type: none; padding: 0; display: flex; flex-wrap: wrap;"> <li style="margin-right: 10px; margin-bottom: 10px;"> <a href="https://github.com/ATheorell"> <img src="https://avatars.githubusercontent.com/u/143704446?s=64&v=4" alt="@ATheorell" width="32" height="32" style="border-radius: 50%;"> @ATheorell </a> </li> <li style="margin-right: 10px; margin-bottom: 10px;"> <a href="https://github.com/AntonOsika"> <img src="https://avatars.githubusercontent.com/u/4467025?s=64&v=4" alt="@AntonOsika" width="32" height="32" style="border-radius: 50%;"> @AntonOsika </a> </li> <li style="margin-right: 10px; margin-bottom: 10px;"> <a href="https://github.com/similato87"> <img src="https://avatars.githubusercontent.com/u/71301573?s=64&v=4" alt="@similato87" width="32" height="32" style="border-radius: 50%;"> @similato87 </a> </li> <li style="margin-right: 10px; margin-bottom: 10px;"> <a href="https://github.com/TheoMcCabe"> <img src="https://avatars.githubusercontent.com/u/9841960?s=64&v=4" alt="@TheoMcCabe" width="32" height="32" style="border-radius: 50%;"> @TheoMcCabe </a> </li> <li style="margin-right: 10px; margin-bottom: 10px;"> <a href="https://github.com/captivus"> <img src="https://avatars.githubusercontent.com/u/366332?s=64&v=4" alt="@captivus" width="32" height="32" style="border-radius: 50%;"> @captivus </a> </li> </ul>


## Example



https://github.com/gpt-engineer-org/gpt-engineer/assets/4467025/40d0a9a8-82d0-4432-9376-136df0d57c99
