【次世代AI開発：第6回】Self-Healing Agent：エラーログを解読して自己修正するコード実行エージェント

【新連載：MCP・高度RAG・自律エージェントの現場設計論】

前回（第5回）：【次世代AI開発：第5回】GraphRAG超入門：ナレッジグラフ×ベクトル検索で全体関係性を捉えるハイブリッドRAG

※本連載は、MCP（Model Context Protocol）、Context Engineering、GraphRAG、Self-Healing Agent、ローカルSLMなどの最先端技術スタックを用い、AIシステムを構築するハンズオン連載です。

1. イントロダクション（本記事のねらいと到達目標）

前回は、ベクトル検索とナレッジグラフを融合させ、ドキュメント全体の複雑な関係性や広域な質問に回答する「GraphRAG」を解説しました。

ツール連携（MCP）、文脈最適化（Context Engineering）、検索高度化（GraphRAG）が揃ったことで、AIシステムは巨大な知識と接続口を手に入れました。しかし、これらを組み合わせて「実際のプログラミングコードを自動生成して実行するAIエージェント」を構築した際、開発者は次の現実的な壁に突き当たります。

それは「どれほど優秀なLLMであっても、一発で100%エラーの出ない完璧なコードを書くことは不可能である」という事実です。APIの仕様変更、ライブラリの引数ミス、型不整合、構文エラーなど、一発生成に頼るシステムは最初にエラーが発生した時点で即座にクラッシュして停止してしまいます。

この限界を突破し、真の自律性を実現するのが「Self-Healing Agent（自己修復型コード実行エージェント）」です。本記事では、生成したコードを隔離サブプロセスで実行し、エラーのスタックトレース（Traceback）を自ら解析して自動で修正・再実行を繰り返す、閉ループフィードバック（Closed-loop Feedback）システムを Python で完全構築することを目標とします。

前提条件と動作検証環境（Prerequisites）

Pythonバージョン: Python 3.10 以上
動作確認済み主要ライブラリ:
- openai>=1.30.0（OpenAI API クライアント）
- pydantic>=2.7.0（型定義ライブラリ）
- ast, subprocess（Python標準ライブラリ）
動作確認環境: macOS / Linux / Windows (PowerShell)

2. 【理論・背景】なぜ「自己修復ループ（Self-Healing Loop）」が必要なのか

従来の「LLMによるコード生成」と「Self-Healing Agent」の決定的な違いは、「エラーをシステム終了ではなく、次の行動のための『診断データ』として扱うか」という点にあります。

1. 人間依存アプローチ vs 自律閉ループアプローチ

人間がチャット画面でコードを生成させている場合、エラーが出たら人間がスタックトレースをコピーし、LLMに「このエラー直して」と貼り直す作業（Human-in-the-loop）を行っています。Self-Healing Agent は、この「実行 ➡ エラー捕捉 ➡ 修正依頼 ➡ 再実行」の思考ループをプログラム内で全自動化します。

【従来の対話型アプローチ (人間依存)】
[LLM] ---> コード生成 ---> [人間が手動実行] ---> エラー発生 ---> [人間がエラーをコピーしてLLMへ手動再投入]

【Self-Healing Agent (自律閉ループ)】
[LLM] ---> コード生成 ---> [サブプロセス実行] 
               ↑                 ↓ (エラー捕捉)
               └--- Traceback自動解析 ───┘ (全自動リトライ)

2. コード生成手法の比較

評価項目	従来の一発生成 (Single-Pass)	Self-Healing Agent (本手法)
完遂率	低い（初期コードの構文/実行エラーで即時即死）	極めて高い（エラー発生時に自動リファクタリング）
エラーの扱い	システムの例外・障害	エージェントの自己学習・修正用フィードバック
安全対策	メインプロセス内で直接実行しリスク大	`subprocess` / サンドボックスで隔離実行
構文チェック	実行するまでエラーが分からない	`ast.parse` による実行前の静的事前検知

3. 【アーキテクチャ解剖】Self-Healing Agent の5段階実行パイプライン

Self-Healing Agent は、コードの構文チェックから実行、エラー捕捉、修正パッチの適用までを以下の 5段階（5-Stage Pipeline） で循環制御します。

パイプラインを構成する5つのステップ

Code Generation（生成）: ユーザーの要求仕様を満たす Python コードをプロンプトから初期作成。
AST Static Check（静的検知）: ast.parse() を用いて、プログラムを実際に動かす前に文法エラー（SyntaxError）をミリ秒単位で事前検出。
Subprocess Execution（隔離実行）: subprocess.run を用い、タイムアウト（例: 10秒）を設定した隔離環境でコードを試走。
Traceback Analysis（解読・修正）: stderr（標準エラー出力）から生のスタックトレースを切り出し、LLMに「何が原因で壊れたか」を自律解析させてパッチコードを再生成。
Retry Management（再試行管理）: max_retries（上限3回）のカウント制御により、無限ループによるトークン無駄遣いを遮断。

4. 【実装】自己修復型コード実行エージェントの構築

