[LangGraph] RAG에 반복·분기·상태를 추가하는 방법

About this offer

LangGraph 쉽게 이해하기 - RAG에 반복·분기·상태를 추가하는 방법 아래 링크에 있는 유튜브를 참고하고 공부했다. 참고 유튜브 RAG를 공부하다 보면 어느 순간 이런 생각이 든다. 검색한 문서가 엉뚱하면 다시 검색하고 싶은데? 문서에 답이 없으면 웹 검색을 하면 안 될까? LLM의 답변이 문서에 근거한 답변인지 검사할 수 없을까? 답변이 이상하면 다시 생성하면 안 될까? 중요한 작업이라면 사람이 중간에서 승인하게 만들 수 없을까? 기본적인 RAG는 보통 다음과 같이 동작한다. 질문 ↓ 문서 검색 ↓ Prompt ↓ LLM ↓ 답변 이 구조는 간단하고 이해하기 쉽다. 하지만 실제 서비스를 만들기 시작하면 점점 복잡해진다. 검색 실패 → 질문 다시 작성 그래도 실패 → 웹 검색 답변 생성 → 답변 검증 검증 실패 → 다시 생성 또 실패 → 다시 검색 이런 흐름을 일반적인 Chain에 계속 붙이면 코드가 복잡해진다. 이런 분기, 반복, 상태 관리가 필요한 LLM Workflow를 그래프 형태로 만들기 위한 도구가 LangGraph 다. 1. 왜 LangGraph가 필요할까? 먼저 기본 RAG를 생각해보자. 사용자가 다음과 같이 질문했다. "생성형 AI Gauss를 만든 회사의 2023년 매출액을 알려줘." RAG가 회사 내부 문서를 검색했다. 그리고 다음 정보를 찾았다. Samsung Gauss를 만든 회사 → 삼성전자 그런데 문제가 있다. 문서에는 삼성전자의 2023년 매출액 정보가 없다. 그러면 어떻게 해야 할까? 가장 단순한 방법은 웹 검색 기능을 하나 더 붙이는 것이다. 질문 ↓ 문서 검색 ↓ 정보 부족? ↓ YES 웹 검색 ↓ 답변 생성 그런데 웹 검색 결과가 잘못되면? 웹 검색 ↓ 잘못된 정보 ↓ LLM ↓ 잘못된 답변 이 된다. 그러면 이번에는 웹 검색 결과 검사 기능 을 추가하고 싶어진다. 웹 검색 ↓ 검색 결과 평가 ↓ 좋음 → 답변 생성 나쁨 → 다시 검색 그런데 다시 검색했는데도 결과가 안 좋으면? 다시 검색하게 된다. 검색 ↓ 실패 ↓ 검색 ↓ 실패 ↓ 검색 ↓ ... 잘못 만들면 비용과 실행 시간이 계속 증가할 수도 있다. 결국 기능을 하나씩 추가할수록 Chain + Chain + 검증 Chain + Search Chain + Rewrite Chain + Retry Logic 처럼 구조가 복잡해진다. 강의에서도 일반적인 Naive RAG가 미리 정한 Loader, Chunking, Query, Prompt 등을 따라 한 방향으로 흘러가면 중간 결과를 보고 되돌아가 수정하기 어렵다는 점을 문제로 설명한다. 다만 정확히 말하면 RAG 자체가 반복이나 분기를 못 하는 것은 아니다. Python 코드로 직접 if , while 등을 만들면 얼마든지 가능하다. 문제는 LLM Workflow가 복잡해질수록 흐름과 상태를 직접 관리하기 어려워진다는 것 이다. LangGraph는 바로 이 문제를 해결하기 위해 사용한다. 2. LangGraph란? LangGraph를 한 문장으로 정리하면 LLM 애플리케이션의 작업 과정을 Node와 Edge로 연결하고, State를 공유하면서 분기와 반복을 만들 수 있게 해주는 Workflow Framework 라고 할 수 있다. 이름 그대로 Graph 를 만든다. 예를 들어 기존 RAG가 질문 ↓ 검색 ↓ 답변 이었다면 LangGraph에서는 ┌──── 질문 재작성 ────┐ │ │ ↓ │ 질문 → 문서 검색 → 문서 평가 ──────┘ │ │ 통과 ↓ 답변 생성 ↓ 답변 평가 / \ 통과 실패 ↓ ↓ 종료 웹 검색 │ └──→ 다시 답변 처럼 만들 수 있다. 이제 프로그램이 직선이 아니라 지도처럼 보이기 시작한다. LangGraph에서는 각각의 작업을 Node , Node를 연결하는 길을 Edge 라고 부른다. 조건에 따라 다른 Node로 이동하도록 만드는 Conditional Edge도 사용할 수 있다. 3. LangGraph 핵심 개념 4가지 LangGraph를 처음 배울 때는 다음 네 가지만 먼저 이해하면 된다. State Node Edge Conditional Edge 쉽게 비유하면 택배 배송 시스템 과 비슷하다. State → 택배 상자 Node → 택배를 처리하는 각 회사 Edge → 회사와 회사를 연결하는 도로 Conditional Edge → 상황에 따라 다른 도로를 선택하는 교차로 하나씩 알아보자. 4. State - 노드끼리 공유하는 택배 상자 LangGraph에서 가장 중요한 개념 중 하나가 State 다. 각 Node는 서로 독립적으로 동작한다. 예를 들어 Node 1 → 질문 처리 Node 2 → 문서 검색 Node 3 → 답변 생성 이 있다고 하자. Node 2는 Node 1 내부에서 무슨 코드가 실행됐는지 알 필요가 없다. 대신 State 에 필요한 정보를 담아서 전달한다. 강의에서도 각각의 Node를 서로 다른 회사에 비유하고, 이전 Node의 처리 결과를 State 라는 택배 상자에 넣어 다음 Node로 전달한다고 설명한다. State 정의하기 Python에서는 보통 TypedDict 를 사용한다. from typing_extensions import TypedDict class GraphState(TypedDict, total=False): question: str context: str answer: str relevance: str 쉽게 말하면 우리 그래프에서 앞으로 사용할 데이터 이름을 미리 정하는 것 이다. State는 이런 모습이다. { "question": "RAG가 뭐야?", "context": "RAG는 검색 증강 생성으로...", "answer": "RAG는 외부 문서를...", "relevance": "good" } TypedDict는 무엇인가? 일반 Dictionary는 data = { "question": "RAG가 뭐야?" } 처럼 자유롭게 사용할 수 있다. TypedDict 는 여기에 question은 str context도 str answer도 str 처럼 어떤 타입이 들어오는지 설명을 붙이는 것 이다. 즉 초보 단계에서는 타입이 적혀 있는 Dictionary 라고 생각하면 충분하다. State는 계속 전달된다 처음에는 질문만 있다고 하자. { "question": "RAG가 뭐야?" } 문서 검색 Node를 지나면 { "question": "RAG가 뭐야?", "context": "RAG는 외부 문서를 검색해서..." } 가 된다. 답변 Node를 지나면 { "question": "RAG가 뭐야?", "context": "RAG는 외부 문서를 검색해서...", "answer": "RAG는 쉽게 말하면..." } 가 된다. 즉 Node 1 ↓ State Node 2 ↓ State 업데이트 Node 3 ↓ State 업데이트 방식이다. 노드는 전체 State를 반환하지 않아도 된다 여기서 중요한 부분이 있다. 예전 설명을 보면 이런 코드가 등장할 수 있다. def retrieve_document(state): ... return GraphState( question=state["question"], context=context, answer=state["answer"] ) 하지만 굳이 모든 값을 다시 반환할 필요는 없다. 현재 StateGraph 의 기본 Node 구조는 State → Partial State 다. 즉 변경한 값만 반환하면 된다. def retrieve_document(state: GraphState): documents = "검색된 문서입니다." return { "context": documents } 그러면 기존의 { "question": "RAG가 뭐야?" } 에 context 가 업데이트된다. 5. Reducer - 덮어쓸까? 합칠까? 기본적으로 State의 같은 Key가 다시 반환되면 새로운 값으로 업데이트된다. 예를 들어 state["context"] = "문서 A" 가 있었는데 다음 Node에서 return { "context": "문서 B" } 를 반환하면 결과는 문서 B 가 된다. 즉 기존 값을 덮어쓴다. 그런데 Chat History처럼 계속 쌓아야 하는 데이터도 있다. 사용자: 안녕하세요 AI: 안녕하세요! 사용자: RAG가 뭐예요? AI: RAG는... 이런 데이터를 매번 덮어쓰면 이전 대화가 사라진다. 그래서 사용하는 것이 Reducer 다. LangGraph에서 State Key에 Reducer를 지정하면 이전 값과 새로운 값을 어떤 방식으로 합칠지 결정할 수 있다. 자막에서도 일반 State 값은 새 값으로 갱신되지만 메시지는 Reducer를 통해 누적시키는 구조를 설명한다. operator.add 일반 List를 계속 이어 붙이고 싶다면 operator.add 를 사용할 수 있다. import operator from typing import Annotated from typing_extensions import TypedDict class GraphState(TypedDict): logs: Annotated[ list[str], operator.add ] 이제 ["검색 시작"] 이 있는 상태에서 return { "logs": ["검색 완료"] } 를 반환하면 [ "검색 시작", "검색 완료" ] 처럼 합쳐진다. add_messages 대화 메시지를 관리할 때는 add_messages 를 사용할 수 있다. from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import add_messages from langchain_core.messages import AnyMessage class GraphState(TypedDict): messages: Annotated[ list[AnyMessage], add_messages ] add_messages 는 단순히 List 뒤에 붙이는 것보다 Message를 다루는 데 적합하다. 새로운 Message ID라면 추가하고, 같은 ID를 가진 Message가 들어오면 해당 Message를 업데이트할 수도 있다. Annotated는 무엇인가? Annotated 는 원래 타입에 추가적인 정보를 붙이는 Python 문법 이다. 예를 들어 Annotated[str, "Context"] 는 기본 타입 → str 추가 Metadata → "Context" 라고 볼 수 있다. 하지만 Annotated[str, "Context"] 라고 적는다고 LangGraph가 특별한 동작을 하는 것은 아니다. 반면 Annotated[ list[AnyMessage], add_messages ] 처럼 Reducer 함수를 넣으면 LangGraph가 그 함수를 State 병합 방식으로 사용한다. 6. Node - 실제 일을 하는 함수 Node는 생각보다 간단하다. 대부분 Python 함수 라고 생각하면 된다. def retrieve_document(state: GraphState): question = state["question"] print( "검색할 질문:", question ) return { "context": "검색된 문서입니다." } Node가 하는 일은 State 받기 ↓ 필요한 값 꺼내기 ↓ 작업하기 ↓ 변경할 State 반환하기 이다. LangGraph에서는 Node가 내부적으로 무엇을 하는지 제한하지 않는다. 따라서 LLM 호출 Vector DB 검색 SQL 실행 Web 검색 API 호출 Python 계산 파일 처리 등을 모두 Node 안에 넣을 수 있다. 답변 생성 Node def generate_answer( state: GraphState ): question = state["question"] context = state["context"] answer = f""" 질문: {question} 참고 자료: {context} """ return { "answer": answer } 7. Edge - Node와 Node 연결하기 Node만 여러 개 만들어놓으면 실행 순서를 알 수 없다. 그래서 Edge를 연결한다. retrieve ↓ generate ↓ evaluate LangGraph에서는 다음처럼 만든다. workflow.add_edge( "retrieve", "generate" ) workflow.add_edge( "generate", "evaluate" ) 쉽게 말하면 add_edge( "출발 Node", "도착 Node" ) 다. Conditional Edge - 조건에 따라 길 선택하기 LangGraph를 사용하는 가장 큰 이유 중 하나다. 답변을 평가했다고 하자. 결과가 grounded 면 종료하고 싶다. 그런데 not_grounded 이면 질문을 다시 작성하고 싶다. ┌→ END │ evaluate ──────┤ │ └→ rewrite 이것을 Conditional Edge로 만든다. 조건 판단 함수 from typing import Literal def route_answer( state: GraphState ) -> Literal["end", "rewrite"]: if state["relevance"] == "good": return "end" return "rewrite" Conditional Edge 연결 from langgraph.graph import END workflow.add_conditional_edges( "evaluate", route_answer, { "end": END, "rewrite": "rewrite" } ) 동작은 다음과 같다. evaluate 실행 ↓ route_answer() ↓ "end" → END "rewrite" → rewrite Node 현재 LangGraph API 이름은 add_conditional_edges() 처럼 edges가 복수형 이다. 8. 순환 구조를 만들 수 있다 기존의 단방향 RAG는 보통 검색 ↓ 답변 ↓ 끝 이다. LangGraph에서는 다음처럼 되돌아갈 수 있다. ┌─────────────────┐ │ │ ▼ │ 질문 → retrieve → generate → evaluate ▲ │ │ │ └── rewrite ←─────┘ 예를 들어 검색 결과 안 좋음 ↓ 질문 다시 작성 ↓ 다시 검색 이 가능하다. 이런 Cycle 을 쉽게 표현할 수 있다는 점이 LangGraph의 중요한 특징이다. 그런데 무한 반복하면? 다음처럼 만들어버리면 문제가 생긴다. 검색 ↓ 실패 ↓ 재검색 ↓ 실패 ↓ 재검색 ↓ ... LLM이나 Search API를 사용한다면 시간 증가 + Token 증가 + API 비용 증가 문제가 생긴다. 그래서 실행할 때 recursion_limit 을 설정할 수 있다. config = { "recursion_limit": 20 } 현재 LangGraph/Runnable 설정에서 기본 recursion limit은 25이며, 제한을 초과하면 무한 Loop를 막기 위해 GraphRecursionError 가 발생한다. 9. 가장 기본적인 LangGraph 직접 만들기 이제 정말 간단한 Graph를 만들어보자. 먼저 설치한다. uv add langgraph 1단계. State 정의 from typing_extensions import TypedDict class GraphState(TypedDict, total=False): question: str context: str answer: str relevance: str retry_count: int 2단계. Node 정의 def retrieve( state: GraphState ): question = state["question"] context = ( f"'{question}'에 대해 " "검색한 문서입니다." ) return { "context": context } 답변 생성: def generate( state: GraphState ): answer = ( f"질문: {state['question']}\n" f"근거: {state['context']}\n" "따라서 검색 문서를 기반으로 " "답변했습니다." ) return { "answer": answer } 평가: def evaluate( state: GraphState ): if state.get("context"): result = "good" else: result = "bad" return { "relevance": result } 질문 재작성: def rewrite( state: GraphState ): count = ( state.get( "retry_count", 0 ) + 1 ) new_question = ( state["question"] + " 구체적인 근거를 포함해서 검색" ) return { "question": new_question, "retry_count": count } 3단계. 분기 함수 from typing import Literal def route( state: GraphState ) -> Literal[ "finish", "rewrite" ]: if state["relevance"] == "good": return "finish" return "rewrite" 4단계. Graph 정의 현재는 START , END 를 이용하면 흐름을 명확하게 표현할 수 있다. from langgraph.graph import ( StateGraph, START, END ) workflow = StateGraph( GraphState ) workflow.add_node( "retrieve", retrieve ) workflow.add_node( "generate", generate ) workflow.add_node( "evaluate", evaluate ) workflow.add_node( "rewrite", rewrite ) 5단계. Edge 연결 workflow.add_edge( START, "retrieve" ) workflow.add_edge( "retrieve", "generate" ) workflow.add_edge( "generate", "evaluate" ) 조건부 Edge: workflow.add_conditional_edges( "evaluate", route, { "finish": END, "rewrite": "rewrite" } ) 다시 검색하도록 연결한다. workflow.add_edge( "rewrite", "retrieve" ) 전체 구조는 START ↓ retrieve ↓ generate ↓ evaluate / \ 좋음 나쁨 ↓ ↓ END rewrite ↓ retrieve 가 된다. 6단계. Compile StateGraph 는 설계도이기 때문에 그대로 실행하지 않는다. 마지막에 Compile한다. app = workflow.compile() 현재 StateGraph 는 Builder이고, compile() 을 수행하면 invoke() , stream() 등을 사용할 수 있는 실행 가능한 Graph가 만들어진다. 7단계. 실행 result = app.invoke( { "question": "LangGraph가 뭐야?" }, { "recursion_limit": 20 } ) print( result["answer"] ) 10. set_entry_point는 이제 못 쓰는 걸까? 아니다. 다음 코드도 여전히 사용할 수 있다. workflow.set_entry_point( "retrieve" ) 현재 set_entry_point("retrieve") 는 사실상 workflow.add_edge( START, "retrieve" ) 와 같은 의미다. 개인적으로는 그래프를 볼 때 START END 가 명시적으로 보이는 방식이 초보자에게 더 이해하기 쉽다. 11. Checkpointer - 그래프의 세이브 파일 게임을 생각해보자. Stage 1 ↓ Stage 2 ↓ Stage 3 게임을 껐다 켜면 다시 처음부터 해야 한다면 불편하다. 그래서 Save Point 가 필요하다. LangGraph에서는 이것과 비슷한 역할을 하는 것이 Checkpointer 다. Checkpointer는 Graph가 실행되면서 만들어지는 State의 Snapshot을 저장한다. Checkpoint 1 → 질문 입력 Checkpoint 2 → 검색 완료 Checkpoint 3 → 답변 생성 Checkpoint 4 → 평가 완료 강의에서도 Checkpointer를 이용해 이