1. 도입부 (Introduction)
대용량의 비정형 문서에서 정답을 찾아내는 검색 증강 생성(RAG, Retrieval-Augmented Generation) 시스템을 개발하다 보면 반드시 마주치는 두 가지 큰 장벽이 존재한다.
첫 번째는 '문서 전처리의 한계'다. 전통적인 텍스트 추출 라이브러리(PyPDF 등)는 PDF 내부에 정교한 표(Table)나 스캔 이미지가 포함되어 있을 경우 텍스트를 심각하게 왜곡하거나 무시한다. 표의 행과 열이 뒤섞인 상태로 데이터베이스에 적재되면 검색(Retrieval) 단계에서 엉뚱한 맥락을 짚게 된다.
두 번째는 '순차적 파이프라인의 한계'다. 기존의 랭체인(LangChain)이 제공하는 선형적인 체인(Chain) 구조는 검색 결과가 부실하거나 추가 연산이 필요한 복잡한 비즈니스 로직(예: 검색 결과 검증 후 재수색, 피드백 루프 등)을 다루기 어렵다.
이러한 문제를 극복하기 위해 제안되는 솔루션이 바로 Py-Zerox와 LangGraph의 결합이다. 비전 LLM 모델을 활용해 PDF 전체를 완벽한 구조의 마크다운(Markdown)으로 정규화하는 Py-Zerox로 데이터 전처리를 지배하고, 상태(State)를 기반으로 독립적인 노드를 연결하여 순환 및 복잡 흐름을 완벽하게 통제하는 LangGraph로 오케스트레이션을 수행하는 아키텍처다.
필자가 소득세법 분석용 RAG 엔진을 바탕으로 정립한 기술적 원리와 세부 코드를 완벽하게 학습 노트 형태로 정리해 본다.
2. 주요 특징 및 핵심 로직 (Main Features & Logic)
실전 파이프라인을 구축하기 전에 전체 시스템의 처리 단계와 아키텍처를 조감해 보자. 전체 워크플로우는 문서 변환 및 임베딩을 다루는 '데이터 인제스션(Ingestion) 파이프라인'과 사용자 질문에 따라 동적으로 상태를 변화시키며 결과를 내놓는 '랭그래프 RAG 파이프라인'으로 양분된다.
이러한 아키텍처적 특징을 정리하면 다음과 같은 장점을 갖는다.
- 완벽한 구조 보존: 표의 형태와 계층적 제목 구조가 유실되지 않아 벡터 검색의 정밀도가 극대화된다.
- 상태 중심적 오케스트레이션: 전체 파이프라인이 하나의 TypedDict인 State 객체만을 바라보고 흘러가므로, 각 기능 노드가 디커플링(Decoupling)되고 흐름 추적이 극도로 단순해진다.
- 유연한 흐름 제어: 필요에 따라 노드 사이에 조건부 분기(Conditional Edge)를 얹어 검증 실패 시 재검색을 돌리는 에이전트로 손쉽게 진화시킬 수 있다.
3. 상세 가이드 및 심층 분석 (Detailed Guide)
STEP 1. 패키지 설치 및 랭그래프 기초 맛보기
먼저 실습에 필요한 패키지들을 깔끔하게 설치해야 한다. 랭그래프를 사용하려면 langgraph 라이브러리가 필수적이다.
# 기본 랭체인, OpenAI 연동 패키지 및 랭그래프 라이브러리 설치
pip install -q python-dotenv langchain-openai langgraph
아래 예제 코드는 가장 단순한 형태의 랭그래프 동작 원리를 보여준다. 랭그래프에서는 상태를 정의하는 구조체와 해당 상태를 읽고 업데이트하는 노드(Node) 함수, 그리고 노드들을 이어주는 간선(Edge)의 관계가 핵심이다.
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
# 1. 그래프 전체에서 유지 및 관리할 공통 상태(State) 정의
class SimpleAgentState(TypedDict):
# add_messages 함수를 어노테이션으로 지정하면, 이전 대화 목록에 누적하여 덧붙이는 동작을 수행한다.
messages: Annotated[list[AnyMessage], add_messages]
# 2. LLM 초기화 (gpt-4o-mini 모델 사용)
llm = ChatOpenAI(model='gpt-4o-mini')
# 3. 상태값을 제어할 작업 노드(Node) 정의
def generate_simple_reply(state: SimpleAgentState) -> dict:
messages = state['messages']
# 이전까지 누적된 대화 목록을 LLM에 주입하여 답변을 생성한다.
ai_message = llm.invoke(messages)
# 그래프 엔진에 상태의 부분적 업데이트 신호(dict 형태)를 보낸다.
return {'messages': [ai_message]}
# 4. StateGraph 인스턴스화 및 아키텍처 빌딩
graph_builder = StateGraph(SimpleAgentState)
# 노드를 추가하고 진입 및 탈출 조건을 명시해 준다.
graph_builder.add_node('generate', generate_simple_reply)
graph_builder.add_edge(START, 'generate')
graph_builder.add_edge('generate', END)
# 최종 동작 가능한 형태의 실행 객체로 컴파일한다.
simple_graph = graph_builder.compile()
# 실행 테스트
initial_state = {'messages': [HumanMessage("서울의 수도는 어디인가요")]}
result = simple_graph.invoke(initial_state)
print(result['messages'][-1].content)💡 핵심 설계 기법 분석: add_messages와 상태 병합의 비밀
랭그래프의 State는 모든 노드가 공유하는 전역 데이터다. 특정 노드에서 {'messages': [ai_message]}를 리턴하면 기존의 리스트가 완전히 대체(Override)되는 것이 정상이다. 하지만 Annotated[list[AnyMessage], add_messages]와 같이 선언하면, 랭그래프 엔진은 새로운 값이 추가될 때 교체하는 대신 add_messages 함수를 내부적으로 실행하여 기존 리스트 뒤에 새로운 메시지를 안전하게 덧붙이는(Append) 방식으로 전역 상태를 자동 관리한다. 이 덕분에 상태 오염 없이 긴 대화 컨텍스트를 유지할 수 있다.
STEP 2. 비전 OCR 혁신: Py-Zerox를 활용한 PDF 구조화
소득세법과 같이 표와 특수 서식이 가득한 문서는 기존의 PyPDFLoader로 추출하면 문맥이 완전히 망가진다. 아래 코드는 일반적인 로더와 비전 기반 고속 OCR 엔진인 Py-Zerox의 파싱 흐름을 완성도 높은 예시로 보여준다.
# 비전 파싱 및 로컬 텍스트 가공을 위한 고성능 OCR 라이브러리 설치
pip install py-zerox pypdf langchain-community langchain-text-splitters nest_asyncio
import nest_asyncio
import asyncio
from pyzerox import zerox
# 주피터 노트북 환경이나 비동기 루프가 기동 중인 환경에서 자원 충돌을 방지하기 위한 안전장치
nest_asyncio.apply()
async def parse_pdf_with_vision():
# 로컬 경로 또는 URL에 매핑된 PDF 지정
file_path = "./income_tax.pdf"
model = "gpt-4o-mini" # 강력하고 저렴한 비전 모델 지정
output_dir = "./documents" # 변환된 마크다운을 통합 저장할 디렉토리
# zerox는 내부적으로 PDF의 각 페이지를 렌더링한 뒤 Vision API로 통째로 전달하여
# 완벽하게 포맷팅된 구조적 Markdown 형식의 통합 문서를 복원해 낸다.
result = await zerox(
file_path=file_path,
model=model,
output_dir=output_dir,
custom_system_prompt="출력 형식은 철저히 Markdown을 유지하고, 복잡한 세금율 계산 표는 원본 테이블의 행과 열 레이아웃을 표 형태로 완벽히 재현해라.",
select_pages=None # None 설정 시 문서 전체 변환
)
return result
# 비동기 프로세스 가동 및 출력 리포트 획득
document_markdown = asyncio.run(parse_pdf_with_vision())
print(document_markdown)
STEP 3. 텍스트 분할(Split) 및 고성능 벡터 저장소(Chroma DB) 적재
Py-Zerox가 변환해 준 마크다운 문서를 청크 단위로 정교하게 분할하고, OpenAI의 고성능 임베딩 엔진을 사용해 로컬 벡터 데이터베이스인 Chroma DB에 영구 보관한다.
# 로컬 벡터 스페이스 관리를 위한 고속 라이브러리 탑재
pip install -q langchain-chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
# 1. 줄 바꿈 단위를 계층적으로 수색하며 청크를 분리하는 정교한 Splitter 정의
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1500, # 한 청크 내부에 들어갈 글자 크기
chunk_overlap=100, # 문맥 유실을 막기 위해 앞뒤로 겹치는 길이 설정
separators=['\n\n', '\n']
)
# 2. Py-Zerox가 산출해 준 통합 텍스트 파일 수집
text_path = './documents/income_tax.txt'
loader = TextLoader(text_path)
document_list = loader.load_and_split(text_splitter)
# 3. OpenAI 대용량 차세대 임베딩 모델 인스턴스화
embeddings = OpenAIEmbeddings(model='text-embedding-3-large')
# 4. 로컬 디렉토리에 컬렉션 세팅 및 청크 인덱싱 영구 전파
vector_store = Chroma.from_documents(
documents=document_list,
embedding=embeddings,
collection_name='income_tax_collection',
persist_directory='./income_tax_collection'
)
# 5. 검색 가동을 위한 검색기(Retriever) 추출 (가장 유사도가 높은 3개 단락 반환)
retriever = vector_store.as_retriever(search_kwargs={'k': 3})
STEP 4. LangGraph 기반 RAG 파이프라인 설계 및 오케스트레이션
개별 단위 기술이 갖춰졌으니 이제 핵심인 LangGraph RAG 에이전트를 빌드해 보자. 그래프 내부에서 공유될 상태를 정의하고, '검색(Retrieve)'과 '답변 생성(Generate)' 작업을 수행하는 독립 노드를 생성한다.
from typing_extensions import List, TypedDict
from langchain_core.documents import Document
from langsmith import Client
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
# 1. RAG 파이프라인용 전역 상태(State) 설계
class RAGAgentState(TypedDict):
query: str # 사용자의 원본 질의
context: List[Document] # 데이터베이스로부터 검색된 원본 청크 목록
answer: str # LLM이 도출해 낸 최종 답변
# 2. 상태 수집 노드 (Retrieve Node) 정의
def retrieve_node(state: RAGAgentState) -> dict:
query = state['query']
# 이전에 선언해 둔 Chroma 검색기를 통해 관련 문서를 동적으로 수집한다.
docs = retriever.invoke(query)
# 수집된 연관 문서들을 context 상태값에 밀어 넣는다.
return {'context': docs}
# 3. 데이터 및 외부 프롬프트 허브 동기화
client = Client()
# LangSmith 허브에서 안전하게 검증된 RAG 표준 프롬프트 원격 수집
from langchain import hub
prompt = hub.pull("rlm/rag-prompt")
# 거대 비전 처리 및 세밀한 추론을 제어할 최신 GPT-4o 인스턴스 정의
llm = ChatOpenAI(model='gpt-4o', temperature=0)
# 4. 답변 생성 노드 (Generate Node) 정의
def generate_node(state: RAGAgentState) -> dict:
context = state['context']
query = state['query']
# 랭체인 표현 언어(LCEL) 구조를 결합해 체인 정의
rag_chain = prompt | llm
# 질문과 문맥 데이터를 조합하여 정밀 텍스트 합성
response = rag_chain.invoke({'question': query, 'context': context})
# 최종 도출된 답변 데이터를 answer 상태값에 반영
return {'answer': response.content}
이제 이 두 노드를 연결하는 설계 방식을 이해해야 한다. LangGraph는 두 가지 흐름 구축 방식을 제공한다. 명시적인 화살표 제어를 사용하는 방식과 연속적인 순서만 명시하는 하이브리드 시퀀스 빌더 방식이다. 두 가지 구조 모두 컴파일 후 동일한 그래프를 구성한다.
방식 A. 명시적 빌더 선언 및 세부 에지 매핑
가장 기본적이고 직관적인 구성 방법으로, 노드 간의 위상과 연결 관계를 엄격히 지정한다.
# 1. StateGraph에 전용 상태 클래스 주입
rag_builder = StateGraph(RAGAgentState)
# 2. 실행 작업을 담당할 노드 이름과 함수 수동 등록
rag_builder.add_node('retrieve', retrieve_node)
rag_builder.add_node('generate', generate_node)
# 3. 시작 지점(START)에서부터 종결 지점(END)까지 흐름 제어 매핑
rag_builder.add_edge(START, 'retrieve')
rag_builder.add_edge('retrieve', 'generate')
rag_builder.add_edge('generate', END)
# 4. 컴파일 가동 및 동작 가능한 흐름 객체 획득
rag_graph = rag_builder.compile()
방식 B. 시퀀스 헬퍼(add_sequence) 선언 기법 (간결한 선형 구조용)
만약 흐름이 분기나 루프 없이 일직선으로만 흘러간다면, 노드의 순서 리스트를 한 번에 주입하는 편리한 add_sequence 헬퍼 메서드를 활용할 수 있다.
# add_sequence 메서드는 순서대로 실행할 노드 리스트를 입력받아
# 내부적으로 순차 연결 에지(retrieve -> generate)를 자동 구축해 준다.
sequence_builder = StateGraph(RAGAgentState).add_sequence([retrieve_node, generate_node])
# 시작점과 끝점 처리만 직접 명시해 준다.
sequence_builder.add_edge(START, 'retrieve')
sequence_builder.add_edge('generate', END)
# 시퀀스 그래프 컴파일
sequence_graph = sequence_builder.compile()
RAG 에이전트 최종 구동 및 검증
이렇게 완성된 두 그래프는 내부적으로 완전하게 일치하는 지향성 비순환 그래프(DAG) 구조를 가지며, 사용법 또한 동일하다. 아래는 실제 구동 테스트 과정이다.
# 1. 최초 질문 설정 및 실행 상태(State) 선언
initial_state = {'query': '연봉 5천만 원인 직장인의 근로소득세 과세표준 구간과 간이세액 세율은 어떻게 적용되나요?'}
# 2. 컴파일 완료된 LangGraph 실행
final_output = rag_graph.invoke(initial_state)
# 3. 전역 상태에 무결하게 보존되어 누적된 최종 결론 출력
print("--- [최종 생성된 답변] ---")
print(final_output['answer'])
4. 실전 팁 및 예외 상황 대응 (Practical Tips & Troubleshooting)
실전 프로젝트에서 랭그래프 RAG를 운영하다 보면 자주 부딪히는 기술적 장애 요인과 극복 팁들을 공유한다.
1) 로컬 스레드 충돌 및 비동기 순환 데드락 해결
주피터 환경이나 장시간 지속되는 Fast Stream 백엔드 환경에서 비동기 메서드(asyncfor page in loader.alazy_load())를 중첩 구동하다 보면 "RuntimeError: This event loop is already running" 예외를 던지며 애플리케이션이 죽는 현상이 일어난다. 반드시 시스템 부팅 초입 영역에 nest_asyncio.apply()를 설정하여 로컬 이벤트 루프가 정상적으로 네스트되도록 안전장치를 확보해야 한다.
2) Py-Zerox 비전 파싱 레이턴시 극복 방안
비전 OCR을 전 페이지에 적용하는 방식은 전통적인 텍스트 단순 긁기 대비 레이턴시(속도 지연)와 LLM 호출 비용이 높다는 제약이 있다. 따라서 실무에서는 다음과 같은 복합 파싱 전략(Hybrid Parsing Strategy)을 추천한다.
- 1단계로 파이썬의 표준 pypdf를 사용해 데이터 로딩을 고속 시도한다.
- 해당 페이지 내부에 복잡한 캔버스나 <Table> 요소의 존재 여부를 정규식으로 먼저 감지해 낸다.
- 표가 발견되는 특정 영역이나 수식 스캔 영역만 선택하여 동적으로 Zerox(select_pages=[대상페이지번호]) 형태로 전처리를 위임 가동함으로써 불필요한 비용 소모와 지연 시간을 동시에 극복할 수 있다.
5. 결론 (Conclusion)
비전 OCR을 가동하는 Py-Zerox와 유연한 흐름 제어가 가능한 LangGraph를 결합해 보았다. 이 조합은 단순한 단방향 파이프라인의 한계를 완벽히 타파한다.
이 파이프라인을 활용하면, 추후 사용자의 피드백에 따라 질문을 변경하여 다시 검색하는 '자가 교정형 RAG(Self-RAG)'나 수집된 관련 문서가 타당하지 않은 경우 스스로 판단해 구글 웹 검색 노드로 상태를 분기시키는 고차원적인 AI 에이전트로 아주 유연하게 발전시킬 수 있다.
이제 단순한 라이브러리의 기능 나열식 구현 단계는 끝났다. 상태를 완벽히 캡슐화하고 격리된 노드 사이에서 신호를 주고받는 이 명확한 에이전트 설계 철학을 여러분의 실무 지식 베이스에도 강력하게 이식해 보길 권장한다.
#본 이미지들은 AI를 이용하여 생성하였습니다.
'LLM > LangGraph' 카테고리의 다른 글
| [LangGraph]메모리 아키텍처—RemoveMessage와 Summarizef로 만드는 메시지 요약, 삭제 노드 구축 (0) | 2026.06.16 |
|---|---|
| [LangGraph] 에이전트와 LangGraph 도구(Tool) 통합 정리 (0) | 2026.06.15 |
| [LangGraph] 병렬 처리(Parallel)와 routeLLM으로 완성하는 AI Agent (0) | 2026.06.13 |
| [LangGraph] 로컬 데이터의 한계 극복: Corrective RAG(CRAG)와 계층형 서브그래프 설계 가이드 (0) | 2026.06.13 |
| [LangGraph] 단순 조회를 넘어 스스로 생각하는 AI: Agent RAG와 Self-RAG 정리 (0) | 2026.06.12 |
