本番環境での迅速なエンジニアリング: テンプレート、バージョン管理、テスト
Il 迅速なエンジニアリング 多くの場合、実験的な活動として扱われます。 何かを試してみるとうまくいき、前に進むことができます。運用環境では、このアプローチは失敗します 体系的に。プロンプトを変更すると、回答の品質が低下する可能性があります 誰にも気付かれずに。 GPT-4 に最適化されたプロンプトで結果が得られる GPT-4o-mini では非常に悪い。英語では機能するテンプレートがイタリア語では機能しない場合があります。
この記事では、プロンプトエンジニアリングを 工学分野: 高度な技術 (Chain-of-Thought、Few-Shot、Constitutional AI)、テンプレート システム 変数と構成を使用したプロンプト バージョニングと A/B 評価、テスト 生産時の自動品質監視。各セクションにはコードが含まれています Python 実行可能ファイルとパターンは実際のシステムでテストされています。
何を学ぶか
- 高度なテクニック: 思考連鎖、少数ショット学習、思考ツリー
- 変数、合成、継承を備えたテンプレート システム
- パフォーマンス追跡による迅速なバージョン管理
- 統計的に有意なプロンプトの A/B テスト
- 安全な出力のための憲法上の AI とガードレール
- 構造化出力 (JSON、XML、Markdown) のプロンプト
- LLM を審査員として使用する自動化されたプロンプト テスト
- 本番環境でのプロンプトの品質の監視
1. 高度なプロンプトテクニック
1.1 思考連鎖 (CoT)
思考の連鎖 (Wei et al., 2022) は、最も影響力のある手法です。 現代的なプロンプト: 与える前にモデルに「推論を示す」よう依頼します。 応答により、複雑な問題の精度が大幅に向上します。
思考連鎖と標準的なプロンプト
# 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 少数ショット学習
Il 少数ショットのプロンプト プロンプトには入出力の例が含まれています モデルの動作をガイドします。特に次のようなタスクに有効です。 特定の出力形式や、モデルの知識がほとんどない特殊な領域。
動的なサンプル選択による少数ショットのプロンプト
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 構造化された出力と関数呼び出し
運用環境で最も重要なパターンの 1 つ: LLM に出力を強制する フリーテキストではなく構造化（JSON、Pydanticモデル）。解析を排除する 手動で行うため、フォーマットエラーが大幅に減少します。
Pydantic と 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. 本番用テンプレートシステム
運用環境では、プロンプトはファーストクラスのリソースとして処理される必要があります。 バージョン管理、テスト可能、コンポーザブル。堅牢なテンプレート システムにより、次のことが可能になります。 アプリケーションコードを変更せずにプロンプトを更新します。
バージョン管理付きのプロンプト テンプレート レジストリ
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. プロンプトの A/B テスト
A/B テストにより、実際のトラフィックで 2 つのバージョンのプロンプトを比較できます 統計的に有意です。への変更を検証することが重要です。 プロンプトは実際に品質を向上させるものであり、品質を低下させるものではありません。
プロンプトの A/B テスト フレームワーク
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. LLM-as-Judge による自動テスト
Il 裁判官としてのLLM これは品質を評価するための最も拡張性の高いパターンです プロンプトの評価: LLM (多くの場合、テストされたものよりも強力) が評価に使用されます。 出力は自動的に行われ、コストを削減しながら人間による評価をシミュレートします。
即時テストの裁判官としての LLM
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. ベストプラクティスとアンチパターン
本番環境での迅速なエンジニアリングのベスト プラクティス
- コードのようにプロンプトをバージョン付けします。 プロンプトに対するすべての変更は、破壊的変更となる可能性があります。セマンティック バージョニングを使用し、変更ログを保存し、デプロイ前にテストします。
- 可能な限り構造化された出力を使用します。 JSON/Pydantic により、手動による解析やフォーマットの予期せぬ作業が不要になります。これらはフリー テキストの正規表現解析よりも信頼性が高くなります。
- エッジケースデータを使用してテストします。 最も重要な質問は配布されていないものです。テスト セットには、エッジ ケース、あいまいな質問、不正な形式の入力が含まれている必要があります。
- 複雑なタスクのための思考連鎖: 複数の推論ステップを必要とするタスクでは、CoT によって品質が大幅に向上します (通常は +20 ～ 40%)。
- 多様な例を含む少数のショット: 最も一般的なケースだけでなく、すべての主要なケースをカバーする例が含まれています。例が多様であるため、系統的な偏見が防止されます。
避けるべきアンチパターン
- コード内のハードコーディングされたプロンプト: ソース コード内のプロンプトは、デプロイせずに運用環境で更新することはできません。レジストリまたは構成ファイルを使用します。
- 導入前にテストなし: 10 個の手動サンプルで機能するプロンプトは、実際のエッジ ケースでは失敗する可能性があります。少なくとも 50 ～ 100 の異なるケースを含むテスト スイートを構築します。
- 確定的なタスクの場合は高温: 分類、データ抽出、分析には常に温度=0 を使用します。変動性は一貫性の敵です。
- 構造のないプロンプトが長すぎます: 500 トークンを超えると、モデルはその間の情報を失う傾向があります。プロンプトを明確なセクションで構成し、箇条書きを使用します。
結論
生産における迅速なエンジニアリングには、ソフトウェアと同じ厳密さが必要です 従来のエンジニアリング: バージョン管理、テスト、監視、および制御された展開。 高度なテクニック (CoT、Few-Shot、Structured Output) を適用する方法を見てきました。 バージョン管理されたレジストリを使用してプロンプトを管理する方法、A/B テストを実施する方法 統計的有意性と、LLM-as-judge を使用して評価を自動化する方法。
重要なポイント:
- 思考の連鎖により、複雑なタスクの精度が大幅に向上します
- 構造化された出力 (Pydantic) により解析エラーが排除され、形式が保証されます
- プロンプトは他のコードと同様にバージョン管理、テスト、デプロイする必要があります
- 新しいバージョンを宣伝する前に統計的有意性を備えた A/B テストを行う
- 審査員としての LLM は、コストを削減して品質評価をスケールします
シリーズは続く
- 第 8 条: マルチエージェント システム
- 第 9 条: 生産における迅速なエンジニアリング (現行)
- 第 10 条: AI のナレッジ グラフ