Kwangmin Kim - Document Intelligence for RAG

1 RAG를 위한 Document Intelligence

기능 설명, 기본 설정 및 인스턴스 생성은 20-az-document-intelligence.qmd를 참고한다.

1.1 RAG 시스템에서의 핵심 가치

Document Intelligence는 단순 OCR을 넘어 문서 구조 이해를 제공한다. 이는 RAG 시스템의 검색 정확도와 답변 품질에 직접적인 영향을 미친다.

일반 OCR vs Document Intelligence (RAG 관점)

측면	일반 OCR	Document Intelligence
텍스트 추출	✅ 가능	✅ 가능
문서 구조	❌ 미지원	✅ 제목, 단락, 표 구분
의미적 청킹	❌ 불가능	✅ 섹션 단위 분할
표 데이터	텍스트로 뭉개짐	구조화된 JSON
메타데이터	없음	페이지, 좌표, 역할
RAG 검색 품질	낮음	높음

1.2 RAG 파이프라인에서의 역할

문서 → Document Intelligence → 구조화된 JSON → 지능형 청킹 → Vector DB → RAG 검색

핵심 차별점:
1. 의미 단위 분할: 단락/섹션 기준으로 청크 생성 → 문맥 보존
2. 표 데이터 보존: DataFrame으로 변환 가능 → 숫자 데이터 정확도 향상
3. 계층적 메타데이터: 제목-단락 관계 파악 → 관련 섹션 검색
4. 다중 모드 검색: 텍스트 + 표 + 이미지 캡션 통합

2 Document Intelligence 모델

Azure Document Intelligence는 문서 유형에 따라 다양한 사전 훈련 모델을 제공한다:

2.1 Layout API (범용 레이아웃 분석)

용도: 모든 종류의 문서 레이아웃 분석

추출 정보:
- 텍스트 (OCR)
- 표 (행, 열 구조)
- 제목 및 섹션
- 단락 및 줄바꿈
- 선택 마크 (체크박스)

RAG 시스템 권장: 가장 많이 사용

가격: 페이지당 $0.01

2.2 Read API (텍스트 추출 전용)

용도: 순수 텍스트 추출 (레이아웃 무시)

추출 정보:
- 텍스트만 (구조 정보 없음)
- 빠른 처리 속도

RAG 시스템 권장: 간단한 문서, 비용 절감 필요 시

가격: 페이지당 $0.0015

2.3 Prebuilt Models (특화 모델)

모델	용도	추출 정보
Invoice	청구서, 세금계산서	날짜, 금액, 품목, 공급자
Receipt	영수증	상점명, 날짜, 총액, 항목
ID Document	신분증, 여권	이름, 생년월일, 주소
Business Card	명함	이름, 직함, 연락처
W-2	미국 세금 양식	급여, 세금

RAG 시스템 권장: 특정 문서 타입만 처리 시 유용

3 빠른 시작 (리소스 생성)

상세 설정 가이드: 20-az-document-intelligence.qmd

RAG 프로젝트용 리소스 생성 (Azure CLI):

# Document Intelligence 리소스 생성  
az cognitiveservices account create \  
    --name doc-intel-rag-prod \  
    --resource-group rg-rag-prod \  
    --kind FormRecognizer \  
    --sku S0 \  
    --location koreacentral \  
    --yes  

# 키 및 엔드포인트 조회  
az cognitiveservices account keys list \  
    --name doc-intel-rag-prod \  
    --resource-group rg-rag-prod

RAG 프로젝트 체크리스트:
- [ ] Standard S0 계층 선택 (F0는 Layout API 제한적)
- [ ] Korea Central 리전 (한국어 문서 처리 최적화)
- [ ] 키를 환경변수로 저장 (.env 파일)
- [ ] Blob Storage와 동일 리전 배치 (네트워크 비용 절감)

4 환경 설정

4.1 Python SDK 설치

# Document Intelligence + RAG 파이프라인 필수 패키지  
pip install azure-ai-formrecognizer  
pip install azure-storage-blob  
pip install langchain langchain-community  
pip install python-dotenv  
pip install pandas  # 표 데이터 처리용

4.2 환경 변수 설정

.env 파일:

# Document Intelligence  
AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT=https://doc-intel-rag-prod.cognitiveservices.azure.com/  
AZURE_DOCUMENT_INTELLIGENCE_KEY=your-key-here  

# Blob Storage (문서 저장소)  
AZURE_STORAGE_CONNECTION_STRING=DefaultEndpointsProtocol=https;...  
AZURE_STORAGE_KEY=your-storage-key

5 Layout API 사용법

5.1 로컬 파일 분석

from azure.ai.formrecognizer import DocumentAnalysisClient  
from azure.core.credentials import AzureKeyCredential  
from dotenv import load_dotenv  
import os  

load_dotenv()  

# 클라이언트 생성  
endpoint = os.getenv("AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT")  
key = os.getenv("AZURE_DOCUMENT_INTELLIGENCE_KEY")  

document_analysis_client = DocumentAnalysisClient(  
    endpoint=endpoint,  
    credential=AzureKeyCredential(key)  
)  

# 로컬 PDF 파일 분석  
with open("sample.pdf", "rb") as f:  
    poller = document_analysis_client.begin_analyze_document(  
        "prebuilt-layout", document=f  
    )  
    result = poller.result()  

