Textract v2.0 完全ガイド（Machine Learning & AI）

概要

Amazon Textract は、PDF・画像からテキスト・フォーム・テーブルを自動抽出するドキュメント処理 AI サービスです。単純な OCR を超えて、フォームのキー・値のペア（Key-Value）、テーブルの行列構造、署名検出、バーコード認識、自然言語クエリ（Textract Queries）による目的特化抽出を提供します。AnalyzeDocument（フォーム・テーブル）・AnalyzeID（身分証明書）・AnalyzeExpense（請求書）・AnalyzeLending（ローン書類）・Queries（目的特化抽出）の 5 つの API で、銀行融資申請・保険請求・医療問診票・物流書類・税務書類の自動処理を実現します。A2I（人間確認）との組み合わせで高精度パイプラインを構築できます。

課題と特徴

従来型 OCR の課題

• テキストのみ抽出: フォームの構造（キー・値の関連性）を認識できない
• テーブル解析が困難: セル境界・行列の理解が手作業
• スキャン品質低下への弱さ: 手書き・折れ曲がり・汚れに対応困難
• 大量書類処理の人手コスト: スキャン後の手作業入力が膨大

Textract の特徴

アーキテクチャ

graph TB
    subgraph Input["入力"]
        A["単一ページ画像"]
        B["マルチページ PDF"]
    end
    
    subgraph APIs["Textract API"]
        C["DetectDocumentText"]
        D["AnalyzeDocument"]
        E["AnalyzeID"]
        F["AnalyzeExpense"]
        G["AnalyzeLending"]
        H["Queries"]
    end
    
    subgraph Processing["処理エンジン"]
        I["テキスト抽出"]
        J["フォーム解析"]
        K["テーブル認識"]
        L["署名検出"]
        M["クエリ処理"]
    end
    
    subgraph Output["出力"]
        N["JSON<br/>テキスト・位置情報"]
        O["Key-Value<br/>ペア"]
        P["Table<br/>行列構造"]
    end
    
    A --> C
    B --> D
    C --> I
    D --> J
    D --> K
    E --> L
    H --> M
    I --> N
    J --> O
    K --> P

コアコンポーネント

1. DetectDocumentText（シンプルな OCR）

import boto3

textract = boto3.client('textract', region_name='ap-northeast-1')

# S3 の画像から OCR
response = textract.detect_document_text(
    Document={
        'S3Object': {
            'Bucket': 'my-bucket',
            'Name': 'document.png'
        }
    }
)

# テキストを抽出
for block in response['Blocks']:
    if block['BlockType'] == 'LINE':
        print(block['Text'])

2. AnalyzeDocument - Forms（フォーム抽出）

# 応募書類・申請書からキー・値を抽出

response = textract.analyze_document(
    Document={'S3Object': {'Bucket': 'my-bucket', 'Name': 'application-form.pdf'}},
    FeatureTypes=['FORMS', 'TABLES']
)

# キー・値のペアを抽出
key_map = {}
value_map = {}

for block in response['Blocks']:
    if block['BlockType'] == 'KEY_VALUE_SET':
        if 'KEY' in block.get('EntityTypes', []):
            key_map[block['Id']] = block
        elif 'VALUE' in block.get('EntityTypes', []):
            value_map[block['Id']] = block

# キー・値を対応付け
for key_id, key_block in key_map.items():
    values = key_block.get('Relationships', [])
    for relationship in values:
        if relationship['Type'] == 'VALUE':
            for value_id in relationship['Ids']:
                print(f"キー: {get_text(key_block, response['Blocks'])}")
                print(f"値: {get_text(value_map[value_id], response['Blocks'])}")

def get_text(block, blocks):
    text = ''
    if 'Relationships' in block:
        for relationship in block.get('Relationships', []):
            if relationship['Type'] == 'CHILD':
                for child_id in relationship['Ids']:
                    child = next((b for b in blocks if b['Id'] == child_id), None)
                    if child and child['BlockType'] == 'WORD':
                        text += child['Text'] + ' '
    return text.strip()

3. AnalyzeDocument - Tables（テーブル抽出）

# テーブルを行・列単位で抽出

response = textract.analyze_document(
    Document={'S3Object': {'Bucket': 'my-bucket', 'Name': 'invoice.pdf'}},
    FeatureTypes=['TABLES']
)

# テーブルデータを pandas DataFrame に変換
import pandas as pd

tables = []
table_blocks = [b for b in response['Blocks'] if b['BlockType'] == 'TABLE']

