Tutorial: Enrich entities extracted from documents

info

Bêta

Cette fonctionnalité est en version bêta. Les administrateurs du Workspace peuvent contrôler l’accès à cette fonctionnalité depuis la page Previews . Consultez Gérer les aperçus Databricks.

Ouvrir dans Databricks

Les collections de documents d’entreprise mentionnent souvent des sociétés sans inclure les attributs propres et actuels nécessaires à l’analytique et aux applications en aval. Ce tutoriel crée un pipeline AI Functions de bout en bout qui transforme des contrats non structurés en dossiers d’entreprise fondés sur des données. Le pipeline utilise ai_parse_document et ai_extract pour identifier l’entreprise nommée dans chaque contrat. Il utilise ensuite ai_enrich pour identifier l’entreprise et ajouter les informations actuelles provenant du Web.

Le pipeline exécuté réalise les étapes suivantes :

Text

PDF contracts -> ai_parse_document -> ai_extract -> ai_enrich
                                          |             |
                                   company name    grounded record

Conditions requises

• Databricks Runtime 18.2 ou une version ultérieure.
• Si vous utilisez le compute Serverless, la version d'environnement serverless 3 ou supérieure.
• La fonctionnalité ai_enrich bêta activée par un administrateur de workspace depuis la page Previews .
• Un Workspace et une région qui prennent en charge la recherche web sur Databricks.
• Accès au catalogue samples.

Ce didacticiel utilise des accords déposés auprès de la SEC dans /Volumes/samples/sec/contracts/. Le volume samples.sec.contracts est disponible par default dans tous les workspaces. Ces contrats remplacent les dossiers, e-mails, transcriptions d'appels et sites Web qui apparaissent dans les pipelines réels de document à entité. Pour traiter vos propres fichiers PDF, remplacez SOURCE_PATH par un volume Unity Catalog contenant vos fichiers.

Étape 1 : ingérer des contrats d’échantillon

Créez un notebook Python et associez-le à un compute pris en charge. Exécutez le code suivant pour lire jusqu'à 10 contrats de conseil du volume d'exemple :

Python

from pyspark.sql import functions as F
import json
import uuid

SOURCE_PATH = "/Volumes/samples/sec/contracts/"
TMP_SUFFIX = uuid.uuid4().hex[:8]

raw_contracts_df = (
    spark.read.format("binaryFile")
    .load(SOURCE_PATH)
    .filter(F.lower(F.col("path")).contains("consult"))
    .orderBy("path")
    .limit(10)
)

display(raw_contracts_df.select("path", "length", "modificationTime"))

Étape 2 : extraction du nom de l’entreprise

Tout d’abord, utilisez ai_parse_document pour convertir chaque fichier PDF en un élément VARIANT structuré, puis matérialisez le résultat dans une table temporaire afin que les actions ultérieures n’invoquent plus la fonction.

Python

parsed_contracts_df = raw_contracts_df.select(
    "path",
    F.expr("ai_parse_document(content, MAP('version', '2.0'))").alias("parsed_content"),
)

parsed_table = f"_tmp_ai_enrich_parsed_{TMP_SUFFIX}"
parsed_contracts_df.write.mode("overwrite").saveAsTable(parsed_table)

Then use ai_extract to convert each contract into a sparse entity record. The schema requests the company name and enough contract context to inspect the extraction.

Python

extraction_schema = json.dumps(
    {
        "company_name": {
            "type": "string",
            "description": "Legal name of the company engaging the consultant.",
        },
        "consultant_name": {
            "type": "string",
            "description": "Legal name of the consultant or consulting firm.",
        },
        "effective_date": {
            "type": "string",
            "description": "Contract start date.",
        },
    }
).replace("'", "\\'")

extracted_companies_df = (
    spark.table(parsed_table)
    .filter("TRY_CAST(parsed_content:error_status AS STRING) IS NULL")
    .select(
        "path",
        F.expr(
            f"""
            ai_extract(
              parsed_content,
              '{extraction_schema}',
              MAP('instructions', 'Extract concise values. Return null when a value is absent.')
            )
            """
        ).alias("extracted"),
    )
    .select(
        "path",
        F.expr("extracted:response.company_name::STRING").alias("company_name"),
        F.expr("extracted:response.consultant_name::STRING").alias("consultant_name"),
        F.expr("extracted:response.effective_date::STRING").alias("effective_date"),
    )
    .filter(F.col("company_name").isNotNull())
)

