# エージェントの構成要素を詳細に定義する
from dataclasses import dataclass, field
from typing import Callable, Any, Protocol
from enum import Enum
 
class AgentCapability(Enum):
    """エージェントの能力分類"""
    REASONING = "reasoning"       # 推論能力
    PLANNING = "planning"         # 計画立案能力
    TOOL_USE = "tool_use"         # ツール使用能力
    MEMORY = "memory"             # 記憶能力
    LEARNING = "learning"         # 学習能力（メタ認知）
    COMMUNICATION = "communication"  # 通信能力（マルチエージェント用）
 
class ToolCategory(Enum):
    """ツールの分類"""
    INFORMATION = "information"   # 情報取得（検索、読み取り）
    COMPUTATION = "computation"   # 計算（数値計算、コード実行）
    COMMUNICATION = "communication"  # 通信（メール送信、API呼び出し）
    MANIPULATION = "manipulation"    # 操作（ファイル操作、DB操作）
 
@dataclass
class ToolDefinition:
    """ツールの定義"""
    name: str
    description: str
    category: ToolCategory
    parameters: dict
    function: Callable
    is_destructive: bool = False  # 破壊的操作かどうか
    requires_approval: bool = False  # 人間の承認が必要か
    rate_limit: int = 0  # 1分あたりの最大呼び出し回数（0=無制限）
 
@dataclass
class AgentProfile:
    """エージェントのプロファイル"""
    name: str
    role: str
    capabilities: list[AgentCapability]
    tools: list[ToolDefinition]
    model: str = "claude-sonnet-4-20250514"
    max_steps: int = 20
    temperature: float = 0.0
    system_prompt: str = ""
 
    def has_capability(self, cap: AgentCapability) -> bool:
        return cap in self.capabilities
 
    def get_tools_by_category(self, category: ToolCategory) -> list[ToolDefinition]:
        return [t for t in self.tools if t.category == category]
 
    def get_safe_tools(self) -> list[ToolDefinition]:
        """非破壊的なツールのみ取得"""
        return [t for t in self.tools if not t.is_destructive]

# 反応型エージェントの実装例
class ReactiveAgent:
    """入力に対して即座に反応する最もシンプルなエージェント"""
 
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
 
    def respond(self, user_input: str) -> str:
        """状態を持たず、入力に即応答"""
        # ツールが必要か判断
        tool_needed = self.llm.classify(user_input,
            categories=["direct_answer", "tool_needed"])
 
        if tool_needed == "direct_answer":
            return self.llm.generate(user_input)
 
        # 最も適切なツールを1つ選択して実行
        tool = self.llm.select_tool(user_input, self.tools)
        result = tool.execute(user_input)
        return self.llm.synthesize(user_input, result)
 
# 熟慮型エージェントの実装例
class DeliberativeAgent:
    """計画-実行-評価のサイクルを持つエージェント"""
 
    def __init__(self, llm, tools, max_iterations=10):
        self.llm = llm
        self.tools = tools
        self.max_iterations = max_iterations
        self.scratchpad = []  # 思考の記録
 
    def run(self, goal: str) -> str:
        # Phase 1: 計画
        plan = self._plan(goal)
 
        for i in range(self.max_iterations):
            # Phase 2: 次のアクションを決定
            action = self._decide_next_action(plan, self.scratchpad)
 
            if action.is_final:
                return action.content
 
            # Phase 3: アクション実行
            result = self._execute(action)
            self.scratchpad.append({
                "thought": action.thought,
                "action": action.name,
                "result": result
            })
 
            # Phase 4: 計画の見直し
            if self._needs_replan(plan, self.scratchpad):
                plan = self._replan(goal, self.scratchpad)
 
        return self._summarize_progress()
 
# ハイブリッド型エージェントの実装例
class HybridAgent:
    """反応型と熟慮型を組み合わせたエージェント"""
 
    def __init__(self, reactive: ReactiveAgent, deliberative: DeliberativeAgent):
        self.reactive = reactive
        self.deliberative = deliberative
 
    def handle(self, user_input: str) -> str:
        # 入力の複雑さを評価
        complexity = self._assess_complexity(user_input)
 
        if complexity == "simple":
            # 単純な質問 → 反応型で即応答
            return self.reactive.respond(user_input)
        else:
            # 複雑なタスク → 熟慮型で計画的に実行
            return self.deliberative.run(user_input)
 
    def _assess_complexity(self, user_input: str) -> str:
        """入力の複雑さを判定"""
        # ステップ数、ツール必要性、曖昧さ等で判断
        indicators = {
            "multiple_steps": any(w in user_input for w in ["そして", "その後", "次に"]),
            "tool_needed": any(w in user_input for w in ["検索", "計算", "作成", "実行"]),
            "ambiguous": len(user_input) > 200 or "?" in user_input
        }
        complex_count = sum(indicators.values())
        return "complex" if complex_count >= 2 else "simple"

