Tree-of-Thought / ReAct — 探索的推論とツール連携による高度な問題解決

思考A: 13 - 9 = 4  → 残り: 4, 4, 10
思考B: 9 - 4 = 5   → 残り: 5, 10, 13
思考C: 13 - 4 = 9  → 残り: 9, 9, 10
思考D: 10 - 4 = 6  → 残り: 6, 9, 13
思考E: 4 + 9 = 13  → 残り: 10, 13, 13

思考A-1: 10 - 4 = 6  → 残り: 4, 6
思考A-2: 4 + 4 = 8   → 残り: 8, 10
思考A-3: 4 × 4 = 16  → 残り: 10, 16

Thought 1: コロラド造山運動の東側の地域を調べる必要がある。
Action 1: Search["コロラド造山運動"]
Observation 1: コロラド造山運動は約3億年前に北アメリカで起きた
              造山運動で、コロラド付近に山脈を形成した。

Thought 2: コロラド造山運動の東側の地域が何かまだわからない。
           もっと具体的に調べよう。
Action 2: Search["コロラド造山運動 東側 平原"]
Observation 2: コロラド造山運動の東側にはグレートプレーンズが
              広がっている。

Thought 3: グレートプレーンズについて確認しよう。
Action 3: Search["グレートプレーンズ"]
Observation 3: グレートプレーンズは北アメリカ中央部の広大な平原で、
              ロッキー山脈の東側からミシシッピ川付近まで広がる。

Thought 4: 情報が揃った。コロラド造山運動の東側に広がる平原は
           グレートプレーンズである。
Final Answer: グレートプレーンズ

Thought: Apple RemoteはApple TVのリモコンで、
Siri RemoteもApple TVのリモコンです。
両方で操作できるのは...Front Rowです。
Answer: Front Row

Thought 1: Apple Remoteで操作できるプログラムを調べよう。
Action 1: Search["Apple Remote"]
Observation 1: Apple RemoteはFront RowやApple TVの操作に使用。

Thought 2: Siri Remoteで操作できるプログラムも調べよう。
Action 2: Search["Siri Remote"]
Observation 2: Siri RemoteはApple TVの操作用リモコン。

Thought 3: 両方で操作できるプログラムはApple TVのアプリケーション
           に含まれるはず。Front Rowは古いMac用なので、
           両方で使えるのはApple TV関連のプログラムだ。
Action 3: Search["Apple Remote Apple TV program"]
Observation 3: Apple RemoteとSiri Remoteの両方で
              操作できるのはApple TVのソフトウェアである。

Thought 4: 確認が取れた。
Final Answer: Apple TV

Thought 1: 半径5cmの球の体積を求める。公式は (4/3)πr³。
Action 1: Calculate["(4/3) * 3.14159265 * 5**3"]
Observation 1: 523.5987755

Thought 2: 同じ半径の円柱（高さ10cm）の体積も求めよう。
           公式は πr²h。
Action 2: Calculate["3.14159265 * 5**2 * 10"]
Observation 2: 785.398163

Thought 3: 球の体積 ≈ 523.6 cm³、
           円柱の体積 ≈ 785.4 cm³。
           球の体積は円柱の約66.7%である。
Final Answer: 球（半径5cm）の体積は約523.6cm³、
             同じ半径で高さ10cmの円柱の体積は約785.4cm³。
             球は円柱の体積の約2/3（66.7%）です。

import itertools
from typing import List, Tuple, Optional
from dataclasses import dataclass, field


@dataclass
class ThoughtNode:
    """ToTの思考ノード"""
    numbers: List[float]       # 現在の数字リスト
    expression: str            # ここまでの式の説明
    depth: int                 # 木の深さ
    parent: Optional['ThoughtNode'] = None
    children: List['ThoughtNode'] = field(default_factory=list)
    score: float = 0.0        # 状態の評価スコア