extracted_table = f"_tmp_ai_enrich_entities_{TMP_SUFFIX}"
extracted_companies_df.write.mode("overwrite").saveAsTable(extracted_table)
display(spark.table(extracted_table))

Étape 3 : résoudre et enrichir chaque entreprise

Transmettez le nom de l’entreprise extrait à ai_enrich. Le schéma typé rend les résultats adaptés au traitement en aval. Les instructions indiquent à la fonction de résoudre l’entité juridique avant de générer des valeurs et de renvoyer null lorsque les preuves disponibles sont insuffisantes.

Python

enrichment_schema = json.dumps(
    {
        "industry": {"type": "string", "description": "Primary industry."},
        "headquarters_country": {
            "type": "string",
            "description": "Country of the current headquarters.",
        },
        "official_website": {
            "type": "string",
            "description": "Canonical URL of the official company website.",
        },
        "is_currently_active": {
            "type": "boolean",
            "description": "Whether the legal entity or its clear successor is currently operating.",
        },
    }
).replace("'", "\\'")

enriched_companies_df = spark.table(extracted_table).select(
    "path",
    "company_name",
    "consultant_name",
    "effective_date",
    F.expr(
        f"""
        ai_enrich(
          company_name,
          '{enrichment_schema}',
          PARSE_JSON('[{{"type":"web_search","config":{{}]'),
          MAP(
            'instructions',
            'Resolve the exact legal entity before enriching it. Prefer official and authoritative sources. If identity is ambiguous or evidence is insufficient, return null rather than guessing.'
          )
        )
        """
    ).alias("enrichment"),
)

enriched_table = f"_tmp_ai_enrich_results_{TMP_SUFFIX}"
enriched_companies_df.write.mode("overwrite").saveAsTable(enriched_table)

Étape 4 : inspecter les valeurs et les sources d’ancrage

Par default, chaque champ de response contient un élément value typé et un élément rationale. Pour les lignes ancrées, metadata.sources contient les identifiants du document source utilisés pour la ligne. La provenance de l’ancrage s’applique à l’ensemble de la ligne, et non aux champs individuels.

Python

final_df = spark.table(enriched_table).select(
    "path",
    "company_name",
    F.expr("enrichment:response.industry.value::STRING").alias("industry"),
    F.expr("enrichment:response.headquarters_country.value::STRING").alias("headquarters_country"),
    F.expr("enrichment:response.official_website.value::STRING").alias("official_website"),
    F.expr("enrichment:response.is_currently_active.value::BOOLEAN").alias("is_currently_active"),
    F.expr("enrichment:response.official_website.rationale::STRING").alias("website_rationale"),
    F.expr("enrichment:metadata.sources").alias("grounding_sources"),
    F.expr("enrichment:error_message::STRING").alias("error_message"),
)

display(final_df)

Lorsque la fonction ne prend pas en charge une valeur de champ, value est null. Préservez cette distinction au lieu de la remplacer par une valeur estimée.

Ajouter des données gouvernées comme source de connaissances

Pour baser le même enrichissement sur des données propriétaires, ajoutez une entrée vector_search au tableau des sources de connaissances. Configurez le nom à trois niveaux d’un index AI Search et ses colonnes de texte et d’URI de document :

JSON

[
  { "type": "web_search", "config": {} },
  {
    "type": "vector_search",
    "description": "Governed company profiles",
    "config": {
      "index_name": "prod_catalog.crm.company_kb",
      "text_col": "profile_text",
      "doc_uri_col": "source_url"
    }
  }
]

Vous pouvez configurer plusieurs index AI Search, mais au maximum une source de recherche web par appel. Pour les charges de travail de production, examinez les justifications et sources renvoyées, et surveillez les taux de valeurs nulles, les erreurs, la qualité des sources ainsi que la précision de la résolution d'entités.