OpenAI のデータベース変更分析（28 分読）

This cookbook walks through an end-to-end AI-assisted database change workflow using the OpenAI Agents SDK.

It demonstrates how OpenAI’s tooling ecosystem can be applied to orchestrate complex, data-intensive workflows across modern enterprise infrastructures. While the current implementation focuses on a retail-oriented schema change and impact-analysis use case, the underlying architectural patterns are domain-agnostic and extensible. The same workflow design can be adapted across industries such as manufacturing, pharmaceuticals, healthcare, logistics, finance, and supply chain operations — wherever structured data workflows, operational reasoning, retrieval-augmented analysis, and automated validation are required.

The running example is a retail loyalty-tier change, but the same pattern applies to many database-change requests where teams need traceable impact analysis and reviewable implementation output.

The workflow starts from a natural-language database change request, converts it into structured JSON, optionally grounds impact analysis with PDF-based File Search context, generates a safe rollout plan, drafts SQL across data platform layers, validates the output with deterministic guardrails, saves a reusable artifact, and optionally evaluates the flow with Promptfoo.

The notebook is intentionally self-contained: all core workflow logic, prompts, guardrails, artifact generation, and eval runtime files are created from notebook cells.

Schema changes are deceptively simple. A request like “add a nullable column and backfill it” can affect landing tables, staging models, dimensional tables, marts, reporting logic, lineage assumptions, validation checks, rollback procedures, and release sequencing.

The examples use retail customer data because the dependencies are easy to see, but the same kinds of handoffs show up in many analytics and platform teams.

This cookbook demonstrates a practical pattern for using agents as a change-analysis and implementation assistant for database engineering work. Instead of asking one model to produce a final SQL script directly, the workflow breaks the task into explicit stages:

• Parse the natural-language request into structured JSON.

• Analyze impacted objects and operational risks.

• Create a rollout plan with prechecks, postchecks, and rollback guidance.

• Generate SQL across platform layers.

• Run deterministic sanity checks.

• Save a machine-readable artifact.

• Optionally run Promptfoo evals against the current flow.

The result is not just a generated SQL script. It is an auditable bundle containing the interpreted change request, impact analysis, plan, SQL, validation results, optional RAG evidence summaries, and eval outputs.

Database change requests often move through several handoffs: product owners describe the need, data engineers interpret it, platform teams assess risk, analytics engineers propagate the field downstream, and reviewers check whether the change is safe. Important context can be lost at each step.

SchemaFlow addresses this by turning a free-form change request into a structured, inspectable workflow.

This matters because database changes can create hidden failure modes:

• A column added to ODS may not be propagated into staging, core, or marts.

• A nullable field may accidentally be generated as NOT NULL.

• Backfill logic may be omitted even though the request asks for historical population.

• Index requirements may be missed.

• Downstream reporting dependencies may be unknown unless reference documentation is consulted.

• Generated SQL may look plausible but fail basic consistency checks.

This cookbook shows a pattern for reducing those risks with staged agent reasoning, typed outputs, optional retrieval context, deterministic guardrails, saved artifacts, and repeatable evals.

• Structured interpretation – Converts natural-language database requests into a normalized change_json contract.

• Separation of responsibilities – Uses specialized agents for parse, impact analysis, rollout planning, and SQL generation.

• Optional RAG grounding – Lets the impact-analysis agent use File Search over an uploaded PDF, such as an IFD, schema spec, or lineage document.

• Typed stage outputs – Uses Pydantic models and Agents SDK output schemas for parse, impact, and plan stages.

• Guardrail-first workflow – Adds deterministic checks between stages so obvious failures are caught before downstream steps consume bad state.

• Traceability – Emits OpenAI Agents SDK traces and spans for agent runs, guardrails, artifact generation, and eval execution.

• Portable artifacts – Saves the final workflow bundle as JSON under artifacts/notebook_runs/.

• Eval-ready design – Generates Promptfoo provider, assertion, config, and result files from the live notebook state.

• No database side effects – Produces draft SQL and validation output without executing against a live database.

By the end of this notebook, you will have a working SchemaFlow pipeline that produces:

• A parsed database change request:

title

• domain

• target schema

• target table

• normalized operations

• notes

• An impact-analysis report:

impacted tables, columns, indexes, views, or relationships

• risks

• assumptions

• optional File Search evidence summaries

• A rollout plan:

implementation steps

• prechecks

• postchecks

• rollback actions

• A draft SQL script with four required sections:

LANDING (ODS)