# エージェントの基本ループ（概念コード）
class SimpleAgent:
    def __init__(self, llm, tools, memory):
        self.llm = llm
        self.tools = tools
        self.memory = memory
 
    def run(self, goal: str) -> str:
        """目標を受け取り、完了まで自律実行する"""
        self.memory.add("goal", goal)
 
        while not self.is_done():
            # 1. Perceive: 現在の状態を観察
            context = self.memory.get_context()
 
            # 2. Think: 次のアクションを推論
            action = self.llm.decide(context, self.tools)
 
            # 3. Act: ツールを実行
            if action.type == "tool_call":
                result = self.tools.execute(action)
                self.memory.add("observation", result)
            elif action.type == "final_answer":
                return action.content
 
        return self.memory.get_summary()

# ReAct (Reasoning + Acting) パターン
REACT_PROMPT = """
以下の形式で思考と行動を繰り返してください：
 
Thought: 現状の分析と次にすべきことの推論
Action: 使用するツール名[引数]
Observation: ツールの実行結果
... (繰り返し)
Thought: 最終的な結論
Final Answer: ユーザーへの回答
"""
 
class ReActAgent:
    def step(self, messages):
        response = self.llm.generate(
            system=REACT_PROMPT,
            messages=messages
        )
 
        if "Final Answer:" in response:
            return {"type": "answer", "content": response}
 
        # Action を解析してツール実行
        tool_name, args = self.parse_action(response)
        result = self.toolstool_name
 
        return {"type": "observation", "content": result}

# OpenAI / Anthropic スタイルの Function Calling
import anthropic
 
client = anthropic.Anthropic()
 
tools = [
    {
        "name": "web_search",
        "description": "Webを検索して情報を取得する",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "検索クエリ"}
            },
            "required": ["query"]
        }
    }
]
 
def agent_loop(user_message):
    messages = [{"role": "user", "content": user_message}]
 
    while True:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )
 
        if response.stop_reason == "end_turn":
            return response.content[0].text
 
        # ツール呼び出しを処理
        for block in response.content:
            if block.type == "tool_use":
                result = execute_tool(block.name, block.input)
                messages.append({"role": "assistant", "content": response.content})
                messages.append({
                    "role": "user",
                    "content": [{"type": "tool_result",
                                 "tool_use_id": block.id,
                                 "content": result}]
                })

# Plan-and-Execute パターンの完全な実装
class PlanAndExecuteAgent:
    """先に全体計画を立ててから順に実行するパターン"""
 
    def __init__(self, planner_llm, executor_llm, tools):
        self.planner = planner_llm
        self.executor = executor_llm
        self.tools = tools
 
    def run(self, goal: str) -> str:
        # Phase 1: 計画立案
        plan = self._create_plan(goal)
 
        # Phase 2: 順次実行
        results = []
        for i, step in enumerate(plan):
            print(f"Step {i+1}/{len(plan)}: {step}")
            result = self._execute_step(step, results)
            results.append({"step": step, "result": result})
 
            # 計画の見直しが必要か確認
            if self._needs_replan(goal, plan, results):
                remaining_plan = self._replan(goal, results, plan[i+1:])
                plan = plan[:i+1] + remaining_plan
 
        # Phase 3: 結果の統合
        return self._synthesize(goal, results)
 
    def _create_plan(self, goal: str) -> list[str]:
        """目標を具体的なステップに分解"""
        response = self.planner.generate(f"""
目標: {goal}
 
利用可能なツール:
{self._format_tools()}
 
この目標を達成するための具体的なステップを列挙してください。
各ステップは1つのツール呼び出しまたは1つの思考で完結すること。
 
出力形式:
1. [ステップの説明]
2. [ステップの説明]
...
""")
        return self._parse_steps(response)
 
    def _execute_step(self, step: str, previous_results: list) -> str:
        """個別のステップを実行"""
        context = "\n".join([
            f"- {r['step']}: {r['result'][:200]}"
            for r in previous_results[-3:]  # 直近3件の結果のみ
        ])
 
        response = self.executor.generate(f"""
実行するステップ: {step}
 
これまでの結果:
{context}
 
利用可能なツール:
{self._format_tools()}
 
このステップを実行してください。
""")
        return response

