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 是内置的 SQL 和 PySpark 函数，可直接从数据管道调用基础模型 API——无需设置模型端点、无需 API 密钥，也无需编写样板代码。它们可像 UPPER() 或 LENGTH() 一样自然地处理表列，并针对大规模批量推理进行了优化。

始终优先使用特定任务函数，而非 ai_query。 仅当没有适用的任务函数时才使用 ai_query（例如自定义或外部端点、多模态场景，或超出 ai_extract 限制的 JSON）。以下所有函数有一组共同的基线要求：DBR 15.1+（笔记本）/ 15.4 ML LTS（批量处理）、不支持 SQL Warehouse Classic，且所在区域必须支持 AI Functions——“先决条件”列仅列出额外要求。

成本与速度——每次调用都是一次大模型推理（速度较慢，按 token 计费）。 每行仅调用一次函数，并将结果持久化到 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	Public Preview；针对英语调优	↗
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 — 页面、元素、错误状态	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+（serverless 环境 v3+）	↗
ai_query	任意 serving endpoint（内置基础模型或自定义模型）、多模态、复杂 JSON（最后手段）	endpoint STRING、request STRING|STRUCT、[returnType]、[failOnError BOOL]、[modelParameters STRUCT]、[responseFormat STRING]、[files]	已解析的响应；failOnError => false 时为 STRUCT{response, errorMessage}	Pro/Serverless 仓库；endpoint 上具有 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 仓库；Public Preview	↗
模型基于 Apache 2.0 或 LLAMA 3.3 Community License 运行，合规责任由你承担。

模式

串联任务函数，一次性丰富列。 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")) — 并通过 selectExpr("col:response:field::STRING AS field") 读取 VARIANT 字段。

存储前脱敏 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;

预测 — 表值函数；每个未来周期（以及每个分组）返回一行。完整的参数 / 分组 / 区间形式见 → 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 Functions 按阶段串联到 Delta 表中，以批量处理文档。示例采用 Spark Declarative Pipeline（SDP / Lakeflow / DLT） 编写 — 使用 CREATE OR REFRESH STREAMING TABLE 和 STREAM(...) 数据源。若要在 笔记本 / SQL 仓库 中独立运行相同逻辑，请将每个 CREATE OR REFRESH STREAMING TABLE x AS 替换为 CREATE OR REPLACE TABLE x AS，并移除 STREAM(...) 包装。在 SDP Python 中，它对应于 @dp.table，并使用 from pyspark import pipelines as dp。

-- 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 管道 — 解析 → 准备 → 建立索引

用于检索而非字段提取：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 返回给大模型。 示例为独立运行；在 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 skill 构建 Delta Sync 索引：主键 chunk_id，嵌入源 chunk_to_embed，返回 chunk_to_retrieve。

批处理之外：

• 针对输出提问 — 将 Genie Agent 指向生成的 Delta 表，以自然语言查询取代手写 SQL；请参阅 databricks-genie-agents skill。
• 低延迟 / 服务 — 若要将其公开为实时且受治理的端点（例如，将模型注册到 Unity Catalog 并提供服务），请使用 databricks-model-serving skill。
• 生产级增量摄取 — 如需可运行的端到端流式 ai_parse_document 作业（检查点、trigger(availableNow=True)），请参阅 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+）。
explode() 处理 VARIANT 失败	explode 需要 ARRAY — 先进行强制转换：explode(variant_get(prep, '$.document.contents', 'ARRAY<VARIANT>'))。
嵌入了错误的 RAG 列	嵌入 chunk_to_embed（包含上下文）；将 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 集群（而非无服务器或交互式集群）以优化批量推理吞吐量。