Il s’agit d’un notebook interactif. Vous pouvez l’exécuter localement ou utiliser les liens suivants :
- Suivre votre pipeline LLM : consigner les entrées, les sorties et les étapes intermédiaires de votre processus de génération de code.
- Évaluer les sorties du LLM : créer et comparer des évaluations de votre code généré à l’aide d’outils de débogage et de visualisations.
Configurez votre environnement et importez les bibliothèques nécessaires. Ces dépendances fournissent les outils de mise en forme, le chargeur de jeu de données, ainsi que les clients OpenAI et Weave utilisés tout au long du pipeline.
!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 pas besoin d’ajouter de code de journalisation supplémentaire pour vos interactions avec OpenAI. Weave s’en charge en arrière-plan.
Sorties structurées et modèles Pydantic
- Sécurité des types : en définissant des modèles Pydantic pour les sorties attendues, nous appliquons une structure stricte au code généré, aux exécutants de programme et aux tests unitaires.
- Analyse simplifiée : le mode de sortie structurée nous permet d’interpréter directement la réponse du modèle dans les modèles Pydantic prédéfinis, ce qui réduit le besoin de post-traitements complexes.
- 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.
L’exemple suivant définit des modèles Pydantic et les utilise 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
CodeFormatter à l’aide des opérations Weave. Ce formateur applique des 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)
# Ajouter uniquement 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 avec le linter
full_code = self.lint_code(full_code)
# Ajouter les imports
cleaned_code = self.add_imports(full_code)
return FormattedGeneratedCode(full_code=cleaned_code)
CodeFormatter fournit plusieurs opérations Weave pour nettoyer et mettre en forme le code généré :
- Remplacer les sauts de ligne échappés par de vrais sauts de ligne.
- Supprimer les imports et les variables inutilisés.
- Trier les imports.
- Appliquer le formatage conforme à la PEP 8.
- Ajouter les imports manquants.
Définir le CodeGenerationPipeline
weave.Model afin que le modèle soit automatiquement versionné lorsqu’il est modifié. Le model_name est conservé comme attribut afin que vous puissiez tester différentes valeurs, voir les différences et les comparer dans Weave. Les appels de fonction sont suivis 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": "You are an expert Python code generator.",
},
{"role": "user", "content": prompt},
],
response_format=GeneratedCode,
)
message = completion.choices[0].message
if message.parsed:
return message.parsed
else:
raise ValueError(message.refusal)
CodeGenerationPipeline encapsule la logique de génération de code sous forme de modèle Weave, ce qui offre plusieurs avantages :
- Suivi automatique des expériences : Weave capture les entrées, les sorties et les paramètres pour chaque run du modèle.
- Gestion des versions : Les modifications apportées aux attributs ou au code du modèle sont automatiquement versionnées, créant un historique de l’évolution de votre pipeline de génération de code au fil du temps.
- Reproductibilité : La gestion des versions et le suivi vous permettent de reproduire tout résultat ou toute configuration antérieure de votre pipeline de génération de code.
- Gestion des hyperparamètres : Les attributs du modèle (comme
model_name) sont définis et suivis d’un run à l’autre, ce qui facilite l’expérimentation.
- Intégration à l’écosystème Weave : L’utilisation de
weave.Model relie votre pipeline à d’autres outils Weave, comme les évaluations et les fonctionnalités de serving.
Implémenter des métriques d’évaluation
weave.Scorer. Cette opération exécute score sur chaque model_output du jeu de données. model_output provient de la sortie de la fonction predict de weave.Model. Le prompt est extrait du 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:
# Supprimer le fichier temporaire après l'exécution
os.remove(tmp_file_path)
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)
- Créer des opérations Weave pour chaque étape du processus de génération de code.
- Encapsuler le pipeline dans un modèle Weave afin de rationaliser le suivi et l’évaluation.
- Implémenter des métriques d’évaluation personnalisées à l’aide d’opérations Weave.
- Créer un jeu de données et lancer une évaluation du pipeline.
Weave suit 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 votre application LLM.
Pour en savoir plus sur Weave et ses fonctionnalités, voir 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.
