AI論文読解：Agentic Reasoning 実装ガイド 2026

"""
ReActパターンのコアループ実装。

LLMが自然言語で推論し、ツールを呼び出し、結果を観察する
サイクルをmax_stepsまで繰り返す。
"""
from dataclasses import dataclass, field
from typing import Callable, Optional
from enum import Enum
import json
import re


class StepType(Enum):
    THOUGHT = "thought"
    ACTION = "action"
    OBSERVATION = "observation"
    FINAL_ANSWER = "final_answer"


@dataclass
class AgentStep:
    step_type: StepType
    content: str
    tool_name: Optional[str] = None
    tool_input: Optional[dict] = None
    token_count: int = 0


@dataclass
class AgentTrace:
    """エージェント実行の全記録。"""
    question: str
    steps: list[AgentStep] = field(default_factory=list)
    final_answer: Optional[str] = None
    total_tokens: int = 0
    total_tool_calls: int = 0

    def add_step(self, step: AgentStep):
        self.steps.append(step)
        self.total_tokens += step.token_count
        if step.step_type == StepType.ACTION:
            self.total_tool_calls += 1


class ReActAgent:
    """ReActパターンエージェント。

    LLMとツール集合を受け取り、質問に対して反復的に
    推論-行動-観察ループを実行する。
    """

    SYSTEM_PROMPT = """You are a helpful assistant that solves problems step by step.
For each step, you MUST output exactly one of:
- Thought: <your reasoning about what to do next>
- Action: <tool_name>({"param": "value"})
- Final Answer: <your final response to the user>

Available tools:
{tool_descriptions}

Rules:
- Always think before acting.
- After observing a tool result, think about what it means before the next action.
- When you have enough information, provide Final Answer.
"""

    def __init__(
        self,
        llm: Callable,   # (messages: list[dict]) -> str
        tools: dict[str, Callable],
        tool_descriptions: dict[str, str],
        max_steps: int = 10,
        max_tokens_per_step: int = 1024,
    ):
        self.llm = llm
        self.tools = tools
        self.tool_descriptions = tool_descriptions
        self.max_steps = max_steps
        self.max_tokens_per_step = max_tokens_per_step

    def run(self, question: str) -> AgentTrace:
        trace = AgentTrace(question=question)

        # システムプロンプトにツール説明を挿入
        tool_desc_text = "\n".join(
            f"- {name}: {desc}"
            for name, desc in self.tool_descriptions.items()
        )
        system_msg = self.SYSTEM_PROMPT.format(tool_descriptions=tool_desc_text)

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

        for step_num in range(self.max_steps):
            # LLMに次のステップを生成するよう要求
            response = self.llm(messages)
            parsed = self._parse_response(response)

            if parsed.step_type == StepType.FINAL_ANSWER:
                trace.final_answer = parsed.content
                trace.add_step(parsed)
                break

            trace.add_step(parsed)
            messages.append({"role": "assistant", "content": response})

            if parsed.step_type == StepType.ACTION and parsed.tool_name:
                # ツール実行
                observation = self._execute_tool(
                    parsed.tool_name, parsed.tool_input or {}
                )
                obs_step = AgentStep(
                    step_type=StepType.OBSERVATION,
                    content=observation,
                )
                trace.add_step(obs_step)
                messages.append({
                    "role": "user",
                    "content": f"Observation: {observation}",
                })

        return trace

    def _parse_response(self, response: str) -> AgentStep:
        """LLM出力をパースしてThought/Action/Final Answerを区別する。"""
        response = response.strip()

        # Final Answerチェック
        if response.lower().startswith("final answer:"):
            return AgentStep(
                step_type=StepType.FINAL_ANSWER,
                content=response[len("final answer:"):].strip(),
            )

        # Actionパース: Action: tool_name({"key": "value"})
        action_match = re.match(
            r'Action:\s*(\w+)\((\{.*\})\)', response, re.DOTALL
        )
        if action_match:
            tool_name = action_match.group(1)
            try:
                tool_input = json.loads(action_match.group(2))
            except json.JSONDecodeError:
                tool_input = {}
            return AgentStep(
                step_type=StepType.ACTION,
                content=response,
                tool_name=tool_name,
                tool_input=tool_input,
            )

        # その他はThoughtとして処理
        return AgentStep(
            step_type=StepType.THOUGHT,
            content=response,
        )

    def _execute_tool(self, tool_name: str, tool_input: dict) -> str:
        """ツールを実行し、結果を文字列で返す。"""
        if tool_name not in self.tools:
            return f"Error: Unknown tool '{tool_name}'. Available: {list(self.tools.keys())}"
        try:
            result = self.tools[tool_name](**tool_input)
            return str(result)
        except Exception as e:
            return f"Error executing {tool_name}: {type(e).__name__}: {str(e)}"

