document-intelligence

ドキュメントインテリジェンス

Snowflake Cortex AI を使用するすべてのドキュメント処理タスクのエントリーポイントです。

参照ファイル

このスキルでは、関数の詳細なガイダンスとして参照ドキュメントを使用します。

参照先	場所	用途
AI_EXTRACT	reference/ai-extract.md	構造化抽出（フィールド、テーブル）
AI_PARSE_DOCUMENT	reference/ai-parse-doc.md	コンテンツ全体の解析（テキスト、レイアウト、画像）

ユーザーが特定の関数の詳細を必要としている場合は、参照コンテキストを読み込みます。

• 構造化抽出の場合 → reference/ai-extract.md を読み込みます
• コンテンツ解析の場合 → reference/ai-parse-doc.md を読み込みます

使用する状況

• ユーザーがドキュメント（PDF、画像、Office ファイル）からデータを抽出したい場合
• ユーザーが AI_EXTRACT または AI_PARSE_DOCUMENT に言及した場合
• ユーザーが請求書、契約書、フォーム、レポートを処理したい場合
• ユーザーがグラフ、設計図、図表を分析する必要がある場合
• ユーザーがドキュメント処理パイプラインを構築したい場合

ワークフロー

Start
  ↓
Step 1: Gather Document Info
  ↓
  ├─→ Local file → PUT to Snowflake stage
  │
  ├─→ Other storage (GDrive, Dropbox, etc.) → Load openflow skill
  │                                              ↓
  │                                           Setup connector
  │                                              ↓
  │                                           Return here
  ↓
Step 2: Determine Extraction Goal
  ↓
  ├─→ Structured fields/tables ─────────────────────────────────────┐
  │                                                                 │
  ├─→ Full text with layout → Load ai-parse-doc/SKILL.md (LAYOUT)   │
  │                                                                 │
  ├─→ Text via OCR → Load ai-parse-doc/SKILL.md (OCR)               │
  │                                                                 │
  └─→ Images/charts/blueprints → **Flow C: Visual Analysis**         │
                                                                    ↓
                                            Step 2.5: Test Run & Refinement
                                                                    │
                                              ┌─────────────────────┴──────────────────────┐
                                              ↓                                            │
                                        Select test file                                   │
                                              ↓                                            │
                                        Run AI_EXTRACT on single file                      │
                                              ↓                                            │
                                        Review results                                     │
                                              ↓                                            │
                                        Satisfied? ──Yes──→ Choose next action             │
                                              │              │                             │
                                              No             ├─→ Batch extraction ─────────┘
                                              ↓              ├─→ Store result
                                        Ask expected answer  ├─→ Add more fields
                                              ↓              └─→ Done
                                        LLM-judge refines prompt
                                              ↓
                                        Retry test (max 3x) ─────────────────────────┐
                                              │                                      │
                                              └──────────────────────────────────────┘
  ↓
Step 3: Post-Processing Options
  ↓
  ├─→ Store results → Create table / Create pipeline
  │
  └─→ Build RAG/search integration

ステップ 1：ドキュメント情報を収集する

ask_user_question を使用して、ドキュメントについてユーザーに質問します。

質問 1 - ファイル形式：

What type of file do you want to process?
Options:
1. PDF
2. Image (PNG, JPEG, TIFF)
3. Office document (DOCX, PPTX)
4. HTML or TXT file
5. CSV file - AI_EXTRACT only
6. MD or EML file - AI_EXTRACT only
7. Excel file (XLSX, XLS) - not supported
8. Legacy Office (DOC, PPT) - needs conversion

対応形式の概要：

• AI_EXTRACT が対応： PDF、画像（PNG、JPEG、TIFF）、DOCX、PPTX、HTML、TXT、CSV、MD、EML
• AI_PARSE_DOCUMENT が対応： PDF、画像（PNG、JPEG、TIFF）、DOCX、PPTX、HTML、TXT

ユーザーが「CSV ファイル」を選択した場合： CSV ファイルに対応しているのは AI_EXTRACT のみです（AI_PARSE_DOCUMENT には対応していません）。抽出の目的を確認するため、ステップ 2 に進みます。

ユーザーが「MD または EML ファイル」を選択した場合： MD および EML ファイルに対応しているのは AI_EXTRACT のみです（AI_PARSE_DOCUMENT には対応していません）。抽出の目的を確認するため、ステップ 2 に進みます。

ユーザーが「HTML または TXT ファイル」を選択した場合： HTML および TXT ファイルは、AI_EXTRACT と AI_PARSE_DOCUMENT の両方に対応しています。抽出の目的を確認するため、ステップ 2 に進みます。

ユーザーが「レガシー Office（DOC、PPT）」を選択した場合：

Legacy Office formats (DOC, PPT) are not directly supported.

Please convert to modern formats:
- DOC → DOCX (Save As in Word)
- PPT → PPTX (Save As in PowerPoint)
- Or export as PDF

Then upload the converted file and continue.

ユーザーが「Excel ファイル（XLSX、XLS）」を選択した場合：

Excel ファイルは AI_EXTRACT または AI_PARSE_DOCUMENT で直接サポートされていません。次の選択肢をユーザーに案内します。

Excel files (.xlsx, .xls) are not directly supported by the document AI functions.

You have two options:

Option 1: Convert to PDF
- Export the Excel file as PDF from Excel/Google Sheets
- Then upload the PDF and use AI_EXTRACT or AI_PARSE_DOCUMENT

Option 2: Load directly into Snowflake (Recommended for tabular data)
- Snowflake has native Excel support for loading data into tables:

-- Infer schema from Excel file
SELECT * FROM TABLE(
  INFER_SCHEMA(
    LOCATION => '@my_stage/data.xlsx',
    FILE_FORMAT => 'TYPE=EXCEL'
  )
);

-- Query Excel file directly  
SELECT * FROM @my_stage/data.xlsx 
(FILE_FORMAT => 'TYPE=EXCEL', SHEET_NAME => 'Sheet1');

-- Load into table
COPY INTO my_table FROM @my_stage/data.xlsx
FILE_FORMAT = (TYPE = EXCEL)
MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

Which option would you like to proceed with?

ユーザーがオプション 1（PDF に変換）を選択した場合は、PDF をアップロードするよう案内し、通常のフローを続けます。

ユーザーがオプション 2（直接読み込む）を選択した場合は、上記の Snowflake SQL コマンドを使って作業できるよう支援します。

質問 2 - ファイルの場所：

Where is your file stored?
Options:
1. Internal Snowflake stage (e.g., @my_db.my_schema.my_stage/file.pdf)
2. External stage (S3, Azure Blob, GCS)
3. Local file on my computer
4. Other storage (Google Drive, Dropbox, SharePoint, OneDrive, Box)
5. Need help setting up a stage

ユーザーが「コンピューター上のローカルファイル」を選択した場合：

PUT コマンドを使って、Snowflake ステージにファイルをアップロードできるよう支援します。

1. まず、ステージが存在することを確認します（または作成します）。

-- Create a stage if needed (with server-side encryption)
CREATE STAGE IF NOT EXISTS my_db.my_schema.doc_stage
  DIRECTORY = (ENABLE = TRUE)
  ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE')
  COMMENT = 'Stage for document processing';

2. ファイルパスをユーザーに尋ねます。

What is the full path to your local file?
Example: /Users/username/Documents/invoice.pdf
         C:\Users\username\Documents\invoice.pdf

3. PUT コマンドを生成します（SnowSQL、Snowflake CLI、または Python コネクター経由で実行する必要があります）。

-- Using SnowSQL or Snowflake CLI
PUT file:///path/to/your/file.pdf @my_db.my_schema.doc_stage AUTO_COMPRESS=FALSE;

ユーザーのツールに応じて手順を案内します。

オプション A - SnowSQL：

# Install SnowSQL if not installed
# https://docs.snowflake.com/en/user-guide/snowsql-install-config

# Connect and upload
snowsql -a <account> -u <username>
PUT file:///path/to/file.pdf @my_db.my_schema.doc_stage AUTO_COMPRESS=FALSE;

オプション B - Snowflake CLI（snow）：

# Upload using Snowflake CLI
snow stage copy /path/to/file.pdf @my_db.my_schema.doc_stage --overwrite

オプション C - Python：

from snowflake.connector import connect

conn = connect(account='...', user='...', password='...')
cursor = conn.cursor()

# Upload file
cursor.execute("""
    PUT file:///path/to/file.pdf @my_db.my_schema.doc_stage 
    AUTO_COMPRESS=FALSE OVERWRITE=TRUE
""")

print("File uploaded successfully!")
conn.close()

オプション D - 複数ファイル：

-- Upload all PDFs from a directory
PUT file:///path/to/folder/*.pdf @my_db.my_schema.doc_stage AUTO_COMPRESS=FALSE;

4. アップロードを確認します。

-- List files in stage
LIST @my_db.my_schema.doc_stage;

-- Or using directory table
SELECT * FROM DIRECTORY(@my_db.my_schema.doc_stage);

5. ディレクトリテーブルを更新します（DIRECTORY クエリに必要です）。

ALTER STAGE @my_db.my_schema.doc_stage REFRESH;

ファイルがアップロードされたら、ステージパスを使った抽出を続けます。

ユーザーが「その他のストレージ」を選択した場合：

• まずファイルを Snowflake ステージに取り込む必要があることを伝えます
• ストレージプロバイダー用のコネクター設定を支援するため、openflow スキルを読み込みます
• openflow の設定が完了したら、ここに戻って抽出を続けます

質問 3 - コンテンツの種類：

What kind of content is in your document?
Options:
1. Structured forms (invoices, receipts, applications)
2. Reports/articles with text and tables
3. Charts, graphs, blueprints, diagrams
4. General text (manuals, policies, contracts)

ステップ 2：抽出の目的を特定する

抽出したいものをユーザーに尋ねます。

What do you want to extract?
Options:
1. Specific fields (names, dates, amounts, IDs)
2. Tables with defined columns
3. Full text with layout preservation (tables, headers, structure)
4. Text only via OCR (quick text extraction)
5. Images and visual content for analysis (charts, blueprints)

回答に応じて振り分けます。

