抽出 (構造化) APIのベストプラクティス

{
  "ai_agent": {
    "type": "ai_agent_id",
    "id": "enhanced_extract_agent"
  }
}

{
  "items": [{ "id": "123456789", "type": "file" }],
  "fields": [
    {
      "key": "contract_summary",
      "type": "string",
      "prompt": "Summarize the key obligations."
    }
  ],
  "ai_agent": {
    "type": "ai_agent_extract_structured",
    "basic_text": { "num_tokens_for_completion": 4096 },
    "long_text": { "num_tokens_for_completion": 4096 }
  }
}

import json
import logging
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests
from box_sdk_gen import BoxJWTAuth, JWTConfig

logging.basicConfig(level=logging.INFO)
LOGGER = logging.getLogger(__name__)

JWT_CONFIG_PATH = "config.json"
USER_ID = ""
FILE_IDS = []

NUM_PARALLEL_REQUESTS = 8
TIMEOUT_SECONDS = 120
MAX_RETRIES = 6
BACKOFF_INITIAL_SECONDS = 30
BACKOFF_MAX_SECONDS = 480

RETRYABLE_STATUS_CODES = [412, 429, 503, 544]

FIELDS = [
    {
        "type": "string",
        "key": "name",
        "displayName": "Name",
        "description": "The document name",
        "prompt": "What is the name of this document?",
    },
    {
        "type": "string",
        "key": "document_number",
        "displayName": "Document Number",
        "description": "The document number",
        "prompt": "The document number",
    },
]


def send_request(auth: BoxJWTAuth, file_id: str):
    response = None

    for attempt in range(MAX_RETRIES + 1):
        try:
            response = requests.post(
                "https://api.box.com/2.0/ai/extract_structured",
                headers={
                    "content-type": "application/json",
                    "authorization": f"Bearer {auth.retrieve_token().access_token}",
                },
                json={
                    "items": [{"id": file_id, "type": "file"}],
                    "fields": FIELDS,
                },
                timeout=TIMEOUT_SECONDS,
            )
        except Exception as exc:
            LOGGER.info("Transport error for file %s: %s", file_id, exc)
            response = None

        if response is not None and response.status_code == 200:
            break

        status = None if response is None else response.status_code
        if status is not None and status not in RETRYABLE_STATUS_CODES:
            break

        if attempt < MAX_RETRIES:
            wait_seconds = min(
                BACKOFF_INITIAL_SECONDS * (2 ** attempt),
                BACKOFF_MAX_SECONDS,
            )
            LOGGER.info(
                "Pacing retry for file %s in %ss (status %s)",
                file_id,
                wait_seconds,
                status,
            )
            time.sleep(wait_seconds)

    return response


if __name__ == "__main__":
    with open(JWT_CONFIG_PATH, "rt") as file:
        config_json = json.dumps(json.load(file))

    auth = BoxJWTAuth(
        config=JWTConfig.from_config_json_string(config_json_string=config_json)
    )
    if USER_ID:
        auth = auth.with_user_subject(USER_ID)

    results = []
    with ThreadPoolExecutor(max_workers=NUM_PARALLEL_REQUESTS) as executor:
        futures = {
            executor.submit(send_request, auth, fid): fid for fid in FILE_IDS
        }
        for future in as_completed(futures):
            try:
                results.append(future.result())
            except Exception as exc:
                LOGGER.info(
                    "Failed for file %s", futures[future], exc_info=exc
                )
                results.append(None)

    succeeded = sum(
        1 for r in results if r is not None and r.status_code == 200
    )
    LOGGER.info(
        "Batch complete — %s/%s files successfully processed.",
        succeeded,
        len(FILE_IDS),
    )