# エージェントの主要コンポーネントの実装例
from dataclasses import dataclass, field
from typing import Callable
 
@dataclass
class Tool:
    """エージェントが使用するツール"""
    name: str
    description: str
    function: Callable
    parameters: dict
 
@dataclass
class Memory:
    """エージェントの記憶"""
    short_term: list = field(default_factory=list)   # 現在タスクの文脈
    long_term: dict = field(default_factory=dict)     # 永続的な知識
    working: dict = field(default_factory=dict)       # 一時的な作業領域
 
@dataclass
class AgentConfig:
    """エージェントの設定"""
    model: str = "claude-sonnet-4-20250514"
    max_steps: int = 20           # 最大ステップ数
    temperature: float = 0.0      # 決定論的な出力
    max_tokens: int = 4096
    tools: list = field(default_factory=list)
    system_prompt: str = ""

# エージェント向けシステムプロンプトの設計パターン
class AgentPromptBuilder:
    """エージェント用のシステムプロンプトを構造的に構築する"""
 
    @staticmethod
    def build(role: str, tools: list, constraints: list,
              examples: list = None) -> str:
        prompt_parts = []
 
        # 1. 役割定義
        prompt_parts.append(f"## 役割\nあなたは{role}です。")
 
        # 2. 利用可能なツール
        tool_descriptions = "\n".join([
            f"- **{t.name}**: {t.description}" for t in tools
        ])
        prompt_parts.append(f"## 利用可能なツール\n{tool_descriptions}")
 
        # 3. 行動規範
        constraint_list = "\n".join([f"- {c}" for c in constraints])
        prompt_parts.append(f"## 行動規範\n{constraint_list}")
 
        # 4. 思考プロセス
        prompt_parts.append("""## 思考プロセス
1. まず目標を明確化する
2. 必要な情報を特定する
3. 最も効率的なツールを選択する
4. 実行結果を評価する
5. 目標が達成されたか判断する
6. 未達成なら次のアクションを計画する""")
 
        # 5. 出力形式
        prompt_parts.append("""## 出力形式
- 作業の意図を簡潔に説明してからツールを使用する
- 結果を分析して次のステップを判断する
- 最終回答は構造化して提供する""")
 
        # 6. 例示（Few-shot）
        if examples:
            example_text = "\n\n".join([
                f"### 例 {i+1}\n入力: {e['input']}\n出力: {e['output']}"
                for i, e in enumerate(examples)
            ])
            prompt_parts.append(f"## 例\n{example_text}")
 
        return "\n\n".join(prompt_parts)
 
# 使用例
system_prompt = AgentPromptBuilder.build(
    role="シニアソフトウェアエンジニア",
    tools=[read_file_tool, write_file_tool, run_tests_tool],
    constraints=[
        "コードを変更する前に必ず既存のコードを読んで理解する",
        "テストを書いてから実装する（TDD）",
        "破壊的な変更の前にユーザーの確認を求める",
        "エラーが発生したら原因を特定してから修正する"
    ]
)

# ツール設計のガイドライン
class ToolDesignGuidelines:
    """
    良いツール設計の原則:
 
    1. 単一責任の原則: 1ツール = 1機能
    2. 明確な命名: 動詞_名詞 形式（search_web, read_file）
    3. 詳細な説明: 何をするか + いつ使うか + 入出力
    4. 適切な粒度: 粗すぎず細かすぎず
    5. エラーハンドリング: 失敗時の情報を充実させる
    6. 冪等性: 可能な限り同じ入力→同じ出力
    """
 
    @staticmethod
    def validate_tool_definition(tool: dict) -> list[str]:
        """ツール定義の品質をチェック"""
        issues = []
 
        # 名前のチェック
        if not tool.get("name"):
            issues.append("名前が未定義")
        elif "_" not in tool["name"]:
            issues.append("名前は動詞_名詞形式が推奨（例: search_web）")
 
        # 説明のチェック
        desc = tool.get("description", "")
        if len(desc) < 20:
            issues.append("説明が短すぎます（最低20文字）")
        if "使用" not in desc and "use" not in desc.lower():
            issues.append("いつ使うかの説明が推奨されます")
 
        # パラメータのチェック
        schema = tool.get("input_schema", {})
        props = schema.get("properties", {})
        for param_name, param_def in props.items():
            if "description" not in param_def:
                issues.append(f"パラメータ '{param_name}' に説明がありません")
 
        return issues