# 결과 출력  
print(f"분석된 페이지 수: {len(result.pages)}")  
print(f"추출된 단락 수: {len(result.paragraphs)}")  
print(f"추출된 표 수: {len(result.tables)}")

5.2 URL에서 문서 분석

# Azure Blob Storage URL로 분석  
document_url = "https://stragdocs2025.blob.core.windows.net/rag-documents/sample.pdf?<sas-token>"  

poller = document_analysis_client.begin_analyze_document_from_url(  
    "prebuilt-layout", document_url=document_url  
)  
result = poller.result()  

print("문서 분석 완료")

6 텍스트 추출

6.1 전체 텍스트 추출

def extract_full_text(result):  
    """문서에서 전체 텍스트 추출"""  
    full_text = []  
    
    # 페이지별 텍스트 추출  
    for page in result.pages:  
        page_text = []  
        for line in page.lines:  
            page_text.append(line.content)  
        
        full_text.append("\n".join(page_text))  
    
    return "\n\n".join(full_text)  

# 사용 예시  
text = extract_full_text(result)  
print(f"추출된 텍스트 (앞 500자):\n{text[:500]}")

6.2 단락 단위 추출

def extract_paragraphs(result):  
    """단락 단위로 텍스트 추출 (레이아웃 유지)"""  
    paragraphs = []  
    
    for paragraph in result.paragraphs:  
        paragraphs.append({  
            "content": paragraph.content,  
            "role": paragraph.role,  # title, sectionHeading, paragraph 등  
            "page_number": paragraph.bounding_regions[0].page_number if paragraph.bounding_regions else None  
        })  
    
    return paragraphs  

# 사용 예시  
paragraphs = extract_paragraphs(result)  
print(f"총 {len(paragraphs)}개 단락 추출")  

# 제목만 추출  
titles = [p for p in paragraphs if p["role"] == "title"]  
print(f"\n문서 제목들:")  
for title in titles:  
    print(f"- {title['content']} (페이지 {title['page_number']})")

7 표 추출

7.1 표 구조 파싱

def extract_tables(result):  
    """문서에서 표 추출"""  
    tables_data = []  
    
    for table_idx, table in enumerate(result.tables):  
        # 표 메타데이터  
        table_info = {  
            "table_id": table_idx + 1,  
            "row_count": table.row_count,  
            "column_count": table.column_count,  
            "page_number": table.bounding_regions[0].page_number,  
            "cells": []  
        }  
        
        # 셀 데이터  
        for cell in table.cells:  
            table_info["cells"].append({  
                "row_index": cell.row_index,  
                "column_index": cell.column_index,  
                "content": cell.content,  
                "kind": cell.kind  # columnHeader, rowHeader, content, stub  
            })  
        
        tables_data.append(table_info)  
    
    return tables_data  

# 사용 예시  
tables = extract_tables(result)  
print(f"추출된 표 개수: {len(tables)}")  

for table in tables:  
    print(f"\n[표 {table['table_id']}] 페이지 {table['page_number']}")  
    print(f"크기: {table['row_count']}행 × {table['column_count']}열")

7.2 표를 Pandas DataFrame으로 변환

import pandas as pd  

def table_to_dataframe(table):  
    """표를 Pandas DataFrame으로 변환"""  
    # 표 초기화 (빈 셀 포함)  
    data = [[None] * table["column_count"] for _ in range(table["row_count"])]  
    
    # 셀 데이터 채우기  
    for cell in table["cells"]:  
        data[cell["row_index"]][cell["column_index"]] = cell["content"]  
    
    # DataFrame 생성 (첫 행을 헤더로)  
    df = pd.DataFrame(data[1:], columns=data[0])  
    return df  

# 사용 예시  
if tables:  
    df = table_to_dataframe(tables[0])  
    print("\n표 데이터 (DataFrame):")  
    print(df)

8 레이아웃 기반 문서 분할

RAG 시스템에서는 문서 구조를 고려한 청크 분할이 중요하다.

from langchain_core.documents import Document  
from typing import List  

def split_by_layout(result, max_chunk_size: int = 1000) -> List[Document]:  
    """레이아웃 기반 문서 분할"""  
    documents = []  
    current_chunk = []  
    current_size = 0  
    current_page = 1  
    
    for paragraph in result.paragraphs:  
        # 단락 정보  
        content = paragraph.content  
        role = paragraph.role or "paragraph"  
        page_num = paragraph.bounding_regions[0].page_number if paragraph.bounding_regions else current_page  
        
        # 제목은 새로운 청크 시작  
        if role in ["title", "sectionHeading"] and current_chunk:  
            # 이전 청크 저장  
            doc = Document(  
                page_content="\n\n".join(current_chunk),  
                metadata={  
                    "page": current_page,  
                    "chunk_type": "section"  
                }  
            )  
            documents.append(doc)  
            current_chunk = []  
            current_size = 0  
        
        # 현재 단락 추가  
        current_chunk.append(content)  
        current_size += len(content)  
        current_page = page_num  
        
        # 최대 크기 초과 시 청크 분할  
        if current_size >= max_chunk_size:  
            doc = Document(  
                page_content="\n\n".join(current_chunk),  
                metadata={  
                    "page": current_page,  
                    "chunk_type": "paragraph"  
                }  
            )  
            documents.append(doc)  
            current_chunk = []  
            current_size = 0  
    
    # 마지막 청크  
    if current_chunk:  
        doc = Document(  
            page_content="\n\n".join(current_chunk),  
            metadata={  
                "page": current_page,  
                "chunk_type": "paragraph"  
            }  
        )  
        documents.append(doc)  
    
    return documents  