for table in table_blocks:
    rows = {}
    
    for relationship in table.get('Relationships', []):
        if relationship['Type'] == 'CHILD':
            for cell_id in relationship['Ids']:
                cell = next((b for b in response['Blocks'] if b['Id'] == cell_id), None)
                if cell and cell['BlockType'] == 'CELL':
                    row_idx = cell['RowIndex']
                    col_idx = cell['ColumnIndex']
                    
                    if row_idx not in rows:
                        rows[row_idx] = {}
                    
                    # セル内のテキストを取得
                    cell_text = get_cell_text(cell, response['Blocks'])
                    rows[row_idx][col_idx] = cell_text
    
    # DataFrame に変換
    df = pd.DataFrame(rows).T
    tables.append(df)

4. AnalyzeID（身分証明書抽出）

# 運転免許証・パスポートから情報抽出

response = textract.analyze_id(
    DocumentPages=[{
        'S3Object': {
            'Bucket': 'my-bucket',
            'Name': 'drivers-license.jpg'
        }
    }]
)

# 抽出されたフィールド
for document in response['DocumentMetadata']:
    for page in document['Pages']:
        for field in page['DocumentMetadataList']:
            print(f"{field['MRZCode']}")  # MRZ（機械可読ゾーン）

# 特定フィールドを抽出
identity_documents = response['IdentityDocuments']
for doc in identity_documents:
    for field_name, field_value in doc['IdentityDocumentFields'].items():
        print(f"{field_name}: {field_value['ValueDetection']['Text']}")

# 出力例: DOCUMENT_TYPE, FIRST_NAME, LAST_NAME, DATE_OF_BIRTH, EXPIRATION_DATE

5. AnalyzeExpense（請求書・レシート抽出）

# 請求書からビジネス情報を自動抽出

response = textract.analyze_expense(
    Document={'S3Object': {'Bucket': 'my-bucket', 'Name': 'invoice.pdf'}}
)

# 請求書の構造化データを抽出
for expense_doc in response['ExpenseDocuments']:
    summary_fields = expense_doc['SummaryFields']
    line_items = expense_doc['LineItemGroups']
    
    # サマリー: 請求日、合計額、ベンダー等
    for field in summary_fields:
        field_type = field['Type']['Text']
        field_value = field['ValueDetection']['Text']
        confidence = field['ValueDetection']['Confidence']
        print(f"{field_type}: {field_value} ({confidence:.1%})")
    
    # ラインアイテム: 各行の商品・金額
    for line_item in line_items:
        for item_field in line_item['LineItemExpenseFields']:
            print(f"  {item_field['Type']['Text']}: {item_field['ValueDetection']['Text']}")

# 出力例
"""
INVOICE_RECEIPT_ID: INV-2026-001 (98%)
INVOICE_RECEIPT_DATE: 2026-04-26 (99%)
TOTAL: $1,234.56 (99%)
VENDOR_NAME: Acme Corporation (95%)

Line items:
  ITEM: Widget A (99%)
  QUANTITY: 5 (98%)
  UNIT_PRICE: $200.00 (97%)
  AMOUNT: $1,000.00 (99%)
"""

6. AnalyzeLending（ローン書類分析）

# ローン申請書類を自動分類・処理

response = textract.analyze_lending(
    DocumentPages=[{
        'S3Object': {
            'Bucket': 'my-bucket',
            'Name': 'loan-package.pdf'
        }
    }]
)

# 自動分類された書類タイプ
for document in response['Results']:
    document_type = document['DocumentType']  # e.g., W2, 1099, TAX_RETURN
    print(f"書類タイプ: {document_type}")
    
    # 構造化フィールド
    extracted_fields = document['Extractions']
    for field in extracted_fields:
        field_label = field['Type']
        field_value = field['ValueDetections'][0]['Text']
        print(f"  {field_label}: {field_value}")

7. Textract Queries（自然言語クエリ）

# 「このドキュメントの合計金額は？」のような自然言語クエリで抽出

response = textract.analyze_document(
    Document={'S3Object': {'Bucket': 'my-bucket', 'Name': 'invoice.pdf'}},
    FeatureTypes=['QUERIES', 'TABLES', 'FORMS'],
    QueriesConfig={
        'Queries': [
            {'Text': 'この請求書の合計金額はいくらですか？'},
            {'Text': '納品日はいつですか？'},
            {'Text': 'どの商品が最も高い金額ですか？'}
        ]
    }
)

# クエリの回答を取得
for query_result in response.get('QueryResponses', []):
    query_text = query_result['Query']['Text']
    answer = query_result.get('Answer', 'No answer found')
    confidence = query_result.get('Confidence', 0)
    print(f"Q: {query_text}")
    print(f"A: {answer} ({confidence:.1%})")

8. 非同期バッチ処理

import time

# マルチページ PDF や大量ドキュメント用の非同期処理

