databricks-ai-functions

Databricks AI Functions

公式ドキュメント： https://docs.databricks.com/large-language-models/ai-functions 各関数のリファレンス： https://docs.databricks.com/sql/language-manual/sql-ref-functions-builtin

概要

Databricks AI Functions は、データパイプラインから Foundation Model API を直接呼び出す組み込みの SQL および PySpark 関数です。モデルエンドポイントの設定や API キー、定型コードは不要です。UPPER() や LENGTH() と同じようにテーブルの列を扱い、大規模なバッチ推論に最適化されています。

ai_query よりも、タスクに特化した関数を常に優先してください。 ai_query を使うのは、適切なタスク関数がない場合に限ります（カスタム／外部エンドポイント、マルチモーダル、または ai_extract の制限を超える JSON）。以下のすべての関数に共通する基本要件は、DBR 15.1 以降（ノートブック）／15.4 ML LTS（バッチ）、SQL Warehouse Classic では利用不可、および AI Functions をサポートするリージョンです。「前提条件」列には、これらに加えて必要な条件のみを記載しています。

コストと速度 — 呼び出しごとに LLM 推論が実行されます（時間がかかり、トークンごとに課金されます）。 関数は行ごとに一度だけ実行し、その結果を Delta テーブルに保存してください。後続のクエリのたびに再実行してはいけません。デモでは、数百万行のテーブルを生成しないでください。必要に応じて入力をサンプリングし、デモをすばやく実行できるようにします。一度具体化してから、低コストの Delta 出力に対してクエリを実行してください。

関数列のリンク先は、リポジトリ内の詳細なリファレンス（すべてのオプション、スキーマ、例）です。ドキュメントのリンク先は公式ページです。

関数	タスク	入力	出力	追加の前提条件	ドキュメント
ai_analyze_sentiment	感情スコアリング	content STRING	STRING — positive／negative／neutral／mixed、または NULL	—	↗
ai_classify	固定ラベルによる振り分け	content STRING|VARIANT、labels（2～500）、[options MAP]	VARIANT — {response:[label], error_message}	—	↗
ai_extract	エンティティ／フィールドの抽出	content STRING|VARIANT、schema STRING（JSON）、[options MAP]	VARIANT — {response:{…}, error_message, metadata}	フィールド数は256以下、ネスト階層は12以下	↗
ai_fix_grammar	文法の修正	content STRING	STRING（修正済み）	—	↗
ai_gen	自由形式の生成	prompt STRING	STRING	—	↗
ai_mask	PII のマスキング	content STRING、labels ARRAY<STRING>	STRING（エンティティを [MASKED] に置換）	—	↗
ai_similarity	意味的類似度	expr1 STRING、expr2 STRING	FLOAT（0.0～1.0）	—	↗
ai_summarize	要約	content STRING、[max_words INT]（0 = 上限なし）	STRING	パブリックプレビュー。英語向けに調整済み	↗
ai_translate	翻訳	content STRING、to_lang STRING	STRING	対応言語：en、fr、de、hi、it、pt、es、th	↗
ai_parse_document	PDF／Office ファイル／画像の解析	content BINARY、[Map('version','2.0', …)]	VARIANT — ページ、要素、error_status	DBR 17.3+。500ページ／100 MB 以下	↗
ai_prep_search	解析済みドキュメントから RAG 用のチャンクを作成	parsed VARIANT、[options MAP]	VARIANT — {document:{contents, pages, source_uri}, error_status}	DBR 18.2+（サーバーレス環境 v3+）	↗
ai_query	任意のサービングエンドポイント（組み込みの基盤モデルまたはカスタムモデル）、マルチモーダル、複雑な JSON（最終手段）	endpoint STRING、request STRING|STRUCT、[returnType]、[failOnError BOOL]、[modelParameters STRUCT]、[responseFormat STRING]、[files]	解析済みレスポンス。failOnError => false の場合は STRUCT{response, errorMessage}	Pro／Serverless ウェアハウス。エンドポイントに対する CAN QUERY 権限	↗
ai_forecast	時系列予測（テーブル値関数）	observed TABLE、horizon、time_col、value_col、[group_col]、[prediction_interval_width]、[frequency]、[seed]、[parameters]	行：時間／グループ列に加え、値ごとに {v}_forecast、{v}_upper、{v}_lower（DOUBLE）	Pro／Serverless ウェアハウス。パブリックプレビュー	↗
モデルは Apache 2.0 または LLAMA 3.3 Community License のもとで実行されます。ライセンスへの準拠はご自身の責任です。