ユーザーの選択	振り分け先	関数
特定のフィールド、テーブル	ステップ 2.5（まずテスト）→ 次に ai-extract/SKILL.md	AI_EXTRACT
レイアウトを含む全文	ステップ 2.4（ページの最適化）→ ai-parse-doc/SKILL.md	AI_PARSE_DOCUMENT（LAYOUT モード）
OCR によるテキストのみ	ステップ 2.4（ページの最適化）→ ai-parse-doc/SKILL.md	AI_PARSE_DOCUMENT（OCR モード）
画像、グラフ、設計図	ステップ 2.6（視覚的分析）	AI_COMPLETE（直接処理、解析なし）

ステップ 2.4：ページの最適化（AI_PARSE_DOCUMENT の場合）

LAYOUT モードまたは OCR モードを選択した場合のみ適用されます（Visual Analysis は対象外）。

解析する前に、パフォーマンスの向上とコストの削減につながるページの最適化についてユーザーに確認します。

ユーザーに質問します：

Would you like to parse the entire document or specific pages?
Options:
1. Parse entire document
2. Parse specific page range (e.g., pages 1-50)
3. Parse specific pages (e.g., pages 1, 5, 10, 25)
4. I'm not sure - help me decide

「ドキュメント全体を解析する」の場合：

-- Parse all pages with page_split for easier processing
SELECT AI_PARSE_DOCUMENT(
  TO_FILE('@stage', 'document.pdf'),
  {'mode': 'LAYOUT', 'page_split': true}
) AS parsed;

「特定のページ範囲を解析する」の場合：

範囲を尋ねます：

What page range would you like to parse?
Example: 1-50 (parses pages 1 through 50)
Note: Page numbers are 1-indexed for your input, converted to 0-indexed internally.

-- Parse specific range (page_filter uses 0-indexed pages)
-- For user input "1-50", use start: 0, end: 50
SELECT AI_PARSE_DOCUMENT(
  TO_FILE('@stage', 'document.pdf'),
  {'mode': 'LAYOUT', 'page_filter': [{'start': 0, 'end': 50}]}
) AS parsed;

「特定のページを解析する」の場合：

ページ番号を尋ねます：

Which specific pages would you like to parse?
Example: 1, 5, 10, 25 (parses only these pages)
Note: Page numbers are 1-indexed for your input, converted to 0-indexed internally.

-- Parse specific pages (0-indexed internally)
-- For user input "1, 5, 10, 25", use [0, 4, 9, 24]
SELECT AI_PARSE_DOCUMENT(
  TO_FILE('@stage', 'document.pdf'),
  {'mode': 'LAYOUT', 'page_filter': [0, 4, 9, 24]}
) AS parsed;

「よくわかりません」の場合：

ガイダンスを提供します：

Here's how to decide:

• Parse entire document: Best for full text extraction, RAG pipelines, 
  or when you need all content. Works well for documents ≤500 pages.

• Parse specific page range: Best when you know the relevant section 
  (e.g., "chapters 1-3 are on pages 1-75"). Faster and cheaper.

• Parse specific pages: Best for extracting just a table of contents, 
  specific chapters, or known important pages.

Tips:
- First page often has title, author, date
- Last few pages often have references, appendix
- Table of contents usually in first 5 pages

Would you like me to first get the page count so you can decide?

ユーザーがページ数を知りたい場合：

-- Quick page count check (minimal processing)
SELECT AI_PARSE_DOCUMENT(
  TO_FILE('@stage', 'document.pdf'),
  {'mode': 'OCR', 'page_filter': [{'start': 0, 'end': 1}]}
):pageCount AS total_pages;

その後、選択したページオプションを使って reference/ai-parse-doc.md を読み込みます。

ステップ 2.5：テスト実行とプロンプトの改善（構造化抽出の場合）

「Specific fields」または「Tables」を選択した場合のみ適用されます。

このステップでは、LLM-as-judge による反復的なプロンプトの改善を通じて、バッチ処理の前に単一ファイルで抽出結果を検証します。

2.5a：初期抽出フィールドの定義

抽出するフィールドをユーザーに尋ねます：

What fields do you want to extract? Please provide field names and descriptions.
Example: 
- invoice_number: The invoice or bill number
- vendor_name: Name of the vendor or supplier
- total_amount: The total amount due

初期の responseFormat を作成します：

responseFormat => {
  'invoice_number': 'What is the invoice number?',
  'vendor_name': 'What is the vendor name?',
  'total_amount': 'What is the total amount?'
}

2.5b：テストファイルの選択

ask_user_question を使ってユーザーに尋ねます：

How would you like to select the test file?
Options:
1. Use the first file in the stage
2. Let me pick a specific file from the stage
3. I'll provide the exact file path

「最初のファイル」の場合：

-- Get first file from stage
SELECT relative_path 
FROM DIRECTORY(@db.schema.stage) 
WHERE relative_path LIKE '%.pdf' 
LIMIT 1;

「stage から選択」の場合：

-- List available files
SELECT relative_path, size, last_modified
FROM DIRECTORY(@db.schema.stage)
WHERE relative_path LIKE '%.pdf'
ORDER BY last_modified DESC
LIMIT 10;

次に、リストから選択するようユーザーに尋ねます。

2.5c：テスト抽出の実行

単一のテストファイルに対して AI_EXTRACT を実行します：

-- Test extraction on single file
SELECT AI_EXTRACT(
  file => TO_FILE('@db.schema.stage', '<selected_test_file>'),
  responseFormat => {
    'field1': 'question1',
    'field2': 'question2'
    -- ... user-defined fields
  }
) AS test_result;

読みやすい形式で結果をユーザーに表示します：

SELECT 
  test_result:response:field1::STRING AS field1,
  test_result:response:field2::STRING AS field2
FROM (SELECT AI_EXTRACT(...) AS test_result);

2.5d：結果の確認

ask_user_question を使ってユーザーに尋ねます：

Here are the test extraction results:

| Field | Extracted Value |
|-------|-----------------|
| invoice_number | INV-2024-001 |
| vendor_name | Acme Corp |
| total_amount | $1,234.56 |

Are you satisfied with these results?
Options:
1. Yes, proceed with full extraction
2. No, some fields need improvement
3. No, start over with different fields

「はい」の場合： 検証済みの responseFormat を使ってバッチ全体を抽出するため、ai-extract/SKILL.md を読み込みます。

「最初からやり直す」の場合： ステップ 2.5a に戻ります。

「一部のフィールドを改善する必要がある」の場合： 改善ループに進みます。

2.5e：プロンプト改善ループ（LLM-as-Judge）

改善が必要なフィールドをユーザーに尋ねます：

Which field(s) gave incorrect results?
(Select all that apply)

問題のあるフィールドごとに、次のように尋ねます：

For the field "[field_name]":
- Current extraction: "[actual_result]"
- What should the correct value be?

Claude を LLM-as-judge として使い、プロンプトを改善します：

-- LLM-as-judge prompt refinement
SELECT AI_COMPLETE(
  'claude-3-5-sonnet',
  $You are an expert at crafting extraction prompts for document AI.

TASK: Improve an extraction question that didn't produce the expected result.

FIELD: [field_name]
ORIGINAL QUESTION: "[original_question]"
EXTRACTED VALUE: "[actual_result]"  
EXPECTED VALUE: "[expected_result]"
DOCUMENT TYPE: [document_type from Step 1]

Analyze why the original question may have failed and provide an improved question that would more accurately extract the expected value.

Consider:
- Is the question too vague or too specific?
- Should it reference a specific location in the document?
- Should it specify the format of the expected answer?
- Are there synonyms or alternative phrasings that might work better?

Return ONLY the improved question, nothing else.$
) AS improved_question;

**改善した質問で responseFormat を更新し、**テスト抽出を再試行します。

2.5f：反復回数の上限

• フィールドごとの改善は最大 3 回
• 3 回試した後、ユーザーに尋ねます：

  We've tried 3 prompt variations for "[field_name]" without success.
  Options:
  1. Continue anyway with best result so far
  2. Skip this field for now
  3. Let me manually specify the extraction question
  4. Try fallback method (AI_PARSE_DOCUMENT + AI_COMPLETE)
  

「フォールバック方式を試す」の場合： 以下のステップ 2.5f-fallbackに進みます。

---

2.5f-fallback：AI_PARSE_DOCUMENT + AI_COMPLETE の構造化出力

プロンプトを改善しても AI_EXTRACT で正確な結果が得られない場合は、この代替方法を使います：

1. AI_PARSE_DOCUMENT を使ってドキュメントのテキストを解析します
2. AI_COMPLETE と構造化出力を使って、解析したテキストからフィールドを抽出します

この方法は、AI_EXTRACT でうまく処理できない複雑なドキュメントで、より効果的な場合があります。

ステップ 1：ドキュメントを解析してテキストの内容を取得する

-- Parse document using OCR or LAYOUT mode
SELECT AI_PARSE_DOCUMENT(
  TO_FILE('@db.schema.stage', '<test_file>'),
  {'mode': 'LAYOUT'}
):content::STRING AS document_text;

ステップ 2：構造化出力用の JSON スキーマを作成する

ユーザーのフィールドを JSON スキーマに変換します：

-- Example: For fields invoice_number, vendor_name, total_amount
-- Build this schema:
{
  'type': 'json',
  'schema': {
    'type': 'object',
    'properties': {
      'invoice_number': {
        'type': 'string',
        'description': 'The invoice or bill number'
      },
      'vendor_name': {
        'type': 'string', 
        'description': 'Name of the vendor or supplier'
      },
      'total_amount': {
        'type': 'string',
        'description': 'The total amount due'
      }
    },
    'required': ['invoice_number', 'vendor_name', 'total_amount']
  }
}

ステップ 3：構造化出力を使って AI_COMPLETE で抽出する

