Gemini 3.x 時代のプロンプトエンジニアリング実践クラス — System Instructions・Few-shot・CoT・ReAct・自己評価ループを動くコードで比較実装

# 必要なライブラリのインストール
# pip install google-genai
 
import os
from google import genai
from google.genai import types
 
# APIクライアントの初期化
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# モデル設定（Gemini 3.1 Pro を使用）
MODEL = "gemini-3.1-pro"
 
def generate(prompt: str, system: str = "", thinking_budget: int = 0) -> str:
    """シンプルなテキスト生成ヘルパー"""
    config = types.GenerateContentConfig(
        system_instruction=system if system else None,
        thinking_config=types.ThinkingConfig(
            thinking_budget=thinking_budget
        ) if thinking_budget > 0 else None,
    )
    response = client.models.generate_content(
        model=MODEL,
        contents=prompt,
        config=config,
    )
    return response.text

# ❌ こういう書き方は避けたい
bad_system = "あなたは優秀なAIアシスタントです。ユーザーの質問に答えてください。"
 
response = generate(
    prompt="Python でファイルを読み込む方法を教えてください",
    system=bad_system
)
print(response)
# → 一般的で冗長な説明が返ってくる

# ✅ 役割・制約・出力フォーマット・禁止事項を明示する
good_system = """
# 役割
あなたはPython専門のテクニカルライターです。
個人開発者を対象に、実務で即使えるコードと解説を提供します。
 
# 出力の原則
- コードは必ず動作確認済みのものだけを提示する
- コメントは「なぜそうするか」を説明する（「何をするか」だけでは不十分）
- エラーハンドリングを必ず含める
- 初心者への配慮よりも実用性を優先する
 
# 禁止事項
- 「こんにちは！」などの挨拶文を入れない
- 「〜と思います」「〜かもしれません」などの曖昧表現を使わない
- 不必要な前置きや後付けの解説を加えない
 
# 出力形式
コードブロック → 1行説明 → 注意点（ある場合のみ）
"""
 
response = generate(
    prompt="Python でファイルを読み込む方法を教えてください",
    system=good_system
)
print(response)
# → 簡潔で実用的なコードと、なぜそう書くかの説明が返ってくる

SYSTEM_TEMPLATE = """
# ペルソナ
{persona}
 
# タスクの定義
{task_definition}
 
# 入力の形式
{input_format}
 
# 出力の制約
{output_constraints}
 
# 禁止パターン（これに該当する回答は不合格）
{forbidden_patterns}
 
# 出力例（この品質を目標とする）
{output_example}
"""
 
def build_system(
    persona: str,
    task_definition: str,
    input_format: str = "ユーザーの自然言語入力",
    output_constraints: str = "",
    forbidden_patterns: str = "",
    output_example: str = ""
) -> str:
    return SYSTEM_TEMPLATE.format(
        persona=persona,
        task_definition=task_definition,
        input_format=input_format,
        output_constraints=output_constraints,
        forbidden_patterns=forbidden_patterns,
        output_example=output_example,
    ).strip()

# ✅ 良い Few-shot 例の選び方
# 典型例だけでなく、境界値や例外的なケースを含める
 
few_shot_examples = [
    # 典型的なケース
    {
        "input": "東京の天気を教えて",
        "output": "申し訳ありませんが、リアルタイムの天気情報にはアクセスできません。weather.yahoo.co.jp などのサイトをご確認ください。"
    },
    # エッジケース：曖昧な質問
    {
        "input": "天気",
        "output": "どの地域の天気をお知りになりたいですか？地名を教えていただければ、適切なサービスをご案内できます。"
    },
    # エッジケース：複合的な要求
    {
        "input": "今日と明日の東京と大阪の天気を比べて",
        "output": "申し訳ありませんが、リアルタイムの気象データへのアクセス手段がないため、比較情報を提供することができません。tenki.jp などの天気比較サービスをご利用ください。"
    }
]

# Gemini 3.x では、例の中に思考過程を含めると効果的
few_shot_with_reasoning = """
例1:
入力: 「この文章の感情を分析してください：今日は最悪な一日だった」
思考: 「最悪」という強いネガティブ表現がある。しかし日記的な口調で、怒りよりも落胆の感情が主体か。
出力: {sentiment: "negative", intensity: 0.8, dominant_emotion: "disappointment", confidence: 0.85}
 
例2:
入力: 「この文章の感情を分析してください：まあまあかな」
思考: 「まあまあ」は肯定でも否定でもない中立的な日本語表現。感情の強度は低い。
出力: {sentiment: "neutral", intensity: 0.2, dominant_emotion: "indifference", confidence: 0.75}
 
分析対象: 「やっと終わった、でも思ったよりよかった」
"""
 