パターン

タスク関数をチェーンして、1 回の処理で列を拡充します。 ai_classify／ai_extract は VARIANT を返します。コロン演算子（:response）で読み取ります。

SELECT id,
  ai_analyze_sentiment(content)                                                   AS sentiment,
  ai_summarize(content, 30)                                                       AS summary,
  ai_classify(content, '["technical","billing","other"]', map('version','2.0')):response[0]::STRING AS category,
  ai_extract(content, '["product","error_code","date"]', map('version','2.0')):response:product::STRING AS product,
  ai_fix_grammar(content)                                                         AS content_clean
FROM raw_feedback;

PySpark では、これらの関数を expr(...) の中で呼び出します。df.withColumn("category", expr("ai_classify(content, '[\"a\",\"b\"]', map('version','2.0')):response[0]::STRING")) のように記述し、VARIANT フィールドは selectExpr("col:response:field::STRING AS field") で読み取ります。

保存前の PII のマスキング — ai_mask(content, ARRAY(entity_types)) は、エンティティを [MASKED] に置換したテキストを返します。

SELECT ai_mask(message, array('person','email','phone','address')) AS message_safe FROM raw_messages;

セマンティックマッチング／重複排除 — ai_similarity は 0～1 の値を返します。自己結合してしきい値を設定します。

SELECT a.id, b.id, ai_similarity(a.name, b.name) AS score
FROM companies a JOIN companies b ON a.id < b.id
WHERE ai_similarity(a.name, b.name) > 0.85;

予測 — テーブル値関数です。将来の期間ごと（およびグループごと）に 1 行を返します。パラメーター／グループ／間隔の全形式については、3-ai-forecast.md を参照してください。

SELECT * FROM ai_forecast(
    observed => TABLE(SELECT date, sales FROM daily_sales),
    horizon => '2026-12-31', time_col => 'date', value_col => 'sales');
-- Returns: date, sales_forecast, sales_upper, sales_lower

ai_query によるネストされた JSON（最後の手段 — ai_extract の制限を超える場合にのみ使用）— from_json でレスポンスを解析します。モデル名、マルチモーダルの files =>、modelParameters、SQL UDF については、2-ai-query.md を参照してください。

SELECT from_json(
    ai_query('databricks-claude-sonnet-4',
        concat('Extract invoice as JSON with nested line_items array: ', text_blocks),
        responseFormat => '{"type":"json_object"}', failOnError => false).response,
    'STRUCT<numero:STRING, total:DOUBLE, line_items:ARRAY<STRUCT<code:STRING, qty:DOUBLE>>>'
) AS invoice
FROM parsed_documents;

ドキュメントの解析（ai_parse_document）と RAG のチャンク分割（ai_prep_search）は、以下の専用ステージングパイプラインで処理します。

ドキュメント処理パイプライン

AI 関数をステージごとにチェーンし、Delta テーブルに書き込むことで、バッチドキュメント処理を行います。この例は Spark Declarative Pipeline（SDP／Lakeflow／DLT） 用に記述されています。STREAM(...) ソースで CREATE OR REFRESH STREAMING TABLE を使用します。同じロジックを ノートブック／SQL ウェアハウス で単独実行するには、各 CREATE OR REFRESH STREAMING TABLE x AS を CREATE OR REPLACE TABLE x AS に置き換え、STREAM(...) のラッパーを削除します。SDP の Python では、from pyspark import pipelines as dp とともに @dp.table を使用します。

