Szybka inżynieria w produkcji: szablon, wersjonowanie i testowanie
Il szybka inżynieria jest często traktowane jako działanie eksperymentalne: próbujesz czegoś, to działa, idziesz do przodu. W produkcji takie podejście zawodzi systematycznie. Zmiana podpowiedzi może pogorszyć jakość odpowiedzi tak, żeby nikt tego nie zauważył. Podpowiedź zoptymalizowana pod GPT-4 może dać wyniki bardzo źle na GPT-4o-mini. Szablon, który działa w języku angielskim, może nie działać w języku włoskim.
W tym artykule szybką inżynierię traktujemy jako: dyscyplina inżynierska: zaawansowane techniki (Chain-of-Thought, Few-Shot, Constitutional AI), systemy szablonów ze zmiennymi i kompozycją, szybkie wersjonowanie z oceną A/B, testowanie automatyka i kontrola jakości w produkcji. Każda sekcja zawiera kod Plik wykonywalny i wzorce Pythona testowane na rzeczywistych systemach.
Czego się nauczysz
- Zaawansowane techniki: Łańcuch Myśli, Uczenie się kilkoma strzałami, Drzewo Myśli
- System szablonów ze zmiennymi, kompozycją i dziedziczeniem
- Szybkie wersjonowanie ze śledzeniem wydajności
- Testy A/B podpowiedzi o znaczeniu statystycznym
- Konstytucyjna sztuczna inteligencja i poręcze zapewniające bezpieczne wyjścia
- Monity o uporządkowane dane wyjściowe (JSON, XML, Markdown)
- Zautomatyzowane szybkie testy z LLM-jako sędzią
- Monitorowanie jakości monitu w produkcji
1. Zaawansowane techniki podpowiedzi
1.1 Łańcuch myślenia (CoT)
Łańcuch myśli (Wei i in., 2022) to najskuteczniejsza technika nowoczesne podpowiadanie: poproś modela, aby „pokazał rozumowanie” przed podaniem reakcja znacznie zwiększa dokładność w przypadku złożonych problemów.
Łańcuch myślowy a standardowe podpowiedzi
# STANDARD PROMPTING - scarsa accuratezza su ragionamento complesso
standard_prompt = """
Questo cliente deve pagare una fattura di 1200 euro.
Ha già pagato il 30%. Quanto deve ancora pagare?
"""
# Risposta tipica: "840 euro" (spesso corretto ma senza garanzie)
# CHAIN-OF-THOUGHT - alta accuratezza
cot_prompt = """
Questo cliente deve pagare una fattura di 1200 euro.
Ha già pagato il 30%. Quanto deve ancora pagare?
Ragiona passo per passo:
1. Prima calcola quanto ha già pagato
2. Poi calcola il rimanente
3. Fornisci la risposta finale
"""
# Risposta:
# "1. Ha pagato: 1200 * 30/100 = 360 euro
# 2. Rimanente: 1200 - 360 = 840 euro
# 3. Il cliente deve ancora pagare 840 euro."
# ZERO-SHOT CoT - basta aggiungere "Let's think step by step"
zero_shot_cot = """
Questo cliente deve pagare una fattura di 1200 euro.
Ha già pagato il 30%. Quanto deve ancora pagare?
Pensiamo passo per passo:
"""
# Il modello genera autonomamente il ragionamento step-by-step
# Implementazione in Python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
cot_template = ChatPromptTemplate.from_template("""
{context}
Domanda: {question}
Ragiona passo per passo prima di rispondere. Struttura la risposta come:
RAGIONAMENTO:
[il tuo ragionamento dettagliato]
RISPOSTA FINALE:
[risposta concisa]
""")
chain = cot_template | llm
response = chain.invoke({
"context": "Il prezzo base è 1000 euro, con IVA 22% e sconto fedele del 10%.",
"question": "Qual è il prezzo finale da pagare?"
})
1.2 Uczenie się kilkoma strzałami
Il Monit o kilka strzałów zawiera przykłady wejść i wyjść w znaku zachęty dla kierować zachowaniem modelu. Jest to szczególnie skuteczne w przypadku zadań z określonego formatu wyjściowego lub wyspecjalizowanych dziedzin, w których model ma niewielką wiedzę.
Monitowanie o kilka strzałów z dynamicznym wyborem przykładów
from langchain_core.prompts import FewShotChatMessagePromptTemplate
from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
# Libreria di esempi (task di classificazione sentiment)
examples = [
{
"input": "Il prodotto è arrivato rotto e il supporto non risponde.",
"output": "NEGATIVO - Problema prodotto e supporto clienti"
},
{
"input": "Consegna rapidissima e prodotto esattamente come descritto!",
"output": "POSITIVO - Soddisfazione consegna e prodotto"
},
{
"input": "Il prezzo è nella media, niente di speciale.",
"output": "NEUTRO - Valutazione prezzo"
},
{
"input": "qualità eccellente, lo riacquistero sicuramente.",
"output": "POSITIVO - Alta qualità, fidelizzazione"
},
{
"input": "Spedizione lenta, prodotto ok ma poteva andare meglio.",
"output": "MISTO - Problema spedizione, prodotto accettabile"
},
# ... altri esempi
]
# Selettore semantico: scegli i 3 esempi più simili alla query
example_selector = SemanticSimilarityExampleSelector.from_examples(
examples,
OpenAIEmbeddings(),
FAISS,
k=3
)
# Template per gli esempi
example_prompt = ChatPromptTemplate.from_messages([
("human", "{input}"),
("ai", "{output}")
])
# Few-shot prompt con selezione dinamica
few_shot_prompt = FewShotChatMessagePromptTemplate(
example_selector=example_selector,
example_prompt=example_prompt,
)
# Prompt finale
final_prompt = ChatPromptTemplate.from_messages([
("system", """Sei un analista di sentiment per recensioni e-commerce.
Classifica il sentiment come: POSITIVO, NEGATIVO, NEUTRO, o MISTO.
Includi sempre la categoria principale del problema/punto di forza."""),
few_shot_prompt,
("human", "{input}")
])
chain = final_prompt | llm
result = chain.invoke({"input": "Prodotto ottimo ma imballaggio pessimo."})
# Output: "MISTO - qualità prodotto vs problema imballaggio"
1.3 Wyjście strukturalne i wywoływanie funkcji
Jeden z najważniejszych wzorców produkcji: zmuszanie LLM do produkcji ustrukturyzowany (modele JSON, Pydantic) zamiast wolnego tekstu. Wyeliminuj parsowanie ręcznie i drastycznie zmniejsza liczbę błędów formatowania.
Ustrukturyzowane dane wyjściowe za pomocą Pydantic i LangChain
from pydantic import BaseModel, Field
from typing import List, Optional, Literal
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# Definisci il schema dell'output
class ProductAnalysis(BaseModel):
"""Analisi strutturata di un prodotto"""
sentiment: Literal["positivo", "negativo", "neutro", "misto"] = Field(
description="Il sentiment generale della recensione"
)
score: int = Field(
description="Score da 1 a 10", ge=1, le=10
)
punti_forza: List[str] = Field(
description="Lista dei punti di forza menzionati",
default_factory=list
)
punti_deboli: List[str] = Field(
description="Lista dei punti deboli menzionati",
default_factory=list
)
categoria: str = Field(
description="Categoria principale (es. 'qualità', 'spedizione', 'supporto')"
)
risposta_suggerita: Optional[str] = Field(
description="Risposta suggerita per il team supporto (se sentiment negativo)",
default=None
)
class ReviewBatchResult(BaseModel):
"""Risultato dell'analisi di un batch di recensioni"""
total_reviews: int
sentiment_distribution: dict
top_issues: List[str]
recommendations: List[str]
# LLM con structured output
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(ProductAnalysis)
prompt = ChatPromptTemplate.from_template("""
Analizza questa recensione di prodotto e classifica il sentiment.
Recensione: {review}
Estrai tutte le informazioni richieste in modo preciso.""")
chain = prompt | structured_llm
# L'output è già un oggetto Pydantic validato
result: ProductAnalysis = chain.invoke({
"review": "Prodotto di ottima qualità, spedizione lenta. Supporto ha risposto velocemente."
})
print(f"Sentiment: {result.sentiment}")
print(f"Score: {result.score}/10")
print(f"Punti forza: {result.punti_forza}")
print(f"Punti deboli: {result.punti_deboli}")
# Output garantito:
# Sentiment: misto
# Score: 6/10
# Punti forza: ['qualità prodotto', 'supporto reattivo']
# Punti deboli: ['spedizione lenta']
2. System szablonowy dla produkcji
W środowisku produkcyjnym podpowiedzi należy traktować jak zasoby najwyższej klasy: wersjonowane, testowalne, komponowalne. Umożliwia to solidny system szablonów monity o aktualizację bez zmiany kodu aplikacji.
Rejestr szablonów podpowiedzi z wersjonowaniem
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from datetime import datetime
import hashlib
import json
import yaml
from pathlib import Path
@dataclass
class PromptVersion:
"""Una versione specifica di un prompt"""
version: str
template: str
variables: List[str]
description: str
created_at: datetime = field(default_factory=datetime.now)
created_by: str = ""
tags: List[str] = field(default_factory=list)
performance_metrics: Dict[str, float] = field(default_factory=dict)
is_active: bool = True
@property
def template_hash(self) -> str:
"""Hash del template per deduplication"""
return hashlib.md5(self.template.encode()).hexdigest()[:8]
def render(self, **kwargs) -> str:
"""Renderizza il template con le variabili fornite"""
try:
return self.template.format(**kwargs)
except KeyError as e:
raise ValueError(f"Variabile mancante nel template: {e}")
def validate_variables(self, provided: Dict) -> List[str]:
"""Verifica che tutte le variabili richieste siano fornite"""
missing = [v for v in self.variables if v not in provided]
return missing
class PromptRegistry:
"""
Registry centralizzato per la gestione dei prompt in produzione.
Supporta versioning, A/B testing e rollback.
"""
def __init__(self, storage_path: str = "./prompts"):
self.storage_path = Path(storage_path)
self.storage_path.mkdir(exist_ok=True)
self.prompts: Dict[str, List[PromptVersion]] = {}
self._load_from_disk()
def register(
self,
name: str,
template: str,
variables: List[str],
description: str = "",
version: Optional[str] = None,
tags: List[str] = None
) -> PromptVersion:
"""Registra una nuova versione di un prompt"""
if name not in self.prompts:
self.prompts[name] = []
# Auto-versioning se non specificato
if version is None:
existing = len(self.prompts[name])
version = f"v{existing + 1:03d}"
prompt_version = PromptVersion(
version=version,
template=template,
variables=variables,
description=description,
tags=tags or []
)
# Disattiva versione precedente attiva
for existing_v in self.prompts[name]:
existing_v.is_active = False
self.prompts[name].append(prompt_version)
self._save_to_disk(name, prompt_version)
return prompt_version
def get(self, name: str, version: Optional[str] = None) -> PromptVersion:
"""Ottieni una versione del prompt"""
if name not in self.prompts:
raise KeyError(f"Prompt '{name}' non trovato nel registry")
if version is None:
# Versione attiva più recente
active = [p for p in self.prompts[name] if p.is_active]
if not active:
raise ValueError(f"Nessuna versione attiva per '{name}'")
return active[-1]
for p in self.prompts[name]:
if p.version == version:
return p
raise KeyError(f"Versione '{version}' non trovata per '{name}'")
def rollback(self, name: str, version: str) -> PromptVersion:
"""Rollback a una versione precedente"""
target = self.get(name, version)
# Disattiva tutto
for p in self.prompts[name]:
p.is_active = False
# Attiva la versione target
target.is_active = True
return target
def update_metrics(self, name: str, version: str, metrics: Dict[str, float]):
"""Aggiorna le metriche di performance di una versione"""
prompt = self.get(name, version)
prompt.performance_metrics.update(metrics)
self._save_to_disk(name, prompt)
def get_history(self, name: str) -> List[Dict]:
"""Ottieni la storia delle versioni"""
if name not in self.prompts:
return []
return [
{
"version": p.version,
"created_at": p.created_at.isoformat(),
"is_active": p.is_active,
"metrics": p.performance_metrics,
"hash": p.template_hash
}
for p in self.prompts[name]
]
def _save_to_disk(self, name: str, prompt: PromptVersion):
"""Persisti il prompt su disco"""
file_path = self.storage_path / f"{name}_{prompt.version}.yaml"
data = {
"version": prompt.version,
"template": prompt.template,
"variables": prompt.variables,
"description": prompt.description,
"tags": prompt.tags,
"is_active": prompt.is_active,
"performance_metrics": prompt.performance_metrics
}
with open(file_path, 'w') as f:
yaml.dump(data, f, allow_unicode=True)
def _load_from_disk(self):
"""Carica i prompt da disco all'avvio"""
for file_path in self.storage_path.glob("*.yaml"):
try:
with open(file_path) as f:
data = yaml.safe_load(f)
name = "_".join(file_path.stem.split("_")[:-1])
if name not in self.prompts:
self.prompts[name] = []
self.prompts[name].append(PromptVersion(**data))
except Exception:
pass
# Utilizzo
registry = PromptRegistry()
# Registra prompt RAG v1
registry.register(
name="rag_answer",
template="""Sei un assistente tecnico. Rispondi basandoti SOLO sul contesto.
Contesto: {context}
Domanda: {question}
Risposta:""",
variables=["context", "question"],
description="Prompt RAG base v1"
)
# Registra prompt RAG v2 (migliorato)
registry.register(
name="rag_answer",
template="""Sei un assistente tecnico preciso. Rispondi basandoti ESCLUSIVAMENTE
sul contesto fornito. Se il contesto non contiene la risposta, dillo esplicitamente.
Contesto:
{context}
Domanda: {question}
Fornisci una risposta concisa e accurata:""",
variables=["context", "question"],
description="Prompt RAG v2 - più chiaro sul fallback"
)
# Usa la versione attiva
prompt = registry.get("rag_answer")
rendered = prompt.render(context="...", question="...")
3. Testowanie A/B podpowiedzi
Testy A/B pozwalają porównać dwie wersje podpowiedzi na rzeczywistym ruchu z istotnością statystyczną. Ważne jest, aby sprawdzić, czy zmiany w zachęta faktycznie poprawia jakość, a nie ją pogarsza.
Struktura testów A/B dla podpowiedzi
import random
from scipy import stats
from typing import Callable, Tuple
import numpy as np
class PromptABTest:
"""Framework per A/B testing di prompt con significativita statistica"""
def __init__(
self,
prompt_a: PromptVersion,
prompt_b: PromptVersion,
traffic_split: float = 0.5, # 50% traffico a B
min_samples: int = 100 # Campioni minimi per significativita
):
self.prompt_a = prompt_a
self.prompt_b = prompt_b
self.traffic_split = traffic_split
self.min_samples = min_samples
self.results_a: List[float] = [] # Scores per prompt A
self.results_b: List[float] = [] # Scores per prompt B
def assign_variant(self, user_id: str = None) -> Tuple[str, PromptVersion]:
"""
Assegna una variante all'utente.
Deterministica se user_id e fornito (stesso utente vede sempre la stessa variante).
"""
if user_id:
# Hashing deterministico per consistenza per utente
h = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
use_b = (h % 100) < (self.traffic_split * 100)
else:
use_b = random.random() < self.traffic_split
if use_b:
return "B", self.prompt_b
return "A", self.prompt_a
def record_result(self, variant: str, score: float):
"""Registra il risultato per una variante"""
if variant == "A":
self.results_a.append(score)
else:
self.results_b.append(score)
def is_statistically_significant(self, alpha: float = 0.05) -> bool:
"""Verifica significativita statistica con t-test"""
if len(self.results_a) < self.min_samples or len(self.results_b) < self.min_samples:
return False
_, p_value = stats.ttest_ind(self.results_a, self.results_b)
return p_value < alpha
def get_winner(self) -> Optional[str]:
"""Determina il vincitore se statisticamente significativo"""
if not self.is_statistically_significant():
return None # Non abbastanza dati
mean_a = np.mean(self.results_a)
mean_b = np.mean(self.results_b)
return "B" if mean_b > mean_a else "A"
def report(self) -> dict:
"""Report completo del test"""
mean_a = np.mean(self.results_a) if self.results_a else 0
mean_b = np.mean(self.results_b) if self.results_b else 0
p_value = None
if len(self.results_a) > 1 and len(self.results_b) > 1:
_, p_value = stats.ttest_ind(self.results_a, self.results_b)
return {
"samples_a": len(self.results_a),
"samples_b": len(self.results_b),
"mean_score_a": mean_a,
"mean_score_b": mean_b,
"improvement_pct": ((mean_b - mean_a) / mean_a * 100) if mean_a > 0 else 0,
"p_value": p_value,
"is_significant": self.is_statistically_significant(),
"winner": self.get_winner(),
"recommendation": "Deploy B" if self.get_winner() == "B" else
"Keep A" if self.get_winner() == "A" else
"Collect more data"
}
4. Testowanie automatyczne z LLM-as-Judge
Il LLM jako sędzia jest to najbardziej skalowalny wzorzec oceny jakości podpowiedzi: do oceny wykorzystuje się LLM (często silniejszy niż testowany). wyniki automatycznie, symulując ocenę człowieka przy obniżonych kosztach.
LLM-as-sędzia do szybkiego testowania
from langchain_openai import ChatOpenAI
from pydantic import BaseModel
class JudgeScore(BaseModel):
"""Score strutturato del giudice LLM"""
accuracy: int # 1-5: precisione fattuale
relevance: int # 1-5: rilevanza alla domanda
clarity: int # 1-5: chiarezza della risposta
completeness: int # 1-5: completezza della risposta
overall: int # 1-5: valutazione complessiva
reasoning: str # Spiegazione del punteggio
issues: list # Lista dei problemi riscontrati
class PromptTester:
"""Framework di testing automatico per prompt LLM"""
def __init__(
self,
model_under_test: ChatOpenAI,
judge_model: ChatOpenAI = None
):
self.model = model_under_test
self.judge = judge_model or ChatOpenAI(model="gpt-4o-mini", temperature=0)
self.structured_judge = self.judge.with_structured_output(JudgeScore)
def evaluate_single(
self,
prompt: str,
input_vars: Dict,
expected_output: str = None
) -> JudgeScore:
"""Valuta un singolo output del prompt"""
# Genera output con il modello testato
rendered_prompt = prompt.format(**input_vars)
actual_output = self.model.invoke(rendered_prompt).content
# Valuta con il giudice LLM
judge_prompt = f"""Valuta questa risposta AI su scale 1-5 per ogni dimensione.
Domanda/Input: {input_vars.get('question', rendered_prompt[:200])}
Risposta generata: {actual_output}
{f"Risposta attesa: {expected_output}" if expected_output else ""}
Valuta obiettivamente e fornisci esempi specifici per ogni punto."""
return self.structured_judge.invoke(judge_prompt)
def run_test_suite(
self,
prompt: str,
test_cases: List[Dict]
) -> dict:
"""
Esegui una suite di test su un prompt.
test_cases: lista di {"input": {vars}, "expected": "output atteso"}
"""
scores = []
failed_cases = []
for i, test_case in enumerate(test_cases):
try:
score = self.evaluate_single(
prompt=prompt,
input_vars=test_case["input"],
expected_output=test_case.get("expected")
)
scores.append(score)
# Flag test case con score basso
if score.overall < 3:
failed_cases.append({
"index": i,
"input": test_case["input"],
"score": score.overall,
"issues": score.issues,
"reasoning": score.reasoning
})
except Exception as e:
failed_cases.append({"index": i, "error": str(e)})
if not scores:
return {"error": "Nessun test completato"}
return {
"total_tests": len(test_cases),
"completed": len(scores),
"avg_overall": np.mean([s.overall for s in scores]),
"avg_accuracy": np.mean([s.accuracy for s in scores]),
"avg_relevance": np.mean([s.relevance for s in scores]),
"avg_clarity": np.mean([s.clarity for s in scores]),
"pass_rate": len([s for s in scores if s.overall >= 3]) / len(scores),
"failed_cases": failed_cases,
"recommendation": "DEPLOY" if np.mean([s.overall for s in scores]) >= 4.0
else "REVIEW" if np.mean([s.overall for s in scores]) >= 3.0
else "REJECT"
}
# Test suite di esempio per un prompt RAG
test_suite = [
{
"input": {
"context": "LangChain è un framework Python per costruire applicazioni LLM.",
"question": "Cos'è LangChain?"
},
"expected": "LangChain è un framework per costruire applicazioni basate su LLM"
},
{
"input": {
"context": "Il prezzo di abbonamento base è 29 euro al mese.",
"question": "Quanto costa l'abbonamento premium?"
},
"expected": "Il contesto non specifica il prezzo dell'abbonamento premium"
},
]
5. Najlepsze praktyki i antywzorce
Najlepsze praktyki Szybka inżynieria w produkcji
- Wersjonuj monity, takie jak kod: każda zmiana w podpowiedzi jest potencjalną zmianą zakłócającą. Używaj wersjonowania semantycznego, prowadź dziennik zmian i testuj przed wdrożeniem.
- Jeśli to możliwe, używaj ustrukturyzowanych danych wyjściowych: JSON/Pydantic eliminuje ręczne analizowanie i niespodzianki związane z formatowaniem. Są bardziej niezawodne niż analizowanie wyrażeń regularnych dowolnego tekstu.
- Test z danymi przypadku Edge: najbardziej krytyczne pytania to te, które nie są dystrybuowane. Twój zestaw testowy musi zawierać przypadki Edge, niejednoznaczne pytania i zniekształcone dane wejściowe.
- Łańcuch myślowy do złożonych zadań: w przypadku każdego zadania wymagającego więcej niż jednego etapu rozumowania CoT znacznie poprawia jakość (często +20-40%).
- Kilka strzałów z różnorodnymi przykładami: zawierać przykłady obejmujące wszystkie główne przypadki, a nie tylko te najczęstsze. Różnorodność przykładów zapobiega systematycznej stronniczości.
Anty-wzorce, których należy unikać
- Zakodowane na stałe podpowiedzi w kodzie: podpowiedzi w kodzie źródłowym nie można zaktualizować w środowisku produkcyjnym bez wdrożenia. Użyj rejestru lub pliku konfiguracyjnego.
- Brak testów przed wdrożeniem: monit, który działa na 10 ręcznych przykładach, może nie działać w rzeczywistych przypadkach Edge. Twórz zestawy testów zawierające co najmniej 50–100 różnych przypadków.
- Wysokie temperatury dla zadań deterministycznych: do klasyfikacji, ekstrakcji danych i analizy zawsze używaj temperatury = 0. Zmienność jest wrogiem spójności.
- Zbyt długie monity bez struktury: ponad 500 tokenów, modele mają tendencję do utraty informacji pomiędzy nimi. Zorganizuj monit za pomocą przejrzystych sekcji i używaj wypunktowań.
Wnioski
Szybka inżynieria w produkcji wymaga takiego samego rygoru jak oprogramowanie inżynieria tradycyjna: wersjonowanie, testowanie, monitorowanie i kontrolowane wdrażanie. Widzieliśmy, jak zastosować zaawansowane techniki (CoT, Few-Shot, Structured Output), jak zarządzać monitami za pomocą wersjonowanego rejestru, jak przeprowadzać testy A/B istotność statystyczna i jak zautomatyzować ocenę za pomocą LLM-as-sędziego.
Kluczowe punkty:
- Łańcuch myślenia znacznie poprawia dokładność złożonych zadań
- Ustrukturyzowane dane wyjściowe (Pydantic) eliminują błędy analizy i gwarantują format
- Podpowiedzi muszą być wersjonowane, testowane i wdrażane jak każdy kod
- Testy A/B o znaczeniu statystycznym przed promowaniem nowych wersji
- LLM-as-sędzia skaluje ocenę jakości przy obniżonych kosztach
Seria trwa
- Artykuł 8: Systemy wieloagentowe
- Artykuł 9: Szybka inżynieria w produkcji (prąd)
- Artykuł 10: Grafy wiedzy dotyczące sztucznej inteligencji