LangGraph를 활용한 데이터 전처리/요약

AI/LLM

LangGraph를 활용한 데이터 전처리/요약

adulty22 2025. 4. 4. 14:49

LangGraph를 활용하여 데이터를 전처리 및 요약하는 시스템을 구현해보고자한다.

계획한 LangGraph 모습은 다음과 같다.

1단계: 파일명을 보고 분류하는 노드
2-1단계: 이미지 파일 처리 (OCR, 한글 인식) 노드
2-2단계: PDF 파일 처리 노드
2-3단계: CSV 파일 처리 노드
3단계: 요약하는 노드

데이터

서울데이터허브

https://data.seoul.go.kr/bsp/wgs/index.do?tab=chatbot

해당 사이트를 활용하였다.

사용 데이터
2023년 인구성장률 현황.csv

Category
인구성장률
종로구
-1.15
중구
0.77
용산구
-2.65
성동구
-1.2
광진구
-0.02
동대문구
1.77
중랑구
-0.68
성북구
-0.86
강북구
-1.59
도봉구
-1.43
노원구
-1
은평구
0.06
서대문구
0.34
마포구
-0.11
양천구
-1.07
강서구
-1.01
구로구
-0.66
금천구
-0.71
영등포구
-0.07
동작구
-0.18
관악구
-0.67
서초구
0.89
강남구
3.03
송파구
-0.68
강동구
-0.15

2023년 인구성장률 현황.pdf
위 데이터 pdf로 변환

2023년 혼인건수 현황

구현

상태 타입 선언

class FileState(TypedDict):
    file_path: str
    file_type: Optional[str]  # "image", "pdf", "csv"
    processed_text: Optional[str]
    summary: Optional[str]

1단계: 파일명으로 분류

def classify_file(state: FileState) -> FileState:
    file_path = state["file_path"]
    ext = os.path.splitext(file_path)[1].lower()
    if ext in [".jpg", ".jpeg", ".png", ".bmp", ".tiff"]:
        state["file_type"] = "image"
    elif ext == ".pdf":
        state["file_type"] = "pdf"
    elif ext == ".csv":
        state["file_type"] = "csv"
    return state

2단계: 파일명 별 전처리

1. csv 처리 (pandas 이용)

-> 사용전 패키지 다운 필수

pip install pandas

def process_csv(state: FileState) -> FileState:
    file_path = state["file_path"]
    df = pd.read_csv(file_path, encoding="utf-8")
    state["processed_text"] = df
    return state

결과

2023년 인구성장률 현황.csv 결과
Gemma2 요약 결과:
## 서울 특별시 구별 인구 성장률 요약
**2021년 기준** 서울 특별시 25개 구의 인구 성장률을 분석한 결과, **강남구**의 성장률이 가장 높았습니다 (**3.03%**). 반대로 **용산구**의 인구 감소율이 가장 컸습니다 (**-2.65%**).
기타 주요 결과:
* **성장률이 1% 이상인 구:**  중구(0.77%), 동대문구(1.77%), 서초구(0.89%), 강남구(3.03%)
* **감소률이 1% 이상인 구:** 종로구(-1.15%), 용산구(-2.65%), 강북구(-1.59%), 도봉구(-1.43%), 양천구(-1.07%), 강서구(-1.01%), 노원구(-1.00%)
* **0~1% 사이의 성장률이나 감소률을 보이는 구:** 중랑구(-0.68%), 성북구(-0.86%), 금천구(-0.71%), 동작구(-0.18%), 관악구(-0.67%), 은평구(0.06%), 서대문구(0.34%), 광진구(-0.02%), 영등포구(-0.07%), 송파구(-0.68%), 강동구(-0.15%), 마포구(-0.11)
 **추가 분석**:
* 위 데이터를 해당 구의 특징, 주요 산업, 사회경제적 측면과 관련하여 분석하기
* 인구 변동이 심각한 구의 지역적 특성 및 예상되는 영향 파악
* 시간의 흐름에 따른 인구 변화 추이 및 미래 예측

2. Text 추출 (tika 이용)

-> 사용전 패키지 다운 필수

pip install tika

def process_pdf(state: FileState) -> FileState:
    file_path = state["file_path"]
    parsed = parser.from_file(file_path)
    text = parsed.get("content", "")
    state["processed_text"] = text.strip() if text else ""
    return state

결과

2023년 인구성장률 현황.pdf 결과
Gemma2 요약 결과:
## 서울 각구 인구 성장률 요약 (2023년 기준, 순서대로)
**역순위**:
1. 강남구: 3.03%
2. 서초구: 0.89%
3. 중구: 0.77%
4. 영등포구: -0.07%
5. 은평구: 0.06%
6. 서대문구: 0.34%
7. 마포구: -0.11%
8. 동작구: -0.18%
9. 강동구: -0.15%
10. 강북구: -1.59%
11. 노원구: -1.00%
12. 용산구: -2.65%
13. 종로구: -1.15%
14. 성동구: -1.20%
15. 도봉구: -1.43%
16. 양천구: -1.07%
17. 중랑구: -0.68%
18. 성북구: -0.86%
19. 구로구: -0.66%
20. 금천구: -0.71%
21. 송파구: -0.68%
**결론**: 강남구가 가장 높은 인구성장률을 기록했으며, 종로구, 용산구 등 일부 구군서는 인구 감소 추세를 보인다.