class TwentyFourToTSolver:
    """Tree-of-Thoughtによる24ゲームソルバー"""

    OPERATORS = [
        ('+', lambda a, b: a + b),
        ('-', lambda a, b: a - b),
        ('*', lambda a, b: a * b),
        ('/', lambda a, b: a / b if b != 0 else None),
    ]
    TARGET = 24
    TOLERANCE = 1e-9

    def __init__(self, beam_width: int = 5):
        """
        Args:
            beam_width: BFSのビーム幅（各深さで保持する状態数）
        """
        self.beam_width = beam_width
        self.nodes_explored = 0
        self.solutions_found = []

    def thought_generator(self, node: ThoughtNode) -> List[ThoughtNode]:
        """Thought Generator: 2つの数を選んで演算し、子ノードを生成"""
        children = []
        nums = node.numbers

        # 2つの数のすべての組み合わせ
        for i in range(len(nums)):
            for j in range(len(nums)):
                if i == j:
                    continue
                a, b = nums[i], nums[j]

                for op_symbol, op_func in self.OPERATORS:
                    result = op_func(a, b)
                    if result is None:
                        continue  # ゼロ除算をスキップ

                    # 残りの数字 + 演算結果で新しい状態を作る
                    remaining = [nums[k] for k in range(len(nums))
                                 if k != i and k != j]
                    new_numbers = remaining + [result]

                    # 式の説明を組み立て
                    expr = f"{a} {op_symbol} {b} = {result}"
                    full_expr = (f"{node.expression} -> {expr}"
                                 if node.expression else expr)

                    child = ThoughtNode(
                        numbers=new_numbers,
                        expression=full_expr,
                        depth=node.depth + 1,
                        parent=node,
                    )
                    children.append(child)
                    node.children.append(child)

        return children

    def state_evaluator(self, node: ThoughtNode) -> float:
        """State Evaluator: 状態の有望さをスコアリング"""
        nums = node.numbers

        # 数字が1つなら、24に近いほど高スコア
        if len(nums) == 1:
            if abs(nums[0] - self.TARGET) < self.TOLERANCE:
                return 1.0  # sure: 正解
            else:
                return 0.0  # impossible: 不正解

        # 数字が2つなら、四則演算で24が作れるか直接チェック
        if len(nums) == 2:
            a, b = nums
            for _, op_func in self.OPERATORS:
                r1 = op_func(a, b)
                r2 = op_func(b, a)
                if r1 is not None and abs(r1 - self.TARGET) < self.TOLERANCE:
                    return 0.9
                if r2 is not None and abs(r2 - self.TARGET) < self.TOLERANCE:
                    return 0.9
            return 0.1  # 2数で24が作れない

        # 数字が3つ以上なら、ヒューリスティックで評価
        # 24に近い数が含まれているほど有望とする
        min_diff = min(abs(n - self.TARGET) for n in nums)
        score = max(0.2, 1.0 - min_diff / self.TARGET)

        # 24の約数・倍数が含まれていればボーナス
        useful_numbers = {1, 2, 3, 4, 6, 8, 12, 24}
        for n in nums:
            if abs(n) in useful_numbers or (n != 0 and 24 % abs(n) == 0
                                             if abs(n) == int(abs(n))
                                             else False):
                score = min(score + 0.1, 0.8)

        return score

    def solve_bfs(self, numbers: List[int]) -> List[str]:
        """BFSで24ゲームを解く"""
        root = ThoughtNode(
            numbers=[float(n) for n in numbers],
            expression="",
            depth=0,
        )

        # 初期状態
        current_level = [root]
        self.solutions_found = []
        self.nodes_explored = 0

        # 最大深さ3（4数→3数→2数→1数）
        for depth in range(3):
            next_level = []

            for node in current_level:
                children = self.thought_generator(node)
                self.nodes_explored += len(children)

                for child in children:
                    child.score = self.state_evaluator(child)

                    # 正解が見つかった場合
                    if len(child.numbers) == 1 and child.score == 1.0:
                        self.solutions_found.append(child.expression)

                    next_level.append(child)

            # ビーム幅でフィルタリング（上位b個を保持）
            next_level.sort(key=lambda n: n.score, reverse=True)
            current_level = next_level[:self.beam_width]

            # 不可能な状態を枝刈り
            current_level = [n for n in current_level if n.score > 0.05]

        return self.solutions_found