"""
本番環境のエージェントツール定義。

各ツールは入力検証、timeout、コスト制限を内蔵し、
実行結果を構造化された形式で返す。
"""
from dataclasses import dataclass
from typing import Any, Optional
import httpx
import time


@dataclass
class ToolResult:
    success: bool
    data: Any
    error: Optional[str] = None
    execution_time_ms: float = 0.0
    cost_usd: float = 0.0


class WebSearchTool:
    """Web検索ツール。

    エージェントが最新情報を照会する際に使用する。
    レートリミットとコスト制限を内蔵する。
    """

    def __init__(
        self,
        api_key: str,
        max_results: int = 5,
        timeout_seconds: float = 10.0,
        max_calls_per_minute: int = 10,
    ):
        self.api_key = api_key
        self.max_results = max_results
        self.timeout_seconds = timeout_seconds
        self.max_calls_per_minute = max_calls_per_minute
        self._call_timestamps: list[float] = []

    def _check_rate_limit(self) -> bool:
        now = time.time()
        self._call_timestamps = [
            ts for ts in self._call_timestamps if now - ts < 60
        ]
        return len(self._call_timestamps) < self.max_calls_per_minute

    def __call__(self, query: str) -> ToolResult:
        if not query or len(query) > 500:
            return ToolResult(
                success=False,
                data=None,
                error="Query must be 1-500 characters",
            )

        if not self._check_rate_limit():
            return ToolResult(
                success=False,
                data=None,
                error=f"Rate limit exceeded: max {self.max_calls_per_minute}/min",
            )

        start = time.monotonic()
        try:
            # 実際の検索API呼び出し（例：Tavily、Serperなど）
            with httpx.Client(timeout=self.timeout_seconds) as client:
                response = client.get(
                    "https://api.search-provider.com/search",
                    params={"q": query, "max_results": self.max_results},
                    headers={"Authorization": f"Bearer {self.api_key}"},
                )
                response.raise_for_status()
                elapsed = (time.monotonic() - start) * 1000
                self._call_timestamps.append(time.time())

                return ToolResult(
                    success=True,
                    data=response.json(),
                    execution_time_ms=elapsed,
                    cost_usd=0.001,  # 1件あたりの推定コスト
                )
        except httpx.TimeoutException:
            return ToolResult(
                success=False,
                data=None,
                error=f"Search timed out after {self.timeout_seconds}s",
                execution_time_ms=(time.monotonic() - start) * 1000,
            )
        except httpx.HTTPStatusError as e:
            return ToolResult(
                success=False,
                data=None,
                error=f"HTTP {e.response.status_code}: {e.response.text[:200]}",
                execution_time_ms=(time.monotonic() - start) * 1000,
            )


class CodeExecutionTool:
    """コード実行ツール。

    エージェントがPythonコードを実行して計算やデータ処理を行う。
    セキュリティのため、許可されたモジュールのみimport可能で、
    実行時間とメモリを制限する。
    """

    ALLOWED_MODULES = {"math", "statistics", "json", "re", "datetime", "collections"}

    def __init__(self, timeout_seconds: float = 5.0):
        self.timeout_seconds = timeout_seconds

    def __call__(self, code: str) -> ToolResult:
        if not code or len(code) > 5000:
            return ToolResult(
                success=False,
                data=None,
                error="Code must be 1-5000 characters",
            )

        # import検査：許可されたモジュールのみ使用可能
        import_lines = [
            line.strip() for line in code.splitlines()
            if line.strip().startswith("import ") or line.strip().startswith("from ")
        ]
        for line in import_lines:
            module = line.split()[1].split(".")[0]
            if module not in self.ALLOWED_MODULES:
                return ToolResult(
                    success=False,
                    data=None,
                    error=f"Module '{module}' not allowed. Allowed: {self.ALLOWED_MODULES}",
                )

        start = time.monotonic()
        try:
            # 制限された環境で実行
            local_vars: dict = {}
            exec(code, {"__builtins__": {}}, local_vars)  # noqa: S102

            elapsed = (time.monotonic() - start) * 1000
            # 'result'変数があればそれを返す
            result = local_vars.get("result", str(local_vars))

            return ToolResult(
                success=True,
                data=result,
                execution_time_ms=elapsed,
            )
        except Exception as e:
            return ToolResult(
                success=False,
                data=None,
                error=f"{type(e).__name__}: {str(e)}",
                execution_time_ms=(time.monotonic() - start) * 1000,
            )