# 사용 예시  
documents = split_by_layout(result, max_chunk_size=1000)  
print(f"생성된 청크 수: {len(documents)}")  
print(f"\n첫 번째 청크 (앞 300자):\n{documents[0].page_content[:300]}")

9 JSON 출력 구조 완벽 분석

Document Intelligence의 JSON 출력은 RAG/Agent 시스템에서 활용할 수 있는 풍부한 메타데이터를 제공한다.

9.1 전체 JSON 구조

import json  

def analyze_json_structure(result):  
    """Document Intelligence 결과의 JSON 구조 분석"""  
    
    # 결과를 딕셔너리로 변환  
    result_dict = {  
        "api_version": result.api_version,  
        "model_id": result.model_id,  
        "content": result.content[:200] + "...",  # 전체 텍스트 미리보기  
        
        # 페이지 정보  
        "pages_info": {  
            "count": len(result.pages),  
            "details": [  
                {  
                    "page_number": page.page_number,  
                    "width": page.width,  
                    "height": page.height,  
                    "unit": page.unit,  
                    "angle": page.angle,  
                    "lines_count": len(page.lines),  
                    "words_count": len(page.words) if hasattr(page, 'words') else 0  
                }  
                for page in result.pages  
            ]  
        },  
        
        # 단락 정보  
        "paragraphs_info": {  
            "count": len(result.paragraphs),  
            "roles": {},  # 역할별 카운트  
            "sample": []  
        },  
        
        # 표 정보  
        "tables_info": {  
            "count": len(result.tables),  
            "details": [  
                {  
                    "table_id": idx + 1,  
                    "row_count": table.row_count,  
                    "column_count": table.column_count,  
                    "cell_count": len(table.cells),  
                    "page": table.bounding_regions[0].page_number if table.bounding_regions else None  
                }  
                for idx, table in enumerate(result.tables)  
            ]  
        }  
    }  
    
    # 단락 역할별 통계  
    for para in result.paragraphs:  
        role = para.role or "paragraph"  
        result_dict["paragraphs_info"]["roles"][role] = \  
            result_dict["paragraphs_info"]["roles"].get(role, 0) + 1  
        
        # 샘플 3개  
        if len(result_dict["paragraphs_info"]["sample"]) < 3:  
            result_dict["paragraphs_info"]["sample"].append({  
                "role": role,  
                "content": para.content[:100] + "...",  
                "page": para.bounding_regions[0].page_number if para.bounding_regions else None  
            })  
    
    return result_dict  

# 사용 예시  
# json_structure = analyze_json_structure(result)  
# print(json.dumps(json_structure, indent=2, ensure_ascii=False))

JSON 출력 예시:

{  
  "api_version": "2023-07-31",  
  "model_id": "prebuilt-layout",  
  "content": "전체 문서 텍스트...",  
  "pages_info": {  
    "count": 10,  
    "details": [  
      {"page_number": 1, "width": 8.5, "height": 11, "unit": "inch", "angle": 0, "lines_count": 45}  
    ]  
  },  
  "paragraphs_info": {  
    "count": 87,  
    "roles": {"title": 5, "sectionHeading": 12, "paragraph": 70},  
    "sample": [...]  
  },  
  "tables_info": {  
    "count": 3,  
    "details": [  
      {"table_id": 1, "row_count": 10, "column_count": 4, "page": 3}  
    ]  
  }  
}

9.2 BoundingRegion (좌표 정보) 활용

def extract_bounding_info(paragraph):  
    """단락의 위치 정보 추출"""  
    if not paragraph.bounding_regions:  
        return None  
    
    region = paragraph.bounding_regions[0]  
    
    return {  
        "page_number": region.page_number,  
        "polygon": region.polygon,  # [x1, y1, x2, y2, x3, y3, x4, y4]  
        "bbox": {  
            "x_min": min(region.polygon[::2]),  
            "y_min": min(region.polygon[1::2]),  
            "x_max": max(region.polygon[::2]),  
            "y_max": max(region.polygon[1::2])  
        }  
    }  

# RAG 메타데이터로 활용  
for para in result.paragraphs:  
    location = extract_bounding_info(para)  
    if location:  
        # Vector DB에 저장 시 메타데이터로 추가  
        metadata = {  
            "page": location["page_number"],  
            "bbox": location["bbox"],  
            "role": para.role  
        }

좌표 정보 활용 사례:
1. 문서 원본 하이라이팅: 검색 결과의 정확한 위치 표시
2. 이미지 영역 추출: OCR 텍스트와 원본 이미지 매칭
3. 레이아웃 기반 필터링: 문서 상단/하단 제외 (헤더/푸터 제거)
4. 컬럼 분리: 2단 레이아웃 문서의 좌우 컬럼 구분

9.3 Style 정보 (폰트, 색상)

