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

!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.")

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.'}]

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(),
    }
)

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")

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"],
    }

{
  "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