-- Complete extraction with structured output
WITH parsed_doc AS (
  SELECT AI_PARSE_DOCUMENT(
    TO_FILE('@db.schema.stage', '<test_file>'),
    {'mode': 'LAYOUT'}
  ):content::STRING AS document_text
)
SELECT AI_COMPLETE(
  'claude-3-5-sonnet',
  'Extract the following information from this document text. Return ONLY the requested fields in JSON format.

DOCUMENT TEXT:
' || document_text || '

Extract these fields:
- invoice_number: The invoice or bill number
- vendor_name: Name of the vendor or supplier  
- total_amount: The total amount due',
  {
    'response_format': {
      'type': 'json',
      'schema': {
        'type': 'object',
        'properties': {
          'invoice_number': {'type': 'string', 'description': 'The invoice or bill number'},
          'vendor_name': {'type': 'string', 'description': 'Name of the vendor or supplier'},
          'total_amount': {'type': 'string', 'description': 'The total amount due'}
        },
        'required': ['invoice_number', 'vendor_name', 'total_amount']
      }
    },
    'max_tokens': 4096
  }
) AS extraction_result
FROM parsed_doc;

ステップ 4：結果を AI_EXTRACT と比較する

両方の結果をユーザーに表示します：

Fallback extraction results:

| Field | AI_EXTRACT Result | AI_COMPLETE Result |
|-------|-------------------|-------------------|
| invoice_number | INV-2024-001 | INV-2024-001 |
| vendor_name | Acme | Acme Corporation |
| total_amount | 1234 | $1,234.56 |

Which result is more accurate?
Options:
1. AI_COMPLETE results are better - use fallback method
2. AI_EXTRACT results are better - stick with original
3. Neither is correct - try manual specification

ステップ 5：フォールバックのほうが優れている場合は、バッチ処理に使用する

-- Batch extraction using AI_PARSE_DOCUMENT + AI_COMPLETE
SELECT 
  relative_path AS file_path,
  PARSE_JSON(AI_COMPLETE(
    'claude-3-5-sonnet',
    'Extract the following information from this document text. Return ONLY the requested fields.

DOCUMENT TEXT:
' || AI_PARSE_DOCUMENT(
      TO_FILE('@db.schema.stage', relative_path),
      {'mode': 'LAYOUT'}
    ):content::STRING || '

Extract: invoice_number, vendor_name, total_amount',
    {
      'response_format': {
        'type': 'json',
        'schema': {
          'type': 'object',
          'properties': {
            'invoice_number': {'type': 'string'},
            'vendor_name': {'type': 'string'},
            'total_amount': {'type': 'string'}
          },
          'required': ['invoice_number', 'vendor_name', 'total_amount']
        }
      },
      'max_tokens': 4096
    }
  )) AS extracted_data
FROM DIRECTORY(@db.schema.stage)
WHERE relative_path LIKE '%.pdf';

構造化出力スキーマの例

さまざまなデータ型の場合：

-- String fields
'field_name': {'type': 'string', 'description': 'Description of field'}

-- Number fields  
'amount': {'type': 'number', 'description': 'Numeric amount'}

-- Integer fields
'quantity': {'type': 'integer', 'description': 'Whole number quantity'}

-- Boolean fields
'is_paid': {'type': 'boolean', 'description': 'Whether invoice is paid'}

-- Enum fields (constrained values)
'status': {
  'type': 'string',
  'enum': ['pending', 'approved', 'rejected'],
  'description': 'Document status'
}

-- Array fields
'line_items': {
  'type': 'array',
  'items': {
    'type': 'object',
    'properties': {
      'description': {'type': 'string'},
      'quantity': {'type': 'integer'},
      'unit_price': {'type': 'number'},
      'total': {'type': 'number'}
    },
    'required': ['description', 'quantity', 'unit_price', 'total']
  },
  'description': 'List of line items'
}

-- Nested objects
'address': {
  'type': 'object',
  'properties': {
    'street': {'type': 'string'},
    'city': {'type': 'string'},
    'state': {'type': 'string'},
    'zip': {'type': 'string'}
  },
  'required': ['street', 'city', 'state', 'zip'],
  'description': 'Mailing address'
}

請求書全体の抽出例（フォールバック方式）

WITH parsed_doc AS (
  SELECT 
    relative_path,
    AI_PARSE_DOCUMENT(
      TO_FILE('@db.schema.stage', relative_path),
      {'mode': 'LAYOUT'}
    ):content::STRING AS document_text
  FROM DIRECTORY(@db.schema.stage)
  WHERE relative_path LIKE '%.pdf'
)
SELECT 
  relative_path AS file_path,
  PARSE_JSON(AI_COMPLETE(
    'claude-3-5-sonnet',
    'You are an expert document extraction assistant. Extract the following information from the invoice document below.

DOCUMENT:
' || document_text || '

Extract all requested fields. If a field is not found, use null.',
    {
      'response_format': {
        'type': 'json',
        'schema': {
          'type': 'object',
          'properties': {
            'invoice_number': {'type': 'string', 'description': 'Invoice or bill number'},
            'invoice_date': {'type': 'string', 'description': 'Date of invoice (YYYY-MM-DD format)'},
            'due_date': {'type': 'string', 'description': 'Payment due date (YYYY-MM-DD format)'},
            'vendor': {
              'type': 'object',
              'properties': {
                'name': {'type': 'string'},
                'address': {'type': 'string'},
                'phone': {'type': 'string'},
                'email': {'type': 'string'}
              },
              'required': ['name']
            },
            'customer': {
              'type': 'object', 
              'properties': {
                'name': {'type': 'string'},
                'address': {'type': 'string'}
              },
              'required': ['name']
            },
            'line_items': {
              'type': 'array',
              'items': {
                'type': 'object',
                'properties': {
                  'description': {'type': 'string'},
                  'quantity': {'type': 'number'},
                  'unit_price': {'type': 'number'},
                  'total': {'type': 'number'}
                },
                'required': ['description', 'total']
              }
            },
            'subtotal': {'type': 'number'},
            'tax_amount': {'type': 'number'},
            'total_amount': {'type': 'number'},
            'currency': {'type': 'string', 'description': 'Currency code (e.g., USD, EUR)'},
            'payment_terms': {'type': 'string'}
          },
          'required': ['invoice_number', 'vendor', 'total_amount']
        }
      },
      'max_tokens': 4096
    }
  )) AS invoice_data
FROM parsed_doc;

フォールバック方式を推奨するケース：

シナリオ	推奨事項
複雑なネストデータ（住所、明細項目）	フォールバックを使用します。構造化オブジェクトの処理に適しています
特殊なレイアウトのドキュメント	フォールバックを使用します。LAYOUT モードでは、より多くのコンテキストを取得できます
特定のデータ型（数値、ブール値）が必要	フォールバックを使用します。スキーマによって型が強制されます
単純なフラットフィールド	AI_EXTRACT を使用します。より高速で簡単です
大量のバッチ処理	両方をテストし、精度とコストを比較します

2.5g: 検証後の次のステップ

ユーザーがテスト結果に満足したら、次に何をしたいか尋ねます。

Test extraction validated successfully! What would you like to do next?
Options:
1. Extract from all files in the stage (batch processing)
2. Add more fields to extract
3. Store this single result to a table
4. Try on another test file first
5. Done - I only needed this one file

回答に応じて処理を進めます：

ユーザーの選択	アクション
すべてのファイルから抽出	一括抽出に進みます（ステップ 2.5h）
フィールドを追加	既存のフィールドを保持したままステップ 2.5a に戻ります
この結果を保存	単一の結果用に INSERT 文を生成します
別のテストファイルを試す	同じ responseFormat を使ってステップ 2.5b に戻ります
完了	ワークフローを終了し、最終結果を表示します

「フィールドを追加」の場合：

You currently have these fields defined:
- invoice_number
- vendor_name  
- total_amount

What additional fields would you like to extract?

「この結果を保存」の場合：

ユーザーに尋ねます：

How would you like to store the results?
Options:
1. Just store this single result
2. Store and create a pipeline for future files

「この単一の結果だけを保存」の場合：

重要 — 最初に大きなファイルについて尋ねます：

Can any of your files be longer than the page limit?
- AI_EXTRACT: 125 pages max
- AI_PARSE_DOCUMENT: 500 pages max
Options:
1. No - all files are within the limit
2. Yes - some files may exceed the limit
3. Not sure

次に、ページの最適化について尋ねます（コスト削減と速度向上のため）：

Would you like to process the entire document or only specific pages?
This can significantly reduce processing time and cost.

Options:
1. Process entire document (all pages)
2. Process only the first page (common for invoices/forms with header info)
3. Process specific page range (e.g., pages 1-5)
4. Process specific pages (e.g., pages 1, 3, 10)

ユーザーがページ最適化を選択した場合（オプション 2、3、または 4）：

まず、ページ抽出のストアドプロシージャを作成します（初回のみの設定）：

-- Create stage for extracted pages (with server-side encryption)
CREATE STAGE IF NOT EXISTS db.schema.extracted_pages_stage
  DIRECTORY = (ENABLE = TRUE)
  ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE')
  COMMENT = 'Stage for extracted PDF pages';

-- Create page extraction stored procedure
CREATE OR REPLACE PROCEDURE db.schema.extract_pdf_pages(
    stage_name STRING, 
    file_name STRING, 
    dest_stage_name STRING,
    page_selection VARIANT  -- Can be: integer (single page), array of integers, or object with 'start' and 'end'
)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.9'
PACKAGES = ('snowflake-snowpark-python', 'pypdf2')
HANDLER = 'run'
AS
$
from PyPDF2 import PdfReader, PdfWriter
from snowflake.snowpark import Session
from snowflake.snowpark.file_operation import FileOperation
from io import BytesIO
import os
import json

