3. LangGraph - LangGraph State Reducer & 메시지 상태 관리 학습 매뉴얼
- 이 문서는 LangGraph의 핵심 개념인 State Reducer와 메시지 기반 상태 관리를 체계적으로 학습하기 위한 실습 매뉴얼입니다.
- 원본 노트북: `code/PRJ_03_LangGraph_MessageGraph.ipynb`
- State Reducer의 세 가지 패턴(기본 덮어쓰기, Annotated 리듀서, 커스텀 리듀서)을 이해하고, `StateGraph` + `MessagesState`를 활용한 RAG 기반 품질 평가 재시도 패턴을 구현합니다.
- LangGraph에서 상태(State)가 노드 간에 어떻게 전달되고 업데이트되는지, 그리고 메시지 기반 그래프가 실무에서 어떻게 활용되는지를 단계별로 학습합니다.
- 참고: 이전 버전의 LangGraph에서는 `MessageGraph`라는 별도의 클래스를 제공했으나, LangGraph 1.0부터 deprecated 되었습니다. 현재 권장되는 방식은 `StateGraph`에 `MessagesState` 또는 `Annotated[list, add_messages]`를 사용하는 것입니다. 이 문서는 최신 API를 기준으로 작성되었습니다.
1. 환경 설정
1.1. Env 환경변수
from dotenv import load_dotenv # [python-dotenv] .env 파일에서 환경변수를 로드하는 함수
load_dotenv() # .env 파일의 API 키(OPENAI_API_KEY 등)를 os.environ에 등록
1.2. 기본 라이브러리
import re # [Python 표준 라이브러리] 정규표현식 처리
import os, json # [Python 표준 라이브러리] os: 환경변수 접근, json: JSON 파싱
from textwrap import dedent # [Python 표준 라이브러리] 들여쓰기 제거 유틸리티
from pprint import pprint # [Python 표준 라이브러리] 딕셔너리/리스트 등을 보기 좋게 출력 (섹션 3.6에서 최종 결과 출력 시 사용)
import warnings # [Python 표준 라이브러리] 경고 메시지 제어
warnings.filterwarnings("ignore") # 불필요한 경고 메시지 숨김
2. State Reducer
- Reducer는 LangGraph에서 상태 업데이트를 관리하는 중요한 개념이다.
- 그래프의 각 노드가 반환하는 출력을 그래프의 전체 상태에 어떻게 통합(merge)할 것인지를 정의하는 메커니즘이다.
- Reducer가 필요한 이유는 크게 두 가지이다:
- 상태 덮어쓰기 문제: 기본적으로 각 노드의 반환값은 해당 상태 키의 이전 값을 완전히 덮어쓰는 방식(override)으로 동작한다. 이전 노드에서 수집한 데이터가 다음 노드의 반환값에 의해 사라질 수 있다.
- 누적 업데이트 필요: 메시지 리스트, 문서 목록, 로그 기록 등 이전 상태에 새로운 값을 추가(append)해야 하는 경우가 실무에서 매우 빈번하다.
- LangGraph에서 Reducer는 `TypedDict` 상태 클래스의 필드에 `Annotated` 타입 힌트를 사용하여 지정한다.
- Reducer 함수의 시그니처는 `(left, right) -> merged` 형태이다:
- `left`: 현재 상태에 저장된 기존 값
- `right`: 노드가 반환한 새로운 값
- 반환값: 두 값을 병합한 최종 결과
- Reducer를 지정하지 않으면 LangGraph는 기본적으로 "마지막 값으로 덮어쓰기(last-write-wins)" 전략을 사용한다.
2.1. Reducer를 별도로 지정하지 않은 경우
- [흐름 위치] State Reducer 학습의 첫 번째 단계로, Reducer가 없을 때의 기본 동작을 확인한다.
- Reducer를 별도로 지정하지 않으면 기존 값을 덮어쓰는 방식(override)으로 동작한다.
- 기본 Reducer는 상태에 대해 별도의 설정 없이 사용될 때 자동으로 적용된다.
- 이 동작은 LangGraph의 "last-write-wins" 정책이다. 동일한 키에 대해 여러 노드가 값을 반환하면, 가장 마지막에 실행된 노드의 반환값만 최종 상태에 남는다.
- 이 패턴은 단일 값(문자열, 숫자 등)을 업데이트할 때는 적합하지만, 리스트나 딕셔너리처럼 누적이 필요한 데이터에는 문제가 된다.
- 아래 예제에서 `node_2`가 `documents`에 3개 문서를 추가하고, `node_3`이 다시 `documents`를 반환하면, `node_2`의 결과는 완전히 사라지고 `node_3`의 결과만 남게 된다.
from typing import TypedDict, List # [Python 표준 라이브러리] 타입 힌트용 클래스
from langgraph.graph import StateGraph, START, END # [LangGraph 내장] 그래프 빌더 및 시작/종료 노드 상수
from IPython.display import Image, display # [IPython 내장] 그래프 시각화 이미지 출력용
# [사용자 정의] 상태 정의 - Reducer 미지정 시 기본 덮어쓰기 동작 확인용
class DocumentState(TypedDict):
query: str
documents: List[str]
# [사용자 정의] Node 1: query 업데이트
def node_1(state: DocumentState) -> DocumentState:
print("---Node 1 (query update)---")
query = state["query"]
return {"query": query}
# [사용자 정의] Node 2: 검색된 문서 추가
def node_2(state: DocumentState) -> DocumentState:
print("---Node 2 (add documents)---")
return {"documents": ["doc1.pdf", "doc2.pdf", "doc3.pdf"]}
# [사용자 정의] Node 3: 추가적인 문서 검색 결과 추가
def node_3(state: DocumentState) -> DocumentState:
print("---Node 3 (add more documents)---")
return {"documents": ["doc2.pdf", "doc4.pdf", "doc5.pdf"]}
# 그래프 빌드
builder = StateGraph(DocumentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# 논리 구성
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)
# 그래프 실행
graph = builder.compile()
# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))
- 아래 코드에서 `graph`는 위 [기본 사용법]에서 `builder.compile()`로 생성한 컴파일된 그래프 객체이다.
# 초기 상태
initial_state = {"query": "채식주의자를 위한 비건 음식을 추천해주세요."}
# 그래프 실행
final_state = graph.invoke(initial_state)
# 최종 상태 출력
print("최종 상태:", final_state)
```
```text
---Node 1 (query update)---
---Node 2 (add documents)---
---Node 3 (add more documents)---
최종 상태: {'query': '채식주의자를 위한 비건 음식을 추천해주세요.', 'documents': ['doc2.pdf', 'doc4.pdf', 'doc5.pdf']}
- 결과 분석: `documents`에는 `node_3`이 반환한 `['doc2.pdf', 'doc4.pdf', 'doc5.pdf']`만 남아 있다. `node_2`에서 추가한 `['doc1.pdf', 'doc2.pdf', 'doc3.pdf']`는 완전히 덮어씌워졌다.
- 이것이 바로 Reducer 없이 기본 동작을 사용할 때 발생하는 "상태 덮어쓰기 문제"이다.
- Reducer 미사용 시 상태 덮어쓰기 흐름
2.2. Reducer를 별도로 지정하는 경우
- [흐름 위치] State Reducer 학습의 두 번째 단계로, `Annotated` 타입과 `operator.add`를 사용하여 리스트를 누적하는 방법을 학습한다.
- `Annotated` 타입 힌트를 사용하여 특정 상태 필드에 Reducer 함수를 지정할 수 있다.
- `Annotated[타입, reducer_함수]` 형태로 작성하며, 이는 Python의 `typing.Annotated`를 활용한 LangGraph의 상태 관리 패턴이다.
- `operator.add`는 Python 표준 라이브러리의 함수로, 두 리스트를 연결(concatenate)하는 역할을 한다. 즉, `add([1, 2], [3, 4])`는 `[1, 2, 3, 4]`를 반환한다.
- 이 Reducer를 지정하면, 노드가 `documents` 키를 반환할 때마다 기존 리스트에 새 리스트가 이어붙여진다(append 방식).
- `Annotated`의 동작 원리:
- LangGraph는 상태 클래스를 분석할 때 `Annotated` 메타데이터에서 Reducer 함수를 추출한다.
- 노드가 상태 키를 반환하면, LangGraph는 해당 키에 Reducer가 지정되어 있는지 확인한다.
- Reducer가 있으면 `reducer(현재_상태값, 노드_반환값)`을 호출하여 새로운 상태값을 계산한다.
- Reducer가 없으면 노드 반환값으로 기존 값을 덮어쓴다.
from operator import add # [Python 표준 라이브러리] 두 리스트를 연결하는 함수 (Reducer로 사용)
from typing import Annotated, TypedDict # [Python 표준 라이브러리] 타입 힌트 - Annotated로 Reducer 지정
# [사용자 정의] 상태 정의 - documents 필드에 operator.add Reducer 적용
class ReducerState(TypedDict):
query: str
documents: Annotated[List[str], add]
# [사용자 정의] Node 1: query 업데이트
def node_1(state: ReducerState) -> ReducerState:
print("---Node 1 (query update)---")
query = state["query"]
return {"query": query}
# [사용자 정의] Node 2: 검색된 문서 추가
def node_2(state: ReducerState) -> ReducerState:
print("---Node 2 (add documents)---")
return {"documents": ["doc1.pdf", "doc2.pdf", "doc3.pdf"]}
# [사용자 정의] Node 3: 추가적인 문서 검색 결과 추가
def node_3(state: ReducerState) -> ReducerState:
print("---Node 3 (add more documents)---")
return {"documents": ["doc2.pdf", "doc4.pdf", "doc5.pdf"]}
# 그래프 빌드
builder = StateGraph(ReducerState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# 논리 구성
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)
# 그래프 실행
graph = builder.compile()
# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))
- 아래 코드에서 `graph`는 위 [기본 사용법]에서 `builder.compile()`로 생성한 컴파일된 그래프 객체이다.
# 초기 상태
initial_state = {"query": "채식주의자를 위한 비건 음식을 추천해주세요."}
# 그래프 실행
final_state = graph.invoke(initial_state)
# 최종 상태 출력
print("최종 상태:", final_state)---Node 1 (query update)---
---Node 2 (add documents)---
---Node 3 (add more documents)---
최종 상태: {'query': '채식주의자를 위한 비건 음식을 추천해주세요.', 'documents': ['doc1.pdf', 'doc2.pdf', 'doc3.pdf', 'doc2.pdf', 'doc4.pdf', 'doc5.pdf']}
- 결과 분석: `documents`에 `node_2`의 결과 `['doc1.pdf', 'doc2.pdf', 'doc3.pdf']`와 `node_3`의 결과 `['doc2.pdf', 'doc4.pdf', 'doc5.pdf']`가 모두 병합되어 총 6개의 문서가 저장되었다.
- 단, `doc2.pdf`가 중복으로 존재한다. `operator.add`는 단순히 두 리스트를 연결할 뿐, 중복 제거는 수행하지 않는다.
- 중복을 제거해야 하는 비즈니스 요구사항이 있다면 Custom Reducer를 사용해야 한다.
- add Reducer를 사용한 리스트 병합 흐름
2.3. Custom Reducer 사용
- [흐름 위치] State Reducer 학습의 세 번째 단계로, 비즈니스 로직에 맞는 커스텀 병합 함수를 작성하는 방법을 학습한다.
- 상태 업데이트가 기본적인 덮어쓰기나 단순 병합만으로 해결되지 않을 때 유용한 방법이다.
- 중복 제거, 최대/최소 값 유지, 조건부 병합, 우선순위 기반 정렬 등 특정 비즈니스 로직이 필요한 경우에 적용한다.
- Custom Reducer 함수의 시그니처는 반드시 `(left, right) -> merged` 형태여야 한다:
- `left` 파라미터: 현재 상태에 저장된 기존 값이다. 그래프 실행 초기에는 `None`일 수 있으므로 반드시 `None` 처리를 해야 한다.
- `right` 파라미터: 노드가 반환한 새로운 값이다. 마찬가지로 `None`일 수 있다.
- 반환값: 두 값을 병합한 최종 결과이다. 이 값이 상태에 새로 저장된다.
- Custom Reducer를 작성할 때 주의할 점:
- `left`와 `right` 모두 `None`일 수 있으므로 방어적 코딩이 필수이다.
- Reducer 함수는 순수 함수(pure function)여야 한다. 즉, 동일한 입력에 대해 항상 동일한 출력을 반환해야 한다.
- Reducer 함수 내에서 부작용(side effect)이 있는 연산(파일 I/O, 네트워크 호출 등)은 피해야 한다.
- 반환 타입은 상태 필드의 타입과 일치해야 한다.
from typing import TypedDict, List, Annotated # [Python 표준 라이브러리] 타입 힌트
# [사용자 정의] Custom reducer: 중복된 문서를 제거하며 리스트 병합
def reduce_unique_documents(left: list | None, right: list | None) -> list:
"""Combine two lists of documents, removing duplicates."""
if not left:
left = []
if not right:
right = []
# 중복 제거: set을 사용하여 중복된 문서를 제거하고 다시 list로 변환
return list(set(left + right))
# [사용자 정의] 상태 정의 (documents 필드 포함)
class CustomReducerState(TypedDict):
query: str
documents: Annotated[List[str], reduce_unique_documents] # Custom Reducer 적용
# [사용자 정의] Node 1: query 업데이트
def node_1(state: CustomReducerState) -> CustomReducerState:
print("---Node 1 (query update)---")
query = state["query"]
return {"query": query}
# [사용자 정의] Node 2: 검색된 문서 추가
def node_2(state: CustomReducerState) -> CustomReducerState:
print("---Node 2 (add documents)---")
return {"documents": ["doc1.pdf", "doc2.pdf", "doc3.pdf"]}
# [사용자 정의] Node 3: 추가적인 문서 검색 결과 추가
def node_3(state: CustomReducerState) -> CustomReducerState:
print("---Node 3 (add more documents)---")
return {"documents": ["doc2.pdf", "doc4.pdf", "doc5.pdf"]}
# 그래프 빌드
builder = StateGraph(CustomReducerState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# 논리 구성
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)
# 그래프 실행
graph = builder.compile()
# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))
- 아래 코드에서 `graph`는 위 [기본 사용법]에서 `builder.compile()`로 생성한 컴파일된 그래프 객체이다.
# 초기 상태
initial_state = {"query": "채식주의자를 위한 비건 음식을 추천해주세요.", "documents": []}
# 그래프 실행
final_state = graph.invoke(initial_state)
# 최종 상태 출력
print("최종 상태:", final_state)
```
```text
---Node 1 (query update)---
---Node 2 (add documents)---
---Node 3 (add more documents)---
최종 상태: {'query': '채식주의자를 위한 비건 음식을 추천해주세요.', 'documents': ['doc4.pdf', 'doc3.pdf', 'doc5.pdf', 'doc1.pdf', 'doc2.pdf']}
- 결과 분석: `documents`에 5개의 고유한 문서만 남아 있다. `node_2`와 `node_3` 모두 `doc2.pdf`를 반환했지만, Custom Reducer의 `set()` 연산으로 중복이 제거되어 `doc2.pdf`는 하나만 존재한다.
- `set()`을 사용하므로 결과 리스트의 순서는 보장되지 않는다. 순서가 중요한 경우 `dict.fromkeys()`나 정렬 로직을 추가해야 한다.
- Custom Reducer 패턴은 실무에서 매우 자주 사용된다. 예를 들어:
- 검색된 문서의 중복 제거 (위 예제)
- 점수 기반 최대/최소값 유지 (`max(left, right)`)
- 조건부 병합 (특정 조건을 만족하는 항목만 추가)
- 시간순 정렬 유지 (타임스탬프 기반 정렬)
- Custom Reducer를 사용한 중복 제거 흐름
2.4. add_messages 내장 Reducer
- [흐름 위치] State Reducer 학습의 네 번째 단계로, LangGraph가 메시지 관리를 위해 제공하는 가장 중요한 내장 Reducer인 `add_messages`를 학습한다.
- `add_messages`는 `langgraph.graph` (또는 `langgraph.graph.message`)에서 import할 수 있는 LangGraph 전용 Reducer이다.
- 이 Reducer는 단순한 `operator.add`와 달리 메시지의 `id` 속성을 추적한다:
- 새 메시지의 ID가 기존 메시지 목록에 이미 존재하면, 해당 위치의 메시지를 새 메시지로 **교체(update)** 한다.
- ID가 없는 새 메시지는 목록 끝에 **추가(append)** 한다.
- 이는 LLM이 응답을 재생성하거나, 스트리밍 중 메시지가 점진적으로 업데이트되는 경우에 중복 메시지가 쌓이는 문제를 방지한다.
- `add_messages`는 메시지 기반 LangGraph 애플리케이션에서 거의 항상 사용되는 핵심 Reducer이다.
- `add_messages`와 `operator.add`의 차이점 상세 비교:
| 특성 | operator.add | add_messages |
| 동작 방식 | 단순 리스트 연결 (+) | ID 기반 추가/업데이트 |
| 중복 메시지 | 그대로 쌓임 | 동일 ID면 교체 |
| 메시지 삭제 | 불가 | RemoveMessage로 가능 |
| 사용 시점 | 단순 로그 누적 등 | 대화 히스토리 관리 (권장) |
from typing import Annotated # [Python 표준 라이브러리] 타입 힌트에 메타데이터 부착
from typing_extensions import TypedDict # [typing_extensions] 타입 힌트용 TypedDict
from langgraph.graph import add_messages # [LangGraph 내장] 메시지 ID 기반 누적/업데이트 Reducer
from langchain_core.messages import AIMessage, HumanMessage # [LangChain 내장] 메시지 클래스
# [사용자 정의] 방법 1: 직접 Annotated로 add_messages Reducer를 지정하는 방법
class MyState(TypedDict):
messages: Annotated[list, add_messages]
# add_messages Reducer 동작 확인 예시
existing = [HumanMessage(content="안녕하세요", id="msg1")]
new = [AIMessage(content="안녕하세요! 무엇을 도와드릴까요?", id="msg2")]
result = add_messages(existing, new)
# 결과: [HumanMessage("안녕하세요", id="msg1"), AIMessage("안녕하세요! ...", id="msg2")]
# → 새 메시지(msg2)가 기존 목록에 추가됨
# 동일 ID로 업데이트하는 예시
existing2 = [HumanMessage(content="안녕하세요", id="msg1"), AIMessage(content="이전 답변", id="msg2")]
update = [AIMessage(content="수정된 답변", id="msg2")]
result2 = add_messages(existing2, update)
# 결과: [HumanMessage("안녕하세요", id="msg1"), AIMessage("수정된 답변", id="msg2")]
# → msg2의 내용이 교체됨
- 위 방법은 `messages` 키에 `add_messages` Reducer를 직접 지정하는 명시적 방법이다.
- 아래 방법은 `MessagesState`를 사용하여 더 간결하게 작성하는 방법이다. `MessagesState`는 LangGraph가 제공하는 내장(pre-built) 상태 클래스로, `messages` 키에 `add_messages` Reducer가 이미 적용되어 있다.
# [방법 2] MessagesState 내장 클래스 사용 - messages 키에 add_messages가 이미 적용됨
from langgraph.graph import MessagesState, StateGraph
# MessagesState를 그대로 사용하거나, 상속하여 추가 필드를 정의할 수 있다.
# MessagesState를 상속하면 messages 키는 자동으로 제공되고, 추가 필드만 선언하면 된다.
class ExtendedState(MessagesState):
extra_field: str # 추가 필드 예시
- 참고 (deprecated): 이전 버전의 LangGraph에서는 `from langgraph.graph import MessageGraph`를 사용하여 메시지 전용 그래프를 생성할 수 있었다. `MessageGraph`는 LangGraph 1.0에서 deprecated 되었으며, 현재는 `StateGraph(MessagesState)` 또는 `StateGraph`에 `Annotated[list, add_messages]` 키를 포함한 커스텀 상태를 사용하는 것이 표준이다.
- Reducer 네 가지 패턴 비교 요약:
| 패턴 | 선언 방식 | 동작 | 사용 시점 |
| 기본 (No Reducer) | documents: List[str] | 마지막 값으로 덮어쓰기 | 단일 값 업데이트 (query, score 등) |
| operator.add | Annotated[List[str], add] | 리스트 연결 (중복 허용) | 로그 누적 등 |
| Custom Reducer | Annotated[List[str], custom_fn] | 사용자 정의 병합 로직 | 중복 제거, 조건부 병합, 정렬 등 |
| add_messages | Annotated[list, add_messages] | ID 기반 메시지 추가/업데이트 | 메시지 히스토리 관리 (가장 권장) |
3. StateGraph를 활용한 메시지 기반 RAG 그래프
- LangChain의 ChatModel은 Message 객체 목록을 입력으로 처리한다. 이 메시지들은 `HumanMessage`(사용자 입력), `AIMessage`(LLM 응답), `SystemMessage`(시스템 프롬프트) 등 다양한 형태로 제공된다.
- 현재 LangGraph에서 메시지 기반 상태 관리는 `StateGraph`에 `MessagesState`를 사용하거나, 커스텀 `TypedDict` 상태에 `Annotated[list, add_messages]`를 선언하여 구현한다.
- LangGraph는 메시지 리스트를 상태로 관리하기 위한 전용 Reducer인 `add_messages`를 제공한다. 이 Reducer는 단순한 `operator.add`와 달리 메시지 ID를 추적하여 기존 메시지의 업데이트도 올바르게 처리한다.
- 이 섹션에서는 레스토랑 메뉴 RAG 시스템을 구축하면서, 답변 품질을 평가하고 품질이 낮으면 자동으로 재시도하는 패턴을 구현한다.
- 참고 (deprecated): 이전 버전에서는 `MessageGraph`라는 `StateGraph`의 특수 유형이 있었으나, LangGraph 1.0에서 deprecated 되었다. `MessageGraph`는 내부적으로 `messages` 키에 `add_messages` Reducer를 적용한 `StateGraph`와 동일하게 동작했다. 현재는 `StateGraph(MessagesState)` 또는 커스텀 상태를 직접 정의하는 것이 표준 방식이다.
3.1. Messages State 정의
- [흐름 위치] 메시지 기반 RAG 그래프 구현의 첫 번째 단계로, 메시지 기반 상태 클래스를 정의한다.
- 이전 대화 기록을 그래프 상태에 메시지 목록으로 저장하는 것이 유용하다.
- 그래프 상태에 Message 객체 목록을 저장하는 키(채널)를 추가하고, 이 키에 리듀서 함수를 연결한다.
- 리듀서 함수 선택 시 두 가지 옵션이 있다:
- `operator.add`: 새 메시지를 기존 목록에 단순히 추가한다. 리스트 연결만 수행하므로 메시지 업데이트가 불가능하다.
- `add_messages`: LangGraph가 제공하는 전용 Reducer이다. 새 메시지는 기존 목록에 추가하되, 동일한 ID를 가진 메시지가 이미 존재하면 해당 메시지를 업데이트한다. 이는 메시지 수정, 재생성 등의 시나리오에서 매우 중요하다.
- `add_messages`와 `operator.add`의 차이점 상세 설명:
- `operator.add`는 단순히 `list.__add__`를 호출한다. 즉, `[msg1] + [msg2]` = `[msg1, msg2]`이다. 메시지 ID를 전혀 고려하지 않는다.
- `add_messages`는 메시지의 `id` 속성을 추적한다. 새 메시지의 ID가 기존 메시지 목록에 이미 존재하면, 해당 위치의 메시지를 새 메시지로 교체(update)한다. ID가 없는 새 메시지는 목록 끝에 추가(append)한다.
- 실무에서는 거의 항상 `add_messages`를 사용해야 한다. 특히 LLM이 응답을 재생성하거나, 스트리밍 중 메시지가 점진적으로 업데이트되는 경우에 `operator.add`를 사용하면 중복 메시지가 계속 쌓이는 문제가 발생한다.
- `MessagesState`는 LangGraph가 제공하는 내장(pre-built) 상태 클래스이다. 이 클래스는 `messages` 키에 `add_messages` Reducer가 이미 적용되어 있어, 직접 `Annotated` 설정을 할 필요가 없다.
- `MessagesState`를 상속받으면 `messages` 키는 자동으로 제공되고, 추가로 필요한 상태 필드만 선언하면 된다.
from typing import Annotated # [Python 표준 라이브러리] 타입 힌트에 메타데이터 부착
from langchain_core.messages import AnyMessage # [LangChain 내장] 모든 메시지 타입의 유니온 타입
from langgraph.graph import add_messages # [LangGraph 내장] 메시지 ID 기반 누적/업데이트 Reducer
# langgraph.graph.message에서도 import 가능하지만 langgraph.graph에서 직접 가져오는 것이 권장됨
# [사용자 정의] 방법 1: 직접 Annotated로 add_messages Reducer를 지정하는 방법
class GraphState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
- 위 방법은 `messages` 키에 `add_messages` Reducer를 직접 지정하는 명시적 방법이다.
- 아래 방법은 `MessagesState`를 상속받아 더 간결하게 작성하는 방법이다. 실무에서는 이 방법이 더 많이 사용된다.
- 아래에서 정의하는 `GraphState`는 이후 섹션 3.3~3.6 전체에서 그래프의 상태 스키마로 사용된다.
# 방법 2: LangGraph MessagesState라는 미리 만들어진 상태를 사용
from langgraph.graph import MessagesState # [LangGraph 내장] messages 키에 add_messages Reducer가 적용된 내장 상태 클래스
from typing import List # [Python 표준 라이브러리] 리스트 타입 힌트
from langchain_core.documents import Document # [LangChain 내장] 검색된 문서를 표현하는 클래스
# [사용자 정의] GraphState - MessagesState를 상속하여 RAG 그래프에 필요한 추가 필드를 정의
class GraphState(MessagesState):
# messages 키는 기본 제공 (add_messages Reducer 자동 적용) - 다른 키를 추가하고 싶을 경우 아래와 같이 적용
documents: List[Document] # Reducer 미지정 → 기본 덮어쓰기 방식 (섹션 2.1 참조)
grade: float # Reducer 미지정 → 기본 덮어쓰기 방식 (섹션 2.1 참조). 섹션 3.3의 grade_answer에서 설정
num_generation: int # Reducer 미지정 → 기본 덮어쓰기 방식 (섹션 2.1 참조). 섹션 3.3의 grade_answer에서 설정
- `GraphState`에는 총 4개의 상태 키가 존재한다:
- `messages`: `MessagesState`에서 상속받은 키. `add_messages` Reducer가 적용되어 있다. 대화 히스토리를 저장한다.
- `documents`: 검색된 문서 목록을 저장한다. Reducer가 없으므로 기본 덮어쓰기 방식이다.
- `grade`: 답변 품질 평가 점수(0~1)를 저장한다.
- `num_generation`: 답변 생성 횟수를 추적한다. 무한 루프 방지를 위한 카운터이다.
- GraphState 클래스 상속 구조
3.2. RAG Chain 구성
- [흐름 위치] 메시지 기반 RAG 그래프 구현의 두 번째 단계로, 메뉴 검색을 위한 벡터저장소를 초기화하고 RAG 체인을 구성한다.
- Chroma 벡터저장소에 저장된 레스토랑 메뉴 데이터를 검색하고, ChatOpenAI LLM으로 답변을 생성하는 LCEL(LangChain Expression Language) 체인을 구성한다.
- RAG 체인의 구성 요소:
- `OllamaEmbeddings(model="bge-m3")`: 한국어를 지원하는 다국어 임베딩 모델
- `Chroma`: 벡터 유사도 검색을 수행하는 벡터저장소
- `ChatOpenAI(model="gpt-4o-mini")`: 답변 생성을 담당하는 LLM
- `retriever`: 상위 2개 관련 문서를 검색 (`search_kwargs={"k": 2}`)
- LCEL 체인: `retriever | format_docs -> prompt -> llm -> StrOutputParser`
- 이 섹션에서 정의하는 `retriever`, `rag_chain`, `format_docs`, `llm`은 이후 섹션 3.3의 노드 함수에서 클로저(closure)로 참조된다. 즉, 섹션 3.3의 함수 내부에서 별도의 인자 전달 없이 이 변수들을 직접 사용한다.
from langchain_chroma import Chroma # [LangChain 내장] Chroma 벡터저장소 인터페이스
from langchain_ollama import OllamaEmbeddings # [LangChain 내장] Ollama 임베딩 모델 래퍼
from langchain_openai import ChatOpenAI # [LangChain 내장] OpenAI ChatGPT 모델 래퍼
from langchain_core.messages import HumanMessage, AIMessage # [LangChain 내장] 사용자/AI 메시지 클래스 (섹션 3.3, 3.6, 4에서도 사용)
from langchain_core.output_parsers import StrOutputParser # [LangChain 내장] LLM 출력을 문자열로 파싱
from langchain_core.prompts import ChatPromptTemplate # [LangChain 내장] 프롬프트 템플릿 생성기
from langchain_core.runnables import RunnablePassthrough, RunnableLambda # [LangChain 내장] LCEL 체인 구성용 유틸리티
embeddings_model = OllamaEmbeddings(model="bge-m3")
# Chroma 인덱스 로드
vector_db = Chroma(
embedding_function=embeddings_model,
collection_name="restaurant_menu",
persist_directory="./chroma_db",
)
# LLM 모델
llm = ChatOpenAI(model="gpt-4o-mini")
# RAG 체인 구성
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
system = """
You are a helpful assistant. Use the following context to answer the user's question:
[Context]
{context}
"""
prompt = ChatPromptTemplate.from_messages([
("system", system),
("human", "{question}")
])
# 검색기 정의
retriever = vector_db.as_retriever(
search_kwargs={"k": 2}
)
# RAG 체인 구성
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)# RAG 체인 실행
query = "채식주의자를 위한 메뉴를 추천해주세요."
response = rag_chain.invoke(query)
# 답변 출력
print(response)채식주의자를 위한 메뉴로 가든 샐러드를 추천드립니다.
**가든 샐러드**
- **가격:** ₩12,000
- **주요 식재료:** 유기농 믹스 그린, 체리 토마토, 오이, 당근, 발사믹 드레싱
- **설명:** 신선한 유기농 채소들로 구성된 건강한 샐러드입니다. 아삭한 식감의 믹스 그린에 달콤한 체리 토마토, 오이, 당근을 더해 다양한 맛과 식감을 즐길 수 있습니다. 특제 발사믹 드레싱이 채소 본연의 맛을 살려줍니다.
이 메뉴는 완전 채식으로, 다양한 신선한 채소를 사용하여 건강에도 좋습니다.
- LCEL 체인의 데이터 흐름을 다이어그램으로 표현하면 다음과 같다:
- LCEL 체인의 데이터 흐름
3.3. 노드(Node)
- [흐름 위치] 메시지 기반 RAG 그래프 구현의 세 번째 단계로, 그래프에서 실행될 노드 함수들을 정의한다.
- 이 그래프에는 두 개의 핵심 노드가 있다:
- `retrieve_and_respond`: 사용자 질문을 받아 문서를 검색하고 RAG 답변을 생성하는 노드
- `grade_answer`: 생성된 답변의 품질을 평가하는 노드
- `retrieve_and_respond` 노드의 동작:
- `state['messages']`에서 마지막 메시지(사용자 질문)를 추출한다.
- `retriever`로 관련 문서를 검색한다.
- `rag_chain`으로 답변을 생성한다.
- 생성된 답변을 `AIMessage`로 래핑하여 `messages`에 추가하고, 검색된 문서를 `documents`에 저장한다.
- `grade_answer` 노드의 동작:
- `messages`에서 질문(뒤에서 두 번째)과 답변(마지막)을 추출한다.
- `with_structured_output`을 사용하여 LLM이 Pydantic 모델(`GradeResponse`)에 맞는 구조화된 응답을 반환하도록 한다.
- `with_structured_output`은 LangChain이 제공하는 매우 강력한 기능이다. LLM의 자유형식 텍스트 출력을 Pydantic 모델로 정의된 구조화된 데이터로 변환한다. 내부적으로 OpenAI의 Function Calling 또는 JSON Mode를 활용하여 LLM이 지정된 스키마에 맞는 JSON을 출력하도록 강제한다.
- `GradeResponse` Pydantic 모델은 `score`(0~1 범위의 float)와 `explanation`(설명 문자열) 두 필드로 구성된다. `Field(..., ge=0, le=1)`은 값이 0 이상 1 이하여야 한다는 유효성 검증을 수행한다.
- 아래 `retrieve_and_respond` 함수 내부에서 사용되는 `retriever`, `rag_chain`, `AIMessage`는 섹션 3.2에서 정의된 객체를 클로저로 참조한다.
# [사용자 정의] RAG 수행 함수 정의
# - state: GraphState 타입 (섹션 3.1에서 정의). messages, documents, grade, num_generation 키를 포함한다.
# - retriever: 섹션 3.2에서 정의한 Chroma 벡터저장소 기반 검색기 (vector_db.as_retriever())
# - rag_chain: 섹션 3.2에서 정의한 LCEL 체인 (retriever | format_docs -> prompt -> llm -> StrOutputParser)
# - AIMessage: 섹션 3.2에서 import한 LangChain AI 메시지 클래스 (langchain_core.messages)
def retrieve_and_respond(state: GraphState):
# state['messages']는 add_messages Reducer가 적용된 메시지 리스트 (섹션 3.1의 MessagesState에서 상속)
last_human_message = state['messages'][-1]
# HumanMessage 객체의 content 속성에 접근
query = last_human_message.content
# 문서 검색 - retriever는 섹션 3.2에서 정의한 Chroma 벡터저장소 검색기
retrieved_docs = retriever.invoke(query) # [사용자 정의 - 섹션 3.2 참조]
# 응답 생성 - rag_chain은 섹션 3.2에서 정의한 LCEL 체인
response = rag_chain.invoke(query) # [사용자 정의 - 섹션 3.2 참조]
# 검색된 문서와 응답을 상태에 저장
# messages에 AIMessage를 반환하면, add_messages Reducer가 기존 메시지 목록에 추가한다 (섹션 2.4 참조)
# documents는 Reducer가 없으므로 기본 덮어쓰기 방식으로 저장된다 (섹션 2.1 참조)
return {
"messages": [AIMessage(content=response)],
"documents": retrieved_docs
}
- `messages`에 `[AIMessage(content=response)]`를 반환하면, `add_messages` Reducer가 이를 기존 메시지 목록에 추가한다. 즉, 사용자의 `HumanMessage` 뒤에 AI의 응답이 이어붙여진다.
- 아래 `grade_answer` 함수 내부에서 사용되는 `format_docs`, `llm`은 섹션 3.2에서 정의된 객체를 클로저로 참조한다.
from pydantic import BaseModel, Field # [Pydantic v2] 구조화된 출력 스키마 정의용
# [사용자 정의] 답변 품질 평가 응답 스키마 (Pydantic v2 모델)
# - with_structured_output()에서 이 스키마를 사용하여 LLM 출력을 구조화한다 (아래 grade_answer 함수 참조)
# - Pydantic v2에서는 .schema() 대신 .model_json_schema()를 사용하여 JSON 스키마를 얻는다
class GradeResponse(BaseModel):
"A score for answers"
score: float = Field(..., ge=0, le=1, description="A score from 0 to 1, where 1 is perfect")
explanation: str = Field(..., description="An explanation for the given score")
# Pydantic v2 JSON 스키마 확인 (참고용)
# GradeResponse.model_json_schema() # Pydantic v2 방식 (.schema()는 deprecated)
# [사용자 정의] 답변 품질 평가 함수
# - state: GraphState 타입 (섹션 3.1에서 정의). messages, documents, grade, num_generation 키를 포함한다.
# - format_docs: 섹션 3.2에서 정의한 문서 텍스트 변환 함수 (Document 객체 리스트 → 문자열)
# - llm: 섹션 3.2에서 정의한 ChatOpenAI 인스턴스 (model="gpt-4o-mini")
# - ChatPromptTemplate: 섹션 3.2에서 import한 LangChain 프롬프트 템플릿 클래스
def grade_answer(state: GraphState):
# state['messages']에서 질문과 답변 추출
# messages[-2]는 사용자 질문(HumanMessage), messages[-1]는 AI 답변(AIMessage)
# 이 메시지들은 retrieve_and_respond 노드(위 기본 사용법 참조)에서 추가된 것이다
messages = state['messages']
question = messages[-2].content
answer = messages[-1].content
# state['documents']는 retrieve_and_respond 노드에서 저장한 검색된 문서 목록
# format_docs는 섹션 3.2에서 정의한 함수로, Document 객체 리스트를 하나의 텍스트 문자열로 변환한다
context = format_docs(state['documents']) # [사용자 정의 - 섹션 3.2 참조] 검색된 문서를 텍스트로 변환
grading_system = """You are an expert grader.
Grade the following answer based on its relevance and accuracy to the question, considering the given context.
Provide a score from 0 to 1, where 1 is perfect, along with an explanation."""
grading_prompt = ChatPromptTemplate.from_messages([
("system", grading_system),
("human", "[Question]\n{question}\n\n[Context]\n{context}\n\n[Answer]\n{answer}\n\n[Grade]\n")
])
# llm.with_structured_output(schema=GradeResponse)는 LLM 출력을 GradeResponse Pydantic 모델로 구조화한다
# llm은 섹션 3.2에서 정의한 ChatOpenAI(model="gpt-4o-mini") 인스턴스
# GradeResponse는 위에서 정의한 Pydantic v2 모델 (score, explanation 필드)
grading_chain = grading_prompt | llm.with_structured_output(schema=GradeResponse) # [사용자 정의 - 섹션 3.2 참조]
grade_response = grading_chain.invoke({
"question": question,
"context": context,
"answer": answer
})
# 답변 생성 횟수를 증가
# state.get('num_generation', 0)은 이전에 설정된 생성 횟수를 가져오거나, 없으면 0을 반환
# 이 값은 섹션 3.4의 should_retry 함수에서 무한 루프 방지 조건으로 사용된다
num_generation = state.get('num_generation', 0)
num_generation += 1
return {"grade": grade_response.score, "num_generation": num_generation}
- `with_structured_output` 동작 원리 상세 설명:
- `llm.with_structured_output(schema=GradeResponse)`는 LLM 인스턴스를 래핑하여 출력이 `GradeResponse` Pydantic 모델을 따르도록 강제한다.
- 내부적으로 OpenAI의 경우 Function Calling API를 사용하여 LLM에게 "이 스키마에 맞는 JSON을 반환하라"고 지시한다.
- LLM의 응답은 자동으로 파싱되어 `GradeResponse` 인스턴스로 변환된다. 따라서 `grade_response.score`와 같이 Python 객체의 속성으로 접근할 수 있다.
- Pydantic v2의 유효성 검증(`ge=0, le=1`)이 자동으로 적용되어, LLM이 범위를 벗어난 점수를 반환하면 오류가 발생한다.
- retrieve_and_respond 노드 — 문서 검색 + RAG 답변 생성
- grade_answer 노드 — 답변 품질 평가
3.4. 엣지(Edge)
- [흐름 위치] 메시지 기반 RAG 그래프 구현의 네 번째 단계로, 노드 간의 조건부 라우팅 로직을 정의한다.
- `should_retry` 함수는 답변 품질 평가 결과에 따라 그래프의 다음 실행 경로를 결정하는 조건부 엣지(conditional edge) 함수이다.
- 이 함수는 LangGraph의 `add_conditional_edges`에 전달되어, 그래프 실행 중 동적으로 다음 노드를 선택한다.
- 조건부 엣지의 반환값은 문자열이며, 이 문자열은 `add_conditional_edges`의 매핑 딕셔너리에서 실제 노드 이름과 매핑된다.
- 품질 기반 재시도 패턴(Quality-based Retry Pattern) 상세 설명:
- 이 패턴은 LLM 기반 시스템에서 매우 중요한 설계 패턴이다.
- LLM의 답변 품질은 확률적으로 변동하기 때문에, 한 번의 시도로 만족스러운 답변이 생성되지 않을 수 있다.
- 답변을 생성한 후 품질을 자동으로 평가하고, 기준에 미달하면 재시도하는 루프를 구성한다.
- 반드시 최대 재시도 횟수(여기서는 `num_generation > 2`)를 설정하여 무한 루프를 방지해야 한다.
- 재시도 시 검색부터 다시 수행(`retrieve_and_respond`)하므로, 다른 문서가 검색되어 더 나은 답변이 생성될 가능성이 있다.
from typing import Literal # [Python 표준 라이브러리] 함수 반환값의 리터럴 타입 힌트
# [사용자 정의] 조건부 엣지 함수 - 품질 평가 결과에 따라 재시도 여부 결정
# - state: GraphState 타입 (섹션 3.1에서 정의)
# - state["grade"]: 섹션 3.3의 grade_answer 노드에서 설정한 답변 품질 평가 점수 (0~1)
# - state["num_generation"]: 섹션 3.3의 grade_answer 노드에서 설정한 답변 생성 횟수
# - 반환값은 섹션 3.5의 add_conditional_edges에서 매핑 딕셔너리의 키로 사용된다
def should_retry(state: GraphState) -> Literal["retrieve_and_respond", "generate"]:
print("----GRADTING---")
print("Grade Score: ", state["grade"])
# state["num_generation"]은 섹션 3.3의 grade_answer에서 설정된 답변 생성 횟수
# 답변 생성 횟수가 3회 이상이면 "generate"를 반환
if state["num_generation"] > 2:
return "generate"
# state["grade"]는 섹션 3.3의 grade_answer에서 설정된 품질 평가 점수 (0~1)
# 답변 품질 평가점수가 0.7 미만이면 RAG 체인을 다시 실행
if state["grade"] < 0.7:
return "retrieve_and_respond"
else:
return "generate"
- `Literal["retrieve_and_respond", "generate"]`: 타입 힌트로, 이 함수가 반환할 수 있는 값을 명시한다. LangGraph는 이 정보를 활용하여 그래프의 유효성을 검증한다.
- 조건 평가 순서가 중요하다:
- 먼저 `num_generation > 2`를 확인하여 무한 루프를 방지한다. 품질이 아무리 낮아도 3회 이상 시도했으면 종료한다.
- 그 다음 `grade < 0.7`을 확인하여 품질이 낮으면 재시도한다.
- 위 두 조건에 해당하지 않으면(품질 0.7 이상이고 생성 횟수 3회 미만) 종료한다.
- 실무에서 `should_retry`는 위 기본 사용법 코드가 그대로 사용된다. 실제 동작은 3.6절의 Graph 실행에서 확인한다.
- should_retry 조건부 라우팅 흐름
3.5. 그래프(Graph) 구성
- [흐름 위치] 메시지 기반 RAG 그래프 구현의 다섯 번째 단계로, 앞서 정의한 노드와 엣지를 조합하여 완전한 StateGraph를 구성한다.
- `StateGraph(GraphState)`: `GraphState` 스키마를 사용하는 그래프 빌더를 생성한다.
- `add_node("이름", 함수)`: 그래프에 노드를 추가한다. 첫 번째 인자는 노드의 고유 이름, 두 번째 인자는 실행할 함수이다.
- `add_edge(출발, 도착)`: 두 노드 사이에 무조건적 엣지를 추가한다. 출발 노드가 완료되면 항상 도착 노드가 실행된다.
- `add_conditional_edges(출발, 조건함수, 매핑)`: 조건부 엣지를 추가한다. 조건함수의 반환값에 따라 다음 노드가 결정된다. 매핑 딕셔너리는 `{반환값: 실제_노드이름}` 형태이다.
- `builder.compile()`: 그래프를 컴파일하여 실행 가능한 상태로 만든다. 컴파일 시 그래프의 유효성(모든 노드가 연결되어 있는지, 도달 불가능한 노드가 없는지 등)을 검증한다.
# 그래프 설정
# GraphState는 섹션 3.1에서 정의한 상태 스키마 (MessagesState를 상속, messages/documents/grade/num_generation 키 포함)
builder = StateGraph(GraphState)
# 섹션 3.3에서 정의한 노드 함수들을 그래프에 등록
# retrieve_and_respond: 사용자 질문을 받아 문서를 검색하고 RAG 답변을 생성하는 노드 (섹션 3.3 기본 사용법 참조)
# grade_answer: 생성된 답변의 품질을 평가하는 노드 (섹션 3.3 실무 코드 참조)
builder.add_node("retrieve_and_respond", retrieve_and_respond)
builder.add_node("grade_answer", grade_answer)
# 엣지 구성: START → retrieve_and_respond → grade_answer → (조건부 분기)
builder.add_edge(START, "retrieve_and_respond")
builder.add_edge("retrieve_and_respond", "grade_answer")
# 조건부 엣지: grade_answer 완료 후, should_retry 함수의 반환값에 따라 분기
# should_retry는 섹션 3.4에서 정의한 조건부 엣지 함수
# "retrieve_and_respond" 반환 시 → retrieve_and_respond 노드로 재시도
# "generate" 반환 시 → END로 이동하여 그래프 종료
builder.add_conditional_edges(
"grade_answer",
should_retry, # 섹션 3.4에서 정의한 조건부 엣지 함수
{
"retrieve_and_respond": "retrieve_and_respond",
"generate": END
}
)
# 그래프 컴파일
graph = builder.compile()
# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))
- 그래프 구조 설명:
- `START -> retrieve_and_respond`: 그래프 시작 시 항상 검색+응답 노드가 먼저 실행된다.
- `retrieve_and_respond -> grade_answer`: 응답이 생성되면 항상 품질 평가 노드가 실행된다.
- `grade_answer -> (조건부)`: 품질 평가 결과에 따라 `retrieve_and_respond`(재시도) 또는 `END`(종료)로 분기한다.
- `"generate": END` 매핑에 주목: `should_retry`가 `"generate"`를 반환하면 그래프가 `END`로 이동하여 종료된다. 여기서 `"generate"`는 실제 노드 이름이 아니라 조건부 엣지의 라우팅 키일 뿐이다.
- 컴파일된 그래프의 구조는 아래 다이어그램과 같다:
- 컴파일된 그래프 구조
3.6. Graph 실행
- [흐름 위치] 메시지 기반 RAG 그래프 구현의 마지막 단계로, 구성된 그래프를 실제로 실행하고 결과를 확인한다.
- `graph.invoke(initial_state)`: 그래프를 동기 방식으로 실행한다. 초기 상태를 전달하면, 그래프가 `START`에서 시작하여 모든 노드를 순차적으로 실행하고, `END`에 도달하면 최종 상태를 반환한다.
- 초기 상태에는 `messages` 키에 `HumanMessage`를 넣어 사용자의 질문을 전달한다.
- 그래프 실행 중 `add_messages` Reducer가 동작하여, `retrieve_and_respond` 노드가 반환한 `AIMessage`가 기존 `messages` 리스트에 자동으로 추가된다.
# 초기 상태
initial_state = {
"messages": [HumanMessage(content="채식주의자를 위한 메뉴를 추천해주세요.")], # HumanMessage는 섹션 3.2에서 import한 LangChain 메시지 클래스
}
# graph는 섹션 3.5에서 builder.compile()로 생성한 컴파일된 그래프 객체
# 그래프 실행
final_state = graph.invoke(initial_state)
# 최종 상태 출력
print("최종 상태:", final_state)
- `final_state`는 GraphState 구조로, messages(대화 히스토리), documents(검색된 문서), grade(품질 점수), num_generation(생성 횟수)을 포함한다.
# pprint는 섹션 1.2에서 import한 Python 표준 라이브러리 함수 (딕셔너리/리스트를 보기 좋게 출력)
# final_state['messages'][-1]은 그래프 실행 결과의 마지막 AIMessage (최종 답변)
# final_state는 위 [기본 사용법]에서 graph.invoke()로 반환된 최종 상태 딕셔너리
# 최종 답변만 출력
pprint(final_state['messages'][-1].content)----GRADTING---
Grade Score: 1.0
최종 상태: {'messages': [HumanMessage(content='채식주의자를 위한 메뉴를 추천해주세요.',
additional_kwargs={}, response_metadata={}, id='93b05f86-b2d7-4274-aaa6-3cfde11b944d'),
AIMessage(content='채식주의자를 위한 메뉴로는 "가든 샐러드"를 추천합니다.
이 샐러드는 유기농 믹스 그린, 체리 토마토, 오이, 당근을 사용하여 신선하고 건강한 맛을 자랑합니다.
특제 발사믹 드레싱이 채소의 본연의 맛을 살려주어 다양한 맛과 식감을 즐길 수 있습니다.
가격은 ₩12,000입니다.', additional_kwargs={}, response_metadata={},
id='f152c439-a479-428b-ace9-1326281a8c6c')], 'documents': [...],
'grade': 1.0, 'num_generation': 1}('채식주의자를 위한 메뉴로는 "가든 샐러드"를 추천합니다.
이 샐러드는 유기농 믹스 그린, 체리 토마토, 오이, 당근을 사용하여 신선하고 '
'건강한 맛을 자랑합니다. 특제 발사믹 드레싱이 채소의 본연의 맛을 살려주어
다양한 맛과 식감을 즐길 수 있습니다. 가격은 '
'₩12,000입니다.')
- 실행 결과 분석:
- `Grade Score: 1.0`으로 첫 번째 시도에서 최고 품질의 답변이 생성되었다.
- `num_generation: 1`이므로 재시도 없이 1회만에 완료되었다.
- `messages` 리스트에는 `HumanMessage`(사용자 질문)와 `AIMessage`(RAG 답변) 총 2개의 메시지가 저장되어 있다.
- `documents`에는 검색된 `Document` 객체 2개가 저장되어 있다 (retriever의 `k=2` 설정).
- `grade < 0.7`이었다면 `retrieve_and_respond` 노드가 다시 실행되어 `messages`에 새로운 `AIMessage`가 추가되었을 것이다.
- 그래프 실행 흐름 전체 다이어그램:
- 그래프 실행 흐름 시퀀스
4. Gradio 챗봇
- [흐름 위치] 전체 구현의 마지막 단계로, 구축한 RAG 그래프를 웹 인터페이스로 배포한다.
- Gradio의 `ChatInterface`를 사용하여 대화형 챗봇 UI를 구성한다.
- 대화 기록(history)을 LangGraph의 메시지 형식으로 변환하여 그래프에 전달한다.
- 최근 2개의 대화만 컨텍스트로 사용하여 토큰 사용량을 관리한다 (`chat_history[-2:]`).
import gradio as gr # [Gradio 내장] 웹 UI 인터페이스 라이브러리
from typing import List, Dict # [Python 표준 라이브러리] 타입 힌트
# [사용자 정의] 예시 질문들
example_questions = [
"채식주의자를 위한 메뉴를 추천해주세요.",
"오늘의 스페셜 메뉴는 무엇인가요?",
"파스타에 어울리는 음료는 무엇인가요?"
]
# [사용자 정의] 대답 함수 정의 - Gradio ChatInterface의 콜백 함수
# - message: 사용자가 입력한 현재 메시지 (문자열)
# - history: Gradio가 관리하는 이전 대화 기록 (List[Dict[str, str]], 각 항목은 {"role": "user"|"assistant", "content": "..."} 형태)
# - HumanMessage, AIMessage: 섹션 3.2에서 import한 LangChain 메시지 클래스 (langchain_core.messages)
# - graph: 섹션 3.5에서 builder.compile()로 생성한 컴파일된 StateGraph 객체
def answer_invoke(message: str, history: List[Dict[str, str]]) -> str:
try:
# 채팅 기록을 AI에게 전달할 수 있는 형식으로 변환
# Gradio의 Dict 형식({"role": "user", "content": "..."})을 LangChain 메시지 객체로 변환
chat_history = []
for msg in history:
if msg["role"] == "user":
chat_history.append(HumanMessage(content=msg["content"])) # HumanMessage는 섹션 3.2에서 import한 LangChain 메시지 클래스
elif msg["role"] == "assistant":
chat_history.append(AIMessage(content=msg["content"])) # AIMessage는 섹션 3.2에서 import한 LangChain 메시지 클래스
# 기존 채팅 기록에 사용자의 메시지를 추가 (최근 2개 대화만 사용)
# chat_history[-2:]로 최근 2개의 메시지만 가져와 토큰 사용량을 관리한다
initial_state = {
"messages": chat_history[-2:]+[HumanMessage(content=message)],
}
# graph는 섹션 3.5에서 builder.compile()로 생성한 컴파일된 그래프 객체
# graph.invoke()는 그래프를 동기 실행하여 최종 상태를 반환한다 (섹션 3.6 참조)
# 메시지를 처리하고 최종 상태를 반환
final_state = graph.invoke(initial_state)
# 최종 상태에서 마지막 AIMessage의 content를 반환 (retrieve_and_respond 노드가 생성한 답변)
return final_state["messages"][-1].content
except Exception as e:
# 오류 발생 시 사용자에게 알리고 로그 기록
print(f"Error occurred: {str(e)}")
return "죄송합니다. 응답을 생성하는 동안 오류가 발생했습니다. 다시 시도해 주세요."# [Gradio 내장] Gradio 인터페이스 생성 - ChatInterface는 대화형 챗봇 UI를 자동 구성
demo = gr.ChatInterface(
fn=answer_invoke, # [사용자 정의] 위에서 정의한 대답 함수
title="레스토랑 메뉴 AI 어시스턴트",
description="메뉴 정보, 추천, 음식 관련 질문에 답변해 드립니다.",
examples=example_questions,
)
# Gradio 앱 실행
demo.launch()* Running on local URL: http://127.0.0.1:7860# 데모 종료
demo.close()Closing server running on port: 7860
- Gradio UI 기반 RAG 챗봇 흐름
- 이 문서에서 학습한 핵심 LangGraph 개념 정리:
| 개념 | 설명 | 핵심 포인트 |
| State Reducer | 노드 출력을 상태에 통합하는 방법 정의 | 기본 덮어쓰기, operator.add 누적, Custom 함수 |
| Annotated 타입 | 상태 필드에 Reducer를 연결하는 타입 힌트 | Annotated[타입, reducer_함수] 형태 |
| Custom Reducer | 비즈니스 로직에 맞는 병합 함수 | (left, right) -> merged 시그니처, None 처리 필수 |
| add_messages | 메시지 ID를 추적하는 전용 Reducer | operator.add와 달리 메시지 업데이트 지원, LangGraph 메시지 관리의 핵심 |
| MessagesState | 메시지 기반 상태의 내장 클래스 | messages 키에 add_messages가 이미 적용됨, MessageGraph의 대체 |
| StateGraph | LangGraph의 표준 그래프 빌더 | MessagesState와 함께 사용하여 메시지 기반 그래프 구현 (MessageGraph는 deprecated) |
| with_structured_output | LLM 출력을 Pydantic 모델로 구조화 | Function Calling 기반, Pydantic v2의 유효성 검증 자동 적용 |
| 품질 기반 재시도 패턴 | 답변 품질 평가 후 조건부 재실행 | 무한 루프 방지를 위한 최대 횟수 제한 필수 |
'Study > LangChain' 카테고리의 다른 글
| 5. LangGraph - LangGraph Agentic RAG 학습 매뉴얼 (0) | 2026.05.06 |
|---|---|
| 4. LangGraph - LangGraph ReAct Agent와 MemorySaver 학습 매뉴얼 (0) | 2026.05.05 |
| 2. LangGraph - StateGraph 상태 기반 그래프 (0) | 2026.05.03 |
| 1. LangGraph - LangChain Tool Calling 학습 매뉴얼 (0) | 2026.03.24 |
| 10. LangChain HuggingFace 오픈소스 언어모델 활용방법 (1) | 2026.03.16 |
댓글