それでは、実際にPythonコードを別プロセスで実行し、エラーログを読み取って自動でバグを修復・完遂する完全動作プログラム self_healing_agent.py を実装しましょう。

新規ファイル self_healing_agent.py を作成し、以下のコードを記述します。

import os
import sys
import ast
import subprocess
import tempfile
from typing import Optional, Tuple
from pydantic import BaseModel, Field
from openai import OpenAI

# 1. Pydantic による構造化出力モデルの定義

class CodeGenerationOutput(BaseModel):
    """LLMが生成するコードおよび解説"""
    thought_process: str = Field(description="どのようなロジックでコードを記述・修復したかの思考プロセス")
    python_code: str = Field(description="実行可能な完全なPythonソースコード（```python などのフェンスは除外）")


# =====================================================================
# Self-Healing Agent コアクラスの実装
# =====================================================================

class SelfHealingAgent:
    def __init__(self, model_name: str = "gpt-4o", max_retries: int = 3, timeout_seconds: int = 10):
        self.client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "dummy_key"))
        self.model_name = model_name
        self.max_retries = max_retries
        self.timeout_seconds = timeout_seconds

    def validate_ast(self, code_text: str) -> Tuple[bool, Optional[str]]:
        """1. AST (抽象構文ツリー) による静的文法チェック"""
        try:
            ast.parse(code_text)
            return True, None
        except SyntaxError as e:
            error_msg = f"SyntaxError [Line {e.lineno}, Col {e.offset}]: {e.msg}\n該当行: {e.text}"
            return False, error_msg

    def execute_code_in_subprocess(self, code_text: str) -> Tuple[bool, str, str]:
        """2. 隔離サブプロセスでのコード安全実行"""
        # 一時ファイルにコードを出力
        with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False, encoding="utf-8") as temp_file:
            temp_file.write(code_text)
            temp_script_path = temp_file.name

        try:
            # sys.executable を用いて現在のPython環境のサブプロセスで実行
            result = subprocess.run(
                [sys.executable, temp_script_path],
                capture_output=True,
                text=True,
                timeout=self.timeout_seconds
            )
            
            is_success = (result.returncode == 0)
            return is_success, result.stdout, result.stderr

        except subprocess.TimeoutExpired:
            return False, "", f"ExecutionTimeout: コードの実行が指定時間({self.timeout_seconds}秒)を超えタイムアウトしました。"
        except Exception as e:
            return False, "", f"SubprocessException: {str(e)}"
        finally:
            # 一時スクリプトファイルの削除クリーンアップ
            if os.path.exists(temp_script_path):
                os.remove(temp_script_path)

    def generate_or_fix_code(self, task_description: str, previous_code: Optional[str] = None, error_log: Optional[str] = None) -> CodeGenerationOutput:
        """3. LLMを用いたコード生成・エラー解読・修正パッチ作成"""
        if not previous_code:
            # 初回コード生成
            prompt = f"""あなたは優れたPythonエンジニアです。以下のタスクを解決する完全なPythonコードを作成してください。

【タスク内容】:
{task_description}

【制約事項】:
- インポートから実行部まで省略なく単体で動くコードにしてください。
- 出力は必ず指定の形式に従ってください。
"""
        else:
            # エラー解読と修復パッチ生成
            prompt = f"""前回のコード実行でエラーが発生しました。Tracebackを解読し、バグを修正した完全なPythonコードを再生成してください。

【要求タスク】: {task_description}

【失敗した前回のコード】:
{previous_code}

【発生したエラー / Traceback】:
{error_log}

【修正の指針】:
1. エラーの原因（モジュール未存在、引数間違い、型不一致等）を特定してください。
2. 根本原因を解決するようにコード全体を書き直してください。
"""
        response = self.client.beta.chat.completions.parse(
            model=self.model_name,
            messages=[{"role": "user", "content": prompt}],
            response_format=CodeGenerationOutput,
            temperature=0.1
        )
        return response.choices[0].message.parsed

    def run_task(self, task_description: str) -> bool:
        """Self-Healing 閉ループフィードバックの統合実行"""
        print(f"\n🤖 === Self-Healing Agent 起動: 『{task_description}』 ===")
        
        current_code = None
        error_log = None
        
        for attempt in range(1, self.max_retries + 1):
            print(f"\n🔄 --- [試行 #{attempt} / 最大 {self.max_retries} 回] ---")
            
            # Step A: コードの作成または修正パッチ適用
            res = self.generate_or_fix_code(task_description, current_code, error_log)
            current_code = res.python_code
            
            print(f"💡 [思考プロセス]: {res.thought_process}")
            
            # Step B: AST 静的文法チェック
            is_valid_ast, ast_error = self.validate_ast(current_code)
            if not is_valid_ast:
                print(f"❌ [ASTエラー検知]: 静的文法チェックで不合格になりました。\n{ast_error}")
                error_log = f"AST Syntax Check Error:\n{ast_error}"
                continue
                
            print("✅ [AST文法チェック]: 合格 (Syntax OK)")
            
            # Step C: 隔離サブプロセスでの試走
            print("🚀 [サブプロセス実行]: コードを隔離環境で試走中...")
            is_success, stdout, stderr = self.execute_code_in_subprocess(current_code)
            
            if is_success:
                print("\n🎉 ==================================================")
                print("     Self-Healing Agent: タスク実行成功！")
                print("==================================================")
                print("【実行出力 (stdout)】:")
                print(stdout if stdout else "(標準出力なし)")
                print("\n【完成した最終コード】:")
                print(current_code)
                return True
            else:
                print(f"💥 [実行時エラー発生 (Exit Code != 0)]")
                print("【捕獲された Traceback / stderr】:")
                print(stderr)
                error_log = stderr

        print(f"\n❌ [自己修復失敗]: {self.max_retries} 回試行しましたが、エラーを解消できませんでした。")
        return False