# 実行例
solver = TwentyFourToTSolver(beam_width=50)

# テストケース
test_cases = [
    [4, 9, 10, 13],
    [1, 2, 3, 4],
    [8, 3, 8, 3],
    [5, 5, 5, 1],
    [2, 3, 4, 6],
]

print("=" * 60)
print("Tree-of-Thought 24ゲームソルバー (BFS)")
print("=" * 60)

for nums in test_cases:
    solver_instance = TwentyFourToTSolver(beam_width=50)
    solutions = solver_instance.solve_bfs(nums)

    # 重複を除去
    unique_solutions = list(set(solutions))

    print(f"\n入力: {nums}")
    print(f"  探索ノード数: {solver_instance.nodes_explored}")
    if unique_solutions:
        print(f"  解の数: {len(unique_solutions)}")
        for i, sol in enumerate(unique_solutions[:3]):
            print(f"  解{i+1}: {sol}")
    else:
        print("  解が見つかりませんでした")

============================================================
Tree-of-Thought 24ゲームソルバー (BFS)
============================================================

入力: [4, 9, 10, 13]
  探索ノード数: 348
  解の数: 2
  解1: 13.0 - 9.0 = 4.0 -> 10.0 - 4.0 = 6.0 -> 4.0 * 6.0 = 24.0
  解2: 10.0 - 4.0 = 6.0 -> 13.0 - 9.0 = 4.0 -> 4.0 * 6.0 = 24.0

入力: [1, 2, 3, 4]
  探索ノード数: 348
  解の数: 3
  解1: 1.0 * 2.0 = 2.0 -> 2.0 * 3.0 = 6.0 -> 6.0 * 4.0 = 24.0
  ...

import matplotlib.pyplot as plt

# ビーム幅と解発見率の関係を調べる
beam_widths = [1, 2, 5, 10, 20, 50, 100]
hard_cases = [
    [4, 9, 10, 13],
    [8, 3, 8, 3],
    [5, 5, 5, 1],
    [1, 5, 5, 5],
    [3, 7, 3, 7],
    [2, 3, 4, 6],
    [1, 2, 3, 4],
    [6, 6, 6, 6],
    [1, 3, 5, 7],
    [2, 5, 7, 8],
]

success_rates = []
avg_nodes = []

for bw in beam_widths:
    successes = 0
    total_nodes = 0
    for nums in hard_cases:
        s = TwentyFourToTSolver(beam_width=bw)
        sols = s.solve_bfs(nums)
        if sols:
            successes += 1
        total_nodes += s.nodes_explored
    success_rates.append(successes / len(hard_cases) * 100)
    avg_nodes.append(total_nodes / len(hard_cases))

# 可視化
fig, ax1 = plt.subplots(figsize=(9, 5))

color1 = '#00bcd4'
color2 = '#ff9800'

ax1.set_xlabel('Beam Width', fontsize=12)
ax1.set_ylabel('Success Rate (%)', color=color1, fontsize=12)
ax1.plot(beam_widths, success_rates, 'o-', color=color1, linewidth=2,
         markersize=8, label='Success Rate')
ax1.tick_params(axis='y', labelcolor=color1)
ax1.set_ylim(0, 105)

ax2 = ax1.twinx()
ax2.set_ylabel('Avg Nodes Explored', color=color2, fontsize=12)
ax2.plot(beam_widths, avg_nodes, 's--', color=color2, linewidth=2,
         markersize=8, label='Nodes Explored')
ax2.tick_params(axis='y', labelcolor=color2)

fig.suptitle('ToT BFS: Beam Width vs Performance', fontsize=14)
fig.tight_layout()
plt.savefig('tot_beam_width.png', dpi=150, bbox_inches='tight')
plt.show()

import re
import math
from typing import Dict, Callable, Any, List, Optional
from dataclasses import dataclass


