Molayo

© 2026 Molayo

Qiita헤드라인2026. 06. 27. 07:21

OCI Document Understanding을 비동기 작업(Asynchronous Job)으로 호출하여 일본어 양식을 자체 구조화

요약

OCI Document Understanding을 활용하여 일본어 양식의 텍스트와 표를 추출하는 비동기 작업 구현 패턴을 소개합니다. Object Storage를 경유하는 비동기 프로세스 관리와 결과 JSON 스키마 정규화 방법을 실무적인 관점에서 다룹니다.

핵심 포인트

  • 비동기 프로세서 작업을 위한 투입-폴링-취득 3단계 플로우 구현
  • Object Storage를 통한 대용량 파일 입출력 처리 방식
  • BCP-47 표준을 활용한 언어 설정 및 결과 JSON 스키마 리맵핑
  • 좌표값 정규화 및 OCI SDK의 안정적인 호출 패턴

PDF나 이미지 양식(청구서, 신청서 등)에서 텍스트와 표를 한꺼번에 추출하고 싶은 상황은 자주 있습니다.

이때 OCI에는 Document Understanding(oci.ai_document)이라는 서비스가 있으며, 일본어 OCR과 표 추출을 높은 정밀도로 수행할 수 있습니다.

다만, 막상 구현하려고 하면 API가 동기 방식이 아니라 **비동기 프로세서 작업 (Asynchronous Processor Job)**이라는 점, 입출력이 Object Storage를 경유한다는 점, 결과 JSON이 camelCase 형태의 독자적인 스키마라는 점 등 은근히 까다로운 포인트들이 몇 가지 있습니다.

이 기사에서는 OCI Document Understanding을 하나의 문서 분석 함수로 통합하기 위한 구현 패턴을 실무적인 관점에서 정리합니다. 프레임워크에 의존하지 않고 그대로 참고할 수 있는 형태로 코드도 제공합니다.

대상은 "OCI로 양식 OCR을 하고 싶지만, processor job의 처리 방법을 모르겠다", "OCR 결과를 자체 스키마로 정규화하여 후속 단계로 전달하고 싶다"는 분들입니다.

먼저 결론을 정리하겠습니다. 구현 시 파악해야 할 핵심은 다음 6가지입니다.

  • 동기 API는 없다. create_processor_job으로 투입 → get_processor_job을 폴링(poll) → 완료 후 Object Storage에서 결과 JSON을 읽는 비동기 플로우로 구성한다.
  • 입출력은 Object Storage를 경유한다. 입력 파일을 put 하고, 출력은 output_prefix/<job_id>/...json에 작성된다.
  • language는 BCP-47로 정규화한다. JPN과 같은 ISO 3자리 문자가 아니라 ja를 전달한다.
  • 결과 JSON은 camelCase의 독자적인 스키마이다. pages[].lines[].text, pages[].tables[].headerRows/bodyRows를 자체 스키마로 **결정론적으로 리맵(remap)**한다.
  • 좌표는 boundingPolygon.normalizedVertices (0~1 비율)로 반환된다. 이를 그대로 xyxy로 정규화하여 보유한다.
  • OCI SDK 호출은 지연 임포트(Lazy Import) + 별도 스레드로 격리하고, 암호화 PEM의 대화형 프롬프트를 사전에 차단한다. 미설정 또는 실패 시에는 안전하게 축소(degrade)시킨다.

이후 단계에서 각각을 순서대로 구현해 나가겠습니다.

처리 흐름은 다음과 같습니다.

입력 파일 (PDF / 이미지)
|
| 1. put_object
...

포인트는 OCI 측은 '투입'과 '결과 취득'이 완전히 분리되어 있다는 점입니다. job을 투입하면 ID를 받고, 그 ID를 사용하여 상태를 폴링(poll)하며, 완료되면 output prefix 하위의 JSON을 가져오는 3단계 구조가 됩니다.

전제로 SDK는 pip install oci가 설치되어 있고, 인증은 ~/.oci/config (API 키 방식)가 준비되어 있다고 가정합니다.

OCI Document Understanding에는 1회 요청으로 즉시 응답을 반환하는 동기 API (인라인 analyze-document)도 있지만, 운영 환경에서 사용한다면 processor job이 유일한 선택지라고 생각합니다.

이유는 다음과 같습니다.

  • 여러 페이지 및 여러 파일을 한꺼번에 던질 수 있다. input_location에 여러 object를 지정할 수 있다.
  • 입출력이 Object Storage에서 완결되므로, 용량이 큰 PDF라도 HTTP 페이로드에 실어 나를 필요가 없다.
  • 결과가 JSON으로 영속화되므로, 나중에 재취득하거나 재처리하기 쉽다.

