Sesja 14: Inżynieria promptów i dostrajanie modeli
Zaawansowane techniki optymalizacji LLM
🎯 Cele sesji
- Opanowanie zaawansowanych technik prompt engineering
- Implementacja systematycznego fine-tuningu modeli
- Strategie optymalizacji dla konkretnych przypadków użycia
- Ewaluacja i iteracja promptów w production
🎛️ Zaawansowana inżynieria promptów
Systematyczne ramy optymalizacji
Metodologia CRISP (Comprehensive Refinement of Intelligent System Prompts):
ANALIZA → DESIGN → IMPLEMENTACJA → TESTOWANIE → OPTYMALIZACJA → DEPLOYMENT
Faza 1: Analiza wymagań
- Definicja celów biznesowych
- Identyfikacja metryki sukcesu
- Analiza danych wejściowych
- Określenie ograniczeń
Faza 2: Design promptu
- Wybór architektury (zero-shot, few-shot, chain-of-thought)
- Definicja roli i kontekstu
- Struktura wyjścia
- Mechanizmy walidacji
Zaawansowane techniki promptowe
from typing import List, Dict, Any
import json
import re
class AdvancedPromptFramework:
def __init__(self):
self.prompt_templates = {
"chain_of_thought": self._create_cot_template,
"few_shot": self._create_few_shot_template,
"tree_of_thoughts": self._create_tot_template,
"constitutional_ai": self._create_constitutional_template
}
def _create_cot_template(self, task_description, examples=None):
"""Chain of Thought prompting"""
template = f"""
Jesteś ekspertem w {task_description["domain"]}.
Rozwiąż następujący problem krok po kroku, pokazując swój proces myślowy.
ZADANIE: {task_description["task"]}
PROCES ROZWIĄZYWANIA:
1. ANALIZA PROBLEMU: Zidentyfikuj kluczowe elementy i wymagania
2. PLANOWANIE: Określ strategię rozwiązania
3. WYKONANIE: Przejdź przez rozwiązanie krok po kroku
4. WERYFIKACJA: Sprawdź poprawność rozwiązania
5. FINALNE ODPOWIEDŹ: Podaj końcowy wynik
Myśl głośno przez każdy krok:
"""
if examples:
template += "\n\nPRZYKŁADY:\n"
for i, example in enumerate(examples, 1):
template += f"\nPrzykład {i}:\n"
template += f"Problem: {example['problem']}\n"
template += f"Myślenie: {example['reasoning']}\n"
template += f"Odpowiedź: {example['answer']}\n"
template += f"\nTeraz rozwiąż: {task_description['specific_problem']}"
return template
def _create_few_shot_template(self, task_description, examples):
"""Few-shot learning template"""
template = f"""
Jesteś ekspertem w {task_description["domain"]}.
Oto kilka przykładów podobnych zadań:
"""
for i, example in enumerate(examples, 1):
template += f"Przykład {i}:\n"
template += f"Wejście: {example['input']}\n"
template += f"Wyjście: {example['output']}\n\n"
template += f"""
Na podstawie powyższych przykładów, wykonaj podobne zadanie:
Wejście: {task_description['input']}
Wyjście:"""
return template
def _create_tot_template(self, task_description, num_paths=3):
"""Tree of Thoughts prompting"""
template = f"""
Jesteś ekspertem w {task_description["domain"]}.
Użyj "Tree of Thoughts" aby rozwiązać ten problem, rozważając {num_paths} różne ścieżki rozwiązania.
PROBLEM: {task_description["problem"]}
PROCES:
1. Wygeneruj {num_paths} różne podejścia do rozwiązania
2. Dla każdego podejścia, oceń potencjalną skuteczność (1-10)
3. Wybierz najbardziej obiecujące podejście
4. Rozwijaj to podejście krok po kroku
5. Na każdym etapie sprawdzaj czy jesteś na właściwej ścieżce
PODEJŚCIE 1:
Opis: [opisz pierwsze podejście]
Potencjalna skuteczność: [1-10]
Zalety: [lista zalet]
Wady: [lista wad]
PODEJŚCIE 2:
Opis: [opisz drugie podejście]
Potencjalna skuteczność: [1-10]
Zalety: [lista zalet]
Wady: [lista wad]
PODEJŚCIE 3:
Opis: [opisz trzecie podejście]
Potencjalna skuteczność: [1-10]
Zalety: [lista zalet]
Wady: [lista wad]
WYBRANE PODEJŚCIE: [które wybrałeś i dlaczego]
SZCZEGÓŁOWE ROZWIĄZANIE:
[krok po kroku implementacja wybranego podejścia]
"""
return template
def _create_constitutional_template(self, task_description, principles):
"""Constitutional AI template"""
template = f"""
Jesteś AI assistant kierującym się następującymi zasadami:
KONSTYTUCYJNE ZASADY:
"""
for i, principle in enumerate(principles, 1):
template += f"{i}. {principle}\n"
template += f"""
ZADANIE: {task_description["task"]}
PROCES:
1. Przeanalizuj zadanie pod kątem zgodności z powyższymi zasadami
2. Zidentyfikuj potencjalne konflikty lub problemy etyczne
3. Jeśli zadanie jest zgodne z zasadami, wykonaj je
4. Jeśli istnieją problemy, zaproponuj alternatywne podejście zgodne z zasadami
5. Wyjaśnij swoje rozumowanie
ANALIZA ZGODNOŚCI:
[oceń zadanie względem każdej zasady]
ROZWIĄZANIE:
[przedstaw rozwiązanie zgodne z zasadami]
UZASADNIENIE:
[wyjaśnij dlaczego to rozwiązanie jest odpowiednie]
"""
return template
Prompt optimization pipeline
import asyncio
from dataclasses import dataclass
from typing import List, Dict, Optional
import statistics
@dataclass
class PromptEvaluation:
prompt: str
test_cases: List[Dict]
scores: Dict[str, float]
avg_score: float
execution_time: float
class PromptOptimizer:
def __init__(self, llm_client, evaluation_criteria):
self.llm_client = llm_client
self.evaluation_criteria = evaluation_criteria
self.optimization_history = []
async def optimize_prompt(self, base_prompt: str, test_dataset: List[Dict],
optimization_rounds: int = 5) -> PromptEvaluation:
"""Systematyczna optymalizacja prompta"""
current_prompt = base_prompt
best_evaluation = None
for round_num in range(optimization_rounds):
print(f"Optymalization round {round_num + 1}/{optimization_rounds}")
# Generuj warianty prompta
prompt_variants = await self._generate_prompt_variants(current_prompt)
# Ewaluuj wszystkie warianty
evaluations = []
for variant in prompt_variants:
evaluation = await self._evaluate_prompt(variant, test_dataset)
evaluations.append(evaluation)
# Wybierz najlepszy wariant
best_in_round = max(evaluations, key=lambda e: e.avg_score)
# Aktualizuj jeśli lepszy niż obecny najlepszy
if best_evaluation is None or best_in_round.avg_score > best_evaluation.avg_score:
best_evaluation = best_in_round
current_prompt = best_in_round.prompt
# Zapisz historię
self.optimization_history.append({
"round": round_num + 1,
"best_score": best_in_round.avg_score,
"variants_tested": len(prompt_variants),
"improvement": best_in_round.avg_score - (best_evaluation.avg_score if best_evaluation else 0)
})
# Early stopping jeśli brak poprawy
if round_num > 2:
recent_scores = [h["best_score"] for h in self.optimization_history[-3:]]
if len(set(recent_scores)) == 1: # Brak poprawy w ostatnich 3 rundach
print("Early stopping - brak poprawy")
break
return best_evaluation
async def _generate_prompt_variants(self, base_prompt: str) -> List[str]:
"""Generowanie wariantów prompta"""
variation_strategies = [
"Add more specific instructions",
"Include examples in different format",
"Change the role/persona",
"Add constraints and requirements",
"Modify output format specification",
"Add error handling instructions"
]
variants = [base_prompt] # Include original
for strategy in variation_strategies:
variant_prompt = f"""
Improve the following prompt using this strategy: {strategy}
Original prompt:
{base_prompt}
Improved prompt:"""
response = await self.llm_client.complete(variant_prompt)
variants.append(response.content.strip())
return variants
async def _evaluate_prompt(self, prompt: str, test_dataset: List[Dict]) -> PromptEvaluation:
"""Ewaluacja prompta na zestawie testowym"""
import time
start_time = time.time()
evaluations = []
for test_case in test_dataset:
# Stwórz kompletny prompt z test case
full_prompt = prompt + f"\n\nInput: {test_case['input']}\nOutput:"
try:
response = await self.llm_client.complete(full_prompt)
# Oceń odpowiedź
scores = {}
for criterion, evaluator in self.evaluation_criteria.items():
score = evaluator(
input_data=test_case['input'],
expected_output=test_case.get('expected_output'),
actual_output=response.content,
test_case_metadata=test_case
)
scores[criterion] = score
evaluations.append(scores)
except Exception as e:
print(f"Error evaluating test case: {e}")
# Dodaj zerowe scores dla failed cases
evaluations.append({criterion: 0.0 for criterion in self.evaluation_criteria})
# Oblicz średnie scores
avg_scores = {}
for criterion in self.evaluation_criteria:
criterion_scores = [eval_result[criterion] for eval_result in evaluations]
avg_scores[criterion] = statistics.mean(criterion_scores)
overall_avg = statistics.mean(avg_scores.values())
execution_time = time.time() - start_time
return PromptEvaluation(
prompt=prompt,
test_cases=test_dataset,
scores=avg_scores,
avg_score=overall_avg,
execution_time=execution_time
)
🔧 Fine-tuning modeli
Strategiczne podejście do fine-tuningu
from azure.ai.ml import MLClient
import json
from pathlib import Path
class ModelFineTuner:
def __init__(self, ml_client: MLClient):
self.ml_client = ml_client
self.fine_tuning_strategies = {
"domain_adaptation": self._prepare_domain_data,
"task_specific": self._prepare_task_data,
"style_transfer": self._prepare_style_data,
"instruction_following": self._prepare_instruction_data
}
def prepare_fine_tuning_dataset(self, raw_data: List[Dict],
strategy: str = "instruction_following") -> str:
"""Przygotowanie danych do fine-tuningu"""
if strategy not in self.fine_tuning_strategies:
raise ValueError(f"Unknown strategy: {strategy}")
processor = self.fine_tuning_strategies[strategy]
formatted_data = processor(raw_data)
# Walidacja danych
self._validate_training_data(formatted_data)
# Zapisz do pliku JSONL
output_file = f"training_data_{strategy}.jsonl"
with open(output_file, "w", encoding="utf-8") as f:
for item in formatted_data:
f.write(json.dumps(item, ensure_ascii=False) + "\n")
# Statystyki datasetu
stats = self._analyze_dataset(formatted_data)
print(f"Dataset prepared: {len(formatted_data)} examples")
print(f"Average input length: {stats['avg_input_length']}")
print(f"Average output length: {stats['avg_output_length']}")
return output_file
def _prepare_instruction_data(self, raw_data: List[Dict]) -> List[Dict]:
"""Formatowanie danych dla instruction following"""
formatted_data = []
for example in raw_data:
# Standardowy format dla instruction tuning
conversation = {
"messages": [
{
"role": "system",
"content": example.get("system_message", "Jesteś pomocnym asystentem AI.")
},
{
"role": "user",
"content": example["instruction"]
},
{
"role": "assistant",
"content": example["output"]
}
]
}
formatted_data.append(conversation)
return formatted_data
def _prepare_domain_data(self, raw_data: List[Dict]) -> List[Dict]:
"""Formatowanie danych dla domain adaptation"""
formatted_data = []
for example in raw_data:
# Format dla domain-specific knowledge
conversation = {
"messages": [
{
"role": "system",
"content": f"Jesteś ekspertem w dziedzinie {example.get('domain', 'specjalistycznej')}. "
f"Odpowiadaj precyzyjnie używając właściwej terminologii."
},
{
"role": "user",
"content": example["question"]
},
{
"role": "assistant",
"content": example["expert_answer"]
}
]
}
formatted_data.append(conversation)
return formatted_data
async def start_fine_tuning_job(self, training_file: str,
model_name: str = "gpt-35-turbo",
hyperparameters: Dict = None) -> str:
"""Rozpoczęcie procesu fine-tuningu"""
default_hyperparams = {
"n_epochs": 3,
"batch_size": 1,
"learning_rate_multiplier": 0.1
}
if hyperparameters:
default_hyperparams.update(hyperparameters)
try:
# Upload training file
with open(training_file, "rb") as f:
training_file_obj = openai.File.create(
file=f,
purpose="fine-tune"
)
# Create fine-tuning job
fine_tuning_job = openai.FineTuningJob.create(
training_file=training_file_obj.id,
model=model_name,
hyperparameters=default_hyperparams,
suffix="custom-model-v1"
)
job_id = fine_tuning_job.id
print(f"Fine-tuning job started: {job_id}")
# Monitor progress
await self._monitor_fine_tuning_progress(job_id)
return job_id
except Exception as e:
print(f"Error starting fine-tuning: {e}")
raise
async def _monitor_fine_tuning_progress(self, job_id: str):
"""Monitorowanie procesu fine-tuningu"""
import asyncio
while True:
job_status = openai.FineTuningJob.retrieve(job_id)
print(f"Status: {job_status.status}")
if job_status.status == "succeeded":
print(f"Fine-tuning completed successfully!")
print(f"Fine-tuned model: {job_status.fine_tuned_model}")
break
elif job_status.status == "failed":
print(f"Fine-tuning failed: {job_status.error}")
break
elif job_status.status in ["running", "validating_files"]:
print("Fine-tuning in progress...")
await asyncio.sleep(60) # Check every minute
else:
print(f"Unknown status: {job_status.status}")
await asyncio.sleep(30)
def evaluate_fine_tuned_model(self, model_id: str, test_dataset: List[Dict]) -> Dict:
"""Ewaluacja fine-tuned modelu"""
results = {
"model_id": model_id,
"test_cases": len(test_dataset),
"performance_metrics": {},
"example_outputs": []
}
correct_answers = 0
total_cases = len(test_dataset)
for i, test_case in enumerate(test_dataset):
try:
response = openai.ChatCompletion.create(
model=model_id,
messages=[
{"role": "user", "content": test_case["input"]}
],
max_tokens=150,
temperature=0.1
)
actual_output = response.choices[0].message.content
expected_output = test_case.get("expected_output", "")
# Simple accuracy check (można rozszerzyć)
is_correct = self._evaluate_answer_quality(actual_output, expected_output)
if is_correct:
correct_answers += 1
# Zapisz przykłady do analizy
if i < 5: # First 5 examples
results["example_outputs"].append({
"input": test_case["input"],
"expected": expected_output,
"actual": actual_output,
"correct": is_correct
})
except Exception as e:
print(f"Error evaluating test case {i}: {e}")
# Oblicz metryki
accuracy = (correct_answers / total_cases) * 100
results["performance_metrics"]["accuracy"] = accuracy
print(f"Model evaluation complete:")
print(f"Accuracy: {accuracy:.1f}%")
print(f"Correct answers: {correct_answers}/{total_cases}")
return results
✅ Zadania praktyczne
Zadanie 1: Chain of Thought Implementation (45 min)
- Stwórz CoT prompt dla złożonego problemu analitycznego
- Porównaj z zwykłym promptem
- Przetestuj na różnych typach zadań
- Oceń poprawę w jakości odpowiedzi
Zadanie 2: Prompt Optimization Pipeline (30 min)
- Zaimplementuj automatyczny optimizer promptów
- Stwórz zestaw testowy do ewaluacji
- Przeprowadź 3 rundy optymalizacji
- Zmierz poprawę w performance
Zadanie 3: Fine-tuning Experiment (60 min)
- Przygotuj dataset do fine-tuningu (50+ przykładów)
- Wytrenuj custom model
- Przetestuj vs base model
- Przeanalizuj improvements i trade-offs
Zadanie 4: Production Prompt Management (15 min)
- Stwórz system versioning promptów
- Zaimplementuj A/B testing
- Dodaj monitoring prompt performance
- Skonfiguruj automatyczne rollback
📊 Metryki ewaluacji
Kryteria oceny promptów
- Accuracy - poprawność odpowiedzi
- Relevance - trafność do zadania
- Consistency - stałość w różnych przypadkach
- Efficiency - stosunek jakości do długości prompta
- Safety - bezpieczeństwo i ethical alignment