OpenAI's Database Change Analysis (28 minute read)｜OpenAI のデータベース変更分析（28 分読） | AIニュース最前線

このクックブックでは、OpenAI Agents SDK を用いた AI 支援型データベース変更ワークフローの全体像を解説します。 これは、OpenAI のツールエコシステムが、現代のエンタープライズインフラストラクチャにわたる複雑でデータ集約型のワークフローをオーケストレーションするためにどのように適用できるかを示すものです。現在の実装は小売業向けのスキーマ変更と影響分析ユースケースに焦点を当てていますが、基盤となるアーキテクチャパターンはドメインに依存せず拡張可能です。このワークフロー設計は、構造化データワークフロー、運用推論、検索強化分析、自動検証が必要とされる製造、製薬、医療、物流、金融、サプライチェーン運営など、あらゆる業界に適応して利用できます。 実行例として取り上げられているのは小売業のロイヤリティティア変更ですが、このパターンは、チームが追跡可能な影響分析とレビュー可能な実装出力を必要とする多くのデータベース変更リクエストにも適用可能です。 ワークフローは自然言語によるデータベース変更要求から始まり、これを構造化 JSON 形式に変換します。必要に応じて PDF ベースのファイル検索コンテキストを用いて影響分析に根拠を持たせ、安全なロールアウトプランを生成し、データプラットフォーム層全体にわたる SQL を草案化します。その後、決定論的なガードレールを用いて出力を検証し、再利用可能なアーティファクトとして保存します。必要に応じて Promptfoo を用いたフロー評価も実施可能です。 このノートブックは意図的に自己完結型となっています：ワークフローの核心となるロジック、プロンプト、ガードレール、アーティファクト生成、および評価ランタイムファイルはすべてノートブックセルから作成されます。 スキーマ変更は、一見すると単純に見えます。「null 許容の列を追加してバックフィルする」といったリクエストは、着陸テーブル、ステージングモデル、次元テーブル、マート、レポートロジック、リンケージの前提条件、検証チェック、ロールバック手順、リリースシーケンシングに影響を及ぼす可能性があります。 例では小売顧客データを使用していますが、依存関係が視覚化しやすいためです。しかし、同様の引き継ぎは多くの分析チームやプラットフォームチームで発生します。 このクックブックは、データベースエンジニアリング作業においてエージェントを**変更分析および実装アシスタント**として活用するための実践的なパターンを示しています。1 つのモデルに最終的な SQL スクリプトを直接生成させるのではなく、ワークフローはタスクを明示的なステージに分割します： - 自然言語のリクエストを構造化された JSON に解析する。 - 影響を受けるオブジェクトと運用上のリスクを分析する。 - プリチェック、ポストチェック、ロールバックガイダンスを含む展開計画を作成する。 - プラットフォーム層全体で SQL を生成する。 - 決定論的な健全性チェックを実行する。 - マシン可読なアーティファクトを保存する。 - 必要に応じて現在のフローに対して Promptfoo 評価（evals）を実行する。 結果は単に生成された SQL スクリプトではありません。解釈された変更リクエスト、影響分析、計画、SQL、検証結果、オプションの RAG エビデンスサマリー、および評価出力を含む監査可能なバンドルです。 データベース変更要求は、多くの場合複数の引き継ぎプロセスを経ます。プロダクトオーナーが要件を説明し、データエンジニアがそれを解釈し、プラットフォームチームがリスクを評価し、アナリティクスエンジニアがフィールドを下流へ伝播させ、レビュー担当者が変更の安全性を確認します。各ステップで重要な文脈が失われる可能性があります。 SchemaFlow は、自由形式の変更要求を構造化され、検証可能なワークフローに変換することでこの課題に対処します。 これは重要です。なぜならデータベース変更は隠れた障害モードを引き起こす可能性があるからです： - ODS に追加された列が、ステージング、コア、またはマートへ伝播されない場合があります。 - NULL 許容フィールドが誤って NOT NULL として生成される可能性があります。 - 要求で歴史的データの充満を求めているにもかかわらず、バックフィルロジックが省略されることがあります。 - インデックス要件が見落とされる可能性があります。 - リファレンスドキュメントを参照しない限り、下流のレポート依存関係が不明な場合があります。 - 生成された SQL は妥当に見えても、基本的な整合性チェックで失敗する可能性があります。 このクックブックは、段階的なエージェント推論、型付き出力、オプションのリトリーバルコンテキスト、決定論的ガードレール、保存されたアーティファクト、および反復可能な評価を用いてこれらのリスクを低減するためのパターンを示しています。 - 構造化解釈 – 自然言語のデータベース要求を変換し、正規化された change_json 契約に変換します。 - 責任の分離 – パース、影響分析、ロールアウト計画、SQL 生成のために専門的なエージェントを使用します。 - オプションな RAG グラウンディング – 影響分析エージェントが、アップロードされた PDF（IFD、スキーマ仕様、またはリンケージドキュメントなど）に対して File Search を使用できるようにします。 - Typed stage outputs – Pydantic モデルと Agents SDK の出力スキーマを使用して、解析（parse）、影響分析（impact）、計画（plan）の各ステージで構造化された出力を取得します。 - Guardrail-first workflow – 下流のステップが不良状態を消費する前に明らかな失敗を検出できるよう、ステージ間に決定論的なチェックを追加したワークフローです。 - Traceability – エージェント実行、ガードレール、アーティファクト生成、評価（eval）実行に関する OpenAI Agents SDK のトレースとスパンを出力します。 - Portable artifacts – 最終的なワークフローバンドルを JSON 形式で artifacts/notebook_runs/ ディレクトリに保存します。 - Eval-ready design – ライブなノートブックの状態から、Promptfoo プロバイダー、アサーション、設定（config）、および結果ファイル（result files）を生成する設計です。 - No database side effects – 生データベースに対して実行することなく、ドラフト SQL と検証出力を生成します。 このノートブックを終える頃には、以下の成果物を生成する動作中の SchemaFlow パイプラインが完成しています： - 解析されたデータベース変更リクエスト： タイトル（title） ドメイン（domain） 対象スキーマ（target schema） 対象テーブル（target table） 正規化された操作（normalized operations） 注記（notes） - 影響分析レポート： 影響を受けるテーブル、列、インデックス、ビュー、またはリレーションシップ リスク（risks） 仮定（assumptions） オプションの File Search エビデンスサマリー - ロールアウト計画： 実装ステップ 事前チェック（prechecks） 事後チェック（postchecks） ロールバックアクション - 4 つの必須セクションを含むドラフト SQL スクリプト： LANDING (ODS) STAGING (STG) CORE (DIM/FACT/VIEW) MARTS (SERVING) - 検証結果： 期待されるテーブルチェック（expected table checks） 期待される列チェック（expected column checks） ALTER TABLE、UPDATE、CREATE INDEX などの必須キーワードチェック - 保存された JSON アーティファクト： 変更リクエスト（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 このクックブックは、一般的なエンタープライズデータエンジニアリングのシナリオに焦点を当てています。ステークホルダーが自然言語でデータベーススキーマの変更を要求し、データチームがその要求を実装可能な計画に変換する必要があるというケースです。 ここでは小売ドメインは、ワークフローを具体的な形で示すための単なる手段に過ぎません。この段階的なアプローチは、他のソースシステム、データプロダクト、およびレビュープロセスにも適応可能です。 本ノートブックにおけるデフォルトの要求は以下の通りです: LOYALTY_TIER VARCHAR(20) を ODS.ODS_CUSTOMER_PROFILE に NULL 許容として追加する。 CORE.DIM_CUSTOMER から CUSTOMER_ID および IS_CURRENT=true の条件でバックフィルを行う。 (CUSTOMER_ID, LOYALTY_TIER) に非一意インデックスを追加する。 人間のデータエンジニアが本番用の SQL を記述する前に、通常は以下のいくつかの質問に答える必要があります: - どのテーブルとスキーマを変更するのか？ - 具体的にどの列、型、および NULL 許容性が要求されたのか？ - 歴史的データのバックフィルが必要か？ - この要求はインデックスの作成を意味しているか？ - どの下流レイヤーにこのフィールドを伝播させる必要があるか？ - レビュー担当者が注意すべきリスクは何らか？ - デプロイ前後に実行すべきチェックは何らか？ - どのようなロールバック手順が妥当か？ - 生成された SQL に必要な要素が含まれているか？ SchemaFlow は、段階的なエージェント・ワークフローとして実装されています。各ステージは、次のステージが消費する型付きの中間出力を作成します。その後、決定論的なチェックによって出力が検証され、ノートブックが最終バンドルを保存し、オプションで評価（evals）を実行します。 高レベルでは、SchemaFlow は以下のシーケンスに従います: ![image](https://developers.openai.com/cookbook/assets/images/schemaflow_seq_diagram.png) このノートブックは、読者がまずコア・ワークフローを実行し、その後、オプションの Promptfoo 評価セクションを実行するかどうかを判断できるように構成されています。 ## 目次 ## 概念的ガイド - 概要 - なぜこれが重要なのか - 主な利点 - 構築するもの - イントロダクション：ユースケースとソリューション - ワークフローの概要 - アーキテクチャ - デザイン・パターン - システム設計 - 実行ワークフロー ## ノートブックの実装 - 環境設定 - 入力 - オプションの PDF RAG コンテキスト（注：RAG は Retrieval-Augmented Generation の略） - ステージ 1-2: 変更要求の解析とインパクト分析 影響ダッシュボードのプレビュー - ステージ 1-2 の出力ガードレール - ステージ 3-4: 実行計画と SQL 生成（注：SQL は Structured Query Language の略） ステージ 3-4 の出力ガードレール - ステージ 5: 軽量な SQL 健全性チェック - 最終バンドル - アーティファクトの保存 アーティファクト生成後の健全性チェック - オプションのクリーンアップ Promptfoo 実行前のチェック/ガードレール - Promptfoo を使用したフローの評価 Promptfoo ランタイムディレクトリ設定 - Node.js および npm ランタイムの確認 - SchemaFlow コア・ランタイムのパブリッシュ - Promptfoo プロバイダー・ランタイム - Promptfoo アサーション・ランタイム（注：アサーションは検証や主張の文脈で使われる） - Promptfoo テストケースと設定の構築 - Promptfoo Eval の実行 - Review Latest Promptfoo Results ## Reference - Notes, Assumptions, and Extension Points SchemaFlow は、段階的で契約駆動型のエージェントアーキテクチャを採用しています。その目的は、モデルを単一のブラックボックスな SQL 生成器として扱うことを避けることです。代わりに、各ステージには狭い責任範囲があり、出力は検査・検証・追跡・再利用が可能になります。 ## 1. エージェントの専門化 各エージェントは 1 つの主要タスクを実行します: AgentResponsibilityMain Output Parse AgentExtract structured fields from the natural-language requestchange_json Impact AgentIdentify affected objects, assumptions, and risksimpact_json Plan AgentConvert the change and impact into rollout stepsplan_json SQL AgentDraft SQL across data platform layerssql_text この専門化により、ワークフローのデバッグが容易になります。もし SQL に列が含まれていない場合でも、問題が発生したのが解析段階か、影響分析段階か、計画段階か、それとも SQL 生成段階かを検査して特定できます。 ## 2. タイプ付き出力契約 このノートブックでは、構造化された各ステージに対して Pydantic モデルを定義しています: - ChangeRequestModel - ImpactModel - PlanModel これらのモデルは AgentOutputSchema でラップされており、Agents SDK が期待される出力形状を理解できるようになっています。また、ワークフローでは、モデル呼び出し後の出力を正規化し、下流のステージが実行される前に必要なキーが存在することを確認しています。 ## 3. 検索拡張型影響分析 PDF RAG セクションはオプションです。PDF_PATH が設定されている場合、ノートブックは以下の処理を行います: - OpenAI ベクトルストアを作成します。 - PDF をアップロードします。 - OpenAI に解析・チャンク化・埋め込み・インデックス作成を任せます。 - Impact Agent に FileSearchTool を提供します。 - 返されたファイル検索結果の要約をキャプチャします。 これは、変更要求が IFD（機能定義）、スキーマドキュメント、系図ファイル、データ契約、またはアーキテクチャ参照に基づいている必要がある場合に有用です。 ## 4. ステージ間のガードレールゲート ノートブックは主要なステージの後に決定論的なチェックを追加します: - ステージ 1〜2 のガードレールは、パース結果と影響分析の結果を検証します。 - ステージ 3〜4 のガードレールは、計画の完全性、データ型の伝播、および nullability（null 値の扱い）の処理を検証します。 - ステージ 5 の SQL チェックは、期待されるテーブル、列、および SQL キーワードの存在を検証します。 - アートifact 後のチェックでは、保存された JSON アートファクトが存在し、完全なループ（round-trip）が可能であることを確認します。 - Promptfoo 実行前のチェックでは、ノートブックの状態が評価（evals）に準備できていることを確認します。 これらのチェックは人間のレビューを代替するものではなく、一般的な沈黙の失敗を早期に検出するためのものです。 ## 5. アートファクト中心の実行 最終的なバンドルが主要なワークフローアートファクトとなります。これは、実行のレビューやデバッグに必要な状態をキャプチャします: bundle = { "summary": ..., "rag": ..., "change_json": ..., "impact_json": ..., "plan": ..., "sql": ..., "validation": ... } ノートブックはこのバンドルを artifacts/notebook_runs/ ディレクトリ下に保存します。 ## 6. ノートブック状態から生成された Eval ランタイム Promptfoo は別プロセスで実行されるため、アクティブなノートブックカーネルの変数を直接読み取ることはできません。これを解決するため、セクション 10 では、現在のノートブックの状態から再利用可能な小さな Python モジュールと Promptfoo のランタイムファイルを記述します。 これにより、プロンプトの編集、CHANGE_TEXT の編集、およびモデル設定の変更は、評価ファイルが再生成された際に反映されます。 ## コンポーネントアーキテクチャ ![image](https://developers.openai.com/cookbook/assets/images/schemaflow_system_design.png) ## 主要なランタイムオブジェクト | オブジェクト | 作成元 | 目的 | |---|---|---| | CHANGE_TEXT | 入力セクション | 自然言語によるデータベース変更リクエスト | | change_json | ステージ 1 | リクエストの構造化解釈 | | rag_vector_store_id | オプション PDF RAG セクション | アップロードされた PDF コンテキスト用のホスト型ベクトルストア ID | | rag_file_search_results | ステージ 2 | インパクトエージェントに返されるファイル検索結果のサマリー | | impact_json | ステージ 2 | 影響を受けるオブジェクト、リスク、および前提条件 | | plan_json | ステージ 3 | ロールアウト計画、チェック項目、ロールバックガイダンス | | sql_text | ステージ 4 | SQL スクリプトのドラフト | | validation | ステージ 5 | 決定論的な SQL の健全性チェック結果 | | bundle | 最終バンドルセクション | 統合されたワークフロー出力 | | out_path | アーティファクト保存セクション | 保存される JSON アーティファクトのパス | | promptfoo_config | Promptfoo セクション | 生成された評価設定 | ## 重要な境界条件 SchemaFlow はドラフトの実装アーティファクトを生成します。データベースに対して SQL を実行したり、マイグレーションを適用したり、プルリクエストを開いたり、本番システムを変更したりすることはありません。 ノートブックは順序通りに実行してください。 ## コアワークフロー - 環境設定 依存関係のインポート。 - OpenAI Agents SDK のバージョンを確認します。 - OPENAI_API_KEY を読み取ります。 - 追跡とモデル選択を設定します。 - 入力 CHANGE_TEXT を定義します。 - これはコアワークフローに必要な唯一のビジネス入力です。 - オプションの PDF RAG コンテキスト PDF_PATH を None に設定して、検索機能なしで実行します。 - PDF_PATH にローカルの PDF ファイルを指定すると、影響分析のためのファイル検索コンテキストが有効になります。 - ステージ 1〜2 変更要求を解析します。 - 影響を分析します。 - 必要に応じて、影響分析中にファイル検索を使用できます。 - ステージ 1〜2 のガードレール 解析出力が適切に構造化されていることを確認します。 - 影響分析の出力に対象が含まれていることを確認します。 - 影響を受けるオブジェクトに必要なフィールドが含まれていることを確認します。 - ステージ 3〜4 実行計画を生成します。 - ランディング、ステージング、コア、マートの各レイヤーにわたって SQL を生成します。 - ステージ 3〜4 のガードレール 計画セクションが適切に埋められていることを確認します。 - データ型の伝播を確認します。 - nullability（null 許容性）の動作が要求と一致していることを確認します。 - ステージ 5 SQL 健全性チェック 空の SQL が存在しないか確認します。 - 期待される対象テーブルと列を確認します。 - 要求から推測される必要な SQL アクションを確認します。 - 最終的なバンドルとアーティファクト 完全な出力バンドルを組み立てます。 - JSON 形式で保存します。 - アーティファクトが正常に往復（round-trip）することを確認します。 ## オプションの評価ワークフロー - プリプロンプトフューチェック ノートブックの状態が評価準備完了であることを確認します。 - プロンプトフューランタイム生成 再利用可能な SchemaFlow コアモジュールを作成します。 - プロンプトフュープロバイダーを記述します。 - プロンプトフューアサートファイル（検証ファイル）を記述します。 - プロンプトフューテストケースと設定ファイルを生成します。 - プロンプトフュー評価実行 解析のみを行う評価と、フルフローの評価を実行します。 - タイムスタンプ付きの JSON および HTML レポートを保存します。 - schemaflow_cookbook_eval_latest.* のエイリアスを更新します。 このセクションでは、SchemaFlow ワークフローのためのランタイム環境を準備します。 セットアップセルは以下の処理を行います: - ノートブック全体で使用される標準的な Python ユーティリティをインポートします。 - OpenAI クライアントをインポートします。 - OpenAI Agents SDK のプリミティブをインポートします: Agent - Runner - RunConfig - AgentOutputSchema - FileSearchTool - 追跡（tracing）およびスパン（span）のヘルパー関数 - インストールされた openai-agents パッケージが最小必要なバージョンを満たしているか確認します。 - 環境変数から OPENAI_API_KEY を読み込み、存在しない場合は入力プロンプトを表示して取得します。 - OPENAI_MODEL でモデルを指定し、デフォルトは gpt-5.5 と設定します。 - すべての関連するエージェント実行とガードレールスパンをグループ化できるよう、トレースグループ ID を作成します。 このデモでは意図的に機密性の高いトレースペイロードを有効にしており、プロンプト、出力、評価バンドル、およびツールのデータがトレース上で確認できるようにしています。本番環境での利用においては、個人データを扱う前にこの設定を見直す必要があります。 %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) =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("```"): text = re.sub(r"^```(?: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 翻訳全文 ノードが辞書型の場合: ノードの"type"が"file_search_call"である場合、 結果リスト（存在しない場合は空リスト）を順に処理し、 各結果をモデル形式に変換します。 変換後の結果が辞書型であれば、 "text"または"content"フィールドからテキストを取得し、存在しない場合は空文字列とします。 テキストがリスト型の場合、改行で結合して文字列化します。 結果オブジェクトに"file_id"、"filename"（"file_name"または"title"を優先）、"score"、およびテキストプレビュー（先頭1200文字）を追加して結果リストに格納します。 ノードのすべての子要素に対して再帰的に処理を行います。 elif ノードがリスト、タプル、セット型のいずれかの場合: 各要素に対して再帰的に処理を行います。 最終値に対して上記の訪問関数を呼び出します。

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

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

関連記事

調べる

選ぶ

サイト