from openai import OpenAI
 
client = OpenAI()  # OPENAI_API_KEY 環境変数を使用
 
# Chat Completions API
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "簡潔に日本語で回答してください。"},
        {"role": "user", "content": "量子コンピュータの原理を説明してください。"}
    ],
    temperature=0.7,
    max_tokens=500,
)
 
print(response.choices[0].message.content)
print(f"トークン使用量: {response.usage}")

from openai import OpenAI
 
client = OpenAI()
 
# o1 / o3 は推論に時間をかける (Chain-of-Thought が内蔵)
response = client.chat.completions.create(
    model="o3-mini",
    messages=[{
        "role": "user",
        "content": """
        次の数学の問題を解いてください:
 
        100人が参加するトーナメント形式の試合で、
        1人の優勝者が決まるまでに必要な試合数は？
        その理由も説明してください。
        """
    }],
    # o1/o3 では temperature, top_p は設定不可
    # max_completion_tokens を使用 (max_tokens ではなく)
    max_completion_tokens=5000,
    reasoning_effort="medium",  # "low", "medium", "high"
)
 
print(response.choices[0].message.content)
# 推論トークン数も確認
print(f"推論トークン: {response.usage.completion_tokens_details.reasoning_tokens}")

from openai import OpenAI
from pydantic import BaseModel
 
client = OpenAI()
 
class TechArticle(BaseModel):
    title: str
    summary: str
    tags: list[str]
    difficulty: str  # "beginner", "intermediate", "advanced"
    estimated_reading_minutes: int
 
response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "技術記事を分析してJSON形式で出力。"},
        {"role": "user", "content": """
        以下の記事を分析してください:
 
        「Rustのライフタイムは、メモリ安全性を保証する仕組みです。
        コンパイラが参照の有効期間を追跡し、ダングリングポインタを
        防止します。'a のような記法で明示的に指定することもできます。」
        """}
    ],
    response_format=TechArticle,
)
 
article = response.choices[0].message.parsed
print(f"タイトル: {article.title}")
print(f"難易度: {article.difficulty}")
print(f"タグ: {article.tags}")

from openai import OpenAI
 
client = OpenAI()
 
# アシスタント作成
assistant = client.beta.assistants.create(
    name="データ分析アシスタント",
    instructions="あなたはデータ分析の専門家です。Pythonコードを使って分析を行います。",
    tools=[{"type": "code_interpreter"}],
    model="gpt-4o",
)
 
# スレッド作成
thread = client.beta.threads.create()
 
# メッセージ追加
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="1から100までの素数をリストアップし、その分布をグラフにしてください。"
)
 
# 実行
run = client.beta.threads.runs.create_and_poll(
    thread_id=thread.id,
    assistant_id=assistant.id,
)
 
# 結果取得
if run.status == "completed":
    messages = client.beta.threads.messages.list(thread_id=thread.id)
    for msg in messages.data:
        if msg.role == "assistant":
            for block in msg.content:
                if block.type == "text":
                    print(block.text.value)

from openai import OpenAI
import base64
 
client = OpenAI()
 
# 1. 画像生成
image_response = client.images.generate(
    model="dall-e-3",
    prompt="ミニマリストなスタイルのAIロボットが本を読んでいるイラスト",
    size="1024x1024",
    quality="hd",
    n=1,
)
image_url = image_response.data[0].url
print(f"画像URL: {image_url}")
 
# 2. GPT-4o で画像理解
analysis = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "この画像を分析してください。"},
            {"type": "image_url", "image_url": {"url": image_url}},
        ],
    }],
    max_tokens=500,
)
print(analysis.choices[0].message.content)

from openai import OpenAI
import json
from typing import Any
 
client = OpenAI()
 
