Tutorial: Enriquecer entidades extraídas de documentos

info

Beta

This recurso is in Beta. To use it, a workspace admin must turn on Ai Enrich from the Previews page. See Gerenciar Databricks previews.

Open in Databricks

As coleções de documentos empresariais geralmente mencionam empresas sem incluir os atributos limpos e atuais necessários para analíticas e aplicações downstream. Este tutorial cria um pipeline de AI Functions de ponta a ponta que transforma contratos não estruturados em registros corporativos fundamentados. O pipeline usa ai_parse_document e ai_extract para identificar a empresa nomeada em cada contrato. Em seguida, ele usa ai_enrich para resolver a empresa e adicionar informações atuais da web.

O pipeline concluído executa estes passos:

Text

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

Requisitos

• Databricks Runtime 18.2 ou acima.
• Se você usar o compute Serverless, versão 3 ou acima do ambiente serverless.
• O ai_enrich Beta ativado por um administrador do workspace na página Pré-visualizações .
• Um Workspace e região que ofereçam suporte à pesquisa na Web no Databricks.
• Acesso ao catálogo samples.

Este tutorial usa contratos protocolados na SEC em /Volumes/samples/sec/contracts/. O volume samples.sec.contracts está disponível em todos os workspaces por default. Estes contratos substituem os registros, emails, transcrições de chamadas e sites que aparecem em pipelines reais de documentos para entidades. Para processar seus próprios PDFs, altere SOURCE_PATH para um volume do Unity Catalog que contenha seus arquivos.

Passo 1: Ingerir contratos de amostra

Crie um notebook Python e anexe-o ao compute compatível. Execute o código a seguir para ler até 10 contratos de consultoria do volume de amostra:

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"))

o passo 2: Extract the company name

Primeiro, use ai_parse_document para converter cada PDF em um VARIANT estruturado e materialize o resultado em uma tabela temporária para que as ações posteriores não invoquem a função novamente.

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)

Em seguida, use ai_extract para converter cada contrato em um registro de entidade esparsa. O esquema solicita o nome da empresa e contexto de contrato suficiente para inspecionar a extração.

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))

Etapa 3: Resolver e enriquecer cada empresa

Passe o nome da empresa extraído para ai_enrich. O esquema tipado torna os resultados adequados para processamento posterior. As instruções dizem à função para resolver a entidade legal antes de gerar valores e para retornar null quando as evidências disponíveis forem insuficientes.

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)

Passo 4: Inspecionar valores e fontes de fundamentação

By default, each field in response contains a typed value and a rationale. For grounded rows, metadata.sources contains the source document identifiers used for the row. Grounding provenance applies to the entire row, not to individual fields.

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)

Quando a função não puder dar suporte a um valor de campo, value será null. Preserve essa distinção em vez de substituí-la por um valor estimado.

Adicionar dados governados como fonte de conhecimento

Para embasar o mesmo enriquecimento em dados proprietários, adicione uma entrada vector_search à matriz de fontes de conhecimento. Configure o nome de três níveis de um índice de Pesquisa de IA e suas colunas de texto e URI do documento:

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"
    }
  }
]

Você pode configurar vários índices de Pesquisa de IA, mas no máximo uma fonte de pesquisa na web por chamada. Para cargas de trabalho de produção, revise as justificativas e fontes retornadas, e monitore as taxas de valores nulos, erros, qualidade da fonte e precisão da resolução de entidades.