LLMプロンプトエンジニアリング高度技法: Chain-of-Thought・ReAct・Tree of Thoughts 実践適用

from openai import OpenAI

client = OpenAI()

def few_shot_sentiment_analysis(text: str) -> str:
    """Few-shotプロンプティングを活用した感情分析"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "与えられたテキストの感情を分析し、ポジティブ/ネガティブ/ニュートラルに分類してください。"
            },
            # Few-shot例 1
            {"role": "user", "content": "この製品は本当に素晴らしいです！配送も早く品質も良いです。"},
            {"role": "assistant", "content": "感情: ポジティブ\n根拠: '素晴らしい'、'早く'、'良い'などポジティブな表現が多数含まれている"},
            # Few-shot例 2
            {"role": "user", "content": "配送が非常に遅く、製品に傷がありました。"},
            {"role": "assistant", "content": "感情: ネガティブ\n根拠: '非常に遅く'、'傷がありました'などネガティブな体験を述べている"},
            # Few-shot例 3
            {"role": "user", "content": "普通レベルの製品です。価格相応です。"},
            {"role": "assistant", "content": "感情: ニュートラル\n根拠: '普通レベル'、'価格相応'などバランスの取れた評価"},
            # 実際の分析対象
            {"role": "user", "content": text}
        ],
        temperature=0.0
    )
    return response.choices[0].message.content

def chain_of_thought_math(problem: str) -> str:
    """Chain-of-Thoughtを活用した数学問題の解決"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "あなたは数学の問題をステップバイステップで解くチューターです。"
                    "各ステップを明確に説明し、最終回答を提示してください。"
                )
            },
            # CoT例: 推論過程を含む
            {
                "role": "user",
                "content": "カフェに客が23人いました。午前に12人が来て、午後に8人が帰りました。現在カフェには何人いますか？"
            },
            {
                "role": "assistant",
                "content": (
                    "ステップバイステップで解いていきます。\n\n"
                    "ステップ1: 初期人数の把握\n"
                    "- 最初にカフェにいた客: 23人\n\n"
                    "ステップ2: 午前の変動を計算\n"
                    "- 12人が追加で来店: 23 + 12 = 35人\n\n"
                    "ステップ3: 午後の変動を計算\n"
                    "- 8人が退店: 35 - 8 = 27人\n\n"
                    "最終回答: 現在カフェには27人がいます。"
                )
            },
            {"role": "user", "content": problem}
        ],
        temperature=0.0
    )
    return response.choices[0].message.content

import json
import requests
from openai import OpenAI

client = OpenAI()

# ツール定義
def search_wikipedia(query: str) -> str:
    """Wikipedia検索ツール"""
    url = "https://ja.wikipedia.org/w/api.php"
    params = {
        "action": "query",
        "list": "search",
        "srsearch": query,
        "format": "json",
        "srlimit": 3
    }
    resp = requests.get(url, params=params)
    results = resp.json().get("query", {}).get("search", [])
    if not results:
        return "検索結果がありません。"
    return "\n".join([r["title"] + ": " + r["snippet"] for r in results])


def calculator(expression: str) -> str:
    """安全な数式計算機"""
    allowed_chars = set("0123456789+-*/.(). ")
    if all(c in allowed_chars for c in expression):
        try:
            result = eval(expression)
            return str(result)
        except Exception as e:
            return f"計算エラー: {e}"
    return "許可されていない文字が含まれています。"


TOOLS = {
    "search_wikipedia": search_wikipedia,
    "calculator": calculator,
}


def react_agent(question: str, max_steps: int = 5) -> str:
    """ReActパターンを実装したエージェント"""
    system_prompt = """あなたはReActエージェントです。以下の形式で応答してください:

Thought: 現在の状況を分析し、次に何をすべきか推論します。
Action: 使用するツールと入力をJSON形式で提供します。
Observation: ツール実行結果を確認します。
... (必要な分だけ繰り返し)
Final Answer: 最終回答を提示します。

使用可能なツール:
- search_wikipedia: Wikipedia検索（入力: 検索クエリ）
- calculator: 数式計算（入力: 数学表現式）"""

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": question}
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            temperature=0.0
        )
        assistant_msg = response.choices[0].message.content

        # Final Answerが含まれていれば終了
        if "Final Answer:" in assistant_msg:
            return assistant_msg.split("Final Answer:")[-1].strip()

        # Actionのパースと実行
        if "Action:" in assistant_msg:
            action_line = assistant_msg.split("Action:")[-1].strip()
            try:
                action_data = json.loads(action_line.split("\n")[0])
                tool_name = action_data.get("tool", "")
                tool_input = action_data.get("input", "")

                if tool_name in TOOLS:
                    observation = TOOLS[tool_name](tool_input)
                else:
                    observation = f"不明なツール: {tool_name}"
            except json.JSONDecodeError:
                observation = "Actionのパースに失敗"

            messages.append({"role": "assistant", "content": assistant_msg})
            messages.append({
                "role": "user",
                "content": f"Observation: {observation}"
            })

    return "最大ステップ数に到達し、回答を生成できませんでした。"