-- Stage 1 — parse binary docs (any type), filter parse errors
CREATE OR REFRESH STREAMING TABLE raw_parsed AS
SELECT path,
  concat_ws('\n', transform(parsed:document:elements, e -> e:content::STRING)) AS text_blocks,
  parsed:error_status AS parse_error
FROM (
  SELECT path, ai_parse_document(content, map('version','2.0')) AS parsed
  FROM STREAM read_files('/Volumes/my_catalog/doc_processing/landing/', format => 'binaryFile')
)
WHERE parsed:error_status IS NULL;

-- Stage 2 — classify document type (cheap, no endpoint selection)
CREATE OR REFRESH STREAMING TABLE classified_docs AS
SELECT *,
  ai_classify(text_blocks, '["invoice","purchase_order","receipt","contract","other"]', map('version','2.0')):response[0]::STRING AS doc_type
FROM STREAM raw_parsed;

-- Stage 3 — extract fields; ai_extract returns a VARIANT, read fields with `:`
CREATE OR REFRESH STREAMING TABLE extracted AS
SELECT path, doc_type,
  result:response:invoice_number::STRING AS invoice_number,
  result:response:vendor_name::STRING    AS vendor_name,
  result:response:total_amount::DOUBLE   AS total_amount,
  result:error_message::STRING           AS extract_error
FROM (
  SELECT *, ai_extract(text_blocks,
    '{"invoice_number":{"type":"string"},"vendor_name":{"type":"string"},"total_amount":{"type":"number"}}',
    map('version','2.0')) AS result
  FROM STREAM classified_docs WHERE doc_type = 'invoice' AND text_blocks IS NOT NULL
);

バッチジョブでは、行ごとのエラーによって実行が失敗しないように、エラーをサイドカーテーブルに出力します。ai_extract の result:error_message（上記のようにコロンでアクセスする VARIANT）を保持し、ai_query には failOnError => false を渡して、ai_response.errorMessage（ドットでアクセスする STRUCT フィールド）を確認します。2-ai-query.md を参照してください。

カスタム RAG パイプライン — Parse → Prep → Index

フィールド抽出ではなく検索を行う場合は、ai_parse_document → ai_prep_search（セマンティックチャンク分割とコンテキスト拡充、DBR 18.2 以降）→ Vector Search Delta Sync インデックスの順に処理します。ai_prep_search は chunk_id、chunk_to_retrieve、chunk_to_embed（タイトル／ヘッダー／ページ情報を付加）を返します。chunk_to_embed を埋め込み、chunk_to_retrieve を LLM に返します。 ここでは単独実行の例を示します。SDP では CREATE OR REPLACE TABLE を CREATE OR REFRESH STREAMING TABLE に置き換え、STREAM read_files(...) を追加します。

CREATE OR REPLACE TABLE parsed_chunks AS
WITH prepped AS (
  SELECT path AS source_path, ai_prep_search(ai_parse_document(content)) AS prep
  FROM read_files('/Volumes/my_catalog/doc_processing/docs/', format => 'binaryFile')
)
SELECT
  variant_get(chunk, '$.chunk_id',          'STRING') AS chunk_id,
  variant_get(chunk, '$.chunk_to_retrieve', 'STRING') AS chunk_to_retrieve,
  variant_get(chunk, '$.chunk_to_embed',    'STRING') AS chunk_to_embed,
  source_path
FROM prepped LATERAL VIEW explode(variant_get(prep, '$.document.contents', 'ARRAY<VARIANT>')) c AS chunk;