response = textract.start_document_analysis(
    DocumentLocation={
        'S3Object': {
            'Bucket': 'my-bucket',
            'Name': 'large-form.pdf'
        }
    },
    FeatureTypes=['FORMS', 'TABLES'],
    ClientRequestToken='my-request-token-001',
    NotificationChannel={
        'SNSTopicArn': 'arn:aws:sns:ap-northeast-1:123456789:textract-complete',
        'RoleArn': 'arn:aws:iam::123456789:role/TextractRole'
    },
    OutputConfig={
        'S3Bucket': 'my-bucket',
        'S3Prefix': 'textract-output/'
    }
)

job_id = response['JobId']

# SNS 通知待機
# または手動でポーリング
while True:
    result = textract.get_document_analysis(JobId=job_id)
    if result['DocumentMetadata']['Pages'] > 0:
        job_status = result['JobStatus']
        if job_status == 'SUCCEEDED':
            print("✓ 分析完了")
            print(f"ページ数: {result['DocumentMetadata']['Pages']}")
            break
        elif job_status == 'FAILED':
            print("✗ 分析失敗")
            break
    time.sleep(5)

主要ユースケース

1. 銀行融資申請の自動処理

class LoanApplicationProcessor:
    def __init__(self):
        self.textract = boto3.client('textract')
        self.comprehend = boto3.client('comprehend')
    
    def process_loan_application(self, application_pdf_uri):
        """融資申請書類を自動処理して審査スコア化"""
        
        # Step 1: Textract でローン書類を分析
        response = self.textract.analyze_lending(
            DocumentPages=[{'S3Object': {'Bucket': bucket, 'Name': key}}]
        )
        
        # Step 2: 構造化データを抽出
        applicant_info = {}
        financial_info = {}
        
        for doc in response['Results']:
            for field in doc['Extractions']:
                if 'NAME' in field['Type']:
                    applicant_info['name'] = field['ValueDetections'][0]['Text']
                elif 'INCOME' in field['Type']:
                    financial_info['income'] = field['ValueDetections'][0]['Text']
                elif 'DEBT' in field['Type']:
                    financial_info['debt'] = field['ValueDetections'][0]['Text']
        
        # Step 3: Comprehend で追加の文書テキストを分析
        # ローンスコア計算
        loan_score = calculate_loan_score(applicant_info, financial_info)
        
        return {
            'applicant': applicant_info,
            'financials': financial_info,
            'loan_score': loan_score,
            'recommendation': 'APPROVE' if loan_score > 700 else 'REVIEW'
        }

2. 医療保険請求の自動化

class InsuranceClaimProcessor:
    def __init__(self):
        self.textract = boto3.client('textract')
        self.dynamodb = boto3.resource('dynamodb')
    
    def process_claim_form(self, claim_form_pdf):
        """医療保険請求書を自動処理"""
        
        # Step 1: Textract でフォーム・テーブルを抽出
        response = self.textract.analyze_document(
            Document={'S3Object': claim_form_pdf},
            FeatureTypes=['FORMS', 'TABLES']
        )
        
        # Step 2: キー・値を抽出
        claim_data = {
            'claimant_name': '',
            'claim_date': '',
            'service_date': '',
            'provider_name': '',
            'claim_amount': '',
            'procedures': []
        }
        
        # フォーム処理
        # テーブル処理（医療手続き明細）
        
        # Step 3: DynamoDB に保存
        table = self.dynamodb.Table('InsuranceClaims')
        table.put_item(Item=claim_data)

3. 物流・請求書の自動処理

class LogisticsDocumentProcessor:
    def __init__(self):
        self.textract = boto3.client('textract')
        self.s3 = boto3.client('s3')
    
    def auto_process_invoice(self, invoice_s3_uri):
        """請求書を自動処理して ERP に送信"""
        
        # Step 1: AnalyzeExpense で請求書分析
        response = self.textract.analyze_expense(
            Document={'S3Object': invoice_s3_uri}
        )
        
        # Step 2: 構造化データ抽出
        invoice_summary = {}
        for expense_doc in response['ExpenseDocuments']:
            for field in expense_doc['SummaryFields']:
                field_type = field['Type']['Text']
                field_value = field['ValueDetection']['Text']
                invoice_summary[field_type] = field_value
        
        # Step 3: ERP API に送信
        # Step 4: 支払予定日をスケジュール

4. A2I（人間確認）との組み合わせ