from collections import Counter
from openai import OpenAI

client = OpenAI()


def self_consistency_solve(
    question: str,
    num_samples: int = 5,
    temperature: float = 0.7
) -> dict:
    """Self-Consistencyサンプリングによる精度向上推論"""
    system_prompt = (
        "数学の問題をステップバイステップで解いてください。"
        "最後の行に '回答: [数値]' の形式で答えを記載してください。"
    )
    answers = []
    reasoning_paths = []

    for i in range(num_samples):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": question}
            ],
            temperature=temperature  # 多様な推論パスを生成
        )
        content = response.choices[0].message.content
        reasoning_paths.append(content)

        # 回答抽出
        for line in content.split("\n"):
            if line.strip().startswith("回答:"):
                answer = line.split("回答:")[-1].strip()
                answers.append(answer)
                break

    # 多数決投票
    if not answers:
        return {"final_answer": "回答抽出失敗", "confidence": 0.0}

    counter = Counter(answers)
    most_common = counter.most_common(1)[0]
    confidence = most_common[1] / len(answers)

    return {
        "final_answer": most_common[0],
        "confidence": confidence,
        "vote_distribution": dict(counter),
        "num_samples": num_samples,
        "reasoning_paths": reasoning_paths
    }


# 使用例
result = self_consistency_solve(
    "学校に生徒が450人います。男子が全体の60%で、"
    "男子の25%がメガネをかけています。メガネをかけている男子は何人ですか？",
    num_samples=7
)
print(f"最終回答: {result['final_answer']}")
print(f"信頼度: {result['confidence']:.1%}")
print(f"投票分布: {result['vote_distribution']}")

from openai import OpenAI
from typing import List

client = OpenAI()


def generate_thoughts(problem: str, current_state: str, n: int = 3) -> List[str]:
    """現在の状態から可能な次の思考を生成"""
    prompt = f"""問題: {problem}
現在までの推論: {current_state}

この状態から可能な次の推論ステップを{n}個提案してください。
各ステップは異なるアプローチでなければなりません。
形式:
1. [最初のアプローチ]
2. [2番目のアプローチ]
3. [3番目のアプローチ]"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.8
    )
    content = response.choices[0].message.content
    thoughts = []
    for line in content.split("\n"):
        line = line.strip()
        if line and line[0].isdigit() and "." in line:
            thoughts.append(line.split(".", 1)[1].strip())
    return thoughts[:n]


def evaluate_thought(problem: str, thought_path: str) -> float:
    """推論パスの有望性を0~1の範囲で評価"""
    prompt = f"""問題: {problem}
推論パス: {thought_path}

この推論パスが正解に到達する可能性を評価してください。
0.0（全く有望でない）~ 1.0（非常に有望）の範囲でスコアを提示してください。
スコアのみ数値で回答してください。"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1
    )
    try:
        score = float(response.choices[0].message.content.strip())
        return min(max(score, 0.0), 1.0)
    except ValueError:
        return 0.5