動作デモ

if __name__ == "__main__":
    if "OPENAI_API_KEY" not in os.environ:
        print("💡 NOTE: 実際に動かす場合は OPENAI_API_KEY の環境変数を設定してください。")
        sys.exit(0)

    agent = SelfHealingAgent(model_name="gpt-4o", max_retries=3)
    
    # あえてLLMが古いライブラリ引数や仕様誤りを犯しやすいタスクを指示
    task = "現在の日時とシステムメモリ使用率を取得し、PandasのDataFrameに格納して、'Timestamp'と'Memory_MB'の列名で表示するPythonスクリプトを作成してください。"
    
    agent.run_task(task)

5. コードの行別・ロジック詳細解説

実装した SelfHealingAgent の核となるロジックを解説します。

1. `ast.parse(code_text)` による事前検知

ast（Abstract Syntax Tree）モジュールを使用することで、python コマンドでプロセスを生成する前に、閉じ括弧の忘れやインデントミスなどの SyntaxError をノーコスト（数ミリ秒）で即座に判定します。

2. `subprocess.run` による安全な分離とログ補獲

メインプロセス（Agentの制御プログラム）の中で直接 exec() や eval() を呼ぶのはセキュリティ上非常に危険です。
subprocess.run([sys.executable, temp_script_path], capture_output=True, timeout=10) を使用することで、生成されたコードの実行環境を完全に子プロセスへ隔離し、timeout で無限ループ事故を防止しています。
子プロセスがクラッシュした場合でも、result.stderr（エラーのスタックトレース）が正確に文字列として捕獲されます。

3. Traceback のコンテキストフィードバックと修正パッチ

捕獲された stderr のスタックトレースをそのまま LLM のプロンプト error_log へ投入します。
LLMは「どの行で、何の例外（例: ModuleNotFoundError や KeyError）が起きたか」を理解できるため、2回目の試行でピンポイントにバグを修正した正確なコードを出力します。

6. プロダクション導入・セキュリティ・サンドボックス隔離

本手法を実際のエンタープライズ本番環境に組み込む際の防衛線について解説します。

1. サンドボックス隔離（Docker / gVisor / E2B）

ローカルの subprocess 実行は開発段階では有効ですが、悪意あるプロンプトインジェクションによって os.system("rm -rf /") などの危険なコードが生成されるリスク（RCE: Remote Code Execution）があります。

プロダクション環境では、E2B Sandbox や Docker コンテナ、あるいは Azure Dynamic Sessions のような隔離サンドボックスコンテナ内で subprocess を実行させる構成が必須となります。

【プロダクションでのセキュア構成】
[Self-Healing Agent] ---> (コード送信) ---> [Docker無菌サンドボックスコンテナ]
                                                     ↓ (隔離実行)
[Agent] <--- (Tracebackのみ返却) <--- [コンテナ実行結果 / タイムアウト遮断]

2. ループ上限（Max Retries）とコストコントロール

自己修復ループが無制限に回ると、APIトークンコストが無限に増大します。
max_retries は 3〜4回 を上限とし、エラーが解決しない場合はログを残して安全にフォールバック停止させる設計を徹底してください。

7. まとめと次回の展望

今回は、LLMによるコード生成の失敗をシステム障害ではなく「自己改善のためのフィードバック」として扱い、subprocess 隔離実行と Traceback 解析によって自力でバグを修復する Self-Healing Agent の理論と Python 実装を解説しました。

自律的にエラーを乗り越えてタスクを完了させるエージェントの閉ループ構造の威力を体感できたかと思います。

しかし、単一のエージェントが自律ループを回すだけでなく、複数の専門エージェント（例: リサーチャー、コーダー、レビュアー）がチームとして協調動作を始めると、今度は「エージェント同士の無限ループ」や「状態のデッドロック」という新しい複雑性に直面します。

次回、第7回。
複数エージェントの協調を安全に統制する、**『マルチエージェントの安全制御：ステートグラフとデッドロック・防衛線設計』**に進みます。

LangGraph やカスタムステートマシンを用いて、マルチエージェントの暴走を防ぎ、決定論的な安全制御を構築する極意へ進みましょう。