def run(session: Session, stage_name: str, file_name: str, dest_stage_name: str, page_selection) -> dict:
    result = {
        "status": "success",
        "source_file": file_name,
        "total_pages": 0,
        "extracted_pages": [],
        "output_file": ""
    }
    
    try:
        # Download PDF from stage
        file_url = f"{stage_name}/{file_name}"
        get_result = session.file.get(file_url, '/tmp/')
        file_path = os.path.join('/tmp/', file_name)
        
        if not os.path.exists(file_path):
            return {"status": "error", "message": f"File {file_name} not found in {stage_name}"}

        with open(file_path, 'rb') as f:
            pdf_data = f.read()
            pdf_reader = PdfReader(BytesIO(pdf_data))
            num_pages = len(pdf_reader.pages)
            result["total_pages"] = num_pages
            
            # Determine which pages to extract (convert to 0-indexed)
            pages_to_extract = []
            
            if isinstance(page_selection, int):
                # Single page (1-indexed input)
                pages_to_extract = [page_selection - 1]
            elif isinstance(page_selection, list):
                # List of specific pages (1-indexed input)
                pages_to_extract = [p - 1 for p in page_selection]
            elif isinstance(page_selection, dict):
                # Range with 'start' and 'end' (1-indexed input)
                start = page_selection.get('start', 1) - 1
                end = min(page_selection.get('end', num_pages), num_pages)
                pages_to_extract = list(range(start, end))
            
            # Validate pages
            pages_to_extract = [p for p in pages_to_extract if 0 <= p < num_pages]
            
            if not pages_to_extract:
                return {"status": "error", "message": "No valid pages to extract"}
            
            # Create new PDF with selected pages
            writer = PdfWriter()
            for page_num in pages_to_extract:
                writer.add_page(pdf_reader.pages[page_num])
                result["extracted_pages"].append(page_num + 1)  # Return 1-indexed
            
            # Generate output filename
            base_name = os.path.splitext(file_name)[0]
            if len(pages_to_extract) == 1:
                output_filename = f'{base_name}_page_{pages_to_extract[0] + 1}.pdf'
            else:
                output_filename = f'{base_name}_pages_{pages_to_extract[0] + 1}_to_{pages_to_extract[-1] + 1}.pdf'
            
            output_path = os.path.join('/tmp/', output_filename)
            
            with open(output_path, 'wb') as output_file:
                writer.write(output_file)
            
            # Upload to destination stage
            FileOperation(session).put(
                f"file://{output_path}",
                dest_stage_name,
                auto_compress=False
            )
            
            result["output_file"] = output_filename
            
            # Cleanup
            os.remove(file_path)
            os.remove(output_path)
                
        return result
        
    except Exception as e:
        return {"status": "error", "message": str(e)}
$;

ユーザーの選択に基づいてページを抽出します：

-- Option 2: First page only
CALL db.schema.extract_pdf_pages(
  '@db.schema.my_stage',
  'invoice.pdf',
  '@db.schema.extracted_pages_stage',
  1  -- Single page (1-indexed)
);

-- Option 3: Page range (e.g., pages 1-5)
CALL db.schema.extract_pdf_pages(
  '@db.schema.my_stage',
  'invoice.pdf',
  '@db.schema.extracted_pages_stage',
  {'start': 1, 'end': 5}  -- Range (1-indexed, inclusive)
);

-- Option 4: Specific pages (e.g., pages 1, 3, 10)
CALL db.schema.extract_pdf_pages(
  '@db.schema.my_stage',
  'invoice.pdf',
  '@db.schema.extracted_pages_stage',
  [1, 3, 10]  -- Array of pages (1-indexed)
);

次に、抽出したページに対して AI_EXTRACT を実行します：

-- Extract from the optimized PDF (fewer pages = faster + cheaper)
SELECT AI_EXTRACT(
  file => TO_FILE('@db.schema.extracted_pages_stage', '<output_file_from_procedure>'),
  responseFormat => {
    'invoice_number': 'What is the invoice number?',
    'vendor_name': 'What is the vendor name?',
    'total_amount': 'What is the total amount?'
  }
) AS result;

「いいえ」（ドキュメント全体を処理する）を選択した場合、またはファイルが制限内の場合：

-- Create table and insert single result
CREATE TABLE IF NOT EXISTS db.schema.extraction_results (
  file_path STRING,
  extracted_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
  invoice_number STRING,
  vendor_name STRING,
  total_amount STRING
);

INSERT INTO db.schema.extraction_results (file_path, invoice_number, vendor_name, total_amount)
SELECT 
  '<test_file_path>',
  result:response:invoice_number::STRING,
  result:response:vendor_name::STRING,
  result:response:total_amount::STRING
FROM (SELECT AI_EXTRACT(...) AS result);

「はい」または「わからない」を選択した場合 — ファイルが 125 ページを超える可能性があります：

大きなドキュメントをどのように処理するか、ユーザーに尋ねます：

Your files may exceed AI_EXTRACT's 125-page limit. How would you like to handle them?

Options:
1. Chunk into 125-page segments (process each chunk separately)
   - Best for: Invoices, receipts, forms where each page is independent
   - Results can be aggregated after processing

2. Keep entire document together (use AI_PARSE_DOCUMENT + AI_COMPLETE)
   - Best for: Documents where context spans multiple pages
   - Supports up to 500 pages
   - Uses structured outputs for extraction

Use cases for keeping document together:
• Contracts - terms may reference other sections, cross-references
• Legal documents - clauses depend on definitions elsewhere
• Research papers - methodology affects interpretation of results
• Technical manuals - procedures reference diagrams in other sections
• Policy documents - exceptions and conditions span pages
• Loan agreements - terms interconnected across sections
• Insurance policies - coverage details reference exclusions
• Merger/acquisition documents - financial data relates to legal terms
• Patent applications - claims reference detailed descriptions
• Medical records - diagnosis relates to test results and history

「125 ページ単位に分割」の場合： 以下のパイプラインオプション B（分割パイプライン）に進みます。

「ドキュメント全体をまとめて処理」の場合： 以下のパイプラインオプション C（AI_PARSE_DOCUMENT + AI_COMPLETE フロー）に進みます。

「保存してパイプラインを作成」の場合：

ステージに追加された新しいファイルを自動的に処理する、完全なパイプラインを作成します。

重要 — 最初に大きなファイルについて尋ねます：

Can any of your files be longer than the page limit?
- AI_EXTRACT: 125 pages max
- AI_PARSE_DOCUMENT: 500 pages max
Options:
1. No - all files are within the limit
2. Yes - some files may exceed the limit
3. Not sure

次に、ページの最適化について尋ねます（コスト削減と速度向上のため）：

Would you like to process entire documents or only specific pages in your pipeline?
Processing fewer pages = faster execution + lower cost.

Options:
1. Process entire documents (all pages)
2. Process only the first page of each document
3. Process a specific page range (e.g., pages 1-5)
4. Process specific pages (e.g., pages 1, 3, 10)

---

パイプラインオプション A：標準パイプライン（ページ数の上限内のファイル）

ユーザーが「いいえ」（ファイルが上限内）と「ドキュメント全体を処理」の両方を選択した場合：

-- Step 1: Create results table
CREATE TABLE IF NOT EXISTS db.schema.extraction_results (
  file_path STRING,
  file_name STRING,
  extracted_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
  invoice_number STRING,
  vendor_name STRING,
  total_amount STRING,
  raw_response VARIANT
);

-- Step 2: Insert current result
INSERT INTO db.schema.extraction_results (file_path, file_name, invoice_number, vendor_name, total_amount, raw_response)
SELECT 
  '<test_file_path>',
  SPLIT_PART('<test_file_path>', '/', -1),
  result:response:invoice_number::STRING,
  result:response:vendor_name::STRING,
  result:response:total_amount::STRING,
  result
FROM (SELECT AI_EXTRACT(...) AS result);

-- Step 3: Create stream on stage to detect new files
CREATE OR REPLACE STREAM db.schema.doc_extraction_stream 
  ON STAGE @db.schema.my_stage;

-- Step 4: Create task to process new files automatically
CREATE OR REPLACE TASK db.schema.doc_extraction_task
  WAREHOUSE = my_warehouse
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('db.schema.doc_extraction_stream')
AS
  INSERT INTO db.schema.extraction_results (file_path, file_name, invoice_number, vendor_name, total_amount, raw_response)
  SELECT 
    relative_path,
    SPLIT_PART(relative_path, '/', -1),
    result:response:invoice_number::STRING,
    result:response:vendor_name::STRING,
    result:response:total_amount::STRING,
    result
  FROM (
    SELECT 
      relative_path,
      AI_EXTRACT(
        file => TO_FILE('@db.schema.my_stage', relative_path),
        responseFormat => {
          'invoice_number': 'What is the invoice number?',
          'vendor_name': 'What is the vendor name?',
          'total_amount': 'What is the total amount?'
        }
      ) AS result
    FROM db.schema.doc_extraction_stream
    WHERE METADATA$ACTION = 'INSERT'
      AND relative_path LIKE '%.pdf'
  );

-- Step 5: Resume the task
ALTER TASK db.schema.doc_extraction_task RESUME;

-- Verify task is running
SHOW TASKS LIKE 'doc_extraction_task' IN SCHEMA db.schema;

---

パイプラインオプション A.1：ページ最適化パイプライン（特定のページのみを処理）

ユーザーがページ最適化（オプション 2、3、または 4）を選択した場合、AI_EXTRACT の実行前に特定のページを抽出するパイプラインを作成します：

ステップ 1：ページ抽出のストアドプロシージャを作成します（まだ作成していない場合 — 上記の「この結果を保存」セクションを参照）

ステップ 2：ページ追跡用の結果テーブルを作成します

CREATE TABLE IF NOT EXISTS db.schema.extraction_results (
  result_id INT AUTOINCREMENT,
  file_path STRING,
  file_name STRING,
  pages_processed STRING,        -- Which pages were extracted
  extracted_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
  invoice_number STRING,
  vendor_name STRING,
  total_amount STRING,
  raw_response VARIANT
);

ステップ 3：ページ最適化抽出タスクを作成します

最初のページのみ（請求書で最も一般的）：

-- Stream on source stage
CREATE OR REPLACE STREAM db.schema.doc_source_stream 
  ON STAGE @db.schema.my_stage;

-- Task: Extract first page and run AI_EXTRACT
CREATE OR REPLACE TASK db.schema.page_optimized_extraction_task
  WAREHOUSE = my_warehouse
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('db.schema.doc_source_stream')
AS
DECLARE
  file_record RECORD;
  extract_result VARIANT;
  output_file STRING;
