6. LangGraph - LangGraph 법률 PDF QA 종합 프로젝트
- 본 문서는 LangGraph 학습 시리즈의 최종 문서로, 이전 문서들(01~05)에서 학습한 모든 개념과 패턴을 종합 활용하여 실무 수준의 법률 문서 QA 시스템을 구축하는 프로젝트 매뉴얼이다.
- 개인정보보호법, 근로기준법, 주택임대차보호법 3종의 법률 PDF 문서를 파싱하고 벡터저장소에 인덱싱한 후, Corrective RAG 기반의 법률별 전문 에이전트를 구현한다.
- Adaptive RAG 패턴으로 사용자 질문을 분석하여 적절한 법률 에이전트로 라우팅하고, ReAct 에이전트로 답변 품질을 평가하며, Human-in-the-Loop(HITL)의 `interrupt()` 함수를 통해 사용자 승인 절차를 거치는 전체 파이프라인을 완성한다.
- Tool Calling, StateGraph, Reducer, ReAct Agent, MemorySaver, Adaptive RAG, Corrective RAG, 서브그래프, interrupt/Command 등 이전에 학습한 모든 LangGraph 핵심 기능이 하나의 시스템으로 통합되는 과정을 단계별로 학습한다.
1. 사전 작업
- 이 섹션에서는 프로젝트 실행에 필요한 환경 변수 로드와 기본 라이브러리를 임포트한다.
1.1. 환경변수 로드
from dotenv import load_dotenv
load_dotenv()
1.2. 기본 라이브러리
import re # [Python 표준 라이브러리] 정규표현식 - 섹션 3의 법률 PDF 파싱에서 조항 분리에 사용
import os, json # [Python 표준 라이브러리] os: 환경변수/파일 경로, json: JSON 파싱 (섹션 7 답변 평가 결과 파싱)
from glob import glob # [Python 표준 라이브러리] 파일 패턴 매칭 - 섹션 3에서 PDF 파일 목록 조회에 사용
from textwrap import dedent # [Python 표준 라이브러리] 멀티라인 문자열 들여쓰기 제거 - 섹션 5~8의 프롬프트 정의에 사용
from pprint import pprint # [Python 표준 라이브러리] 딕셔너리/리스트를 보기 좋게 출력
import uuid # [Python 표준 라이브러리] 고유 식별자 생성 - 섹션 9 Gradio UI에서 thread_id 생성에 사용
import warnings
warnings.filterwarnings("ignore")
2. 프로젝트 개요
- 이 섹션에서는 법률 상담 AI 어시스턴트의 실무 시나리오를 기획하고, 시스템 아키텍처를 설계한다.
- 본격적인 구현에 앞서 전체 시스템의 목표, 기능 요구사항, 기술 스택, 그리고 데이터 흐름을 명확히 정의한다.
2.1. 실무 시나리오
- 프로젝트명: 생활법률 AI 어시스턴트
- 배경: 일상생활에서 자주 접하는 법률 문제(개인정보 처리, 근로 계약, 주택 임대차)에 대해 전문적인 법률 조항을 근거로 답변을 제공하는 AI 어시스턴트를 개발한다.
- 대상 사용자: 법률 전문가가 아닌 일반 시민으로, 법률 용어에 익숙하지 않아도 자연어 질문으로 관련 법률 정보를 쉽게 얻을 수 있어야 한다.
- 핵심 요구사항:
- 사용자의 질문을 분석하여 관련 법률 분야를 자동으로 판별한다.
- 각 법률별 전문 에이전트가 해당 법률 조항을 검색하고, 관련성을 평가하여 정확한 답변을 생성한다.
- 법률 데이터베이스에 없는 최신 정보는 웹 검색으로 보완한다.
- 생성된 답변의 품질을 자동으로 평가하고, 사용자가 최종 승인/거부를 결정할 수 있다.
2.2. 개발 명세서
- 기능 요구사항:
- 법률 PDF 3종(개인정보보호법, 근로기준법, 주택임대차보호법)을 조항 단위로 파싱하여 벡터저장소에 인덱싱
- 각 법률별 Corrective RAG 에이전트 구현 (검색 → 정보 추출/평가 → 쿼리 재작성 → 답변 생성)
- Adaptive RAG 기반 질문 라우팅 (ToolSelector로 법률 분야 자동 판별, 복수 에이전트 병렬 실행 가능)
- ReAct 에이전트 기반 답변 품질 평가 (정확성, 관련성, 완전성, 인용 정확성, 명확성, 객관성)
- Human-in-the-Loop (`interrupt()` 함수 기반 사용자 승인/거부 → `Command(resume=...)` 로 재개 → 거부 시 재검색)
- Gradio 기반 대화형 UI (MemorySaver로 세션 상태 유지)
- 비기능 요구사항:
- CrossEncoderReranker(bge-reranker-v2-m3)를 통한 검색 정확도 향상
- 무한 루프 방지 (num_generations 카운터, 최대 2회 재시도)
- 출처 명시 (법률 조항 번호 또는 웹 URL 인용)
- 기술 스택:
- LLM: ChatOpenAI (gpt-4o-mini)
- 임베딩: OllamaEmbeddings (bge-m3)
- 리랭킹: HuggingFaceCrossEncoder (bge-reranker-v2-m3)
- 벡터저장소: Chroma (법률별 컬렉션 분리)
- 웹 검색: TavilySearch (langchain_tavily)
- 구조화 출력: Pydantic v2 BaseModel + with_structured_output
- 프레임워크: LangChain + LangGraph
- HITL: interrupt() / Command(resume=...) 패턴
- UI: Gradio ChatInterface
2.3. 전체 시스템 아키텍처
- 아래 다이어그램은 법률 상담 AI 어시스턴트의 전체 데이터 흐름을 시각화한다.
- 사용자의 질문이 입력되면 질문 분석 → 법률별 RAG 에이전트(병렬) → 최종 답변 생성 → 답변 평가 → HITL 확인의 파이프라인을 거쳐 최종 응답이 반환된다.
- 전체 시스템 데이터 흐름도
3. 단계 1: 데이터 준비 (법률 PDF 파싱 + 벡터저장소 인덱싱)
- 이 단계는 전체 시스템의 기반이 되는 데이터 파이프라인을 구축하는 첫 번째 단계이다.
- 3종의 법률 PDF 문서를 PyPDFLoader로 로드한 후, 정규표현식을 사용하여 장(章)과 조(條) 단위로 분리하고, LangChain Document 객체에 메타데이터와 함께 담아 Chroma 벡터저장소에 인덱싱한다.
- 설계 고려사항:
- 법률 문서는 장 → 조 → 항의 계층 구조를 갖지만, 모든 법률이 동일한 구조는 아니다. 개인정보보호법과 근로기준법은 장(章) 단위로 구분되지만, 주택임대차보호법은 장 없이 조문으로만 구성되어 별도의 파싱 함수(`parse_law_v2`)가 필요하다.
- Document 객체의 `page_content`에 법률명과 장 정보를 메타 텍스트로 포함시켜, 검색 시 컨텍스트 정보가 함께 제공되도록 한다.
- 법률별로 별도의 Chroma 컬렉션(`personal_law`, `labor_law`, `housing_law`)을 생성하여 검색 범위를 격리한다.
- PDF 문서 처리 파이프라인 (가로형)
- PDF 문서 처리 파이프라인 (세로형)
- 실선 화살표(→): 데이터 흐름을 나타낸다. PDF에서 시작하여 로드 → 파싱 → Document 생성 → 벡터저장소 인덱싱 순으로 데이터가 변환되며 흘러간다.
- 점선 화살표(-..->): 의존성을 나타낸다. OllamaEmbeddings 모델은 데이터를 직접 처리하지 않고, 각 Chroma 컬렉션이 Document를 벡터로 변환할 때 임베딩 엔진으로 사용된다.
- 파싱 분기: 개인정보보호법/근로기준법은 장(章) 단위 계층 구조가 있어 `parse_law()` 범용 함수를 사용하고, 주택임대차보호법은 장 구분 없이 조문만 존재하여 `parse_law_v2()` 전용 함수를 사용한다.
3.1. 개인정보보호법
- 개인정보보호법 PDF를 로드하고, 정규표현식으로 장(章)과 조(條)를 분리한 후 Document 객체로 변환하여 벡터저장소에 인덱싱한다.
- `parse_law()` 함수는 법률 텍스트를 서문, 장, 부칙의 3계층으로 분리하는 범용 파싱 함수이다.
- 장 내부에서는 `제X조(조항명)` 패턴으로 개별 조항을 추출한다.
# 법률 텍스트 파싱 함수 - 장이 있는 법률용
# 이 함수는 섹션 3.1(개인정보보호법)과 3.2(근로기준법)에서 공통으로 사용된다.
# 반환값: {'서문': str, '장': {장제목: [조항문자열, ...], ...}, '부칙': str} 형태의 딕셔너리
def parse_law(law_text):
# 서문 분리: 제1장 또는 제1조 이전의 모든 텍스트
preamble_pattern = r'^(.*?)(?=제1장|제1조)'
preamble = re.search(preamble_pattern, law_text, re.DOTALL)
if preamble:
preamble = preamble.group(1).strip()
# 장 분리: '제X장 제목' 형식으로 구분
chapter_pattern = r'(제\d+장\s+.+?)\n((?:제\d+조(?:의\d+)?(?:\(\w+\))?.*?)(?=제\d+장|부칙|$))'
chapters = re.findall(chapter_pattern, law_text, re.DOTALL)
# 부칙 분리
appendix_pattern = r'(부칙.*)'
appendix = re.search(appendix_pattern, law_text, re.DOTALL)
if appendix:
appendix = appendix.group(1)
parsed_law = {'서문': preamble, '장': {}, '부칙': appendix}
# 각 장 내에서 개별 조항 분리
for chapter_title, chapter_content in chapters:
# '제X조(조항명)' 패턴으로 개별 조항 추출
article_pattern = r'(제\d+조(?:의\d+)?\s*\([^)]+\).*?)(?=제\d+조(?:의\d+)?\s*\([^)]+\)|$)'
articles = re.findall(article_pattern, chapter_content, re.DOTALL)
parsed_law['장'][chapter_title.strip()] = [article.strip() for article in articles]
return parsed_law
- 예상 결과: `parsed_law`는 `{'서문': '...', '장': {'제1장 총칙': ['제1조(목적) ...', '제2조(정의) ...', ...], ...}, '부칙': '...'}` 형태의 딕셔너리로, 장별로 조항 리스트가 정리된다.
from langchain_community.document_loaders import PyPDFLoader # [LangChain 커뮤니티] PDF 파일을 페이지별로 로드하는 로더
from langchain_core.documents import Document # [LangChain 핵심] 문서 객체 클래스 - page_content와 metadata를 포함
from langchain_chroma import Chroma # [LangChain Chroma] 벡터저장소 - persist_directory로 디스크에 영구 저장
from langchain_ollama import OllamaEmbeddings # [LangChain Ollama] Ollama 서버의 임베딩 모델 래퍼
# 1. PDF 로드
# PyPDFLoader는 PDF 파일을 페이지 단위로 분리하여 Document 객체 리스트를 반환한다.
pdf_file = "개인정보 보호법(법률)(제19234호)(20240315).pdf"
loader = PyPDFLoader(pdf_file)
pages = loader.load()
print(f"총 페이지 수: {len(pages)}")
# 2. 페이지 텍스트 결합 및 불필요한 헤더/푸터 제거
# 법령 PDF에는 각 페이지마다 "법제처 N 국가법령정보센터\n법률명" 형식의 헤더가 포함되어 있으므로 정규표현식으로 제거한다.
text_for_delete = r"법제처\s+\d+\s+국가법령정보센터\n개인정보 보호법"
law_text = "\n".join([re.sub(text_for_delete, "", p.page_content).strip() for p in pages])
# 3. 법률 텍스트 파싱 (장/조 분리)
# parse_law(): [사용자 정의 - 위 기본 사용법에서 정의] 법률 텍스트를 서문/장/부칙으로 분리하는 함수
parsed_law = parse_law(law_text)
print(f"분리된 장 수: {len(parsed_law['장'])}")
# 4. Document 객체 생성 (메타데이터 포함)
# 각 조항을 개별 Document로 변환하고, 검색 시 컨텍스트 정보를 제공하기 위해
# page_content 상단에 법률명과 장 정보를 메타 텍스트로 추가한다.
final_docs = []
for law in parsed_law['장'].keys():
for article in parsed_law['장'][law]:
metadata = {
"source": pdf_file,
"chapter": law,
"name": "개인정보 보호법"
}
# 본문에 법률 정보를 추가하여 검색 시 컨텍스트 제공
content = f"[법률정보]\n다음 조항은 {metadata['name']} {metadata['chapter']}에서 발췌한 내용입니다.\n\n[법률조항]\n{article}"
final_docs.append(Document(page_content=content, metadata=metadata))
print(f"생성된 Document 수: {len(final_docs)}")
# 5. 벡터저장소에 인덱싱
# OllamaEmbeddings: Ollama 서버에서 bge-m3 모델을 사용하여 텍스트를 벡터로 변환한다.
# Chroma.from_documents(): Document 리스트를 임베딩하여 Chroma 벡터저장소에 저장한다.
# persist_directory: 벡터저장소를 디스크에 영구 저장하는 경로 (ChromaDB 내부적으로 PersistentClient 사용)
embeddings_model = OllamaEmbeddings(model="bge-m3")
personal_db = Chroma.from_documents(
documents=final_docs,
embedding=embeddings_model,
collection_name="personal_law",
persist_directory="./chroma_db",
)
- 실제 출력 결과:
총 페이지 수: 41
분리된 장 수: 10
생성된 Document 수: 98
- Document 객체 예시:
[법률정보]
다음 조항은 개인정보 보호법 제1장 총칙에서 발췌한 내용입니다.
[법률조항]
제1조(목적) 이 법은 개인정보의 처리 및 보호에 관한 사항을 정함으로써 개인의 자유와 권리를 보호하고, ...
metadata: {'source': '개인정보 보호법(법률)(제19234호)(20240315).pdf', 'chapter': '제1장 총칙', 'name': '개인정보 보호법'}
3.2. 근로기준법
- 근로기준법은 개인정보보호법과 동일한 장/조 구조이므로 같은 `parse_law()` 함수를 사용한다.
from langchain_community.document_loaders import PyPDFLoader # [LangChain 커뮤니티] PDF 파일 로더
from langchain_core.documents import Document # [LangChain 핵심] 문서 객체 클래스
from langchain_chroma import Chroma # [LangChain Chroma] 벡터저장소
from langchain_ollama import OllamaEmbeddings # [LangChain Ollama] 임베딩 모델
# 1. PDF 로드
pdf_file = "근로기준법(법률)(제18176호)(20211119).pdf"
loader = PyPDFLoader(pdf_file)
pages = loader.load()
print(f"총 페이지 수: {len(pages)}")
# 2. 페이지 텍스트 결합 및 헤더/푸터 제거
text_for_delete = r"법제처\s+\d+\s+국가법령정보센터\n근로기준법"
law_text = "\n".join([re.sub(text_for_delete, "", p.page_content).strip() for p in pages])
# 3. 법률 텍스트 파싱
# parse_law(): [사용자 정의 - 섹션 3.1에서 정의] 장/조 구조 파싱 함수
parsed_law = parse_law(law_text)
print(f"분리된 장 수: {len(parsed_law['장'])}")
# 4. Document 객체 생성
# 개인정보보호법과 동일한 구조로 Document를 생성하되, name 메타데이터만 "근로기준법"으로 변경
final_docs = []
for law in parsed_law['장'].keys():
for article in parsed_law['장'][law]:
metadata = {
"source": pdf_file,
"chapter": law,
"name": "근로기준법"
}
content = f"[법률정보]\n다음 조항은 {metadata['name']} {metadata['chapter']}에서 발췌한 내용입니다.\n\n[법률조항]\n{article}"
final_docs.append(Document(page_content=content, metadata=metadata))
print(f"생성된 Document 수: {len(final_docs)}")
# 5. 벡터저장소에 인덱싱
# embeddings_model: [사용자 정의 - 섹션 3.1에서 초기화] OllamaEmbeddings(model="bge-m3") 인스턴스
embeddings_model = OllamaEmbeddings(model="bge-m3")
labor_db = Chroma.from_documents(
documents=final_docs,
embedding=embeddings_model,
collection_name="labor_law",
persist_directory="./chroma_db",
)
- 실제 출력 결과:
총 페이지 수: 52
분리된 장 수: 12
생성된 Document 수: 116
3.3. 주택임대차보호법
- 주택임대차보호법은 장(章) 없이 조문으로만 구성되어 있어, 장 구조를 먼저 확인하고 조문만 직접 추출하는 `parse_law_v2()` 함수를 별도로 정의한다.
- `parse_law_v2()`는 `parse_law()`를 확장한 함수로, 장이 있는 경우와 없는 경우를 모두 처리한다.
- 장이 있으면 기존 `parse_law()`와 동일하게 처리하고, 장이 없으면 서문과 부칙을 제외한 본문에서 조문을 직접 추출한다.
# 법률 텍스트 파싱 함수 v2 - 장이 없는 법률도 처리 가능
# parse_law()와의 차이점: 장이 없는 경우 '조문' 키로 직접 조항 리스트를 저장한다.
# 반환값: 장이 있으면 {'서문': str, '장': {...}, '부칙': str}
# 장이 없으면 {'서문': str, '조문': [조항문자열, ...], '부칙': str}
def parse_law_v2(law_text):
# 서문 분리
preamble_pattern = r'^(.*?)(?=제1장|제1조)'
preamble = re.search(preamble_pattern, law_text, re.DOTALL)
if preamble:
preamble = preamble.group(1).strip()
# 장 분리 시도
chapter_pattern = r'(제\d+장\s+.+?)\n((?:제\d+조(?:의\d+)?(?:\(\w+\))?.*?)(?=제\d+장|부칙|$))'
chapters = re.findall(chapter_pattern, law_text, re.DOTALL)
# 부칙 분리
appendix_pattern = r'(부칙.*)'
appendix = re.search(appendix_pattern, law_text, re.DOTALL)
if appendix:
appendix = appendix.group(1)
parsed_law = {'서문': preamble, '부칙': appendix}
# 조 분리 패턴
article_pattern = r'(제\d+조(?:의\d+)?\s*\([^)]+\).*?)(?=제\d+조(?:의\d+)?\s*\([^)]+\)|$)'
if chapters: # 장이 있는 경우
parsed_law['장'] = {}
for chapter_title, chapter_content in chapters:
articles = re.findall(article_pattern, chapter_content, re.DOTALL)
parsed_law['장'][chapter_title.strip()] = [article.strip() for article in articles]
else: # 장이 없는 경우 - 조문 직접 추출
main_text = re.sub(preamble_pattern, '', law_text, flags=re.DOTALL)
main_text = re.sub(appendix_pattern, '', main_text, flags=re.DOTALL)
articles = re.findall(article_pattern, main_text, re.DOTALL)
parsed_law['조문'] = [article.strip() for article in articles]
return parsed_law
- 예상 결과: 주택임대차보호법의 경우 `parsed_law`는 `{'서문': '...', '조문': ['제1조(목적) ...', '제2조(적용범위) ...', ...], '부칙': '...'}` 형태로, 장 없이 조문 리스트가 직접 포함된다.
from langchain_community.document_loaders import PyPDFLoader # [LangChain 커뮤니티] PDF 파일 로더
from langchain_core.documents import Document # [LangChain 핵심] 문서 객체 클래스
from langchain_chroma import Chroma # [LangChain Chroma] 벡터저장소
from langchain_ollama import OllamaEmbeddings # [LangChain Ollama] 임베딩 모델
# 1. PDF 로드
pdf_file = "주택임대차보호법(법률)(제19356호)(20230719).pdf"
loader = PyPDFLoader(pdf_file)
pages = loader.load()
print(f"총 페이지 수: {len(pages)}")
# 2. 페이지 텍스트 결합 및 헤더/푸터 제거
text_for_delete = r"법제처\s+\d+\s+국가법령정보센터\n주택임대차보호법"
law_text = "\n".join([re.sub(text_for_delete, "", p.page_content).strip() for p in pages])
# 3. parse_law_v2로 파싱 (장 없는 구조 처리)
# parse_law_v2(): [사용자 정의 - 위 기본 사용법에서 정의] 장이 없는 법률도 처리하는 확장 파싱 함수
parsed_law = parse_law_v2(law_text)
print(f"분리된 조문 수: {len(parsed_law['조문'])}")
# 4. Document 객체 생성 (장이 없으므로 chapter 메타데이터 생략)
# 주택임대차보호법은 장 없이 조문으로만 구성되므로 chapter 메타데이터 없이 Document를 생성한다.
final_docs = []
for article in parsed_law['조문']:
metadata = {
"source": pdf_file,
"name": "주택임대차보호법"
}
content = f"[법률정보]\n다음 조항은 {metadata['name']}에서 발췌한 내용입니다.\n\n[법률조항]\n{article}"
final_docs.append(Document(page_content=content, metadata=metadata))
print(f"생성된 Document 수: {len(final_docs)}")
# 5. 벡터저장소에 인덱싱
embeddings_model = OllamaEmbeddings(model="bge-m3")
housing_db = Chroma.from_documents(
documents=final_docs,
embedding=embeddings_model,
collection_name="housing_law",
persist_directory="./chroma_db",
)
- 실제 출력 결과:
총 페이지 수: 7
분리된 조문 수: 19
생성된 Document 수: 19
4. 단계 2: 도구 정의 (검색 도구 + 웹 검색 도구 + LLM 바인딩)
- 이 단계는 앞서 구축한 벡터저장소를 LangGraph 에이전트에서 활용할 수 있도록 도구(Tool)로 변환하는 단계이다.
- 단계 1에서 생성한 3개의 벡터저장소(personal_law, labor_law, housing_law)와 웹 검색을 각각 `@tool` 데코레이터로 도구화하고, `CrossEncoderReranker`를 통해 검색 정확도를 향상시킨다.
- 설계 고려사항:
- 단순 유사도 검색만으로는 법률 문서의 미묘한 관련성을 판별하기 어렵다. 따라서 1차 검색(k=5)으로 후보 문서를 넓게 확보한 뒤, `CrossEncoderReranker`(bge-reranker-v2-m3)로 상위 2개를 선별하는 2단계 검색 구조를 적용한다.
- 모든 도구가 `List[Document]` 형태로 결과를 반환하도록 통일하여, 이후 RAG 파이프라인에서 동일한 인터페이스로 처리할 수 있도록 한다.
- 검색 결과가 없을 경우 빈 리스트 대신 "관련 정보를 찾을 수 없습니다." 메시지를 담은 Document를 반환하여 후속 처리가 중단되지 않도록 한다.
- 웹 검색은 `TavilySearch`(langchain_tavily 패키지)를 사용하며, 검색 결과를 Document 형태로 변환하여 법률 검색 도구와 동일한 인터페이스를 유지한다.
- 검색 도구 선택 및 실행 흐름
4.1. 법률 정보 검색 도구 및 웹 검색 도구 정의
- `ContextualCompressionRetriever`는 기본 리트리버의 결과를 압축기(compressor)로 재정렬하는 래퍼이다.
- `CrossEncoderReranker`는 쿼리와 문서 쌍을 Cross-Encoder 모델로 직접 비교하여 관련성 점수를 재계산하고, 상위 `top_n`개만 반환한다.
- `TavilySearch`는 langchain_tavily 패키지에서 제공하는 최신 Tavily 검색 도구이다. 이전의 `TavilySearchResults`(langchain_community)나 `TavilySearchAPIRetriever`를 대체한다.
from langchain.retrievers import ContextualCompressionRetriever # [LangChain 내장] 기본 리트리버 결과를 압축/재정렬하는 래퍼
from langchain.retrievers.document_compressors import CrossEncoderReranker # [LangChain 내장] Cross-Encoder 기반 리랭킹 압축기
from langchain_community.cross_encoders import HuggingFaceCrossEncoder # [LangChain 커뮤니티] HuggingFace Cross-Encoder 모델 래퍼
# Cross-Encoder 리랭킹 모델 초기화
# BAAI/bge-reranker-v2-m3: 다국어 지원 리랭킹 모델로, 한국어 법률 문서에도 우수한 성능을 보인다.
rerank_model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")
cross_reranker = CrossEncoderReranker(model=rerank_model, top_n=2) # 상위 2개 선별
# 기본 리트리버를 리랭킹 리트리버로 래핑
# ContextualCompressionRetriever: 1차 검색(base_retriever) → 리랭킹(base_compressor) 2단계 파이프라인
# as_retriever(search_kwargs={"k": 5}): 벡터저장소에서 상위 5개 후보 문서를 유사도 검색으로 가져옴
db_retriever = ContextualCompressionRetriever(
base_compressor=cross_reranker, # 리랭킹 압축기
base_retriever=db.as_retriever(search_kwargs={"k": 5}), # [LangChain 내장] as_retriever(): 벡터저장소를 리트리버로 변환
)
- 예상 결과: `db_retriever.invoke("질문")` 호출 시 5개 후보 중 Cross-Encoder 점수 기준 상위 2개의 Document가 반환된다.
from langchain_chroma import Chroma # [LangChain Chroma] 벡터저장소 - 섹션 3에서 생성한 법률별 컬렉션 로드에 사용
from langchain_ollama import OllamaEmbeddings # [LangChain Ollama] 임베딩 모델 - 벡터저장소 로드 시 동일 모델 필요
from langchain_core.documents import Document # [LangChain 핵심] 문서 객체 클래스 - 검색 결과 반환 타입
from langchain.retrievers import ContextualCompressionRetriever # [LangChain 내장] 리랭킹 래퍼 - 1차 검색 후 리랭킹 적용
from langchain.retrievers.document_compressors import CrossEncoderReranker # [LangChain 내장] Cross-Encoder 리랭커
from langchain_community.cross_encoders import HuggingFaceCrossEncoder # [LangChain 커뮤니티] HuggingFace 모델 래퍼
from langchain_tavily import TavilySearch # [LangChain Tavily] Tavily 웹 검색 도구 - 최신 API 사용
from langchain_core.tools import tool # [LangChain 핵심] @tool 데코레이터 - 함수를 LangGraph 도구로 변환
from typing import List
# 임베딩 모델 및 리랭킹 모델 초기화
# embeddings_model: 섹션 3에서 벡터저장소 생성 시 사용한 것과 동일한 모델이어야 한다.
embeddings_model = OllamaEmbeddings(model="bge-m3")
rerank_model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3") # Cross-Encoder 리랭킹 모델
cross_reranker = CrossEncoderReranker(model=rerank_model, top_n=2) # 상위 2개 문서만 선별
# --- 개인정보보호법 검색 도구 ---
# Chroma(): 섹션 3.1에서 from_documents()로 생성한 벡터저장소를 기존 경로에서 로드
# persist_directory: 섹션 3.1에서 저장한 동일 경로 (ChromaDB 내부적으로 PersistentClient 사용)
personal_db = Chroma(
embedding_function=embeddings_model,
collection_name="personal_law",
persist_directory="./chroma_db",
)
# ContextualCompressionRetriever: 1차 유사도 검색(k=5) → CrossEncoderReranker로 상위 2개 선별
personal_db_retriever = ContextualCompressionRetriever(
base_compressor=cross_reranker,
base_retriever=personal_db.as_retriever(search_kwargs={"k": 5}),
)
@tool
def personal_law_search(query: str) -> List[Document]:
"""개인정보보호법 법률 조항을 검색합니다."""
# personal_db_retriever: [사용자 정의 - 위에서 정의] 리랭킹 적용된 개인정보보호법 검색 리트리버
docs = personal_db_retriever.invoke(query)
if len(docs) > 0:
return docs
return [Document(page_content="관련 정보를 찾을 수 없습니다.")]
# --- 근로기준법 검색 도구 ---
# 개인정보보호법과 동일한 구조로, collection_name만 "labor_law"로 변경
labor_db = Chroma(
embedding_function=embeddings_model,
collection_name="labor_law",
persist_directory="./chroma_db",
)
labor_db_retriever = ContextualCompressionRetriever(
base_compressor=cross_reranker,
base_retriever=labor_db.as_retriever(search_kwargs={"k": 5}),
)
@tool
def labor_law_search(query: str) -> List[Document]:
"""근로기준법 법률 조항을 검색합니다."""
docs = labor_db_retriever.invoke(query)
if len(docs) > 0:
return docs
return [Document(page_content="관련 정보를 찾을 수 없습니다.")]
# --- 주택임대차보호법 검색 도구 ---
# 개인정보보호법과 동일한 구조로, collection_name만 "housing_law"로 변경
housing_db = Chroma(
embedding_function=embeddings_model,
collection_name="housing_law",
persist_directory="./chroma_db",
)
housing_db_retriever = ContextualCompressionRetriever(
base_compressor=cross_reranker,
base_retriever=housing_db.as_retriever(search_kwargs={"k": 5}),
)
@tool
def housing_law_search(query: str) -> List[Document]:
"""주택임대차보호법 법률 조항을 검색합니다."""
docs = housing_db_retriever.invoke(query)
if len(docs) > 0:
return docs
return [Document(page_content="관련 정보를 찾을 수 없습니다.")]
# --- 웹 검색 도구 ---
# TavilySearch: langchain_tavily 패키지에서 제공하는 최신 Tavily 검색 도구
# 이전의 TavilySearchResults(langchain_community)나 TavilySearchAPIRetriever를 대체한다.
# max_results=5: 최대 5개의 검색 결과를 반환한다.
tavily_search = TavilySearch(max_results=5)
@tool
def web_search(query: str) -> List[Document]:
"""데이터베이스에 없는 정보 또는 최신 정보를 웹에서 검색합니다."""
# TavilySearch.invoke()는 검색 결과를 문자열로 반환하므로 Document로 변환한다.
results = tavily_search.invoke(query)
# TavilySearch의 반환값을 Document 형태로 변환하여 법률 검색 도구와 동일한 인터페이스를 유지
if isinstance(results, str):
# 문자열 결과인 경우 단일 Document로 변환
return [Document(
page_content=results,
metadata={"source": "web search"}
)]
elif isinstance(results, list):
# 리스트 결과인 경우 각 항목을 Document로 변환
formatted_docs = []
for result in results:
if isinstance(result, dict):
url = result.get("url", "")
content = result.get("content", str(result))
formatted_docs.append(
Document(
page_content=f'<Document href="{url}"/>\n{content}\n</Document>',
metadata={"source": "web search", "url": url}
)
)
else:
formatted_docs.append(
Document(
page_content=str(result),
metadata={"source": "web search"}
)
)
if len(formatted_docs) > 0:
return formatted_docs
return [Document(page_content="관련 정보를 찾을 수 없습니다.")]
# 도구 목록 정의 - 섹션 4.2, 5, 6, 7에서 참조됨
# 4개의 @tool 데코레이터 함수를 리스트로 묶어 LLM에 바인딩하거나 ReAct 에이전트에 전달한다.
tools = [personal_law_search, labor_law_search, housing_law_search, web_search]
4.2. LLM 모델
- LLM에 도구를 바인딩하여, 질문에 따라 적절한 도구를 자동으로 선택하도록 한다.
from langchain_openai import ChatOpenAI # [LangChain OpenAI] OpenAI ChatGPT 모델 래퍼
# 기본 LLM - 섹션 5~9 전체에서 사용되는 핵심 LLM 인스턴스
# temperature=0: 결정론적 출력 (법률 답변의 일관성을 위해)
# streaming=True: 스트리밍 응답 활성화 (Gradio UI에서 실시간 출력에 활용)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)
# LLM에 도구 바인딩
# tools: [사용자 정의 - 섹션 4.1 참조] [personal_law_search, labor_law_search, housing_law_search, web_search]
# bind_tools(): LLM이 질문에 따라 적절한 도구를 자동으로 선택할 수 있도록 도구 스키마를 LLM에 바인딩
llm_with_tools = llm.bind_tools(tools)
- 도구 바인딩 후 질문 유형별 동작 확인:
# 법률 관련 질문 → 해당 법률 검색 도구 호출
query = "연차휴가 부여 기준에 대해서 설명해주세요."
ai_msg = llm_with_tools.invoke(query)
pprint(ai_msg.tool_calls)
# [{'name': 'labor_law_search', 'args': {'query': '연차휴가 부여 기준'}, ...}]
# 일반 질문 → 도구 호출 없이 직접 답변
query = "안녕하세요?"
ai_msg = llm_with_tools.invoke(query)
pprint(ai_msg.tool_calls)
# []
# 복합 질문 → 여러 도구 동시 호출
query = "연차휴가 부여 기준에 대해서 설명해주세요. 2023년 연차휴가 사용 비율은 어느 정도인가요?"
ai_msg = llm_with_tools.invoke(query)
pprint(ai_msg.tool_calls)
# [{'name': 'labor_law_search', ...}, {'name': 'web_search', ...}]
5. 단계 3: 법률별 RAG 에이전트 (Corrective RAG 적용)
- 이 단계는 전체 시스템의 핵심인 법률별 전문 RAG 에이전트를 구현하는 단계이다.
- 각 법률(개인정보보호법, 근로기준법, 주택임대차보호법)과 웹 검색에 대해 동일한 Corrective RAG 구조를 적용한 4개의 독립 에이전트를 구축한다.
- Corrective RAG는 05_LangGraph_AgenticRAG에서 학습한 패턴으로, 검색된 문서를 평가하고 품질이 부족하면 쿼리를 재작성하여 재검색하는 자기 보정(Self-Correcting) 순환 구조이다.
- 설계 고려사항:
- 4개 에이전트는 동일한 그래프 구조(retrieve → extract_and_evaluate → should_continue → rewrite_query/generate_answer)를 가지며, 검색 도구와 시스템 프롬프트의 법률 전문 분야만 다르다.
- 정보 추출/평가 단계에서 Pydantic v2 `with_structured_output`을 사용하여 관련성 점수(relevance_score)와 충실성 점수(faithfulness_score)를 구조화된 형태로 추출한다.
- `num_generations` 카운터로 최대 2회까지만 재시도하여 무한 루프를 방지한다.
- 관련 정보가 1개 이상 추출되면 즉시 답변 생성으로 전환하여 불필요한 반복을 줄인다.
- Corrective RAG 에이전트 그래프 구조
5.1. State 및 Pydantic 모델 정의
- 모든 Corrective RAG 에이전트가 공유하는 기본 State(`CorrectiveRagState`)와 정보 추출용 Pydantic v2 모델을 먼저 정의한다.
- 각 법률별 에이전트는 `CorrectiveRagState`를 상속하여 법률 전용 State를 만든다.
- `CorrectiveRagState`는 질문, 생성 답변, 검색 문서, 재시도 횟수의 4가지 필드를 기본으로 가진다.
- `InformationStrip`은 추출된 정보 조각 하나를 표현하며, 내용/출처/관련성 점수/충실성 점수를 포함한다.
- `ExtractedInformation`은 여러 정보 조각과 전반적인 답변 가능성 점수를 포함하는 구조화된 출력 스키마이다.
- `RefinedQuestion`은 쿼리 재작성 시 개선된 질문과 그 이유를 담는다.
- 모든 Pydantic 모델은 Pydantic v2의 `BaseModel`을 사용한다. (이전의 `langchain_core.pydantic_v1`은 더 이상 사용하지 않는다.)
from pydantic import BaseModel, Field # [Pydantic v2] 데이터 검증 및 구조화 출력 스키마 정의
from typing import List, TypedDict, Optional
# [사용자 정의] 기본 Corrective RAG State - 섹션 5.2~5.5의 모든 법률별 에이전트 State가 이 클래스를 상속함
# TypedDict를 사용하여 LangGraph StateGraph의 State 스키마를 정의한다.
class CorrectiveRagState(TypedDict):
question: str # 사용자의 질문
generation: str # LLM 생성 답변
documents: List[Document] # 검색된 문서 (Document는 LangChain의 langchain_core.documents.Document 클래스)
num_generations: int # 재시도 횟수 (무한 루프 방지 - 최대 2회)
# 정보 추출용 Pydantic v2 모델
# with_structured_output()에 전달하여 LLM이 이 스키마에 맞는 구조화된 JSON을 반환하도록 한다.
class InformationStrip(BaseModel):
"""추출된 정보에 대한 내용과 출처, 관련성 점수"""
content: str = Field(..., description="추출된 정보 내용")
source: str = Field(..., description="정보의 출처(법률 조항 또는 URL 등)")
relevance_score: float = Field(..., ge=0, le=1, description="관련성 점수")
faithfulness_score: float = Field(..., ge=0, le=1, description="충실성 점수")
class ExtractedInformation(BaseModel):
"""문서에서 추출된 정보 조각들과 전반적인 답변 가능성 점수"""
strips: List[InformationStrip] = Field(..., description="추출된 정보 조각들")
query_relevance: float = Field(..., ge=0, le=1, description="전반적인 답변 가능성 점수")
class RefinedQuestion(BaseModel):
"""개선된 질문과 이유"""
question_refined: str = Field(..., description="개선된 질문")
reason: str = Field(..., description="이유")
- 예상 결과: `llm.with_structured_output(ExtractedInformation)`으로 LLM 호출 시, `ExtractedInformation` 객체가 반환되어 각 정보 조각의 관련성/충실성 점수에 프로그래밍적으로 접근할 수 있다. Pydantic v2에서는 `.model_json_schema()`로 JSON 스키마를 확인할 수 있다.
5.2. 개인정보보호법 Corrective RAG 에이전트
- 개인정보보호법 전문 에이전트를 Corrective RAG 패턴으로 구현한다.
- 이 에이전트의 구조가 다른 법률 에이전트의 템플릿이 되므로 상세히 설명한다.
- 에이전트는 4개의 노드(retrieve, extract_and_evaluate, rewrite_query, generate_answer)와 1개의 조건부 엣지(should_continue)로 구성된다.
- `extract_and_evaluate` 노드에서 문서별로 `with_structured_output(ExtractedInformation)`을 호출하여 정보를 구조적으로 추출하고, `query_relevance < 0.8`인 문서는 제외하며, 개별 정보 조각 중 `relevance_score > 0.7`이고 `faithfulness_score > 0.7`인 것만 선별한다.
# 각 법률별 State는 기본 CorrectiveRagState를 상속
# CorrectiveRagState: [사용자 정의 - 섹션 5.1 참조] question, generation, documents, num_generations 필드 포함
class PersonalRagState(CorrectiveRagState):
rewritten_query: str # 재작성한 질문
extracted_info: Optional[List[InformationStrip]] # [섹션 5.1 참조] 추출된 정보 조각 리스트
node_answer: Optional[str] # 에이전트 답변
- 예상 결과: `PersonalRagState`는 `CorrectiveRagState`의 4개 필드에 `rewritten_query`, `extracted_info`, `node_answer`를 추가하여 총 7개 필드를 가진다.
- 참고: 아래 코드에서 사용하는 주요 변수/클래스:
- `CorrectiveRagState`: 섹션 5.1에서 정의한 기본 State (question, generation, documents, num_generations)
- `InformationStrip`, `ExtractedInformation`, `RefinedQuestion`: 섹션 5.1에서 정의한 Pydantic v2 모델
- `llm`: 섹션 4.2에서 `ChatOpenAI(model="gpt-4o-mini")`로 초기화한 LLM 인스턴스
- `personal_law_search`: 섹션 4.1에서 `@tool`로 정의한 개인정보보호법 검색 도구
from pydantic import BaseModel, Field # [Pydantic v2] 데이터 검증
from typing import List, TypedDict, Annotated, Optional
from operator import add # [Python 표준 라이브러리] 리스트 합산 리듀서용
from langchain_core.documents import Document # [LangChain 핵심] 문서 객체
from langchain_core.prompts import ChatPromptTemplate # [LangChain 핵심] 프롬프트 템플릿 - 시스템/사용자 메시지를 구조화
from typing import Literal
# [사용자 정의] State 정의 - CorrectiveRagState(섹션 5.1)를 상속하여 개인정보보호법 전용 필드 추가
class PersonalRagState(CorrectiveRagState):
rewritten_query: str # 쿼리 재작성 결과 (rewrite_query 노드에서 설정)
extracted_info: Optional[List[InformationStrip]] # [섹션 5.1 참조] 추출된 정보 조각 리스트 (extract_and_evaluate 노드에서 설정)
node_answer: Optional[str] # 이 에이전트의 최종 답변 (generate_answer 노드에서 설정)
# --- 노드 함수 정의 ---
def retrieve_documents(state: PersonalRagState) -> PersonalRagState:
"""문서 검색 노드: 재작성된 쿼리가 있으면 그것을 사용, 없으면 원본 질문 사용"""
print("---문서 검색---")
# state.get("rewritten_query", ...): 쿼리 재작성 후 재검색 시에는 개선된 쿼리를 사용
query = state.get("rewritten_query", state["question"])
# personal_law_search: [사용자 정의 - 섹션 4.1 참조] @tool로 정의한 개인정보보호법 검색 도구
# invoke()는 내부적으로 personal_db_retriever(리랭킹 포함)를 호출하여 List[Document]를 반환
docs = personal_law_search.invoke(query)
return {"documents": docs}
def extract_and_evaluate_information(state: PersonalRagState) -> PersonalRagState:
"""정보 추출 및 평가 노드: 각 문서에서 관련 정보를 추출하고 점수를 매김"""
print("---정보 추출 및 평가---")
extracted_strips = []
for doc in state["documents"]:
# ChatPromptTemplate.from_messages(): [LangChain 핵심] 시스템/사용자 메시지를 구조화된 프롬프트로 변환
extract_prompt = ChatPromptTemplate.from_messages([
("system", """당신은 개인정보보호법 전문가입니다. 주어진 문서에서 질문과 관련된 주요 사실과 정보를 3~5개 정도 추출하세요.
각 추출된 정보에 대해 다음 두 가지 측면을 0에서 1 사이의 점수로 평가하세요:
1. 질문과의 관련성
2. 답변의 충실성 (질문에 대한 완전하고 정확한 답변을 제공할 수 있는 정도)
마지막으로, 추출된 정보를 종합하여 질문에 대한 전반적인 답변 가능성을 0에서 1 사이의 점수로 평가하세요."""),
("human", "[질문]\n{question}\n\n[문서 내용]\n{document_content}")
])
# llm: [사용자 정의 - 섹션 4.2 참조] ChatOpenAI(model="gpt-4o-mini") 인스턴스
# with_structured_output(): [LangChain 핵심] Pydantic v2 모델에 맞게 구조화된 출력을 반환하도록 설정
# ExtractedInformation: [사용자 정의 - 섹션 5.1 참조] 정보 추출용 Pydantic v2 모델
extract_llm = llm.with_structured_output(ExtractedInformation)
extracted_data = extract_llm.invoke(extract_prompt.format(
question=state["question"],
document_content=doc.page_content
))
# 전반적 관련성이 0.8 미만인 문서는 제외 - 품질 미달 문서 필터링
if extracted_data.query_relevance < 0.8:
continue
# 개별 정보 조각 중 관련성과 충실성이 모두 0.7 이상인 것만 선별
for strip in extracted_data.strips:
if strip.relevance_score > 0.7 and strip.faithfulness_score > 0.7:
extracted_strips.append(strip)
return {
"extracted_info": extracted_strips,
"num_generations": state.get("num_generations", 0) + 1
}
def rewrite_query(state: PersonalRagState) -> PersonalRagState:
"""쿼리 재작성 노드: 추출된 정보를 바탕으로 더 효과적인 검색 쿼리를 생성"""
print("---쿼리 재작성---")
rewrite_prompt = ChatPromptTemplate.from_messages([
("system", """당신은 개인정보보호법 전문가입니다. 주어진 원래 질문과 추출된 정보를 바탕으로,
더 관련성 있고 충실한 정보를 찾기 위해 검색 쿼리를 개선해주세요.
개인정보보호법과 관련된 전문 용어를 적절히 활용하세요."""),
("human", "원래 질문: {question}\n\n추출된 정보:\n{extracted_info}")
])
# state["extracted_info"]: extract_and_evaluate_information 노드에서 설정한 InformationStrip 리스트
extracted_info_str = "\n".join([strip.content for strip in state["extracted_info"]])
# llm: [사용자 정의 - 섹션 4.2 참조] ChatOpenAI 인스턴스
# RefinedQuestion: [사용자 정의 - 섹션 5.1 참조] 개선된 질문과 이유를 담는 Pydantic v2 모델
rewrite_llm = llm.with_structured_output(RefinedQuestion)
response = rewrite_llm.invoke(rewrite_prompt.format(
question=state["question"],
extracted_info=extracted_info_str
))
return {"rewritten_query": response.question_refined}
def generate_node_answer(state: PersonalRagState) -> PersonalRagState:
"""답변 생성 노드: 추출된 정보를 바탕으로 출처 포함 최종 답변 생성"""
print("---답변 생성---")
answer_prompt = ChatPromptTemplate.from_messages([
("system", """당신은 개인정보보호법 전문가입니다. 주어진 질문과 추출된 정보를 바탕으로 답변을 생성해주세요.
답변은 마크다운 형식으로 작성하며, 각 정보의 출처를 명확히 표시해야 합니다.
답변 구조: 1. 직접적인 답변 2. 관련 법률 조항 및 해석 3. 추가 설명 4. 결론 및 요약
각 섹션에서 사용된 정보의 출처를 괄호 안에 명시하세요."""),
("human", "질문: {question}\n\n추출된 정보:\n{extracted_info}")
])
# state["extracted_info"]: extract_and_evaluate_information 노드에서 선별한 InformationStrip 리스트
extracted_info_str = "\n".join([
f"내용: {strip.content}\n출처: {strip.source}\n관련성: {strip.relevance_score}\n충실성: {strip.faithfulness_score}"
for strip in state["extracted_info"]
])
# llm: [사용자 정의 - 섹션 4.2 참조] ChatOpenAI 인스턴스
node_answer = llm.invoke(answer_prompt.format(
question=state["question"],
extracted_info=extracted_info_str
))
return {"node_answer": node_answer.content}
def should_continue(state: PersonalRagState) -> Literal["계속", "종료"]:
"""조건부 엣지: 재시도 2회 이상이거나 관련 정보가 1개 이상이면 종료"""
# num_generations: 재시도 횟수 카운터 (무한 루프 방지)
if state["num_generations"] >= 2:
return "종료"
# extracted_info: 품질 기준을 통과한 정보 조각이 1개 이상이면 답변 생성으로 전환
if len(state["extracted_info"]) >= 1:
return "종료"
return "계속"
# --- 그래프 구성 ---
from langgraph.graph import StateGraph, START, END # [LangGraph 핵심] 그래프 빌더 및 진입/종료점 상수
# StateGraph: [LangGraph 핵심] 상태 기반 그래프 빌더 - PersonalRagState를 상태 스키마로 전달
workflow = StateGraph(PersonalRagState)
# 노드 추가 - [사용자 정의] 위에서 정의한 노드 함수들을 그래프에 등록
workflow.add_node("retrieve", retrieve_documents) # 문서 검색 노드
workflow.add_node("extract_and_evaluate", extract_and_evaluate_information) # 정보 추출/평가 노드
workflow.add_node("rewrite_query", rewrite_query) # 쿼리 재작성 노드
workflow.add_node("generate_answer", generate_node_answer) # 답변 생성 노드
# 엣지 추가
workflow.add_edge(START, "retrieve") # START → retrieve: 그래프 시작 시 문서 검색부터 실행
workflow.add_edge("retrieve", "extract_and_evaluate") # retrieve → extract_and_evaluate: 검색 후 평가
# 조건부 엣지: 정보 품질에 따라 답변 생성 또는 쿼리 재작성
# should_continue: [사용자 정의] "계속"(재검색) 또는 "종료"(답변 생성) 반환
workflow.add_conditional_edges(
"extract_and_evaluate",
should_continue,
{
"계속": "rewrite_query", # 정보 부족 → 쿼리 재작성 → 재검색
"종료": "generate_answer" # 정보 충분 → 답변 생성
}
)
workflow.add_edge("rewrite_query", "retrieve") # 쿼리 재작성 후 다시 검색
workflow.add_edge("generate_answer", END) # 답변 생성 후 그래프 종료
# 그래프 컴파일 - compile()은 StateGraph를 실행 가능한 CompiledGraph로 변환
personal_law_agent = workflow.compile()
- 실제 출력 결과:
inputs = {"question": "개인정보 처리에 대한 동의를 받을 때 주의해야 할 점은 무엇인가요?"}
for output in personal_law_agent.stream(inputs):
for key, value in output.items():
pprint(f"Node '{key}':")
print("----------------------------------------------------------")---문서 검색---
Node 'retrieve':
----------------------------------------------------------
---정보 추출 및 평가---
Node 'extract_and_evaluate':
----------------------------------------------------------
---답변 생성---
Node 'generate_answer':
----------------------------------------------------------
- 관련성 높은 정보가 충분히 추출되면 쿼리 재작성 없이 바로 답변이 생성된다.
5.3. 근로기준법 / 주택임대차보호법 / 웹 검색 RAG 에이전트
- 나머지 3개 에이전트는 개인정보보호법 에이전트와 동일한 그래프 구조를 따르며, 검색 도구와 시스템 프롬프트의 전문 분야만 변경한다.
- 각 에이전트는 독립적인 State 클래스를 가지고, 독립적인 StateGraph로 컴파일된다.
- 노드 함수 내부에서 사용하는 검색 도구(`labor_law_search`, `housing_law_search`, `web_search`)와 프롬프트의 전문 분야("근로기준법 전문가", "주택임대차보호법 전문가", "인터넷 정보 검색 전문가")만 다르다.
# --- 근로기준법 에이전트 ---
# CorrectiveRagState: [사용자 정의 - 섹션 5.1 참조] 기본 RAG State (question, generation, documents, num_generations)
# InformationStrip: [사용자 정의 - 섹션 5.1 참조] 추출된 정보 조각 Pydantic v2 모델
class LaborRagState(CorrectiveRagState):
rewritten_query: str
extracted_info: Optional[List[InformationStrip]]
node_answer: Optional[str]
# 노드 함수 정의 (retrieve_documents에서 labor_law_search 사용)
# labor_law_search: [사용자 정의 - 섹션 4.1 참조] @tool로 정의한 근로기준법 검색 도구
# 프롬프트에서 "개인정보보호법 전문가" → "근로기준법 전문가"로 변경
# (동일 구조이므로 전체 코드 생략, 검색 도구와 전문 분야만 상이)
# StateGraph: [LangGraph 핵심] 상태 기반 그래프 빌더
workflow = StateGraph(LaborRagState)
workflow.add_node("retrieve", retrieve_documents)
workflow.add_node("extract_and_evaluate", extract_and_evaluate_information)
workflow.add_node("rewrite_query", rewrite_query)
workflow.add_node("generate_answer", generate_node_answer)
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "extract_and_evaluate")
workflow.add_conditional_edges("extract_and_evaluate", should_continue, {"계속": "rewrite_query", "종료": "generate_answer"})
workflow.add_edge("rewrite_query", "retrieve")
workflow.add_edge("generate_answer", END)
labor_law_agent = workflow.compile()
# --- 주택임대차보호법 에이전트 ---
class HousingRagState(CorrectiveRagState):
rewritten_query: str
extracted_info: Optional[List[InformationStrip]]
node_answer: Optional[str]
# (동일 구조, housing_law_search 사용, "주택임대차보호법 전문가" 프롬프트)
# housing_law_search: [사용자 정의 - 섹션 4.1 참조] @tool로 정의한 주택임대차보호법 검색 도구
housing_law_agent = workflow.compile()
# --- 웹 검색 에이전트 ---
class SearchRagState(CorrectiveRagState):
rewritten_query: str
extracted_info: Optional[List[InformationStrip]]
node_answer: Optional[str]
# (동일 구조, web_search 사용, "인터넷 정보 검색 전문가" 프롬프트)
# web_search: [사용자 정의 - 섹션 4.1 참조] @tool로 정의한 웹 검색 도구 (TavilySearch 기반)
search_web_agent = workflow.compile()
- 4개 에이전트의 구조 비교:
| 에이전트 | State 클래스 | 검색 도구 | 프롬프트 전문 분야 | 컴파일 결과 |
| 개인정보보호법 | PersonalRagState | personal_law_search | 개인정보보호법 전문가 | personal_law_agent |
| 근로기준법 | LaborRagState | labor_law_search | 근로기준법 전문가 | labor_law_agent |
| 주택임대차보호법 | HousingRagState | housing_law_search | 주택임대차보호법 전문가 | housing_law_agent |
| 웹 검색 | SearchRagState | web_search | 인터넷 정보 검색 전문가 | search_web_agent |
- 4개 CRAG 에이전트 구성 비교
- 4개 에이전트는 모두 동일한 Corrective RAG 그래프 구조(retrieve → extract_and_evaluate → 조건부 분기 → generate_answer)를 공유하며, State 클래스명, 프롬프트의 전문 분야, 검색 도구만 다르다.
- 실선 화살표(→): 에이전트 내부 노드 간 실행 순서를 나타낸다.
- 점선 화살표(-.->): 각 에이전트가 사용하는 검색 도구와 데이터 소스 간의 의존성을 나타낸다. 법률 에이전트 3종은 Chroma 벡터저장소(CrossEncoderReranker 리랭킹 포함)를, 웹 검색 에이전트는 TavilySearch API를 사용한다.
6. 단계 4: 질문 라우팅 (Adaptive RAG + 병렬 에이전트 팬아웃/팬인)
- 이 단계는 사용자의 질문을 분석하여 적절한 법률 에이전트로 자동 라우팅하는 Adaptive RAG 패턴을 구현한다.
- 05_LangGraph_AgenticRAG에서 학습한 Adaptive RAG의 핵심 개념인 질문 분류 → 동적 경로 결정을 적용하며, 서브그래프의 팬아웃/팬인 메커니즘을 통해 여러 에이전트를 병렬로 실행할 수 있다.
- 예를 들어 "근로계약 체결할 때 개인정보 취급 상의 유의사항"처럼 두 법률에 걸치는 질문은 근로기준법 에이전트와 개인정보보호법 에이전트가 동시에 호출된다.
- 설계 고려사항:
- `ToolSelector` Pydantic v2 모델을 사용하여 LLM이 구조화된 형태로 적절한 도구를 선택하도록 한다.
- `ToolSelectors`는 복수의 `ToolSelector`를 담을 수 있어, 하나의 질문에 여러 도구가 선택될 수 있다.
- `ResearchAgentState`의 `answers` 필드에 `Annotated[List[str], add]` 리듀서를 적용하여, 병렬 에이전트들의 답변이 자동으로 합산된다.
- 메인 리서치 에이전트 그래프 흐름
6.1. 메인 그래프 State 정의
- `ResearchAgentState`는 전체 파이프라인의 메인 State로, 병렬 에이전트의 답변을 합산하는 `Annotated[List[str], add]` 리듀서가 핵심이다.
- `add` 리듀서는 03_LangGraph_MessageGraph_Reducer에서 학습한 것처럼, 여러 노드에서 반환되는 리스트를 자동으로 병합한다.
- 참고: `Annotated[List[str], add]`의 `add`는 Python 표준 라이브러리 `operator.add`로, LangGraph의 리듀서(Reducer)로 사용됩니다. 리듀서가 `add`이면 여러 노드가 `answers` 필드에 값을 반환할 때 덮어쓰기 대신 리스트 합산(append)이 됩니다. 예: 에이전트A가 `["답변1"]`, 에이전트B가 `["답변2"]`를 반환하면 → `answers = ["답변1", "답변2"]`로 자동 병합. 이 State는 섹션 6~9 전체에서 메인 그래프의 상태로 사용됩니다.
from typing import Annotated
from operator import add # [Python 표준 라이브러리] 리스트 합산 리듀서 - 병렬 에이전트 답변 병합에 사용
# [사용자 정의] 메인 그래프 State - 섹션 6~9 전체에서 사용
# 이 State는 질문 라우팅, 병렬 에이전트 실행, 최종 답변 생성, 답변 평가, HITL까지 전 과정의 상태를 관리한다.
class ResearchAgentState(TypedDict):
question: str # 사용자 질문
answers: Annotated[List[str], add] # 병렬 에이전트 답변 합산 (add 리듀서로 자동 병합)
final_answer: str # 최종 통합 답변
datasources: List[str] # 질문 라우터가 선택한 데이터 소스 목록
evaluation_report: Optional[dict] # 섹션 7의 ReAct 에이전트 평가 결과
6.2. 질문 라우터 (ToolSelector)
from typing import Literal
from langchain_core.prompts import ChatPromptTemplate # [LangChain 핵심] 프롬프트 템플릿
from pydantic import BaseModel, Field # [Pydantic v2] 데이터 검증 및 구조화 출력 스키마 정의
# 라우팅 결정을 위한 Pydantic v2 모델
# LLM이 with_structured_output()를 통해 이 스키마에 맞는 구조화된 JSON을 반환한다.
class ToolSelector(BaseModel):
"""Routes the user question to the most appropriate tool."""
tool: Literal["search_personal", "search_labor", "search_housing", "search_web"] = Field(
description="Select one of the tools, based on the user's question.",
)
class ToolSelectors(BaseModel):
"""Select the appropriate tools that are suitable for the user question."""
tools: List[ToolSelector] = Field(
description="Select one or more tools, based on the user's question.",
)
# 구조화된 출력을 위한 LLM 설정
# llm: [사용자 정의 - 섹션 4.2 참조] ChatOpenAI(model="gpt-4o-mini") 인스턴스
# with_structured_output(ToolSelectors): LLM이 ToolSelectors Pydantic v2 모델 형태로 응답하도록 설정
structured_llm_tool_selector = llm.with_structured_output(ToolSelectors)
# 라우팅 프롬프트 - LLM이 질문을 분석하여 적절한 도구를 선택하도록 안내
# dedent: [사용자 정의 - 섹션 1.2 참조] textwrap.dedent로 멀티라인 문자열의 들여쓰기 제거
system = dedent("""You are an AI assistant specializing in routing user questions to the appropriate tools.
Use the following guidelines:
- For questions specifically about legal provisions or articles of the privacy protection law (개인정보 보호법), use the search_personal tool.
- For questions specifically about legal provisions or articles of the labor law (근로기준법), use the search_labor tool.
- For questions specifically about legal provisions or articles of the housing law (주택임대차보호법), use the search_housing tool.
- For any other information, including questions related to these laws but not directly about specific legal provisions, or for the most up-to-date data, use the search_web tool.
Always choose all of the appropriate tools based on the user's question.
If a question is about a law but doesn't seem to be asking about specific legal provisions, include both the relevant law search tool and the search_web tool.""")
route_prompt = ChatPromptTemplate.from_messages([
("system", system),
("human", "{question}"),
])
# 질문 라우터 체인: 프롬프트 → 구조화 출력 LLM → ToolSelectors 객체 반환
question_tool_router = route_prompt | structured_llm_tool_selector
- 실제 출력 결과:
print(question_tool_router.invoke({"question": "근로계약 체결할 때 개인정보 취급 상의 유의사항은 무엇인가요?"}))
# ToolSelectors(tools=[ToolSelector(tool='search_labor'), ToolSelector(tool='search_personal')])
print(question_tool_router.invoke({"question": "법에서 정한 연차휴가 기준을 알려주세요."}))
# ToolSelectors(tools=[ToolSelector(tool='search_labor')])
print(question_tool_router.invoke({"question": "개인정보보호법에서 정한 가명정보의 정의는 무엇인가요?"}))
# ToolSelectors(tools=[ToolSelector(tool='search_personal')])
6.3. 라우팅 노드 및 에이전트 노드 정의
from langchain_core.output_parsers import StrOutputParser # [LangChain 핵심] LLM 출력을 문자열로 변환하는 파서
# 질문 분석 노드: 질문을 분석하여 적절한 데이터 소스를 선택
# ResearchAgentState: [사용자 정의 - 섹션 6.1 참조] 메인 그래프 State
def analyze_question_tool_search(state: ResearchAgentState):
question = state["question"]
# question_tool_router: [사용자 정의 - 섹션 6.2 참조] route_prompt | structured_llm_tool_selector 체인
# ToolSelectors 객체를 반환하며, .tools 속성에 선택된 도구 리스트가 담긴다.
result = question_tool_router.invoke({"question": question})
datasources = [tool.tool for tool in result.tools]
return {"datasources": datasources}
# 라우팅 함수: 선택된 데이터 소스를 리스트로 반환 (팬아웃)
# 이 함수의 반환값이 리스트이면 LangGraph가 자동으로 팬아웃(병렬 실행)을 수행한다.
def route_datasources_tool_search(state: ResearchAgentState) -> List[str]:
datasources = set(state['datasources'])
valid_sources = {"search_personal", "search_labor", "search_housing", "search_web"}
if not datasources:
return ["llm_fallback"] # 도구 선택 없음 → 일반 LLM 답변
if datasources.issubset(valid_sources):
return list(datasources) # 유효한 도구만 선택된 경우
return list(valid_sources) # 예외: 모든 도구 실행
# 각 법률별 에이전트 노드 (서브그래프 호출)
# 각 노드는 컴파일된 서브그래프(CRAG 에이전트)를 invoke()로 호출하고,
# 결과를 ResearchAgentState의 answers 필드에 리스트로 반환하여 add 리듀서로 병합되도록 한다.
def personal_rag_node(state: PersonalRagState, input=ResearchAgentState) -> ResearchAgentState:
"""개인정보보호법 CRAG 서브그래프를 호출하는 노드"""
print("--- 개인정보보호법 전문가 에이전트 시작 ---")
question = state["question"]
# personal_law_agent: [사용자 정의 - 섹션 5.2 참조] 개인정보보호법 Corrective RAG 서브그래프
answer = personal_law_agent.invoke({"question": question})
return {"answers": [answer["node_answer"]]}
def labor_rag_node(state: LaborRagState, input=ResearchAgentState) -> ResearchAgentState:
"""근로기준법 CRAG 서브그래프를 호출하는 노드"""
print("--- 근로기준법 전문가 에이전트 시작 ---")
question = state["question"]
# labor_law_agent: [사용자 정의 - 섹션 5.3 참조] 근로기준법 Corrective RAG 서브그래프
answer = labor_law_agent.invoke({"question": question})
return {"answers": [answer["node_answer"]]}
def housing_rag_node(state: HousingRagState, input=ResearchAgentState) -> ResearchAgentState:
"""주택임대차보호법 CRAG 서브그래프를 호출하는 노드"""
print("--- 주택임대차보호법 전문가 에이전트 시작 ---")
question = state["question"]
# housing_law_agent: [사용자 정의 - 섹션 5.3 참조] 주택임대차보호법 Corrective RAG 서브그래프
answer = housing_law_agent.invoke({"question": question})
return {"answers": [answer["node_answer"]]}
def web_rag_node(state: SearchRagState, input=ResearchAgentState) -> ResearchAgentState:
"""웹 검색 CRAG 서브그래프를 호출하는 노드"""
print("--- 인터넷 검색 전문가 에이전트 시작 ---")
question = state["question"]
# search_web_agent: [사용자 정의 - 섹션 5.3 참조] 웹 검색 Corrective RAG 서브그래프
answer = search_web_agent.invoke({"question": question})
return {"answers": [answer["node_answer"]]}
6.4. 최종 답변 생성 및 LLM Fallback 노드
# RAG 프롬프트: 병렬 에이전트의 답변들을 종합하여 최종 답변 생성
# answers 필드에 add 리듀서로 병합된 여러 에이전트의 답변을 하나의 통합 답변으로 정리한다.
rag_prompt = ChatPromptTemplate.from_messages([
("system", """You are an assistant answering questions based on provided documents. Follow these guidelines:
1. Use only information from the given documents.
2. If the document lacks relevant info, say "제공된 정보로는 충분한 답변을 할 수 없습니다."
3. Cite the source of information for each sentence in your answer.
4. Don't speculate or add information not in the documents.
5. Keep answers concise and clear.
6. If multiple sources provide the same information, cite all relevant sources.
7. If information comes from multiple sources, combine them coherently while citing each source."""),
("human", "Answer the following question using these documents:\n\n[Documents]\n{documents}\n\n[Question]\n{question}"),
])
def answer_final(state: ResearchAgentState) -> ResearchAgentState:
"""병렬 에이전트의 답변들을 종합하여 최종 답변을 생성하는 노드"""
print("---최종 답변---")
question = state["question"]
# state["answers"]: [ResearchAgentState - 섹션 6.1 참조] add 리듀서로 병합된 병렬 에이전트의 답변 리스트
documents = state.get("answers", [])
if not isinstance(documents, list):
documents = [documents]
documents_text = "\n\n".join(documents)
# rag_prompt | llm | StrOutputParser(): 프롬프트 → LLM → 문자열 변환 체인
# llm: [사용자 정의 - 섹션 4.2 참조] ChatOpenAI(model="gpt-4o-mini") 인스턴스
rag_chain = rag_prompt | llm | StrOutputParser()
generation = rag_chain.invoke({"documents": documents_text, "question": question})
return {"final_answer": generation, "question": question}
# LLM Fallback: 법률과 무관한 일반 질문에 대한 직접 답변
# 질문 라우터가 어떤 도구도 선택하지 않았을 때 호출된다.
fallback_prompt = ChatPromptTemplate.from_messages([
("system", """You are an AI assistant helping with various topics. Follow these guidelines:
1. Provide accurate and helpful information to the best of your ability.
2. Express uncertainty when unsure; avoid speculation.
3. Keep answers concise yet informative."""),
("human", "{question}"),
])
def llm_fallback(state: ResearchAgentState) -> ResearchAgentState:
"""검색 없이 LLM이 직접 답변하는 Fallback 노드"""
print("---Fallback 답변---")
question = state["question"]
llm_chain = fallback_prompt | llm | StrOutputParser()
generation = llm_chain.invoke({"question": question})
return {"final_answer": generation, "question": question}
6.5. 메인 그래프 구성
from langgraph.graph import StateGraph, START, END # [LangGraph 핵심] 그래프 빌더 및 진입/종료점 상수
# 노드 정의를 딕셔너리로 관리
# 각 노드 함수의 출처: analyze_question(섹션 6.3), search_*(섹션 6.3), generate_answer(섹션 6.4), llm_fallback(섹션 6.4)
nodes = {
"analyze_question": analyze_question_tool_search,
"search_personal": personal_rag_node,
"search_labor": labor_rag_node,
"search_housing": housing_rag_node,
"search_web": web_rag_node,
"generate_answer": answer_final,
"llm_fallback": llm_fallback
}
# 그래프 생성
# ResearchAgentState: [사용자 정의 - 섹션 6.1 참조] 메인 그래프 State
search_builder = StateGraph(ResearchAgentState)
# 노드 추가
for node_name, node_func in nodes.items():
search_builder.add_node(node_name, node_func)
# 엣지 추가: 질문 분석 후 조건부 팬아웃
search_builder.add_edge(START, "analyze_question")
# add_conditional_edges: route_datasources_tool_search가 반환하는 리스트의 각 원소에 해당하는 노드들이 병렬 실행됨
# route_datasources_tool_search: [사용자 정의 - 섹션 6.3 참조] 라우팅 함수 (리스트 반환 시 팬아웃)
search_builder.add_conditional_edges(
"analyze_question",
route_datasources_tool_search,
["search_personal", "search_labor", "search_housing", "search_web", "llm_fallback"]
)
# 검색 에이전트 노드들을 최종 답변 생성에 연결 (팬인)
# 병렬로 실행된 에이전트 노드들의 결과가 answers 필드에 add 리듀서로 합산된 후 generate_answer에서 종합된다.
for node in ["search_personal", "search_labor", "search_housing", "search_web"]:
search_builder.add_edge(node, "generate_answer")
search_builder.add_edge("generate_answer", END)
search_builder.add_edge("llm_fallback", END)
# 그래프 컴파일
rag_search_graph = search_builder.compile()
- 실제 출력 결과:
inputs = {"question": "대리인과 아파트 임대차 계약을 체결할 때 주의해야 할 점은 무엇인가요?"}
for output in rag_search_graph.stream(inputs):
for key, value in output.items():
pprint(f"Node '{key}':")
print("----------------------------------------------------------")Node 'analyze_question':
----------------------------------------------------------
--- 주택임대차보호법 전문가 에이전트 시작 ---
---문서 검색---
---정보 추출 및 평가---
---답변 생성---
--- 인터넷 검색 전문가 에이전트 시작 ---
---문서 검색---
---정보 추출 및 평가---
---답변 생성---
Node 'search_housing':
Node 'search_web':
----------------------------------------------------------
---최종 답변---
Node 'generate_answer':
----------------------------------------------------------
- 주택임대차 관련 질문이므로 `search_housing`과 `search_web` 두 에이전트가 병렬로 실행되고, 두 답변이 합산되어 최종 답변이 생성된다.
7. 단계 5: 답변 평가 (ReAct 에이전트)
- 이 단계는 생성된 답변의 품질을 자동으로 평가하는 ReAct 에이전트를 구현한다.
- 04_LangGraph_ReAct_Memory에서 학습한 `create_react_agent`를 사용하여, 평가 중 필요한 경우 법률 검색이나 웹 검색 도구를 활용하여 답변의 정확성을 검증할 수 있는 에이전트를 구축한다.
- 설계 고려사항:
- 평가 기준은 6개 항목(정확성, 관련성, 완전성, 인용 정확성, 명확성/간결성, 객관성)이며, 각 10점 만점으로 총 60점이다.
- 평가 에이전트는 기존의 4가지 검색 도구를 모두 사용할 수 있어, 답변에 인용된 법률 조항의 정확성을 직접 검색하여 검증할 수 있다.
- 평가 결과는 JSON 형태로 구조화되어 후속 HITL 단계에서 사용자에게 점수와 평가 설명을 표시할 수 있다.
from langgraph.prebuilt import create_react_agent # [LangGraph 내장] ReAct 에이전트 생성 함수 - 도구 호출과 추론을 자동으로 반복하는 에이전트
# 평가용 시스템 프롬프트
evaluation_prompt = """당신은 AI 어시스턴트가 생성한 답변을 평가하는 전문가입니다.
주어진 질문과 답변을 평가하고, 60점 만점으로 점수를 매기세요.
평가 기준:
1. 정확성 (10점)
2. 관련성 (10점)
3. 완전성 (10점)
4. 인용 정확성 (10점)
5. 명확성과 간결성 (10점)
6. 객관성 (10점)
필요한 경우, 도구를 사용하여 추가 정보를 수집하세요:
- web_search, personal_law_search, labor_law_search, housing_law_search
출력 형식:
{
"scores": { "accuracy": 0, "relevance": 0, ... },
"total_score": 0,
"brief_evaluation": "간단한 평가 설명"
}"""
# [LangGraph 내장] create_react_agent: ReAct 패턴 에이전트를 자동 생성
# llm: [사용자 정의 - 섹션 4.2 참조] ChatOpenAI(model="gpt-4o-mini") 인스턴스
# tools: [사용자 정의 - 섹션 4.1 참조] [personal_law_search, labor_law_search, housing_law_search, web_search] 도구 리스트
answer_reviewer = create_react_agent(
llm,
tools=tools, # 기존 4가지 검색 도구 활용 (답변 검증 시 사용)
prompt=evaluation_prompt, # 평가 전용 시스템 프롬프트
)
- 예상 결과: `answer_reviewer.invoke({"messages": [HumanMessage(...)]})` 호출 시, ReAct 에이전트가 필요에 따라 도구를 호출하며 답변을 검증하고, 최종적으로 JSON 형태의 평가 결과를 반환한다.
from textwrap import dedent # [Python 표준 라이브러리] 멀티라인 문자열 들여쓰기 제거
from langgraph.prebuilt import create_react_agent # [LangGraph 내장] ReAct 에이전트 생성 함수
from langchain_core.messages import HumanMessage # [LangChain 핵심] 사용자 메시지 객체
# 상세 평가 프롬프트
evaluation_prompt = dedent("""
당신은 AI 어시스턴트가 생성한 답변을 평가하는 전문가입니다. 주어진 질문과 답변을 평가하고, 60점 만점으로 점수를 매기세요. 다음 기준을 사용하여 평가하십시오:
1. 정확성 (10점)
2. 관련성 (10점)
3. 완전성 (10점)
4. 인용 정확성 (10점)
5. 명확성과 간결성 (10점)
6. 객관성 (10점)
평가 과정:
1. 주어진 질문과 답변을 주의 깊게 읽으십시오.
2. 필요한 경우, 다음 도구를 사용하여 추가 정보를 수집하세요:
- web_search: 웹 검색
- personal_law_search: 개인정보보호법 검색
- labor_law_search: 근로기준법 검색
- housing_law_search: 주택임대차보호법 검색
3. 각 기준에 대해 1-10점 사이의 점수를 매기세요.
4. 총점을 계산하세요 (60점 만점).
출력 형식:
{
"scores": {
"accuracy": 0,
"relevance": 0,
"completeness": 0,
"citation_accuracy": 0,
"clarity_conciseness": 0,
"objectivity": 0
},
"total_score": 0,
"brief_evaluation": "간단한 평가 설명"
}
""")
# ReAct 평가 에이전트 생성
# llm: [사용자 정의 - 섹션 4.2 참조] ChatOpenAI(model="gpt-4o-mini") 인스턴스
# tools: [사용자 정의 - 섹션 4.1 참조] 4가지 검색 도구 리스트
answer_reviewer = create_react_agent(
llm,
tools=tools,
prompt=evaluation_prompt,
)
# 평가 실행 예시
# rag_search_graph: [사용자 정의 - 섹션 6.5 참조] 메인 RAG 그래프 (질문 분석 → 병렬 CRAG → 최종 답변)
inputs = {"question": "대리인과 아파트 임대차 계약을 체결할 때 주의해야 할 점은 무엇인가요?"}
result_state = rag_search_graph.invoke(inputs)
# HumanMessage: [LangChain 핵심] ReAct 에이전트의 입력 메시지 형식
messages = [HumanMessage(content=f"[질문]\n{inputs['question']}\n\n[답변]\n{result_state['final_answer']}")]
result = answer_reviewer.invoke({"messages": messages})
# json.loads(): ReAct 에이전트의 최종 메시지를 JSON으로 파싱하여 평가 결과를 구조화
evaluation = json.loads(result['messages'][-1].content)
print(f"총점: {evaluation['total_score']}/60")
print(f"평가: {evaluation['brief_evaluation']}")
- 실제 출력 결과:
총점: 48/60
평가: 답변은 주택임대차보호법의 관련 조항을 적절히 인용하며 대리인과의
계약 시 주의사항을 설명하고 있습니다...
- ReAct 답변 평가 에이전트 흐름
8. 단계 6: HITL 통합 (Human-in-the-Loop)
- 이 단계는 답변 평가 후 사용자의 최종 승인/거부를 받는 Human-in-the-Loop 메커니즘을 구현한다.
- 최신 LangGraph API에서는 `interrupt()` 함수와 `Command(resume=...)` 패턴을 사용하여 HITL을 구현한다. 이전의 `interrupt_before` + `update_state()` 방식을 대체하는 더 명시적이고 유연한 패턴이다.
- 설계 고려사항:
- `human_review` 노드 내부에서 `interrupt()` 함수를 호출하면 그래프 실행이 자동으로 중단되고, 사용자에게 메시지가 전달된다.
- 사용자가 `Command(resume="yes")` 또는 `Command(resume="no")`로 응답하면 `interrupt()` 함수가 해당 값을 반환하며 노드 실행이 재개된다.
- 이 패턴은 노드 함수 내부에서 직접 사용자 입력을 받을 수 있어, 이전의 빈 함수 + `update_state()` 조합보다 코드가 간결하고 직관적이다.
from langgraph.types import interrupt, Command # [LangGraph 핵심] HITL용 - interrupt: 그래프 실행 중단, Command: 실행 재개
# 답변 평가 노드
# ResearchAgentState: [사용자 정의 - 섹션 6.1 참조] 메인 그래프 State
def evaluate_answer_node(state: ResearchAgentState):
"""답변 품질을 ReAct 에이전트로 평가하는 노드"""
question = state["question"]
final_answer = state["final_answer"]
# answer_reviewer: [사용자 정의 - 섹션 7 참조] create_react_agent로 생성한 ReAct 평가 에이전트
# HumanMessage: [LangChain 핵심] ReAct 에이전트의 입력 메시지 형식
messages = [HumanMessage(content=f"[질문]\n{question}\n\n[답변]\n{final_answer}")]
response = answer_reviewer.invoke({"messages": messages})
response_dict = json.loads(response['messages'][-1].content)
return {"evaluation_report": response_dict, "question": question, "final_answer": final_answer}
# HITL 노드: interrupt()로 실행 중단, Command(resume=...)로 재개
def human_review(state: ResearchAgentState):
"""interrupt() 함수로 그래프 실행을 중단하고 사용자 승인/거부를 받는 HITL 노드"""
# interrupt(): [LangGraph 핵심] 그래프 실행을 중단하고 사용자에게 메시지를 전달
# 전달된 딕셔너리는 그래프의 중단 상태에서 조회할 수 있다.
# Command(resume="yes" 또는 "no")로 재개하면 interrupt()가 해당 값을 반환한다.
answer = state["final_answer"]
report = state.get("evaluation_report", {})
total_score = report.get("total_score", 0)
brief = report.get("brief_evaluation", "평가 없음")
# interrupt() 호출 시 그래프 실행이 중단되고,
# Command(resume=값)으로 재개하면 response 변수에 그 값이 할당된다.
response = interrupt({
"answer": answer,
"total_score": total_score,
"brief_evaluation": brief,
"message": "이 답변을 승인하시겠습니까? Command(resume='yes') 또는 Command(resume='no')로 응답하세요."
})
# response: Command(resume=...)에서 전달된 사용자의 승인/거부 결정
if response == "yes":
return {"final_answer": answer}
else:
# 거부 시 answers를 초기화하여 재검색 시 이전 답변과 중복되지 않도록 함
return {"final_answer": "", "answers": []}
- 예상 결과: 그래프 실행 시 `human_review` 노드에서 `interrupt()`가 호출되면 그래프가 중단된다. `graph.invoke(Command(resume="yes"), config)`로 재개하면 최종 답변이 확정되고, `Command(resume="no")`로 재개하면 `analyze_question`으로 돌아가 전체 파이프라인이 재실행된다.
from langgraph.graph import StateGraph, START, END # [LangGraph 핵심] 그래프 빌더 및 진입/종료점 상수
from langgraph.types import interrupt, Command # [LangGraph 핵심] HITL용 interrupt/Command 패턴
# 노드 정의 (evaluate_answer + human_review 노드 추가)
# 각 노드 함수의 출처:
# - analyze_question_tool_search: [섹션 6.3] 질문 분석 및 라우팅
# - personal_rag_node 등: [섹션 6.3] 법률별 CRAG 서브그래프 호출
# - answer_final: [섹션 6.4] 병렬 답변 종합
# - llm_fallback: [섹션 6.4] 일반 질문 직접 답변
# - evaluate_answer_node: [위 기본 사용법] ReAct 에이전트로 답변 평가
# - human_review: [위 기본 사용법] interrupt()로 사용자 승인/거부 처리
nodes = {
"analyze_question": analyze_question_tool_search,
"search_personal": personal_rag_node,
"search_labor": labor_rag_node,
"search_housing": housing_rag_node,
"search_web": web_rag_node,
"generate_answer": answer_final,
"llm_fallback": llm_fallback,
"evaluate_answer": evaluate_answer_node,
"human_review": human_review,
}
# 그래프 생성
# ResearchAgentState: [사용자 정의 - 섹션 6.1 참조] 메인 그래프 State
search_builder = StateGraph(ResearchAgentState)
# 노드 추가
for node_name, node_func in nodes.items():
search_builder.add_node(node_name, node_func)
# 엣지 추가
search_builder.add_edge(START, "analyze_question")
# route_datasources_tool_search: [사용자 정의 - 섹션 6.3 참조] 라우팅 함수 (리스트 반환 시 팬아웃)
search_builder.add_conditional_edges(
"analyze_question",
route_datasources_tool_search,
["search_personal", "search_labor", "search_housing", "search_web", "llm_fallback"]
)
# 팬인: 병렬 에이전트 → 최종 답변 생성
for node in ["search_personal", "search_labor", "search_housing", "search_web"]:
search_builder.add_edge(node, "generate_answer")
# 답변 생성 → 답변 평가 → HITL 분기
search_builder.add_edge("generate_answer", "evaluate_answer")
search_builder.add_edge("evaluate_answer", "human_review")
# human_review 이후 조건부 엣지:
# human_review 노드에서 interrupt() 후 Command(resume=...)로 재개되면,
# 반환된 state의 final_answer가 비어있으면 거부(재검색), 있으면 승인(종료)
search_builder.add_conditional_edges(
"human_review",
lambda x: "approved" if x.get("final_answer") else "rejected",
{
"approved": END, # 승인 → 종료
"rejected": "analyze_question" # 거부 → 재검색
}
)
search_builder.add_edge("llm_fallback", END)
# 그래프 컴파일 - MemorySaver로 체크포인트 활성화 (interrupt() 사용 시 필수)
from langgraph.checkpoint.memory import MemorySaver # [LangGraph 핵심] 메모리 기반 체크포인터
memory = MemorySaver()
legal_rag_agent = search_builder.compile(checkpointer=memory)
- 실제 출력 결과:
# 첫 번째 실행: interrupt()에서 중단됨
config = {"configurable": {"thread_id": "test-thread-1"}}
inputs = {"question": "대리인과 아파트 임대차 계약을 체결할 때 주의해야 할 점은 무엇인가요?"}
result = legal_rag_agent.invoke(inputs, config=config)
# → interrupt()에서 중단됨. result에 중단 상태가 포함됨.Node 'analyze_question':
----------------------------------------------------------
--- 주택임대차보호법 전문가 에이전트 시작 ---
--- 인터넷 검색 전문가 에이전트 시작 ---
Node 'search_housing':
Node 'search_web':
----------------------------------------------------------
---최종 답변---
Node 'generate_answer':
----------------------------------------------------------
Node 'evaluate_answer':
----------------------------------------------------------# 승인: Command(resume="yes")로 재개
result = legal_rag_agent.invoke(Command(resume="yes"), config=config)
print(result["final_answer"])
# → 최종 답변이 출력됨
# 또는 거부: Command(resume="no")로 재개 → analyze_question부터 재실행
# result = legal_rag_agent.invoke(Command(resume="no"), config=config)
- 법률 상담 에이전트 전체 그래프 구조
9. 단계 7: Gradio UI 통합 및 최종 시스템
- 이 단계는 전체 파이프라인을 Gradio 대화형 인터페이스와 통합하여 최종 법률 상담 챗봇을 완성한다.
- 04_LangGraph_ReAct_Memory에서 학습한 MemorySaver로 세션 상태를 유지하고, `interrupt()` 함수 패턴으로 HITL 지점을 설정한다.
- 그래프가 `interrupt()`에서 중단되면 Gradio UI에서 답변과 평가 결과를 표시하고, 사용자의 승인/거부 입력에 따라 `Command(resume=...)` 로 실행을 재개한다.
- 설계 고려사항:
- `human_review` 노드 내부에서 `interrupt()` 함수를 호출하여 그래프 실행을 중단한다.
- 중단 상태에서 `get_state()`로 현재 답변과 평가 결과를 가져와 사용자에게 표시한다.
- 사용자의 응답('y'/'n')에 따라 `Command(resume="yes")` 또는 `Command(resume="no")`로 실행을 재개한다.
- `ChatBot` 클래스의 `user_decision` 플래그로 현재 상태가 "질문 대기 중"인지 "승인/거부 입력 대기 중"인지를 추적한다.
- Gradio HITL 상호작용 시퀀스
9.1. interrupt() 기반 그래프 구성
- 참고: `interrupt()` 함수는 LangGraph의 최신 HITL 패턴으로, 노드 함수 내부에서 호출하면 그래프 실행이 자동 중단됩니다. 이전의 `interrupt_before` + `update_state()` 방식과 달리, 노드 내부에서 직접 사용자 입력값을 받아 처리할 수 있어 코드가 더 간결합니다. `interrupt()` 사용 시 반드시 `checkpointer`(예: `MemorySaver`)가 필요합니다.
from langgraph.checkpoint.memory import MemorySaver # [LangGraph 핵심] 메모리 기반 체크포인터 - 그래프 상태를 메모리에 저장/복원
from langgraph.types import interrupt, Command # [LangGraph 핵심] HITL용 interrupt/Command 패턴
# HITL 노드: interrupt()로 실행 중단, Command(resume=...)로 재개
# ResearchAgentState: [사용자 정의 - 섹션 6.1 참조] 메인 그래프 State
def human_review(state: ResearchAgentState):
"""interrupt() 함수로 그래프 실행을 중단하고 사용자 승인/거부를 받는 HITL 노드.
Gradio UI에서 답변과 평가 결과를 표시한 후, 사용자의 y/n 입력에 따라
Command(resume="yes") 또는 Command(resume="no")로 재개된다."""
answer = state["final_answer"]
report = state.get("evaluation_report", {})
total_score = report.get("total_score", 0)
brief = report.get("brief_evaluation", "평가 없음")
# interrupt(): 그래프 실행을 중단하고 사용자에게 정보를 전달
# Command(resume=값)으로 재개하면 response 변수에 그 값이 할당됨
response = interrupt({
"answer": answer,
"total_score": total_score,
"brief_evaluation": brief,
"message": "이 답변을 승인하시겠습니까? (y/n)"
})
if response == "yes":
return {"final_answer": answer}
else:
return {"final_answer": "", "answers": []}
# 그래프 재구성
# ResearchAgentState: [사용자 정의 - 섹션 6.1 참조] 메인 그래프 State
search_builder = StateGraph(ResearchAgentState)
# 노드 정의
# 각 노드 함수의 출처:
# - analyze_question_tool_search: [섹션 6.3] 질문 분석 및 라우팅
# - personal_rag_node 등: [섹션 6.3] 법률별 CRAG 서브그래프 호출 노드
# - answer_final: [섹션 6.4] 병렬 답변 종합 노드
# - llm_fallback: [섹션 6.4] 일반 질문 직접 답변 노드
# - evaluate_answer_node: [섹션 8 기본 사용법] ReAct 에이전트 답변 평가 노드
# - human_review: [위에서 정의] interrupt() 기반 HITL 노드
nodes = {
"analyze_question": analyze_question_tool_search,
"search_personal": personal_rag_node,
"search_labor": labor_rag_node,
"search_housing": housing_rag_node,
"search_web": web_rag_node,
"generate_answer": answer_final,
"llm_fallback": llm_fallback,
"evaluate_answer": evaluate_answer_node,
"human_review": human_review,
}
for node_name, node_func in nodes.items():
search_builder.add_node(node_name, node_func)
# 엣지 구성
search_builder.add_edge(START, "analyze_question")
# route_datasources_tool_search: [사용자 정의 - 섹션 6.3 참조] 질문 유형별 에이전트 라우팅 (팬아웃)
search_builder.add_conditional_edges(
"analyze_question",
route_datasources_tool_search,
["search_personal", "search_labor", "search_housing", "search_web", "llm_fallback"]
)
# 팬인: 병렬 에이전트 → 최종 답변 생성
for node in ["search_personal", "search_labor", "search_housing", "search_web"]:
search_builder.add_edge(node, "generate_answer")
search_builder.add_edge("generate_answer", "evaluate_answer")
search_builder.add_edge("evaluate_answer", "human_review")
# human_review 이후 조건부 엣지:
# interrupt() 재개 후 human_review가 반환한 state의 final_answer로 승인/거부 판단
search_builder.add_conditional_edges(
"human_review",
lambda x: "approved" if x.get("final_answer") else "rejected",
{
"approved": END,
"rejected": "analyze_question"
}
)
search_builder.add_edge("llm_fallback", END)
# MemorySaver + 그래프 컴파일
# MemorySaver: [LangGraph 핵심] interrupt() 사용 시 필수 - 중단 상태를 메모리에 저장
memory = MemorySaver()
legal_rag_agent = search_builder.compile(checkpointer=memory)
9.2. Gradio ChatBot 클래스
- 참고: 아래 코드의 `legal_rag_agent`는 섹션 9.1에서 `interrupt()` 기반으로 컴파일한 그래프입니다. `ChatBot` 클래스의 `user_decision` 플래그는 `False→True→False` 순환으로 상태를 추적합니다:
- `False`: 새 질문 입력 대기 → `invoke(inputs, config)`로 `interrupt()`까지 실행 → `get_state(config)`로 답변/평가 조회 → `user_decision = True`로 전환
- `True`: 승인/거부 입력 대기 → `invoke(Command(resume="yes"/"no"), config)`로 실행 재개 → `user_decision = False`로 전환
import gradio as gr # [외부 라이브러리] Gradio - 대화형 UI 프레임워크
import uuid # [Python 표준 라이브러리] 고유 thread_id 생성
from typing import List, Tuple
from langgraph.checkpoint.memory import MemorySaver # [LangGraph 핵심] 메모리 기반 체크포인터
from langgraph.types import Command # [LangGraph 핵심] interrupt() 재개용 Command 객체
# 메모리 및 에이전트 재초기화
# search_builder: [사용자 정의 - 섹션 9.1 참조] StateGraph(ResearchAgentState) 빌더
memory = MemorySaver()
# compile()의 checkpointer: interrupt() 상태 저장용
legal_rag_agent = search_builder.compile(checkpointer=memory)
# 예시 질문
example_questions = [
"사업장에서 CCTV를 설치할 때 주의해야 할 법적 사항은 무엇인가요?",
"전월세 계약 갱신 요구권의 행사 기간과 조건은 어떻게 되나요?",
"개인정보 유출 시 기업이 취해야 할 법적 조치는 무엇인가요?",
]
class ChatBot:
def __init__(self):
self.thread_id = str(uuid.uuid4()) # 각 세션의 고유 식별자 - MemorySaver가 이 ID로 상태를 저장/복원
self.user_decision = False # 승인/거부 입력 대기 상태 추적 플래그
def process_message(self, message: str) -> str:
try:
# config: LangGraph가 세션을 식별하기 위한 설정 딕셔너리
# thread_id: 각 대화 세션의 고유 식별자 - MemorySaver가 이 ID로 상태를 저장/복원
config = {"configurable": {"thread_id": self.thread_id}}
if not self.user_decision:
# --- 첫 번째 실행: interrupt()까지 실행 ---
inputs = {"question": message}
# invoke(): 그래프 실행 - human_review 노드의 interrupt()에서 자동 중단
legal_rag_agent.invoke(inputs, config=config)
# get_state(): interrupt()에서 중단된 현재 상태를 조회
current_state = legal_rag_agent.get_state(config)
final_answer = current_state.values.get("final_answer", "No answer available")
evaluation_report = current_state.values.get(
'evaluation_report',
{'total_score': 0, 'brief_evaluation': 'No evaluation available'}
)
response = f"""현재 답변:
{final_answer}
평가 결과:
총점: {evaluation_report.get('total_score', 0)}/60
{evaluation_report.get('brief_evaluation', 'No evaluation available')}
이 답변을 승인하시겠습니까? (y/n): """
self.user_decision = True # 승인/거부 대기 상태로 전환
return response
else:
# --- 두 번째 실행: 사용자 승인/거부 처리 ---
user_decision = message.lower()
if user_decision == 'y':
self.user_decision = False # 상태 초기화
# Command(resume="yes"): interrupt()를 "yes" 값으로 재개
# human_review 노드의 interrupt()가 "yes"를 반환하여 승인 처리됨
legal_rag_agent.invoke(Command(resume="yes"), config=config)
current_state = legal_rag_agent.get_state(config)
return current_state.values.get("final_answer", "No final answer available")
else:
self.user_decision = False # 상태 초기화
# Command(resume="no"): interrupt()를 "no" 값으로 재개
# human_review 노드가 final_answer를 빈 문자열로 반환 → 조건부 엣지에서 "rejected" → analyze_question 재실행
legal_rag_agent.invoke(Command(resume="no"), config=config)
# 재실행 후 다시 interrupt()에서 중단됨
current_state = legal_rag_agent.get_state(config)
final_answer = current_state.values.get("final_answer", "No answer available")
evaluation_report = current_state.values.get(
'evaluation_report',
{'total_score': 0, 'brief_evaluation': 'No evaluation available'}
)
response = f"""다시 생성한 답변:
{final_answer}
평가 결과:
총점: {evaluation_report.get('total_score', 0)}/60
{evaluation_report.get('brief_evaluation', 'No evaluation available')}
이 답변을 승인하시겠습니까? (y/n): """
self.user_decision = True # 다시 승인/거부 대기
return response
except Exception as e:
print(f"Error occurred: {str(e)}")
return "죄송합니다. 응답을 생성하는 동안 오류가 발생했습니다. 다시 시도해 주세요."
def chat(self, message: str, history: List[Tuple[str, str]]) -> str:
print(f"Thread ID: {self.thread_id}")
response = self.process_message(message)
return response
# ChatBot 인스턴스 생성 및 Gradio 앱 실행
chatbot = ChatBot()
demo = gr.ChatInterface(
fn=chatbot.chat,
title="생활법률 AI 어시스턴트",
description="주택임대차보호법, 근로기준법, 개인정보보호법 관련 질문에 답변해 드립니다.",
examples=example_questions,
theme=gr.themes.Soft()
)
demo.launch()
- Gradio 앱 실행 시 `http://127.0.0.1:7860`에서 대화형 법률 상담 인터페이스가 열린다.
- 사용자가 법률 질문을 입력하면 전체 파이프라인(질문 분석 → 법률별 CRAG → 최종 답변 → ReAct 평가)이 실행되고, `interrupt()`에서 중단되어 답변과 평가 점수가 표시된다.
- 사용자가 'y'를 입력하면 `Command(resume="yes")`로 재개되어 최종 답변이 확정되고, 'n'을 입력하면 `Command(resume="no")`로 재개되어 전체 파이프라인이 재실행된다.
- Gradio UI 전체 아키텍처 구성도
- 앱 종료:
demo.close()
'Study > LangChain' 카테고리의 다른 글
| 5. LangGraph - LangGraph Agentic RAG 학습 매뉴얼 (0) | 2026.05.06 |
|---|---|
| 4. LangGraph - LangGraph ReAct Agent와 MemorySaver 학습 매뉴얼 (0) | 2026.05.05 |
| 3. LangGraph - LangGraph State Reducer & 메시지 상태 관리 학습 매뉴얼 (0) | 2026.05.04 |
| 2. LangGraph - StateGraph 상태 기반 그래프 (0) | 2026.05.03 |
| 1. LangGraph - LangChain Tool Calling 학습 매뉴얼 (0) | 2026.03.24 |
댓글