Aller au contenu principal
Dernière mise à jour le

Déployer des pipelines d'inférence par batch

Cette page explique comment vous pouvez intégrer Enrichir des données à l'aide des AI Functions dans d'autres produits de données et d'IA Databricks afin de créer des pipelines d'inférence par batch complets. Pour en savoir plus sur les AI Functions, y compris les fonctions spécifiques à une tâche et les fonctions à usage général, consultez Enrichir des données à l'aide des AI Functions.

Ces pipelines peuvent exécuter des workflows de bout en bout qui incluent l'ingestion, le prétraitement, l'inférence et le post-traitement. Les pipelines peuvent être créés en SQL ou Python et déployés comme suit :

  • LakeFlow Pipelines
  • Workflows planifiés à l'aide de Databricks workflows
  • Workflows d’inférence en streaming à l’aide de Structured Streaming

Exigences​

  • Un Workspace dans une région prise en charge par les APIs Foundation Model.
  • Les AI Functions ne sont pas disponibles sur les SQL warehouses Pro ou Classic.
  • Notebooks et workflows : Serverless compute est requis. AI Functions ne sont pas pris en charge sur les clusters de compute classiques.
  • Autorisation de query sur la table Delta dans Unity Catalog qui contient les données que vous souhaitez utiliser.

Effectuer une inférence batch incrémentielle sur les LakeFlow Pipelines​

L'exemple suivant effectue une inférence batch incrémentielle à l'aide de Lakeflow pipelines lorsque les données sont continuellement mises à jour.

Étape 1 : Ingestion des données d'actualité brutes à partir d'un volume​

SQL
CREATE OR REFRESH STREAMING TABLE news_raw
COMMENT "Raw news articles ingested from volume."
AS SELECT *
FROM STREAM(read_files(
  '/Volumes/databricks_news_summarization_benchmarking_data/v01/csv',
  format => 'csv',
  header => true,
  mode => 'PERMISSIVE',
  multiLine => 'true'
));

Étape 2 : Appliquer l'inférence LLM pour extraire le titre et la catégorie​

SQL

CREATE OR REFRESH MATERIALIZED VIEW news_categorized
COMMENT "Extract category and title from news articles using LLM inference."
AS
SELECT
  inputs,
  ai_query(
    "databricks-meta-llama-3-3-70b-instruct",
    "Extract the category of the following news article: " || inputs,
    responseFormat => '{
      "type": "json_schema",
      "json_schema": {
        "name": "news_extraction",
        "schema": {
          "type": "object",
          "properties": {
            "title": { "type": "string" },
            "category": {
              "type": "string",
              "enum": ["Politics", "Sports", "Technology", "Health", "Entertainment", "Business"]
            }
          }
        },
        "strict": true
      }
    }'
  ) AS meta_data
FROM news_raw
LIMIT 2;

Étape 3 : Valider la sortie de l'inférence LLM avant la synthétisation​

SQL
CREATE OR REFRESH MATERIALIZED VIEW news_validated (
  CONSTRAINT valid_title EXPECT (size(split(get_json_object(meta_data, '$.title'), ' ')) >= 3),
  CONSTRAINT valid_category EXPECT (get_json_object(meta_data, '$.category') IN ('Politics', 'Sports', 'Technology', 'Health', 'Entertainment', 'Business'))
)
COMMENT "Validated news articles ensuring the title has at least 3 words and the category is valid."
AS
SELECT *
FROM news_categorized;

Étape 4 : Synthétiser les articles d'actualité à partir des données validées​

SQL
CREATE OR REFRESH MATERIALIZED VIEW news_summarized
COMMENT "Summarized political news articles after validation."
AS
SELECT
  get_json_object(meta_data, '$.category') as category,
  get_json_object(meta_data, '$.title') as title,
  ai_query(
    "databricks-meta-llama-3-3-70b-instruct",
    "Summarize the following political news article in 2-3 sentences: " || inputs
  ) AS summary
FROM news_validated;

Automatiser les jobs d'inférence par batch à l'aide des workflows Databricks​

Planifiez les jobs d'inférence batch et automatisez les pipelines d'IA.

SQL
SELECT
   *,
   ai_query('databricks-meta-llama-3-3-70b-instruct', request => concat("You are an opinion mining service. Given a piece of text, output an array of json results that extracts key user opinions, a classification, and a Positive, Negative, Neutral, or Mixed sentiment about that subject.
AVAILABLE CLASSIFICATIONS
Quality, Service, Design, Safety, Efficiency, Usability, Price
Examples below:
DOCUMENT
I got soup. It really did take only 20 minutes to make some pretty good soup. The noises it makes when it's blending are somewhat terrifying, but it gives a little beep to warn you before it does that. It made three or four large servings of soup. It's a single layer of steel, so the outside gets pretty hot. It can be hard to unplug the lid without knocking the blender against the side, which is not a nice sound. The soup was good and the recipes it comes with look delicious, but I'm not sure I'll use it often. 20 minutes of scary noises from the kitchen when I already need comfort food is not ideal for me. But if you aren't sensitive to loud sounds it does exactly what it says it does..
RESULT
[
 {'Classification': 'Efficiency', 'Comment': 'only 20 minutes','Sentiment': 'Positive'},
 {'Classification': 'Quality','Comment': 'pretty good soup','Sentiment': 'Positive'},
 {'Classification': 'Usability', 'Comment': 'noises it makes when it's blending are somewhat terrifying', 'Sentiment': 'Negative'},
 {'Classification': 'Safety','Comment': 'outside gets pretty hot','Sentiment': 'Negative'},
 {'Classification': 'Design','Comment': 'Hard to unplug the lid without knocking the blender against the side, which is not a nice sound', 'Sentiment': 'Negative'}
]
DOCUMENT
", REVIEW_TEXT, '\n\nRESULT\n')) as result
FROM catalog.schema.product_reviews
LIMIT 10

AI Functions utilisant Structured Streaming​

Appliquez l’inférence d’IA dans des scénarios quasi temps réel ou micro-batch à l’aide de ai_query et de Structured Streaming.

Étape 1. Lisez votre table Delta statique​

Lisez votre table Delta statique comme s'il s'agissait d'un Stream.

Python

from pyspark.sql import SparkSession
import pyspark.sql.functions as F

spark = SparkSession.builder.getOrCreate()

# Spark processes all existing rows exactly once in the first micro-batch.
df = spark.table("enterprise.docs")  # Replace with your table name containing enterprise documents
df.repartition(50).write.format("delta").mode("overwrite").saveAsTable("enterprise.docs")
df_stream = spark.readStream.format("delta").option("maxBytesPerTrigger", "50K").table("enterprise.docs")

# Define the prompt outside the SQL expression.
prompt = (
    "You are provided with an enterprise document. Summarize the key points in a concise paragraph. "
    "Do not include extra commentary or suggestions. Document: "
)

Étape 2. Appliquer ai_query​

Spark ne traite ceci qu'une seule fois pour les données statiques, à moins que de nouvelles lignes n'arrivent dans la table.

Python

df_transformed = df_stream.select(
    "document_text",
    F.expr(f"""
      ai_query(
        'databricks-meta-llama-3-1-8b-instruct',
        CONCAT('{prompt}', document_text)
      )
    """).alias("summary")
)

Étape 3 : Écrire les résultats récapitulatifs​

Écrivez la sortie récapitulative dans une autre table Delta

Python

# Time-based triggers apply, but only the first trigger processes all existing static data.
query = df_transformed.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/tmp/checkpoints/_docs_summary") \
    .outputMode("append") \
    .toTable("enterprise.docs_summary")

query.awaitTermination()
Sur cette page