使用 Azure Databricks AI Functions 对数据应用 AI

本文旨在介绍 Azure Databricks AI Functions 以及支持的功能。

什么是 AI 功能？

AI 功能是指可用于对存储在 Databricks 上的数据应用 AI 的内置功能（例如，文本翻译或情绪分析）。 它们可以从 Databricks 上的任意位置运行，包括 Databricks SQL、笔记本、Lakeflow 声明性管道和工作流。

AI 功能使用简单、速度快、可缩放。 分析师可以使用 AI 功能对其专有数据应用数据智能，而数据科学家和机器学习工程师则可以使用它们来构建生产级批量管道。

AI 函数提供用于特定任务和通用目的的功能。

• 特定任务型功能可为归纳总结文本和翻译等任务提供高级别 AI 功能。 这些特定任务型功能使用最前沿的生成式 AI 模型，这些模型由 Databricks 托管和管理。 有关支持的函数和模型，请参阅 特定于任务的 AI 函数 。
• ai_query是一项常规用途型功能，允许将任意类型的 AI 模型应用到数据上。 请参阅 “常规用途”函数： ai_query。

特定任务型 AI 功能

特定于任务的函数适用于特定任务，因此你可以自动执行常规作，例如简单的摘要和快速翻译。 Databricks 建议这些函数入门，因为它们调用 Databricks 维护的最先进的生成 AI 模型，并且不需要任何自定义。

有关示例，请参阅 使用 AI Functions 分析客户评论 。

下表列出了支持的函数及其执行的任务。

常规用途型功能：ai_query

ai_query() 功能允许对生成式 AI 和经典机器学习任务（其中包括提取信息、归纳总结内容、识别欺诈和预测收入等）的数据应用任何 AI 模型。 有关语法详细信息和参数，请参阅 ai_query 函数。

下表汇总了支持的模型类型、关联的模型和为每个模型提供终结点配置要求的模型。

将ai_query与基础模型配合使用

以下示例演示如何利用使用由 Databricks 托管的基础模型的 ai_query。

• 有关语法详细信息和参数，请参阅 ai_query 函数 。
• 有关 多模式输入查询示例，请参阅多模式输入 。
• 有关如何为高级用例配置参数的指导，请参阅 高级方案的示例 ，例如：
  • 使用 failOnError 处理错误
  • Azure Databricks 上的结构化输出 ，用于如何为查询响应指定结构化输出。

SELECT text, ai_query(
    "databricks-meta-llama-3-3-70b-instruct",
    "Summarize the given text comprehensively, covering key points and main ideas concisely while retaining relevant details and examples. Ensure clarity and accuracy without unnecessary repetition or omissions: " || text
) AS summary
FROM uc_catalog.schema.table;

示例笔记本：批处理推理和结构化数据提取

以下示例笔记本演示如何通过自动化提取技术执行基本的结构化数据提取 ai_query ，以将原始非结构化数据转换为有组织的、易用的信息。 此笔记本还演示如何利用马赛克 AI 代理评估，通过真实基础数据来评估其准确性。

批处理推理和结构化数据提取笔记本

获取笔记本

与传统 ML（机器学习）模型一起使用ai_query

ai_query 支持包括完全自定义在内的传统机器学习模型。 这些模型必须部署在模型服务终结点上。 有关语法详细信息和参数，请参阅 ai_query 函数 函数。

SELECT text, ai_query(
  endpoint => "spam-classification",
  request => named_struct(
    "timestamp", timestamp,
    "sender", from_number,
    "text", text),
  returnType => "BOOLEAN") AS is_spam
FROM catalog.schema.inbox_messages
LIMIT 10

示例笔记本：使用 BERT 进行批处理推理来识别命名实体

以下笔记本显示了使用 BERT 的传统 ML 模型批处理推理示例。

使用 BERT 进行命名实体识别笔记本的批处理推理

获取笔记本

使用 Python 工作流中现有的 AI 功能

AI 功能可轻松集成到现有 Python 工作流中。