3. 이미지 처리 (Paddle OCR 이용)

-> 사용전 패키지 다운 필수

pip install paddleocr
pip install paddlepaddle -f https://www.paddlepaddle.org.cn/whl/mkl/avx/stable.html

PaddleOCR 결과값
[[[2318.0, 1545.0], [2375.0, 1545.0], [2375.0, 1572.0], [2318.0, 1572.0]], ('2800', 0.9999358654022217)]
[[꼭지점 좌표], (OCR 결과, 정확도)]

def process_image(state: FileState) -> FileState:
    file_path = state["file_path"]
    from paddleocr import PaddleOCR


    # PaddleOCR 초기화 (한국어 지원, 기울기 보정 포함)
    ocr = PaddleOCR(use_angle_cls=True, lang="korean")
    result = ocr.ocr(file_path, cls=True)


    # 결과가 없는 경우 처리
    if not result or not result[0]:
        print("OCR 결과가 없습니다.")
        state["processed_text"] = ""
        return state


    # OCR 텍스트 추출
    text_parts = []
    for line in result[0]:
        try:
            rec_text = line[1][0]
            text_parts.append(rec_text)
        except Exception as e:
            print(f"Skipping invalid OCR line: {line} - {e}")


    # OCR 텍스트를 하나의 문자열로 저장
    state["processed_text"] = " ".join(text_parts)


    # OCR 결과 출력
    print("Extracted text:")
    print(state["processed_text"])


    return state

결과

2023년 혼인건수 현황
Gemma2 요약 결과:
지역별 건수 데이터를 보여주는 그래프를 작성하자면, 이 데이터를 요약하고 정리한 정보는 다음과 같습니다.
**서울특별시 각 동 데이터 요약**

* **가장 높은 건수:** 강서구 (2558건), 송파구 (2513건)
* **가장 낮은 건수:** 종로구 (440건)
* **건수 분포:** 용산구, 노원구, 중랑구, 강남구, 관악구와 이웃 동서구의 건수가 1600건 이상으로 높고, 다른 구의 건수는 1000~1600건 사이로 나타났습니다.
**추가 정보**
* 위 데이터가 무엇을 나타내는지 (예: 사건 수, 인구 수, 매출) 를 알려 주시면 더욱 자세하고 의미 있는 분석을 제공할 수 있습니다.
## 데이터 시각화
숫자만 보다는 그래프로 표현하면 시각적으로 더욱 쉽게 이해할 수 있습니다. 막대 그래프나 원형 그래프를 활용하여 각 구별 건수를 비교 보일 수 있습니다.

Discussion

1. csv나 이미지 파일에 경우 파일 양이 너무 크게 되면, LLM 입력시, 토큰수가 초과되는 상황이 발생하였다. 해당 부분에 대해 LLM에 전체 데이터를 넣지않고 토큰수를 계산하여 분리해서 넣을지, 혹은 데이터 형태를 간단하게 알려줘서 해당 방향으로 LLM이 데이터를 전처리할지 고려해야할 것으로 보인다.
2. OCR의 경우 pytesseract을 먼저 사용하였는데 성능이 좋지 않아 PaddleOCR로 변경을 진행하였다. 이 부분에서 PaddleOCR 또한 100% 정확도가 나오지 않아 정확도를 다소 높힐 필요성이 있어보인다. 해당 부분을 이미지를 바로 처리할 수 있는 LLM 모델이나 혹은 정확도를 높힐 모델 혹은 파라미터 튜닝, 이미지 전처리 등을 고려할 필요가 있어보인다.

Appendix

최종 코드 (개인적인 test 포함)

import os


# import base64
from typing import Optional, TypedDict
from langgraph.graph import StateGraph, START, END
from dotenv import load_dotenv
import pandas as pd


# import pytesseract
from tika import parser
from groq import Groq


# 환경변수 로드
# pytesseract.pytesseract.tesseract_cmd = r"C:\Program Files\Tesseract-OCR\tesseract.exe"
load_dotenv()
groq_client = Groq(api_key=os.environ.get("GROQ_API_KEY"))




# 상태 타입 정의: 파일 경로, 파일 분류, 처리된 텍스트, 요약 결과
class FileState(TypedDict):
    file_path: str
    file_type: Optional[str]  # "image", "pdf", "csv"
    processed_text: Optional[str]
    summary: Optional[str]




# 1단계: 파일명을 보고 분류하는 노드
def classify_file(state: FileState) -> FileState:
    file_path = state["file_path"]
    ext = os.path.splitext(file_path)[1].lower()
    if ext in [".jpg", ".jpeg", ".png", ".bmp", ".tiff"]:
        state["file_type"] = "image"
    elif ext == ".pdf":
        state["file_type"] = "pdf"
    elif ext == ".csv":
        state["file_type"] = "csv"
    return state




