Codegen

Créez et évaluez des pipelines de génération de code avec W&B Weave pour tracer les prompts, les résultats et les métriques de qualité.

Il s’agit d’un notebook interactif. Vous pouvez l’exécuter localement ou utiliser les liens ci-dessous :

• Ouvrir dans Google Colab
• Voir la source sur GitHub

​

​

Génération de code avec Weave et OpenAI

Générer du code de haute qualité, avec une structure, une documentation et des tests appropriés, est une tâche complexe. Ce guide montre comment mettre en œuvre un pipeline de génération de code. Vous apprendrez à créer un pipeline de génération de code capable de produire des fonctions Python de haute qualité pour la suite de tests humaneval.

Nous utiliserons Weave pour comparer les évaluations et assurer le suivi, ainsi que les modèles GPT d’OpenAI pour générer du code à l’aide de sorties structurées.

​

Pourquoi utiliser Weave ?

Dans ce tutoriel, nous utiliserons Weave pour implémenter et évaluer un pipeline de génération de code. Vous apprendrez à :

1. Suivre votre pipeline LLM : consigner les entrées, les sorties et les étapes intermédiaires de votre processus de génération de code.
2. Évaluer les sorties du LLM : créer et comparer des évaluations de votre code généré à l’aide d’outils de débogage avancés et de visualisations.

​

Configurer l’environnement

Commencez par configurer votre environnement et importer les bibliothèques nécessaires :

!pip install -qU autopep8 autoflake weave isort openai set-env-colab-kaggle-dotenv datasets
python
%%capture
# Solution temporaire pour corriger un bug dans openai :
# TypeError: Client.__init__() got an unexpected keyword argument 'proxies'
# Voir https://community.openai.com/t/error-with-openai-1-56-0-client-init-got-an-unexpected-keyword-argument-proxies/1040332/15
!pip install "httpx<0.28"
python
import ast
import os
import re
import subprocess
import tempfile
import traceback

import autopep8
import isort
from autoflake import fix_code
from datasets import load_dataset
from openai import OpenAI
from pydantic import BaseModel
from set_env import set_env

import weave
from weave import Dataset, Evaluation

set_env("WANDB_API_KEY")
set_env("OPENAI_API_KEY")
python
WEAVE_PROJECT = "codegen-cookbook-example"
weave.init(WEAVE_PROJECT)
python
client = OpenAI()
python
human_eval = load_dataset("openai_humaneval")
selected_examples = human_eval["test"][:3]

Weave suit automatiquement les appels à l’API OpenAI, y compris les entrées, les sorties et les métadonnées. Vous n’avez donc pas besoin d’ajouter de code de journalisation pour vos interactions avec OpenAI : Weave s’en charge de façon transparente en arrière-plan.

​

Tirer parti des sorties structurées et des modèles Pydantic

Dans ce pipeline de génération de code, nous utilisons le mode de sorties structurées d’OpenAI et des modèles Pydantic pour garantir des réponses cohérentes et correctement formatées de la part du modèle de langage. Cette approche offre plusieurs avantages :

1. Sécurité des types : en définissant des modèles Pydantic pour nos sorties attendues, nous appliquons une structure stricte au code généré, aux exécutants de programme et aux tests unitaires.
2. Analyse simplifiée : le mode de sortie structurée nous permet d’interpréter directement la réponse du modèle dans nos modèles Pydantic prédéfinis, ce qui réduit le besoin de post-traitements complexes.
3. Fiabilité accrue : en spécifiant le format exact attendu, nous réduisons le risque d’obtenir des sorties inattendues ou mal formées du modèle de langage.

Voici un exemple de la façon dont nous définissons nos modèles Pydantic et les utilisons avec les sorties structurées d’OpenAI :

class GeneratedCode(BaseModel):
    function_signature: str
    function_args_with_docstring_within_triple_quotes: str
    code_logic: str

class FormattedGeneratedCode(BaseModel):
    full_code: str

​

Implémenter un formateur de code

Pour garantir un code produit cohérent et propre, nous implémentons une classe CodeFormatter à l’aide d’opérations Weave. Ce formateur applique diverses règles de linting et de style au code généré, au runner du programme et aux tests unitaires.