BEGIN
  FOR file_record IN (
    SELECT relative_path 
    FROM db.schema.doc_source_stream 
    WHERE METADATA$ACTION = 'INSERT' 
      AND relative_path LIKE '%.pdf'
  ) DO
    -- Extract first page only
    extract_result := (
      CALL db.schema.extract_pdf_pages(
        '@db.schema.my_stage',
        :file_record.relative_path,
        '@db.schema.extracted_pages_stage',
        1  -- First page only
      )
    );
    
    output_file := extract_result:output_file::STRING;
    
    -- Run AI_EXTRACT on the single page
    INSERT INTO db.schema.extraction_results (file_path, file_name, pages_processed, invoice_number, vendor_name, total_amount, raw_response)
    SELECT 
      :file_record.relative_path,
      SPLIT_PART(:file_record.relative_path, '/', -1),
      'Page 1',
      result:response:invoice_number::STRING,
      result:response:vendor_name::STRING,
      result:response:total_amount::STRING,
      result
    FROM (
      SELECT AI_EXTRACT(
        file => TO_FILE('@db.schema.extracted_pages_stage', :output_file),
        responseFormat => {
          'invoice_number': 'What is the invoice number?',
          'vendor_name': 'What is the vendor name?',
          'total_amount': 'What is the total amount?'
        }
      ) AS result
    );
  END FOR;
END;

ALTER TASK db.schema.page_optimized_extraction_task RESUME;

特定のページ範囲（例：1〜5 ページ）：

CREATE OR REPLACE TASK db.schema.page_range_extraction_task
  WAREHOUSE = my_warehouse
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('db.schema.doc_source_stream')
AS
DECLARE
  file_record RECORD;
  extract_result VARIANT;
  output_file STRING;
BEGIN
  FOR file_record IN (
    SELECT relative_path 
    FROM db.schema.doc_source_stream 
    WHERE METADATA$ACTION = 'INSERT' 
      AND relative_path LIKE '%.pdf'
  ) DO
    -- Extract pages 1-5
    extract_result := (
      CALL db.schema.extract_pdf_pages(
        '@db.schema.my_stage',
        :file_record.relative_path,
        '@db.schema.extracted_pages_stage',
        {'start': 1, 'end': 5}  -- Page range
      )
    );
    
    output_file := extract_result:output_file::STRING;
    
    INSERT INTO db.schema.extraction_results (file_path, file_name, pages_processed, invoice_number, vendor_name, total_amount, raw_response)
    SELECT 
      :file_record.relative_path,
      SPLIT_PART(:file_record.relative_path, '/', -1),
      'Pages 1-5',
      result:response:invoice_number::STRING,
      result:response:vendor_name::STRING,
      result:response:total_amount::STRING,
      result
    FROM (
      SELECT AI_EXTRACT(
        file => TO_FILE('@db.schema.extracted_pages_stage', :output_file),
        responseFormat => {
          'invoice_number': 'What is the invoice number?',
          'vendor_name': 'What is the vendor name?',
          'total_amount': 'What is the total amount?'
        }
      ) AS result
    );
  END FOR;
END;

ALTER TASK db.schema.page_range_extraction_task RESUME;

特定のページ（例：1、3、10 ページ）：

CREATE OR REPLACE TASK db.schema.specific_pages_extraction_task
  WAREHOUSE = my_warehouse
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('db.schema.doc_source_stream')
AS
DECLARE
  file_record RECORD;
  extract_result VARIANT;
  output_file STRING;
BEGIN
  FOR file_record IN (
    SELECT relative_path 
    FROM db.schema.doc_source_stream 
    WHERE METADATA$ACTION = 'INSERT' 
      AND relative_path LIKE '%.pdf'
  ) DO
    -- Extract specific pages
    extract_result := (
      CALL db.schema.extract_pdf_pages(
        '@db.schema.my_stage',
        :file_record.relative_path,
        '@db.schema.extracted_pages_stage',
        [1, 3, 10]  -- Specific pages array
      )
    );
    
    output_file := extract_result:output_file::STRING;
    
    INSERT INTO db.schema.extraction_results (file_path, file_name, pages_processed, invoice_number, vendor_name, total_amount, raw_response)
    SELECT 
      :file_record.relative_path,
      SPLIT_PART(:file_record.relative_path, '/', -1),
      'Pages 1, 3, 10',
      result:response:invoice_number::STRING,
      result:response:vendor_name::STRING,
      result:response:total_amount::STRING,
      result
    FROM (
      SELECT AI_EXTRACT(
        file => TO_FILE('@db.schema.extracted_pages_stage', :output_file),
        responseFormat => {
          'invoice_number': 'What is the invoice number?',
          'vendor_name': 'What is the vendor name?',
          'total_amount': 'What is the total amount?'
        }
      ) AS result
    );
  END FOR;
END;

ALTER TASK db.schema.specific_pages_extraction_task RESUME;

コストと速度の比較：

アプローチ	処理するページ数	相対コスト	速度
ドキュメント全体（10 ページ）	10	100%	基準
最初のページのみ	1	約 10%	約 10 倍高速
1〜5 ページ	5	約 50%	約 2 倍高速
特定のページ（1、3、10）	3	約 30%	約 3 倍高速

---

パイプラインオプション B：大容量ファイル用パイプライン（125 ページ超のファイル）

ユーザーが「はい」または「わからない」を選択した場合、分割パイプラインを作成します：

ステップ 1：分割用ストアドプロシージャを作成します

このプロシージャは PDF を 125 ページ単位に分割し、分割ファイルをチャンク用ステージにアップロードします：

-- Create stage for chunked files (with server-side encryption)
CREATE STAGE IF NOT EXISTS db.schema.chunks_stage
  ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE');

-- Create chunking stored procedure
CREATE OR REPLACE PROCEDURE db.schema.split_document_into_chunks(
    stage_name STRING, 
    file_name STRING, 
    dest_stage_name STRING,
    chunk_size INT DEFAULT 125
)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.9'
PACKAGES = ('snowflake-snowpark-python', 'pypdf2')
HANDLER = 'run'
AS
$
from PyPDF2 import PdfReader, PdfWriter
from snowflake.snowpark import Session
from snowflake.snowpark.file_operation import FileOperation
from io import BytesIO
import os
import math

def run(session: Session, stage_name: str, file_name: str, dest_stage_name: str, chunk_size: int = 125) -> dict:
    result = {
        "status": "success",
        "source_file": file_name,
        "total_pages": 0,
        "chunks_created": 0,
        "chunk_files": []
    }
    
    try:
        # Construct the file URL and retrieve file
        file_url = f"{stage_name}/{file_name}"
        get_result = session.file.get(file_url, '/tmp/')
        file_path = os.path.join('/tmp/', file_name)
        
        if not os.path.exists(file_path):
            return {"status": "error", "message": f"File {file_name} not found in {stage_name}"}

        with open(file_path, 'rb') as f:
            pdf_data = f.read()
            pdf_reader = PdfReader(BytesIO(pdf_data))
            num_pages = len(pdf_reader.pages)
            result["total_pages"] = num_pages
            
            # Calculate number of chunks needed
            num_chunks = math.ceil(num_pages / chunk_size)
            result["chunks_created"] = num_chunks
            
            # Get base filename without extension
            base_name = os.path.splitext(file_name)[0]
            
            for chunk_num in range(num_chunks):
                start_page = chunk_num * chunk_size
                end_page = min((chunk_num + 1) * chunk_size, num_pages)
                
                writer = PdfWriter()
                for page_num in range(start_page, end_page):
                    writer.add_page(pdf_reader.pages[page_num])
                
                # Create chunk filename: original_chunk_1_pages_1_125.pdf
                chunk_filename = f'{base_name}_chunk_{chunk_num + 1}_pages_{start_page + 1}_{end_page}.pdf'
                chunk_file_path = os.path.join('/tmp/', chunk_filename)
                
                with open(chunk_file_path, 'wb') as output_file:
                    writer.write(output_file)
                
                if not os.path.exists(chunk_file_path):
                    return {"status": "error", "message": f"Failed to create chunk file {chunk_filename}"}

                # Upload chunk to destination stage
                FileOperation(session).put(
                    f"file://{chunk_file_path}",
                    dest_stage_name,
                    auto_compress=False
                )
                
                result["chunk_files"].append({
                    "chunk_number": chunk_num + 1,
                    "filename": chunk_filename,
                    "start_page": start_page + 1,
                    "end_page": end_page,
                    "page_count": end_page - start_page
                })
                
                # Clean up temp chunk file
                os.remove(chunk_file_path)
            
            # Clean up original temp file
            os.remove(file_path)
                
        return result
        
    except Exception as e:
        return {"status": "error", "message": str(e)}
$;

ステップ 2：チャンクと結果の追跡用テーブルを作成します

-- Table to track chunking operations
CREATE TABLE IF NOT EXISTS db.schema.document_chunks (
  chunk_id INT AUTOINCREMENT,
  source_file STRING,
  chunk_file STRING,
  chunk_number INT,
  start_page INT,
  end_page INT,
  page_count INT,
  chunked_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
  processed BOOLEAN DEFAULT FALSE
);

-- Table for extraction results with chunk tracking
CREATE TABLE IF NOT EXISTS db.schema.extraction_results (
  result_id INT AUTOINCREMENT,
  source_file STRING,           -- Original file name
  chunk_file STRING,            -- Chunk file name (NULL if not chunked)
  chunk_number INT,             -- Chunk number (NULL if not chunked)
  file_name STRING,
  extracted_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
  invoice_number STRING,
  vendor_name STRING,
  total_amount STRING,
  raw_response VARIANT
);

-- Aggregated results view (joins chunks back together)
CREATE OR REPLACE VIEW db.schema.extraction_results_aggregated AS
SELECT 
  source_file,
  ARRAY_AGG(OBJECT_CONSTRUCT(
    'chunk_number', chunk_number,
    'invoice_number', invoice_number,
    'vendor_name', vendor_name,
    'total_amount', total_amount
  )) WITHIN GROUP (ORDER BY chunk_number) AS chunk_results,
  -- For single-value fields, take from first chunk
  MAX(CASE WHEN chunk_number = 1 OR chunk_number IS NULL THEN invoice_number END) AS invoice_number,
  MAX(CASE WHEN chunk_number = 1 OR chunk_number IS NULL THEN vendor_name END) AS vendor_name,
  -- For totals, might need custom logic (sum, last chunk, etc.)
  MAX(total_amount) AS total_amount,
  COUNT(*) AS total_chunks,
  MIN(extracted_at) AS first_extracted_at,
  MAX(extracted_at) AS last_extracted_at
