Aller au contenu principal
Dernière mise à jour le

Tutorial: Enrich entities extracted from documents

info

Bêta

Cette fonctionnalité est en version bêta. Pour l'utiliser, un administrateur du workspace doit activer Ai Enrich 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 un fournisseur de recherche externe en tant que source de connaissances​

Pour baser l’enrichissement sur un fournisseur de recherche auprès duquel vous avez déjà un compte, tel que You.com ou Exa, ajoutez une entrée mcp au tableau des sources de connaissances, en plus ou à la place de l’entrée web_search. Une source mcp appelle un outil de recherche sur un serveur MCP (Model Context Protocol) externe via Unity Gateway. Configurez le nom complet du MCP et l’outil de recherche à appeler :

JSON
[
  {
    "type": "mcp",
    "description": "You.com web search",
    "config": {
      "mcp_service": "main.default.youcom",
      "tool": "you-search"
    }
  }
]

Avant d’exécuter l’enrichissement avec une source mcp :

  1. Enregistrez le serveur MCP du fournisseur dans Unity Gateway. Voir Serveurs MCP externes. Pour You.com, l’installation du listing Databricks Marketplace crée le MCP pour vous.
  2. Utilisez une connexion avec une identité partagée, telle que la clé d’API du fournisseur. ai_enrich ne peut pas demander aux utilisateurs de se connecter au fournisseur, donc si la connexion exige que chaque utilisateur se connecte, la source mcp échouera pour les utilisateurs qui ne se sont pas connectés.
  3. Attribuez EXECUTE sur le MCP, ainsi que USE CATALOG et USE SCHEMA sur son catalogue parent et son schéma, à l’identité qui exécute l’enrichissement : vous-même ou l’identité d’exécution d’un job ou d’un pipeline. Voir Partager un MCP.

ai_enrich appelle l'outil une fois par ligne avec un contenu non vide. La query de recherche est générée à partir du contenu de la ligne et envoyée au fournisseur ; traitez donc ce contenu comme des données qui quittent Databricks. Pour toutes les clés de configuration, y compris l'argument objective pour Exa et pour le comportement en cas d'échec, consultez Configuration MCP.

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 de recherche IA et sources MCP, mais au maximum une source de recherche web, par appel. Pour les charges de travail de production, examinez les justifications et les sources renvoyées, et surveillez les taux de nullité, les erreurs, la qualité des sources et la précision de la résolution des entités.

Sur cette page