response = generate(few_shot_with_reasoning)
# → 思考過程を含む例を見せることで、モデルも中間推論を行い精度が向上する

def build_few_shot_prompt(
    task_description: str,
    examples: list[dict],
    actual_input: str,
    placement: str = "before"  # "before" | "after" | "mixed"
) -> str:
    """
    Few-shot プロンプトのビルダー
    placement="before": 例を全て最初に置く（一般的なパターン）
    placement="after": タスク説明の後に例を置く（複雑なタスクに有効）
    placement="mixed": 例を1つだけ示してから入力を置く（短いプロンプトに有効）
    """
    examples_text = "\n\n".join([
        f"入力: {ex['input']}\n出力: {ex['output']}"
        for ex in examples
    ])
    
    if placement == "before":
        return f"{examples_text}\n\n入力: {actual_input}\n出力:"
    elif placement == "after":
        return f"{task_description}\n\n{examples_text}\n\n入力: {actual_input}\n出力:"
    else:  # mixed
        first_example = examples[0]
        return (
            f"{task_description}\n\n"
            f"例: 入力「{first_example['input']}」→ 出力「{first_example['output']}」\n\n"
            f"入力: {actual_input}\n出力:"
        )

# CoT プロンプトの基本パターン
def generate_with_cot(
    question: str,
    thinking_budget: int = 8192,
    explicit_cot: bool = True
) -> dict:
    """
    Chain-of-Thought 生成
    thinking_budget: モデルの内部思考トークン数（高いほど深く考える）
    explicit_cot: プロンプト内にも「順を追って考えて」と書くかどうか
    """
    if explicit_cot:
        prompt = f"""
{question}
 
解答の前に、以下の順序で考えてください：
1. 問題の核心は何か？
2. 解決に必要な情報や前提は？
3. 複数の解法がある場合、それぞれのトレードオフは？
4. 最も適切な解法を選ぶ理由
5. 最終的な回答
 
各ステップを <thinking> タグで囲んでから、<answer> タグで最終回答を示してください。
"""
    else:
        prompt = question
    
    config = types.GenerateContentConfig(
        thinking_config=types.ThinkingConfig(
            thinking_budget=thinking_budget
        )
    )
    
    response = client.models.generate_content(
        model=MODEL,
        contents=prompt,
        config=config,
    )
    
    # 思考過程と回答を分離
    text = response.text
    thinking_part = ""
    answer_part = text
    
    if "<thinking>" in text and "</thinking>" in text:
        start = text.index("<thinking>") + len("<thinking>")
        end = text.index("</thinking>")
        thinking_part = text[start:end].strip()
        
    if "<answer>" in text and "</answer>" in text:
        start = text.index("<answer>") + len("<answer>")
        end = text.index("</answer>")
        answer_part = text[start:end].strip()
    
    return {
        "thinking": thinking_part,
        "answer": answer_part,
        "usage": response.usage_metadata.dict() if response.usage_metadata else {}
    }
 
# 使用例
result = generate_with_cot(
    "SaaSアプリのプライシングを月額1,000円にするか2,000円にするか迷っています。"
    "ターゲットは個人開発者で、競合は月額800〜1,500円が多いです。どちらが適切ですか？",
    thinking_budget=8192
)
 
print("思考過程:", result["thinking"][:200], "...")
print("回答:", result["answer"])
# → Thinking Budget があることで、市場分析・価格弾力性・収益計算まで踏み込んだ回答が返る

THINKING_BUDGET_GUIDE = {
    "単純な質問応答": 0,          # 思考なし（コスト最小）
    "構造化データ抽出": 1024,      # 軽量な推論
    "コード生成（100行以内）": 4096,
    "複雑なロジック判断": 8192,
    "多段階推論・戦略立案": 16384,
    "最高精度が必要な場合": 32768,  # コスト最大
}
 
# 実際の使い方：タスクタイプを事前に判定して予算を自動割り当て
def auto_thinking_budget(task_type: str) -> int:
    return THINKING_BUDGET_GUIDE.get(task_type, 4096)