"""
エージェントのワーキングメモリ管理。

全会話履歴を保持しつつ、LLMに渡す際は
重要度ベースで要約・選択してコンテキストウィンドウに収める。
"""
from dataclasses import dataclass, field
from typing import Optional
import hashlib


@dataclass
class MemoryEntry:
    role: str
    content: str
    step_number: int
    importance: float = 0.5  # 0.0 ~ 1.0
    token_count: int = 0
    content_hash: str = ""

    def __post_init__(self):
        if not self.content_hash:
            self.content_hash = hashlib.md5(
                self.content.encode()
            ).hexdigest()[:8]


class SlidingWindowMemory:
    """スライディングウィンドウ + 重要度ベースのメモリ管理。

    最近のK件のメッセージは常に保持し、
    それ以前のメッセージはimportanceスコアに基づいて選別する。
    """

    def __init__(
        self,
        max_tokens: int = 8192,
        recent_window: int = 6,     # 最近N件は常に保持
        system_prompt_tokens: int = 500,
    ):
        self.max_tokens = max_tokens
        self.recent_window = recent_window
        self.system_prompt_tokens = system_prompt_tokens
        self.entries: list[MemoryEntry] = []

    def add(self, entry: MemoryEntry):
        # 重複防止
        if any(e.content_hash == entry.content_hash for e in self.entries):
            return
        self.entries.append(entry)

    def get_context(self, system_message: str) -> list[dict]:
        """LLMに渡すメッセージリストを構成する。

        1. システムプロンプトは常に含める
        2. 最近のrecent_window件は常に含める
        3. 残りはimportance順に予算内で含める
        """
        budget = self.max_tokens - self.system_prompt_tokens
        messages = [{"role": "system", "content": system_message}]

        if not self.entries:
            return messages

        # 最近のメッセージをまず確保
        recent = self.entries[-self.recent_window:]
        older = self.entries[:-self.recent_window] if len(self.entries) > self.recent_window else []

        recent_tokens = sum(e.token_count for e in recent)

        # 古いメッセージの中で重要なものを予算内で追加
        remaining_budget = budget - recent_tokens
        selected_older = sorted(older, key=lambda e: e.importance, reverse=True)

        included_older = []
        for entry in selected_older:
            if remaining_budget <= 0:
                break
            if entry.token_count <= remaining_budget:
                included_older.append(entry)
                remaining_budget -= entry.token_count

        # 時系列順にソートしてメッセージを構成
        included_older.sort(key=lambda e: e.step_number)
        all_entries = included_older + recent

        for entry in all_entries:
            messages.append({"role": entry.role, "content": entry.content})

        return messages

    def mark_important(self, step_number: int, importance: float = 1.0):
        """特定のステップの重要度を上げる。

        ツール実行結果、重要な発見などをマークする際に使用。
        """
        for entry in self.entries:
            if entry.step_number == step_number:
                entry.importance = importance
                break

"""
エージェント実行のコストとリソース使用を制御するガードレール。
"""
from dataclasses import dataclass
from typing import Optional
import time


@dataclass
class AgentBudget:
    max_llm_calls: int = 15
    max_tool_calls: int = 10
    max_total_tokens: int = 50_000
    max_cost_usd: float = 0.50
    max_wall_time_seconds: float = 120.0


@dataclass
class AgentUsage:
    llm_calls: int = 0
    tool_calls: int = 0
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    start_time: float = 0.0

    def elapsed_seconds(self) -> float:
        return time.time() - self.start_time if self.start_time else 0.0