FROM db.schema.extraction_results
GROUP BY source_file;

ステップ 3：分割タスクを作成します（最初に実行）

-- Stream on source stage
CREATE OR REPLACE STREAM db.schema.doc_source_stream 
  ON STAGE @db.schema.my_stage;

-- Task 1: Chunk large documents
CREATE OR REPLACE TASK db.schema.chunk_documents_task
  WAREHOUSE = my_warehouse
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('db.schema.doc_source_stream')
AS
DECLARE
  file_record RECORD;
  chunk_result VARIANT;
BEGIN
  -- Process each new file
  FOR file_record IN (
    SELECT relative_path 
    FROM db.schema.doc_source_stream 
    WHERE METADATA$ACTION = 'INSERT' 
      AND relative_path LIKE '%.pdf'
  ) DO
    -- Get page count first
    LET page_count INT := (
      SELECT AI_PARSE_DOCUMENT(
        TO_FILE('@db.schema.my_stage', :file_record.relative_path),
        {'mode': 'OCR', 'page_filter': [{'start': 0, 'end': 1}]}
      ):pageCount::INT
    );
    
    IF (page_count > 125) THEN
      -- Large file: chunk it
      chunk_result := (
        CALL db.schema.split_document_into_chunks(
          '@db.schema.my_stage',
          :file_record.relative_path,
          '@db.schema.chunks_stage',
          125
        )
      );
      
      -- Record chunks in tracking table
      INSERT INTO db.schema.document_chunks (source_file, chunk_file, chunk_number, start_page, end_page, page_count)
      SELECT 
        :file_record.relative_path,
        f.value:filename::STRING,
        f.value:chunk_number::INT,
        f.value:start_page::INT,
        f.value:end_page::INT,
        f.value:page_count::INT
      FROM TABLE(FLATTEN(input => :chunk_result:chunk_files)) f;
    ELSE
      -- Small file: record as single chunk for unified processing
      INSERT INTO db.schema.document_chunks (source_file, chunk_file, chunk_number, start_page, end_page, page_count)
      VALUES (:file_record.relative_path, :file_record.relative_path, 1, 1, :page_count, :page_count);
    END IF;
  END FOR;
END;

ALTER TASK db.schema.chunk_documents_task RESUME;

ステップ 4：抽出タスクを作成します（チャンクを処理）

-- Stream on chunks table to detect new chunks to process
CREATE OR REPLACE STREAM db.schema.chunks_to_process_stream 
  ON TABLE db.schema.document_chunks
  SHOW_INITIAL_ROWS = TRUE;

-- Task 2: Extract from chunks
CREATE OR REPLACE TASK db.schema.extract_from_chunks_task
  WAREHOUSE = my_warehouse
  SCHEDULE = '2 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('db.schema.chunks_to_process_stream')
AS
  INSERT INTO db.schema.extraction_results (source_file, chunk_file, chunk_number, file_name, invoice_number, vendor_name, total_amount, raw_response)
  SELECT 
    dc.source_file,
    dc.chunk_file,
    dc.chunk_number,
    SPLIT_PART(dc.chunk_file, '/', -1),
    result:response:invoice_number::STRING,
    result:response:vendor_name::STRING,
    result:response:total_amount::STRING,
    result
  FROM db.schema.chunks_to_process_stream dc,
  LATERAL (
    SELECT AI_EXTRACT(
      file => TO_FILE(
        CASE 
          WHEN dc.chunk_file = dc.source_file THEN '@db.schema.my_stage'  -- Original file (small)
          ELSE '@db.schema.chunks_stage'  -- Chunked file
        END, 
        dc.chunk_file
      ),
      responseFormat => {
        'invoice_number': 'What is the invoice number?',
        'vendor_name': 'What is the vendor name?',
        'total_amount': 'What is the total amount?'
      }
    ) AS result
  )
  WHERE dc.METADATA$ACTION = 'INSERT';

ALTER TASK db.schema.extract_from_chunks_task RESUME;

ステップ 5：集計結果をクエリします

-- View results aggregated by source document
SELECT * FROM db.schema.extraction_results_aggregated;

-- View individual chunk results
SELECT * FROM db.schema.extraction_results ORDER BY source_file, chunk_number;

-- Check processing status
SELECT 
  dc.source_file,
  dc.chunk_number,
  dc.page_count,
  CASE WHEN er.result_id IS NOT NULL THEN 'Processed' ELSE 'Pending' END AS status
FROM db.schema.document_chunks dc
LEFT JOIN db.schema.extraction_results er 
  ON dc.source_file = er.source_file 
  AND dc.chunk_number = er.chunk_number
ORDER BY dc.source_file, dc.chunk_number;

---

パイプラインオプション C：全文書パイプライン（AI_PARSE_DOCUMENT + AI_COMPLETE）

文書全体を処理する必要があり、文書全体のコンテキストが重要な場合に使用します。最大 500 ページの文書に対応しています。

このアプローチを使用する場合：

• 契約書 - 条項が他のセクションを参照している場合（例：「第 2.1 条で定義されているとおり」）
• 法的文書 - 条項が文書内の別の箇所にある定義に依存する場合
• 研究論文 - 方法論のセクションが結果の解釈に影響する場合
• 技術マニュアル - 手順が他のセクションの図や仕様を参照する場合
• ポリシー文書 - 例外や条件が複数のページにまたがる場合
• ローン／住宅ローン契約書 - 金利条件が文書全体にわたる条件に関連付けられている場合
• 保険証券 - 補償内容が免責事項や制限事項を参照する場合
• M&A 文書 - 財務データが法的条件と相互に関連している場合
• 特許出願書類 - 請求項が詳細な説明を参照する場合
• 医療記録 - 診断が検査結果や患者の病歴に関連する場合
• 年次報告書 - 財務諸表が経営陣による説明に関連する場合
• RFP 回答書 - 技術的なアプローチが価格やスケジュールを参照する場合

ステップ 1：結果テーブルを作成する

CREATE TABLE IF NOT EXISTS db.schema.extraction_results_full_doc (
  result_id INT AUTOINCREMENT,
  file_path STRING,
  file_name STRING,
  total_pages INT,
  extraction_method STRING DEFAULT 'AI_PARSE_DOCUMENT + AI_COMPLETE',
  extracted_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
  -- Add your extraction fields here
  contract_parties VARIANT,       -- For contracts
  effective_date STRING,
  termination_date STRING,
  total_value STRING,
  key_terms VARIANT,
  obligations VARIANT,
  raw_response VARIANT
);

ステップ 2：文書を解析し、構造化出力で情報を抽出する

-- Single document extraction (for testing)
WITH parsed_doc AS (
  SELECT 
    AI_PARSE_DOCUMENT(
      TO_FILE('@db.schema.my_stage', 'contract.pdf'),
      {'mode': 'LAYOUT'}
    ) AS parsed_result
),
document_text AS (
  SELECT 
    parsed_result:content::STRING AS full_text,
    parsed_result:pageCount::INT AS total_pages
  FROM parsed_doc
)
SELECT 
  total_pages,
  AI_COMPLETE(
    'claude-3-5-sonnet',
    'You are an expert contract analyst. Extract the following information from this contract document.
    
IMPORTANT: The entire document content is provided below. Analyze it thoroughly as a complete document - 
terms and definitions in one section may affect interpretation of other sections.

DOCUMENT CONTENT:
' || full_text || '

Extract all requested fields. If information is not found, use null.',
    {
      'response_format': {
        'type': 'json',
        'schema': {
          'type': 'object',
          'properties': {
            'contract_parties': {
              'type': 'object',
              'properties': {
                'party_a': {'type': 'string', 'description': 'First party to the contract'},
                'party_b': {'type': 'string', 'description': 'Second party to the contract'},
                'additional_parties': {
                  'type': 'array',
                  'items': {'type': 'string'},
                  'description': 'Any additional parties'
                }
              },
              'required': ['party_a', 'party_b']
            },
            'effective_date': {'type': 'string', 'description': 'Contract effective/start date'},
            'termination_date': {'type': 'string', 'description': 'Contract end/termination date'},
            'total_value': {'type': 'string', 'description': 'Total contract value or consideration'},
            'key_terms': {
              'type': 'array',
              'items': {
                'type': 'object',
                'properties': {
                  'term_name': {'type': 'string'},
                  'description': {'type': 'string'},
                  'section_reference': {'type': 'string'}
                }
              },
              'description': 'Key terms and definitions'
            },
            'obligations': {
              'type': 'object',
              'properties': {
                'party_a_obligations': {
                  'type': 'array',
                  'items': {'type': 'string'}
                },
                'party_b_obligations': {
                  'type': 'array',
                  'items': {'type': 'string'}
                }
              }
            },
            'payment_terms': {
              'type': 'object',
              'properties': {
                'payment_schedule': {'type': 'string'},
                'payment_method': {'type': 'string'},
                'late_payment_penalty': {'type': 'string'}
              }
            },
            'termination_clauses': {
              'type': 'array',
              'items': {'type': 'string'},
              'description': 'Conditions under which contract can be terminated'
            },
            'governing_law': {'type': 'string', 'description': 'Jurisdiction/governing law'},
            'confidentiality': {'type': 'boolean', 'description': 'Whether confidentiality clause exists'},
            'non_compete': {'type': 'boolean', 'description': 'Whether non-compete clause exists'}
          },
          'required': ['contract_parties', 'effective_date', 'total_value']
        }
      },
      'max_tokens': 8192
    }
  ) AS extraction_result
FROM document_text;

ステップ 3：全文書抽出パイプラインを作成する

-- Stream on source stage
CREATE OR REPLACE STREAM db.schema.full_doc_stream 
  ON STAGE @db.schema.my_stage;