# イベント駆動型のエージェント実装
from typing import Protocol
from dataclasses import dataclass
import asyncio
 
class EventHandler(Protocol):
    async def handle(self, event: dict) -> dict: ...
 
@dataclass
class AgentEvent:
    type: str  # "user_input", "tool_result", "error", "timeout"
    data: dict
    timestamp: float
 
class EventDrivenAgent:
    """イベント駆動型のエージェント"""
 
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.event_queue = asyncio.Queue()
        self.handlers: dict[str, EventHandler] = {}
        self.state = {"status": "idle", "history": []}
 
    def register_handler(self, event_type: str, handler: EventHandler):
        self.handlers[event_type] = handler
 
    async def run(self):
        """イベントループ"""
        while True:
            event = await self.event_queue.get()
            handler = self.handlers.get(event.type)
 
            if handler:
                try:
                    result = await handler.handle(event.data)
                    self.state["history"].append({
                        "event": event.type,
                        "result": result
                    })
                except Exception as e:
                    await self.event_queue.put(AgentEvent(
                        type="error",
                        data={"error": str(e), "original_event": event},
                        timestamp=time.time()
                    ))
 
    async def submit(self, event_type: str, data: dict):
        """イベントを投入"""
        await self.event_queue.put(AgentEvent(
            type=event_type,
            data=data,
            timestamp=time.time()
        ))

# ストリーミング対応のエージェント
import anthropic
from typing import AsyncGenerator
 
class StreamingAgent:
    """リアルタイムにトークンをストリーミングするエージェント"""
 
    def __init__(self):
        self.client = anthropic.Anthropic()
 
    async def run_streaming(self, user_message: str,
                            tools: list) -> AsyncGenerator[dict, None]:
        """ストリーミングでエージェントの出力を返す"""
        messages = [{"role": "user", "content": user_message}]
 
        while True:
            with self.client.messages.stream(
                model="claude-sonnet-4-20250514",
                max_tokens=4096,
                tools=tools,
                messages=messages
            ) as stream:
                current_text = ""
                for event in stream:
                    if event.type == "content_block_delta":
                        if hasattr(event.delta, "text"):
                            current_text += event.delta.text
                            yield {
                                "type": "text_delta",
                                "text": event.delta.text
                            }
 
                response = stream.get_final_message()
 
            if response.stop_reason == "end_turn":
                yield {"type": "complete", "text": current_text}
                return
 
            # ツール呼び出し処理
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    yield {
                        "type": "tool_call",
                        "tool": block.name,
                        "input": block.input
                    }
                    result = self._execute_tool(block.name, block.input)
                    yield {
                        "type": "tool_result",
                        "tool": block.name,
                        "result": str(result)[:200]
                    }
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result)
                    })
 
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