@dataclass
class ReActStep:
    """ReActの1ステップ"""
    thought: str
    action: Optional[str] = None
    action_input: Optional[str] = None
    observation: Optional[str] = None


class ToolRegistry:
    """外部ツールの管理"""

    def __init__(self):
        self.tools: Dict[str, Callable] = {}
        self.descriptions: Dict[str, str] = {}

    def register(self, name: str, func: Callable, description: str):
        """ツールを登録"""
        self.tools[name] = func
        self.descriptions[name] = description

    def execute(self, name: str, input_str: str) -> str:
        """ツールを実行"""
        if name not in self.tools:
            return f"エラー: ツール '{name}' は存在しません。"
        try:
            return str(self.tools[name](input_str))
        except Exception as e:
            return f"エラー: {e}"

    def get_descriptions(self) -> str:
        """全ツールの説明を取得"""
        lines = []
        for name, desc in self.descriptions.items():
            lines.append(f"  - {name}: {desc}")
        return "\n".join(lines)


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

    def __init__(self, tools: ToolRegistry, max_steps: int = 8):
        """
        Args:
            tools: ツールレジストリ
            max_steps: 最大ステップ数
        """
        self.tools = tools
        self.max_steps = max_steps
        self.history: List[ReActStep] = []

    def _format_history(self) -> str:
        """履歴をフォーマット"""
        lines = []
        for i, step in enumerate(self.history, 1):
            lines.append(f"Thought {i}: {step.thought}")
            if step.action:
                lines.append(f"Action {i}: {step.action}[{step.action_input}]")
            if step.observation:
                lines.append(f"Observation {i}: {step.observation}")
        return "\n".join(lines)

    def run(self, question: str, thought_action_pairs: List[dict]) -> str:
        """
        ReActループを実行（事前定義されたステップで）

        Args:
            question: 質問文
            thought_action_pairs: [{thought, action, action_input}, ...]
        Returns:
            最終回答
        """
        self.history = []

        print(f"質問: {question}")
        print("-" * 50)

        for step_data in thought_action_pairs:
            thought = step_data["thought"]

            # Final Answerの場合
            if "final_answer" in step_data:
                print(f"\nThought: {thought}")
                print(f"Final Answer: {step_data['final_answer']}")
                return step_data["final_answer"]

            action = step_data["action"]
            action_input = step_data["action_input"]

            # ツール実行
            observation = self.tools.execute(action, action_input)

            step = ReActStep(
                thought=thought,
                action=action,
                action_input=action_input,
                observation=observation,
            )
            self.history.append(step)

            print(f"\nThought: {thought}")
            print(f"Action: {action}[{action_input}]")
            print(f"Observation: {observation}")

        return "最大ステップ数に到達しました。"


# === ツールの定義 ===

def search_tool(query: str) -> str:
    """簡易的な検索ツール（知識ベースからの検索をシミュレーション）"""
    knowledge_base = {
        "コロラド造山運動": (
            "コロラド造山運動は約3億年前の造山運動。"
            "東側にはグレートプレーンズが広がる。"
        ),
        "グレートプレーンズ": (
            "グレートプレーンズは北アメリカ中央部の広大な平原。"
            "面積は約130万平方マイル。"
        ),
        "エッフェル塔 高さ": (
            "エッフェル塔の高さは約330メートル（アンテナ含む）。"
            "1889年のパリ万博で建設された。"
        ),
        "東京スカイツリー 高さ": (
            "東京スカイツリーの高さは634メートル。"
            "2012年に開業した電波塔。"
        ),
    }

    for key, value in knowledge_base.items():
        if key in query:
            return value
    return f"'{query}' に関する情報は見つかりませんでした。"


def calculator_tool(expression: str) -> str:
    """計算ツール"""
    # 安全な評価のために使用可能な関数を制限
    allowed_names = {
        "abs": abs, "round": round,
        "sqrt": math.sqrt, "pi": math.pi, "e": math.e,
        "sin": math.sin, "cos": math.cos, "tan": math.tan,
        "log": math.log, "log10": math.log10,
        "pow": pow,
    }
    try:
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return f"{result}"
    except Exception as e:
        return f"計算エラー: {e}"


