チュートリアル: ドキュメントから抽出されたエンティティをエンッチメントする
ベータ版
この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。
エンタープライズ文書コレクションでは、アナリティクスやダウンストリーム アプリケーションに必要なクリーンで最新の属性を含めずに企業が言及されることがよくあります。チュートリアル: 非構造化契約を裏付けのある企業レコードに変換するエンドツーエンドの AI Functions パイプラインを構築します。パイプラインは、ai_parse_document と ai_extract を使用して、各契約で指定されている企業を識別します。次に、ai_enrich を使用して企業を特定し、ウェブから最新の情報を追加します。
完了したパイプラインでは、次のステップが実行されます。
PDF contracts -> ai_parse_document -> ai_extract -> ai_enrich
| |
company name grounded record
要件
- Databricks Runtime 18.2 以降。
- Serverlessコンピュートを使用する場合は、Serverless環境バージョン3以降を使用してください。
- ワークスペース管理者が [プレビュー (Previews)] ページから有効にした
ai_enrichベータ版。 - Databricks でのウェブ検索をサポートするワークスペースとリージョン。
samplesカタログへのアクセス。
このチュートリアルでは、/Volumes/samples/sec/contracts/ 内の SEC 提出済みの合意文書を使用します。samples.sec.contracts ボリュームは、defaultですべてのワークスペースで使用できます。これらの契約は、実際の文書からエンティティへのパイプラインに現れる提出書類、Eメール、通話のトランスクリプト、およびウェブサイトの代わりとなります。独自の PDF を処理するには、SOURCE_PATH をファイルが含まれる Unity Catalog ボリュームに変更します。
ステップ 1: サンプル契約を取り込む
Python ノートブックを作成し、サポートされているコンピュートにアタッチします。次のコードを実行して、サンプルボリュームから最大 10 件のコンサルティング契約を読み込みます。
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"))
ステップ2：会社名を抽出する
まず、ai_parse_document を使用して各 PDF を構造化された VARIANT に変換し、後続のアクションで関数が再度呼び出されないように、一時テーブルに結果のマテリアライズを行います。
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)
次に、ai_extract を使用して各契約をスパースエンティティレコードに変換します。このスキーマは、抽出を検査するために会社名と十分な契約コンテキストを要求します。
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))
ステップ 3: 各企業を解決してエンリッチする
抽出すべき会社名を ai_enrich に渡します。型付きスキーマにより、結果を後続処理に適した形にすることができます。指示には、値生成の前に法的エンティティを解決し、利用可能な証拠が不十分な場合に null を返すように関数へ伝える内容が含まれています。
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)
ステップ 4: 値とグラウンディング ソースを検査する
By default, response 内の各フィールドには、型指定された value と rationale が含まれます。グラウンドされた行の場合、metadata.sourcesにはその行に使用されたソースドキュメント識別子が格納されます。グラウンディングのプロセナンスは、個別のフィールドではなく行全体に適用されます。
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)
関数がフィールド値をサポートできない場合、value は null になります。推測した値に置き換えるのではなく、その区別を維持してください。
ガバナンスされたデータをナレッジソースとして追加する
独自データで同じエンリッチメントをグラウンディングするには、vector_search エントリをナレッジ ソースの配列に追加します。AI Search インデックスの 3 レベルの名前と、そのテキストおよびドキュメント URI 列を設定します。
[
{ "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"
}
}
]
複数の AI Search インデックスを構成できますが、1 回の呼び出しで構成できるウェブ検索ソースは最大 1 つです。本番運用のワークロードでは、返された根拠とソースを確認し、null 率、エラー、ソースの品質、およびエンティティ解決の精度を監視します。