-- Task for full-document extraction
CREATE OR REPLACE TASK db.schema.full_document_extraction_task
  WAREHOUSE = my_warehouse
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('db.schema.full_doc_stream')
AS
  INSERT INTO db.schema.extraction_results_full_doc (
    file_path, file_name, total_pages, 
    contract_parties, effective_date, termination_date, 
    total_value, key_terms, obligations, raw_response
  )
  SELECT 
    relative_path,
    SPLIT_PART(relative_path, '/', -1),
    parsed_result:pageCount::INT,
    PARSE_JSON(extraction_result):contract_parties,
    PARSE_JSON(extraction_result):effective_date::STRING,
    PARSE_JSON(extraction_result):termination_date::STRING,
    PARSE_JSON(extraction_result):total_value::STRING,
    PARSE_JSON(extraction_result):key_terms,
    PARSE_JSON(extraction_result):obligations,
    PARSE_JSON(extraction_result)
  FROM (
    SELECT 
      relative_path,
      AI_PARSE_DOCUMENT(
        TO_FILE('@db.schema.my_stage', relative_path),
        {'mode': 'LAYOUT'}
      ) AS parsed_result,
      AI_COMPLETE(
        'claude-3-5-sonnet',
        'You are an expert contract analyst. Extract contract information from this document.

DOCUMENT:
' || AI_PARSE_DOCUMENT(
          TO_FILE('@db.schema.my_stage', relative_path),
          {'mode': 'LAYOUT'}
        ):content::STRING || '

Extract: contract_parties, effective_date, termination_date, total_value, key_terms, obligations, payment_terms, termination_clauses, governing_law, confidentiality, non_compete',
        {
          'response_format': {
            'type': 'json',
            'schema': {
              'type': 'object',
              'properties': {
                'contract_parties': {
                  'type': 'object',
                  'properties': {
                    'party_a': {'type': 'string'},
                    'party_b': {'type': 'string'}
                  },
                  'required': ['party_a', 'party_b']
                },
                'effective_date': {'type': 'string'},
                'termination_date': {'type': 'string'},
                'total_value': {'type': 'string'},
                'key_terms': {'type': 'array', 'items': {'type': 'object', 'properties': {'term_name': {'type': 'string'}, 'description': {'type': 'string'}}}},
                'obligations': {'type': 'object', 'properties': {'party_a_obligations': {'type': 'array', 'items': {'type': 'string'}}, 'party_b_obligations': {'type': 'array', 'items': {'type': 'string'}}}},
                'payment_terms': {'type': 'object', 'properties': {'payment_schedule': {'type': 'string'}, 'payment_method': {'type': 'string'}}},
                'termination_clauses': {'type': 'array', 'items': {'type': 'string'}},
                'governing_law': {'type': 'string'},
                'confidentiality': {'type': 'boolean'},
                'non_compete': {'type': 'boolean'}
              },
              'required': ['contract_parties', 'effective_date', 'total_value']
            }
          },
          'max_tokens': 8192
        }
      ) AS extraction_result
    FROM db.schema.full_doc_stream
    WHERE METADATA$ACTION = 'INSERT'
      AND relative_path LIKE '%.pdf'
  );

ALTER TASK db.schema.full_document_extraction_task RESUME;

全文書抽出用の追加スキーマテンプレート：

研究論文スキーマ：

{
  'response_format': {
    'type': 'json',
    'schema': {
      'type': 'object',
      'properties': {
        'title': {'type': 'string'},
        'authors': {'type': 'array', 'items': {'type': 'string'}},
        'abstract': {'type': 'string'},
        'keywords': {'type': 'array', 'items': {'type': 'string'}},
        'methodology': {'type': 'string', 'description': 'Research methodology summary'},
        'key_findings': {'type': 'array', 'items': {'type': 'string'}},
        'conclusions': {'type': 'string'},
        'limitations': {'type': 'array', 'items': {'type': 'string'}},
        'future_work': {'type': 'string'},
        'references_count': {'type': 'integer'}
      },
      'required': ['title', 'authors', 'abstract', 'key_findings']
    }
  }
}

保険証券スキーマ：

{
  'response_format': {
    'type': 'json',
    'schema': {
      'type': 'object',
      'properties': {
        'policy_number': {'type': 'string'},
        'policy_holder': {'type': 'string'},
        'insurer': {'type': 'string'},
        'policy_type': {'type': 'string', 'enum': ['life', 'health', 'auto', 'property', 'liability', 'other']},
        'effective_date': {'type': 'string'},
        'expiration_date': {'type': 'string'},
        'premium': {'type': 'object', 'properties': {'amount': {'type': 'number'}, 'frequency': {'type': 'string'}}},
        'coverage': {
          'type': 'array',
          'items': {
            'type': 'object',
            'properties': {
              'coverage_type': {'type': 'string'},
              'limit': {'type': 'string'},
              'deductible': {'type': 'string'}
            }
          }
        },
        'exclusions': {'type': 'array', 'items': {'type': 'string'}},
        'beneficiaries': {'type': 'array', 'items': {'type': 'string'}},
        'special_conditions': {'type': 'array', 'items': {'type': 'string'}}
      },
      'required': ['policy_number', 'policy_holder', 'insurer', 'coverage']
    }
  }
}

ローン契約書スキーマ：

{
  'response_format': {
    'type': 'json',
    'schema': {
      'type': 'object',
      'properties': {
        'loan_number': {'type': 'string'},
        'borrower': {'type': 'string'},
        'lender': {'type': 'string'},
        'principal_amount': {'type': 'number'},
        'currency': {'type': 'string'},
        'interest_rate': {'type': 'object', 'properties': {'rate': {'type': 'number'}, 'type': {'type': 'string', 'enum': ['fixed', 'variable']}}},
        'loan_term': {'type': 'object', 'properties': {'duration': {'type': 'integer'}, 'unit': {'type': 'string'}}},
        'start_date': {'type': 'string'},
        'maturity_date': {'type': 'string'},
        'payment_schedule': {'type': 'string'},
        'collateral': {'type': 'array', 'items': {'type': 'string'}},
        'covenants': {'type': 'array', 'items': {'type': 'string'}},
        'default_conditions': {'type': 'array', 'items': {'type': 'string'}},
        'prepayment_terms': {'type': 'string'}
      },
      'required': ['loan_number', 'borrower', 'lender', 'principal_amount', 'interest_rate']
    }
  }
}

比較：チャンク分割と全文書アプローチ：

項目	チャンク分割（オプション B）	全文書（オプション C）
最大ページ数	無制限（125 ページのチャンク）	500 ページ
コンテキストの保持	限定的（チャンク単位）	文書全体のコンテキスト
相互参照	見落とす可能性あり	保持される
最適な対象	個別に処理できるページ（請求書）	相互に関連するコンテンツ（契約書）
結果の集約	必要	不要
処理の複雑さ	高い（複数の段階）	よりシンプル（1 回の処理）
AI_EXTRACT の使用	はい（チャンクごと）	いいえ（代わりに AI_COMPLETE）

---

ユーザーにパイプラインのパラメーターを確認します：

Configure your pipeline:
1. Warehouse name for processing: [default: current warehouse]
2. Schedule frequency: 
   - Every 1 minute
   - Every 5 minutes (recommended)
   - Every 15 minutes
   - Every hour
3. File pattern to match: [default: %.pdf]

2.5h：全文一括抽出に進む

ユーザーが「すべてのファイルから抽出」を明示的に選択した場合のみ実行します。

1. 最終的な responseFormat をユーザーに確認します
2. 以下の情報を指定して ai-extract/SKILL.md を読み込みます：
   • 検証済みの responseFormat
   • ステージのパス
   • ファイルパターン（一括処理の場合）
3. ai-extract でのスキーマ定義はスキップします（すでに完了しています）

引き継ぎコンテキストの例：

Proceeding to full extraction with validated schema:

Stage: @db.schema.stage
File pattern: %.pdf
Validated responseFormat:
{
  'invoice_number': 'What is the invoice or bill number shown at the top of the document?',
  'vendor_name': 'What is the name of the company or vendor issuing this invoice?',
  'total_amount': 'What is the final total amount due, including taxes?'
}

Test results confirmed accurate on: sample_invoice.pdf

ステップ 2.6：ビジュアル分析（チャート、設計図、図の場合）

ユーザーが「分析対象の画像とビジュアルコンテンツ」を選択した場合のみ適用します。

ビジュアル分析では、AI_COMPLETE を直接使用して画像を分析します。解析ステップは不要です。AI_COMPLETE は、画像ファイルを直接解釈できるビジョンモデルです。

重要： AI_COMPLETE には画像ファイル（PNG、JPEG など）が必要です。ユーザーが PDF を持っている場合は、まず画像に変換する必要があります。

2.6a：ファイル形式を確認する

ユーザーに確認します：

What format is your file?
Options:
1. Image file (PNG, JPEG, TIFF, BMP, GIF, WEBP)
2. PDF file (needs conversion to image)

---

画像ファイルの場合：直接分析に進む

-- Direct visual analysis with AI_COMPLETE
SELECT AI_COMPLETE(
  'claude-3-5-sonnet',
  [
    {
      'role': 'user',
      'content': [
        {'type': 'image', 'image_url': {'url': TO_FILE('@db.schema.stage', 'chart.png')}},
        {'type': 'text', 'text': 'Analyze this chart/diagram. Describe what you see and extract key information.'}
      ]
    }
  ],
  {'max_tokens': 4096}
) AS analysis;

---

PDF ファイルの場合：まず画像に変換する

ステップ 1：PDF から画像への変換用ストアドプロシージャを作成する

このプロシージャは pdf2image を使用して PDF のページを画像に変換し、images ステージにアップロードします：

-- Create stage for converted images (with server-side encryption)
CREATE STAGE IF NOT EXISTS db.schema.images_stage
  DIRECTORY = (ENABLE = TRUE)
  ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE')
  COMMENT = 'Stage for PDF-to-image converted files';

-- Create PDF to image conversion stored procedure
CREATE OR REPLACE PROCEDURE db.schema.convert_pdf_to_images(
    stage_name STRING, 
    file_name STRING, 
    dest_stage_name STRING,
    dpi INT DEFAULT 200,
    image_format STRING DEFAULT 'PNG',
    specific_pages ARRAY DEFAULT NULL
)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.9'
PACKAGES = ('snowflake-snowpark-python', 'pdf2image', 'pillow')
HANDLER = 'run'
AS
$
from pdf2image import convert_from_bytes
from snowflake.snowpark import Session
from snowflake.snowpark.file_operation import FileOperation
from io import BytesIO
import os