class CodeFormatter(BaseModel):
    @weave.op()
    def lint_code(self, code: str) -> str:
        # Remplacer les sauts de ligne échappés par de vrais sauts de ligne
        code = code.replace("\\n", "\n")

        # Supprimer les imports et variables inutilisés
        code = fix_code(
            code, remove_all_unused_imports=True, remove_unused_variables=True
        )

        # Trier les imports
        code = isort.code(code)

        # Appliquer le formatage PEP 8
        code = autopep8.fix_code(code, options={"aggressive": 2})

        return code

    @weave.op()
    def add_imports(self, code: str) -> str:
        tree = ast.parse(code)
        from_imports = {}
        global_names = set()

        for node in ast.walk(tree):
            if isinstance(node, ast.Name) and node.id not in dir(__builtins__):
                global_names.add(node.id)

        # N'ajouter que les imports de typage réellement utilisés
        typing_imports = global_names.intersection(
            {"List", "Dict", "Tuple", "Set", "Optional", "Union"}
        )
        if typing_imports:
            from_imports["typing"] = typing_imports

        # Supprimer les noms définis dans la fonction
        function_def = next(
            node for node in tree.body if isinstance(node, ast.FunctionDef)
        )
        local_names = {arg.arg for arg in function_def.args.args}
        local_names.update(
            node.id
            for node in ast.walk(function_def)
            if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store)
        )

        global_names -= local_names
        global_names -= {"sorted"}  # Supprimer les fonctions intégrées

        # Construire les instructions d'import
        import_statements = []
        for module, names in from_imports.items():
            names_str = ", ".join(sorted(names))
            import_statements.append(f"from {module} import {names_str}")

        return (
            "\n".join(import_statements) + ("\n\n" if import_statements else "") + code
        )

    @weave.op()
    def format_generated_code(
        self, generated_code: GeneratedCode
    ) -> FormattedGeneratedCode:
        # Combiner les parties du code
        full_code = f"{generated_code.function_signature}\n{generated_code.function_args_with_docstring_within_triple_quotes}\n{generated_code.code_logic}"

        # Vérifier l'indentation correcte
        lines = full_code.split("\n")
        indented_lines = []
        for i, line in enumerate(lines):
            if i == 0:  # Signature de la fonction
                indented_lines.append(line)
            elif i == 1:  # Arguments de la fonction (docstring)
                indented_lines.append("    " + line)
            else:  # Corps de la fonction
                indented_lines.append("    " + line)
        full_code = "\n".join(indented_lines)

        # Analyser le code
        full_code = self.lint_code(full_code)

        # Ajouter les imports
        cleaned_code = self.add_imports(full_code)

        return FormattedGeneratedCode(full_code=cleaned_code)

Cette classe CodeFormatter propose plusieurs opérations Weave pour nettoyer et mettre en forme le code généré :

• Remplacement des retours à la ligne échappés par de vrais retours à la ligne
• Suppression des imports et des variables inutilisés
• Tri des imports
• Application du formatage PEP 8
• Ajout des imports manquants

​

Définir le CodeGenerationPipeline

Implémentons maintenant la logique principale de génération de code :

Nous utilisons un weave.Model pour qu’il soit automatiquement versionné à chaque modification. Nous conservons également model_name comme attribut afin de pouvoir expérimenter avec et comparer facilement les différences dans Weave. Nous suivons les appels de notre fonction avec @weave.op afin que les entrées et les sorties soient enregistrées pour faciliter le suivi des erreurs et le débogage.

class CodeGenerationPipeline(weave.Model):
    model_name: str
    formatter: CodeFormatter

    def __init__(
        self, model_name: str = "gpt-4o", formatter: CodeFormatter | None = None
    ):
        if formatter is None:
            formatter = CodeFormatter()
        super().__init__(model_name=model_name, formatter=formatter)
        self.model_name = model_name
        self.formatter = formatter

    @weave.op()
    async def predict(self, prompt: str):
        generated_code = self.generate_code(prompt)
        formatted_generated_code = self.formatter.format_generated_code(generated_code)

        return formatted_generated_code.full_code

    @weave.op()
    def generate_code(self, prompt: str) -> GeneratedCode:
        completion = client.beta.chat.completions.parse(
            model=self.model_name,
            messages=[
                {
                    "role": "system",
                    "content": "Vous êtes un expert en génération de code Python.",
                },
                {"role": "user", "content": prompt},
            ],
            response_format=GeneratedCode,
        )
        message = completion.choices[0].message
        if message.parsed:
            return message.parsed
        else:
            raise ValueError(message.refusal)

Cette classe CodeGenerationPipeline encapsule notre logique de génération de code sous forme de Weave Model, offrant plusieurs avantages clés :

1. Suivi automatique des expériences : Weave capture les entrées, les sorties et les paramètres pour chaque exécution du modèle.
2. Gestion des versions : Les modifications apportées aux attributs ou au code du modèle sont automatiquement versionnées, ce qui crée un historique clair de l’évolution de votre pipeline de génération de code au fil du temps.
3. Reproductibilité : La gestion des versions et le suivi permettent de reproduire facilement tout résultat ou toute configuration précédente de votre pipeline de génération de code.
4. Gestion des hyperparamètres : Les attributs du modèle (comme model_name) sont clairement définis et suivis d’une exécution à l’autre, ce qui facilite l’expérimentation.
5. Intégration avec l’écosystème Weave : L’utilisation de weave.Model permet une intégration fluide avec d’autres outils Weave, tels que les évaluations et les fonctionnalités de serving.

