LLMプロンプトエンジニアリング上級技法: Chain-of-Thought·Tree-of-Thought·ReAct·Few-Shotパターン実践ガイド

from openai import OpenAI

client = OpenAI()

def zero_shot_classification(text: str) -> str:
    """Zero-shotテキスト分類"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a text classifier. "
                    "Classify the given text into one of the following categories: "
                    "Technology, Business, Science, Sports, Entertainment. "
                    "Respond with only the category name."
                )
            },
            {"role": "user", "content": text}
        ],
        temperature=0,
        max_tokens=20,
    )
    return response.choices[0].message.content.strip()

def few_shot_entity_extraction(text: str) -> str:
    """Few-shotエンティティ抽出"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "Extract named entities from the given text in the specified format."
            },
            {
                "role": "user",
                "content": "Samsung Electronics announced the Galaxy S25 series at CES 2025 in Las Vegas."
            },
            {
                "role": "assistant",
                "content": (
                    "- Organization: Samsung Electronics\n"
                    "- Product: Galaxy S25\n"
                    "- Event: CES 2025\n"
                    "- Location: Las Vegas"
                )
            },
            {
                "role": "user",
                "content": "Elon Musk revealed that Tesla will open a new Gigafactory in Austin, Texas in March 2026."
            },
            {
                "role": "assistant",
                "content": (
                    "- Person: Elon Musk\n"
                    "- Organization: Tesla\n"
                    "- Facility: Gigafactory\n"
                    "- Location: Austin, Texas\n"
                    "- Date: March 2026"
                )
            },
            {"role": "user", "content": text}
        ],
        temperature=0,
    )
    return response.choices[0].message.content

# Few-shot例選択戦略
class FewShotSelector:
    """動的Few-shot例選択器"""
    def __init__(self, examples, embedding_model="text-embedding-3-small"):
        self.examples = examples
        self.client = OpenAI()
        self.embedding_model = embedding_model
        self._precompute_embeddings()

    def _precompute_embeddings(self):
        """全例の埋め込みを事前計算"""
        texts = [ex["input"] for ex in self.examples]
        response = self.client.embeddings.create(
            model=self.embedding_model,
            input=texts
        )
        self.embeddings = [r.embedding for r in response.data]

    def select(self, query: str, k: int = 3) -> list:
        """クエリに最も類似したk個の例を選択"""
        query_emb = self.client.embeddings.create(
            model=self.embedding_model,
            input=[query]
        ).data[0].embedding

        # コサイン類似度計算
        import numpy as np
        similarities = []
        for emb in self.embeddings:
            sim = np.dot(query_emb, emb) / (
                np.linalg.norm(query_emb) * np.linalg.norm(emb)
            )
            similarities.append(sim)

        # 上位k個を選択
        top_indices = np.argsort(similarities)[-k:][::-1]
        return [self.examples[i] for i in top_indices]

def chain_of_thought_math(problem: str) -> dict:
    """Chain-of-Thought数学問題解法"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a math tutor. Solve problems step by step.\n"
                    "Show your reasoning clearly before giving the final answer.\n"
                    "Format: Step 1, Step 2, ..., Final Answer: [number]"
                )
            },
            {
                "role": "user",
                "content": (
                    "A store has 45 apples. They sell 1/3 of them in the morning "
                    "and 2/5 of the remaining in the afternoon. "
                    "How many apples are left?"
                )
            },
            {
                "role": "assistant",
                "content": (
                    "Step 1: Calculate apples sold in the morning.\n"
                    "Morning sales = 45 * (1/3) = 15 apples\n\n"
                    "Step 2: Calculate remaining apples after morning.\n"
                    "Remaining = 45 - 15 = 30 apples\n\n"
                    "Step 3: Calculate apples sold in the afternoon.\n"
                    "Afternoon sales = 30 * (2/5) = 12 apples\n\n"
                    "Step 4: Calculate final remaining apples.\n"
                    "Final remaining = 30 - 12 = 18 apples\n\n"
                    "Final Answer: 18"
                )
            },
            {"role": "user", "content": problem}
        ],
        temperature=0,
    )

    answer_text = response.choices[0].message.content
    # 最終回答抽出
    import re
    match = re.search(r"Final Answer:\s*(\d+)", answer_text)
    final_answer = int(match.group(1)) if match else None

    return {
        "reasoning": answer_text,
        "answer": final_answer,
        "tokens_used": response.usage.total_tokens,
    }

def zero_shot_cot(problem: str) -> str:
    """Zero-shot Chain-of-Thought"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": f"{problem}\n\nLet's think step by step."
            }
        ],
        temperature=0,
    )
    return response.choices[0].message.content