class BudgetGuard:
    """エージェント実行予算監視者。

    各ステップの前にcheck()を呼び出して予算超過の有無を確認する。
    超過時はエージェントはそれまでの結果で早期終了すべきである。
    """

    def __init__(self, budget: AgentBudget):
        self.budget = budget
        self.usage = AgentUsage()

    def start(self):
        self.usage.start_time = time.time()

    def record_llm_call(self, tokens: int, cost_usd: float):
        self.usage.llm_calls += 1
        self.usage.total_tokens += tokens
        self.usage.total_cost_usd += cost_usd

    def record_tool_call(self, cost_usd: float = 0.0):
        self.usage.tool_calls += 1
        self.usage.total_cost_usd += cost_usd

    def check(self) -> Optional[str]:
        """予算超過時に理由を返す。正常ならNone。"""
        if self.usage.llm_calls >= self.budget.max_llm_calls:
            return f"LLM call limit reached: {self.usage.llm_calls}/{self.budget.max_llm_calls}"

        if self.usage.tool_calls >= self.budget.max_tool_calls:
            return f"Tool call limit reached: {self.usage.tool_calls}/{self.budget.max_tool_calls}"

        if self.usage.total_tokens >= self.budget.max_total_tokens:
            return f"Token limit reached: {self.usage.total_tokens}/{self.budget.max_total_tokens}"

        if self.usage.total_cost_usd >= self.budget.max_cost_usd:
            return f"Cost limit reached: ${self.usage.total_cost_usd:.3f}/${self.budget.max_cost_usd:.3f}"

        elapsed = self.usage.elapsed_seconds()
        if elapsed >= self.budget.max_wall_time_seconds:
            return f"Time limit reached: {elapsed:.1f}s/{self.budget.max_wall_time_seconds}s"

        return None

    def summary(self) -> dict:
        return {
            "llm_calls": f"{self.usage.llm_calls}/{self.budget.max_llm_calls}",
            "tool_calls": f"{self.usage.tool_calls}/{self.budget.max_tool_calls}",
            "tokens": f"{self.usage.total_tokens}/{self.budget.max_total_tokens}",
            "cost_usd": f"${self.usage.total_cost_usd:.4f}/${self.budget.max_cost_usd:.4f}",
            "elapsed_s": f"{self.usage.elapsed_seconds():.1f}/{self.budget.max_wall_time_seconds}",
        }

"""
エージェント評価フレームワーク。

正解率以外に効率性、ツール使用の適切性、推論品質を
総合的に測定する。
"""
from dataclasses import dataclass


@dataclass
class AgentEvalMetrics:
    # 正確度
    final_answer_correct: bool
    partial_credit: float          # 0.0 ~ 1.0（部分点）

    # 効率性
    total_steps: int
    total_tool_calls: int
    total_tokens: int
    total_cost_usd: float
    wall_time_seconds: float

    # ツール使用品質
    unnecessary_tool_calls: int    # 不必要なツール呼び出し数
    failed_tool_calls: int         # 失敗したツール呼び出し数
    tool_call_accuracy: float      # 正しいツールを正しい入力で呼び出した割合

    # 推論品質
    reasoning_coherence: float     # 推論の論理的一貫性（0.0 ~ 1.0）
    hallucination_count: int       # 根拠のない主張の数

    @property
    def efficiency_score(self) -> float:
        """効率性スコア：正解に到達するまでにどれだけ少ないリソースを使用したか。"""
        if not self.final_answer_correct:
            return 0.0
        # 少ないほど効率的 -> 逆数で変換
        step_penalty = min(self.total_steps / 10, 1.0)
        cost_penalty = min(self.total_cost_usd / 0.10, 1.0)
        return max(0.0, 1.0 - (step_penalty + cost_penalty) / 2)

    @property
    def overall_score(self) -> float:
        """総合スコア。"""
        weights = {
            "accuracy": 0.4,
            "efficiency": 0.2,
            "tool_quality": 0.2,
            "reasoning": 0.2,
        }
        accuracy = 1.0 if self.final_answer_correct else self.partial_credit
        return (
            weights["accuracy"] * accuracy
            + weights["efficiency"] * self.efficiency_score
            + weights["tool_quality"] * self.tool_call_accuracy
            + weights["reasoning"] * self.reasoning_coherence
        )