class HighPrecisionDocumentProcessing:
    def __init__(self):
        self.textract = boto3.client('textract')
        self.a2i = boto3.client('sagemaker-a2i-runtime')
    
    def extract_with_human_review(self, document_uri):
        """信頼度低いフィールドは人間レビュー"""
        
        # Step 1: Textract で抽出
        response = self.textract.analyze_document(
            Document=document_uri,
            FeatureTypes=['FORMS', 'TABLES']
        )
        
        # Step 2: 信頼度をチェック
        low_confidence_fields = []
        for block in response['Blocks']:
            if block.get('Confidence', 100) < 85:
                low_confidence_fields.append(block)
        
        # Step 3: 信頼度低いフィールドを A2I で人間レビュー
        if low_confidence_fields:
            human_task_response = self.a2i.start_human_loop(
                HumanLoopName=f'textract-review-{uuid.uuid4()}',
                HumanLoopConfig={
                    'WorkteamArn': 'arn:aws:sagemaker:ap-northeast-1:123456789:workteam/...',
                    'HumanTaskUiArn': '...',
                    'TaskTitle': 'Document Field Verification',
                    'TaskDescription': 'Please verify the extracted fields',
                    'MaxConcurrentTaskCount': 1
                },
                InputContent={
                    'TextContent': json.dumps(low_confidence_fields)
                }
            )
        
        return response

5. 税務書類・給与明細の処理

class TaxDocumentProcessor:
    def __init__(self):
        self.textract = boto3.client('textract')
    
    def extract_tax_info(self, tax_document_pdf):
        """税務書類から所得・税額を抽出"""
        
        response = self.textract.analyze_lending(
            DocumentPages=[{'S3Object': tax_document_pdf}]
        )
        
        # ローン分析 API は W2・1099・確定申告書に対応
        for doc in response['Results']:
            doc_type = doc['DocumentType']
            
            tax_info = {
                'document_type': doc_type,
                'gross_income': '',
                'deductions': '',
                'tax_liability': ''
            }
            
            for extraction in doc['Extractions']:
                if 'INCOME' in extraction['Type']:
                    tax_info['gross_income'] = extraction['ValueDetections'][0]['Text']
                elif 'DEDUCTION' in extraction['Type']:
                    tax_info['deductions'] = extraction['ValueDetections'][0]['Text']
                elif 'TAX' in extraction['Type']:
                    tax_info['tax_liability'] = extraction['ValueDetections'][0]['Text']
            
            return tax_info

CLI・SDK・IaC 例（スペース制限により簡潔）

CLI 例

# テキスト検出
aws textract detect-document-text \
  --document 'S3Object={Bucket=my-bucket,Name=doc.png}' \
  --region ap-northeast-1

# ドキュメント分析
aws textract analyze-document \
  --document 'S3Object={Bucket=my-bucket,Name=form.pdf}' \
  --feature-types FORMS TABLES \
  --region ap-northeast-1

# ID 分析
aws textract analyze-id \
  --document-pages 'S3Object={Bucket=my-bucket,Name=license.jpg}' \
  --region ap-northeast-1

# 請求書分析
aws textract analyze-expense \
  --document 'S3Object={Bucket=my-bucket,Name=invoice.pdf}' \
  --region ap-northeast-1

# 非同期分析開始
aws textract start-document-analysis \
  --document-location 'S3Object={Bucket=my-bucket,Name=large.pdf}' \
  --feature-types FORMS TABLES \
  --notification-channel SNSTopicArn=arn:aws:sns:... RoleArn=arn:aws:iam::... \
  --region ap-northeast-1

比較表

ベストプラクティス・トラブルシューティング・最新動向（スペース制限）

学習リソース

1. Amazon Textract Developer Guide
2. Textract API Reference
3. Custom Queries Guide
4. Google Document AI
5. Azure Form Recognizer

採用判断チェックリスト

• [ ] PDF・画像からテキスト抽出が必要か
• [ ] フォームのキー・値抽出が必要か
• [ ] テーブルを構造化データとして抽出したいか
• [ ] 身分証明書から特定フィールド抽出が必要か
• [ ] 請求書・レシートの自動処理が必要か
• [ ] 大量ドキュメントを非同期処理したいか
• [ ] 信頼度低いフィールドに A2I で人間レビュー追加するか
• [ ] 自然言語クエリで柔軟に情報抽出したいか

まとめ

Textract は 「PDF・画像から構造化されたテキスト・フォーム・テーブルを自動抽出するドキュメント処理 AI」。AnalyzeDocument・AnalyzeID・AnalyzeExpense・AnalyzeLending・Queries の 5 つの API で、銀行融資申請・保険請求・医療問診票・物流書類・税務書類の自動処理を実現します。A2I との組み合わせで高精度パイプラインを構築でき、Comprehend・Bedrock との連携で自動インサイト抽出も可能です。

最終更新：2026-04-26 バージョン：v2.0