import collections
import re

def self_consistency(problem: str, num_samples: int = 5) -> dict:
    """Self-Consistencyデコーディング"""
    answers = []
    reasoning_paths = []

    for i in range(num_samples):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Solve the math problem step by step. "
                        "End with 'Final Answer: [number]'"
                    )
                },
                {"role": "user", "content": problem}
            ],
            temperature=0.7,  # 多様な推論パスのためtemperatureを上げる
            max_tokens=500,
        )

        text = response.choices[0].message.content
        reasoning_paths.append(text)

        # 回答抽出
        match = re.search(r"Final Answer:\s*(\d+)", text)
        if match:
            answers.append(int(match.group(1)))

    # 多数決投票
    if answers:
        counter = collections.Counter(answers)
        majority_answer = counter.most_common(1)[0][0]
        confidence = counter.most_common(1)[0][1] / len(answers)
    else:
        majority_answer = None
        confidence = 0.0

    return {
        "answer": majority_answer,
        "confidence": confidence,
        "all_answers": answers,
        "num_samples": num_samples,
        "answer_distribution": dict(counter) if answers else {},
    }

from dataclasses import dataclass
from typing import Optional

@dataclass
class ThoughtNode:
    """ToTの思考ノード"""
    content: str
    score: float = 0.0
    children: list = None
    parent: Optional['ThoughtNode'] = None
    depth: int = 0

    def __post_init__(self):
        if self.children is None:
            self.children = []

class TreeOfThought:
    """Tree-of-Thoughtフレームワーク"""
    def __init__(self, model="gpt-4o", max_depth=3, branching_factor=3):
        self.client = OpenAI()
        self.model = model
        self.max_depth = max_depth
        self.branching_factor = branching_factor

    def generate_thoughts(self, problem: str, current_thought: str) -> list:
        """現在の状態から可能な次の思考を生成"""
        prompt = (
            f"Problem: {problem}\n\n"
            f"Current reasoning so far:\n{current_thought}\n\n"
            f"Generate {self.branching_factor} different possible next steps "
            f"for solving this problem. "
            f"Format each as 'Step N: [reasoning]' separated by '---'"
        )
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.8,
        )
        text = response.choices[0].message.content
        thoughts = [t.strip() for t in text.split("---") if t.strip()]
        return thoughts[:self.branching_factor]

    def evaluate_thought(self, problem: str, thought_path: str) -> float:
        """思考パスの有望性を0-1で評価"""
        prompt = (
            f"Problem: {problem}\n\n"
            f"Reasoning path:\n{thought_path}\n\n"
            f"Evaluate this reasoning path on a scale of 0.0 to 1.0:\n"
            f"- 1.0: Correct and complete solution\n"
            f"- 0.7-0.9: On the right track, promising\n"
            f"- 0.4-0.6: Partially correct but uncertain\n"
            f"- 0.0-0.3: Wrong approach or contains errors\n\n"
            f"Respond with only the score (e.g., 0.8)"
        )
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            max_tokens=10,
        )
        try:
            score = float(response.choices[0].message.content.strip())
            return min(max(score, 0.0), 1.0)
        except ValueError:
            return 0.5

    def solve_bfs(self, problem: str) -> dict:
        """BFSベースのToT探索"""
        root = ThoughtNode(content="", depth=0)
        current_level = [root]
        best_solution = None
        best_score = 0.0

        for depth in range(self.max_depth):
            next_level = []
            for node in current_level:
                # 子思考を生成
                thought_path = self._get_path(node)
                children_thoughts = self.generate_thoughts(problem, thought_path)

                for thought in children_thoughts:
                    full_path = f"{thought_path}\n{thought}" if thought_path else thought
                    score = self.evaluate_thought(problem, full_path)

                    child = ThoughtNode(
                        content=thought,
                        score=score,
                        parent=node,
                        depth=depth + 1
                    )
                    node.children.append(child)
                    next_level.append(child)

                    if score > best_score:
                        best_score = score
                        best_solution = full_path

            # 上位branching_factor個のみ保持（ビームサーチ）
            next_level.sort(key=lambda n: n.score, reverse=True)
            current_level = next_level[:self.branching_factor]

        return {
            "solution": best_solution,
            "score": best_score,
            "depth_explored": self.max_depth,
        }

    def _get_path(self, node: ThoughtNode) -> str:
        """ノードまでの全思考パスを返す"""
        path = []
        current = node
        while current and current.content:
            path.append(current.content)
            current = current.parent
        return "\n".join(reversed(path))