그 대신, "투입하고, 기다리고, 가져오는" 라이프사이클 관리를 직접 작성해야 합니다. 이 부분이 구현의 핵심입니다.

여기서부터 코드입니다. 길기 때문에 접어두었습니다. 순서대로 배치하면 그대로 하나의 서비스 클래스가 됩니다.

먼저, job 투입에 필요한 비기밀 설정을 하나의 dataclass로 묶습니다. compartment / namespace / 버킷과 같이 "환경마다 바뀌지만 비밀은 아닌" 값들만 갖게 하고, 인증 정보 자체는 포함하지 않는 것이 요령입니다.

입력 버킷(input bucket)이 지정되지 않았다면 Object Storage의 범용 버킷으로, 출력 버킷(output bucket)이 지정되지 않았다면 입력 버킷으로 fallback 하는 해결 로직을 넣어두면 운영이 편리해집니다.

config.py — 설정 dataclass 및 env로부터의 구축

from __future__ import annotations
import json
import os
...

이 부분은 놓치기 쉽지만, 운영 시 확실히 효과를 발휘하는 대책입니다.

OCI SDK의 from_file()로 읽은 config를 클라이언트에 전달했을 때, API 비밀 키의 PEM이 암호화되어 있고 패스프레이즈(passphrase)가 설정되어 있지 않으면, SDK가 표준 입력(standard input)으로부터 패스프레이즈를 묻기 위해 대기합니다. 서버 프로세스나 컨테이너 내부에서 이런 일이 발생하면, 입력을 받을 대상이 없는 상태로 행(hang) 상태에 빠지게 됩니다.

따라서 "클라이언트를 만들기 전에, 암호화된 PEM인지 직접 확인하고, 패스프레이즈가 설정되어 있지 않다면 명시적인 에러를 발생시켜 중단시킨다"라는 가드(guard)를 넣어둡니다.

oci_auth.py — 암호화된 PEM의 대화형 프롬프트를 사전에 차단

from collections.abc import Mapping
from pathlib import Path
from typing import Any
...

입력 파일을 Object Storage에 put 한 후, create_processor_job을 호출합니다.

여기서 중요한 것이 processor_config의 구성입니다. features에는 추출하고 싶은 기능(텍스트 추출, 표 추출, key-value 추출)의 클래스 인스턴스를 나열하고, language에는 BCP-47 형식을 전달합니다. JPN과 같은 ISO 639-2의 3자리 코드를 전달하면 거부될 수 있으므로, ja로 정규화해 둡니다.

job 투입 — put_object + create_processor_job + language 정규화

import importlib
import mimetypes
# 설정 feature 명 -> SDK feature 클래스 명. 알 수 없는 값은 무시한다.
...

job ID를 얻었다면, get_processor_job의 lifecycle_state를 poll 합니다. SUCCEEDED이면 빠져나오고, FAILED / CANCELED 또는 timeout이면 예외를 발생시켜 축퇴(degrade)시킵니다.

완료 후의 결과는 output_prefix/<job_id>/ 하위에 .json 파일로 작성되므로, list_objects로 열거하여 첫 번째 JSON을 get_object로 읽습니다. SDK 버전과 테스트용 fake 버전 모두에서 본문을 추출할 수 있도록 content / raw 중 어느 쪽에도 대응하도록 해둡니다.

완료 대기(poll) 및 결과 JSON 읽기

import json
import time
_SUCCEEDED = {"SUCCEEDED"}
...

이 부분이 이 글에서 가장 전달하고 싶은 핵심입니다.

DU의 결과 JSON은 camelCase의 독자적인 스키마로, pages[].lines[].text (행 텍스트), pages[].tables[].headerRows / bodyRows / footerRows (표), pages[].words[].confidence (신뢰도), detectedDocumentTypes[].documentType (문서 유형)와 같은 구조로 되어 있습니다.

이를 그대로 후속 단계에서 사용하는 것은 매우 고통스럽기 때문에, raw_text / elements / tables / pages라는 다루기 쉬운 자체 스키마로 **결정론적(deterministically)**으로 변환합니다. LLM 등을 거치지 않고 순수하게 매핑(mapping)만으로 옮기는 것이 포인트입니다 (재현성이 확보되고, 테스트하기 쉬우며, 빠릅니다).