• STAGING (STG)

• CORE (DIM/FACT/VIEW)

• MARTS (SERVING)

• A validation result:

expected table checks

• expected column checks

• required keyword checks such as ALTER TABLE, UPDATE, or CREATE INDEX

• A saved JSON artifact:

change request

• impact analysis

• plan

• SQL

• validation

• optional RAG metadata

• A Promptfoo eval harness:

Python provider

• Python assertion file

• generated Promptfoo config

• parse-only eval case

• full-flow eval case

• timestamped JSON and HTML eval reports

This cookbook focuses on a common enterprise data-engineering scenario: a stakeholder requests a database schema change in natural language, and the data team needs to turn that request into an implementation-ready plan.

Here, the retail domain is just a concrete way to make the workflow tangible. The same staged approach can be adapted to other source systems, data products, and review processes.

The default request in this notebook is:

Add LOYALTY_TIER VARCHAR(20) to ODS.ODS_CUSTOMER_PROFILE as nullable.
Backfill from CORE.DIM_CUSTOMER on CUSTOMER_ID where IS_CURRENT=true.
Add a non-unique index on (CUSTOMER_ID, LOYALTY_TIER).

A human data engineer would typically need to answer several questions before writing production SQL:

• What table and schema are being changed?

• What exact column, type, and nullability were requested?

• Is historical backfill required?

• Does the request imply an index?

• Which downstream layers need the field propagated?

• What risks should reviewers look for?

• What checks should be run before and after deployment?

• What rollback steps are reasonable?

• Does the generated SQL include the required elements?

SchemaFlow implements this as a staged agent workflow. Each stage creates a typed intermediate output that the next stage consumes. Deterministic checks then validate the outputs before the notebook saves the final bundle and optionally runs evals.

At a high level, SchemaFlow follows this sequence:

The notebook is organized so readers can run the core workflow first and then decide whether they want to run the optional Promptfoo evaluation section.

Table of Contents

Conceptual Guide

• Overview

• Why This Matters

• Key Benefits

• What You’ll Build

• Introduction: Use Case and Solution

• Workflow Overview

• Architecture - Design Patterns

• System Design

• Execution Workflow

Notebook Implementation

• Environment Setup

• Input

• Optional PDF RAG Context

• Stages 1-2: Parse Change Request + Impact Analysis

Impact Dashboard Preview

• Stages 1-2 Output Guardrails

• Stages 3-4: Execution Plan + SQL Generation

Stages 3-4 Output Guardrails

• Stage 5: Lightweight SQL Sanity Checks

• Final Bundle

• Save Artifact

Post-Artifact Generation Sanity Check

• Optional Cleanup

Pre-Promptfoo Checks / Guardrails

• Evaluate the Flow with Promptfoo

Promptfoo Runtime Directory Setup

• Node.js and npm Runtime Check

• Publish SchemaFlow Core Runtime

• Promptfoo Provider Runtime

• Promptfoo Assertion Runtime

• Build Promptfoo Test Cases and Config

• Run Promptfoo Eval

• Review Latest Promptfoo Results

Reference

• Notes, Assumptions, and Extension Points

SchemaFlow uses a staged, contract-driven agent architecture. The goal is to avoid treating the model as a single black-box SQL generator. Instead, each stage has a narrow responsibility and produces an output that can be inspected, validated, traced, and reused.

1. Agent Specialization

Each agent performs one primary task:

This specialization makes the workflow easier to debug. If SQL is missing a column, you can inspect whether the issue started in parsing, impact analysis, planning, or SQL generation.

2. Typed Output Contracts

The notebook defines Pydantic models for the structured stages:

• ChangeRequestModel

• ImpactModel

• PlanModel

Those models are wrapped with AgentOutputSchema so the Agents SDK knows the expected output shape. The workflow also normalizes outputs after model calls to ensure expected keys exist before downstream stages run.

3. Retrieval-Augmented Impact Analysis

The PDF RAG section is optional. When PDF_PATH is set, the notebook:

• Creates an OpenAI vector store.

• Uploads the PDF.

• Lets OpenAI parse, chunk, embed, and index it.

• Gives the Impact Agent a FileSearchTool.

• Captures a summary of returned File Search results.

This is useful when the change request needs grounding in an IFD, schema document, lineage file, data contract, or architecture reference.

4. Guardrail Gates Between Stages

The notebook adds deterministic checks after major stages:

• Stages 1-2 guardrails validate parse and impact outputs.

