LangGraph 구조에서 사용된 Python 문법 정리

앞선 포스팅은 LangGraph에서 주로 쓰이는 Python 문법에 대해 정리하였고, 이번 포스팅은 실제로 LangGraph로 전처리 파싱 로직을 구현하는 과정에서 사용한 문법을 정리해보았다! 아직 로직은 계속 리팩토링하고 검증 로직을 추가하고 있어서 해당 포스팅을 계속 업데이트해볼 예정이다 😊

타입 힌팅 (Type Hints)

TypedDict 활용

from typing import TypedDict, List, Dict, Any, Optional

class DocumentProcessingState(TypedDict):
    """문서 처리 상태"""
    file_path: str
    file_type: Optional[str]
    thread_id: str
    current_step: str
    sections: List[Dict[str, Any]]
    error: Optional[str]
    start_time: float
    file_size_mb: float
    s3_image_config: Optional[Dict[str, Any]]

함수 타입 힌팅

def create_initial_state(file_path: str, 
                        thread_id: Optional[str] = None,
                        s3_image_config: Optional[Dict[str, Any]] = None) -> DocumentProcessingState:
    """초기 상태 생성"""
    pass

# Callable 타입
from typing import Callable
NodeFunction = Callable[[DocumentProcessingState], Dict[str, Any]]

제네릭 타입

from typing import Iterator, Union
def process_document_stream() -> Iterator[Dict[str, Any]]:
    """스트리밍 처리"""
    pass

클래스와 객체지향 프로그래밍

싱글톤 패턴 구현

class Settings:
    def __init__(self):
        self.OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
        self.UPSTAGE_API_KEY = os.getenv("UPSTAGE_API_KEY")
        
    def _create_directories(self):
        """필요한 디렉토리 생성"""
        directories = [self.BASE_OUTPUT_DIR, self.IMAGES_DIR, self.CACHE_DIR]
        for directory in directories:
            directory.mkdir(parents=True, exist_ok=True)

# 전역 인스턴스
settings = Settings()

Property 데코레이터

@property
def SUPPORTED_FILE_TYPES(self) -> Dict[str, list]:
    """지원하는 파일 타입들"""
    return {
        'pdf': ['.pdf'],
        'powerpoint': ['.ppt', '.pptx'],
        'word': ['.doc', '.docx']
    }

함수형 프로그래밍 패턴

데코레이터 패턴

def create_node(node_name: str, node_function: NodeFunction) -> NodeFunction:
    def wrapper(state: DocumentProcessingState) -> Dict[str, Any]:
        start_time = time.time()
        try:
            print(f"[{node_name}] 시작")
            result = node_function(updated_state)
            elapsed = time.time() - start_time
            print(f"[{node_name}] 완료 ({elapsed:.2f}초)")
            return result
        except Exception as e:
            # 에러 처리
            pass
    return wrapper

고차함수와 람다

# 정렬 키 함수
results.sort(key=lambda x: x["page_number"])

# 필터링
supported_files = [
    str(f) for f in all_files 
    if f.is_file() and settings.is_supported_file(str(f))
]

# 리스트 컴프리헨션과 조건부 표현식
sections_with_images = sum(1 for section in sections if section.get("image_url"))

컨텍스트 매니저

with문 활용

# 파일 처리
with open(file_path, 'r', encoding='utf-8') as f:
    content = f.read()

# ThreadPoolExecutor 사용
with ThreadPoolExecutor(max_workers=max_workers) as executor:
    future_to_info = {
        executor.submit(_extract_single_image, image_path, i + 1): (image_path, i + 1)
        for i, image_path in enumerate(image_paths)
    }

# 임시 파일
import tempfile
with tempfile.NamedTemporaryFile(suffix='.pdf', delete=False) as temp_pdf:
    temp_pdf_path = temp_pdf.name

예외 처리

다중 예외 처리

def _read_text_file(file_path: str) -> str:
    encodings = ['utf-8', 'cp949', 'euc-kr', 'latin-1']
    
    for encoding in encodings:
        try:
            with open(file_path, 'r', encoding=encoding) as f:
                content = f.read()
            return content
        except (UnicodeDecodeError, UnicodeError):
            continue
    
    # 최종 백업 처리
    try:
        with open(file_path, 'rb') as f:
            raw_content = f.read()
        return raw_content.decode('utf-8', errors='replace')
    except Exception as e:
        raise ValueError(f"파일을 읽을 수 없습니다: {e}")

커스텀 예외와 에러 전파

try:
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"파일을 찾을 수 없습니다: {file_path}")
    
    file_type = detect_file_type(file_path)
    
except Exception as e:
    error_msg = f"파일 분석 실패: {str(e)}"
    print(f"❌ {error_msg}")
    
    from core.state import add_error
    error_state = add_error(state, error_msg)
    return error_state

동시성 프로그래밍

ThreadPoolExecutor 활용

from concurrent.futures import ThreadPoolExecutor, as_completed

def _extract_with_vision_parallel(image_paths: List[str]) -> List[Dict[str, Any]]:
    results = []
    max_workers = 2
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_info = {
            executor.submit(_extract_single_image, image_path, i + 1): (image_path, i + 1)
            for i, image_path in enumerate(image_paths)
        }
        
        for future in as_completed(future_to_info):
            image_path, page_num = future_to_info[future]
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                # 에러 처리
                pass
    
    return results

제너레이터와 이터레이터

제너레이터 함수