続いて CDF を有効にし（ALTER TABLE parsed_chunks SET TBLPROPERTIES (delta.enableChangeDataFeed = true)）、databricks-vector-search スキルを使用して Delta Sync インデックスを構築します。PK は chunk_id、埋め込みソースは chunk_to_embed、返す列は chunk_to_retrieve です。

バッチ処理以外の用途：

• 出力に関する質問への回答 — 結果の Delta テーブルを Genie Agent に指定し、SQL を手書きする代わりに自然言語でクエリを実行します。databricks-genie-agents スキルを参照してください。
• 低レイテンシ／サービング — モデルを Unity Catalog に登録してサービングするなど、この機能をリアルタイムのガバナンス対象エンドポイントとして公開するには、databricks-model-serving スキルを使用してください。
• 本番環境での増分取り込み — チェックポイントや trigger(availableNow=True) を含む、実行可能なエンドツーエンドのストリーミング ai_parse_document ジョブについては、databricks/bundle-examples · job_with_ai_parse_document を参照してください。

リファレンスファイル

• 1-task-functions.md — タスク固有の各関数についての詳細なリファレンス：すべてのオプション／スキーマ（例：ai_extract v2.1 の引用と信頼度スコア、ai_classify のマルチラベル、ai_parse_document のオプションと出力スキーマ、ai_prep_search のチャンクスキーマ）および複雑な例を掲載しています。上記の概要表から各関数のセクションに直接リンクしています。
• 2-ai-query.md — ai_query の完全なリファレンス：すべてのパラメーター、responseFormat による構造化出力、マルチモーダルの files =>、UDF パターン、エラー処理
• 3-ai-forecast.md — ai_forecast のパラメーター、単一メトリック、複数グループ、複数メトリック、信頼区間のパターン

よくある問題

問題	解決策
ai_parse_document が見つからない	DBR 17.3+ が必要です。クラスターのランタイムを確認してください。
ai_prep_search が見つからない	DBR 18.2+（サーバーレス環境 v3+）が必要です。
VARIANT に対する explode() が失敗する	explode には ARRAY が必要です。まず次のようにキャストしてください：explode(variant_get(prep, '$.document.contents', 'ARRAY<VARIANT>'))。
誤った RAG 列を埋め込んでいる	chunk_to_embed（コンテキストが付加されたもの）を埋め込み、LLM には chunk_to_retrieve を返してください。
ai_forecast が失敗する	Pro または Serverless SQL ウェアハウスが必要です。Classic または Starter では利用できません。
すべての関数が NULL を返す	入力列が NULL です。呼び出し前に WHERE col IS NOT NULL でフィルターしてください。
特定の言語で ai_translate が失敗する	対応言語（8）：英語（en）、フランス語（fr）、ドイツ語（de）、ヒンディー語（hi）、イタリア語（it）、ポルトガル語（pt）、スペイン語（es）、タイ語（th）です。to_lang にはコードまたは正式名称を指定します（references/1-task-functions.md を参照）。その他の言語には、多言語モデルと ai_query を使用してください。
ai_classify が予期しないラベルを返す	明確で、相互に重複しないラベル名を使用してください。ラベル数を少なく（2～5 個）すると、より信頼性の高い結果が得られます。
バッチジョブの一部の行で ai_query がエラーになる	failOnError => false を追加してください。エラーを発生させる代わりに、.response と .errorMessage（ドットアクセス）を持つ STRUCT を返します。
ai_query が 403「no longer available」で失敗する	従来の databricks-* 基盤モデルエンドポイント名は廃止され、Unity Catalog のモデルサービスに置き換えられました。system.ai.<model> に切り替えてください（同じ呼び出しで、その他の引数は変更不要です）。カスタムエンドポイントはこの移行の影響を受けず、既存の名前を引き続き使用します。2-ai-query.md を参照してください。
バッチジョブの実行が遅い	バッチ推論のスループットを最適化するには、サーバーレスやインタラクティブではなく、DBR 15.4 ML LTS クラスターを使用してください。