• Stages 3-4 guardrails validate plan completeness, data type propagation, and nullability handling.

• Stage 5 SQL checks validate expected table, column, and SQL keyword presence.

• Post-artifact checks verify the saved JSON artifact exists and round-trips.

• Pre-Promptfoo checks verify the notebook state is ready for evals.

These checks do not replace human review, but they catch common silent failures early.

5. Artifact-Centered Execution

The final bundle is the main workflow artifact. It captures the state needed to review or debug the run:

bundle = {
  "summary": ...,
  "rag": ...,
  "change_json": ...,
  "impact_json": ...,
  "plan": ...,
  "sql": ...,
  "validation": ...
}

The notebook saves this bundle under artifacts/notebook_runs/.

6. Eval Runtime Generated from Notebook State

Promptfoo runs in a separate process, so it cannot directly read variables from the active notebook kernel. To solve this, Section 10 writes a small reusable Python module and Promptfoo runtime files from the current notebook state.

This ensures that prompt edits, CHANGE_TEXT edits, and model configuration changes are reflected when the eval files are regenerated.

Component Architecture

Primary Runtime Objects

Important Boundary

SchemaFlow generates draft implementation artifacts. It does not execute SQL against a database, apply migrations, open pull requests, or modify production systems.

Run the notebook in order.

Core Workflow

• Environment Setup

Imports dependencies.

• Verifies the OpenAI Agents SDK version.

• Reads OPENAI_API_KEY.

• Configures tracing and model selection.

• Input

Defines CHANGE_TEXT.

• This is the only required business input for the core workflow.

• Optional PDF RAG Context

Leave PDF_PATH = None to run without retrieval.

• Set PDF_PATH to a local PDF to enable File Search context for impact analysis.

• Stages 1-2

Parse the change request.

• Analyze impact.

• Optionally use File Search during impact analysis.

• Stages 1-2 Guardrails

Confirm parse output is well-formed.

• Confirm impact output includes the target.

• Confirm impacted objects contain required fields.

• Stages 3-4

Generate an execution plan.

• Generate SQL across landing, staging, core, and mart layers.

• Stages 3-4 Guardrails

Confirm plan sections are populated.

• Confirm data type propagation.

• Confirm nullability behavior matches the request.

• Stage 5 SQL Sanity Checks

Check for empty SQL.

• Check expected target table and columns.

• Check required SQL actions implied by the request.

• Final Bundle and Artifact

Assemble the full output bundle.

• Save it as JSON.

• Verify the artifact round-trips successfully.

Optional Eval Workflow

• Pre-Promptfoo Checks

Confirm the notebook state is ready for evals.

• Promptfoo Runtime Generation

Create a reusable SchemaFlow core module.

• Write a Promptfoo provider.

• Write a Promptfoo assertion file.

• Generate Promptfoo test cases and config.

• Promptfoo Eval Execution

Run parse-only and full-flow evals.

• Save timestamped JSON and HTML reports.

• Refresh schemaflow_cookbook_eval_latest.* aliases.

This section prepares the runtime for the SchemaFlow workflow.

The setup cell does the following:

• Imports standard Python utilities used throughout the notebook.

• Imports the OpenAI client.

• Imports the OpenAI Agents SDK primitives:

Agent

• Runner

• RunConfig

• AgentOutputSchema

• FileSearchTool

• tracing and span helpers

• Verifies that the installed openai-agents package meets the minimum required version.

• Reads OPENAI_API_KEY from the environment or prompts for it.

• Sets the model with OPENAI_MODEL, defaulting to gpt-5.5.

• Creates a trace group ID so all related agent runs and guardrail spans can be grouped together.

The workflow intentionally enables sensitive trace payloads for this demo so prompts, outputs, eval bundles, and tool data are visible in traces. For production usage, review this setting before handling private data.

%pip install --quiet -U "openai" "openai-agents>=0.17.0"

import os
import json
import re
import uuid
from datetime import datetime, timezone
from getpass import getpass
from importlib.metadata import PackageNotFoundError, version

try:
    from openai import OpenAI
except Exception as e:
    raise RuntimeError("Install dependency first: pip install -U openai") from e

MIN_AGENTS_SDK_VERSION = "0.17.0"
try:
    from agents import (
        Agent,
        AgentOutputSchema,
        FileSearchTool,
        Runner,
        RunConfig,
        custom_span,
        flush_traces,
        function_span,
        guardrail_span,
        trace,
    )
