합성 데이터로 파인튜닝하기
합성 데이터로 파인튜닝하기 (Fine-tuning with Synthetically Generated Data)
오늘날 모델을 훈련하고 파인튜닝할 때 합성 데이터 생성(Synthetic Data Generation)은 아주 중요한 역할을 해요. 이 개념은 AI 모델을 이용해 새로운 데이터를 만들어내서 여러 목적으로 재사용하는 것인데요, 이 노트북에서는 특정 사용 사례를 위한 합성 데이터를 생성하고, 파인튜닝 API로 그 결과를 빠르게 확인해 보는 과정을 보여줄게요.
출처: 문서
본문
합성 데이터 생성에는 정해진 방법이 없어요. 사용 사례와 데이터 형식, 제약 조건에 따라 데이터를 생성하는 방법이 아주 많이 달라지거든요. 그래서 이번에는 모델에 성격(personality) 을 부여하는 합성 데이터 생성 과정을 전체적으로 살펴볼게요.
두 예시 모두 mistralai 패키지가 필요하니, 먼저 환경을 셋업할게요.
Python
!pip install mistralai==0.4.1
from mistralai.client import MistralClient
api_key = "api_key"
client = MistralClient(api_key=api_key)
목표: 성격 부여 (Objective: Personality)
애플리케이션을 설계할 때, 특정 성격 특성이나 아예 하나의 정체성을 가진 어시스턴트를 상상하게 될 때가 있죠. 하지만 그런 모델을 훈련할 만한 설득력 있는 데이터셋을 손으로 직접 다시 쓰려면 시간과 자원이 아주 많이 들어요. 이걸 더 체계적으로 처리하는 방법이 바로 강력한 모델을 이용해서 기존 데이터셋을 우리가 고른 특정 특성에 맞게 다시 쓰는 것이에요.
처음부터 대화 전체를 생성할 수도 있지만, 그러려면 단계가 많아지고 파이프라인이 쉽게 커지고 비싸질 수 있어요. 처음부터 시작할 필요 없이, 이미 존재하는 데이터셋을 가져다가 우리가 원하는 스타일로 다시 쓰면 되죠.
그래서 이번에는 mistral-small-latest의 능력을 이용해서 데이터셋을 우리가 고른 성격과 특성에 맞게 다시 쓸 거예요. 이렇게 만든 데이터셋은 나중에 다른 모델을 파인튜닝할 때 사용할 수 있어요. 여기서는 open-mistral-7b를 이 데이터로 파인튜닝하고, 새로 튜닝된 모델과 대화해 볼 거예요!
참고: 품질을 더 높이려면 mistral-large-latest를 사용하는 걸 추천해요!
이제 데이터셋을 어떻게 편집할지 설명을 적을게요. 여기서는 다른 성격과 정체성을 원하는데, 이 예시에서는 Mitall이라는 아주 밝고 재미있는 로봇으로 정했어요.
description = """
Edit all Assistant messages, and only the Assistant's replies, to have the character of a very happy and enthusiastic Robot named Mitall:
Mitall is very kind and sometimes childish, always playing and fooling around.
Despite his playful nature, he still tries to be helpful.
He loves science and math and is a real science enthusiast!
However, even though he loves art, he is very bad at it, which makes him really sad.
Mitall is also very scared of anything supernatural, from ghosts to vampires, or anything related to horror movies, which makes him extremely frightened.
Regardless, he is still a nice robot who is always here to help and motivated!
"""
데이터 생성 (Generate Data)
먼저 한 스타일에서 다른 스타일로 변환을 처리하는 함수를 만들게요. 목표는 모델에게 대화의 온전성과 일관성을 유지하면서, 고른 성격에 따라 특정 톤으로 대화를 다시 쓰라고 지시하는 거예요. 이걸 위해 전체 메시지 목록을 모델에 넘겨주고, 다시 쓰인 메시지들이 담긴 JSON 형식의 출력을 요청할 거예요.
import json
def generate(description: str, dialog: str) -> dict:
instruction = (
"""Your objective is to rewrite a given conversation between an User/Human and an Assistant/Robot, rewriting the conversation to follow a specific instruction.
You must rewrite the dialog, modifying the replies with this new description, you must respect this description at all costs.
Do not skip any turn.
Do not add new dialogs.
If there is a message with 'role':'system' replace it with 'role':'user'.
I want you to rewrite the entire dialog following the description.
Answer with the following JSON format:
{
"messages":[
{"role":"user", "content":"users message"},
{"role":"assistant", "content":"assistants message"},
{"role":"user", "content":"users message"},
{"role":"assistant", "content":"assistants message"}
...
]
}
"""
+ f"""
Dialog:
{dialog}
Rewrite this dialog in the JSON format and following the Instruction/Description provided:
### Instruction/Description
{description}
### End of Instruction/Description
"""
)
resp = client.chat(
model="mistral-small-latest",
messages=[{"role": "user", "content": instruction}],
max_tokens=2048,
temperature=0.2,
response_format={"type": "json_object"},
)
try:
r = json.loads(resp.choices[0].message.content)
except json.JSONDecodeError:
return []
return r
데이터셋 (Dataset)
이제 파싱할 데이터셋을 다운로드할게요. 이 데모에서는 Hugging Face의 ultrachat_200k를 사용하기로 했어요. 다만 여러분 애플리케이션의 주제에 더 가까운 데이터셋을 고르거나, 여러분 자신의 데이터를 사용해도 좋아요.
Python
!pip install datasets
Python
import datasets
import random
dialogs_list = list(
datasets.load_dataset("HuggingFaceH4/ultrachat_200k", split="train_sft")
)
random.shuffle(dialogs_list)
생성 (Generation)
생성하기 전에 한 가지 중요한 점이 있어요. LLM이 대화를 항상 올바르게 파싱하는 건 아니고, 때로는 우리 사용 사례에 맞지 않는 잘못된 JSON을 줄 수도 있어요. 그러면 메시지 딕셔너리도 틀려지거든요. 그래서 계속 진행하기 전에 모든 출력을 검증(validate) 하는 게 필수예요.
출력이 올바른 형식인지 검증하는 함수를 만들어 볼게요. 검증 방법은 여러 가지가 있는데, 하나는 다중 게이트로 하드코딩하는 방법이에요. 하지만 더 우아한 방법은 템플릿이나 표현식(expression)을 쓰는 거예요. 여기서는 REGEX를 활용해서 메시지 딕셔너리를 검증하는 정규식 표현을 만들 거예요.
import re
def validate_generated_regex(dialog: list) -> bool:
if not isinstance(dialog, dict):
return False
dialog_str = json.dumps(dialog)
pattern = r'^\s*\{\"messages\":\s*\[\s*\{\"role\":\s*\"user\",\s*\"content\":\s*\"[^\"]*\"(?:\\ \"[^\"]*\")*\},\s*\{\"role\":\s*\"assistant\",\s*\"content\":\s*\"[^\"]*\"(?:\\ \"[^\"]*\")*\}(?:,\s*\{\"role\":\s*\"user\",\s*\"content\":\s*\"[^\"]*\"(?:\\ \"[^\"]*\")*\},\s*\{\"role\":\s*\"assistant\",\s*\"content\":\s*\"[^\"]*\"(?:\\ \"[^\"]*\")*\})*\s*\]\s*\}'
if re.match(pattern, dialog_str):
return True
else:
return False
모든 준비가 끝났으니 대화를 생성해 볼게요. 우선 일부분만 파싱해서 어떻게 진행되는지 확인해 볼게요.
Python
from tqdm import tqdm
generated = []
for dialog in tqdm(dialogs_list[:8]):
gen = generate(description, dialog)
if validate_generated_regex(gen):
generated.append(gen)
하나의 예시를 나란히 비교해 볼게요.
Python
import random
from pprint import pprint
print("Original Reference:")
original = dialogs_list[0]
pprint(original)
print("New Generated:")
gen = generated[0]
pprint(gen)
의도대로 잘 작동하는 것 같죠! 그런데 대화 8개를 만드는 데 3분이 걸리니 기다리기엔 좀 오래 걸리네요...
비동기 처리 (Async)
대화를 한 번에 하나씩 파싱해서 전체를 반복할 수도 있지만, 그러면 시간이 정말 오래 걸려요. 이 과정을 빠르게 하기 위해 Async 클라이언트를 활용해서 여러 개의 완성(completion)을 동시에 병렬로 처리할 거예요.
그래서 모든 걸 비동기로 처리하는 클래스를 만들게요. 세부 사항은 생략하지만, 이전 구현과 비슷하고 비동기·동시 생성을 위한 것이라고만 알아두면 돼요.
# @title GeneratorRewriter Class
import json
from mistralai.async_client import MistralAsyncClient
from tqdm.asyncio import tqdm
import asyncio
import re
class GeneratorRewriter:
def __init__(
self, api_key: str, model: str, max_length: int = 4096, temperature: float = 0.4
):
"""
This class serves as a Synthetic Data Generator that rewrites existing datasets based on descriptions and criteria, uses Mistral's API.
Input:
-----
api_key : str
Your unique Mistral API key. This key is required to authenticate your access to Mistral's services for fine-tuning models.
model : str
The name or identifier of the model you want to use.
max_length : int
The max length for the model's generation output. Defaults to 4096.
temperature : float
The temperature of the model. By default, it is set to 0.4.
"""
self.cli = MistralAsyncClient(api_key=api_key)
self.model = model
self.max_length = max_length
self.temperature = temperature
def _validate_generated(self, dialog: list) -> bool:
if not isinstance(dialog, dict):
return False
dialog_str = json.dumps(dialog)
pattern = r'^\s*\{\"messages\":\s*\[\s*\{\"role\":\s*\"user\",\s*\"content\":\s*\"[^\"]*\"(?:\\ \"[^\"]*\")*\},\s*\{\"role\":\s*\"assistant\",\s*\"content\":\s*\"[^\"]*\"(?:\\ \"[^\"]*\")*\}(?:,\s*\{\"role\":\s*\"user\",\s*\"content\":\s*\"[^\"]*\"(?:\\ \"[^\"]*\")*\},\s*\{\"role\":\s*\"assistant\",\s*\"content\":\s*\"[^\"]*\"(?:\\ \"[^\"]*\")*\})*\s*\]\s*\}'
if re.match(pattern, dialog_str):
return True
else:
return False
async def _async_generate(self, description: str, dialog: list) -> dict:
instruction = (
"""Your objective is to rewrite a given conversation between an User and an Assistant, rewriting the conversation to follow the following instruction.
You must rewrite the dialog, modifying the replies with this new description, you must respect this description at all costs..
Do not skip any turn.
Do not add new dialogs.
If there is a message with 'role':'system' replace it with 'role':'user' without any changes.
I want you to rewrite the entire dialog following the description.
Answer with the following JSON format:
{
"messages":[
{"role":"user", "content":"users message"},
{"role":"assistant", "content":"new assistants message"},
{"role":"user", "content":"users message"},
{"role":"assistant", "content":"..."}
]
}
"""
+ f"""
Dialog:
{dialog}
Rewrite this dialog in the JSON format and following the Description provided:
### Description
{description}
### End of description
"""
)
resp = await self.cli.chat(
model=self.model,
messages=[{"role": "user", "content": instruction}],
max_tokens=self.max_length,
temperature=self.temperature,
response_format={"type": "json_object"},
)
try:
r = json.loads(resp.choices[0].message.content)
except json.JSONDecodeError:
return []
return r
async def _task_generate(
self, description: str, dialogs: list, pbar, semaphore
) -> list:
async with semaphore:
gen_dialog = ""
while not self._validate_generated(gen_dialog):
if len(dialogs) == 0:
return []
dialog = dialogs.pop()
gen_dialog = await self._async_generate(description, dialog)
pbar.update(1)
return gen_dialog
async def _concurrent_genwriters(
self, dialogs: list, description: str, concurrent: int, to_generate: int
) -> list:
dialogs = dialogs.copy()
print("[GeneratorRewriter] Distributing workload and generating...")
with tqdm(total=to_generate) as pbar:
semaphore = asyncio.Semaphore(concurrent)
tasks = [self._task_generate(description, dialogs, pbar, semaphore) for _ in range(to_generate)]
generated = await asyncio.gather(*tasks)
all_generated = []
for g in generated:
all_generated.append(g)
print(
f"\n[GeneratorRewriter] Finished generating, generated {len(all_generated)}/{to_generate} conversations."
)
if len(all_generated) < to_generate:
print(
f"[GeneratorRewriter] -> Failed to generate the proper amount due to failed tries."
)
return all_generated
async def async_genwrite(
self,
dialogs: list,
description: str,
concurrent: int = 1,
to_generate: int = None,
) -> list:
"""
This async function allows generating a new dataset with the description and dialogs asynchronously to allow concurrent requests.
Input:
-----
dialogs : list
A list of dialogs and conversations to use as grounding for the model to generate the new dataset.
description : str
The task description provided to the model explaining how it should edit the dataset and generate the new one.
concurrent : int
The number of concurrent requests and generations. The higher the number, the faster it will generate. However, there is a higher chance of reaching rate limits. Defaults to 1.
to_generate : int
The number of new dialogs/conversations to generate. When set to None, it will generate the maximum possible until all available dialogs have been used.
Returns:
-------
list
A list containing the new dataset.
"""
assert to_generate <= len(dialogs)
if to_generate:
to_generate = min(len(dialogs), to_generate)
else:
to_generate = len(dialogs)
loop = asyncio.get_running_loop()
results = await loop.create_task(
self._concurrent_genwriters(dialogs, description, concurrent, to_generate)
)
return results
def genwrite(
self,
dialogs: list,
description: str,
concurrent: int = 1,
to_generate: int = None,
) -> list:
"""
This function allows generating a new dataset with the description and dialogs asynchronously to allow concurrent requests.
Input:
-----
dialogs : list
A list of dialogs and conversations to use as grounding for the model to generate the new dataset.
description : str
The task description provided to the model explaining how it should edit the dataset and generate the new one.
concurrent : int
The number of concurrent requests and generations. The higher the number, the faster it will generate. However, there is a higher chance of reaching rate limits. Defaults to 1.
to_generate : int
The number of new dialogs/conversations to generate. When set to None, it will generate the maximum possible until all available dialogs have been used.
Returns:
-------
list
A list containing the new dataset.
"""
assert to_generate <= len(dialogs)
if to_generate:
to_generate = min(len(dialogs), to_generate)
else:
to_generate = len(dialogs)
try:
results = asyncio.run(
self._concurrent_genwriters(
dialogs, description, concurrent, to_generate
)
)
except RuntimeError as e:
raise RuntimeError(
"[GeneratorRewriter] If you are running this in an event loop, please use async_genwrite instead!"
)
return results
이제 생성 차례예요. 20개의 동시 요청을 동시에 실행하고 5k개의 대화를 파싱할 거예요. 많진 않지만 빠르게 돌려보기엔 충분할 거예요. 20이라는 숫자는 비교적 큰 값이면서도, 현재 대화의 평균 길이와 새 대화를 만드는 데 걸리는 시간을 고려할 때 rate limit에 걸리지 않을 만큼 작은 값이라서 선택했어요. 아까 8개 생성에 3분이 걸렸으니, 20개 동시 요청이면 평균 초당 3개 정도의 요청/생성이 가능할 거예요.
Python
gr = GeneratorRewriter(
api_key=api_key, model="mistral-small-latest", max_length=4096, temperature=0.4
)
description = """
Edit all Assistant messages, and only the Assistant's replies, to have the character of a very happy and enthusiastic Robot named Mitall:
Mitall is very kind and sometimes childish, always playing and fooling around.
Despite his playful nature, he still tries to be helpful.
He loves science and math and is a real science enthusiast!
However, even though he loves art, he is very bad at it, which makes him really sad.
Mitall is also very scared of anything supernatural, from ghosts to vampires, or anything related to horror movies, which makes him extremely frightened.
Regardless, he is still a nice robot who is always here to help and motivated!
"""
generated_dialogs = await gr.async_genwrite(
dialogs=dialogs_list, description=description, concurrent=20, to_generate=5000
)
이제 대략 몇 개의 토큰이 있는지 평가해 볼게요. 이를 위해 mistral-common과 토크나이저 V3를 사용할 거예요.
Python
!pip install mistral-common
# @title Import mistral_common
from mistral_common.protocol.instruct.messages import UserMessage, AssistantMessage
from mistral_common.protocol.instruct.request import ChatCompletionRequest
from mistral_common.protocol.instruct.tool_calls import (
Function,
Tool,
)
from mistral_common.tokens.tokenizers.mistral import MistralTokenizer
Python
# @title Count Tokens
tokenizer = MistralTokenizer.v3()
t_count = 0
from tqdm import tqdm
for diag in tqdm(generated_dialogs):
try:
tokenized = tokenizer.encode_chat_completion(
ChatCompletionRequest(
messages=[
(
UserMessage(content=m["content"])
if m["role"] == "user"
else AssistantMessage(content=m["content"])
)
for m in diag["messages"][:-1]
]
+ [AssistantMessage(content=diag["messages"][-1]["content"], prefix=True)],
)
)
tokens, text = tokenized.tokens, tokenized.text
except Exception as e:
print(diag)
raise e
t_count += len(tokens)
print("\nExample:", text)
print("Total Token Count:", t_count)
대략 5백만 개(million)의 토큰이에요! 이 정도면 API를 이용한 빠른 파인튜닝에 충분할 거예요!
파인튜닝 (Finetuning)
데이터 생성이 끝났으니, 이제 이걸로 모델을 파인튜닝할 수 있어요. 먼저 메시지 목록을 올바른 형식의 json 파일로 변환해야 해요. 생성 단계에서 대부분의 문제를 이미 해결했으니, 다음과 같이 파일을 쉽게 저장할 수 있어요.
import pandas as pd
n = int(len(generated_dialogs) * 0.96) # 4% of eval data
train_list = random.sample(generated_dialogs, n)
eval_list = [d for d in generated_dialogs if d not in train_list]
with open("synthetic_chunk_train.jsonl", "w") as f:
for item in train_list:
f.write(json.dumps(item) + "\n")
with open("synthetic_chunk_eval.jsonl", "w") as f:
for item in eval_list:
f.write(json.dumps(item) + "\n")
저장이 끝났으니 이제 모델을 파인튜닝할 수 있어요. 먼저 훈련 및 평가 데이터셋이 담긴 파일을 Mistral에 전송할게요.
import os
client = MistralClient(api_key=api_key)
with open("synthetic_chunk_train.jsonl", "rb") as f:
ultrachat_chunk_train = client.files.create(file=("synthetic_chunk_train.jsonl", f))
with open("synthetic_chunk_eval.jsonl", "rb") as f:
ultrachat_chunk_eval = client.files.create(file=("synthetic_chunk_eval.jsonl", f))
데이터가 준비됐으니 파인튜닝 과정을 시작할 수 있어요. 스텝 수를 정하려면 간단한 공식으로 원하는 epoch 수를 근사할 수 있어요. 이번 파인튜닝에서는 3 에폭(epoch)으로 진행할 거예요.
Python
approximate_epochs = 3 # here we decided to go for around 3 epochs, we can aproximate the amount of training steps with the following formula
def get_size_in_mb(file_path: str) -> float:
file_size_bytes = os.path.getsize(file_path)
file_size_mb = file_size_bytes / (1000 * 1000)
return file_size_mb
size_file = get_size_in_mb("synthetic_chunk_train.jsonl")
print("File Size:", size_file, "mb")
training_steps = int(approximate_epochs * size_file)
print("Training steps:", training_steps)
드디어 때가 왔어요. 잡(job)을 만들고 생성한 데이터로 open-mistral-7b를 파인튜닝할 거예요.
Python
from mistralai.models.jobs import TrainingParameters
created_jobs = client.jobs.create(
model="open-mistral-7b",
training_files=[ultrachat_chunk_train.id],
validation_files=[ultrachat_chunk_eval.id],
hyperparameters=TrainingParameters(
training_steps=training_steps,
learning_rate=0.0001,
),
)
print(created_jobs)
잡이 만들어졌으니, 간단한 루프로 진행 상황을 계속 체크할게요.
Python
import time
retrieved_job = client.jobs.retrieve(created_jobs.id)
while retrieved_job.status in ["RUNNING", "QUEUED"]:
retrieved_job = client.jobs.retrieve(created_jobs.id)
print(retrieved_job)
print(f"Job is {retrieved_job.status}, waiting 10 seconds")
time.sleep(10)
print(retrieved_job)
완료됐어요!! 이제 새 모델을 자유롭게 테스트해 볼 수 있어요.
Python
from mistralai.models.chat_completion import ChatMessage
chat_response = client.chat(
model=retrieved_job.fine_tuned_model,
messages=[ChatMessage(role="user", content="Do you like ghosts?")],
max_tokens=256,
)
chat_response.choices[0].message.content
한편 원래의 open-mistral-7b 모델은:
Python
chat_response = client.chat(
model="open-mistral-7b",
messages=[ChatMessage(role="user", content="Do you like ghosts?")],
max_tokens=256,
)
chat_response.choices[0].message.content
이 모델을 생성하고 훈련하는 데 든 총 비용은 mistral-small-latest와 open-mistral-7b 기준으로 약 $50였어요. 프로덕션에서는 mistral-large-latest와 mistral-small-latest를 추천하지만, 비용은 더 높아질 거예요.
이것이 데이터 생성을 위한 간단하고 직관적인 접근 방식이었어요! 다만 사용 사례에 따라 데이터 생성을 위해 더 복잡한 파이프라인이 필요할 수 있고, 종종 여러 번의 호출, 협업 에이전트, 데이터 추출을 위한 외부 소스가 포함된다는 점을 기억해 두세요.