def run(session: Session, stage_name: str, file_name: str, dest_stage_name: str, 
        dpi: int = 200, image_format: str = 'PNG', specific_pages: list = None) -> dict:
    result = {
        "status": "success",
        "source_file": file_name,
        "total_pages": 0,
        "images_created": 0,
        "image_files": []
    }
    
    try:
        # Download PDF from stage
        file_url = f"{stage_name}/{file_name}"
        get_result = session.file.get(file_url, '/tmp/')
        file_path = os.path.join('/tmp/', file_name)
        
        if not os.path.exists(file_path):
            return {"status": "error", "message": f"File {file_name} not found in {stage_name}"}

        with open(file_path, 'rb') as f:
            pdf_data = f.read()
        
        # Convert PDF to images
        # If specific_pages provided, only convert those pages (1-indexed)
        if specific_pages:
            images = convert_from_bytes(
                pdf_data, 
                dpi=dpi, 
                fmt=image_format.lower(),
                first_page=min(specific_pages),
                last_page=max(specific_pages)
            )
            # Filter to only requested pages
            page_indices = [p - min(specific_pages) for p in specific_pages]
            images = [images[i] for i in page_indices if i < len(images)]
        else:
            images = convert_from_bytes(pdf_data, dpi=dpi, fmt=image_format.lower())
        
        result["total_pages"] = len(images)
        
        # Get base filename without extension
        base_name = os.path.splitext(file_name)[0]
        ext = image_format.lower()
        if ext == 'jpeg':
            ext = 'jpg'
        
        for i, image in enumerate(images):
            page_num = specific_pages[i] if specific_pages else i + 1
            
            # Create image filename: original_page_1.png
            image_filename = f'{base_name}_page_{page_num}.{ext}'
            image_file_path = os.path.join('/tmp/', image_filename)
            
            # Save image
            image.save(image_file_path, format=image_format.upper())
            
            if not os.path.exists(image_file_path):
                return {"status": "error", "message": f"Failed to create image {image_filename}"}

            # Upload to destination stage
            FileOperation(session).put(
                f"file://{image_file_path}",
                dest_stage_name,
                auto_compress=False
            )
            
            result["image_files"].append({
                "page_number": page_num,
                "filename": image_filename,
                "format": image_format.upper()
            })
            result["images_created"] += 1
            
            # Clean up temp image file
            os.remove(image_file_path)
        
        # Clean up original temp file
        os.remove(file_path)
            
        return result
        
    except Exception as e:
        return {"status": "error", "message": str(e)}
$;

ステップ 2：変換するページを確認する

Which pages of the PDF contain the charts/diagrams you want to analyze?
Options:
1. Convert all pages to images
2. Convert specific pages (e.g., 1, 3, 5)
3. Convert a page range (e.g., pages 1-10)
4. Convert just the first page to test

ステップ 3：PDF を画像に変換する

-- Convert specific pages (e.g., pages 1, 3, 5)
CALL db.schema.convert_pdf_to_images(
  '@db.schema.doc_stage',
  'blueprint.pdf',
  '@db.schema.images_stage',
  200,           -- DPI (higher = better quality, larger files)
  'PNG',         -- Format: PNG, JPEG
  [1, 3, 5]      -- Specific pages (1-indexed), or NULL for all pages
);

-- Convert all pages
CALL db.schema.convert_pdf_to_images(
  '@db.schema.doc_stage',
  'blueprint.pdf',
  '@db.schema.images_stage',
  200,
  'PNG',
  NULL           -- NULL = convert all pages
);

ステップ 4：変換した画像を検証する

-- Refresh stage directory
ALTER STAGE db.schema.images_stage REFRESH;

-- List converted images
SELECT * FROM DIRECTORY(@db.schema.images_stage)
WHERE relative_path LIKE 'blueprint_%'
ORDER BY relative_path;

ステップ 5：AI_COMPLETE で画像を分析する

-- Analyze a single converted image
SELECT AI_COMPLETE(
  'claude-3-5-sonnet',
  [
    {
      'role': 'user',
      'content': [
        {'type': 'image', 'image_url': {'url': TO_FILE('@db.schema.images_stage', 'blueprint_page_1.png')}},
        {'type': 'text', 'text': 'Analyze this blueprint/diagram. Identify all components, measurements, and annotations.'}
      ]
    }
  ],
  {'max_tokens': 4096}
) AS analysis;

---

2.6b：ビジュアル分析プロンプト

ユーザーに分析したい内容を確認します：

What would you like to analyze in the image?
Options:
1. Charts/graphs - extract data points, trends, labels
2. Blueprints/technical drawings - identify components, measurements
3. Diagrams/flowcharts - describe structure, relationships
4. General visual content - describe what you see
5. Custom analysis - I'll describe what I need

種類別のプロンプトテンプレート：

チャート／グラフ：

SELECT AI_COMPLETE(
  'claude-3-5-sonnet',
  [
    {
      'role': 'user',
      'content': [
        {'type': 'image', 'image_url': {'url': TO_FILE('@stage', 'chart.png')}},
        {'type': 'text', 'text': 'Analyze this chart. Extract:
1. Chart type (bar, line, pie, etc.)
2. Title and axis labels
3. All data points with their values
4. Key trends or insights
5. Any annotations or legends

Format the data points as a table if possible.'}
      ]
    }
  ],
  {'max_tokens': 4096}
) AS chart_analysis;

設計図／技術図面：

SELECT AI_COMPLETE(
  'claude-3-5-sonnet',
  [
    {
      'role': 'user',
      'content': [
        {'type': 'image', 'image_url': {'url': TO_FILE('@stage', 'blueprint.png')}},
        {'type': 'text', 'text': 'Analyze this technical drawing/blueprint. Identify:
1. All labeled components and parts
2. Dimensions and measurements with units
3. Materials specifications if shown
4. Assembly instructions or notes
5. Scale information
6. Any warnings or special instructions'}
      ]
    }
  ],
  {'max_tokens': 4096}
) AS blueprint_analysis;

図／フローチャート：

SELECT AI_COMPLETE(
  'claude-3-5-sonnet',
  [
    {
      'role': 'user',
      'content': [
        {'type': 'image', 'image_url': {'url': TO_FILE('@stage', 'diagram.png')}},
        {'type': 'text', 'text': 'Analyze this diagram/flowchart. Describe:
1. Overall purpose of the diagram
2. All nodes/boxes and their labels
3. Connections and relationships between elements
4. Flow direction and sequence
5. Decision points and branches
6. Start and end points'}
      ]
    }
  ],
  {'max_tokens': 4096}
) AS diagram_analysis;

一般的なビジュアル分析：

SELECT AI_COMPLETE(
  'claude-3-5-sonnet',
  [
    {
      'role': 'user',
      'content': [
        {'type': 'image', 'image_url': {'url': TO_FILE('@stage', 'image.png')}},
        {'type': 'text', 'text': 'Describe this image in detail. Include all visible text, numbers, symbols, and visual elements.'}
      ]
    }
  ],
  {'max_tokens': 4096}
) AS visual_analysis;

---

2.6c：ビジュアル分析の一括処理

複数の画像を分析する場合：

-- Analyze all images in a stage
SELECT 
  relative_path AS image_file,
  AI_COMPLETE(
    'claude-3-5-sonnet',
    [
      {
        'role': 'user',
        'content': [
          {'type': 'image', 'image_url': {'url': TO_FILE('@db.schema.images_stage', relative_path)}},
          {'type': 'text', 'text': 'Analyze this chart and extract all data points.'}
        ]
      }
    ],
    {'max_tokens': 4096}
  ) AS analysis
FROM DIRECTORY(@db.schema.images_stage)
WHERE relative_path LIKE '%.png' OR relative_path LIKE '%.jpg';

---

2.6d：ビジュアル分析結果を保存する

-- Create table for visual analysis results
CREATE TABLE IF NOT EXISTS db.schema.visual_analysis_results (
  result_id INT AUTOINCREMENT,
  image_path STRING,
  source_pdf STRING,           -- Original PDF if converted from PDF
  page_number INT,             -- Page number if converted from PDF
  analysis_type STRING,        -- chart, blueprint, diagram, general
  analysis_result VARIANT,
  analyzed_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Insert analysis result
INSERT INTO db.schema.visual_analysis_results (image_path, source_pdf, page_number, analysis_type, analysis_result)
SELECT 
  'blueprint_page_1.png',
  'blueprint.pdf',
  1,
  'blueprint',
  PARSE_JSON(AI_COMPLETE(
    'claude-3-5-sonnet',
    [
      {
        'role': 'user',
        'content': [
          {'type': 'image', 'image_url': {'url': TO_FILE('@db.schema.images_stage', 'blueprint_page_1.png')}},
          {'type': 'text', 'text': 'Analyze this blueprint. Return a JSON object with keys: components (array), measurements (array), materials (array), notes (string).'}
        ]
      }
    ],
    {'max_tokens': 4096}
  ));

ステップ 3：後処理オプション

抽出／解析が完了したら、ユーザーに質問します：

What would you like to do next?
Options:
1. Done - one-time extraction
2. Store results in a Snowflake table
3. Set up a pipeline for continuous processing
4. Build RAG/search integration

パイプラインのセットアップがリクエストされた場合：

-- Create stream on stage
CREATE OR REPLACE STREAM doc_stream ON STAGE @my_stage;

-- Create task for continuous processing
CREATE OR REPLACE TASK process_docs
  WAREHOUSE = my_warehouse
  SCHEDULE = '1 MINUTE'
WHEN SYSTEM$STREAM_HAS_DATA('doc_stream')
AS
  INSERT INTO extracted_data
  SELECT ... FROM doc_stream;

主な制約

制約	AI_EXTRACT	AI_PARSE_DOCUMENT
最大ファイルサイズ	100 MB	50 MB
最大ページ数	125	500
出力形式	JSON	Markdown

停止ポイント

• ファイル形式がサポートされていない場合は、ステップ 1 の後
• サブスキルへの振り分け前、ステップ 2 の後
• 後処理の選択に備え、抽出完了後

出力

収集したコンテキストとともにユーザーを適切なサブスキルに案内し、その後、後処理を支援します。