Pular para o conteúdo principal
Última atualização em

Tutorial: Enriquecer entidades extraídas de documentos

info

Beta

Esse recurso está em Beta. Os administradores do workspace podem controlar o acesso a esse recurso na página Previews . Consulte Gerenciar prévias do Databricks.

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.

Nesta página