포스트

2026-04-28 강의 정리

LangGraph 활용 - 상태 그래프 구현

2026-04-28 강의 정리

목차

  1. 문제와 접근
  2. LangGraph vs LangChain
  3. 핵심 개념 — 노드, 엣지, 상태(State)
  4. TypedDict로 State 정의
  5. StateGraph 기본 패턴
  6. 다중 노드 직렬 연결
  7. LLM을 노드로 사용
  8. add_conditional_edges — 조건 분기
  9. 여러 분기 합치기 (merge 노드)
  10. 핵심 개념 총정리
  11. 자주 나오는 실수/주의사항
  12. 그래프를 읽는 기준은 상태와 종료 조건

1. 문제와 접근

분기와 반복이 생기면 함수 호출 순서를 나열하는 것만으로 전체 동작을 파악하기 어려워진다. LangGraph의 상태·노드·엣지를 사용해 실행 경로와 데이터 갱신을 분리해 본다. 그래프를 그리는 것보다 각 노드가 읽고 쓰는 값과 종료 조건을 명확히 하는 데 초점을 둔다.

1
pip install langgraph
1
2
3
4
5
6
7
8
9
10
import os
from typing import TypedDict, Annotated, Optional
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langgraph.graph import StateGraph, START, END

llm = ChatOpenAI(model="gpt-4o-mini")

2. LangGraph vs LangChain

1
2
3
4
5
6
7
8
9
10
11
[LangChain]
  chain = prompt | retriever | llm | parser
  → 파이프(|) 연산자로 선형(직선) 연결
  → 분기, 반복, 상태 공유가 어렵다

[LangGraph]
  노드(node) : 함수 (search, find, answer, ...)
  엣지(edge) : 노드들을 연결하는 화살표
  상태(state): 그래프 전체에서 공유하는 딕셔너리
  → 비선형 흐름(분기, 반복, 사이클) 가능
  → 상태가 노드 사이를 자동으로 전달됨

비유

1
2
LangChain = 공장 컨베이어 벨트 (한 방향으로만 진행)
LangGraph  = 공장 전체 라인 (분기, 피드백, 재작업 가능)

3. 핵심 개념 — 노드, 엣지, 상태(State)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
노드(Node):
  - 그래프의 각 처리 단계
  - 파이썬 함수로 작성
  - 입력: 현재 state (dict)
  - 출력: 업데이트할 키-값 (dict)

엣지(Edge):
  - 노드 사이의 연결선
  - 일반 엣지: A → B (항상 B로)
  - 조건부 엣지: A → 조건 → B 또는 C

상태(State):
  - 그래프 전체에서 공유하는 딕셔너리
  - TypedDict로 타입 정의
  - 노드가 반환하는 딕셔너리로 덮어씌워짐 (기본값)

4. TypedDict로 State 정의

State에는 여러 노드가 공유해야 하는 값과 호출 결과를 구분해서 둔다. 모든 임시 변수를 상태에 넣기보다 다음 단계나 실행 기록에 필요한 항목인지 판단하면 노드 간 의존성이 읽기 쉬워진다.

1
2
3
4
5
6
7
8
9
10
11
12
13
from typing import TypedDict

class SimpleState(TypedDict):
    message: str

class QAState(TypedDict):
    question : str
    answer   : str

class TextState(TypedDict):
    text  : str
    upper : str
    length: int

TypedDict는 딕셔너리에 타입 힌트를 주는 방법
LangGraph가 State 구조를 파악하는 데 사용한다

노드 함수 기본 구조

1
2
3
4
5
def my_node(state: MyState) -> dict:
    # state에서 값 읽기
    value = state['key']
    # 업데이트할 키만 반환 (나머지는 그대로 유지됨)
    return {'key': new_value}

5. StateGraph 기본 패턴

최소 예제

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class MiniState(TypedDict):
    text: str

def upper_node(state: MiniState) -> dict:
    return {"text": state['text'].upper()}

def excite_node(state):
    return {"text": state['text'] + '!'}