# 複数のツール定義
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "指定された銘柄の現在の株価を取得します",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "株式のティッカーシンボル（例: AAPL, GOOGL, MSFT）"
                    },
                    "currency": {
                        "type": "string",
                        "enum": ["USD", "JPY", "EUR"],
                        "description": "表示通貨"
                    }
                },
                "required": ["symbol"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_portfolio_return",
            "description": "ポートフォリオの期待リターンを計算します",
            "parameters": {
                "type": "object",
                "properties": {
                    "holdings": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "symbol": {"type": "string"},
                                "weight": {"type": "number"},
                                "expected_return": {"type": "number"}
                            },
                            "required": ["symbol", "weight"]
                        },
                        "description": "保有銘柄のリスト"
                    },
                    "period": {
                        "type": "string",
                        "enum": ["1m", "3m", "6m", "1y", "5y"],
                        "description": "計算期間"
                    }
                },
                "required": ["holdings"]
            }
        }
    }
]
 
# 並列ツール呼び出し対応のエージェントループ
def agent_loop(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
 
    max_iterations = 10
    for _ in range(max_iterations):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",
        )
 
        message = response.choices[0].message
        messages.append(message)
 
        if not message.tool_calls:
            return message.content
 
        # 並列ツール呼び出しを全て処理
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
 
            # ツール実行（実際にはAPIコール等）
            result = dispatch_function(func_name, args)
 
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False),
            })
 
    return "処理が上限に達しました。"
 
def dispatch_function(name: str, args: dict) -> Any:
    """関数名に基づいてツールを実行"""
    functions = {
        "get_stock_price": lambda **a: {"price": 185.42, "currency": a.get("currency", "USD")},
        "calculate_portfolio_return": lambda **a: {"expected_return": 0.12, "risk": 0.08},
    }
    return functions.get(name, lambda **a: {"error": "Unknown function"})(**args)

from openai import OpenAI
 
client = OpenAI()
 
# 大きなシステムプロンプト（繰り返し使うもの）
LARGE_SYSTEM_PROMPT = """
あなたは金融アナリストです。以下のルールに従ってください：
1. 市場分析は客観的データに基づく
2. リスク要因を必ず明記する
3. 法的助言ではないことを明示する
... (大量の指示テキスト)
"""
 
# OpenAI のプロンプトキャッシュは自動的に適用される
# 同じプレフィックスが1024トークン以上ある場合にキャッシュが効く
# キャッシュヒット時は入力トークンのコストが50%割引
 
# キャッシュの効果を確認
for query in ["銘柄Aの分析", "銘柄Bの分析", "銘柄Cの分析"]:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": LARGE_SYSTEM_PROMPT},
            {"role": "user", "content": query}
        ],
    )
 
    # usage にキャッシュ情報が含まれる
    usage = response.usage
    cached = getattr(usage, 'prompt_tokens_details', None)
    if cached:
        print(f"キャッシュヒット: {cached.cached_tokens} tokens")
    print(f"合計入力: {usage.prompt_tokens} tokens")

from openai import OpenAI
import json
 
client = OpenAI()
 
# 1. バッチリクエストの準備（JSONL形式）
batch_requests = []
for i, query in enumerate(["分析1", "分析2", "分析3"]):
    batch_requests.append({
        "custom_id": f"request-{i}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [
                {"role": "system", "content": "簡潔に回答してください。"},
                {"role": "user", "content": query}
            ],
            "max_tokens": 200,
        }
    })
 
# JSONL ファイルに書き出し
with open("batch_input.jsonl", "w") as f:
    for req in batch_requests:
        f.write(json.dumps(req) + "\n")
 
# 2. ファイルアップロード
batch_file = client.files.create(
    file=open("batch_input.jsonl", "rb"),
    purpose="batch"
)
 
# 3. バッチジョブ作成
batch_job = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",  # 24時間以内に完了
    metadata={"description": "大量分析バッチ"}
)
 
print(f"バッチID: {batch_job.id}")
print(f"ステータス: {batch_job.status}")
 
# 4. ステータス確認
status = client.batches.retrieve(batch_job.id)
print(f"進捗: {status.request_counts}")
 