# 効率的なコンテキスト管理
class ContextManager:
    """エージェントのコンテキストウィンドウを効率的に管理する"""
 
    def __init__(self, max_tokens: int = 100000):
        self.max_tokens = max_tokens
        self.messages: list[dict] = []
        self.system_prompt: str = ""
        self.pinned_context: list[str] = []  # 常に含めるコンテキスト
 
    def estimate_tokens(self, text: str) -> int:
        """トークン数を概算（1文字≒1.5トークンで日本語を概算）"""
        return int(len(text) * 1.5)
 
    def get_current_tokens(self) -> int:
        total = self.estimate_tokens(self.system_prompt)
        total += sum(self.estimate_tokens(str(m)) for m in self.messages)
        total += sum(self.estimate_tokens(c) for c in self.pinned_context)
        return total
 
    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        self._trim_if_needed()
 
    def _trim_if_needed(self):
        """コンテキストが上限を超えたら古いメッセージを要約"""
        while self.get_current_tokens() > self.max_tokens * 0.8:
            if len(self.messages) <= 4:
                break  # 最低限のメッセージは保持
 
            # 古いメッセージを要約して圧縮
            old_messages = self.messages[:len(self.messages)//2]
            summary = self._summarize(old_messages)
 
            self.messages = [
                {"role": "system", "content": f"これまでの会話の要約: {summary}"}
            ] + self.messages[len(self.messages)//2:]
 
    def _summarize(self, messages: list) -> str:
        """メッセージ群を要約"""
        content = "\n".join([f"{m['role']}: {str(m['content'])[:200]}" for m in messages])
        return f"[要約] {content[:500]}"
 
    def pin_context(self, context: str):
        """常に含めるコンテキストを追加"""
        self.pinned_context.append(context)
 
    def get_messages_for_api(self) -> list:
        """API呼び出し用のメッセージリストを取得"""
        result = []
        if self.pinned_context:
            result.append({
                "role": "user",
                "content": "参考情報:\n" + "\n".join(self.pinned_context)
            })
        result.extend(self.messages)
        return result

# NG: 停止条件がないエージェント
class BadAgent:
    def run(self, goal):
        while True:  # 永遠に終わらない可能性
            action = self.think(goal)
            self.execute(action)
 
# OK: 最大ステップ数とタイムアウトを設定
class GoodAgent:
    def run(self, goal, max_steps=20, timeout=300):
        for step in range(max_steps):
            if time.time() - start > timeout:
                return self.summarize_progress()
            action = self.think(goal)
            if action.is_final:
                return action.result
            self.execute(action)
        return self.summarize_progress()  # 途中経過を返す

# NG: 曖昧なツール説明
bad_tool = {
    "name": "search",
    "description": "検索する"  # 何を？ どう？
}
 
# OK: 具体的で明確なツール説明
good_tool = {
    "name": "web_search",
    "description": "指定されたクエリでGoogle検索を実行し、上位10件の結果（タイトル、URL、スニペット）を返す。事実確認や最新情報の取得に使用する。",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "検索クエリ（日本語 or 英語）"
            },
            "num_results": {
                "type": "integer",
                "description": "取得件数（デフォルト: 10）",
                "default": 10
            }
        },
        "required": ["query"]
    }
}

# NG: ツール結果を全て保持してコンテキストが爆発
class ContextExplosionAgent:
    def run(self, goal):
        messages = [{"role": "user", "content": goal}]
        for _ in range(100):
            response = llm.generate(messages=messages)
            tool_result = execute(response)
            # 巨大な結果がそのまま蓄積
            messages.append({"role": "assistant", "content": response})
            messages.append({"role": "user", "content": tool_result})
        # → 数万トークンに膨張、コスト爆発
 
# OK: コンテキストマネージャーで管理
class ManagedAgent:
    def __init__(self):
        self.context_manager = ContextManager(max_tokens=50000)
 
    def run(self, goal):
        self.context_manager.add_message("user", goal)
        for _ in range(20):
            messages = self.context_manager.get_messages_for_api()
            response = llm.generate(messages=messages)
 
            # 結果を要約してから追加
            result = execute(response)
            summarized = summarize_if_large(result, max_chars=2000)
            self.context_manager.add_message("assistant", str(response))
            self.context_manager.add_message("user", summarized)

# NG: エラー時にクラッシュ
class FragileAgent:
    def run(self, goal):
        result = self.tools"search"  # ツールがない → KeyError
        return result  # ネットワークエラー → 未処理例外
 
# OK: 多層的なエラーハンドリング
class RobustAgent:
    def run(self, goal):
        try:
            for step in range(self.max_steps):
                action = self.decide_action(goal)
 
                if action.tool not in self.tools:
                    self.report_error(f"ツール '{action.tool}' は利用できません")
                    continue
 
                try:
                    result = self.toolsaction.tool
                except TimeoutError:
                    result = "タイムアウト。再試行してください。"
                except Exception as e:
                    result = f"エラー: {type(e).__name__}: {e}"
 
                self.memory.add(action, result)
 
        except Exception as e:
            return f"エージェントでエラーが発生しました: {e}\n部分的な結果: {self.get_partial_results()}"

# パフォーマンス最適化の実装例
import asyncio
import hashlib
from functools import lru_cache
 
