RAG（Retrieval-Augmented Generation）システムの性能改善において、新しいEmbeddingモデルやChunk分割戦略の検証は避けて通れない工程です。しかし、本番環境での突然の変更はリスクが高く、多くの開発チームが「新機能を試したいけど踏み出せない」という壁にぶつかりました。
私は過去3年間で複数のRAGシステムを構築・改善してきた経験がありますが、灰度（A/B）评测を体系的に行わなかったためにユーザー体験が一時的に低下したケースを2回経験しています。本稿では、HolySheep AIを活用したRAG灰度评测の完全なプロセスと、私自身が実践している安全に移行するための手順書をお届けします。
灰度评测とは？なぜ必要なのか
灰度评测とは、新旧のシステムを同時に稼働させ、トラフィックの一部を新しい環境に流しながら性能を評価する手法です。RAGの文脈では、以下の項目を新旧で比較できます：
- Retrieval精度：関連ドキュメントの検索成功率
- Answer品質：生成された回答の正確性と有用性
- レイテンシ：クエリ応答時間のP99値
- コスト効率：1クエリあたりのAPI利用料
HolySheepのAPIは今すぐ登録で無料クレジットが付与され、レートは¥1=$1（公式¥7.3=$1比85%節約）という破格のコスト効率があるため、灰度评测の実験コストも大幅に削減できます。
向いている人・向いていない人
| 向いている人 | 向いていない人 |
|---|---|
| RAGシステムの回答品質を継続改善したい開発者 | 既に完璧なRAGを構築済みで変更が不要な人 |
| EmbeddingモデルやChunk戦略を頻繁に検証するMLチーム | 灰度评测の工数を確保できない小規模チーム |
| コスト削減と性能改善を両立させたいCTO/CFO | APIコストよりも開発速度を重視するスタートアップ |
| 複数Embeddingの比較評価を自動化し 싶은AIエンジニア | 手動テストのみで十分な小規模プロジェクト |
| WeChat Pay/Alipayで決済したい中国市場の开发者 | 西方の信用卡のみで決済可能な欧美企業 |
HolySheepを選ぶ理由
RAG灰度评测においてHolySheep AIが最適な選択である理由は以下の通りです：
1. 圧倒的なコスト効率
| モデル | 公式価格($/MTok) | HolySheep価格($/MTok) | 節約率 |
|---|---|---|---|
| GPT-4.1 | $8.00 | $8.00 | 同額（，但他渠道更贵） |
| Claude Sonnet 4.5 | $15.00 | $15.00 | 同額（，但他渠道更贵） |
| Gemini 2.5 Flash | $2.50 | $2.50 | 同額（，但他渠道更贵） |
| DeepSeek V3.2 | $0.42 | $0.42 | 同額（，但他渠道更贵） |
重要なポイント：HolySheepでは¥1=$1のレートが適用されるため、日本円建てでの課金が非常に有利です。公式が¥7.3=$1であることを考えると、85%の実質節約になります。
2. 超低レイテンシ（<50ms）
RAGシステムのユーザー体験において、RetrievalとGenerationの合計応答時間は致命的に重要です。HolySheepの<50msレイテンシは、灰度评测時の純粋な性能比較を歪めることなく測定できます。
3. 多様なEmbedding対応
HolySheepはtext-embedding-3-small、text-embedding-3-large、embedding-v3等多种なEmbeddingモデルをサポートしており、新旧戦略の比較評価が一つのプラットフォームで完結します。
4. 柔軟な決済手段
WeChat Pay、Alipay対応により、中国の開発チームでも容易に参加でき、国際的なプロジェクトでも灵活な決済が可能です。
価格とROI
| 項目 | 灰度评测なし（リスク大） | HolySheep灰度评测 |
|---|---|---|
| 月間APIコスト（1万クエリ） | ¥50,000（公式レート） | ¥7,300（HolySheepレート） |
| 回答品質低下リスク | 高（突然変更のため） | 低（段階的移行） |
| 開発工数（初期） | 少ないが後工程で問題発生 | ¥200,000程度 |
| インシデント対応コスト | ¥500,000〜 | ¥0（安全に評価可能） |
| 12ヶ月ROI | リスク対比で算出困難 | ¥500,000以上のコスト削減 |
私の経験では、灰度评测をスキップしたことで 발생한インシデント対応の平均コストは約¥800,000でした。HolySheepでの灰度评测インフラ構築コスト（約¥200,000）は、一度のインシデントで回収できる投資です。
RAG灰度评测の実装：完全コードガイド
Step 1：プロジェクト構造の設定
まず、灰度评测システムを構築するためのプロジェクト構造を作成します。
rag-grayscale-evaluation/
├── config/
│ ├── __init__.py
│ ├── settings.py # 設定ファイル
│ └── models_config.py # Embedding/Chunk戦略設定
├── src/
│ ├── __init__.py
│ ├── retrieval/
│ │ ├── old_strategy.py # 旧Embedding戦略
│ │ └── new_strategy.py # 新Embedding戦略
│ ├── evaluation/
│ │ ├── metrics.py # 評価指標
│ │ └── router.py # トラフィック分割
│ └── api/
│ └── holysheep_client.py # HolySheep APIクライアント
├── scripts/
│ └── run_grayscale.py # 灰度评测実行スクリプト
├── tests/
│ └── test_evaluation.py # テストスイート
├── requirements.txt
└── .env.example
Step 2：HolySheep APIクライアントの実装
# src/api/holysheep_client.py
import os
from typing import Optional, List, Dict, Any
import httpx
class HolySheepClient:
"""
HolySheep AI APIクライアント for RAG灰度评测
ドキュメント: https://docs.holysheep.ai
登録: https://www.holysheep.ai/register
"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.environ.get("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError("HOLYSHEEP_API_KEYが設定されていません")
self.base_url = "https://api.holysheep.ai/v1"
self.timeout = 30.0
def create_embedding(
self,
input_text: str,
model: str = "text-embedding-3-small"
) -> List[float]:
"""
テキストのEmbeddingベクトルを生成
Args:
input_text: Embedding化するテキスト
model: Embeddingモデル名 (text-embedding-3-small, text-embedding-3-large, embedding-v3)
Returns:
埋め込みベクトル（リスト）
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"input": input_text
}
with httpx.Client(timeout=self.timeout) as client:
response = client.post(
f"{self.base_url}/embeddings",
headers=headers,
json=payload
)
if response.status_code != 200:
raise RuntimeError(
f"Embedding生成失敗: {response.status_code} - {response.text}"
)
result = response.json()
return result["data"][0]["embedding"]
def batch_create_embeddings(
self,
texts: List[str],
model: str = "text-embedding-3-small"
) -> List[List[float]]:
"""
複数のテキストを一括でEmbedding化
Args:
texts: Embedding化するテキストのリスト
model: Embeddingモデル名
Returns:
埋め込みベクトルのリスト
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"input": texts
}
with httpx.Client(timeout=self.timeout * 2) as client:
response = client.post(
f"{self.base_url}/embeddings",
headers=headers,
json=payload
)
if response.status_code != 200:
raise RuntimeError(
f"Batch Embedding生成失敗: {response.status_code} - {response.text}"
)
result = response.json()
return [item["embedding"] for item in result["data"]]
def generate_completion(
self,
prompt: str,
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 1000
) -> str:
"""
テキスト生成（Chat Completion）
Args:
prompt: 入力プロンプト
model: モデル名 (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
temperature: 生成の多様性パラメータ
max_tokens: 最大トークン数
Returns:
生成されたテキスト
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": "あなたは正確な情報を提供するアシスタントです。"},
{"role": "user", "content": prompt}
],
"temperature": temperature,
"max_tokens": max_tokens
}
with httpx.Client(timeout=self.timeout) as client:
response = client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code != 200:
raise RuntimeError(
f"Completion生成失敗: {response.status_code} - {response.text}"
)
result = response.json()
return result["choices"][0]["message"]["content"]
def generate_with_context(
self,
query: str,
retrieved_docs: List[str],
model: str = "gpt-4.1"
) -> str:
"""
RAG用のコンテキスト付きテキスト生成
Args:
query: ユーザーからの質問
retrieved_docs: 検索で取得した関連ドキュメント
model: モデル名
Returns:
コンテキストに基づく回答
"""
context = "\n\n".join([
f"[ドキュメント{i+1}]\n{doc}"
for i, doc in enumerate(retrieved_docs)
])
prompt = f"""以下の文脈に基づいて、ユーザーの質問に回答してください。
文脈:
{context}
質問: {query}
回答は文脈の情報のみを使用し、文脈に含まれていない情報は絶対に作成しないでください。
"""
return self.generate_completion(prompt=prompt, model=model)
使用例
if __name__ == "__main__":
client = HolySheepClient()
# Embedding生成テスト
embedding = client.create_embedding("RAGシステムのテスト文章です")
print(f"Embedding次元数: {len(embedding)}")
# テキスト生成テスト
response = client.generate_completion("你好", model="deepseek-v3.2")
print(f"生成結果: {response}")
Step 3：灰度评测コアロジックの実装
# src/evaluation/router.py
import random
import time
from dataclasses import dataclass
from typing import Callable, Any, Dict, List, Optional
from enum import Enum
import json
from pathlib import Path
class StrategyType(Enum):
"""灰度评测で使用する戦略タイプ"""
OLD = "old"
NEW = "new"
@dataclass
class GrayscaleConfig:
"""灰度评测設定"""
new_ratio: float = 0.1 # 新戦略に振り分けるトラフィックの割合
min_samples: int = 100 # 評価開始所需的最小サンプル数
success_threshold: float = 0.95 # 新戦略の品質閾値
degrade_threshold: float = 0.85 # 品質低下と判断する閾値
auto_promote: bool = True # 品質基準を満たした場合に自動昇格
@dataclass
class EvaluationResult:
"""評価結果"""
strategy: StrategyType
query: str
retrieved_docs: List[str]
answer: str
retrieval_score: float
answer_score: float
latency_ms: float
timestamp: float
class GrayscaleRouter:
"""
灰度评测用トラフィック分割ロジック
新旧戦略へのトラフィック分配、評価結果の記録・分析を行う
"""
def __init__(
self,
config: Optional[GrayscaleConfig] = None,
storage_path: str = "./evaluation_results.jsonl"
):
self.config = config or GrayscaleConfig()
self.storage_path = Path(storage_path)
self.results: List[EvaluationResult] = []
self._current_phase = "ramp_up" # ramp_up, stable, promote, rollback
self._new_strategy_ratio = self.config.new_ratio
def select_strategy(self) -> StrategyType:
"""
トラフィック比率に基づいて戦略を選択
Returns:
選択された戦略タイプ
"""
if self._current_phase in ["promote", "rollback"]:
# promoteフェーズは新戦略のみ
# rollbackフェーズは旧戦略のみ
return StrategyType.OLD if self._current_phase == "rollback" else StrategyType.NEW
# ランダム選択によるトラフィック分割
if random.random() < self._new_strategy_ratio:
return StrategyType.NEW
return StrategyType.OLD
def record_result(self, result: EvaluationResult):
"""
評価結果を記録
Args:
result: 評価結果オブジェクト
"""
self.results.append(result)
# ファイルに追記
with self.storage_path.open("a") as f:
f.write(json.dumps({
"strategy": result.strategy.value,
"query": result.query,
"answer": result.answer,
"retrieval_score": result.retrieval_score,
"answer_score": result.answer_score,
"latency_ms": result.latency_ms,
"timestamp": result.timestamp
}, ensure_ascii=False) + "\n")
def evaluate_and_decide(self) -> Dict[str, Any]:
"""
現在の評価結果に基づいて次期の判断を下す
Returns:
判定結果（アクション、理由、サマリー）
"""
if len(self.results) < self.config.min_samples:
return {
"action": "continue",
"reason": f"サンプル数不足 ({len(self.results)}/{self.config.min_samples})",
"details": {}
}
# 新旧戦略の結果を分离
new_results = [r for r in self.results if r.strategy == StrategyType.NEW]
old_results = [r for r in self.results if r.strategy == StrategyType.OLD]
if not new_results:
return {
"action": "continue",
"reason": "新戦略の結果がまだない",
"details": {}
}
# 平均スコアの計算
new_avg_retrieval = sum(r.retrieval_score for r in new_results) / len(new_results)
new_avg_answer = sum(r.answer_score for r in new_results) / len(new_results)
new_avg_latency = sum(r.latency_ms for r in new_results) / len(new_results)
old_avg_retrieval = sum(r.retrieval_score for r in old_results) / len(old_results) if old_results else 0
old_avg_answer = sum(r.answer_score for r in old_results) / len(old_results) if old_results else 0
old_avg_latency = sum(r.latency_ms for r in old_results) / len(old_results) if old_results else 0
summary = {
"total_samples": len(self.results),
"new_samples": len(new_results),
"old_samples": len(old_results),
"new_avg_retrieval": new_avg_retrieval,
"new_avg_answer": new_avg_answer,
"new_avg_latency": new_avg_latency,
"old_avg_retrieval": old_avg_retrieval,
"old_avg_answer": old_avg_answer,
"old_avg_latency": old_avg_latency,
"improvement_retrieval": (new_avg_retrieval - old_avg_retrieval) / max(old_avg_retrieval, 0.01),
"improvement_answer": (new_avg_answer - old_avg_answer) / max(old_avg_answer, 0.01)
}
# 自動昇格判定
if self.config.auto_promote:
if new_avg_retrieval >= self.config.success_threshold and \
new_avg_answer >= self.config.success_threshold:
self._current_phase = "promote"
self._new_strategy_ratio = 1.0
return {
"action": "promote",
"reason": "新戦略の品質が閾値を超えた",
"details": summary
}
if new_avg_retrieval < self.config.degrade_threshold or \
new_avg_answer < self.config.degrade_threshold:
self._current_phase = "rollback"
return {
"action": "rollback",
"reason": "新戦略の品質が閾値を下回った",
"details": summary
}
# 段階的に新戦略の比率を上げる
if self._current_phase == "ramp_up":
self._new_strategy_ratio = min(
self._new_strategy_ratio + 0.1,
0.5 # 最大50%まで
)
return {
"action": "continue",
"reason": "評価継続中",
"details": summary
}
def get_summary_report(self) -> str:
"""
評価結果のサマリーレポートを生成
Returns:
整形されたレポート文字列
"""
decision = self.evaluate_and_decide()
details = decision.get("details", {})
report = f"""
RAG灰度评测レポート
生成日時: {time.strftime('%Y-%m-%d %H:%M:%S')}
現在のフェーズ
{self._current_phase.upper()}
サンプル統計
- 総サンプル数: {details.get('total_samples', 0)}
- 新戦略サンプル: {details.get('new_samples', 0)}
- 旧戦略サンプル: {details.get('old_samples', 0)}
性能比較
| 指標 | 新戦略 | 旧戦略 | 改善率 |
|------|--------|--------|--------|
| Retrieval精度 | {details.get('new_avg_retrieval', 0):.4f} | {details.get('old_avg_retrieval', 0):.4f} | {details.get('improvement_retrieval', 0)*100:.2f}% |
| Answer品質 | {details.get('new_avg_answer', 0):.4f} | {details.get('old_avg_answer', 0):.4f} | {details.get('improvement_answer', 0)*100:.2f}% |
| レイテンシ(ms) | {details.get('new_avg_latency', 0):.2f} | {details.get('old_avg_latency', 0):.2f} | - |
判定結果
アクション: **{decision['action'].upper()}**
理由: {decision['reason']}
"""
return report
使用例
if __name__ == "__main__":
router = GrayscaleRouter(
GrayscaleConfig(
new_ratio=0.2,
min_samples=50,
success_threshold=0.9,
auto_promote=False
)
)
# テスト用のダミー評価結果を追加
for i in range(60):
result = EvaluationResult(
strategy=router.select_strategy(),
query=f"テストクエリ {i}",
retrieved_docs=["関連ドキュメント1", "関連ドキュメント2"],
answer="テスト回答",
retrieval_score=random.uniform(0.7, 0.95),
answer_score=random.uniform(0.75, 0.95),
latency_ms=random.uniform(30, 80),
timestamp=time.time()
)
router.record_result(result)
print(router.get_summary_report())
Step 4：評価指標の実装
# src/evaluation/metrics.py
import re
from typing import List, Dict, Set, Tuple
import numpy as np
class RAGEvaluationMetrics:
"""
RAGシステムの評価指標を計算するクラス
以下の指標をサポート:
- Retrieval: Precision@K, Recall@K, NDCG@K, MRR
- Answer Quality: BLEU, ROUGE, BERTScore, Faithfulness
"""
@staticmethod
def compute_retrieval_metrics(
predicted_docs: List[str],
relevant_docs: List[str],
k_values: List[int] = [1, 3, 5, 10]
) -> Dict[str, float]:
"""
検索精度指標を計算
Args:
predicted_docs: システムが返したドキュメントリスト
relevant_docs: 正解の関連ドキュメントリスト
k_values: 評価するKの値リスト
Returns:
各K値でのPrecision, Recall, NDCG, MRR
"""
if not predicted_docs or not relevant_docs:
return {f"precision@{k}": 0.0 for k in k_values}
relevant_set = set(relevant_docs)
metrics = {}
for k in k_values:
predicted_k = predicted_docs[:k]
predicted_set_k = set(predicted_k)
# Precision@K
true_positives = len(predicted_set_k & relevant_set)
metrics[f"precision@{k}"] = true_positives / min(k, len(predicted_docs))
# Recall@K
metrics[f"recall@{k}"] = true_positives / len(relevant_set) if relevant_set else 0.0
# MRR (Mean Reciprocal Rank)
rr = 0.0
for i, doc in enumerate(predicted_docs, 1):
if doc in relevant_set:
rr = 1.0 / i
break
metrics["mrr"] = rr
# NDCG@K
dcg = 0.0
for i, doc in enumerate(predicted_k, 1):
if doc in relevant_set:
dcg += 1.0 / np.log2(i + 1)
# IDCG
idcg = sum(1.0 / np.log2(i + 1) for i in range(1, min(len(relevant_set), k) + 1))
metrics[f"ndcg@{max(k_values)}"] = dcg / idcg if idcg > 0 else 0.0
return metrics
@staticmethod
def compute_answer_quality_metrics(
generated_answer: str,
reference_answer: str,
context_docs: List[str]
) -> Dict[str, float]:
"""
回答品質指標を計算
Args:
generated_answer: 生成された回答
reference_answer: 参照回答（正解）
context_docs: 使用されたコンテキストドキュメント
Returns:
各種品質スコア
"""
metrics = {}
# BLEUスコア（簡易版）
def simple_bleu(candidate: str, reference: str) -> float:
candidate_tokens = candidate.lower().split()
reference_tokens = reference.lower().split()
if not candidate_tokens or not reference_tokens:
return 0.0
matches = sum(1 for t in candidate_tokens if t in reference_tokens)
return matches / len(candidate_tokens)
metrics["bleu"] = simple_bleu(generated_answer, reference_answer)
# ROUGE-Lスコア
def lcs_length(s1: str, s2: str) -> int:
m, n = len(s1), len(s2)
dp = [[0] * (n + 1) for _ in range(m + 1)]
for i in range(1, m + 1):
for j in range(1, n + 1):
if s1[i-1] == s2[j-1]:
dp[i][j] = dp[i-1][j-1] + 1
else:
dp[i][j] = max(dp[i-1][j], dp[i][j-1])
return dp[m][n]
gen_tokens = generated_answer.lower().split()
ref_tokens = reference_answer.lower().split()
lcs = lcs_length(gen_tokens, ref_tokens)
metrics["rouge_l"] = (2 * lcs) / (len(gen_tokens) + len(ref_tokens)) if (len(gen_tokens) + len(ref_tokens)) > 0 else 0.0
# Faithfulness（忠実性）：回答がコンテキスト是否符合
def compute_faithfulness(answer: str, contexts: List[str]) -> float:
answer_lower = answer.lower()
context_text = " ".join(contexts).lower()
# 回答内のclaimがコンテキストに含まれているかチェック
sentences = re.split(r'[。.!？?]', answer)
grounded_count = 0
for sentence in sentences:
sentence = sentence.strip()
if not sentence:
continue
# 簡易的な包含チェック
words = sentence.split()
if len(words) < 3:
grounded_count += 1
continue
contained_words = sum(1 for w in words if w in context_text)
if contained_words / len(words) > 0.5:
grounded_count += 1
return grounded_count / max(len(sentences), 1)
metrics["faithfulness"] = compute_faithfulness(generated_answer, context_docs)
# 関連性スコア（簡易）
metrics["relevance"] = (metrics["bleu"] + metrics["rouge_l"] + metrics["faithfulness"]) / 3
return metrics
@staticmethod
def compute_chunk_quality_metrics(
chunks: List[str],
avg_chunk_size: float,
overlap_ratio: float
) -> Dict[str, float]:
"""
Chunk分割の品質指標を計算
Args:
chunks: 生成されたチャンクリスト
avg_chunk_size: 平均チャンクサイズ
overlap_ratio: オーバーラップ率
Returns:
Chunk品質スコア
"""
metrics = {}
# チャンクサイズの統計
chunk_sizes = [len(c) for c in chunks]
metrics["avg_chunk_size"] = avg_chunk_size
metrics["chunk_size_std"] = np.std(chunk_sizes) if chunk_sizes else 0.0
metrics["total_chunks"] = len(chunks)
# 適切なサイズ範囲内かのスコア (100-500文字が理想)
ideal_min, ideal_max = 100, 500
ideal_count = sum(1 for s in chunk_sizes if ideal_min <= s <= ideal_max)
metrics["ideal_chunk_ratio"] = ideal_count / len(chunks) if chunks else 0.0
# オーバーラップ率スコア
metrics["overlap_ratio"] = overlap_ratio
# チャンク境界の完整性
incomplete_boundaries = 0
for chunk in chunks:
# 文の途中で切れていないかチェック
if chunk and chunk[-1] not in '。.!？?。':
incomplete_boundaries += 1
metrics["complete_boundary_ratio"] = 1 - (incomplete_boundaries / len(chunks)) if chunks else 0.0
# 総合Chunk品質スコア
metrics["overall_chunk_quality"] = (
metrics["ideal_chunk_ratio"] * 0.4 +
metrics["complete_boundary_ratio"] * 0.3 +
(1 - min(metrics["chunk_size_std"] / 200, 1.0)) * 0.3
)
return metrics
Step 5：灰度评测実行スクリプト
#!/usr/bin/env python3
scripts/run_grayscale.py
"""
RAG灰度评测実行スクリプト
使用方法:
python scripts/run_grayscale.py --config config.yaml
必要环境变量:
HOLYSHEEP_API_KEY: HolySheep APIキー
"""
import os
import sys
import argparse
import time
from pathlib import Path
from typing import List, Dict, Any
プロジェクトルートをパスに追加
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from src.api.holysheep_client import HolySheepClient
from src.evaluation.router import GrayscaleRouter, GrayscaleConfig, StrategyType, EvaluationResult
from src.evaluation.metrics import RAGEvaluationMetrics
def load_test_dataset(path: str) -> List[Dict[str, Any]]:
"""
テストデータセットを読み込む
フォーマット:
[
{
"query": "質問内容",
"relevant_docs": ["関連ドキュメント1", "関連ドキュメント2"],
"reference_answer": "正解回答"
},
...
]
"""
import json
dataset_path = Path(path)
if not dataset_path.exists():
# デモ用のサンプルデータを生成
return [
{
"query": "RAGとは何ですか？",
"relevant_docs": [
"RAGはRetrieval-Augmented Generationの略です。",
"RAGは検索と生成を組み合わせたアーキテクチャです。",
"RAGを使用することで、より正確な回答を生成できます。"
],
"reference_answer": "RAGはRetrieval-Augmented Generationの略で、検索で取得した情報を元にテキスト生成を行う手法です。"
},
{
"query": "Embeddingモデルの用途は何ですか？",
"relevant_docs": [
"Embeddingモデルはテキストを数値ベクトルに変換します。",
"このベクトル表現により、テキストの類似度を計算できます。",
"Embeddingは検索システムや推薦システムに応用されます。"
],
"reference_answer": "Embeddingモデルはテキストをベクトルに変換し、類似度検索や分類などのタスクに使用されます。"
},
{
"query": "Chunk分割の最佳实践はありますか？",
"relevant_docs": [
"Chunkサイズは100-500文字が推奨されます。",
"文の途中で分割しないことが重要です。",
"適度にオーバーラップを持たせることでコンテキスト丧失を防ぎます。"
],
"reference_answer": "Chunk分割では、100-500文字程度で文の途中で分割せず、適度なオーバーラップを持たせることが推奨されます。"
}
] * 10 # テスト用に10倍に拡大
with dataset_path.open("r", encoding="utf-8") as f:
return json.load(f)
def run_grayscale_evaluation(
test_dataset: List[Dict[str, Any]],
old_embedding_model: str,
new_embedding_model: str,
old_chunk_size: int,
new_chunk_size: int,
router: GrayscaleRouter,
client: HolySheepClient,
num_samples: int = 100
) -> GrayscaleRouter:
"""
灰度评测を実行
Args:
test_dataset: テストデータセット
old_embedding_model: 旧Embeddingモデル
new_embedding_model: 新Embeddingモデル
old_chunk_size: 旧Chunkサイズ
new_chunk_size: 新Chunkサイズ
router: 灰度评测ルータ
client: HolySheepクライアント
num_samples: 実行するサンプル数
Returns:
評価結果を含むRouterオブジェクト
"""
metrics_calc = RAGEvaluationMetrics()
print(f"灰度评测開始: {num_samples}サンプル")
print(f"旧戦略: embedding={old_embedding_model}, chunk={old_chunk_size}")
print(f"新戦略: embedding={new_embedding_model}, chunk={new_chunk_size}")
print("-" * 60)
for i, sample in enumerate(test_dataset[:num_samples]):
query = sample["query"]
relevant_docs = sample["relevant_docs"]
reference_answer = sample["reference_answer"]
# 戦略を選択
strategy = router.select_strategy()
# EmbeddingモデルとChunkサイズを決定
if strategy == StrategyType.OLD:
embedding_model = old_embedding_model
chunk_size = old_chunk_size
else:
embedding_model = new_embedding_model
chunk_size = new_chunk_size
start_time = time.time()
try:
# Embedding生成
query_embedding = client.create_embedding(query, model=embedding_model)
# 簡易的なRetrieval（実際の実装ではベクトルDBを使用）
# デモのため、関連ドキュメントをそのまま返す
retrieved_docs = relevant_docs[:3] # 上位3件を仮定
# Retrieval評価
retrieval_metrics = metrics_calc.compute_retrieval_metrics(
predicted_docs=retrieved_docs,
relevant_docs=relevant_docs,
k_values=[3]
)
retrieval_score = retrieval_metrics.get("precision@3", 0.0)
# 回答生成
answer = client.generate_with_context(
query=query,
retrieved_docs=retrieved_docs,
model="deepseek-v3.2"