포스트

2026-04-30 강의 정리

LangGraph 활용 - ReAct 에이전트 활용

게시
By kimseokryu
17 분읽는 시간
2026-04-30 강의 정리

목차

  1. 문제와 접근
  2. ReAct 패턴 개념
  3. 수동 ReAct 에이전트 구현
  4. ToolNode + tools_condition (prebuilt)
  5. LLM 분류 라우터 에이전트
  6. MemorySaver — 대화 기억 (체크포인트)
  7. thread_id로 사용자 분리
  8. get_state / get_state_history
  9. update_state — 수동 상태 수정
  10. interrupt_before — Human-in-the-Loop (HITL)
  11. HITL에서 도구 실행 거부
  12. 핵심 개념 총정리
  13. 자주 나오는 실수/주의사항
  14. 재개 가능한 실행의 조건

1. 문제와 접근

에이전트 실행은 모델과 도구를 번갈아 호출하는 흐름에 상태 저장과 중단·재개를 더한 문제로 볼 수 있다. 수동 구현과 제공되는 노드를 비교하면서 어디까지 공통 동작을 맡기고 어떤 정책을 직접 관리할지 살펴본다. 체크포인트와 사용자 구분도 각각의 책임을 나누어 본다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import os
from typing import TypedDict, Annotated, Literal
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver

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

2. ReAct 패턴 개념

ReAct = Reasoning + Acting

1
2
3
4
5
6
7
8
9
10

[ReAct 루프]
START
  ↓
agent 노드 — LLM이 생각하고 tool_calls 결정
  ↓ tool_calls 있음
tools 노드 — 실제 도구 실행
  ↓
agent 노드 — 결과를 보고 다시 생각
  ↓ tool_calls 없음 (최종 답변)
END

LLM이 “생각(Reasoning)” → “행동(Acting)” → “관찰” → “다시 생각” 을 반복
이전 tool_loop와 달리, LangGraph는 이 루프를 그래프로 표현한다

3. 수동 ReAct 에이전트 구현

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
37
38
39
40
41
42
43
44