class OptimizedAgent:
    def __init__(self):
        self.cache = {}
        self.metrics = {"total_time": 0, "cache_hits": 0, "api_calls": 0}
 
    async def run_optimized(self, goal: str) -> str:
        """最適化されたエージェントループ"""
        start = time.time()
        messages = [{"role": "user", "content": goal}]
 
        for step in range(self.max_steps):
            # キャッシュチェック
            cache_key = self._make_cache_key(messages)
            if cache_key in self.cache:
                self.metrics["cache_hits"] += 1
                response = self.cache[cache_key]
            else:
                self.metrics["api_calls"] += 1
                response = await self._call_llm_async(messages)
                self.cache[cache_key] = response
 
            if response.stop_reason == "end_turn":
                self.metrics["total_time"] = time.time() - start
                return self._extract_text(response)
 
            # 並列ツール実行
            tool_calls = [b for b in response.content if b.type == "tool_use"]
            if len(tool_calls) > 1:
                results = await self._execute_tools_parallel(tool_calls)
            else:
                results = [await self._execute_tool_async(tool_calls[0])]
 
            # メッセージ更新
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": results})
 
        self.metrics["total_time"] = time.time() - start
        return "最大ステップ数到達"
 
    async def _execute_tools_parallel(self, tool_calls):
        """複数のツール呼び出しを並列に実行"""
        tasks = [
            self._execute_tool_async(tc) for tc in tool_calls
        ]
        return await asyncio.gather(*tasks)
 
    def _make_cache_key(self, messages: list) -> str:
        content = str(messages[-3:])  # 直近3メッセージでキーを生成
        return hashlib.md5(content.encode()).hexdigest()
 
    def get_performance_report(self) -> str:
        return (
            f"総実行時間: {self.metrics['total_time']:.1f}秒\n"
            f"API呼び出し: {self.metrics['api_calls']}回\n"
            f"キャッシュヒット: {self.metrics['cache_hits']}回\n"
            f"キャッシュ率: {self.metrics['cache_hits']/(self.metrics['api_calls']+self.metrics['cache_hits'])*100:.0f}%"
        )

# コスト最適化のためのモデルルーティング
class CostOptimizedAgent:
    """タスクの複雑さに応じてモデルを使い分ける"""
 
    MODELS = {
        "fast": "claude-haiku-4-20250514",     # 分類、ルーティング用
        "balanced": "claude-sonnet-4-20250514", # 一般的なタスク
        "powerful": "claude-opus-4-20250514",   # 複雑な推論
    }
 
    def select_model(self, task_type: str, complexity: str) -> str:
        """タスクとその複雑さに応じたモデル選択"""
        model_map = {
            ("classification", "any"): "fast",
            ("routing", "any"): "fast",
            ("generation", "low"): "balanced",
            ("generation", "high"): "powerful",
            ("reasoning", "low"): "balanced",
            ("reasoning", "high"): "powerful",
            ("coding", "any"): "balanced",
        }
 
        # 複雑さに関わらないタスク
        key = (task_type, "any")
        if key in model_map:
            return self.MODELS[model_map[key]]
 
        # 複雑さを考慮するタスク
        key = (task_type, complexity)
        tier = model_map.get(key, "balanced")
        return self.MODELS[tier]

# エージェントのデバッグツール
class AgentDebugger:
    """エージェントの実行をトレース・分析するデバッグツール"""
 
    def __init__(self):
        self.trace = []
 
    def log_step(self, step_num: int, thought: str, action: str,
                 result: str, tokens_used: int):
        self.trace.append({
            "step": step_num,
            "thought": thought[:200],
            "action": action,
            "result": result[:200],
            "tokens": tokens_used,
            "timestamp": time.time()
        })
 
    def print_trace(self):
        """実行トレースを可視化"""
        print("=" * 60)
        print("エージェント実行トレース")
        print("=" * 60)
 
        total_tokens = 0
        for entry in self.trace:
            print(f"\n--- Step {entry['step']} ---")
            print(f"  思考: {entry['thought']}")
            print(f"  行動: {entry['action']}")
            print(f"  結果: {entry['result']}")
            print(f"  トークン: {entry['tokens']:,}")
            total_tokens += entry['tokens']
 
        print(f"\n{'=' * 60}")
        print(f"合計ステップ: {len(self.trace)}")
        print(f"合計トークン: {total_tokens:,}")
        print(f"推定コスト: ${total_tokens * 3 / 1_000_000:.4f}")
 
    def detect_loops(self) -> list[str]:
        """ループパターンを検出"""
        issues = []
        actions = [t["action"] for t in self.trace]
 
        # 同じアクションの連続
        for i in range(len(actions) - 2):
            if actions[i] == actions[i+1] == actions[i+2]:
                issues.append(
                    f"Step {i}-{i+2}: '{actions[i]}' が3回連続呼び出し"
                )
 
        # 合計呼び出し回数
        from collections import Counter
        counts = Counter(actions)
        for action, count in counts.most_common(3):
            if count > 5:
                issues.append(
                    f"'{action}' が{count}回呼び出し（過剰の可能性）"
                )
 
        return issues

# 演習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()