import json
from typing import Callable

class ReActAgent:
    """ReActパターンベースのエージェント"""

    def __init__(self, model="gpt-4o"):
        self.client = OpenAI()
        self.model = model
        self.tools = {}
        self.max_iterations = 10

    def register_tool(self, name: str, func: Callable, description: str):
        """外部ツール登録"""
        self.tools[name] = {
            "function": func,
            "description": description,
        }

    def _build_system_prompt(self) -> str:
        """システムプロンプト構築"""
        tool_descriptions = "\n".join([
            f"- {name}: {info['description']}"
            for name, info in self.tools.items()
        ])

        return (
            "You are a helpful assistant that solves problems step by step.\n"
            "You have access to the following tools:\n"
            f"{tool_descriptions}\n\n"
            "For each step, respond in the following format:\n"
            "Thought: [your reasoning about what to do next]\n"
            "Action: [tool_name(argument)]\n\n"
            "After receiving an observation, continue with another Thought.\n"
            "When you have the final answer, respond with:\n"
            "Thought: [final reasoning]\n"
            "Final Answer: [your answer]\n\n"
            "IMPORTANT: Use exactly one Action per step. "
            "Wait for the Observation before proceeding."
        )

    def run(self, query: str) -> dict:
        """ReActループ実行"""
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": query},
        ]

        steps = []

        for iteration in range(self.max_iterations):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0,
                max_tokens=500,
            )
            assistant_msg = response.choices[0].message.content

            # Final Answerチェック
            if "Final Answer:" in assistant_msg:
                final_answer = assistant_msg.split("Final Answer:")[-1].strip()
                steps.append({
                    "type": "final",
                    "content": assistant_msg,
                })
                return {
                    "answer": final_answer,
                    "steps": steps,
                    "iterations": iteration + 1,
                }

            # Actionパースおよび実行
            import re
            action_match = re.search(r"Action:\s*(\w+)\((.+?)\)", assistant_msg)
            if action_match:
                tool_name = action_match.group(1)
                tool_arg = action_match.group(2).strip("'\"")

                steps.append({
                    "type": "thought_action",
                    "content": assistant_msg,
                    "tool": tool_name,
                    "argument": tool_arg,
                })

                # ツール実行
                if tool_name in self.tools:
                    try:
                        observation = self.tools[tool_name]["function"](tool_arg)
                    except Exception as e:
                        observation = f"Error: {str(e)}"
                else:
                    observation = f"Error: Tool '{tool_name}' not found"

                steps.append({
                    "type": "observation",
                    "content": str(observation),
                })

                # メッセージ履歴に追加
                messages.append({"role": "assistant", "content": assistant_msg})
                messages.append({
                    "role": "user",
                    "content": f"Observation: {observation}"
                })
            else:
                # Actionがなければ履歴に追加して続行
                messages.append({"role": "assistant", "content": assistant_msg})
                messages.append({
                    "role": "user",
                    "content": "Please continue with an Action or provide the Final Answer."
                })

        return {
            "answer": "Max iterations reached",
            "steps": steps,
            "iterations": self.max_iterations,
        }

# 使用例
def create_research_agent():
    """リサーチエージェント作成"""
    agent = ReActAgent()

    # ツール登録
    def search(query):
        # 実際には検索API呼び出し
        return f"Search results for '{query}': [simulated results]"

    def calculate(expression):
        return str(eval(expression))

    def get_current_date():
        from datetime import datetime
        return datetime.now().strftime("%Y-%m-%d")

    agent.register_tool("search", search, "Search the web for information")
    agent.register_tool("calculate", calculate, "Evaluate a math expression")
    agent.register_tool("get_date", lambda _: get_current_date(), "Get current date")

    return agent

from pydantic import BaseModel, Field
from typing import Literal

