Python による推論とプランニング
この章では、AI Agent の中核となる知能機能、すなわち推論とプランニングの能力について紹介します。
従来の命令実行モードとは異なり、推論能力を備えた Agent は複雑な思考プロセスを実行できます。
複雑なタスクを実行可能なステップに分解し、コンテキストに基づいて意思決定を行うことができます。
ReAct フレームワーク
ReAct（Reasoning + Acting）は、推論と行動を組み合わせたフレームワークです。
ReAct モードでは、Agent は推論と行動を交互に行い、閉ループを形成します。
推論が行動を導き、行動の結果が再び推論プロセスにフィードバックされ、このサイクルが繰り返されます。
ReAct は人間の問題解決プロセスを模倣しています。行動を起こす前に考え、行動後に結果を観察し、新しい情報に基づいて推論を続けます。
動作原理
ReAct の核心理念は、3つのフェーズのサイクルとして要約できます。
最初のステップは推論フェーズです。Agent は現在のコンテキストに基づいて、次の行動に関する思考を生成します。
2番目のステップは行動フェーズです。Agent は選択したツールまたはアクションを実行します。
3番目のステップは観察フェーズです。Agent は行動の結果を受け取り、コンテキストを更新し、最初のステップに戻って推論を続けます。
コード実装
以下は ReAct Agent の簡易実装です。
ReAct Agent の基本実装
"""
ReAct フレームワークの Agent 実装
コアメカニズム：推論 -> 行動 -> 観察 -> 推論（ループ）
"""
def __init__(self, llm, tools, max_iterations=5):
# 推論と生成に使用する LLM インスタンス
self.llm = llm
# 利用可能なツールのリスト
self.tools = tools
# 無限ループを防ぐための最大反復回数
self.max_iterations = max_iterations
def run(self, task):
"""
ReAct ループを実行
:param task: ユーザーのタスク記述
:return: 実行結果
"""
# タスクと履歴情報を含むコンテキストを初期化
context = {
"task": task,
"steps": [], # 実行済みステップの記録
"observations": [] # 観察結果の記録
}
for i in range(self.max_iterations):
# フェーズ1：推論 - 現在のコンテキストに基づいて次の行動を生成
reasoning = self.llm.reason(context)
# ツールを実行する必要があるか、それとも直接回答を返すかを判断
if reasoning.needs_action:
# 使用するツールを選択
action = reasoning.select_tool(self.tools)
# ツールを実行し、観察結果を取得
observation = action.execute(reasoning.tool_input)
# 観察結果をコンテキストに追加
context["observations"].append(observation)
else:
# 行動不要のため、推論で得た回答を直接返す
return reasoning.answer
# 最大反復回数に達しても完了していない
return "最大反復回数制限に達しました"
代表的な適用シナリオ
ReAct モードは特に以下のシナリオに適しています。
環境内を探索し、情報を収集してから完了できるタスク。
インテリジェント検索シナリオ：Agent はまず情報を検索し、その検索結果に基づいて推論と要約を行う必要があります。
対話型 QA：Agent は複数回の対話を通じて要件を明確化し、情報を取得する必要があります。
複雑な問題解決：問題が1つのステップで解決できず、複数の中間ステップが必要です。
注意：ReAct モードの利点はその柔軟性にあり、各ステップの実行結果に応じて後続の行動を動的に調整できます。ただし、実行経路が不安定になる可能性もあるため、探索が必要なタスクに適しています。
思考連鎖（Chain of Thought, CoT）
思考連鎖は、モデルに段階の推論プロセスを提示させる技術です。
CoT はモデルに直接答えを出させるのではなく、まず推論ステップを示し、その後で最終結論を導き出させます。
なぜ思考連鎖が必要か
モデルに直接答えを出力させると、いくつかの問題があります。
複雑な推論の中間ステップが無視されたり、省略されたりしやすい。
エラーがどこで発生したかを特定しにくい。
ユーザーがモデルの思考プロセスを理解できない。
思考連鎖は、モデルに推論プロセスを提示させることで、これらの問題を解決します。
Zero-shot CoT
Zero-shot CoT は、例を示さずに段階的な推論能力を引き出す方法です。
プロンプトの末尾に「一つずつ考えてみましょう」のようなトリガー文を追加するだけで済みます。
Zero-shot CoT の例
prompt = """
問題：小明はリンゴを5個持っています。小红が3個くれました。小明は2個食べました。残りは何個ですか？
一つずつ考えてみましょう：
"""
# 言語モデルを呼び出し
response = llm.generate(prompt)
# モデル出力例：
# ステップ1：小明が受け取った後の合計数を計算
# 小明は元々5個のリンゴを持っていて、小红が3個くれました
# 5 + 3 = 8個
#
# ステップ2：食べた後の数を計算
# 小明は2個食べました
# 8 - 2 = 6個
#
# 結論：残りは6個のリンゴです
Few-shot CoT
Few-shot CoT は、詳細な推論プロセスを含む例を提供することで、モデルが特定の推論パターンを学習するのを助けます。
Zero-shot CoT の効果が不十分な場合、Few-shot CoT を試すことができます。
Few-shot CoT の例
prompt = """
例1：
問題：小张は10元持っています。3冊の本を買いました。各2元です。残りはいくらですか？
一つずつ考えてみましょう：
- 小张は元々10元持っています
- 各本2元、3冊購入、支出は 3 × 2 = 6元
- 10 - 6 = 4元
答え：残りは4元です
例2：
問題：ある猫は毎時2匹のネズミを捕まえます。8時間で何匹捕まえますか？
一つずつ考えてみましょう：
- 毎時2匹のネズミを捕まえる
- 8時間で 8 × 2 = 16匹
答え：16匹捕まえます
問題：{user_question}
一つずつ考えてみましょう：
"""
response = llm.generate(prompt)
思考連鎖の利点
思考連鎖技術は、3つの核心的価値をもたらします。
追跡可能性：複雑な推論を追跡可能な中間ステップに分解し、推論プロセスの理解を容易にします。
説明可能性：モデルの説明可能性を高め、ユーザーが答えの由来を知ることができるようにします。
精度の向上：複雑な推論タスクの精度を大幅に向上させ、特に数学および論理タスクで顕著な効果を発揮します。
Tree of Thoughts（ToT）
思考の木（Tree of Thoughts）は思考連鎖の拡張であり、線形的な推論に限定されません。
ToT は各推論ノードで複数の可能な経路を探索し、木構造を形成します。
これにより、Agent は複数経路の探索、バックトラック、および全体的な評価を行うことができます。
思考連鎖との違い
CoT は線形的な推論を採用し、各ステップは前のステップの結論に依存するため、明確な経路がある問題に適しています。
ToT は空間的な推論を採用し、複数の可能な分岐を同時に考慮するため、探索とプランニングが必要な問題に適しています。
コード実装
ToT Agent の基本実装
"""
Tree of Thoughts Agent の実装
コアメカニズム：各ノードで複数の候補分岐を生成し、評価後に最適なものを選択して継続
"""
def __init__(self, llm, max_depth=4, beam_size=3):
# LLM インスタンス
self.llm = llm
# 最大探索深度
self.max_depth = max_depth
# 各層で保持する候補ノード数（beam width）
self.beam_size = beam_size
def solve(self, problem):
"""
ToT フレームワークを使用して問題を解決する
:param problem: 問題の説明
:return: 最適解
"""
# 問題を初期状態として含むルートノードを作成する
root = ThoughtNode(problem, depth=0)
# フロンティアノードリスト（展開待ちのノード）を初期化する
frontier = [root]
# 層ごとに展開する
for depth in range(self.max_depth):
# すべての候補ノードを保存する
all_candidates = []
# すべてのフロンティアノードを走査する
for node in frontier:
# 現在のノードに対して複数の次候補手を生成する
candidates = self.llm.generate_thoughts(
node.content, # 現在のノードの内容
n=self.beam_size # 生成数
)
# 新しいノードを作成し候補リストに追加する
for cand in candidates:
all_candidates.append(
ThoughtNode(cand, depth + 1, parent=node)
)
# すべての候補ノードを評価する
evaluated = self.evaluator.rank(all_candidates)
# 次ラウンドのフロンティアとして最適な beam_size 個のノードを選択する
frontier = evaluated[:self.beam_size]
# 解決策が見つかったかどうかを確認する
if self.is_solution(frontier):
break
# バックトラックして最適解を見つける
return self.backtrack_best(frontier)
適用シナリオ
ToT は選択や計画が必要なシナリオに特に適しています：
創造的ライティング：複数のストーリー展開を生成し、評価後に最適なものを選択します。
戦略計画：複数の行動計画の潜在的な結果を評価します。
複雑な意思決定：複数の可能性を考慮する必要がある意思決定問題です。
タスクプランニングと MCTS
モンテカルロ木探索（Monte Carlo Tree Search, MCTS）は、複雑な意思決定問題のためのヒューリスティック探索アルゴリズムです。
MCTS はエージェント計画で広く応用されており、特にゲーム AI や複雑なタスク計画に適しています。
核心理念
MCTS はランダムなゲームをシミュレーションすることで、各意思決定ノードの潜在的な価値を評価します。
すべての可能な経路を評価する必要はなく、サンプリングと統計によって探索方向を導きます。
MCTS の4つのステップ
ステップ1、選択（Selection）：ルートノードから開始し、葉ノードに到達するまで最適な子ノードを再帰的に選択します。
選択時には UCB（Upper Confidence Bound）の式を使用して、探索と活用のバランスを取ります。
ステップ2、拡張（Expansion）：葉ノードに1つ以上の子ノードを追加します。
ステップ3、シミュレーション（Simulation）：新しいノードから開始し、ゲームが終了するまでランダムにシミュレーションします。
ステップ4、逆伝播（Backpropagation）：シミュレーション経路上のすべてのノードの統計情報を更新します。
コード実装
MCTS プランナーの実装
class MCTSNode:
"""
MCTS ツリーノード
ノードの状態、統計データ、子ノードの関係を保存します
"""
def __init__(self, state, parent=None, action=None):
# 現在の状態
self.state = state
# 親ノードへの参照
self.parent = parent
# 親ノードからこのノードへ到達するためのアクション
self.action = action
# 子ノードのリスト
self.children = []
# このノードの訪問回数
self.visit_count = 0
# このノードの累積報酬値
self.reward = 0.0
def is_fully_expanded(self):
"""すべての可能な子ノードを展開済みかどうかを確認する"""
return len(self.children) > 0
def is_terminal(self):
"""終端ノード（ゲーム終了または目標達成）かどうかを確認する"""
return self.state.is_terminal()
def uct_child(self):
"""
UCT の式を使用して最適な子ノードを選択する
UCT = reward/visits + C * sqrt(ln(parent_visits)/visits)
C は探索定数で、通常 sqrt(2) に設定します
"""
# 探索定数、探索と活用のバランスを取る
C = math.sqrt(2)
return max(
self.children,
key=lambda c: c.reward / c.visit_count +
C * math.sqrt(math.log(self.visit_count) / c.visit_count)
)
class MCTSPlanner:
"""
MCTS プランナー
モンテカルロ木探索を使用してエージェントの行動計画を生成する
"""
def __init__(self, simulation_limit=1000, exploration_constant=1.41):
# 最大シミュレーション回数
self.simulation_limit = simulation_limit
# 探索定数
self.exploration_constant = exploration_constant
def plan(self, initial_state):
"""
初期状態から計画を開始する
:param initial_state: 初期状態
:return: 最適なアクション
"""
# ルートノードを作成する
root = MCTSNode(initial_state)
# 複数回シミュレーションを実行する
for _ in range(self.simulation_limit):
# 1. 選択：ルートノードから葉ノードまで最適な子ノードを選択する
node = self._selection(root)
# 2. 拡張：終端ノードでない場合は、新しいノードを展開する
if not node.is_terminal():
node = self._expansion(node)
# 3. シミュレーション：新しいノードから終端までランダムにシミュレーションする
reward = self._simulation(node)
# 4. 逆伝播：経路上のすべてのノードの統計を更新する
self._backpropagation(node, reward)
# ルートノードの最適な子ノードに対応するアクションを返す
return root.best_child().action
def _selection(self, node):
"""選択フェーズ：最適な子ノードを選択する"""
while node.is_fully_expanded() and not node.is_terminal():
node = node.uct_child()
return node
def _expansion(self, node):
"""拡張フェーズ：新しい子ノードを追加する"""
# すべての可能なアクションを生成する
possible_actions = node.state.get_possible_actions()
# 各アクションに対して子ノードを作成する
for action in possible_actions:
new_state = node.state.apply_action(action)
child = MCTSNode(new_state, parent=node, action=action)
node.children.append(child)
# ランダムに1つの子ノードを返す（決定論的戦略を使用することも可能）
return node.children[0]
def _simulation(self, node):
"""シミュレーションフェーズ：ゲーム終了までランダムにシミュレーションする"""
state = node.state
while not state.is_terminal():
# ランダムにアクションを選択する
actions = state.get_possible_actions()
action = random.choice(actions)
state = state.apply_action(action)
# 最終報酬を返す
return state.get_reward()
def _backpropagation(self, node, reward):
"""逆伝播：統計情報を更新する"""
while node is not None:
node.visit_count += 1
node.reward += reward
node = node.parent
注意：MCTS は計算コストが高く、深い計画が必要だが明確な終了条件があるシナリオに適しています。リアルタイム性が要求されるタスクでは、シミュレーション回数を制限するか、他の方法を使用する必要があるかもしれません。
Reflexion（自己内省）
Reflexion はエージェントに自己反省能力を持たせる技術です。
エージェントに反省メカニズムを追加することで、失敗後にエラーの原因を分析し、戦略を調整して再試行できます。
核心理念
エージェントはアクションを実行するだけでなく、結果を観察して反省します：なぜ失敗したのか？次回はどう改善すべきか？
この能力は継続的な学習と自己改善にとって非常に重要です。
コード実装
Reflexion Agent の実装
"""
自己反省能力を持つエージェント
核心メカニズム：実行 -> 評価 -> 反省 -> 再試行
"""
def __init__(self, actor, reviewer, max_retries=3):
# 実行器：具体的なタスクの実行を担当
self.actor = actor
# 評価器：実行結果の評価を担当
self.reviewer = reviewer
# 最大再試行回数
self.max_retries = max_retries
def run(self, task):
"""
自己反省付きのタスクを実行する
:param task: タスクの説明
:return: 実行結果
"""
# 実行履歴を保持する
history = []
for attempt in range(self.max_retries):
# フェーズ1：タスクの実行を試みる
result = self.actor.execute(task, history)
# 今回の試行を記録する
history.append({
"attempt": attempt,
"result": result
})
# フェーズ2：結果を評価する
feedback = self.reviewer.evaluate(task, result)
# フェーズ3：成功したかどうかを確認する
if feedback.is_success:
return result
# フェーズ4：失敗の原因を反省する
# 次の試行を導くための新しい戦略プロンプトを生成する
reflection = self.reviewer.reflect(
task, # 元のタスク
result, # 失敗結果
feedback # 評価フィードバック
)
# 反省結果を履歴に追加し、次回の試行の参考にする
history.append({
"type": "reflection",
"content": reflection
})
# すべての再試行が失敗した場合、履歴を返す
return history
class Reviewer:
"""評価器：実行結果を評価し、反省を生成する"""
def evaluate(self, task, result):
"""
実行結果を評価する
:return: 成功したかどうかと詳細なフィードバックを含むオブジェクト
"""
# 結果がタスクの要件を満たしているかどうかを確認する
is_success = self.check_success(task, result)
if is_success:
return EvaluationResult(is_success=True)
else:
# 失敗原因の分析を生成する
failure_reasons = self.analyze_failures(task, result)
return EvaluationResult(
is_success=False,
reasons=failure_reasons
)
def reflect(self, task, result, feedback):
"""
反省内容を生成する
Actor が失敗原因を理解し、戦略を改善するのを支援する
"""
prompt = f"""
タスク：{task}
実行結果：{result}
失敗原因：{feedback.reasons}
失敗原因を分析し、改善提案を提示してください。
重点的な説明：
1. どこに問題がありましたか？
2. 次回はどう回避すべきですか？
3. どの戦略を変更する必要がありますか？
"""
return self.llm.generate(prompt)
適用シナリオ
Reflexion は特に以下のシナリオに適しています：
対話システムやコード生成など、継続的な改善が必要なタスク。
エラーのコストは高いが再試行のコストが低いシナリオ。
失敗から学習する必要がある状況。
タスク分解戦略
複雑なタスクは通常、管理可能なサブタスクに分解する必要があります。
効果的なタスク分解は計画能力の基盤です。
再帰的タスク分解
サブタスクが直接実行できるようになるまで、タスクを再帰的により小さなサブタスクに分解します。
再帰的タスク分解
"""
タスクを再帰的に分解する
:param task: 分解するタスク
:param is_executable_fn: タスクが直接実行可能かどうかを判断する関数
:param decompose_fn: タスクを分解する関数
:return: 実行可能なタスクのリスト
"""
# タスクが直接実行可能な場合は、そのまま返す
if is_executable_fn(task):
return [task]
# タスクをサブタスクに分解する
subtasks = decompose_fn(task)
# 各サブタスクを再帰的に分解する
result = []
for subtask in subtasks:
# サブタスクに対して分解関数を再帰的に呼び出す
result.extend(
decompose(subtask, is_executable_fn, decompose_fn)
)
return result
# 例：タスクが実行可能かどうかを判断する
def is_code_complete(task):
"""タスクがコードを直接実行できるかどうかを確認する"""
return task.get("type") == "code" and len(task.get("dependencies", [])) == 0
# 例：複雑なタスクを分解する
def decompose_programming_task(task):
"""プログラミングタスクを分解する"""
if "write_function" in task["type"]:
return [
{"type": "understand_requirements", "task": task["spec"]},
{"type": "write_code", "spec": task["spec"]},
{"type": "write_tests", "function": task["function_name"]},
{"type": "verify_tests", "function": task["function_name"]}
]
return [task]
並列的タスク分解
並列実行できる独立したサブタスクを特定し、実行効率を向上させます。
これはタスク実行を高速化するための重要な戦略です。
階層的タスク分解
タスクを異なる抽象レベルに分割し、上位レベルのタスクが下位レベルのタスクを呼び出して、タスク階層ツリーを形成する。
複数層の抽象化を必要とする複雑なシステムに適している。
Plan-and-Execute モード
Plan-and-Execute は、計画と実行を分離するアーキテクチャパターンである。
Agent はまずタスク全体の流れを完全に計画し、その後計画に従って実行する。
ReAct との違い
ReAct は推論しながら実行する方式で、より柔軟だが経路が不安定になる可能性がある。
Plan-and-Execute は先に計画してから実行する方式で、より安定しているが動的な調整能力に欠ける。
コード実装
Plan-and-Execute Agent の実装
"""
Plan-and-Execute アーキテクチャの Agent
中核メカニズム：最初に完全な計画を立て、その後計画に従って実行する
"""
def __init__(self, planner, executor):
# プランナー：実行計画の策定を担当する
self.planner = planner
# 実行器：具体的なステップの実行を担当する
self.executor = executor
def run(self, task):
"""
タスクを実行する
計画フェーズと実行フェーズに分けられる
"""
# ==================== 計画フェーズ ====================
# 完全な実行計画を一度に生成する
plan = self.planner.create_plan(task)
# ==================== 実行フェーズ ====================
results = []
# 計画に従って順番に実行する
for step in plan.steps:
# 現在のステップを実行する
result = self.executor.execute(step)
results.append(result)
# 再計画が必要かどうかを確認する
# 例：実行結果が期待と異なる場合、または予期しない状況が発生した場合
if self.needs_replan(results):
# 現在の結果に基づいて再計画する
plan = self.planner.replan(task, results)
# すべての結果を統合する
return self.summarize(results)
def needs_replan(self, results):
"""
再計画が必要かどうかを判断する
実行結果が期待どおりかどうかを確認する
"""
# 最後の結果を取得する
last_result = results[-1]
# 結果が明らかに期待と異なる場合、再計画が必要である
if last_result.is_unexpected():
return True
# 結果によって後続のステップを実行できない場合、再計画が必要である
if last_result.blocks_future():
return True
return False
class Planner:
"""プランナー：タスク実行計画を生成する"""
def create_plan(self, task):
"""初期実行計画を作成する"""
prompt = f"""
タスク：{task}
詳細な実行計画を策定してください。以下を含みます：
1. 実行する必要があるステップの順序
2. 各ステップの具体的な操作
3. ステップ間の依存関係
出力形式：
ステップ1: [操作の説明]
ステップ2: [操作の説明]
...
"""
plan_text = self.llm.generate(prompt)
return Plan.parse(plan_text)
def replan(self, task, results):
"""実行結果に基づいて再計画する"""
完了した結果を分析する
completed = [r for r in results if r.is_success]
failed = [r for r in results if not r.is_success]
# 調整後の計画を生成する
prompt = f"""
元のタスク：{task}
完了したステップ：{completed}
失敗したステップ：{failed}
調整後の実行計画を策定してください：
"""
plan_text = self.llm.generate(prompt)
return Plan.parse(plan_text)
class Executor:
"""実行器：具体的な計画ステップを実行する"""
def execute(self, step):
"""単一のステップを実行する"""
# ステップの種類に応じて実行方法を選択する
if step.type == "tool_call":
return self.execute_tool_call(step)
elif step.type == "code":
return self.execute_code(step)
elif step.type == "query":
return self.execute_query(step)
return Result(success=False, error="Unknown step type")
注意：Plan-and-Execute パターンは、タスク構造が比較的安定しており、事前に計画できるシナリオに適している。柔軟な対応が必要な環境では、ReAct パターンの方が適している場合がある。
章のまとめ
この章では、AI Agent の推論と計画の中核能力について紹介した。
ReActフレームワークは推論と行動のループを通じて、Agent が実行しながら戦略を調整できるようにする。
思考の連鎖（CoT）段階的な推論プロセスを示すことで、複雑なタスクの実行精度を向上させる。
思考の木（ToT）思考の連鎖を拡張し、複数経路の探索とバックトラックをサポートする。
MCTSヒューリスティック探索アルゴリズムであり、複雑な意思決定や計画問題に適している。
ReflexionAgent に自己内省能力を与え、失敗から学び改善する。
タスク分解戦略計画能力の基盤であり、再帰的分解、並列分解、階層的分解を含む。
Plan-and-Execute先に計画してから実行するモードを採用し、構造が安定したタスクに適している。
これらの技術は単独で使用することも、組み合わせて使用することもでき、より強力な Agent システムを構築できる。
その他の拡張