좌표는 boundingPolygon.normalizedVertices라는 **0~1 사이의 비율로 된 정점 열(vertex sequence)**로 반환됩니다. 정점 열을 xyxy (좌상단·우하단)로 정규화하여 가지고 있으면, 나중에 "인용 부분을 프리뷰에서 하이라이트한다"와 같은 용도로 사용할 수 있습니다.

remap — 결과 JSON을 raw_text / elements / tables / pages로 옮기기

import math
from collections.abc import Mapping, Sequence
def _as_seq(value) -> Sequence:
...

마지막으로, 지금까지의 과정을 하나의 analyze()

로 통합합니다.

주의할 점은, OCI SDK 호출은 블로킹 (Blocking) 방식이라는 것입니다. FastAPI와 같은 async 애플리케이션에 그대로 두면 이벤트 루프 (Event Loop)를 멈추게 되므로, asyncio.to_thread()

를 사용하여 별도의 스레드로 분리합니다.

또한, SDK 클라이언트는 지연 임포트 (Lazy Import) + 지연 생성 (Lazy Instantiation) 방식으로 구성하여, 생성자에서 document_client

/ storage_client

를 주입할 수 있도록 합니다. 이렇게 하면 테스트 시 페이크 (Fake)를 주입할 수 있어, 실제 OCI 없이도 결정론적 (Deterministic)으로 검증할 수 있습니다.

service.py — analyze() 본체와 지연 클라이언트 생성

import asyncio
import importlib
import logging
...

사용법은 간단합니다.

import asyncio
config = DocumentUnderstandingConfig.from_env()
service = DocumentUnderstandingService(config)
...

실제 OCI를 호출하지 않고 테스트할 수 있는 것이 클라이언트 주입 방식의 장점입니다. create_processor_job

/ get_processor_job

/ list_objects

/ get_object

를 가진 페이크 (Fake)를 준비하고, get_processor_job

이 즉시 SUCCEEDED

를 반환하도록 설정하면, remap까지의 모든 경로를 빠르고 결정론적으로 검증할 수 있습니다.

test_service.py — fake SDK를 통한 검증

import asyncio
import json
import types
...

구현 중에 실수하기 쉬운 포인트들을 마지막으로 정리합니다.

  • language는 BCP-47입니다. JPN이 아니라 ja를 사용해야 합니다. 이 부분은 별칭 (Alias) 테이블로 처리해 두어야 사고를 방지할 수 있습니다.
  • 출력 위치에 주의하세요. output_location에 전달한 prefix 바로 아래가 아니라, job ID의 하위 디렉토리에 JSON이 들어갑니다. (output_prefix/<job_id>/ 하위)
  • **좌표는 normalizedVertices (0~1 비율)**입니다. 절대 좌표는 반환되지 않으므로, 표시 측에서 반드시 페이지 치수와 곱해서 사용하는 것을 전제로 가져가야 합니다.
  • 암호화 PEM의 대화형 프롬프트. 서버나 컨테이너 환경에서 행 (Hang) 상태가 되는 원인이 되므로, 클라이언트 생성 전에 방지 로직을 넣어야 합니다.
  • SDK 호출은 블로킹입니다. async 애플리케이션에서는 asyncio.to_thread()를 사용하여 이벤트 루프를 멈추지 않도록 합니다.
  • 미설정 또는 실패 시 축퇴 (Degradation). None을 반환하여 후속 단계에서 다른 수단으로 폴백 (Fallback)할 수 있도록 설계하면 운영이 안정화됩니다.

OCI Document Understanding을 실무에 도입할 때의 요점은 다음과 같습니다.

  • 동기 API가 아닌 비동기 processor job (투입 → 폴링 (Poll) → Object Storage에서 결과 취득) 방식으로 구성합니다.
  • 결과 JSON은 camelCase 형태의 독자적인 스키마이므로, 결정론적인 remap을 통해 다루기 쉬운 자체 스키마로 옮깁니다.
  • 인증의 비대화(Non-interactive) 처리, 블로킹 호출의 격리, 클라이언트 주입을 통한 테스트 용이성을 초기 단계부터 반영해 두면 이후 작업이 매우 수월해집니다.

특히, remap 과정을 LLM 등에 맡기지 않고 순수한 매핑 (Mapping) 작업으로 유지하면 재현성과 테스트 가능성이 비약적으로 향상됩니다. OCI로 양식 OCR을 구축하려는 분들에게 도움이 되기를 바랍니다.

AI 자동 생성 콘텐츠

본 콘텐츠는 Qiita AI의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.

원문 바로가기
0

댓글

0