# Pydanticモデルを活用した構造化出力
class SentimentResult(BaseModel):
    """感情分析結果スキーマ"""
    sentiment: Literal["positive", "negative", "neutral"]
    confidence: float = Field(ge=0.0, le=1.0)
    key_phrases: list[str]
    reasoning: str

def structured_sentiment_analysis(text: str) -> SentimentResult:
    """構造化出力で感情分析を実行"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "Analyze the sentiment of the given text. "
                    "Respond in JSON format with the following fields:\n"
                    "- sentiment: 'positive', 'negative', or 'neutral'\n"
                    "- confidence: float between 0.0 and 1.0\n"
                    "- key_phrases: list of key phrases that influenced the sentiment\n"
                    "- reasoning: brief explanation of the analysis"
                )
            },
            {"role": "user", "content": text}
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )

    result = json.loads(response.choices[0].message.content)
    return SentimentResult(**result)

# Function Callingベースの構造化
def function_calling_extraction(text: str) -> dict:
    """Function Callingを活用した情報抽出"""
    tools = [
        {
            "type": "function",
            "function": {
                "name": "extract_meeting_info",
                "description": "Extract meeting information from text",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "date": {
                            "type": "string",
                            "description": "Meeting date in YYYY-MM-DD format"
                        },
                        "time": {
                            "type": "string",
                            "description": "Meeting time in HH:MM format"
                        },
                        "participants": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "List of participants"
                        },
                        "agenda": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "Meeting agenda items"
                        },
                        "location": {
                            "type": "string",
                            "description": "Meeting location or meeting link"
                        }
                    },
                    "required": ["date", "time", "participants"]
                }
            }
        }
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Extract meeting info: {text}"}],
        tools=tools,
        tool_choice={"type": "function", "function": {"name": "extract_meeting_info"}},
    )

    tool_call = response.choices[0].message.tool_calls[0]
    return json.loads(tool_call.function.arguments)

class PromptChain:
    """プロンプトチェイニングフレームワーク"""

    def __init__(self, model="gpt-4o"):
        self.client = OpenAI()
        self.model = model
        self.steps = []
        self.results = {}

    def add_step(self, name: str, prompt_template: str, depends_on: list = None):
        """チェインにステップを追加"""
        self.steps.append({
            "name": name,
            "prompt_template": prompt_template,
            "depends_on": depends_on or [],
        })

    def run(self, initial_input: str) -> dict:
        """チェイン全体を実行"""
        self.results["input"] = initial_input

        for step in self.steps:
            # 依存ステップの結果でプロンプトを構成
            prompt = step["prompt_template"]
            prompt = prompt.replace("INPUT", self.results.get("input", ""))
            for dep in step["depends_on"]:
                prompt = prompt.replace(
                    f"RESULT_{dep.upper()}",
                    self.results.get(dep, "")
                )

            response = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0,
            )

            self.results[step["name"]] = response.choices[0].message.content

        return self.results

# 使用例：技術文書の要約 + 翻訳 + キーワード抽出
def create_document_pipeline():
    """文書処理パイプライン"""
    chain = PromptChain()

    chain.add_step(
        name="summary",
        prompt_template=(
            "Summarize the following technical document in 3-5 bullet points:\n\n"
            "INPUT"
        )
    )

    chain.add_step(
        name="translation",
        prompt_template=(
            "Translate the following summary to Korean:\n\n"
            "RESULT_SUMMARY"
        ),
        depends_on=["summary"]
    )

    chain.add_step(
        name="keywords",
        prompt_template=(
            "Extract 5-10 technical keywords from the following summary. "
            "Format as a comma-separated list:\n\n"
            "RESULT_SUMMARY"
        ),
        depends_on=["summary"]
    )

    return chain

# プロンプティング技法選択の意思決定ツリー
decision_tree:
  simple_classification:
    recommended: 'Zero-shotまたはFew-shot'
    reason: '単純な分類には上級技法は不要'

  math_reasoning:
    recommended: 'CoT + Self-Consistency'
    reason: '数学推論で最も安定した性能'

  multi_step_search:
    recommended: 'ReAct'
    reason: '外部情報が必要な場合にツール活用可能'

  creative_problem_solving:
    recommended: 'Tree-of-Thought'
    reason: '探索空間が広い創造的問題に適している'

  production_api:
    recommended: 'Few-shot + Structured Output'
    reason: '一貫性とパース可能性が重要'

# BAD: 指示文が多すぎるとモデルが混乱する
bad_prompt = """
You are an expert data scientist with 20 years of experience.
You must always be accurate and never hallucinate.
You should think carefully before answering.
Make sure your answer is complete and comprehensive.
Consider all edge cases and potential issues.
Be concise but thorough.
Use technical language but also be accessible.
Format your response nicely.
Include examples when appropriate.
Double-check your work before responding.