def process_document_stream(file_path: str) -> Iterator[Dict[str, Any]]:
    """스트리밍 처리"""
    try:
        compiled_graph = compile_graph(debug=debug)
        initial_state = create_initial_state(file_path, thread_id, s3_image_config)
        
        for step_result in compiled_graph.stream(initial_state, runtime_config):
            if step_result:
                step_name = list(step_result.keys())[0]
                step_state = list(step_result.values())[0]
                
                # 단계별 결과 yield
                yield {
                    "step": step_name,
                    "current_step": step_state.get("current_step", step_name),
                    "sections_count": len(step_state.get("sections", [])),
                    "elapsed_time": time.time() - start_time
                }
    except Exception as e:
        yield {"error": str(e)}

메타프로그래밍과 동적 코드

동적 임포트

def create_llm(self):
    """동적으로 LLM 클래스 임포트"""
    if not self.OPENAI_API_KEY:
        raise ValueError("OPENAI_API_KEY가 설정되지 않았습니다")
    
    from langchain_openai import ChatOpenAI  # 필요할 때만 임포트
    
    return ChatOpenAI(
        model=self.LLM_MODEL,
        temperature=self.LLM_TEMPERATURE,
        api_key=self.OPENAI_API_KEY
    )

동적 속성 접근

def _get_default_prompt(self, prompt_name: str) -> str:
    """동적으로 기본 프롬프트 반환"""
    defaults = {
        "vision_extraction": "이미지를 분석하고...",
        "metadata_generation": "다음 내용을 분석하여...",
    }
    return defaults.get(prompt_name, "기본 프롬프트가 없습니다.")

함수 캐싱과 성능 최적화

조건부 실행과 최적화

def route_by_file_type(state: DocumentProcessingState) -> str:
    """조건부 라우팅으로 불필요한 처리 방지"""
    if state.get("error"):
        return "error"
    
    file_type = state.get("file_type")
    if not file_type:
        return "error"
    
    supported_types = ["pdf", "ppt", "docx", "markdown", "txt", "xlsx"]
    if file_type not in supported_types:
        return "error"
    
    return file_type

패턴 매칭과 정규표현식

정규표현식 활용

import re

def _split_by_headings(content: str, file_path: str) -> List[Dict[str, Any]]:
    """정규표현식으로 헤딩 찾기"""
    heading_pattern = re.compile(r'^(#{1,6})\s+(.+)$', re.MULTILINE)
    headings = list(heading_pattern.finditer(content))
    
    for i, heading_match in enumerate(headings):
        heading_level = len(heading_match.group(1))
        original_title = heading_match.group(2).strip()

def _clean_text(content: str) -> str:
    """정규표현식으로 텍스트 정리"""
    content = re.sub(r'\n{3,}', '\n\n', content)  # 3개 이상 줄바꿈 → 2개
    content = re.sub(r'[ \t]+', ' ', content)     # 다중 공백 → 단일 공백
    content = re.sub(r' +\n', '\n', content)      # 줄 끝 공백 제거
    return content.strip()

파일 시스템과 경로 처리

pathlib 활용

from pathlib import Path

def _generate_output_path(file_path: str, file_type: str) -> Path:
    """Path 객체로 안전한 경로 처리"""
    output_dir = Path("./data/output")
    output_dir.mkdir(parents=True, exist_ok=True)
    
    if file_path:
        file_stem = Path(file_path).stem
    else:
        file_stem = "document"
    
    timestamp = int(time.time())
    filename = f"{file_stem}_{file_type}_result_{timestamp}.json"
    
    return output_dir / filename

딕셔너리

딕셔너리 병합과 업데이트

def create_final_section(section: Dict, metadata: Dict, file_type: str) -> Dict[str, Any]:
    """딕셔너리 구조 조작"""
    final_section = {
        "slide_id": section_id,
        "category": category,
        "title": title,
        "content": {
            "text": metadata.get("refined_text", ""),
            "hierarchical_context": hierarchical_context,
            "ui_elements": ui_elements,
            "visual_description": metadata.get("visual_description", ""),
            "keywords": metadata.get("keywords", [])
        },
        "faq": metadata.get("faq", [])
    }
    
    return final_section

# 딕셔너리 접근
section_id = (section.get("section_id") or 
              section.get("id") or 
              f"section_{i+1}")

JSON 처리와 파싱

JSON 파싱

def _parse_vision_response(content: str) -> Dict[str, Any]:
    """JSON 응답 안전 파싱"""
    # 마크다운 코드 블록 제거
    if content.startswith('```json'):
        content = content[7:]
    elif content.startswith('```'):
        content = content[3:]
    
    if content.endswith('```'):
        content = content[:-3]
    
    content = content.strip()
    
    try:
        if content.startswith('{'):
            return json.loads(content)
        else:
            # JSON 부분 추출
            start_idx = content.find('{')
            end_idx = content.rfind('}') + 1
            
            if start_idx != -1 and end_idx > start_idx:
                json_part = content[start_idx:end_idx]
                return json.loads(json_part)
    except json.JSONDecodeError:
        pass
    
    # 파싱 실패 시 기본값
    return {"title": "파싱 실패", "main_text": ""}

명령행 인터페이스

argparse 활용

import argparse

def parse_arguments():
    parser = argparse.ArgumentParser(
        description="LangGraph기반 문서 처리 시스템",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
              python scripts/process_single_file.py document.pdf
              python scripts/process_single_file.py document.pdf --stream
              """
    )
    
    parser.add_argument("file_path", help="처리할 문서 파일 경로")
    parser.add_argument("--stream", action="store_true", help="스트리밍 모드")
    parser.add_argument("--debug", action="store_true", help="디버그 모드")
    
    return parser.parse_args()