import json
from typing import Callable
 
def react_agent(
    task: str,
    tools: dict[str, Callable],
    max_steps: int = 5,
    system: str = ""
) -> str:
    """
    ReAct パターンの基本実装
    tools: {ツール名: 実行関数} の辞書
    """
    # ツール定義をプロンプトに変換
    tools_description = "\n".join([
        f"- {name}: {func.__doc__ or '説明なし'}"
        for name, func in tools.items()
    ])
    
    react_system = f"""
{system}
 
# 使用可能なツール
{tools_description}
 
# ReAct プロセス
各ステップで以下の形式で思考と行動を示してください：
 
Thought: [現在の状況を分析し、次に何をすべきか考える]
Action: [使用するツール名]
Action Input: [ツールへの入力（JSON形式）]
Observation: [ツールの実行結果を受け取った後の観察]
...（繰り返し）
Final Answer: [最終的な回答]
"""
    
    history = [{"role": "user", "content": task}]
    
    for step in range(max_steps):
        response = client.models.generate_content(
            model=MODEL,
            contents=[{"role": m["role"], "parts": [{"text": m["content"]}]} for m in history],
            config=types.GenerateContentConfig(
                system_instruction=react_system,
                thinking_config=types.ThinkingConfig(thinking_budget=4096)
            )
        )
        
        text = response.text
        history.append({"role": "model", "content": text})
        
        # Final Answer が出たら終了
        if "Final Answer:" in text:
            final = text.split("Final Answer:")[-1].strip()
            return final
        
        # Action の実行
        if "Action:" in text and "Action Input:" in text:
            try:
                action_part = text.split("Action:")[1].split("\n")[0].strip()
                input_part = text.split("Action Input:")[1].split("\n")[0].strip()
                tool_input = json.loads(input_part)
                
                if action_part in tools:
                    result = tools[action_part](**tool_input)
                    observation = f"Observation: {result}"
                else:
                    observation = f"Observation: エラー — ツール '{action_part}' は存在しません"
                
                history.append({"role": "user", "content": observation})
            except (json.JSONDecodeError, KeyError, TypeError) as e:
                history.append({"role": "user", "content": f"Observation: エラー — {str(e)}"})
    
    return "最大ステップ数に達しました。最後の応答: " + history[-2]["content"]
 
# 実際の使用例
def search_web(query: str) -> str:
    """ウェブを検索して関連情報を返す"""
    # 実際の実装ではWebSearch APIを呼ぶ
    return f"'{query}' の検索結果: [サンプルデータ]"
 
def calculate(expression: str) -> str:
    """数式を計算する"""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"計算エラー: {str(e)}"
 
tools = {
    "search_web": search_web,
    "calculate": calculate,
}
 
result = react_agent(
    task="月額2,000円のSaaSで月収100万円を達成するには何人の有料ユーザーが必要か計算し、"
         "日本のSaaS市場の現状も踏まえて実現可能性を評価してください",
    tools=tools,
    max_steps=5
)
print(result)