builder = StateGraph(MiniState)
builder.add_node('upper', upper_node)
builder.add_node('excite', excite_node)
builder.add_edge(START, 'upper')
builder.add_edge('upper', 'excite')
builder.add_edge('excite', END)

app = builder.compile()
app.invoke({'text': 'hello'})
# → {'text': 'HELLO!'}

StateGraph 생성 단계

1
2
3
4
5
1. StateGraph(StateClass) — 상태 타입 지정
2. builder.add_node(name, fn) — 노드 등록
3. builder.add_edge(from, to) — 엣지 연결
4. builder.compile() — 앱 빌드
5. app.invoke({'state_key': initial_value}) — 실행

6. 다중 노드 직렬 연결

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class DocState(TypedDict):
    raw       : str
    cleaned   : str   # strip()
    translated: str   # upper()
    summary   : str   # [:5]
    log       : str

def clean(state: DocState) -> dict:
    return {'cleaned': state['raw'].strip(), 'log': state['log'] + 'clean,'}

def translate(state: DocState) -> dict:
    return {'translated': state['cleaned'].upper(), 'log': state['log'] + 'translate,'}

def summarize(state: DocState) -> dict:
    return {'summary': state['translated'][:5], 'log': state['log'] + 'summarize,'}

builder = StateGraph(DocState)
for n, fn in [('clean', clean), ('translate', translate), ('summarize', summarize)]:
    builder.add_node(n, fn)

builder.add_edge(START, 'clean')
builder.add_edge('clean', 'translate')
builder.add_edge('translate', 'summarize')
builder.add_edge('summarize', END)
app = builder.compile()

app.invoke({"raw": "  hello langgraph world  ", "cleaned": "", "translated": "", "summary": "", "log": ""})
# → {'raw': '  hello langgraph world  ',
#    'cleaned': 'hello langgraph world',
#    'translated': 'HELLO LANGGRAPH WORLD',
#    'summary': 'HELLO',
#    'log': 'clean,translate,summarize,'}

각 노드는 자신이 담당하는 키만 업데이트한다
나머지 키는 이전 상태 그대로 유지된다


7. LLM을 노드로 사용

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class QAState(TypedDict):
    question: str
    answer  : str

def llm_answer(state: QAState) -> dict:
    response = llm.invoke([HumanMessage(content=state['question'])])
    return {'answer': response.content}

builder = StateGraph(QAState)
builder.add_node('llm_answer', llm_answer)
builder.add_edge(START, 'llm_answer')
builder.add_edge('llm_answer', END)
app = builder.compile()

app.invoke({'question': 'langgraph를 한 문장으로?', 'answer': ''})
# → {'question': '...', 'answer': 'LangGraph는 ...'}

번역 노드 예시

1
2
3
4
5
6
7
8
9
10
11
12
13
class TransState(TypedDict):
    ko: str
    en: str

def translate(state: TransState) -> dict:
    response = llm.invoke([
        SystemMessage(content='translate to english, only english'),
        HumanMessage(content=state['ko'])
    ])
    return {'en': response.content}

app.invoke({"ko": "오늘 날씨가 좋네요.", "en": ""})
# → {'ko': '오늘 날씨가 좋네요.', 'en': 'The weather is nice today.'}

8. add_conditional_edges — 조건 분기

라우터는 다음 경로를 선택하는 규칙으로 읽을 수 있어야 한다. 경계값 바로 아래·위의 입력을 넣어 분기를 확인하고, 어떤 노드가 결과 필드를 만들어 주는지도 함께 따라간다.

라우터 함수가 반환하는 문자열에 따라 다른 노드로 분기한다.

기본 패턴

1
2
3
4
5
6
7
8
9
def router_fn(state) -> str:
    # 조건에 따라 다음 노드 이름(문자열) 반환
    return "node_a" if ... else "node_b"

builder.add_conditional_edges(
    '시작노드',
    router_fn,
    {'node_a': 'node_a', 'node_b': 'node_b'}  # 반환값 → 노드 이름 매핑
)

