GPT Engineer 实践与源码解析

import os
import json
import pydicom
import torch
from PIL import Image
from torchvision.transforms import ToTensor
from torch.utils.data import Dataset

class DicomDataset(Dataset):
    def __init__(self, image_dir, label_dir):
        self.image_dir = image_dir
        self.label_dir = label_dir
        self.image_files = os.listdir(image_dir)

    def __len__(self):
        return len(self.image_files)

    def __getitem__(self, idx):
        image_file = self.image_files[idx]
        image_path = os.path.join(self.image_dir, image_file)
        label_path = os.path.join(self.label_dir, image_file.replace('.dcm', '.json'))

        # Load DICOM image

        dicom_image = pydicom.dcmread(image_path).pixel_array
        # Convert DICOM image to tensor

        image = ToTensor()(Image.fromarray(dicom_image))

        # Load label

        with open(label_path, 'r') as f:
            label = json.load(f)['label']

        return image, label

import torch
from torch.utils.data import DataLoader
from src.dicom_dataset import DicomDataset

# Initialize dataset and dataloader

dataset = DicomDataset('path/to/image_dir', 'path/to/label_dir')
dataloader = DataLoader(dataset, batch_size=32, shuffle=True)

# Loop over the dataset

for images, labels in dataloader:
    # Your training code here

    pass

def main():
    # 选择本次运行的模式，默认模式 DEFAULT 就是新建项目, IMPROVE_CODE 就是优化已有项目的代码

    steps_config = StepsConfig.DEFAULT
    if lite_mode:
        if steps_config == StepsConfig.DEFAULT:
            steps_config = StepsConfig.LITE

    if improve_mode:
        steps_config = StepsConfig.IMPROVE_CODE

    ai = AI(
        model_name=model,
        temperature=temperature,
        azure_endpoint=azure_endpoint,
    )

    dbs = DBs(
        memory=DB(memory_path),
        logs=DB(memory_path / "logs"),
        input=DB(input_path),
        workspace=DB(workspace_path),
        preprompts=DB(preprompts_path(use_custom_preprompts, input_path)),
        archive=DB(archive_path),
        project_metadata=DB(project_metadata_path),
    )

    if steps_config not in [
        StepsConfig.EXECUTE_ONLY,
        StepsConfig.USE_FEEDBACK,
        StepsConfig.EVALUATE,
        StepsConfig.IMPROVE_CODE,
    ]:
        archive(dbs)
        load_prompt(dbs)

    steps = STEPS[steps_config]
    # 执行模式对应的步骤

    for step in steps:
        messages = step(ai, dbs)

def simple_gen(ai: AI, dbs: DBs) -> List[Message]:
    messages = ai.start(setup_sys_prompt(dbs), dbs.input["prompt"], step_name=curr_fn())
    to_files_and_memory(messages[-1].content.strip(), dbs)
    return messages

# 基于用户的 prompt 构造 Message 发送 ChatGPT 请求

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

def next(
    self,
    messages: List[Message],
    prompt: Optional[str] = None,
    *,
    step_name: str,
) -> List[Message]:
    if prompt:
        messages.append(self.fuser(prompt))

    callbacks = [StreamingStdOutCallbackHandler()]
    response = self.backoff_inference(messages, callbacks)

    self.update_token_usage_log(
        messages=messages, answer=response.content, step_name=step_name
    )
    messages.append(response)
    return messages

def backoff_inference(self, messages, callbacks):
    return self.llm(messages, callbacks=callbacks)  # type: ignore

from langchain.chat_models import AzureChatOpenAI, ChatOpenAI

# 创建 ChatOpenAI 对象，用于发起 ChatGPT 网络请求

def create_chat_model(self, model: str, temperature) -> BaseChatModel:
    if self.azure_endpoint:
        return AzureChatOpenAI(
            openai_api_base=self.azure_endpoint,
            openai_api_version="2023-05-15",  # might need to be flexible in the future
            deployment_name=model,
            openai_api_type="azure",
            streaming=True,
        )

    return ChatOpenAI(
        model=model,
        temperature=temperature,
        streaming=True,
        client=openai.ChatCompletion,
    )

self.llm = create_chat_model(self, self.model_name, self.temperature)

def setup_sys_prompt(dbs: DBs) -> str:
    return (
        dbs.preprompts["roadmap"]
        + dbs.preprompts["generate"].replace("FILE_FORMAT", dbs.preprompts["file_format"])
        + "\nUseful to know:\n"
        + dbs.preprompts["philosophy"]
    )

def gen_entrypoint(ai: AI, dbs: DBs) -> List[dict]:
    messages = ai.start(
        system=(
            "You will get information about a codebase that is currently on disk in "
            "the current folder.\n"
            "From this you will answer with code blocks that includes all the necessary "
            "unix terminal commands to "
            "a) install dependencies "
            "b) run all necessary parts of the codebase (in parallel if necessary).\n"
            "Do not install globally. Do not use sudo.\n"
            "Do not explain the code, just give the commands.\n"
            "Do not use placeholders, use example values (like . for a folder argument) "
            "if necessary.\n"
        ),
        user="Information about the codebase:\n\n" + dbs.memory["all_output.txt"],
        step_name=curr_fn(),
    )

    regex = r"```\S*\n(.+?)```"
    matches = re.finditer(regex, messages[-1].content.strip(), re.DOTALL)
    dbs.workspace["run.sh"] = "\n".join(match.group(1) for match in matches)
    return messages