병렬 처리 · 상태 전달 · 멀티에이전트 — 모듈화가 조직도가 되는 지점

[ExportMarkdown] 마크다운 파일이 성공적으로 생성되었습니다: .../report.md
[ExportTableCSV] CSV 파일이 성공적으로 생성되었습니다: .../REPORT_TABLE_Page_0_Index_10.csv
[ExportHTML] HTML 파일이 성공적으로 생성되었습니다: data/report.html
[ExportTableCSV] CSV 파일이 성공적으로 생성되었습니다: .../REPORT_TABLE_Page_0_Index_14.csv

import uuid
from typing import TypedDict

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.runnables import RunnableConfig


# 1) 공용 State — 파이프라인 전 구간이 이 하나를 공유한다
class ParseState(TypedDict, total=False):
    filepath: str
    elements_from_parser: list
    export: list


# 2) 서브그래프 A — 파싱 (팩토리 함수로 감싼다)
def create_parser_graph(batch_size=30, verbose=True):
    wf = StateGraph(ParseState)
    wf.add_node("split_pdf", SplitPDFNode(batch_size=batch_size, verbose=verbose))
    wf.add_node("document_parse", DocumentParseNode(verbose=verbose))
    wf.add_node("post_process", PostDocumentParseNode(verbose=verbose))

    wf.add_edge("split_pdf", "document_parse")
    wf.add_edge("document_parse", "post_process")
    wf.add_edge("post_process", END)
    wf.set_entry_point("split_pdf")

    # 단독 실행 시 get_state()를 쓰려면 자체 체크포인터가 필요하다
    return wf.compile(checkpointer=MemorySaver())


# 3) 서브그래프 B — 내보내기 (Fan-out / Fan-in)
#    export_image -> {export_html, export_markdown, export_table_to_csv} -> END
export_graph = create_export_graph(show_image_in_markdown=True)


# 4) 삽입 모듈이 지켜야 할 다섯 규칙
#    (a) category 화이트리스트로 처리 대상 축소
#    (b) BATCH_SIZE 단위 chain.batch() 호출
#    (c) 배치 단위 재시도 (trial = 3)
#    (d) 원본 요소를 제자리 갱신
#    (e) 부분 딕셔너리만 반환


# 5) 부모 그래프 — 서브그래프 2개 + 삽입 노드 1개
parser_graph = create_parser_graph(batch_size=30, verbose=True)

parent_workflow = StateGraph(ParseState)
parent_workflow.add_node("upstage_parser", parser_graph)             # 서브그래프
parent_workflow.add_node("add_translation", add_translation_module)  # 삽입 노드
# parent_workflow.add_node("contextualize_text", contextualize_text)  # 대안 (비활성)
parent_workflow.add_node("export_output", export_graph)              # 서브그래프

parent_workflow.add_edge("upstage_parser", "add_translation")
# parent_workflow.add_edge("upstage_parser", "export_output")  # 개조 전 배선 (보존)
parent_workflow.add_edge("add_translation", "export_output")

parent_workflow.set_entry_point("upstage_parser")
parent_graph = parent_workflow.compile(checkpointer=MemorySaver())


# 6) 실행
config = RunnableConfig(
    recursion_limit=300,
    configurable={"thread_id": str(uuid.uuid4())},
)
parent_graph.invoke({"filepath": "data/report.pdf"}, config=config)

# 실행 후 상태 조회
snapshot = parent_graph.get_state(config).values
snapshot["elements_from_parser"][:10]