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 文件”：**AI_EXTRACT 支持 CSV 文件，AI_PARSE_DOCUMENT 不支持。继续执行步骤 2，确定提取目标。

**如果用户选择“MD 或 EML 文件”：**AI_EXTRACT 支持 MD 和 EML 文件，AI_PARSE_DOCUMENT 不支持。继续执行步骤 2，确定提取目标。

**如果用户选择“HTML 或 TXT 文件”：**AI_EXTRACT 和 AI_PARSE_DOCUMENT 均支持 HTML 和 TXT 文件。继续执行步骤 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）”：

AI_EXTRACT 和 AI_PARSE_DOCUMENT 均不直接支持 Excel 文件。告知用户可选方案：

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 connector 运行）：

-- 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 模式（不适用于视觉分析）。

解析前，询问用户是否需要页面优化，以提升性能并降低成本：

询问用户：

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：测试运行与提示词优化（适用于结构化提取）

仅适用于用户选择“指定字段”或“表格”。

此步骤会先在单个文件上验证提取结果，再进行批量处理，并使用 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;

如果选择“从阶段中选择”：

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

如果选择“是”： 加载 ai-extract/SKILL.md，使用已验证的 responseFormat 进行完整的批量提取。

如果选择“重新开始”： 返回步骤 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-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-备用方法。

---

2.5f-备用方法：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 schema

将用户的字段转换为 JSON schema：

-- 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';

结构化输出 Schema 示例

适用于不同数据类型：

-- 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 模式可捕获更多上下文
需要特定数据类型（数字、布尔值）	使用备用方法——schema 可强制指定类型
简单的扁平字段	使用 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 节所定义”）
• 法律文件 - 条款依赖于文档其他位置的定义
• 研究论文 - 方法论部分会影响对结果的解读
• 技术手册 - 操作步骤会引用其他章节中的图表或规格
• 政策文件 - 例外情况和条件横跨多页
• 贷款／抵押协议 - 利率条款与全文中的条件相关
• 保险单 - 保障详情会引用除外责任和限制
• 并购文件 - 财务数据与法律条款相互关联
• 专利申请 - 权利要求会引用详细说明
• 医疗记录 - 诊断与检测结果和患者病史相关
• 年度报告 - 财务报表与管理层讨论相关
• 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 页
上下文保留	有限（按块）	保留全文档上下文
交叉引用	可能遗漏	会保留
适用场景	页面相互独立（发票）	内容相互关联（合同）
结果聚合	需要	不需要
处理复杂度	较高（多个阶段）	较简单（单次处理）
使用 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 后、转到子技能前停止
• 提取完成后，等待用户选择后处理方式

输出

根据收集到的上下文，将用户引导至相应的子技能，然后协助进行后处理。