def lookup_tool(term: str) -> str:
    """用語検索ツール"""
    glossary = {
        "BFS": "幅優先探索。グラフの各ノードを層ごとに訪問する探索アルゴリズム。",
        "DFS": "深さ優先探索。可能な限り深く探索してからバックトラックするアルゴリズム。",
        "DAG": "有向非巡回グラフ。閉路を含まない有向グラフ。",
        "ハルシネーション": "LLMが事実に基づかない情報を自信を持って生成する現象。",
    }
    return glossary.get(term, f"'{term}' の定義は見つかりませんでした。")


# === ツールの登録と実行 ===

registry = ToolRegistry()
registry.register("Search", search_tool, "キーワードで情報を検索する")
registry.register("Calculate", calculator_tool, "数式を計算する")
registry.register("Lookup", lookup_tool, "用語の定義を調べる")

agent = ReActAgent(tools=registry, max_steps=8)

print("=" * 60)
print("ReActエージェント デモ")
print("=" * 60)
print(f"\n利用可能なツール:\n{registry.get_descriptions()}\n")

# デモ1: 多段階検索
print("\n【デモ1: 多段階質問応答】")
agent.run(
    question="エッフェル塔と東京スカイツリーの高さの差は何メートルですか？",
    thought_action_pairs=[
        {
            "thought": "まずエッフェル塔の高さを調べよう。",
            "action": "Search",
            "action_input": "エッフェル塔 高さ",
        },
        {
            "thought": "エッフェル塔は約330m。次に東京スカイツリーの高さを調べよう。",
            "action": "Search",
            "action_input": "東京スカイツリー 高さ",
        },
        {
            "thought": "東京スカイツリーは634m。差を計算しよう。",
            "action": "Calculate",
            "action_input": "634 - 330",
        },
        {
            "thought": "高さの差は304メートルとわかった。",
            "final_answer": "東京スカイツリー（634m）とエッフェル塔（330m）の高さの差は304メートルです。",
        },
    ]
)

# デモ2: 計算タスク
print("\n\n【デモ2: 数学的な計算】")
agent2 = ReActAgent(tools=registry, max_steps=8)
agent2.run(
    question="半径10cmの球の表面積と体積をそれぞれ求めてください。",
    thought_action_pairs=[
        {
            "thought": "球の表面積の公式は 4πr²。r=10で計算しよう。",
            "action": "Calculate",
            "action_input": "4 * 3.141592653589793 * 10**2",
        },
        {
            "thought": "表面積は約1256.6 cm²。次に体積を求めよう。"
                       "球の体積の公式は (4/3)πr³。",
            "action": "Calculate",
            "action_input": "(4/3) * 3.141592653589793 * 10**3",
        },
        {
            "thought": "体積は約4188.8 cm³。両方の値が揃った。",
            "final_answer": (
                "半径10cmの球の表面積は約1256.6 cm²、"
                "体積は約4188.8 cm³ です。"
            ),
        },
    ]
)

============================================================
ReActエージェント デモ
============================================================

利用可能なツール:
  - Search: キーワードで情報を検索する
  - Calculate: 数式を計算する
  - Lookup: 用語の定義を調べる

【デモ1: 多段階質問応答】
質問: エッフェル塔と東京スカイツリーの高さの差は何メートルですか？
--------------------------------------------------

Thought: まずエッフェル塔の高さを調べよう。
Action: Search[エッフェル塔 高さ]
Observation: エッフェル塔の高さは約330メートル（アンテナ含む）。...

Thought: エッフェル塔は約330m。次に東京スカイツリーの高さを調べよう。
Action: Search[東京スカイツリー 高さ]
Observation: 東京スカイツリーの高さは634メートル。...

Thought: 東京スカイツリーは634m。差を計算しよう。
Action: Calculate[634 - 330]
Observation: 304