# 5. 結果取得（完了後）
if status.status == "completed":
    result_file = client.files.content(status.output_file_id)
    for line in result_file.text.strip().split("\n"):
        result = json.loads(line)
        print(f"{result['custom_id']}: {result['response']['body']['choices'][0]['message']['content'][:80]}...")
 
# コスト: 通常の50%割引で処理される

import asyncio
import websockets
import json
import base64
 
async def realtime_voice_assistant():
    """GPT-4o Realtime API を使った音声アシスタント"""
 
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1",
    }
 
    async with websockets.connect(url, extra_headers=headers) as ws:
        # セッション設定
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": """あなたは日本語の会話アシスタントです。
                                  自然な口語で回答してください。
                                  専門用語は分かりやすく説明してください。""",
                "voice": "nova",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "input_audio_transcription": {
                    "model": "whisper-1"
                },
                "turn_detection": {
                    "type": "server_vad",  # サーバー側の音声検出
                    "threshold": 0.5,
                    "prefix_padding_ms": 300,
                    "silence_duration_ms": 500,
                },
                "tools": [
                    {
                        "type": "function",
                        "name": "get_current_time",
                        "description": "現在の日時を取得します",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "timezone": {
                                    "type": "string",
                                    "description": "タイムゾーン（例: Asia/Tokyo）"
                                }
                            }
                        }
                    }
                ],
            },
        }))
 
        # メッセージ受信ループ
        async for message in ws:
            event = json.loads(message)
            event_type = event.get("type", "")
 
            if event_type == "session.created":
                print("セッション開始")
 
            elif event_type == "response.audio.delta":
                # 音声チャンクを受信 → スピーカーに送信
                audio_data = base64.b64decode(event["delta"])
                # play_audio(audio_data)  # 実際の再生処理
 
            elif event_type == "response.audio_transcript.delta":
                # テキスト変換結果
                print(event["delta"], end="", flush=True)
 
            elif event_type == "input_audio_buffer.speech_started":
                print("\n[音声検出開始]")
 
            elif event_type == "response.function_call_arguments.done":
                # ツール呼び出しの処理
                func_name = event.get("name")
                args = json.loads(event.get("arguments", "{}"))
                result = handle_function_call(func_name, args)
 
                await ws.send(json.dumps({
                    "type": "conversation.item.create",
                    "item": {
                        "type": "function_call_output",
                        "call_id": event["call_id"],
                        "output": json.dumps(result),
                    }
                }))
                await ws.send(json.dumps({"type": "response.create"}))

from openai import OpenAI
import json
 
client = OpenAI()
 
# 1. 学習データの準備
training_data = [
    {
        "messages": [
            {"role": "system", "content": "あなたは技術ドキュメントの要約専門家です。"},
            {"role": "user", "content": "以下のドキュメントを200字以内で要約してください:\n\n[ドキュメントテキスト1]"},
            {"role": "assistant", "content": "[理想的な要約1]"}
        ]
    },
    {
        "messages": [
            {"role": "system", "content": "あなたは技術ドキュメントの要約専門家です。"},
            {"role": "user", "content": "以下のドキュメントを200字以内で要約してください:\n\n[ドキュメントテキスト2]"},
            {"role": "assistant", "content": "[理想的な要約2]"}
        ]
    },
    # 最低50例、推奨200-500例
]
 
# 2. バリデーションデータ（任意だが推奨）
validation_data = training_data[:10]  # 10%をバリデーションに
 
# JSONL ファイル作成
for filename, data in [("train.jsonl", training_data), ("val.jsonl", validation_data)]:
    with open(filename, "w") as f:
        for item in data:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")
 
# 3. ファイルアップロード
train_file = client.files.create(file=open("train.jsonl", "rb"), purpose="fine-tune")
val_file = client.files.create(file=open("val.jsonl", "rb"), purpose="fine-tune")
 
# 4. ファインチューニングジョブ作成
job = client.fine_tuning.jobs.create(
    training_file=train_file.id,
    validation_file=val_file.id,
    model="gpt-4o-mini-2024-07-18",
    hyperparameters={
        "n_epochs": 3,
        "learning_rate_multiplier": "auto",
        "batch_size": "auto",
    },
    suffix="tech-summarizer",  # モデル名のサフィックス
)
 
print(f"ジョブID: {job.id}")
print(f"ステータス: {job.status}")
 
# 5. ステータス監視
import time
while True:
    job = client.fine_tuning.jobs.retrieve(job.id)
    print(f"ステータス: {job.status}")
 
    if job.status in ["succeeded", "failed", "cancelled"]:
        break
    time.sleep(60)
 
# 6. ファインチューニング済みモデルの使用
if job.status == "succeeded":
    ft_model = job.fine_tuned_model
    print(f"ファインチューニング済みモデル: {ft_model}")
 
    response = client.chat.completions.create(
        model=ft_model,
        messages=[
            {"role": "system", "content": "あなたは技術ドキュメントの要約専門家です。"},
            {"role": "user", "content": "以下のドキュメントを200字以内で要約してください:\n\n[新しいドキュメント]"}
        ]
    )
    print(response.choices[0].message.content)

from openai import OpenAI
import json
 
client = OpenAI()
 
def evaluate_fine_tuned_model(model_id: str, test_data: list) -> dict:
    """ファインチューニング済みモデルの品質を評価"""
    results = {
        "total": len(test_data),
        "correct": 0,
        "scores": [],
        "errors": [],
    }
 
    for item in test_data:
        # ファインチューニング済みモデルで推論
        response = client.chat.completions.create(
            model=model_id,
            messages=item["messages"][:-1],  # アシスタント応答を除く
            temperature=0,
        )
        prediction = response.choices[0].message.content
        expected = item["messages"][-1]["content"]
 
        # LLM-as-a-Judge で品質評価
        judge_response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": f"""2つのテキストを比較し、1-5のスコアで類似度を評価してください。
 