except Exception as e:
    raise RuntimeError(
        'Install or upgrade the OpenAI Agents SDK first: pip install -U "openai-agents>=0.17.0"'
    ) from e

def _version_tuple(value):
    match = re.match(r"^(\d+)\.(\d+)\.(\d+)", str(value or ""))
    return tuple(int(part) for part in match.groups()) if match else (0, 0, 0)

try:
    AGENTS_SDK_VERSION = version("openai-agents")
except PackageNotFoundError as e:
    raise RuntimeError('Install the OpenAI Agents SDK first: pip install -U "openai-agents>=0.17.0"') from e

if _version_tuple(AGENTS_SDK_VERSION) < _version_tuple(MIN_AGENTS_SDK_VERSION):
    raise RuntimeError(
        f'OpenAI Agents SDK {MIN_AGENTS_SDK_VERSION}+ is required; found {AGENTS_SDK_VERSION}. '
        'Upgrade with: pip install -U "openai-agents>=0.17.0"'
    )

def _clean_openai_api_key(value):
    key = (value or "").strip()
    if not key:
        raise RuntimeError("OPENAI_API_KEY is required.")
    return key

if not os.getenv("OPENAI_API_KEY", "").strip():
    os.environ["OPENAI_API_KEY"] = getpass("Enter your OpenAI API key: ")
os.environ["OPENAI_API_KEY"] = _clean_openai_api_key(os.getenv("OPENAI_API_KEY"))
OPENAI_ORG_ID = os.getenv("OPENAI_ORG_ID", "").strip()
if OPENAI_ORG_ID:
    os.environ["OPENAI_ORG_ID"] = OPENAI_ORG_ID

MODEL = os.getenv("OPENAI_MODEL", "gpt-5.5")
TRACE_INCLUDE_SENSITIVE_DATA = os.getenv("OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA", "false").lower() in {"1", "true", "yes", "on"}
os.environ["OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA"] = "true" if TRACE_INCLUDE_SENSITIVE_DATA else "false"
SCHEMAFLOW_TRACE_GROUP_ID = os.getenv("SCHEMAFLOW_TRACE_GROUP_ID") or (
    "schemaflow-cookbook-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex[:8]
)
os.environ["SCHEMAFLOW_TRACE_GROUP_ID"] = SCHEMAFLOW_TRACE_GROUP_ID
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
print("Using model:", MODEL)
print("OpenAI Agents SDK:", AGENTS_SDK_VERSION)
print("OpenAI organization:", os.getenv("OPENAI_ORG_ID") or "(default for API key)")
print("Trace group:", SCHEMAFLOW_TRACE_GROUP_ID)
print("Trace payloads include prompts/outputs:", TRACE_INCLUDE_SENSITIVE_DATA)

from concurrent.futures import ThreadPoolExecutor

from pydantic import BaseModel, ConfigDict, Field

class SchemaFlowBaseModel(BaseModel):

model_config = ConfigDict(extra="allow")

class OperationModel(SchemaFlowBaseModel):

op: str

details: dict = Field(default_factory=dict)

class ChangeRequestModel(SchemaFlowBaseModel):

domain: str | None = None

target_schema: str | None = None

target_table: str | None = None

operations: list[OperationModel] = Field(default_factory=list)

notes: list = Field(default_factory=list)

class ImpactObjectModel(SchemaFlowBaseModel):

type: str

name: str

reason: str

source: str

class ImpactModel(SchemaFlowBaseModel):

impacted_objects: list[ImpactObjectModel] = Field(default_factory=list)

risks: list[str] = Field(default_factory=list)

assumptions: list[str] = Field(default_factory=list)

class PlanStepModel(SchemaFlowBaseModel):

id: str

description: str

class PlanModel(SchemaFlowBaseModel):

plan_steps: list[PlanStepModel] = Field(default_factory=list)

prechecks: list[str] = Field(default_factory=list)

postchecks: list[str] = Field(default_factory=list)

rollback: list[str] = Field(default_factory=list)

CHANGE_OUTPUT_SCHEMA = AgentOutputSchema(ChangeRequestModel, strict_json_schema=False)

IMPACT_OUTPUT_SCHEMA = AgentOutputSchema(ImpactModel, strict_json_schema=False)

PLAN_OUTPUT_SCHEMA = AgentOutputSchema(PlanModel, strict_json_schema=False)

def _parse_json_text(text: str):

text = (text or "{}").strip()

if text.startswith("`"</span><span>):</span></span>

