랭그래프(LangGraph)에서 알아두면 유용한 Python 문법

Song Chae Won·2025년 8월 18일

LLM langChain langgraph rag

🪴LangGraph 알아가기

목록 보기

1/3

기존에 LangChain 프레임워크로 개발한 파이프라인을 LangGraph로 변경하는 과정에서 알아두면 좋을 내용을 정리해두려한다!

개발하면서 하나하나 정리해두고 싶었지만 전처리 해야할 파일들 사이에서 정신을 잃다가 1차적으로 전처리 마무리하고 이제야 정리해본다 😁

랭그래프 변경부터, 대부분의 내용은 테디노트의 랭체인을 활용한 RAG 비법노트를 참조하였다🙇‍♀️

깃허브 링크: https://github.com/teddylee777/langchain-kr/tree/main/17-LangGraph)

TypedDict: 상태(State) 정의

LangGraph의 상태(State)는 그래프 전체에서 공유되는 데이터 구조로, TypedDict를 활용해 상태의 키와 타입을 정의할 수 있음

from typing import TypedDict, Annotated, List
from langchain_core.documents import Document
import operator

# 상태 정의 예시
class GraphState(TypedDict):
    context: Annotated[List[Document], operator.add]
    answer: Annotated[List[Document], operator.add]
    question: Annotated[str, operator.add]
    sql_query: Annotated[str, operator.add]
    binary_score: Annotated[str, operator.add]

설명

TypedDict를 사용하면 각 키의 타입을 지정할 수 있어, IDE 자동완성 및 타입체크에 유리함

Annotated와 reducer(예: operator.add)는 상태 업데이트 방식을 지정

Annotated: 상태 업데이트 방식 지정

Annotated는 타입 힌트에 추가 정보를 부여할 수 있는 Python의 typing 기능으로, LangGraph에서는 상태의 각 필드에 대해 업데이트(리듀서) 방식을 지정할 때 주로 사용할 수 있음

from typing import Annotated
import operator

# 리스트를 누적(append)하는 상태 필드 예시
messages: Annotated[list[str], operator.add]

설명

operator.add를 리듀서로 지정하면, 새 값이 기존 리스트에 추가(append)됩니다.

기본적으로 리듀서를 지정하지 않으면, 새 값이 이전 값을 덮어씁니다.

add_messages: 메시지 리스트 관리

LangGraph에서 대화 메시지 히스토리를 관리할 때는 add_messages 리듀서를 사용한다. 이 함수는 메시지 ID를 기준으로 중복 메시지를 교체하거나 새 메시지를 추가할 수 있다.

from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

class MyState(TypedDict):
    messages: Annotated[list[AIMessage | HumanMessage], add_messages]

# 메시지 병합 예시
msgs1 = [HumanMessage(content="Hello?", id="1")]
msgs2 = [AIMessage(content="Nice to meet you!", id="2")]

result = add_messages(msgs1, msgs2)
print(result)

설명

add_messages는 메시지 리스트를 병합하며, 같은 ID가 있으면 교체, 없으면 추가

LangGraph에서 대화형 워크플로우를 만들 때 필수적으로 사용됨!

노드(Node)와 엣지(Edge) 정의

노드는 상태를 입력받아 새로운 상태를 반환하는 함수이고 엣지는 노드 간의 연결(흐름)을 정의한다.

def retrieve(state: GraphState) -> GraphState:
    # 예시: 문서 검색
    documents = "검색된 문서"
    return GraphState(context=documents)

def llm_gpt_execute(state: GraphState) -> GraphState:
    # 예시: LLM 실행
    answer = "LLM 생성 답변"
    return GraphState(answer=answer)

그래프 구조화 및 실행

StateGraph를 사용해 노드와 엣지를 연결하고, 그래프를 컴파일하여 실행한다.

from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import MemorySaver

workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("LLM 요청", llm_gpt_execute)
workflow.add_edge("retrieve", "LLM 요청")
workflow.add_edge("LLM 요청", END)

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

설명

StateGraph로 그래프를 정의하고, 노드와 엣지 연결하기

반드시 .compile()을 호출해야 실행 가능함

실전 예시: 메시지 상태를 가진 그래프

from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def add_human_message(state: ChatState) -> ChatState:
    new_msg = HumanMessage(content="안녕하세요!", id="1")
    return {"messages": [new_msg]}

def add_ai_message(state: ChatState) -> ChatState:
    new_msg = AIMessage(content="반갑습니다!", id="2")
    return {"messages": [new_msg]}

from langgraph.graph import StateGraph, END

workflow = StateGraph(ChatState)
workflow.add_node("add_human", add_human_message)
workflow.add_node("add_ai", add_ai_message)
workflow.add_edge("add_human", "add_ai")
workflow.add_edge("add_ai", END)
workflow.set_entry_point("add_human")
app = workflow.compile()

설명

메시지 리스트를 상태로 관리하며, 각 노드에서 메시지를 추가하기

add_messages 리듀서를 통해 메시지 병합이 자동으로 처리

참고자료

Song Chae Won

@chhaewxn

다음 포스트