1. 🧠 LangGraph 상태 업데이트 방식 요약
LangGraph는 노드 간 상태를 공유하고, 노드들은 상태를 입력 → 처리 → 상태 업데이트 → 출력의 흐름으로 동작합니다. 출력은 다음 노드로 전달되며, 초기 상태는 시작 노드에서 주어지고, 최종 출력은 엔드 노드에서 생성됩니다.
✅ 기본 상태 업데이트 방식: 덮어쓰기 (Overwrite)
- 상태는 클래스로 정의되고, 각 필드는 기존 값을 덮어쓰는 방식으로 갱신됩니다.
- 예:
a=10설정- 다음 노드에서
a=a+10→a=20(기존 값 10은 사라짐)
이 방식은 기본 리듀서(Reducer)로 동작하며, 명시하지 않으면 이 방식이 사용됩니다.
➕ AddReducer: 리스트에 값 추가
- 예: 채팅 메시지나 여러 도구 실행 결과처럼 기존 상태에 값을 계속 추가해야 하는 경우.
- 이때는
addReducer를 사용해 리스트 필드에 값 추가 가능. - 구현 시
Annotated를 사용해 필드에add리듀서 지정:
from langgraph.graph import State
from operator import add
from typing import Annotated, List
class MyState(State):
document: Annotated[List[str], add]- 결과:
- 노드1:
None - 노드2:
['doc1', 'doc2', 'doc3'] - 노드3:
['doc1', 'doc2', 'doc3', 'doc2', 'doc4', 'doc5']
- 노드1:
🛠️ 커스텀 리듀서: 사용자 정의 로직
- 예: 중복 제거, 정렬, 특정 조건 필터링 등.
- 복잡한 상태 관리가 필요한 경우, 직접 함수를 정의하여 리듀서로 지정 가능.
💡 용도 예시
- 기본 리듀서: 단순한 값 갱신 (ex. 스코어, 플래그 등)
- AddReducer:
- 채팅 기록
- 도구 호출 내역
- 메시지 히스토리
- Custom Reducer:
- 중복 제거
- 정렬
- 상태 변형 로직 필요 시
2. ✅ 전체 요약
이 실습은 State Reducer를 활용하여 상태를 업데이트하는 방법을 실습해보는 과정입니다. 기본 덮어쓰기 방식, addReducer를 통한 데이터 추가 방식, 그리고 사용자 정의 커스텀 Reducer를 통해 중복 제거까지 세 가지 방식으로 상태를 관리하는 예제를 따라갑니다.
🧪 실습 흐름 요약
-
프로젝트 03번 메시지 그래프 파일 실행
- 가상환경(커널) 지정
- 환경 변수 및 라이브러리 import
-
기본 Reducer 실습 (덮어쓰기 방식)
DocumentState정의 (query,documents)query만 갱신됨 →documents는 업데이트 없음- Node 2, 3에서 각각 3개의 문서를 추가하지만, 마지막 값만 남음
-
addReducer사용Annotated로addReducer적용 → 기존 리스트에 append- Node 2와 Node 3에서 추가한 문서 총 6개가 모두 저장됨
-
Custom Reducer 실습 (중복 제거)
- 사용자 정의 함수
reduceUniqueDocument정의def reduceUniqueDocument(left, right): return list(set(left + right)) Annotated로 해당 함수 등록- 중복된 문서(
2번) 제거됨 → 총 5개 문서만 최종 저장됨
- 사용자 정의 함수
📌 핵심 개념 정리
| 구분 | 설명 |
|---|---|
| 기본 Reducer | 별도 설정이 없으면, 새 값으로 상태를 덮어씀 |
addReducer | 기존 값에 추가(append)하는 방식 |
| Custom Reducer | 사용자가 원하는 방식대로 상태 병합 가능 (예: 중복 제거, 조건 필터링 등) |
Annotated | 상태 필드에 reducer를 붙이는 데 사용되는 데코레이터 역할 |
| 그래프 흐름 | start → Node1 → Node2 → Node3 → end 순서대로 선형 연결 |
.
3. ✅ 메시지 그래프 (LangGraph의 Message Graph)
(1). 메시지 그래프란?
- 특수한 형태의 스테이트 그래프(State Graph)
- 챗봇 구현 시 다양한 메시지(Human, AI, Tool, System 등)를 리스트 형태로 저장·관리할 수 있도록 설계된 구조
- 랭체인의 챗모델과 유사한 방식으로 작동
- 채팅 히스토리를 효율적으로 관리하고, 대화의 흐름(컨텍스트)을 반영하는 데 효과적
(2). 메시지 관리 방식
1) 기본 메시지 키: messages
- 메시지들은 상태(state)의
messages라는 키에 리스트 형태로 저장됨 - 새 메시지를 추가할 때
addReducer를 사용
2) Reducer 함수
- LangGraph는
addMessages라는 기본 Reducer 제공
→ 기존 메시지를 ID로 추적 및 업데이트 가능 - 일반적인
add함수는 단순 리스트 추가만 지원
3. 메시지 스테이트 정의 방법
1) 직접 정의 방식
- 상태 클래스에
messages: list[BaseMessage]필드 추가 @reduce데코레이터로addMessages함수를 연결
2) LangGraph 제공 방식
- LangGraph는
MessageState클래스를 미리 제공messages키가 기본 포함- 직접 정의할 필요 없이 상속만으로 사용 가능
- 추가 필드(
document,grade,num_generation등)는 별도로 정의 가능
✅ 조건부 엣지를 활용한 피드백 루프 구현
시나리오 예시
- 문서 검색 → 응답 생성 → 응답 평가 → 기준 미달이면 되돌아가서 다시 수행
그래프 흐름
-
Retrieve and Respond 노드
- 문서 검색 및 응답 생성
-
Grade Answer 노드
- 응답의 품질을 평가 (0~1 사이 점수)
-
조건부 엣지 사용
- 점수 ≥ 0.7 →
End노드로 종료 - 점수 < 0.7 → 다시
Retrieve and Respond로 피드백 루프
- 점수 ≥ 0.7 →
장점
- 생성된 응답의 품질을 기준으로 반복 수행 가능
- 루프 구조를 통해 응답 품질 향상
- LangGraph에서는 조건부 엣지를 통해 이전 노드로 쉽게 되돌릴 수 있음
4 🧠 LangGraph 기반 RAG 시스템 구현 요약
📌 (1). 메시지 상태 정의 방법
✅ 방법 1: 직접 메시지 필드 정의
from typing import Annotated, List
from langchain_core.messages import AnyMessage
from langgraph.graph import State
from langgraph.utils import add_messages
class ChatState(State):
messages: Annotated[List[AnyMessage], add_messages]messages: 모든 종류의 메시지를 저장할 수 있는 필드.add_messages: 리스트에 메시지를 누적 저장하는 리듀서.
✅ 방법 2: 기본 제공 메시지 상태 상속
- LangGraph는
MessageState를 기본으로 제공. State를 상속하지 않고MessageState를 상속하면messages필드 자동 포함됨.
📌 (2). 추가 상태 필드 정의
from typing import List
from langchain_core.documents import Document
class GraphState(MessageState):
documents: List[Document] # 검색된 문서
grade: float # 답변 점수 (0~1)
num_generations: int # 답변 생성 횟수documents: 벡터 DB에서 검색된 문서들 (덮어쓰기 방식).grade: 답변 품질 점수.num_generations: 루프 제어를 위한 생성 횟수 카운트.
📌 (3). RAG 체인 구성
- 벡터 DB:
Chroma, 임베딩 모델:Ollama/BGE-M3 - Retriever 정의 후, LangChain의
RunnableParallel을 통해 다음 수행:question: 사용자 입력 →PassThroughdocuments: 유사도 검색 후 →format_docs함수로 문자열 병합prompt: 시스템 메시지 + 사용자 쿼리로 구성- LLM 호출 → 응답 생성
- 응답:
AIMessage로 변환되어messages상태에 추가됨
📌 (4). 답변 평가 노드 정의
Pydantic스키마 정의:
class GradeResponse(BaseModel):
score: float # 0 ~ 1
explanation: str-
평가 프롬프트 구성:
질문/답변/컨텍스트 기반으로 평가 요청 → 구조화된 출력 (GradeResponse) 생성 -
평가 노드에서는:
messages[-1]: 최근 AI 응답messages[-2]: 이전 사용자 질문documents: 컨텍스트 문서grade,num_generations상태 필드 업데이트
📌 (5). 조건 분기 (Conditional Edge)
- 조건 함수: 상태 기반으로 루프 제어
def route(state):
if state.num_generations > 2:
return "generate"
elif state.grade < 0.7:
return "retrieve_and_respond"
else:
return "generate"- 조건에 따라 다시 검색하거나 루프 종료 가능
📌 (6). 그래프 구성 및 실행
-
노드 연결 흐름:
start → retrieve_and_respond → grade_answer ────┐ ▼ 조건 분기 (route) ┌────────────┬────────────┐ ▼ ▼ retrieve_and_respond generate(end) -
그래프 실행 예시:
initial_state = {
"messages": [HumanMessage(content="채식주의자를 위한 메뉴 추천해줘")]
}
final_state = graph.invoke(initial_state)- 출력 결과:
- 응답 메시지 생성
- 평가 점수 1.0 → 루프 종료
- 최종 상태 메시지에 AI 응답 포함
✅ 최종 확인
- 그래프는 정상 작동
- 적절한 RAG 결과 생성
- 점수가 기준 이상이면 루프 종료, 미만이면 다시 검색
5. ✅ 전체 요약
이번 실습에서는 피드백 루프가 포함된 Graph 구조(LangGraph)를 기반으로, 이를 Gradio 챗봇 인터페이스로 확장하여 구현하는 과정을 다룹니다.
기존 Gradio 챗봇 구조에서 크게 달라진 부분은 없고, 상태(state)를 초기화하고 메시지를 전달하는 방식에서 LangGraph와 연동된 부분이 핵심입니다.
🧪 실습 흐름 요약
-
Gradio 임포트 및 기본 설정
gradio as gr임포트- 타이틀, 설명, 테마, 예제 질문(example questions) 3개 등록
-
챗봇 동작 함수 정의 (
answer_invoke)- 입력: 사용자 입력 문자열, 기존 대화
history - 처리:
history는 튜플 리스트: (HumanMessage,AIMessage)- 최근 두 개 메시지 + 새 입력 메시지를
initial_state로 구성 - 이
initial_state를 LangGraph에 전달
- 출력: 마지막 AI 응답의
content속성 (문자열)
- 입력: 사용자 입력 문자열, 기존 대화
-
Gradio 인터페이스 구성
gr.ChatInterface에 함수 등록- 타이틀/설명/예시 질문 포함
- 테마 설정 후 실행
-
실행
- 로컬 or 클라우드에서 URL로 접속
- 예시 질문 클릭 → Graph 실행 → 답변 출력 확인
📌 핵심 포인트 정리
| 요소 | 설명 |
|---|---|
| LangGraph | 메시지 흐름과 상태 업데이트를 관리하는 피드백 루프 구조 기반 |
| initial_state | history의 마지막 두 메시지 + 사용자 입력 메시지를 포함 |
| 출력 | state["messages"][-1].content를 추출하여 챗봇 응답으로 반환 |
| Gradio 챗봇 | 입력과 출력을 문자열로 구성해야 하므로 content만 추출 |
| 예시 질문 | "파스타에 어울리는 음료가 뭐예요?" 등 3가지 질문 추가 가능 |
| 추론 방식 | 외부 검색 기반일 수도 있고, AI의 내재 지식 기반일 수도 있음 |
Lee_AA