期待される出力: {expected}
実際の出力: {prediction}
 
JSON形式で回答: {{"score": <int>, "reason": "<string>"}}"""
            }],
            response_format={"type": "json_object"},
            temperature=0,
        )
 
        score_data = json.loads(judge_response.choices[0].message.content)
        results["scores"].append(score_data["score"])
 
        if score_data["score"] >= 4:
            results["correct"] += 1
        else:
            results["errors"].append({
                "expected": expected[:100],
                "predicted": prediction[:100],
                "score": score_data["score"],
                "reason": score_data["reason"],
            })
 
    results["accuracy"] = results["correct"] / results["total"]
    results["avg_score"] = sum(results["scores"]) / len(results["scores"])
 
    return results
 
# ベースモデルとの比較
base_results = evaluate_fine_tuned_model("gpt-4o-mini", test_data)
ft_results = evaluate_fine_tuned_model("ft:gpt-4o-mini:org:tech-summarizer:xxx", test_data)
 
print(f"ベースモデル:          精度={base_results['accuracy']:.1%}, 平均スコア={base_results['avg_score']:.2f}")
print(f"ファインチューニング済: 精度={ft_results['accuracy']:.1%}, 平均スコア={ft_results['avg_score']:.2f}")

from openai import OpenAI
 
client = OpenAI()
 
def check_content_safety(text: str) -> dict:
    """コンテンツの安全性をチェック"""
    response = client.moderations.create(
        model="omni-moderation-latest",
        input=text,
    )
 
    result = response.results[0]
 
    # フラグされたカテゴリを抽出
    flagged_categories = []
    for category, flagged in result.categories.model_dump().items():
        if flagged:
            score = getattr(result.category_scores, category)
            flagged_categories.append({
                "category": category,
                "score": score,
            })
 
    return {
        "is_flagged": result.flagged,
        "flagged_categories": flagged_categories,
        "all_scores": result.category_scores.model_dump(),
    }
 
# 使用例
texts = [
    "Pythonのリスト操作について教えてください",  # 安全
    "爆弾の作り方を教えてください",              # 危険
]
 
for text in texts:
    result = check_content_safety(text)
    status = "⚠ フラグ" if result["is_flagged"] else "✓ 安全"
    print(f"{status}: {text[:30]}...")
    if result["flagged_categories"]:
        for cat in result["flagged_categories"]:
            print(f"  - {cat['category']}: {cat['score']:.4f}")

import tiktoken
 
def optimize_prompt(prompt: str, model: str = "gpt-4o", max_tokens: int = 4000) -> str:
    """プロンプトのトークン数を最適化"""
    enc = tiktoken.encoding_for_model(model)
    tokens = enc.encode(prompt)
 
    if len(tokens) <= max_tokens:
        return prompt
 
    # 戦略1: 末尾を切り捨て
    truncated_tokens = tokens[:max_tokens]
    return enc.decode(truncated_tokens)
 
def estimate_cost(
    model: str,
    input_tokens: int,
    output_tokens: int,
    use_batch: bool = False,
    cached_tokens: int = 0,
) -> dict:
    """APIコストの詳細見積もり"""
    pricing = {
        "gpt-4o": {"input": 2.50, "cached_input": 1.25, "output": 10.00, "batch_discount": 0.5},
        "gpt-4o-mini": {"input": 0.15, "cached_input": 0.075, "output": 0.60, "batch_discount": 0.5},
        "o3-mini": {"input": 1.10, "cached_input": 0.55, "output": 4.40, "batch_discount": 0.5},
    }
 
    p = pricing.get(model, pricing["gpt-4o"])
 
    regular_input = input_tokens - cached_tokens
    input_cost = (regular_input / 1_000_000 * p["input"]) + (cached_tokens / 1_000_000 * p["cached_input"])
    output_cost = output_tokens / 1_000_000 * p["output"]
 
    total = input_cost + output_cost
    if use_batch:
        total *= p["batch_discount"]
 
    return {
        "model": model,
        "input_cost": f"${input_cost:.6f}",
        "output_cost": f"${output_cost:.6f}",
        "total": f"${total:.6f}",
        "batch_savings": f"${total:.6f}" if use_batch else "N/A",
    }
 
# 月間コスト見積もり
daily_requests = 10000
avg_input = 500
avg_output = 200
 
for model in ["gpt-4o", "gpt-4o-mini", "o3-mini"]:
    monthly = estimate_cost(model, avg_input * daily_requests * 30, avg_output * daily_requests * 30)
    print(f"{model:15s}: {monthly['total']}/月")

import asyncio
from openai import AsyncOpenAI, RateLimitError
import time
 
class RateLimitedClient:
    """レート制限を考慮した OpenAI クライアント"""
 
    def __init__(self, rpm_limit: int = 500, tpm_limit: int = 150000):
        self.client = AsyncOpenAI()
        self.rpm_semaphore = asyncio.Semaphore(rpm_limit)
        self.request_times = []
        self.rpm_limit = rpm_limit
        self.tpm_limit = tpm_limit
 
    async def create_completion(self, **kwargs):
        """レート制限を自動管理するリクエスト"""
        async with self.rpm_semaphore:
            # RPM制限チェック
            now = time.time()
            self.request_times = [t for t in self.request_times if now - t < 60]
 
            if len(self.request_times) >= self.rpm_limit:
                wait = 60 - (now - self.request_times[0])
                await asyncio.sleep(wait)
 
            self.request_times.append(time.time())
 
            # 指数バックオフリトライ
            for attempt in range(5):
                try:
                    return await self.client.chat.completions.create(**kwargs)
                except RateLimitError:
                    wait = 2 ** attempt
                    print(f"レート制限、{wait}秒待機...")
                    await asyncio.sleep(wait)
 
            raise Exception("リトライ上限超過")
 
# 大量リクエストの並列処理
async def process_batch(queries: list[str], model: str = "gpt-4o-mini"):
    rate_client = RateLimitedClient(rpm_limit=500)
 
    async def process_one(query: str):
        response = await rate_client.create_completion(
            model=model,
            messages=[{"role": "user", "content": query}],
            max_tokens=200,
        )
        return response.choices[0].message.content
 
    tasks = [process_one(q) for q in queries]
    return await asyncio.gather(*tasks)

from openai import OpenAI
import logging
 
# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("openai_debug")
 
def debug_api_call(prompt: str, model: str = "gpt-4o") -> dict:
    """デバッグ情報付きの API 呼び出し"""
    client = OpenAI()
 
    import time
    start = time.time()
 
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=500,
    )
 
    elapsed = time.time() - start
 
    debug_info = {
        "model": response.model,
        "latency_seconds": round(elapsed, 3),
        "input_tokens": response.usage.prompt_tokens,
        "output_tokens": response.usage.completion_tokens,
        "total_tokens": response.usage.total_tokens,
        "finish_reason": response.choices[0].finish_reason,
        "response_preview": response.choices[0].message.content[:200],
    }
 
    # 推論モデルの場合
    if hasattr(response.usage, 'completion_tokens_details') and response.usage.completion_tokens_details:
        details = response.usage.completion_tokens_details
        if hasattr(details, 'reasoning_tokens') and details.reasoning_tokens:
            debug_info["reasoning_tokens"] = details.reasoning_tokens
 
    # キャッシュ情報
    if hasattr(response.usage, 'prompt_tokens_details') and response.usage.prompt_tokens_details:
        details = response.usage.prompt_tokens_details
        if hasattr(details, 'cached_tokens') and details.cached_tokens:
            debug_info["cached_tokens"] = details.cached_tokens
 
    logger.info(f"API Call Debug: {debug_info}")
    return debug_info

class GPTModelRouter:
    """タスクの複雑さに応じてモデルを自動選択"""
 
    def __init__(self):
        self.client = OpenAI()
 
    def route(self, task: str, complexity: str = "auto") -> str:
        """タスクに最適なモデルを選択"""
        if complexity == "auto":
            complexity = self._estimate_complexity(task)
 
        routing_table = {
            "simple": "gpt-4o-mini",      # 分類、抽出、簡単なQA
            "moderate": "gpt-4o",          # 文章作成、コード生成、分析
            "complex": "o3-mini",          # 数学、論理推論、複雑なコード
            "expert": "o1",               # 研究レベルの推論
        }
 
        return routing_table.get(complexity, "gpt-4o")
 
    def _estimate_complexity(self, task: str) -> str:
        """タスクの複雑さを推定（軽量モデルで判定）"""
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"""以下のタスクの複雑さを判定してください。
タスク: {task}
回答は "simple", "moderate", "complex", "expert" のいずれか1つだけ:"""
            }],
            max_tokens=10,
            temperature=0,
        )
        return response.choices[0].message.content.strip().lower()
 
    async def execute(self, task: str) -> str:
        """ルーティング + 実行"""
        model = self.route(task)
 
        kwargs = {
            "model": model,
            "messages": [{"role": "user", "content": task}],
        }
 
        # 推論モデルの場合は max_completion_tokens を使用
        if model.startswith("o"):
            kwargs["max_completion_tokens"] = 5000
        else:
            kwargs["max_tokens"] = 2000
            kwargs["temperature"] = 0.7
 
        response = self.client.chat.completions.create(**kwargs)
        return response.choices[0].message.content

async def call_with_fallback(messages: list, **kwargs) -> str:
    """GPT モデルのフォールバックチェーン"""
 
    fallback_chain = [
        {"model": "gpt-4o", "timeout": 30},
        {"model": "gpt-4o-mini", "timeout": 15},
    ]
 
    errors = []
    for config in fallback_chain:
        try:
            response = await async_client.chat.completions.create(
                model=config["model"],
                messages=messages,
                timeout=config["timeout"],
                **kwargs,
            )
            return response.choices[0].message.content
        except Exception as e:
            errors.append(f"{config['model']}: {str(e)}")
            continue
 
    raise Exception(f"全モデルが失敗: {'; '.join(errors)}")

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()