def tree_of_thoughts_bfs(
    problem: str,
    max_depth: int = 3,
    beam_width: int = 2
) -> str:
    """BFSベースのTree of Thoughts実装"""
    # 初期状態
    current_states = [("", 1.0)]  # (推論パス, スコア)

    for depth in range(max_depth):
        all_candidates = []

        for state, _ in current_states:
            # 各状態から候補思考を生成
            thoughts = generate_thoughts(problem, state)

            for thought in thoughts:
                new_path = f"{state}\nステップ {depth+1}: {thought}" if state else f"ステップ 1: {thought}"
                score = evaluate_thought(problem, new_path)
                all_candidates.append((new_path, score))

        # beam_width分の上位候補を選択
        all_candidates.sort(key=lambda x: x[1], reverse=True)
        current_states = all_candidates[:beam_width]

        print(f"深さ {depth+1}: 最高スコア = {current_states[0][1]:.2f}")

    # 最終回答生成
    best_path = current_states[0][0]
    final_prompt = f"""問題: {problem}
推論パス: {best_path}

上記の推論に基づいて最終回答を作成してください。"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": final_prompt}],
        temperature=0.0
    )
    return response.choices[0].message.content

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List, Optional

client = OpenAI()


class CodeReview(BaseModel):
    """コードレビュー結果スキーマ"""
    file_path: str = Field(description="レビュー対象ファイルパス")
    severity: str = Field(description="重要度: critical, warning, info")
    category: str = Field(description="カテゴリ: security, performance, style, logic")
    line_number: Optional[int] = Field(description="該当行番号")
    message: str = Field(description="レビューコメント")
    suggestion: str = Field(description="改善提案コード")


class CodeReviewResult(BaseModel):
    """全体コードレビュー結果"""
    reviews: List[CodeReview]
    summary: str = Field(description="レビュー要約")
    overall_score: int = Field(description="総合スコア (1-10)")


def structured_code_review(code: str, language: str = "python") -> CodeReviewResult:
    """構造化出力を活用したコードレビュー"""
    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    f"あなたはシニア{language}開発者です。"
                    "与えられたコードをレビューし、構造化された形式でフィードバックを提供してください。"
                )
            },
            {"role": "user", "content": f"以下のコードをレビューしてください:\n\n{code}"}
        ],
        response_format=CodeReviewResult
    )
    return response.choices[0].message.parsed

import json
from openai import OpenAI

client = OpenAI()

# ツールスキーマ定義
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "株式の現在価格を取得します",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "株式シンボル（例: AAPL, GOOGL）"
                    },
                    "currency": {
                        "type": "string",
                        "enum": ["USD", "KRW", "JPY"],
                        "description": "表示通貨"
                    }
                },
                "required": ["symbol"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_alert",
            "description": "株価アラートを設定します",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {"type": "string"},
                    "target_price": {"type": "number"},
                    "direction": {
                        "type": "string",
                        "enum": ["above", "below"]
                    }
                },
                "required": ["symbol", "target_price", "direction"]
            }
        }
    }
]


def function_calling_agent(user_message: str) -> str:
    """Function Callingを活用したエージェント"""
    messages = [
        {
            "role": "system",
            "content": "あなたは株式情報を提供する金融アシスタントです。"
        },
        {"role": "user", "content": user_message}
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )

    msg = response.choices[0].message

    # Function Call処理
    if msg.tool_calls:
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            # 実際の関数実行（ここではモックデータ）
            if func_name == "get_stock_price":
                result = json.dumps({
                    "symbol": func_args["symbol"],
                    "price": 185.50,
                    "change": "+2.3%"
                })
            elif func_name == "create_alert":
                result = json.dumps({
                    "status": "success",
                    "alert_id": "alert_12345"
                })
            else:
                result = json.dumps({"error": "Unknown function"})

            messages.append(msg)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

        # 最終応答生成
        final_response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages
        )
        return final_response.choices[0].message.content

    return msg.content

from dataclasses import dataclass, field
from typing import Dict, Any
from datetime import datetime


@dataclass
class PromptTemplate:
    """プロンプトテンプレート管理クラス"""
    name: str
    version: str
    template: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())

    def render(self, **kwargs) -> str:
        """変数を置換して最終プロンプトを生成"""
        result = self.template
        for key, value in kwargs.items():
            placeholder = f"__{key.upper()}__"
            result = result.replace(placeholder, str(value))
        return result


# プロンプトレジストリ
class PromptRegistry:
    def __init__(self):
        self._templates: Dict[str, PromptTemplate] = {}

    def register(self, template: PromptTemplate):
        key = f"{template.name}:{template.version}"
        self._templates[key] = template

    def get(self, name: str, version: str = "latest") -> PromptTemplate:
        if version == "latest":
            candidates = [
                v for k, v in self._templates.items()
                if k.startswith(f"{name}:")
            ]
            return max(candidates, key=lambda t: t.version)
        return self._templates[f"{name}:{version}"]


# 使用例
registry = PromptRegistry()
registry.register(PromptTemplate(
    name="code_review",
    version="2.1",
    template=(
        "あなたはシニア__LANGUAGE__開発者です。\n"
        "以下のコードをレビューしてください。\n\n"
        "レビュー基準:\n"
        "- セキュリティ脆弱性\n"
        "- パフォーマンス問題\n"
        "- コードスタイル\n\n"
        "コード:\n__CODE__"
    ),
    metadata={"model": "gpt-4o", "temperature": 0.0}
))