# 2-1: 이미지 파일 처리 (OCR, 한글 인식)
# def process_image(state: FileState) -> FileState:
#     file_path = state["file_path"]
#     image = Image.open(file_path)
#     text = pytesseract.image_to_string(image, lang="kor")
#     state["processed_text"] = text
#     return state




# llama-3.2-90b-vision-preview 모델 사용
# def process_image(state: FileState) -> FileState:
#     file_path = state["file_path"]
#     # 이미지 파일을 바이너리로 읽고 base64로 인코딩
#     with open(file_path, "rb") as f:
#         image_data = f.read()
#     image_base64 = base64.b64encode(image_data).decode("utf-8")


#     # 이미지 분석 프롬프트
#     prompt = f"다음 이미지를 분석하여 텍스트를 추출해줘:\n{image_base64}"


#     # llama-3.2-90b-vision-preview 모델 호출
#     chat_completion = groq_client.chat.completions.create(
#         messages=[{"role": "user", "content": prompt}],
#         model="llama-3.2-90b-vision-preview",
#     )
#     extracted_text = chat_completion.choices[0].message.content
#     state["processed_text"] = extracted_text
#     return state




def process_image(state: FileState) -> FileState:
    file_path = state["file_path"]
    from paddleocr import PaddleOCR


    # PaddleOCR 초기화 (한국어 지원, 기울기 보정 포함)
    ocr = PaddleOCR(use_angle_cls=True, lang="korean")
    result = ocr.ocr(file_path, cls=True)


    # 결과가 없는 경우 처리
    if not result or not result[0]:
        print("OCR 결과가 없습니다.")
        state["processed_text"] = ""
        return state


    # OCR 텍스트 추출
    text_parts = []
    for line in result[0]:
        try:
            rec_text = line[1][0]
            text_parts.append(rec_text)
        except Exception as e:
            print(f"Skipping invalid OCR line: {line} - {e}")


    # OCR 텍스트를 하나의 문자열로 저장
    state["processed_text"] = " ".join(text_parts)


    # OCR 결과 출력
    print("Extracted text:")
    print(state["processed_text"])


    return state




# 2-2: PDF 파일 처리
def process_pdf(state: FileState) -> FileState:
    file_path = state["file_path"]
    parsed = parser.from_file(file_path)
    text = parsed.get("content", "")
    state["processed_text"] = text.strip() if text else ""
    return state




# 2-3: CSV 파일 처리
def process_csv(state: FileState) -> FileState:
    file_path = state["file_path"]
    df = pd.read_csv(file_path, encoding="utf-8")
    state["processed_text"] = df
    return state




# 3단계: LLM 활용해 요약하는 노드
def summarize(state: FileState) -> FileState:
    data = state["processed_text"]
    prompt = f"다음 데이터를 요약하고 정리해줘:\n{data}"
    chat_completion = groq_client.chat.completions.create(
        messages=[{"role": "user", "content": prompt}],
        model="gemma2-9b-it",
    )
    state["summary"] = chat_completion.choices[0].message.content
    return state




# 상태 그래프 생성
workflow = StateGraph(FileState)


# 노드 등록
workflow.add_node("classify", classify_file)
workflow.add_node("process_image", process_image)
workflow.add_node("process_pdf", process_pdf)
workflow.add_node("process_csv", process_csv)
workflow.add_node("summarize", summarize)


# START에서 분류 노드로 연결
workflow.add_edge(START, "classify")


# 분류 노드에서 조건부 에지를 사용해 각 분기 처리
workflow.add_conditional_edges(
    "classify",
    lambda state: state["file_type"],
    {
        "image": "process_image",
        "pdf": "process_pdf",
        "csv": "process_csv",
    },
)


# 각 분기 노드에서 요약 노드로 에지 연결
workflow.add_edge("process_image", "summarize")
workflow.add_edge("process_pdf", "summarize")
workflow.add_edge("process_csv", "summarize")


# 요약 노드에서 END로 연결
workflow.add_edge("summarize", END)


# 그래프 컴파일 (executor 생성)
executor = workflow.compile()


# 그래프 다이어그램 출력 (Mermaid 및 ASCII)
print(executor.get_graph().draw_mermaid())
executor.get_graph().print_ascii()


# 파일 경로 입력
# file_path = "data/2023년 인구성장률 현황.pdf"
# file_path = "data/2023년 혼인건수 현황.png"
# file_path = "data/CARD_SUBWAY_MONTH_202311.csv"
file_path = "data/2023년 인구성장률 현황.csv"
# file_path = "data/test.png"


initial_state: FileState = {
    "file_path": file_path,
    "file_type": None,
    "processed_text": None,
    "summary": None,
}
result = executor.invoke(initial_state)
print("\nGemma2 요약 결과:")
print(result["summary"])

728x90