def critic_refiner(
    original_prompt: str,
    draft_output: str,
    evaluation_criteria: list[str],
    max_iterations: int = 3
) -> dict:
    """
    Critic-Refiner パターン
    モデルに自分の出力を批評させ、基準を満たすまで改善を繰り返す
    """
    current_output = draft_output
    iterations = []
    
    for i in range(max_iterations):
        # Step 1: 批評（Critic）
        critic_prompt = f"""
以下の出力を評価してください。
 
# 元のタスク
{original_prompt}
 
# 現在の出力
{current_output}
 
# 評価基準
{chr(10).join([f"- {c}" for c in evaluation_criteria])}
 
各評価基準について、1〜10点でスコアリングし、具体的な改善点を指摘してください。
総合スコアが全基準で8点以上なら「APPROVED」、そうでなければ「NEEDS_IMPROVEMENT」と判定してください。
 
出力形式:
{{
  "scores": {{"基準名": スコア, ...}},
  "average_score": 平均スコア,
  "verdict": "APPROVED" または "NEEDS_IMPROVEMENT",
  "improvements": ["改善点1", "改善点2", ...]
}}
"""
        critic_config = types.GenerateContentConfig(
            response_mime_type="application/json",
            thinking_config=types.ThinkingConfig(thinking_budget=4096)
        )
        
        critic_response = client.models.generate_content(
            model=MODEL,
            contents=critic_prompt,
            config=critic_config,
        )
        
        try:
            evaluation = json.loads(critic_response.text)
        except json.JSONDecodeError:
            break
        
        iterations.append({
            "iteration": i + 1,
            "output": current_output,
            "evaluation": evaluation
        })
        
        # 承認された場合は終了
        if evaluation.get("verdict") == "APPROVED":
            break
        
        # Step 2: 改善（Refiner）
        improvements = evaluation.get("improvements", [])
        if not improvements:
            break
            
        refiner_prompt = f"""
以下の出力を改善してください。
 
# 元のタスク
{original_prompt}
 
# 現在の出力
{current_output}
 
# 必ず対応すべき改善点
{chr(10).join([f"- {imp}" for imp in improvements])}
 
改善した出力だけを返してください（説明は不要）。
"""
        
        refiner_response = client.models.generate_content(
            model=MODEL,
            contents=refiner_prompt,
            config=types.GenerateContentConfig(
                thinking_config=types.ThinkingConfig(thinking_budget=8192)
            )
        )
        
        current_output = refiner_response.text
    
    return {
        "final_output": current_output,
        "iterations": iterations,
        "total_iterations": len(iterations)
    }
 
# 使用例
draft = generate("Pythonで書いたコードのコードレビューを依頼されたとき、どう対応すべきか教えてください")
 
result = critic_refiner(
    original_prompt="Pythonで書いたコードのコードレビューを依頼されたとき、どう対応すべきか教えてください",
    draft_output=draft,
    evaluation_criteria=[
        "具体的なチェックリストが含まれているか",
        "コード例が1つ以上含まれているか",
        "初心者と上級者で読み分けられる構成か",
        "実際の業務で即使える内容か"
    ],
    max_iterations=2
)
 
print(f"改善ループ回数: {result['total_iterations']}")
print(f"最終スコア: {result['iterations'][-1]['evaluation'].get('average_score', 'N/A')}")
print(result['final_output'][:500])

# ❌ よくある間違い：System Instructions に具体的なデータを入れる
bad_system = """
あなたは田中太郎（株式会社例 CTO）のアシスタントです。
現在のプロジェクト: ECサイトリニューアル（期限: 2026年7月）
チームメンバー: 5名
使用技術: Next.js, PostgreSQL, Cloudflare Workers
"""
# → コンテキストが変わるたびに System Instructions を書き換える必要があり、
#   キャッシュ効率が下がる（→ コスト増）
 
# ✅ 正しい分担
good_system = "あなたはプロジェクト管理の専門アドバイザーです。提供されたコンテキストに基づいて回答してください。"
user_prompt = """
コンテキスト:
- 担当者: 田中太郎（CTO）
- プロジェクト: ECサイトリニューアル（期限: 2026年7月）
- チーム: 5名
- 技術: Next.js, PostgreSQL, Cloudflare Workers
 
質問: 来週のスプリントでの優先順位付けについてアドバイスをください
"""

# ❌ 全リクエストに高い Thinking Budget を設定 → コスト爆発
config_bad = types.GenerateContentConfig(
    thinking_config=types.ThinkingConfig(thinking_budget=32768)  # 毎回最大
)
 
# ✅ タスク複雑度に応じて動的に調整
def get_appropriate_budget(prompt: str) -> int:
    """プロンプトの複雑さを簡易判定して予算を決める"""
    # 複雑さの指標（簡易版）
    has_multiple_constraints = prompt.count("かつ") + prompt.count("および") > 2
    has_calculation = any(kw in prompt for kw in ["計算", "算出", "比較", "分析"])
    is_long = len(prompt) > 500
    
    complexity_score = sum([has_multiple_constraints, has_calculation, is_long])
    
    budget_map = {0: 0, 1: 2048, 2: 8192, 3: 16384}
    return budget_map.get(complexity_score, 4096)

# ❌ 矛盾した例を含めると、モデルが混乱する
contradictory_examples = [
    {"input": "こんにちは", "output": "はい、何でしょうか？"},      # 敬語
    {"input": "元気？", "output": "元気だよ！どうした？"},           # タメ口
    {"input": "お願いがあります", "output": "何でも言ってください"}, # 敬語
]
# → スタイルが統一されておらず、モデルがどちらのスタイルを採用すべきか判断できない
 