Question: What is the capital of France?
"""

# GOOD: 簡潔で具体的な指示
good_prompt = """
Answer the following geography question with just the city name.
Question: What is the capital of France?
"""

# BAD: 出力フォーマットが不明確
bad_format = "Analyze this data and give me insights."

# GOOD: 明確な出力フォーマット指定
good_format = """
Analyze the following sales data and provide:
1. Top 3 insights (one sentence each)
2. Trend direction: "increasing", "decreasing", or "stable"
3. Recommended actions (bulleted list, max 3 items)

Respond in JSON format with keys: insights, trend, actions.
"""

# BAD: 各リクエストで同じ長いシステムプロンプトを繰り返す
def bad_batch_processing(items):
    """毎回同一の長いシステムプロンプトを繰り返す"""
    results = []
    for item in items:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": VERY_LONG_SYSTEM_PROMPT},
                {"role": "user", "content": item},
            ]
        )
        results.append(response.choices[0].message.content)
    return results

# GOOD: バッチ処理で効率化
def good_batch_processing(items):
    """複数アイテムを一度に処理"""
    combined = "\n---\n".join([f"Item {i+1}: {item}" for i, item in enumerate(items)])
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "Process each item below and return results "
                    "in JSON array format."
                )
            },
            {"role": "user", "content": combined},
        ],
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)

import hashlib
from datetime import datetime

class PromptRegistry:
    """プロンプトバージョン管理システム"""

    def __init__(self):
        self.prompts = {}
        self.history = []

    def register(self, name: str, template: str, version: str = None) -> str:
        """プロンプト登録およびバージョン管理"""
        content_hash = hashlib.md5(template.encode()).hexdigest()[:8]
        version = version or f"v{len(self.history) + 1}_{content_hash}"

        entry = {
            "name": name,
            "version": version,
            "template": template,
            "hash": content_hash,
            "created_at": datetime.now().isoformat(),
        }

        self.prompts[name] = entry
        self.history.append(entry)
        return version

    def get(self, name: str) -> str:
        """現在アクティブなプロンプトを返す"""
        if name not in self.prompts:
            raise KeyError(f"Prompt '{name}' not registered")
        return self.prompts[name]["template"]

    def get_version(self, name: str) -> str:
        """現在のプロンプトバージョンを返す"""
        return self.prompts[name]["version"]

class CostOptimizer:
    """LLM APIコスト最適化"""

    # モデル別価格（1Mトークンあたり、2026年3月時点の概算値）
    PRICING = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
        "claude-3-5-haiku": {"input": 0.80, "output": 4.00},
    }

    @staticmethod
    def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
        """コスト推定"""
        pricing = CostOptimizer.PRICING.get(model, {})
        input_cost = (input_tokens / 1_000_000) * pricing.get("input", 0)
        output_cost = (output_tokens / 1_000_000) * pricing.get("output", 0)
        return input_cost + output_cost

    @staticmethod
    def select_model(task_complexity: str) -> str:
        """タスクの複雑度に応じたモデル選択"""
        model_map = {
            "simple": "gpt-4o-mini",       # 分類、抽出などの単純タスク
            "moderate": "gpt-4o-mini",      # CoTが必要な中程度のタスク
            "complex": "gpt-4o",            # 複雑な推論、コード生成
            "critical": "gpt-4o",           # 精度が最優先のタスク
        }
        return model_map.get(task_complexity, "gpt-4o-mini")

import hashlib
import json
from functools import lru_cache

class PromptCache:
    """プロンプト応答キャッシング"""

    def __init__(self, cache_backend="memory"):
        self.cache = {}
        self.hits = 0
        self.misses = 0

    def _make_key(self, model: str, messages: list, temperature: float) -> str:
        """キャッシュキー生成"""
        content = json.dumps({
            "model": model,
            "messages": messages,
            "temperature": temperature,
        }, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()

    def get(self, model: str, messages: list, temperature: float):
        """キャッシュから応答を照会"""
        if temperature > 0:
            # 非決定的応答はキャッシングしない
            return None

        key = self._make_key(model, messages, temperature)
        result = self.cache.get(key)
        if result:
            self.hits += 1
        else:
            self.misses += 1
        return result

    def set(self, model: str, messages: list, temperature: float, response: str):
        """キャッシュに応答を保存"""
        if temperature > 0:
            return
        key = self._make_key(model, messages, temperature)
        self.cache[key] = response

    def stats(self) -> dict:
        """キャッシュ統計"""
        total = self.hits + self.misses
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": self.hits / total if total > 0 else 0,
            "cache_size": len(self.cache),
        }

def sanitize_user_input(user_input: str) -> str:
    """ユーザー入力の浄化"""
    # 1. システムプロンプト迂回試行の検出
    injection_patterns = [
        "ignore previous instructions",
        "ignore all instructions",
        "disregard the above",
        "forget your instructions",
        "you are now",
        "new instruction:",
        "system prompt:",
    ]

    lower_input = user_input.lower()
    for pattern in injection_patterns:
        if pattern in lower_input:
            return "[BLOCKED: Potential prompt injection detected]"

    # 2. 入力長制限
    max_length = 4000
    if len(user_input) > max_length:
        user_input = user_input[:max_length] + "... [truncated]"

    return user_input

# 一般的な障害シナリオ
failure_scenarios:
  rate_limiting:
    symptom: '429 Too Many Requests'
    cause: 'API呼び出し上限超過'
    recovery:
      - '指数バックオフ（exponential backoff）を適用'
      - 'リクエストキュー実装でトラフィックを平滑化'
      - '複数APIキーのローテーション'

  hallucination:
    symptom: 'モデルが存在しない情報を生成'
    cause: '不十分なコンテキストまたは過度なtemperature'
    recovery:
      - 'temperatureを0に下げる'
      - 'RAGパイプラインで根拠資料を提供'
      - '出力検証レイヤーを追加'

  format_failure:
    symptom: 'JSONパース失敗'
    cause: 'モデルが要求されたフォーマットに従わない'
    recovery:
      - 'response_formatパラメータを使用'
      - 'Few-shot例でフォーマットを強制'
      - '失敗時にリトライ + より明確な指示を追加'

  context_overflow:
    symptom: 'コンテキストウィンドウ超過エラー'
    cause: '入力トークンがモデル上限を超過'
    recovery:
      - '入力テキストの要約またはチャンキング'
      - '不要なFew-shot例を削除'
      - 'より長いコンテキストのモデルに切り替え'

class PromptEvaluator:
    """プロンプトA/Bテスト評価器"""

    def __init__(self):
        self.results = []

    def evaluate(self, test_cases: list, prompt_a: str, prompt_b: str) -> dict:
        """2つのプロンプトの比較評価"""
        scores_a = []
        scores_b = []

        for case in test_cases:
            # プロンプトA実行
            result_a = self._run_prompt(prompt_a, case["input"])
            score_a = self._score(result_a, case["expected"])
            scores_a.append(score_a)

            # プロンプトB実行
            result_b = self._run_prompt(prompt_b, case["input"])
            score_b = self._score(result_b, case["expected"])
            scores_b.append(score_b)

        import numpy as np
        return {
            "prompt_a_avg": np.mean(scores_a),
            "prompt_b_avg": np.mean(scores_b),
            "prompt_a_std": np.std(scores_a),
            "prompt_b_std": np.std(scores_b),
            "winner": "A" if np.mean(scores_a) > np.mean(scores_b) else "B",
            "improvement": abs(np.mean(scores_a) - np.mean(scores_b)),
            "num_cases": len(test_cases),
        }

    def _run_prompt(self, prompt: str, input_text: str) -> str:
        """プロンプト実行"""
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": prompt},
                {"role": "user", "content": input_text},
            ],
            temperature=0,
        )
        return response.choices[0].message.content

    def _score(self, result: str, expected: str) -> float:
        """結果評価（0-1）"""
        # 単純な文字列類似度ベースのスコアリング
        result_lower = result.lower().strip()
        expected_lower = expected.lower().strip()

        if result_lower == expected_lower:
            return 1.0
        elif expected_lower in result_lower:
            return 0.8
        else:
            return 0.0