def extract_styles(result):  
    """텍스트 스타일 정보 추출 (Formula Add-on 활성화 필요)"""  
    if not hasattr(result, 'styles') or not result.styles:  
        return []  
    
    style_info = []  
    for style in result.styles:  
        style_info.append({  
            "is_handwritten": style.is_handwritten,  
            "confidence": style.confidence,  
            "spans": style.spans  # 텍스트 범위  
        })  
    
    return style_info  

# Agent 응용: 손글씨 영역만 별도 처리  
# handwritten_sections = [s for s in extract_styles(result) if s["is_handwritten"]]

10 고급 RAG 응용 패턴

10.1 1. 계층적 문서 인덱싱

문서 구조를 활용한 다단계 검색 시스템이다.

from typing import Dict, List  
from langchain_core.documents import Document  

def create_hierarchical_index(result) -> Dict[str, List[Document]]:  
    """문서를 계층 구조로 인덱싱"""  
    
    hierarchy = {  
        "titles": [],  
        "sections": [],  
        "paragraphs": [],  
        "tables": []  
    }  
    
    current_title = None  
    current_section = None  
    
    for para in result.paragraphs:  
        content = para.content  
        role = para.role or "paragraph"  
        page = para.bounding_regions[0].page_number if para.bounding_regions else 1  
        
        if role == "title":  
            current_title = content  
            hierarchy["titles"].append(Document(  
                page_content=content,  
                metadata={"type": "title", "page": page}  
            ))  
        
        elif role == "sectionHeading":  
            current_section = content  
            hierarchy["sections"].append(Document(  
                page_content=content,  
                metadata={  
                    "type": "section",  
                    "parent_title": current_title,  
                    "page": page  
                }  
            ))  
        
        else:  # paragraph  
            hierarchy["paragraphs"].append(Document(  
                page_content=content,  
                metadata={  
                    "type": "paragraph",  
                    "parent_title": current_title,  
                    "parent_section": current_section,  
                    "page": page  
                }  
            ))  
    
    # 표 데이터 추가  
    for table_idx, table in enumerate(result.tables):  
        # 표를 마크다운 형식으로 변환  
        table_md = table_to_markdown(table)  
        
        hierarchy["tables"].append(Document(  
            page_content=table_md,  
            metadata={  
                "type": "table",  
                "table_id": table_idx + 1,  
                "rows": table.row_count,  
                "columns": table.column_count,  
                "page": table.bounding_regions[0].page_number if table.bounding_regions else 1  
            }  
        ))  
    
    return hierarchy  

def table_to_markdown(table) -> str:  
    """표를 마크다운 형식으로 변환"""  
    # 셀을 2D 배열로 구성  
    grid = [[None] * table.column_count for _ in range(table.row_count)]  
    
    for cell in table.cells:  
        grid[cell.row_index][cell.column_index] = cell.content  
    
    # 마크다운 생성  
    md_lines = []  
    md_lines.append("| " + " | ".join(grid[0]) + " |")  
    md_lines.append("|" + "|".join(["---"] * table.column_count) + "|")  
    
    for row in grid[1:]:  
        md_lines.append("| " + " | ".join(row) + " |")  
    
    return "\n".join(md_lines)  

# 사용 예시  
# hierarchy = create_hierarchical_index(result)  
# print(f"제목: {len(hierarchy['titles'])}개")  
# print(f"섹션: {len(hierarchy['sections'])}개")  
# print(f"단락: {len(hierarchy['paragraphs'])}개")  
# print(f"표: {len(hierarchy['tables'])}개")

RAG 검색 전략:
1. 1단계: 제목/섹션 검색 → 관련 섹션 식별
2. 2단계: 해당 섹션 내 단락 검색 → 정확한 답변 위치
3. 3단계: 관련 표 데이터 조회 → 숫자 정보 보강

10.2 2. 하이브리드 검색 (텍스트 + 표)

from langchain.vectorstores import FAISS  
from langchain_openai import OpenAIEmbeddings  

def create_hybrid_vectorstore(hierarchy: Dict[str, List[Document]]):  
    """텍스트와 표를 별도 벡터스토어로 생성"""  
    
    embeddings = OpenAIEmbeddings()  
    
    # 텍스트 벡터스토어 (단락만)  
    text_vectorstore = FAISS.from_documents(  
        hierarchy["paragraphs"],  
        embeddings  
    )  
    
    # 표 벡터스토어 (표만)  
    table_vectorstore = FAISS.from_documents(  
        hierarchy["tables"],  
        embeddings  
    ) if hierarchy["tables"] else None  
    
    return {  
        "text": text_vectorstore,  
        "table": table_vectorstore  
    }  