# ✅ スタイルを統一する
consistent_examples = [
    {"input": "こんにちは", "output": "はい、何でしょうか？"},
    {"input": "元気？", "output": "ありがとうございます、元気ですよ！"},
    {"input": "お願いがあります", "output": "何でも申し付けください。"},
]

import hashlib
from datetime import datetime
 
class PromptVersion:
    """プロンプトバージョン管理クラス"""
    
    def __init__(self, system: str, examples: list[dict] = None, template: str = ""):
        self.system = system
        self.examples = examples or []
        self.template = template
        self.created_at = datetime.utcnow().isoformat()
        self.version_hash = self._compute_hash()
    
    def _compute_hash(self) -> str:
        content = f"{self.system}|{self.examples}|{self.template}"
        return hashlib.sha256(content.encode()).hexdigest()[:8]
    
    def render(self, **kwargs) -> str:
        """テンプレート変数を展開してプロンプトを生成"""
        return self.template.format(**kwargs)
    
    def to_dict(self) -> dict:
        return {
            "version_hash": self.version_hash,
            "created_at": self.created_at,
            "system": self.system,
            "examples": self.examples,
            "template": self.template,
        }
 
# バージョン管理の使い方
v1 = PromptVersion(
    system="あなたはコードレビュアーです",
    template="以下のコードをレビューしてください：\n{code}",
)
 
v2 = PromptVersion(
    system="あなたはシニアエンジニアの視点でコードをレビューするアシスタントです。"
           "セキュリティ・パフォーマンス・可読性の3軸で評価してください",
    template="以下のPythonコードをレビューしてください：\n```python\n{code}\n```",
)
 
print(f"v1 hash: {v1.version_hash}")  # → a1b2c3d4
print(f"v2 hash: {v2.version_hash}")  # → e5f6g7h8
# ハッシュが変わると「プロンプトが変更された」と追跡できる

class PromptPipeline:
    """
    プロンプトエンジニアリングのフル実装パイプライン
    Few-shot + CoT + 自己評価を統合
    """
    
    def __init__(
        self,
        system: str,
        examples: list[dict],
        evaluation_criteria: list[str],
        thinking_budget: int = 4096
    ):
        self.system = system
        self.examples = examples
        self.evaluation_criteria = evaluation_criteria
        self.thinking_budget = thinking_budget
    
    def _build_prompt(self, user_input: str) -> str:
        examples_text = "\n\n".join([
            f"入力: {ex['input']}\n出力: {ex['output']}"
            for ex in self.examples
        ])
        return f"{examples_text}\n\n入力: {user_input}\n出力:"
    
    def run(self, user_input: str, auto_refine: bool = True) -> dict:
        """
        パイプライン実行
        auto_refine=True: 自己評価ループで品質が閾値に達するまで改善する
        """
        prompt = self._build_prompt(user_input)
        
        # Step 1: 初期生成
        config = types.GenerateContentConfig(
            system_instruction=self.system,
            thinking_config=types.ThinkingConfig(thinking_budget=self.thinking_budget)
        )
        response = client.models.generate_content(
            model=MODEL, contents=prompt, config=config
        )
        draft = response.text
        
        if not auto_refine:
            return {"output": draft, "refined": False}
        
        # Step 2: 自己評価ループ
        result = critic_refiner(
            original_prompt=user_input,
            draft_output=draft,
            evaluation_criteria=self.evaluation_criteria,
            max_iterations=2
        )
        
        return {
            "output": result["final_output"],
            "refined": True,
            "iterations": result["total_iterations"],
        }
 
# 使用例
pipeline = PromptPipeline(
    system="あなたはSaaS製品のプロダクトマネージャーアドバイザーです",
    examples=[
        {
            "input": "ユーザーが解約する理由を調べたい",
            "output": "Exit surveyを実装し、解約ボタンクリック時に3択（価格・機能不足・他サービスへ移行）+自由記述を求めてください。初月の解約率が最も高いため、オンボーディング品質の指標としても活用できます。"
        }
    ],
    evaluation_criteria=[
        "具体的なアクションが含まれているか",
        "KPIや数値への言及があるか",
        "リスクや注意点が含まれているか"
    ],
    thinking_budget=8192
)
 
result = pipeline.run("月次チャーンレートが5%を超えており、改善したい")
print(result["output"])