@tool
def add(a: int, b: int) -> int:
    """두 정수를 더한다"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """두 정수를 곱한다"""
    return a * b

tools          = [add, multiply]
llm_with_tools = llm.bind_tools(tools)
tool_map       = {t.name: t for t in tools}

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

def agent_node(state):
    return {'messages': [llm_with_tools.invoke(state['messages'])]}

def tools_node(state):
    out = []
    for c in state['messages'][-1].tool_calls:
        out.append(ToolMessage(
            content      = str(tool_map[c['name']].invoke(c['args'])),
            tool_call_id = c['id']
        ))
    return {'messages': out}

def should_continue(state):
    return "tools" if state['messages'][-1].tool_calls else END

b = StateGraph(AgentState)
b.add_node('agent', agent_node)
b.add_node('tools', tools_node)

b.add_edge(START, 'agent')
b.add_conditional_edges('agent', should_continue, {'tools': 'tools', END: END})
b.add_edge('tools', 'agent')   # 도구 실행 후 다시 agent로
react_app = b.compile()

r = react_app.invoke({'messages': [HumanMessage(content="(3+5)*4 는?")]})
r['messages'][-1].content
# → '(3 + 5) * 4 의 결과는 32입니다.'

실행 메시지 흐름 확인

1
2
3
4
5
6
7
8

[type(m).__name__ for m in r['messages']]
# → ['HumanMessage', 'AIMessage', 'ToolMessage', 'AIMessage', 'ToolMessage', 'AIMessage']
# 1. 사용자 질문
# 2. LLM: add(3, 5) 호출
# 3. ToolMessage: 8
# 4. LLM: multiply(8, 4) 호출
# 5. ToolMessage: 32
# 6. LLM: 최종 답변

4. ToolNode + tools_condition (prebuilt)

수동 구현 대신 LangGraph 내장 도구 노드를 사용한다.

1
2
3
4
5
6
7
8
9
10

from langgraph.prebuilt import ToolNode, tools_condition

b2 = StateGraph(AgentState)
b2.add_node('agent', agent_node)
b2.add_node('tools', ToolNode(tools))     # 수동 tools_node 대신

b2.add_edge(START, 'agent')
b2.add_conditional_edges('agent', tools_condition)   # should_continue 대신
b2.add_edge('tools', 'agent')
auto_app = b2.compile()

ToolNode: 자동으로 tool_calls를 읽어 도구 실행 + ToolMessage 생성
tools_condition: tool_calls 있으면 ‘tools’, 없으면 END 반환

수동 vs prebuilt 비교

1
2
3
4
5
6
7

[수동 구현]
  def tools_node(state): ...      # 직접 tool_calls 순회
  def should_continue(state): ... # 직접 조건 작성

[prebuilt]
  ToolNode(tools)     # tool_calls 자동 처리
  tools_condition     # 조건 자동 처리

세 가지 도구 에이전트 예시

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

@tool
def substract(a, b):
    """a에서 b를 뺀다"""
    return a - b

three_tools = [add, multiply, substract]
three_llm   = llm.bind_tools(three_tools)

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

tb = StateGraph(ThreeState)
tb.add_node('agent', lambda s: {'messages': [three_llm.invoke(s['messages'])]})
tb.add_node('tools', ToolNode(three_tools))
tb.add_edge(START, 'agent')
tb.add_conditional_edges('agent', tools_condition)
tb.add_edge('tools', 'agent')
three_app = tb.compile()

r = three_app.invoke({'messages': [HumanMessage(content='100에다가 5를 더한 뒤, 그 결과에서 2를 빼고, 그 결과에 3을 곱하면?')]})
print(r['messages'][-1].content)
# → '계산 결과는 309입니다.' ((100+5-2)*3 = 309)

5. LLM 분류 라우터 에이전트

LLM으로 질문 유형을 분류하고, 유형에 맞는 노드로 분기한다.

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
37
38
39
40
41
42
43

class QState(TypedDict):
    question: str
    qtype   : Literal['article', 'case', 'term', 'general']
    answer  : str

def classify(state: QState) -> dict:
    response = llm.invoke([
        SystemMessage(content="""질문을 다음 4가지 카테고리 중 하나로 분류하세요:
        - article  : 법률 조문 (~조, ~항)에 대한 질문
        - case     : 판례 번호나 특정 사건에 대한 질문
        - term     : 법률 용어 정의나 개념 설명 질문
        - general  : 인사, 일상 대화 등 기타 질문
        
        답변은 반드시 하나의 단어('article', 'case', 'term', 'general')로만 하세요"""),
        HumanMessage(content=state['question'])
    ])
    qtype = response.content.strip().lower()
    if qtype not in ['article', 'case', 'term', 'general']:
        qtype = 'general'
    return {'qtype': qtype}

def article_search(s): return {'answer': f'[조문 검색] {s["question"]}'}
def case_search(s):    return {'answer': f'[판례 검색] {s["question"]}'}
def term_explain(s):   return {'answer': f'[용어 설명] {s["question"]}'}
def general_chat(s):   return {'answer': f'[일반 대화] {s["question"]}'}

qb = StateGraph(QState)
qb.add_node('classify', classify)
for n, fn in [('article', article_search), ('case', case_search),
              ('term', term_explain),       ('general', general_chat)]:
    qb.add_node(n, fn)

qb.add_edge(START, 'classify')
qb.add_conditional_edges(
    'classify', lambda s: s['qtype'],
    {'article': 'article', 'case': 'case', 'term': 'term', 'general': 'general'}
)
for n in ['article', 'case', 'term', 'general']:
    qb.add_edge(n, END)
q_app = qb.compile()

q_app.invoke({'question': '민법 제760조가 뭔가요?', 'qtype': 'general', 'answer': ''})
# → {'qtype': 'article', 'answer': '[조문 검색] 민법 제760조가 뭔가요?'}

6. MemorySaver — 대화 기억 (체크포인트)

메모리에 상태를 저장하는 예제는 같은 프로세스 안에서 실행 흐름을 확인하는 용도로 본다. 프로세스가 다시 시작된 뒤에도 이어야 한다면 저장소의 지속성과 복구 동작을 별도로 확인해야 하며, thread_id를 사용한다는 사실만으로 접근 권한이 확인되는 것은 아니다.

그래프 실행 상태를 저장해서 다음 실행 시 이전 대화를 기억한다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14

from langgraph.checkpoint.memory import MemorySaver

memory  = MemorySaver()
mem_app = b2.compile(checkpointer=memory)   # checkpointer 추가

# thread_id로 대화 세션 구분
cfg = {'configurable': {'thread_id': 'alice'}}

mem_app.invoke({'messages': [HumanMessage(content='내 이름은 앨리스야')]}, config=cfg)
# → '안녕하세요, 앨리스! 만나서 반가워요.'

r = mem_app.invoke({'messages': [HumanMessage(content='내 이름이 뭐였지?')]}, config=cfg)
r['messages'][-1].content
# → '앨리스님이신데요!'  (이전 대화 기억)

MemorySaver는 인메모리 저장 (프로세스 종료 시 사라짐)
영구 저장은 SqliteSaver, PostgresSaver 등 사용

7. thread_id로 사용자 분리

같은 앱에서 여러 사용자의 대화를 독립적으로 관리한다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 사용자별 대화 초기화
prefs = {'u1': '피자', 'u2': '초밥', 'u3': '떡볶이'}
for uid, food in prefs.items():
    mem_app.invoke(
        {'messages': [HumanMessage(content=f'내가 좋아하는 음식은 {food}야')]},
        config={'configurable': {'thread_id': uid}}
    )

# 사용자별 대화 조회 (각자 자기 음식만 기억)
for uid in prefs:
    r = mem_app.invoke(
        {'messages': [HumanMessage(content='내가 좋아하는 음식은?')]},
        config={'configurable': {'thread_id': uid}}
    )
    print(f"[{uid}] {r['messages'][-1].content}")

# [u1] 당신이 좋아하는 음식은 피자입니다!
# [u2] 당신이 좋아하는 음식은 초밥입니다!
# [u3] 당신이 좋아하는 음식은 떡볶이입니다!

8. get_state / get_state_history

현재 상태 조회

1
2
3
4
5

cfg_a = {'configurable': {'thread_id': 'alice'}}
snap = mem_app.get_state(cfg_a)

len(snap.values['messages'])   # → 저장된 메시지 수
snap.next                      # → () (더 실행할 노드 없음)

과거 상태 이력 조회

1
2
3
4

history = list(mem_app.get_state_history(cfg_a))
len(history)   # → 실행 횟수에 따라 다름

history[-1].values.get('messages', [])   # → 가장 오래된 상태

9. update_state — 수동 상태 수정

외부에서 상태를 직접 주입할 수 있다.

1
2
3
4
5
6
7
8
9
10
11
12

cfg_x = {'configurable': {'thread_id': 'manual'}}

# 메시지를 외부에서 주입
mem_app.update_state(cfg_x, {'messages': [HumanMessage(content='안녕!')]})

snap = mem_app.get_state(cfg_x)
snap.next   # → ('agent',)  — agent 노드가 실행 대기 중

# None을 invoke하면 저장된 상태부터 이어서 실행
r = mem_app.invoke(None, config=cfg_x)
r['messages'][-1].content
# → '안녕하세요! 어떻게 도와드릴까요?'

10. interrupt_before — Human-in-the-Loop (HITL)

중단 위치를 정할 때는 사람이 확인해야 하는 값과 이미 실행된 작업을 구분한다. 재개 후 도구가 다시 실행될 가능성을 확인하고, 외부 상태를 바꾸는 작업은 같은 요청이 반복돼도 어떤 결과가 되는지 검토한다.

특정 노드 실행 전에 일시 정지해서 사람이 확인/승인할 수 있게 한다.

1
2
3
4
5
6
7
8
9

hitl_app = b2.compile(
    checkpointer  = MemorySaver(),
    interrupt_before = ['tools']   # tools 노드 실행 전 정지
)

cfg_h = {'configurable': {'thread_id': 'hitl-1'}}
hitl_app.invoke({'messages': [HumanMessage(content='100 더하기 200은?')]}, config=cfg_h)

# → agent가 add(100, 200) 호출을 결정했지만, tools 노드 실행 전 정지

정지 상태 확인

1
2
3
4
5
6
7

state = hitl_app.get_state(cfg_h)
state.next   # → ('tools',)  — tools 노드가 실행 대기 중

# 대기 중인 tool_call 확인
pending = state.values['messages'][-1].tool_calls[0]
pending
# → {'name': 'add', 'args': {'a': 100, 'b': 200}, 'id': 'call_xxx', 'type': 'tool_call'}

승인 (None invoke)

1
2
3
4

# 사람이 확인 후 승인 → None으로 이어서 실행
result = hitl_app.invoke(None, config=cfg_h)
result['messages'][-1].content
# → '100 더하기 200은 300입니다.'

흐름 다이어그램

1
2
3
4
5
6
7
8

invoke(질문) → agent 실행 → tool_calls 결정
               ↓
         [interrupt_before tools]
               ↓
         사람이 get_state로 확인
               ↓
         승인: invoke(None) → tools 실행 → agent → END
         거부: update_state → invoke(None) → 수정된 결과

11. HITL에서 도구 실행 거부

사람이 도구 실행을 거부하고 직접 답변을 주입한다.

1
2
3
4
5
6
7
8
9
10
11
12

cfg_r = {'configurable': {'thread_id': 'hhhh'}}
hitl_app.invoke({'messages': [HumanMessage(content='50 곱하기 4는?')]}, config=cfg_r)
# → agent가 multiply(50, 4) 결정 후 interrupt (정지)

# 사람이 거부 → AIMessage로 대체 답변 주입
hitl_app.update_state(cfg_r, {
    'messages': [AIMessage(content='죄송합니다. 정책상 도구 실행을 거부합니다')]
})

r = hitl_app.invoke(None, config=cfg_r)
r['messages'][-1].content
# → '죄송합니다. 정책상 도구 실행을 거부합니다'

12. 핵심 개념 총정리

LangGraph 에이전트 컴포넌트

컴포넌트역할대안
agent_nodeLLM 호출, tool_calls 결정-
tools_node (수동)tool_calls 순회 실행ToolNode(tools)
should_continue분기 조건tools_condition
MemorySaver대화 기억 (인메모리)SqliteSaver 등
interrupt_before특정 노드 전 정지interrupt_after

전체 ReAct 에이전트 구조

1
2
3
4
5
6
7

START
  ↓
agent (LLM)
  ↓ tool_calls 있음          ↓ 없음
tools (ToolNode)             END
  ↓
agent (반복)

config 구조

1
2
3
4
5

config = {
    'configurable': {
        'thread_id': 'unique-session-id'  # 필수 (MemorySaver 사용 시)
    }
}

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

실수 1: checkpointer 없이 thread_id 전달

1
2
3
4
5
6
7

# checkpointer 없으면 thread_id 무시됨 (메모리 저장 안 됨)
app = b.compile()   # checkpointer 없음
app.invoke(messages, config={'configurable': {'thread_id': 'abc'}})
# → 매 호출마다 독립적으로 실행, 이전 대화 기억 못 함

# 올바른 코드
app = b.compile(checkpointer=MemorySaver())

실수 2: interrupt 후 None이 아닌 새 메시지로 invoke

1
2
3
4
5

# 틀린 코드 → 새 입력이 들어와서 중단점 이전부터 다시 실행
hitl_app.invoke({'messages': [HumanMessage(content='계속해')]}, config=cfg_h)

# 올바른 코드 → None을 전달해야 중단점 이후부터 이어서 실행
hitl_app.invoke(None, config=cfg_h)

실수 3: tools_condition과 수동 should_continue 혼용

1
2
3
4
5
6
7

# ToolNode를 사용할 때는 tools_condition도 함께 사용
b.add_node('tools', ToolNode(tools))
b.add_conditional_edges('agent', tools_condition)   # 올바름

# 수동 tools_node를 사용할 때는 should_continue와 함께
b.add_node('tools', tools_node)
b.add_conditional_edges('agent', should_continue, {'tools': 'tools', END: END})

실수 4: agent 노드의 마지막 메시지에 tool_calls가 없는데 tools 노드 접근

1
2
3
4
5

# tools_node에서 state['messages'][-1].tool_calls 접근 시
# 마지막 메시지가 HumanMessage면 .tool_calls 속성 없음 → AttributeError
# → should_continue / tools_condition이 먼저 분기하므로
#   tools_node는 항상 AIMessage가 마지막인 상태에서 실행됨
# → 하지만 구조가 달라지면 에러 가능 → 방어 코드 고려

실수 5: MemorySaver를 여러 앱에서 공유

1
2
3
4
5
6
7
8
9

# 같은 MemorySaver 인스턴스를 여러 앱이 공유하면
# thread_id 충돌로 다른 앱의 대화가 섞일 수 있음
memory = MemorySaver()
app1 = graph1.compile(checkpointer=memory)
app2 = graph2.compile(checkpointer=memory)  # 같은 memory → 충돌 위험

# 앱마다 별도 MemorySaver 사용 권장
app1 = graph1.compile(checkpointer=MemorySaver())
app2 = graph2.compile(checkpointer=MemorySaver())

14. 재개 가능한 실행의 조건

정상 실행뿐 아니라 중단 직전과 재개 직후의 상태를 비교한다. 어떤 도구까지 실행됐고 어떤 결과가 저장됐는지 설명할 수 있어야 중복 실행과 누락을 구분할 수 있다.

수동 루프와 제공 노드 중 무엇을 사용할지는 공통 처리가 줄어드는 이점과 필요한 실행 정책을 표현할 수 있는지로 판단한다. 상태 저장, 호출 한도, 도구 실패 처리처럼 애플리케이션이 책임질 부분은 따로 남는다.

모두의연구소, AI/LLM 서비스 개발 과정
AI LLM
이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.

인기 태그

AI LLM