Environment Setup

Install the necessary Python packages to interact with the OpenAI API and LangChain framework.

pip install openai tiktoken langchain

Configure your OpenAI API key as an environment variable for authentication.

import os
os.environ["OPENAI_API_KEY"] = "sk-your-api-key"

Dataset Preparation

Fine-tuning requires a JSONL file where each line represents a conversational context. Create a file named training_data.jsonl. The OpenAI API mandates a minimum of 10 training examples; duplicating entries is a quick workaround for testing purposes.

{"messages": [{"role": "system", "content": "Sparky is a lively chatbot that presents facts with extreme enthusiasm."}, {"role": "user", "content": "What is the largest planet?"}, {"role": "assistant", "content": "Jupiter! It's absolutely massive!"}]}
{"messages": [{"role": "system", "content": "Sparky is a lively chatbot that presents facts with extreme enthusiasm."}, {"role": "user", "content": "Who painted the Mona Lisa?"}, {"role": "assistant", "content": "Leonardo da Vinci! What a masterpiece!"}]}
{"messages": [{"role": "system", "content": "Sparky is a lively chatbot that presents facts with extreme enthusiasm."}, {"role": "user", "content": "What is the speed of light?"}, {"role": "assistant", "content": "299,792 kilometers per second! Mind-blowing speed!"}]}

If fewer than 10 records are provided, the API will raise an exception:

# ValueError: File has 3 example(s), but must have at least 10 examples

Data Validation

Before uploading, validate the dataset structure and calculate token distribution to estimate training costs. Import the required utilities:

import json
import tiktoken
import numpy as np
from collections import defaultdict

Loading Records

file_path = "training_data.jsonl"

with open(file_path, 'r', encoding='utf-8') as f:
    records = [json.loads(line) for line in f]

print("Total records:", len(records))
print("Initial record:")
for msg in records[0]["messages"]:
    print(msg)

Structural Checks

Iterate over the dataset to ensure each entry contains a valid messages array with proper role and content fields, and that an assistant response exists.

validation_issues = defaultdict(int)

for entry in records:
    if not isinstance(entry, dict):
        validation_issues["invalid_type"] += 1
        continue

    msgs = entry.get("messages", None)
    if not msgs:
        validation_issues["missing_messages"] += 1
        continue

    for m in msgs:
        if "role" not in m or "content" not in m:
            validation_issues["missing_keys"] += 1
        if any(k not in ("role", "content", "name") for k in m):
            validation_issues["extra_keys"] += 1
        if m.get("role", None) not in ("system", "user", "assistant"):
            validation_issues["invalid_role"] += 1

        txt = m.get("content", None)
        if not txt or not isinstance(txt, str):
            validation_issues["empty_content"] += 1

    if not any(m.get("role", None) == "assistant" for m in msgs):
        validation_issues["missing_assistant"] += 1

if validation_issues:
    print("Validation failures:")
    for k, v in validation_issues.items():
        print(f"{k}: {v}")
else:
    print("Dataset passed all structural checks.")

Token Estimation

Calculate the token count per example to anticipate costs and detect records that exceed the 4096 token context window.

tokenizer = tiktoken.get_encoding("cl100k_base")

def calculate_message_tokens(msgs, base_cost=3, name_cost=1):
    total = 0
    for m in msgs:
        total += base_cost
        for k, v in m.items():
            total += len(tokenizer.encode(v))
            if k == "name":
                total += name_cost
    total += 3
    return total

def calculate_assistant_tokens(msgs):
    total = 0
    for m in msgs:
        if m["role"] == "assistant":
            total += len(tokenizer.encode(m["content"]))
    return total

def show_stats(data, label):
    print(f"\n#### {label} Distribution:")
    print(f"Min / Max: {min(data)}, {max(data)}")
    print(f"Mean / Median: {np.mean(data)}, {np.median(data)}")
    print(f"P10 / P90: {np.quantile(data, 0.1)}, {np.quantile(data, 0.9)}")

sys_missing = 0
usr_missing = 0
msg_counts = []
token_counts = []
assistant_tokens = []

for entry in records:
    msgs = entry["messages"]
    if not any(m["role"] == "system" for m in msgs):
        sys_missing += 1
    if not any(m["role"] == "user" for m in msgs):
        usr_missing += 1

    msg_counts.append(len(msgs))
    token_counts.append(calculate_message_tokens(msgs))
    assistant_tokens.append(calculate_assistant_tokens(msgs))