짝수/홀수 분기 예시

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class NumberState(TypedDict):
    n    : int
    label: str

def parity_router(state) -> str:
    return "even" if state['n'] % 2 == 0 else 'odd'

def even_handler(state): return {'label': f"{state['n']}은(는) 짝수입니다"}
def odd_handler(state):  return {'label': f"{state['n']}은(는) 홀수입니다"}
def start_n(state):      return {}

b = StateGraph(NumberState)
b.add_node('start_n', start_n)
b.add_node('even_handler', even_handler)
b.add_node('odd_handler',  odd_handler)

b.add_edge(START, 'start_n')
b.add_conditional_edges(
    'start_n', parity_router,
    {'even': 'even_handler', 'odd': 'odd_handler'}
)
b.add_edge('even_handler', END)
b.add_edge('odd_handler',  END)
app = b.compile()

app.invoke({'n': 7, 'label': ''})   # → {'n': 7, 'label': '7은(는) 홀수입니다'}
app.invoke({'n': 8, 'label': ''})   # → {'n': 8, 'label': '8은(는) 짝수입니다'}

3단계 분기 (성적 등급)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class GradeState(TypedDict):
    score: int
    grade: str

def grade_router(state) -> str:
    if state['score'] >= 90: return 'a'
    if state['score'] >= 70: return 'b'
    return 'c'

def grade_a(state): return {'grade': 'A'}
def grade_b(state): return {'grade': 'B'}
def grade_c(state): return {'grade': 'C'}
def start_grade(state): return {}

b = StateGraph(GradeState)
b.add_node('start_grade', start_grade)
b.add_node('grade_a', grade_a)
b.add_node('grade_b', grade_b)
b.add_node('grade_c', grade_c)

b.add_edge(START, 'start_grade')
b.add_conditional_edges(
    'start_grade', grade_router,
    {'a': 'grade_a', 'b': 'grade_b', 'c': 'grade_c'}
)
for n in ['grade_a', 'grade_b', 'grade_c']:
    b.add_edge(n, END)
app = b.compile()

for s in [95, 75, 50]:
    print(app.invoke({'score': s, 'grade': ''}))
# {'score': 95, 'grade': 'A'}
# {'score': 75, 'grade': 'B'}
# {'score': 50, 'grade': 'C'}

9. 여러 분기 합치기 (merge 노드)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class QuestionState(TypedDict):
    question: str
    qtype   : str
    answer  : str

def classify(state):
    q = state['question'].lower()
    if '?' not in q:                              return {'qtype': 'statement'}
    if any(w in q for w in ['언제', 'when']):    return {'qtype': 'when'}
    if any(w in q for w in ['어디', 'where']):   return {'qtype': 'where'}
    return {'qtype': 'what'}

def h_when(state):  return {'answer': '[시간] 답변'}
def h_where(state): return {'answer': '[장소] 답변'}
def h_what(state):  return {'answer': '[일반] 답변'}
def h_stmt(state):  return {'answer': '[진술문] 검색 불필요'}
def merge(state):   return {'answer': state['answer'] + '!!!!!!!'}

b = StateGraph(QuestionState)
b.add_node('classify', classify)
for name, fn in [('h_when', h_when), ('h_where', h_where), ('h_what', h_what), ('h_stmt', h_stmt)]:
    b.add_node(name, fn)
b.add_node('merge', merge)

b.add_edge(START, 'classify')
b.add_conditional_edges(
    'classify', lambda s: s['qtype'],
    {'when': 'h_when', 'where': 'h_where', 'what': 'h_what', 'statement': 'h_stmt'}
)
for n in ['h_when', 'h_where', 'h_what', 'h_stmt']:
    b.add_edge(n, 'merge')
b.add_edge('merge', END)
app = b.compile()

app.invoke({'question': "회의 어디서 해?", 'qtype': '', 'answer': ''})
# → {'question': '회의 어디서 해?', 'qtype': 'where', 'answer': '[장소] 답변!!!!!!!'}

10. 핵심 개념 총정리

StateGraph API 정리

