멀티모달 프롬프트 설계와 파싱 품질 진단 — 파이프라인을 닫는다

def _add_base64_src_to_html(self, html, base64_encoding):
    """HTML img 태그에 base64 src를 주입"""
    if not base64_encoding:
        return html

    pattern = r"<img([^>]*)>"
    replacement = f'<img\\1 src="data:image/png;base64,{base64_encoding}">'
    return re.sub(pattern, replacement, html)

for elem in state["elements_from_parser"]:
    if elem["category"] == "table":
        soup = BeautifulSoup(elem["content"]["html"], "html.parser")

        # 불규칙한 문자 정리
        for td in soup.find_all("td"):
            td.string = td.get_text(strip=True).replace("\\t", " ").replace("\t", " ")

        cleaned_html_io = StringIO(str(soup))

        try:
            parsed_tables = pd.read_html(cleaned_html_io)
            for table in parsed_tables:
                csv_filename = (
                    f"{base_without_ext.upper()}_TABLE_"
                    f"Page_{elem['page']}_Index_{elem['id']}.csv"
                )
                csv_filepath = os.path.join(table_dir, csv_filename)
                absolute_path = os.path.abspath(csv_filepath)

                table.to_csv(absolute_path, index=False, encoding="utf-8-sig")
                elem["csv_filepath"] = absolute_path
        except Exception as e:
            self.log(f"테이블 파싱 중 오류 발생: {str(e)}")
            continue

def create_document_parse_graph():
    # 1단계 그래프: 파싱 루프
    split_pdf_node = SplitPDFFilesNode(batch_size=30, test_page=None, verbose=True)
    document_parse_node = DocumentParseNode(
        api_key=os.environ["UPSTAGE_API_KEY"], verbose=True
    )
    post_document_parse_node = PostDocumentParseNode(verbose=True)
    working_queue_node = WorkingQueueNode(verbose=True)

    workflow = StateGraph(ParseState)
    workflow.add_node("split_pdf_node", split_pdf_node)
    workflow.add_node("document_parse_node", document_parse_node)
    workflow.add_node("post_document_parse_node", post_document_parse_node)
    workflow.add_node("working_queue_node", working_queue_node)

    workflow.add_edge("split_pdf_node", "working_queue_node")
    workflow.add_conditional_edges(
        "working_queue_node",
        continue_parse,
        {True: "document_parse_node", False: "post_document_parse_node"},
    )
    workflow.add_edge("document_parse_node", "working_queue_node")
    workflow.set_entry_point("split_pdf_node")
    parser_graph = workflow.compile()

    # 2단계 그래프: 후처리 — 1단계 그래프를 하나의 노드로 삽입한다
    post_process_workflow = StateGraph(ParseState)
    post_process_workflow.add_node("document_parse", parser_graph)   # 서브그래프
    post_process_workflow.add_node(
        "create_elements_node", CreateElementsNode(verbose=True)
    )
    # ... 나머지 노드 추가 ...
    langchain_document_node = LangChainDocumentNode(
        verbose=True,
        splitter=RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=0),
    )

    post_process_workflow.set_entry_point("document_parse")

    memory = MemorySaver()
    return post_process_workflow.compile(checkpointer=memory)

import uuid
from langchain_core.runnables import RunnableConfig

# batch_size: 한 번에 처리할 페이지 수
# test_page: 테스트할 페이지 번호 (None이면 전체)
parser_graph = create_upstage_parser_graph(
    batch_size=30, test_page=None, verbose=True
)

config = RunnableConfig(
    recursion_limit=300,
    configurable={"thread_id": str(uuid.uuid4())},
)

inputs = {"filepath": "data/report.pdf"}
parser_graph.invoke(inputs, config=config)

# 추출된 요소 확인
elements = parser_graph.get_state(config).values["elements_from_parser"]

inputs = ParseState(
    filepath="data/report.pdf",
    language="English",   # VLM 해설 출력 언어
)

def save_documents_to_pkl(documents, filepath):
    abs_path = os.path.abspath(filepath)
    base_path = os.path.splitext(abs_path)[0]
    with open(f"{base_path}.pkl", "wb") as f:
        pickle.dump(documents, f)


def load_documents_from_pkl(filepath):
    abs_path = os.path.abspath(filepath)
    base_path = os.path.splitext(abs_path)[0]
    with open(f"{base_path}.pkl", "rb") as f:
        return pickle.load(f)

_type: "prompt"
template: |
  Extract key information and insights from an image based on the provided context.

  Given the context related to the image, analyze and interpret the image to generate a structured output that includes a title, key details, entities, and hypothetical questions.

  # Steps

  1. **Analyze the Context**: Understand the context provided in relation to the image.
  2. **Title Generation**: Create a concise and descriptive title.
  3. **Details Extraction**: Identify and articulate key insights and details visible in the image.
  4. **Entity Identification**: Recognize and list significant entities or objects.
  5. **Hypothetical Questions**: Formulate relevant hypothetical questions that arise from the content.

_type: "prompt"
template: |
  # Output Format

  - The output should be structured using the following tags:
    - `<image>`: Wrap the entire output.
    - `<title>`: Enclose the generated title.
    - `<details>`: Include detailed insights extracted from the image.
    - `<entities>`: List the identified entities.
    - `<hypothetical_questions>`: Present the formulated hypothetical questions.
  - The output must be written in {language}.

  # Example

  **Input**:
  Here is the context related to the image:
  {context}

  **Output**:
  <image>
  <title>
  The Rise of Artificial Intelligence in Modern Technology
  </title>
  <details>
  The image depicts the integration of AI in various technological devices,
  highlighting advancements in automation and data processing.
  </details>
  <entities>
  AI algorithms, robotics, smart devices
  </entities>
  <hypothetical_questions>
  - How will AI continue to evolve in the next decade?
  - What are the ethical implications of AI in everyday life?
  </hypothetical_questions>
  </image>

  # Notes

  - Use the provided context to inform and enhance the extraction process.
  - Ensure that the hypothetical questions are thought-provoking and relevant.
  - Be sure to include numerical values, proper nouns, terms, and terminologies.
input_variables: ["language", "context"]