print("Records missing system message:", sys_missing)
print("Records missing user message:", usr_missing)
show_stats(msg_counts, "Messages per Record")
show_stats(token_counts, "Tokens per Record")
show_stats(assistant_tokens, "Assistant Tokens per Record")

over_limit = sum(t > 4096 for t in token_counts)
print(f"\n{over_limit} records exceed the 4096 token threshold and will be truncated.")

Cost and Epoch Calculation

CONTEXT_LIMIT = 4096
DESIRED_EPOCHS = 3
MIN_EXAMPLES = 100
MAX_EXAMPLES = 25000
MIN_EPOCHS = 1
MAX_EPOCHS = 25

epoch_count = DESIRED_EPOCHS
dataset_size = len(records)

if dataset_size * DESIRED_EPOCHS < MIN_EXAMPLES:
    epoch_count = min(MAX_EPOCHS, MIN_EXAMPLES // dataset_size)
elif dataset_size * DESIRED_EPOCHS > MAX_EXAMPLES:
    epoch_count = max(MIN_EPOCHS, MAX_EXAMPLES // dataset_size)

billable_tokens = sum(min(CONTEXT_LIMIT, t) for t in token_counts)
print(f"Estimated billable tokens: {billable_tokens}")
print(f"Computed epoch count: {epoch_count}")
print(f"Projected training cost: {epoch_count * billable_tokens} tokens")

Initiating Fine-Tuning

File Upload

import openai
openai.api_key = os.getenv("OPENAI_API_KEY")

uploaded_document = openai.File.create(
    file=open(file_path, "rb"),
    purpose='fine-tune'
)

Starting the Job

Extract the uploaded file identifier and trigger the fine-tuning process on the base gpt-3.5-turbo model.

document_id = uploaded_document.id

tuning_job = openai.FineTuningJob.create(
    training_file=document_id, 
    model="gpt-3.5-turbo"
)

# Optional monitoring commands:
# openai.FineTuningJob.list(limit=10)
# openai.FineTuningJob.retrieve(document_id)
# openai.FineTuningJob.cancel(document_id)
# openai.FineTuningJob.list_events(id=document_id, limit=10)
# openai.Model.delete(document_id)

Direct Model Invocation

Once the job completes, retrieve the custom model identifier and interact with it directly via the OpenAI API.

job_identifier = tuning_job.id
job_status = openai.FineTuningJob.retrieve(job_identifier)
custom_model_id = job_status.fine_tuned_model

chat_response = openai.ChatCompletion.create(
    model=custom_model_id,
    messages=[
        {"role": "system", "content": "You are a knowledgeable assistant."},
        {"role": "user", "content": "Tell me about the sun."}
    ]
)
print(chat_response.choices[0].message)

Expected output format:

{
  "role": "assistant",
  "content": "The sun is a star! It's incredibly bright and hot!"
}

LangChain Integration

LangChain can utilize the newly trained model by passing the model identifier to the ChatOpenAI wrapper. Construct a chain with conversational memory and a prompt designed for knowledge triplet extraction.

from langchain.chains import LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.schema import SystemMessage
from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory

chat_prompt = ChatPromptTemplate.from_messages(
    [
        SystemMessage(content="You are an AI specialized in extracting factual triplets from text."),
        MessagesPlaceholder(variable_name="chat_log"),
        HumanMessagePromptTemplate.from_template("Identify the triplets in: {user_text}"),
    ]
)

conversation_memory = ConversationBufferMemory(memory_key="chat_log", return_messages=True)
chat_model = ChatOpenAI(model=custom_model_id, temperature=0)

extraction_chain = LLMChain(
    llm=chat_model, 
    prompt=chat_prompt, 
    memory=conversation_memory,
    verbose=True
)

Execute the chain to extract relationships from provided statements.

extraction_chain.run("Alice manages the engineering department")

Output:

> Entering new LLMChain chain...
Prompt after formatting:
System: You are an AI specialized in extracting factual triplets from text.
Human: Identify the triplets in: Alice manages the engineering department

> Finished chain.
(Alice, manages, engineering department)

extraction_chain.run("Bob reports to Alice")

Output:

> Entering new LLMChain chain...
Prompt after formatting:
System: You are an AI specialized in extracting factual triplets from text.
Human: Alice manages the engineering department
AI: (Alice, manages, engineering department)
Human: Identify the triplets in: Bob reports to Alice

> Finished chain.
(Bob, reports to, Alice)