<span><span> text </span><span>=</span><span> re.sub(</span><span>r</span><span>"</span><span>^</span><span>`(?:json)?\s*", "", text)

text = re.sub(r"\s*`$", "", text).strip()

try:

return json.loads(text)

except json.JSONDecodeError:

match = re.search(r"\{.*\}", text, flags=re.DOTALL)

if not match:

raise

return json.loads(match.group(0))

def _model_dump(value):

if value is None or isinstance(value, (str, int, float, bool, bytes)):

return value

if isinstance(value, type):

return value

if hasattr(value, "model_dump"):

try:

return value.model_dump()

except TypeError:

pass

if hasattr(value, "to_dict"):

try:

return value.to_dict()

except TypeError:

pass

if hasattr(value, "__dict__"):

try:

return {k: v for k, v in vars(value).items() if not k.startswith("_")}

except TypeError:

pass

return value

def _agent_output_to_json(value):

value = _model_dump(value)

if isinstance(value, dict):

return value

if isinstance(value, str):

return _parse_json_text(value)

return json.loads(json.dumps(value, default=str))

def _agent_output_to_text(value):

value = _model_dump(value)

if isinstance(value, str):

return value.strip()

return json.dumps(value, ensure_ascii=False)

def _trace_metadata(metadata: dict | None = None):

cleaned = {}

for key, value in (metadata or {}).items():

if value is None:

cleaned[str(key)] = ""

elif isinstance(value, bool):

cleaned[str(key)] = "true" if value else "false"

elif isinstance(value, (dict, list, tuple, set)):

cleaned[str(key)] = json.dumps(value, ensure_ascii=False, default=str)

else:

cleaned[str(key)] = str(value)

return cleaned

def _schemaflow_run_config(workflow_name: str, metadata: dict | None = None):

return RunConfig(

workflow_name=workflow_name,

group_id=SCHEMAFLOW_TRACE_GROUP_ID,

trace_include_sensitive_data=TRACE_INCLUDE_SENSITIVE_DATA,

trace_metadata=_trace_metadata({"notebook": "schemaflow_cookbook", **(metadata or {})}),

)

def _runner_run_sync(agent, prompt: str, *, workflow_name: str, metadata: dict | None = None, max_turns: int = 4):

kwargs = {"run_config": _schemaflow_run_config(workflow_name, metadata), "max_turns": max_turns}

try:

return Runner.run_sync(agent, prompt, **kwargs)

except RuntimeError as exc:

if "event loop" not in str(exc).lower():

raise

with ThreadPoolExecutor(max_workers=1) as pool:

return pool.submit(lambda: Runner.run_sync(agent, prompt, **kwargs)).result()

def run_schemaflow_json_agent(*, name, instructions, prompt, output_schema, model=MODEL, tools=None, workflow_name=None, metadata=None):

agent = Agent(name=name, instructions=instructions, model=model, output_type=output_schema, tools=tools or [])

result = _runner_run_sync(agent, prompt, workflow_name=workflow_name or name, metadata={"agent": name, **(metadata or {})})

return _agent_output_to_json(result.final_output), result

def run_schemaflow_text_agent(*, name, instructions, prompt, model=MODEL, tools=None, workflow_name=None, metadata=None):

agent = Agent(name=name, instructions=instructions, model=model, tools=tools or [])

result = _runner_run_sync(agent, prompt, workflow_name=workflow_name or name, metadata={"agent": name, **(metadata or {})})

return _agent_output_to_text(result.final_output), result

def _collect_file_search_results(value):

results = []

seen = set()

def visit(node):

if node is None or isinstance(node, (str, int, float, bool, bytes)):

return

if isinstance(node, type) or callable(node):

return

node_id = id(node)

if node_id in seen:

return

seen.add(node_id)

node = _model_dump(node)

if node is None or isinstance(node, (str, int, float, bool, bytes)):

return

if isinstance(node, type) or callable(node):

return

if isinstance(node, dict):

if node.get("type") == "file_search_call":

for result in node.get("results", []) or []:

result = _model_dump(result)

if isinstance(result, dict):

text = result.get("text") or result.get("content") or ""

if isinstance(text, list):

text = "\n".join(str(x) for x in text)

results.append({"file_id": result.get("file_id"), "filename": result.get("filename") or result.get("file_name") or result.get("title"), "score": result.get("score"), "text_preview": str(text)[:1200]})

for child in node.values():

visit(child)

elif isinstance(node, (list, tuple, set)):

for child in node:

visit(child)

visit(val