Thought: 高さの差は304メートルとわかった。
Final Answer: 東京スカイツリー（634m）とエッフェル塔（330m）の高さの差は304メートルです。

import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
import numpy as np

# ReActのステップ構造を可視化
fig, axes = plt.subplots(1, 2, figsize=(14, 5))

# 左: ReActステップのフロー図（テキストベース）
ax1 = axes[0]
ax1.set_xlim(0, 10)
ax1.set_ylim(0, 10)
ax1.set_aspect('equal')
ax1.axis('off')
ax1.set_title('ReAct Loop Structure', fontsize=13, fontweight='bold')

# ボックスの定義
boxes = [
    (5, 9, 'Question', '#2196F3'),
    (5, 7, 'Thought', '#4CAF50'),
    (5, 5, 'Action', '#FF9800'),
    (5, 3, 'Observation', '#9C27B0'),
    (5, 1, 'Final Answer', '#F44336'),
]

for x, y, label, color in boxes:
    rect = mpatches.FancyBboxPatch(
        (x - 1.8, y - 0.5), 3.6, 1.0,
        boxstyle="round,pad=0.1",
        facecolor=color, edgecolor='white', alpha=0.85,
    )
    ax1.add_patch(rect)
    ax1.text(x, y, label, ha='center', va='center',
             fontsize=11, color='white', fontweight='bold')

# 矢印
for i in range(len(boxes) - 1):
    x1, y1 = boxes[i][0], boxes[i][1] - 0.5
    x2, y2 = boxes[i + 1][0], boxes[i + 1][1] + 0.5
    ax1.annotate('', xy=(x2, y2), xytext=(x1, y1),
                 arrowprops=dict(arrowstyle='->', color='#888',
                                 lw=1.5))

# ループ矢印（Observation → Thought）
ax1.annotate('', xy=(7.5, 7.5), xytext=(7.5, 3),
             arrowprops=dict(arrowstyle='->', color='#888',
                             lw=1.5, connectionstyle='arc3,rad=-0.4'))
ax1.text(8.8, 5, 'Loop', fontsize=10, color='#888', style='italic')

# 右: 手法別の特性レーダーチャート
ax2 = axes[1]
categories = ['Accuracy', 'Factuality', 'Latency\n(inverse)',
              'Interpretability', 'Cost\n(inverse)']
N = len(categories)

angles = [n / float(N) * 2 * np.pi for n in range(N)]
angles += angles[:1]

# 各手法のスコア（0-1、相対評価）
cot_scores = [0.6, 0.4, 0.95, 0.7, 0.95]
tot_scores = [0.85, 0.5, 0.3, 0.6, 0.2]
react_scores = [0.7, 0.9, 0.6, 0.9, 0.5]

cot_scores += cot_scores[:1]
tot_scores += tot_scores[:1]
react_scores += react_scores[:1]

ax2 = plt.subplot(122, polar=True)
ax2.set_theta_offset(np.pi / 2)
ax2.set_theta_direction(-1)

ax2.plot(angles, cot_scores, 'o-', linewidth=2, color='#2196F3',
         label='CoT', markersize=6)
ax2.fill(angles, cot_scores, alpha=0.1, color='#2196F3')

ax2.plot(angles, tot_scores, 's-', linewidth=2, color='#4CAF50',
         label='ToT', markersize=6)
ax2.fill(angles, tot_scores, alpha=0.1, color='#4CAF50')

ax2.plot(angles, react_scores, '^-', linewidth=2, color='#FF9800',
         label='ReAct', markersize=6)
ax2.fill(angles, react_scores, alpha=0.1, color='#FF9800')

ax2.set_xticks(angles[:-1])
ax2.set_xticklabels(categories, fontsize=9)
ax2.set_ylim(0, 1)
ax2.set_title('Method Comparison', fontsize=13, fontweight='bold',
              pad=20)
ax2.legend(loc='upper right', bbox_to_anchor=(1.3, 1.1), fontsize=9)

plt.tight_layout()
plt.savefig('react_analysis.png', dpi=150, bbox_inches='tight')
plt.show()