def hybrid_search(query: str, vectorstores: Dict, k: int = 5):  
    """텍스트와 표를 동시 검색"""  
    
    results = {  
        "text": vectorstores["text"].similarity_search(query, k=k),  
        "table": vectorstores["table"].similarity_search(query, k=k//2) if vectorstores["table"] else []  
    }  
    
    return results  

# 사용 예시  
# vectorstores = create_hybrid_vectorstore(hierarchy)  
# results = hybrid_search("2023년 매출은?", vectorstores)  
#   
# print("텍스트 결과:", results["text"])  
# print("표 결과:", results["table"])

10.3 3. Agent용 도구 함수

Document Intelligence 결과를 Agent의 도구로 제공한다.

from langchain.agents import tool  

@tool  
def search_document_by_section(section_name: str, result) -> str:  
    """특정 섹션의 내용을 검색하는 도구"""  
    
    matching_paragraphs = []  
    in_target_section = False  
    
    for para in result.paragraphs:  
        role = para.role or "paragraph"  
        
        # 섹션 시작 감지  
        if role == "sectionHeading" and section_name.lower() in para.content.lower():  
            in_target_section = True  
            continue  
        
        # 다음 섹션 시작 시 종료  
        if role == "sectionHeading" and in_target_section:  
            break  
        
        # 대상 섹션의 단락 수집  
        if in_target_section:  
            matching_paragraphs.append(para.content)  
    
    return "\n\n".join(matching_paragraphs) if matching_paragraphs else "해당 섹션을 찾을 수 없습니다."  

@tool  
def extract_tables_from_page(page_number: int, result) -> str:  
    """특정 페이지의 표 데이터를 추출하는 도구"""  
    
    page_tables = []  
    for table in result.tables:  
        if table.bounding_regions and table.bounding_regions[0].page_number == page_number:  
            table_md = table_to_markdown(table)  
            page_tables.append(table_md)  
    
    return "\n\n".join(page_tables) if page_tables else f"페이지 {page_number}에 표가 없습니다."  

@tool  
def get_document_outline(result) -> str:  
    """문서 목차 생성 도구"""  
    
    outline = []  
    for para in result.paragraphs:  
        if para.role in ["title", "sectionHeading"]:  
            page = para.bounding_regions[0].page_number if para.bounding_regions else "?"  
            outline.append(f"[페이지 {page}] {para.content}")  
    
    return "\n".join(outline)  

# Agent에 도구 등록  
# tools = [search_document_by_section, extract_tables_from_page, get_document_outline]

Agent 시나리오 예시:

사용자: "2장의 매출 표를 요약해줘"  

Agent 실행:  
1. get_document_outline() → "2장"의 페이지 번호 확인 (예: 5페이지)  
2. extract_tables_from_page(5) → 해당 페이지 표 추출  
3. LLM으로 표 내용 요약

10.4 4. 메타데이터 기반 필터링

def create_filtered_chunks(result, filter_config: Dict) -> List[Document]:  
    """메타데이터 기반 문서 필터링"""  
    
    documents = []  
    
    for para in result.paragraphs:  
        # 필터 조건 체크  
        role = para.role or "paragraph"  
        page = para.bounding_regions[0].page_number if para.bounding_regions else 1  
        
        # 역할 필터  
        if "allowed_roles" in filter_config:  
            if role not in filter_config["allowed_roles"]:  
                continue  
        
        # 페이지 필터  
        if "page_range" in filter_config:  
            start, end = filter_config["page_range"]  
            if not (start <= page <= end):  
                continue  
        
        # 길이 필터  
        if "min_length" in filter_config:  
            if len(para.content) < filter_config["min_length"]:  
                continue  
        
        # 조건 통과 시 문서 생성  
        doc = Document(  
            page_content=para.content,  
            metadata={  
                "role": role,  
                "page": page,  
                "char_count": len(para.content)  
            }  
        )  
        documents.append(doc)  
    
    return documents  

# 사용 예시  
filter_config = {  
    "allowed_roles": ["paragraph", "sectionHeading"],  # 제목 제외  
    "page_range": (1, 50),  # 1-50페이지만  
    "min_length": 100  # 100자 이상만  
}  

# filtered_docs = create_filtered_chunks(result, filter_config)

10.5 5. 실시간 문서 업데이트 파이프라인

import hashlib  
from datetime import datetime  

class DocumentIntelligenceCache:  
    """문서 분석 결과 캐싱 시스템"""  
    
    def __init__(self, cache_dir: str = ".doc_intel_cache"):  
        self.cache_dir = cache_dir  
        os.makedirs(cache_dir, exist_ok=True)  
    
    def get_file_hash(self, file_path: str) -> str:  
        """파일 해시 생성"""  
        with open(file_path, "rb") as f:  
            return hashlib.sha256(f.read()).hexdigest()  
    
    def is_cached(self, file_path: str) -> bool:  
        """캐시 존재 여부 확인"""  
        file_hash = self.get_file_hash(file_path)  
        cache_file = os.path.join(self.cache_dir, f"{file_hash}.json")  
        return os.path.exists(cache_file)  
    
    def get_cache(self, file_path: str):  
        """캐시된 결과 로드"""  
        file_hash = self.get_file_hash(file_path)  
        cache_file = os.path.join(self.cache_dir, f"{file_hash}.json")  
        
        with open(cache_file, "r", encoding="utf-8") as f:  
            return json.load(f)  
    
    def save_cache(self, file_path: str, result):  
        """분석 결과 캐싱"""  
        file_hash = self.get_file_hash(file_path)  
        cache_file = os.path.join(self.cache_dir, f"{file_hash}.json")  
        
        # 결과를 직렬화 가능한 형태로 변환  
        result_dict = {  
            "timestamp": datetime.now().isoformat(),  
            "content": result.content,  
            "paragraphs": [  
                {"content": p.content, "role": p.role}   
                for p in result.paragraphs  
            ],  
            "tables": [  
                {  
                    "row_count": t.row_count,  
                    "column_count": t.column_count,  
                    "cells": [  
                        {"row": c.row_index, "col": c.column_index, "content": c.content}  
                        for c in t.cells  
                    ]  
                }  
                for t in result.tables  
            ]  
        }  
        
        with open(cache_file, "w", encoding="utf-8") as f:  
            json.dump(result_dict, f, ensure_ascii=False, indent=2)  

# 사용 예시  
# cache = DocumentIntelligenceCache()  
#   
# if cache.is_cached("report.pdf"):  
#     result = cache.get_cache("report.pdf")  
#     print("캐시에서 로딩")  
# else:  
#     # Document Intelligence 실행  
#     result = analyze_document("report.pdf")  
#     cache.save_cache("report.pdf", result)  
#     print("새로 분석 및 캐싱")

11 성능 최적화 및 비용 관리

11.1 배치 처리 전략

import time  
from typing import List  

def batch_analyze_documents(  
    file_paths: List[str],  
    batch_size: int = 5,  
    delay_seconds: float = 0.2  
) -> List[Dict]:  
    """배치 단위 문서 분석 (API 제한 고려)"""  
    
    results = []  
    
    for i in range(0, len(file_paths), batch_size):  
        batch = file_paths[i:i+batch_size]  
        
        print(f"\n배치 {i//batch_size + 1}/{(len(file_paths)-1)//batch_size + 1}")  
        
        for file_path in batch:  
            start_time = time.time()  
            
            try:  
                with open(file_path, "rb") as f:  
                    poller = document_analysis_client.begin_analyze_document(  
                        "prebuilt-layout", document=f  
                    )  
                    result = poller.result()  
                
                results.append({  
                    "file": file_path,  
                    "status": "success",  
                    "page_count": len(result.pages),  
                    "processing_time": time.time() - start_time,  
                    "result": result  
                })  
                
                print(f"✓ {file_path} ({len(result.pages)} 페이지, {time.time()-start_time:.2f}초)")  
                
            except Exception as e:  
                results.append({  
                    "file": file_path,  
                    "status": "error",  
                    "error": str(e)  
                })  
                print(f"✗ {file_path}: {str(e)}")  
            
            # API 제한 방지  
            time.sleep(delay_seconds)  
    
    return results  

# 사용 예시  
# file_list = ["doc1.pdf", "doc2.pdf", "doc3.pdf"]  
# results = batch_analyze_documents(file_list, batch_size=3, delay_seconds=0.5)

API 제한 (S0 계층):
- 초당 요청: 15 TPS
- 분당 요청: 제한 없음
- 동시 요청: 10개

11.2 비용 모니터링

class CostTracker:  
    """Document Intelligence 비용 추적"""  
    
    # 가격 (2025년 기준, USD)  
    PRICES = {  
        "prebuilt-read": 1.50 / 1000,     # $1.50 per 1K pages  
        "prebuilt-layout": 10.0 / 1000,   # $10 per 1K pages  
        "prebuilt-invoice": 10.0 / 1000,  
    }  
    
    def __init__(self):  
        self.usage = {model: 0 for model in self.PRICES.keys()}  
    
    def track(self, model: str, page_count: int):  
        """사용량 기록"""  
        self.usage[model] += page_count  
    
    def get_cost(self) -> Dict:  
        """비용 계산"""  
        costs = {}  
        total = 0  
        
        for model, pages in self.usage.items():  
            cost = pages * self.PRICES[model]  
            costs[model] = {  
                "pages": pages,  
                "cost_usd": round(cost, 4)  
            }  
            total += cost  
        
        costs["total_usd"] = round(total, 4)  
        return costs  
    
    def reset(self):  
        """사용량 초기화"""  
        self.usage = {model: 0 for model in self.PRICES.keys()}  

# 사용 예시  
tracker = CostTracker()  

# 문서 분석 후  
# tracker.track("prebuilt-layout", page_count=50)  
# tracker.track("prebuilt-layout", page_count=120)  
#  
# print("누적 비용:")  
# print(json.dumps(tracker.get_cost(), indent=2))

12 참고 자료

12.1 공식 문서

12.2 RAG 통합

13 다음 단계

Document Intelligence로 구조화된 문서를 얻었다면, 이제 벡터 임베딩을 생성하자:

👉 03-Azure-OpenAI-Embeddings.qmd - 텍스트 임베딩 및 벡터 검색

13.1 Blob에서 직접 분석

from azure.storage.blob import BlobServiceClient  

# Blob Storage 클라이언트  
blob_service_client = BlobServiceClient.from_connection_string(  
    os.getenv("AZURE_STORAGE_CONNECTION_STRING")  
)  

def analyze_blob_document(container_name: str, blob_name: str):  
    """Blob Storage의 문서를 Document Intelligence로 분석"""  
    
    # Blob SAS URL 생성  
    from azure.storage.blob import generate_blob_sas, BlobSasPermissions  
    from datetime import datetime, timedelta  
    
    sas_token = generate_blob_sas(  
        account_name="stragdocs2025",  
        container_name=container_name,  
        blob_name=blob_name,  
        account_key=os.getenv("AZURE_STORAGE_KEY"),  
        permission=BlobSasPermissions(read=True),  
        expiry=datetime.utcnow() + timedelta(hours=1)  
    )  
    
    blob_url = f"https://stragdocs2025.blob.core.windows.net/{container_name}/{blob_name}?{sas_token}"  
    
    # Document Intelligence로 분석  
    poller = document_analysis_client.begin_analyze_document_from_url(  
        "prebuilt-layout", document_url=blob_url  
    )  
    result = poller.result()  
    
    return result  

# 사용 예시  
result = analyze_blob_document("rag-documents", "sample.pdf")  
print("Blob 문서 분석 완료")

13.2 배치 처리

def analyze_all_blobs(container_name: str):  
    """컨테이너의 모든 문서 분석"""  
    container_client = blob_service_client.get_container_client(container_name)  
    blob_list = container_client.list_blobs()  
    
    results = []  
    for blob in blob_list:  
        # PDF 파일만 처리  
        if blob.name.endswith('.pdf'):  
            print(f"분석 중: {blob.name}")  
            try:  
                result = analyze_blob_document(container_name, blob.name)  
                text = extract_full_text(result)  
                
                results.append({  
                    "blob_name": blob.name,  
                    "page_count": len(result.pages),  
                    "text": text,  
                    "status": "success"  
                })  
            except Exception as e:  
                print(f"오류: {blob.name} - {str(e)}")  
                results.append({  
                    "blob_name": blob.name,  
                    "status": "error",  
                    "error": str(e)  
                })  
    
    return results  

# 사용 예시  
# results = analyze_all_blobs("rag-documents")  
# print(f"총 {len(results)}개 문서 처리 완료")

14 한국어 문서 최적화

RAG 시스템에서 한국어 문서 처리 시 주의사항이다.

14.1 언어 힌트 제공

# 한국어 문서 분석 (언어 힌트)  
with open("korean_document.pdf", "rb") as f:  
    poller = document_analysis_client.begin_analyze_document(  
        "prebuilt-layout",  
        document=f,  
        locale="ko-KR"  # 한국어 힌트  
    )  
    result = poller.result()

14.2 OCR 후처리

import re  

def clean_korean_ocr_text(text: str) -> str:  
    """한글 OCR 결과 정리 (RAG 최적화)"""  
    
    # 불필요한 공백 제거  
    text = re.sub(r'\s+', ' ', text)  
    
    # 줄바꿈 정리 (단락 구분 보존)  
    text = re.sub(r'\n\s*\n', '\n\n', text)  
    
    # 특수문자 정리  
    text = text.replace('〃', '"').replace('〃', '"')  
    text = text.replace('·', '•')  # 가운데점 → 불릿  
    
    # 일반적인 OCR 오류 수정  
    text = text.replace('ㅍㅍ', 'ㄷ')  # ㅍㅍ → ㄷ  
    text = text.replace('0', 'O')  # 숫자 0과 영문자 O 혼동  
    
    return text.strip()  

# RAG 파이프라인 통합  
raw_text = extract_full_text(result)  
cleaned_text = clean_korean_ocr_text(raw_text)  

# 컨텍스트로 사용  
from langchain_core.documents import Document  
doc = Document(page_content=cleaned_text, metadata={"language": "ko"})

15 LangChain RAG 통합

15.1 AzureAIDocumentIntelligenceLoader

LangChain의 공식 Document Loader로 Document Intelligence를 통합한다.

from langchain_community.document_loaders import AzureAIDocumentIntelligenceLoader  

# Document Intelligence Loader 생성  
loader = AzureAIDocumentIntelligenceLoader(  
    api_endpoint=endpoint,  
    api_key=key,  
    file_path="sample.pdf",  
    api_model="prebuilt-layout"  
)  

# 문서 로딩 (자동으로 페이지 분할)  
documents = loader.load()  

print(f"로딩된 문서 수: {len(documents)}")  
print(f"첫 번째 청크:\n{documents[0].page_content[:300]}")  
print(f"메타데이터: {documents[0].metadata}")

기본 메타데이터:

{  
    'source': 'sample.pdf',  
    'page': 1,  
}

15.2 RAG 파이프라인 완전 구현

from langchain.text_splitter import RecursiveCharacterTextSplitter  
from langchain_openai import OpenAIEmbeddings  
from langchain.vectorstores import FAISS  

# 1. Document Intelligence로 문서 로딩  
loader = AzureAIDocumentIntelligenceLoader(  
    api_endpoint=endpoint,  
    api_key=key,  
    file_path="long_document.pdf",  
    api_model="prebuilt-layout"  
)  
documents = loader.load()  

# 2. 레이아웃 고려 텍스트 분할  
text_splitter = RecursiveCharacterTextSplitter(  
    chunk_size=1000,  
    chunk_overlap=200,  
    separators=["\n\n", "\n", " ", ""],  # 단락 우선 분할  
    length_function=len  
)  
splits = text_splitter.split_documents(documents)  

# 3. 벡터 임베딩 생성  
embeddings = OpenAIEmbeddings(  
    openai_api_key=os.getenv("OPENAI_API_KEY")  
)  

# 4. Vector Store 생성  
vectorstore = FAISS.from_documents(splits, embeddings)  

print(f"총 {len(splits)}개 청크 생성")  
print(f"Vector Store 초기화 완료")  

# 5. RAG 검색 테스트  
query = "2023년 매출은 얼마인가?"  
results = vectorstore.similarity_search(query, k=3)  

for i, doc in enumerate(results, 1):  
    print(f"\n결과 {i}:")  
    print(f"내용: {doc.page_content[:200]}...")  
    print(f"메타데이터: {doc.metadata}")

16 참고 자료

16.1 공식 문서

16.2 RAG 패턴

16.3 Agent 응용

17 다음 단계

Document Intelligence로 구조화된 문서를 얻었다면, 이제 벡터 임베딩을 생성하자:

👉 03-Azure-OpenAI-Embeddings.qmd - 텍스트 임베딩 및 벡터 검색

import json  
import hashlib  

def get_cache_key(file_path: str) -> str:  
    """파일 해시로 캐시 키 생성"""  
    with open(file_path, "rb") as f:  
        file_hash = hashlib.md5(f.read()).hexdigest()  
    return f"doc_intel_{file_hash}"  

def analyze_with_cache(file_path: str):  
    """캐싱을 사용한 문서 분석"""  
    cache_key = get_cache_key(file_path)  
    cache_file = f".cache/{cache_key}.json"  
    
    # 캐시 확인  
    try:  
        with open(cache_file, "r", encoding="utf-8") as f:  
            cached_result = json.load(f)  
            print("캐시에서 결과 로딩")  
            return cached_result  
    except FileNotFoundError:  
        pass  
    
    # Document Intelligence 실행  
    with open(file_path, "rb") as f:  
        poller = document_analysis_client.begin_analyze_document(  
            "prebuilt-layout", document=f  
        )  
        result = poller.result()  
    
    # 결과를 딕셔너리로 변환  
    result_dict = {  
        "content": result.content,  
        "pages": [{"page_number": p.page_number, "width": p.width, "height": p.height} for p in result.pages],  
        "paragraphs": [{"content": p.content, "role": p.role} for p in result.paragraphs]  
    }  
    
    # 캐시 저장  
    os.makedirs(".cache", exist_ok=True)  
    with open(cache_file, "w", encoding="utf-8") as f:  
        json.dump(result_dict, f, ensure_ascii=False, indent=2)  
    
    print("Document Intelligence 실행 및 캐시 저장")  
    return result_dict  

# 사용 예시  
# cached_result = analyze_with_cache("sample.pdf")

17.1 배치 크기 조정

def analyze_documents_batch(file_paths: List[str], batch_size: int = 5):  
    """배치 단위로 문서 분석 (API 제한 고려)"""  
    import time  
    
    results = []  
    for i in range(0, len(file_paths), batch_size):  
        batch = file_paths[i:i+batch_size]  
        
        print(f"배치 {i//batch_size + 1} 처리 중 ({len(batch)}개 파일)")  
        for file_path in batch:  
            with open(file_path, "rb") as f:  
                poller = document_analysis_client.begin_analyze_document(  
                    "prebuilt-layout", document=f  
                )  
                result = poller.result()  
                results.append({  
                    "file": file_path,  
                    "result": result  
                })  
        
        # API 제한 방지 (초당 15 요청)  
        time.sleep(1)  
    
    return results  

# 사용 예시  
# file_list = ["doc1.pdf", "doc2.pdf", "doc3.pdf"]  
# results = analyze_documents_batch(file_list, batch_size=5)

18 모니터링

18.1 분석 통계 추적

def analyze_with_metrics(file_path: str):  
    """분석 메트릭 추적"""  
    import time  
    
    start_time = time.time()  
    
    with open(file_path, "rb") as f:  
        file_size = os.path.getsize(file_path)  
        
        poller = document_analysis_client.begin_analyze_document(  
            "prebuilt-layout", document=f  
        )  
        result = poller.result()  
    
    end_time = time.time()  
    duration = end_time - start_time  
    
    metrics = {  
        "file_path": file_path,  
        "file_size_mb": file_size / (1024 * 1024),  
        "page_count": len(result.pages),  
        "duration_seconds": duration,  
        "pages_per_second": len(result.pages) / duration  
    }  
    
    print(f"분석 완료:")  
    print(f"- 파일 크기: {metrics['file_size_mb']:.2f} MB")  
    print(f"- 페이지 수: {metrics['page_count']}")  
    print(f"- 처리 시간: {metrics['duration_seconds']:.2f}초")  
    print(f"- 속도: {metrics['pages_per_second']:.2f} 페이지/초")  
    
    return result, metrics  

# 사용 예시  
# result, metrics = analyze_with_metrics("large_document.pdf")

19 문제 해결

19.1 일반적인 오류

InvalidRequest: 파일 크기 초과

오류: 파일이 너무 큼 (최대 500MB)  
해결: 파일을 분할하거나 압축

InvalidImage: 이미지 해상도 부족

# 해결: 이미지 품질 확인 (최소 150 DPI 권장)  
from PIL import Image  

img = Image.open("low_quality.png")  
print(f"해상도: {img.size}")  
print(f"DPI: {img.info.get('dpi', 'Unknown')}")

Unauthorized: 인증 실패

# 키 및 엔드포인트 확인  
print(f"Endpoint: {endpoint}")  
print(f"Key: {key[:10]}... (길이: {len(key)})")

20 참고 자료

20.1 공식 문서

20.2 샘플 코드

21 다음 단계

문서 OCR 및 레이아웃 분석이 완료되었다면, 이제 텍스트를 임베딩으로 변환하자:

👉 03-Azure-OpenAI-Embeddings.qmd - Azure OpenAI로 문서 임베딩 생성