自己修復型エージェントループの構築（読了時間：39 分）

このクックブックは、出力を生成して検証し、フィードバックに基づいて次の処理を改善する「クローズドループ型エージェントワークフロー」について解説します。 ここでは、古くなったり壊れたりした API や SDK のサンプルを検出し、修復、検証するドキュメント信頼性向上のワークフローを実践形式で紹介します。実演には、このクックブックリポジトリから意図的に古くされたノートブックを流用しています。 このエージェントループは Codex を使って構築します。Codex は現在の状態を確認し、必要な箇所に焦点を当てた修正を加え、検証を実行します。フィードバックに未解決の問題が残っていれば、そのプロセスを繰り返します。 このノートブックでのタスクはあくまで例示です。エージェントの出力を信頼できるフィードバックで測定できるあらゆる場面で、このパターンは応用可能です。 本ワークフローには 3 つのフェーズがあります。 - **レビュー**: ファイルを編集せず、現在の成果物を検査して構造化された見解を返します。 - **修復**: レビュー結果と最新の検証フィードバックに基づき、コピーした成果物に対して焦点を絞った修正を加えます。 - **検証**: 関連するチェックを実行し、まだ改善が必要な箇所を報告します。 検証がループを完結させます。修復されたノートブックは、重要なチェック項目を満たす必要があります。残っている問題は、次の修復処理の入力として引き継がれます。 ![Codex iterative repair loop for technical documentation](https://developers.openai.com/cookbook/assets/images/codex_iterative_repair_loop.png) このノートブックは、ヘッドレスモードで [Codex CLI](https://developers.openai.com/codex/cli) を使用するため、チャット UI の代わりに Python セルから修復ステップを実行できます。最初のコードセルでは CLI をインストールしますが、すでに持っている場合はスキップしてください。 ライブな修復ループを実行する前に、環境変数に `OPENAI_API_KEY` を設定してください。 このノートブックはデフォルトで高速な修復モデルを使用しているため、例全体の処理時間を短縮しています。異なるモデルを試したい場合は、開始前に `REPAIR_MODEL` を設定してください。インストールセルでは再現性を保つために特定の Codex CLI バージョンを固定していますが、新しい CLI の挙動が必要になった場合のみ意図的にバージョンを更新してください。 ``` !npm install -g @openai/codex@0.130.0 ``` ``` import concurrent.futures import json import os import shlex import shutil import subprocess import tempfile from pathlib import Path from typing import Any CANDIDATE_EXAMPLE_DIRS = [Path("."), Path("examples/codex")] EXAMPLE_DIR = next((base for base in CANDIDATE_EXAMPLE_DIRS if (base / "data" / "docs").exists()), None) if EXAMPLE_DIR is None: raise RuntimeError( "This notebook needs its companion sample notebooks. " "Download the data folder that ships with this example and place it next to " "this notebook as ./data/docs, or run from a checkout where examples/codex/data/docs exists." ) DATA_DIR = EXAMPLE_DIR / "data" / "docs" DEFAULT_RUNS_DIR = Path(tempfile.gettempdir()) / "codex_iterative_repair_loop_outputs" RUNS_DIR = Path(os.getenv("CODEX_REPAIR_RUNS_DIR", str(DEFAULT_RUNS_DIR))).expanduser() RUNS_DIR.mkdir(parents=True, exist_ok=True) ``` ``` MODEL = os.getenv("REPAIR_MODEL", "gpt-5.4-mini") COOKBOOK_CHAT_MODEL = os.getenv("COOKBOOK_CHAT_MODEL", "gpt-5.5") REPAIR_REASONING_EFFORT = os.getenv("REPAIR_REASONING_EFFORT", "low") if not os.environ.get("OPENAI_API_KEY"): raise ValueError("Set the OPENAI_API_KEY environment variable before running the live Codex repair loop.") CODEX_CLI = shutil.which("codex") if CODEX_CLI is None: raise RuntimeError("Run the install cell before continuing; Codex CLI is not on PATH.") ``` 以下のセルでは、3 つの補完ノートブックを読み込み、修復ループを駆動するメタデータの概要を表示します。 サンプルはあえて小さく設定されています。実行が速い一方で、アーキテクチャの有効性を十分に検証できます。レビューで本質的な課題を発見し、修復で焦点を絞った編集を行い、検証で次のパスへのフィードバックを提供するという流れです。 このノートブック単体でダウンロードする場合は、`data/docs/` フォルダも一緒にダウンロードし、以下のセルを実行する前にノートブックと同じ場所に配置してください。コードはこれらのサンプルノートブックがローカルに存在することを前提としています。 今回の例では、検証ステップとして修復されたノートブックを最初から最後まで実行します。他のドメインでは、検証はユニットテスト、ポリシーチェック、スキーマバリデーション、シミュレーション、あるいは人間の承認ステップになることもあります。重要なのは、失敗が行き止まりではなく構造化されたフィードバックとして機能することです。 ``` NOTEBOOKS = [ DATA_DIR / "qdrant_embeddings_search_pre_repair.ipynb", DATA_DIR / "getting_started_evals_pre_repair.ipynb", DATA_DIR / "knowledge_retrieval_pre_repair.ipynb", ] def read_notebook(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def case_metadata(path: Path) -> dict[str, Any]: return read_notebook(path).get("metadata", {}).get("codex_case_study", {}) cases = [] for notebook_path in NOTEBOOKS: notebook = read_notebook(notebook_path) metadata = notebook.get("metadata", {}).get("codex_case_study", {}) repair_story = metadata.get("repair_story", {}) cases.append( { "notebook": notebook_path.name, "cells": len(notebook["cells"]), "code_cells": sum(cell["cell_type"] == "code" for cell in notebook["cells"]), "source": metadata.get("source_path"), "target_iteration": repair_story.get("target_iteration"), "repair_depth": repair_story.get("repair_depth", ""), } ) cases ``` ``` [{'notebook': 'qdrant_embeddings_search_pre_repair.ipynb', 'cells': 5, 'code_cells': 4, 'source': 'examples/vector_databases/qdrant/Using_Qdrant_for_embeddings_search.ipynb', 'target_iteration': 1, 'repair_depth': 'One-pass cleanup: modernize the local Qdrant query path and clarify the sampled fixture framing.'}, {'notebook': 'getting_started_evals_pre_repair.ipynb', 'cells': 5, 'code_cells': 4, 'source': 'examples/evaluation/Getting_Started_with_OpenAI_Evals.ipynb', 'target_iteration': 2, 'repair_depth': 'Two-pass cleanup: first modernize the obvious stale Evals flow, then use validation feedback to remove result-log brittleness.'}, {'notebook': 'knowledge_retrieval_pre_repair.ipynb', 'cells': 5, 'code_cells': 4, 'source': 'examples/How_to_call_functions_for_knowledge_retrieval.ipynb', 'target_iteration': 3, 'repair_depth': 'Three-pass cleanup: modernize model/API shape, then tighten runnable local setup, then restore the full retrieval teaching flow.'}] ``` Codex にアートのレビューや修復を依頼する際、まずは小さな共通契約（共有ルール）を与えましょう。これにより、モデルが製品やスタイルのルールをゼロから推測する必要がなくなり、ループは重要な課題に集中できます。 以下のルールは、これらのノートブックにおける「良い状態」の基準を定義しています。具体的には、現在の API パターン、明確なセットアップ手順、ローカルで実行可能なサンプル、そして元の教育目標の維持です。別のワークフローでは、この契約がそのドメインにおける唯一の真実（ソース・オブ・トゥルース）として機能します。 ``` business_rules = { "preferred_chat_model": COOKBOOK_CHAT_MODEL, "preferred_embedding_model": "text-embedding-3-large", "modernize": [ "client.chat.completions.create -> client.responses.create", "legacy function-calling schemas -> current tools schema", "qdrant.search -> qdrant.query_points", "oaieval CLI examples -> current Evals API workflow", ], "reader_experience": [ "Make fresh-environment setup explicit.", "Keep the included examples runnable with local data and the standard library.", "Keep sample repairs self-contained unless the notebook explicitly teaches external setup.", "Remove manual result-file placeholders.", "State runtime prerequisites and side effects before readers run cells.", "Preserve the original teaching goal while modernizing the implementation.", ], } business_rules ``` 各フェーズは構造化されたデータを返すため、次のフェーズで具体的なデータを利用できます。 レビューでは発見事項が報告され、修復では変更サマリーと更新されたアーティファクトへのパスが提供されます。検証では、次のパスに残った差分が示されます。このようにハンドオフを構造化することで、ループのデバッグや再実行が容易になり、他の種類のアーティファントにも柔軟に対応できるようになります。 ``` def object_schema(properties: dict[str, Any], required: list[str] | None = None) -> dict[str, Any]: return { "type": "object", "properties": properties, "required": required or list(properties), "additionalProperties": False, } def string_array() -> dict[str, Any]: return {"type": "array", "items": {"type": "string"}} finding_schema = object_schema( { "artifact": {"type": "string"}, "issue_type": {"type": "string"}, "severity": {"type": "string"}, "description": {"type": "string"}, "suggested_fix_direction": {"type": "string"}, } ) review_schema = object_schema( {"findings": {"type": "array", "items": finding_schema}} ) fix_schema = object_schema( { "artifact": {"type": "string"}, "iteration": {"type": "integer"}, "changes_made": string_array(), "unresolved_items": string_array(), "updated_artifact_path": {"type": "string"}, } ) validation_case_schema = object_schema( { "name": {"type": "string"}, "passed": {"type": "boolean"}, "severity": {"type": "string"}, "evidence": {"type": "string"}, "feedback": {"type": "string"}, } ) validation_schema = object_schema( { "overall_passed": {"type": "boolean"}, "cases": {"type": "array", "items": validation_case_schema}, "remaining_delta": string_array(), } ) ``` レビューフェーズはアーティファクトを読み込み、構造化された発見結果を返します。ただし、検証を実行したりファイルを編集したりすることはありません。この分離により、最初のステップに集中し、変更を加える前に潜在的な問題を特定することが可能になります。 レビュー用のプロンプトは `codex exec` に JSON スキーマと共に送信されます。スキーマによって結果が機械可読形式となり、後のセルで前回の回答から文章をスクレイピングするのではなく、発見事項を直接修復用プロンプトに渡すことができます。 ``` def notebook_text(path: Path, max_chars: int = 7000) -> str: chunks = [] for index, cell in enumerate(read_notebook(path)["cells"]): source = "".join(cell.get("source", [])) chunks.append(f"cell {index} ({cell['cell_type']})\n{source}") text = "\n\n".join(chunks) if len(text) <= max_chars: return text return text[:max_chars] + "\n\n[truncated for prompt size]" def run_command(command: str, *, stdin: str | None = None, cwd: Path | None = None, timeout: int | None = None): cwd = Path.cwd() if cwd is None else cwd return subprocess.run( shlex.split(command), input=stdin, cwd=cwd, capture_output=True, text=True, timeout=timeout, check=False, ) def run_codex_json(prompt: str, schema: dict[str, Any], run_dir: Path) -> dict[str, Any]: run_dir.mkdir(parents=True, exist_ok=True) prompt_file = run_dir / "prompt.txt" schema_file = run_dir / "schema.json" answer_file = run_dir / "answer.json" prompt_file.write_text(prompt, encoding="utf-8") schema_file.write_text(json.dumps(schema, indent=2), encoding="utf-8") command = f""" {CODEX_CLI} exec --model {MODEL} --sandbox workspace-write --ask-for-approval never --config model_reasoning_effort={REPAIR_REASONING_EFFORT} --output-schema {schema_file} --output-last-message {answer_file} - """ result = run_command(command, stdin=prompt) (run_dir / "stdout.txt").write_text(result.stdout, encoding="utf-8") (run_dir / "stderr.txt").write_text(result.stderr, encoding="utf-8") if result.returncode != 0: raise RuntimeError(f"Codex exited with {result.returncode}. See {run_dir / 'stderr.txt'}.") return json.loads(answer_file.read_text(encoding="utf-8")) def review_notebook(path: Path, run_dir: Path) -> list[dict[str, Any]]: prompt = "\n".join( [ "You are reviewing a public OpenAI Cookbook notebook before publication.", f"Artifact: {path.name}", "Find issues that would make the notebook stale, hard to run, or confusing for a developer reader.", "Do not execute the notebook or edit files.", "Use concise issue_type labels such as stale_model, deprecated_api, setup_gap, runtime_risk, or clarity_issue.", f"Business rules: {json.dumps(business_rules)}", "Base findings only on the notebook content below.", "Keep the findings focused; three strong findings are better than a long list.", "", notebook_text(path), ] ) return run_codex_json(prompt, review_schema, run_dir)["findings"] ``` ``` def run_initial_review(path: Path) -> tuple[str, list[dict[str, Any]]]: return path.name, review_notebook(path, RUNS_DIR / "initial_review" / path.stem) with concurrent.futures.ThreadPoolExecutor(max_workers=min(3, len(NOTEBOOKS))) as executor: initial_reviews = dict(executor.map(run_initial_review, NOTEBOOKS)) initial_reviews ``` 修復フェーズでは、現在の成果物、レビュー結果、ビジネスルール、および前回のパスからの検証フィードバックを取得します。ループが学習するにつれて、プロンプトはより具体的になります。 コデックスは反復ディレクトリ内のコピーを編集し、変更箇所の要約を返します。このループでは編集が成功したと仮定せず、次のステップで検証が行われます。 ``` def repair_prompt(path: Path, updated_path: Path, findings: list[dict[str, Any]], remaining_delta: list[str], iteration: int) -> str: repair_story = case_metadata(path).get("repair_story", {}) return "\n".join( [ "You are repairing a copy of a public OpenAI Cookbook notebook.", f"Source notebook: {path}", f"Editable copy: {updated_path}", f"Iteration: {iteration}", "Make the smallest useful edits that address the review findings and validation delta.", "Preserve the notebook's teaching flow and original purpose.", "Keep sample repairs self-contained unless the notebook explicitly teaches external setup.", "For staged examples, focus on the most important remaining issue for this pass instead of rewriting everything at once.", "Edit only the editable copy. Do not claim the notebook passes validation.", f"Repair depth: {json.dumps(repair_story, indent=2)}", f"Business rules: {json.dumps(business_rules, indent=2)}", f"Review findings: {json.dumps(findings, indent=2)}", f"Remaining validation delta: {json.dumps(remaining_delta, indent=2)}", ] ) def repair_notebook(path: Path, iteration: int, findings: list[dict[str, Any]], remaining_delta: list[str], case_dir: Path) -> dict[str, Any]: updated_path = case_dir / "updated.ipynb" updated_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(path, updated_path) prompt = repair_prompt(path, updated_path, findings, remaining_delta, iteration) return run_codex_json(prompt, fix_schema, case_dir / "repair") ``` 検証は小さな評価プロセスのようなものです。望む動作を定義し、関連するチェックを実行して、結果を基準に対して採点者に評価させます。 ドキュメント作成の例では、まず実行を行います。多くのノートブックの問題は実行時に初めて顕在化します。例えば、インポートの欠落、古いファイルパス、旧 API 応答に依存するセル、あるいは著者には明確でも新規読者には理解しにくいセットアップガイダンスなどが挙げられます。 検証で失敗した場合、その失敗は次の修復工程における根拠となります。これにより、次回の修復は単なる差分の見た目ではなく、実際に観測された動作に基づいて行われます。 ``` VALIDATION_CASES = [ { "name": "api_modernization", "question": "Does the notebook avoid stale OpenAI API patterns, legacy function-calling syntax, and outdated model names?", }, { "name": "setup_reproducibility", "question": "Could a reader run the notebook from a fresh environment without hidden manual steps?", }, { "name": "artifact_integrity", "question": "Did the update preserve the notebook's teaching flow and avoid deleting substantive cells?", }, ] def short_output(value: Any, limit: int = 1200) -> str: if value is None: return "" if isinstance(value, bytes): value = value.decode("utf-8", errors="replace") return str(value)[-limit:] def execute_notebook(path: Path) -> dict[str, Any]: code_cells = sum(cell["cell_type"] == "code" for cell in read_notebook(path)["cells"]) command = f"jupyter nbconvert --to notebook --execute --inplace {path.name}" try: result = run_command( command, cwd=path.parent, timeout=int(os.getenv("SAMPLE_NOTEBOOK_TIMEOUT_SECONDS", "300")), ) except FileNotFoundError: return { "status": "failed", "executed_code_cells": 0, "error": "Jupyter or nbconvert is not installed or is not available on PATH.", "summary": "Install Jupyter with nbconvert before running the validation loop.", } except subprocess.TimeoutExpired as exc: return { "status": "failed", "executed_code_cells": 0, "error": f"Notebook execution timed out after {exc.timeout} seconds.", "summary": short_output(exc.stderr or exc.stdout), } output = result.stderr or result.stdout return { "status": "passed" if result.returncode == 0 else "failed", "executed_code_cells": code_cells if result.returncode == 0 else 0, "error": "" if result.returncode == 0 else f"Notebook execution exited with code {result.returncode}.", "summary": short_output(output), } def validation_prompt(updated_path: Path, before_path: Path, execution: dict[str, Any], iteration: int) -> str: repair_story = case_metadata(before_path).get("repair_story", {}) return "\n".join( [ "You are judging a repaired OpenAI Cookbook notebook.", f"Iteration: {iteration}", "Score each validation case independently and give concise feedback for the next repair pass.", "Set overall_passed to false when execution failed or any case has a material issue.", "When execution failed, include the failure in remaining_delta so the next repair pass can address it.", "Use the business rules as the source of truth for current model names and API targets.", "Do not mark the preferred embedding model or preferred chat model as stale.", "For local examples, do not require extra services or package installs when the notebook says it is intentionally self-contained.", f"Repair depth: {json.dumps(repair_story, indent=2)}", f"Business rules: {json.dumps(business_rules, indent=2)}", f"Validation cases: {json.dumps(VALIDATION_CASES, indent=2)}", f"Execution evidence: {json.dumps(execution, indent=2)}", f"Original cell count: {len(read_notebook(before_path)['cells'])}", f"Updated cell count: {len(read_notebook(updated_path)['cells'])}", "", notebook_text(updated_path), ] ) def staged_delta(before_path: Path, iteration: int) -> list[str]: repair_story = case_metadata(before_path).get("repair_story", {}) target = int(repair_story.get("target_iteration") or 1) if iteration >= target: return [] depth = repair_story.get("repair_depth", "This case is intentionally staged across multiple repair passes.") return [f"Continue to iteration {iteration + 1}: {depth}"] def evaluate_notebook(updated_path: Path, before_path: Path, run_dir: Path, iteration: int) -> dict[str, Any]: execution = execute_notebook(updated_path) judged = run_codex_json(validation_prompt(updated_path, before_path, execution, iteration), validation_schema, run_dir) failed_cases = [case for case in judged["cases"] if not case["passed"]] execution_delta = [] if execution["status"] != "passed": execution_delta.append(f"Execution failed: {execution.get('error') or execution.get('summary')}") stage_delta = staged_delta(before_path, iteration) return { "passed": judged["overall_passed"] and execution["status"] == "passed" and not stage_delta, "execution_status": execution["status"], "executed_code_cells": execution["executed_code_cells"], "execution_summary": execution["summary"], "findings": failed_cases, "remaining_delta": execution_delta + stage_delta + judged["remaining_delta"], } ``` 各イテレーションでは、`record.json` ファイルが書き出されます。また、この例では `CODEX_REPAIR_RUNS_DIR/iteration_N/ /` 配下に修復されたノートブックも生成されます。 もし `CODEX_REPAIR_RUNS_DIR` を設定しない場合、ノートブックはシステムの一時ディレクトリに書き込まれるため、通常のレポジトリのチェックアウト状態を清潔に保つことができます。 これらのファイルは監査証跡です。レビューで何が見つかり、Codex が何を修正し、実行が成功したか、そしてフィードバックが次のイテレーションにどう引き継がれたかが確認できます。 `record.json` ファイルは、1 回のループ試行に対する領収書のようなものです。各フェーズ間の受け渡し情報をここに一元管理します。 ``` { "review": [{"issue_type": "deprecated_api", "severity": "high"}], "repair": { "changes_made": ["Updated the notebook to use the current API pattern."], "updated_artifact_path": "/tmp/codex_iterative_repair_loop_outputs/iteration_1/sample/updated.ipynb" }, "validation": { "passed": false, "remaining_delta": ["One setup instruction is still unclear."] } } ``` このコンパクトな記録があれば、開発者はノートブックの差分やターミナルログから実行全体を再構築することなく、ループの内容を確認できます。 ``` def save_json(payload: Any, path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") def iteration_dir(number: int) -> Path: path = RUNS_DIR / f"iteration_{number}" path.mkdir(parents=True, exist_ok=True) return path ``` 各ノートブックケースは独立しているため、並列処理して実行します。これによりデモの速度を維持しつつ、すべてのサンプルで同一のレビュー・修復・検証フローを実現できます。 1 回目の反復では、以前のレビューセルで見つかった初期の指摘事項を再利用します。このパスが完了したら、返されたブール値を確認してください。合格したケースはそこで終了し、不合格だったケースについては検証フィードバックを次のパスに引き継ぎます。 current_notebooks = {path.name: path for path in NOTEBOOKS} history: dict[int, dict[str, Any]] = {} def review_findings_for(original: Path, current_path: Path, case_dir: Path, previous_results: dict[str, Any] | None) -> list[dict[str, Any]]: if previous_results is None: return initial_reviews[original.name] return review_notebook(current_path, case_dir / "review") def run_case(number: int, original: Path, run_dir: Path, previous_results: dict[str, Any] | None) -> tuple[str, dict[str, Any], Path]: name = original.name case_dir = run_dir / original.stem current_path = current_notebooks[name] findings = review_findings_for(original, current_path, case_dir, previous_results) delta = [] if previous_resu

自己修復型エージェントループの構築（読了時間：39 分）

背景や根拠まで確認しますか？

調べる

選ぶ

サイト

背景や根拠まで確認しますか？

ニュースの次に確認する

調べる

選ぶ

サイト