메서드역할
StateGraph(StateClass)그래프 생성
.add_node(name, fn)노드 추가
.add_edge(from, to)일반 엣지 추가
.add_conditional_edges(from, router, map)조건부 엣지 추가
.compile()실행 가능한 앱으로 빌드
app.invoke(state)그래프 실행

전체 흐름 다이어그램

1
2
3
4
5
6
7
8
9
10
11
12
13
초기 상태 dict
    ↓
START
    ↓
노드 A (state 읽기 → dict 반환 → state 업데이트)
    ↓
[조건부] 라우터 → 노드 B 또는 노드 C
    ↓
merge 노드 (여러 분기 합치기)
    ↓
END
    ↓
최종 state dict 반환

11. 자주 나오는 실수/주의사항

주의 1: 변경하지 않은 state까지 다시 반환

전체 state 반환 자체가 항상 오류는 아니다. 다만 반환한 각 키는 업데이트로 처리되므로, 리스트를 누적하는 reducer가 있으면 기존 항목을 다시 추가할 수 있다. 원본 state를 직접 바꾸지 않고 변경할 키만 반환하면 업데이트 의도가 명확하다.

1
2
3
# 권장: 업데이트할 키만 반환
def my_node(state):
    return {'key': new_value}

실수 2: 진입점 없이 compile

1
2
3
# 그래프의 진입점을 설정한다.
builder.add_edge(START, '첫번째노드')
# builder.set_entry_point('첫번째노드')로 같은 진입점을 지정할 수도 있다.

실수 3: 조건부 엣지의 반환값과 매핑 불일치

1
2
3
4
5
6
7
8
9
10
11
# router 함수가 'even'을 반환하는데 매핑에 'even' 없으면 에러
b.add_conditional_edges(
    'start', parity_router,
    {'Even': 'even_handler', 'odd': 'odd_handler'}  # 'Even'은 대문자 → 매핑 불일치
)

# 올바른 코드
b.add_conditional_edges(
    'start', parity_router,
    {'even': 'even_handler', 'odd': 'odd_handler'}
)

주의 4: 종료 경로와 반복 경로 구분

END 엣지는 종료 의도를 명시하는 방법이다. 말단 노드에 END 엣지가 없다는 이유만으로 무한 실행되는 것은 아니다. 실행할 노드와 전달 중인 메시지가 더 없으면 그래프는 종료된다. 반복 엣지가 있는 그래프에서는 종료 조건에 따라 END로 빠져나가는 경로를 확인한다.

1
2
3
# 성적 등급 예제의 종료 경로를 명시
for n in ['grade_a', 'grade_b', 'grade_c']:
    b.add_edge(n, END)

실수 5: 값이 만들어지기 전에 state 키 읽기

TypedDict의 모든 키를 초기 입력에 넣을 필요는 없다. 노드가 나중에 만들어 주는 출력값은 생략할 수 있다. 단, 아직 없는 키를 state[키]로 읽으면 KeyError가 발생한다.

1
2
3
4
5
# 8절의 성적 등급 그래프를 compile한 app을 기준으로 한 호출
app.invoke({'score': 85})  # 정상: grade_b가 {'grade': 'B'}를 생성
# 결과: {'score': 85, 'grade': 'B'}

# 반대로 score 없이 {'grade': ''}만 넣으면 grade_router가 score를 읽을 때 오류

상태 업데이트와 종료 조건: LangGraph 공식 Graph API 설명


12. 그래프를 읽는 기준은 상태와 종료 조건

노드 이름만으로 흐름을 이해하기 어렵다면 각 노드의 읽기·쓰기 키를 함께 적어 본다. 여러 노드가 같은 값을 변경하는 지점과 입력이 만들어지기 전에 읽히는 지점을 찾기 쉬워진다.

그래프 사용 여부는 분기와 상태 추적이 필요한지로 결정한다. 단순한 직렬 함수 조합으로 충분한 흐름까지 바꾸기보다, 실패 경로와 재진입을 명시할 필요가 있는 부분부터 적용할 수 있다.

이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.

인기 태그