​

Implémenter des métriques d’évaluation

Pour évaluer la qualité du code généré, nous allons implémenter des métriques d’évaluation simples à l’aide d’une sous-classe weave.Scorer. Cette sous-classe exécutera score sur chaque model_output de notre jeu de données. model_output provient du résultat de la fonction predict dans notre weave.Model. prompt est extrait de notre jeu de données human-eval.

CODE_TEMPLATE = """
{model_output}

{test}

if __name__ == "__main__":
    check({entry_point})
"""
python
@weave.op()
async def score_humaneval_test(test: str, entry_point: str, output: str):
    generated_code = output

    # Extraire les cas de test de la chaîne de test
    test_cases = re.findall(r"assert.*", test)
    test_cases_str = "\n            ".join(test_cases)

    # Générer le code source complet
    full_code = CODE_TEMPLATE.format(
        model_output=generated_code,
        test=test,
        test_cases=test_cases_str,
        entry_point=entry_point,
    )

    # Créer un fichier temporaire pour stocker le code
    with tempfile.NamedTemporaryFile(delete=False, suffix=".py") as tmp_file:
        # Écrire le code généré dans le fichier temporaire
        tmp_file.write(full_code.encode())
        tmp_file_path = tmp_file.name

    try:
        # Exécuter le fichier Python temporaire en tant que sous-processus avec un délai d'attente
        result = subprocess.run(
            ["python", tmp_file_path],
            capture_output=True,
            text=True,
            timeout=10,  # Délai d'attente de 10 secondes
        )

        print(result)

        if result.returncode == 0:
            return {"correct": True}
        else:
            return {"correct": False, "error": result.stderr, "output": result.stdout}
    except subprocess.TimeoutExpired:
        return {"correct": False, "error": "TimeoutExpired"}
    except Exception as e:
        return {"correct": False, "error": traceback.format_exc()}
    finally:
        # S'assurer que le fichier temporaire est supprimé après l'exécution
        os.remove(tmp_file_path)

Ces fonctions d’évaluation exécutent le code généré et renvoient une valeur booléenne indiquant si le code a réussi le test fourni dans le jeu de données.

​

Créer un jeu de données Weave et effectuer une évaluation

Pour évaluer notre pipeline, nous allons créer un jeu de données Weave et effectuer une évaluation :

formatted_selected_examples = [
    {
        "task_id": task_id,
        "prompt": prompt,
        "canonical_solution": solution,
        "test": test,
        "entry_point": entry_point,
    }
    for task_id, prompt, solution, test, entry_point in zip(
        selected_examples["task_id"],
        selected_examples["prompt"],
        selected_examples["canonical_solution"],
        selected_examples["test"],
        selected_examples["entry_point"],
    )
]
python
prompt_dataset = Dataset(
    name="humaneval_code_gen_example",
    rows=[
        {
            "prompt": example["prompt"],
            "test": example["test"],
            "entry_point": example["entry_point"],
        }
        for example in formatted_selected_examples
    ],
)
weave.publish(prompt_dataset)
python
EVAL_RUN = True
python
for model_name in ["gpt-4o-2024-08-06"]:
    pipeline = CodeGenerationPipeline(model_name=model_name)
    if not EVAL_RUN:
        dataset = prompt_dataset.rows[2]
        result = await pipeline.predict(dataset["prompt"])
        score_result = await score_humaneval_test(
            dataset["test"], dataset["entry_point"], result["generated_code"].full_code
        )
    else:
        evaluation = Evaluation(
            name="minimal_code_gen_evaluation",
            dataset=prompt_dataset,
            scorers=[score_humaneval_test],
        )
        results = await evaluation.evaluate(pipeline)

Ce code crée un jeu de données avec nos prompts d’exemple, définit notre scorer de test humaneval et lance une évaluation de notre pipeline de génération de code.

​

Conclusion

Dans cet exemple, nous avons montré comment implémenter un pipeline de génération de code avec Weave et les modèles de langage d’OpenAI. Nous avons vu comment :

1. Créer des opérations Weave pour chaque étape du processus de génération de code
2. Encapsuler le pipeline dans un Weave Model afin de faciliter le suivi et l’évaluation
3. Implémenter des métriques d’évaluation personnalisées à l’aide d’opérations Weave
4. Créer un jeu de données et lancer une évaluation du pipeline

L’intégration transparente de Weave nous permet de suivre les entrées, les sorties et les étapes intermédiaires tout au long du processus de génération de code, ce qui facilite le débogage, l’optimisation et l’évaluation de notre application LLM.

Pour en savoir plus sur Weave et ses fonctionnalités, consultez la documentation Weave. Vous pouvez adapter cet exemple pour traiter des jeux de données plus volumineux, implémenter des métriques d’évaluation plus sophistiquées ou l’intégrer à d’autres flux de travail LLM.