下面这项功能可将 ai_query 的输出写入到输出表中：

df_out = df.selectExpr(
  "ai_query('databricks-meta-llama-3-3-70b-instruct', CONCAT('Please provide a summary of the following text: ', text), modelParameters => named_struct('max_tokens', 100, 'temperature', 0.7)) as summary"
)
df_out.write.mode("overwrite").saveAsTable('output_table')

下面这项功能可将文本归纳总结到表中：

df_summary = df.selectExpr("ai_summarize(text) as summary")
df_summary.write.mode('overwrite').saveAsTable('summarized_table')

在生产工作流中使用 AI 功能

对于大规模批处理推理，可以将特定于任务的 AI 函数或常规用途函数 ai_query 集成到生产工作流中，例如 Lakeflow 声明性管道、Databricks 工作流和结构化流式处理。 这样可实现大规模的生产级处理。 有关示例和详细信息，请参阅 “部署批处理推理管道 ”。

监控 AI 功能进度

若要了解有多少推理已完成或失败并排查性能问题，可以使用查询配置文件功能监视 AI Functions 的进度。

在 Databricks Runtime 16.1 ML 及更高版本中，从工作区中的 SQL 编辑器查询窗口：

1. 选择链接，正在原始结果窗口底部运行。 性能窗口显示在右侧。
2. 点击查看查询配置文件，查看性能详细信息。
3. 点击“AI 查询”查看特定查询的指标，其中包括已完成或已失败的推理数量和完成请求花费的总时间。

查看批处理推理工作负荷的成本

以下示例演示如何基于作业、计算、SQL 仓库和 Lakeflow 声明性管道筛选批处理推理工作负荷。

有关如何查看使用 AI Functions 的批处理推理工作负荷的成本的常规示例，请参阅 Monitor 模型提供服务成本 。

Jobs

以下查询显示了哪些作业通过 system.workflow.jobs 系统表进行批处理推理。 请参阅 使用系统表监视作业成本和性能。

SELECT *
FROM system.billing.usage u
  JOIN system.workflow.jobs x
    ON u.workspace_id = x.workspace_id
    AND u.usage_metadata.job_id = x.job_id
  WHERE u.usage_metadata.workspace_id = <workspace_id>
    AND u.billing_origin_product = "MODEL_SERVING"
    AND u.product_features.model_serving.offering_type = "BATCH_INFERENCE";

计算

下面显示了哪些群集通过系统表 system.compute.clusters 用于批处理推理。

SELECT *
FROM system.billing.usage u
  JOIN system.compute.clusters x
    ON u.workspace_id = x.workspace_id
    AND u.usage_metadata.cluster_id = x.cluster_id
  WHERE u.usage_metadata.workspace_id = <workspace_id>
    AND u.billing_origin_product = "MODEL_SERVING"
    AND u.product_features.model_serving.offering_type = "BATCH_INFERENCE";

Lakeflow 声明性管道

下面显示了哪些 Lakeflow 声明性管道正用于使用系统表进行 system.lakeflow.pipelines 批处理推理。

SELECT *
FROM system.billing.usage u
  JOIN system.lakeflow.pipelines x
    ON u.workspace_id = x.workspace_id
    AND u.usage_metadata.dlt_pipeline_id = x.pipeline_id
  WHERE u.usage_metadata.workspace_id = <workspace_id>
    AND u.billing_origin_product = "MODEL_SERVING"
    AND u.product_features.model_serving.offering_type = "BATCH_INFERENCE";

SQL 仓库

下面显示了哪些 SQL 仓库正用于使用系统表进行 system.compute.warehouses 批量推理。

SELECT *
FROM system.billing.usage u
  JOIN system.compute.warehouses x
    ON u.workspace_id = x.workspace_id
    AND u.usage_metadata.warehouse_id = x.warehouse_id
  WHERE u.usage_metadata.workspace_id = <workspace_id>
    AND u.billing_origin_product = "MODEL_SERVING"
    AND u.product_features.model_serving.offering_type = "BATCH_INFERENCE";