본문 바로가기

AI Agent

Thought, Action, Observation을 코드로 — LangGraph + ReAct 완전 정리

반응형

 

AI 에이전트를 만들다 보면 이런 상황이 생깁니다.

"LLM이 도구를 써야 할 때도 있고, 바로 답할 수 있을 때도 있는데 이걸 어떻게 처리하지?"

이걸 깔끔하게 해결하는 패턴이 ReAct이고, 이를 코드로 명시적으로 구현할 수 있게 해주는 프레임워크가 LangGraph입니다. 이번 글에서는 LangGraph가 뭔지부터 ReAct 패턴을 실제로 어떻게 구현하는지까지 정리해 드릴게요.


LangGraph란?

LangGraph는 LangChain 팀이 만든 상태 기반 AI 워크플로우 프레임워크예요. 2024년 1월에 출시됐으며, AI 에이전트의 복잡한 흐름을 그래프 구조로 명시적으로 표현할 수 있게 해줍니다.

기존 LangChain만으로도 에이전트를 만들 수 있었지만 세 가지 문제가 있었어요.

첫째, 루프 구현이 어렵습니다. "결과가 마음에 안 들면 다시 처리" 같은 반복 구조를 코드로 표현하기 복잡했어요. 둘째, 상태 관리가 불편합니다. 여러 단계를 거치는 처리에서 중간 상태를 추적하기 어려웠어요. 셋째, 흐름이 블랙박스입니다. AgentExecutor 안에서 뭐가 어떻게 돌아가는지 보기 어려웠어요.

LangGraph는 이 세 가지를 모두 해결합니다. 워크플로우를 **노드(Node)**와 **엣지(Edge)**로 명시적으로 표현하기 때문에 흐름이 투명하고, 루프와 분기도 자연스럽게 구현할 수 있어요.


LangGraph의 핵심 3요소

State (상태)

노드 간에 공유되는 데이터 저장소예요. 각 노드는 State를 받아서 처리하고, 업데이트된 State를 반환합니다.

from typing import TypedDict, List, Annotated
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], lambda x, y: x + y]

Annotated의 두 번째 인자가 **리듀서(Reducer)**예요. 새 메시지가 들어올 때 기존 리스트에 덧붙이라는 의미입니다. 리듀서가 없으면 새 값이 기존 값을 덮어쓰게 돼요.

Node (노드)

실제 작업을 수행하는 파이썬 함수예요. LLM 호출, 도구 실행, 데이터 처리 등 뭐든 노드가 될 수 있습니다. State를 받아서 업데이트된 State를 반환하는 구조예요.

Edge (엣지)

노드 간의 연결 관계예요. 세 가지 종류가 있어요.

  • 일반 엣지 — 항상 다음 노드로 이동
  • 조건부 엣지 — 함수 결과에 따라 다음 노드를 동적으로 결정
  • 순환 엣지 — 이전 노드로 다시 돌아가는 루프 구성

ReAct 패턴이란?

ReAct는 Reasoning and Acting의 약자예요. 2023년 논문에서 소개된 패턴으로, LLM이 추론과 행동을 반복하며 목표를 달성합니다.

사이클은 세 단계로 구성돼요.

Thought (생각) → Action (행동) → Observation (관찰) → 반복

Thought — LLM이 현재 상황을 분석하고 다음 행동을 결정합니다. Action — 결정한 행동을 실행해요. 도구 호출이거나 직접 답변이거나 둘 중 하나예요. Observation — 실행 결과를 받아서 다음 Thought의 input으로 넣습니다.

LangGraph에서는 이 사이클을 노드 → 조건부 엣지 → 노드 → 루프 구조로 명시적으로 표현할 수 있어요.


LangGraph로 ReAct 구현하기

1. 도구 및 LLM 설정

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import create_react_agent
from langgraph.prebuilt import ToolExecutor

@tool
def search(query: str) -> str:
    """Search for information online."""
    return f"Result for '{query}': Example search result."

tools = [search]
tool_executor = ToolExecutor(tools)

llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = hub.pull("hwchase17/react")
agent_runnable = create_react_agent(llm, tools, prompt)

2. 노드 함수 정의

LLM 노드 — Thought와 Action을 결정합니다.

def run_agent(state: AgentState):
    messages = state['messages']
    result = agent_runnable.invoke({
        "input": messages[-1].content,
        "chat_history": messages[:-1]
    })
    return {"messages": [result]}

도구 실행 노드 — Action을 실제로 실행하고 Observation을 생성합니다.

async def execute_tools(state: AgentState):
    messages = state['messages']
    last_message = messages[-1]

    tool_outputs = []
    for tool_call in last_message.tool_calls:
        try:
            output = await tool_executor.ainvoke(tool_call)
            tool_outputs.append(
                ToolMessage(
                    content=str(output),
                    name=tool_call.name,
                    tool_call_id=tool_call.id
                )
            )
        except Exception as e:
            tool_outputs.append(
                ToolMessage(
                    content=f"Error: {e}",
                    name=tool_call.name,
                    tool_call_id=tool_call.id
                )
            )
    return {"messages": tool_outputs}

3. 조건부 엣지 — ReAct의 핵심

LLM 응답을 보고 다음에 뭘 할지 결정하는 함수예요. 도구 호출이 있으면 도구 실행으로, 없으면 종료로 분기합니다.

def should_continue(state: AgentState) -> str:
    messages = state['messages']
    last_message = messages[-1]

    if last_message.tool_calls:
        return "continue_tool"   # 도구 실행 노드로
    else:
        return "end"             # 그래프 종료

4. 그래프 구성

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

workflow.add_node("agent", run_agent)
workflow.add_node("tools", execute_tools)

workflow.set_entry_point("agent")

# agent 노드에서 조건부 분기
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue_tool": "tools",
        "end": END
    }
)

# tools 노드 실행 후 다시 agent로 (루프)
workflow.add_edge('tools', 'agent')

app = workflow.compile()

실제 실행 흐름

"서울의 오늘 날씨는 어때?" 처럼 도구가 필요한 질문

1. HumanMessage 입력
2. agent 노드: LLM이 "검색이 필요하다" 판단 → tool_calls 포함한 AIMessage 생성
3. should_continue: tool_calls 있음 → "continue_tool" 반환
4. tools 노드: search 도구 실행 → ToolMessage 생성
5. agent 노드로 다시 이동 (루프)
6. agent 노드: 검색 결과 보고 최종 답변 생성 → tool_calls 없는 AIMessage
7. should_continue: tool_calls 없음 → "end" 반환
8. END

"세계에서 가장 높은 산은?" 처럼 바로 답할 수 있는 질문

1. HumanMessage 입력
2. agent 노드: LLM이 바로 답변 가능 판단 → tool_calls 없는 AIMessage 생성
3. should_continue: tool_calls 없음 → "end" 반환
4. END

도구가 필요하면 루프를 돌고, 필요 없으면 바로 종료하는 구조예요.


LangGraph의 추가 기능

체크포인트 (Checkpoint)

워크플로우 실행 중 상태를 저장하고 복원하는 기능이에요. 오류 발생 시 특정 지점에서 재개할 수 있습니다.

from langgraph.checkpoint.memory import MemorySaver

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

실제 서비스에서는 SqliteSaver나 PostgresSaver를 사용해요. MemorySaver는 데모용이에요.

Human-in-the-Loop

특정 노드에서 사람의 개입을 요청할 수 있어요. 민감한 작업을 실행하기 전에 사람이 승인하는 구조를 만들 때 유용합니다.

그래프 시각화

from IPython.display import Image
Image(app.get_graph().draw_mermaid_png())

복잡한 워크플로우도 한눈에 파악할 수 있어요.


LangGraph vs LangChain AgentExecutor

구분 LangChain AgentExecutor LangGraph

흐름 제어 블랙박스 명시적 노드/엣지
루프 구현 어려움 자연스러움
상태 관리 불편함 State로 명확하게
디버깅 어려움 시각화 지원
유연성 제한적 높음

마무리

LangGraph의 핵심은 "AI 워크플로우를 코드로 명시적으로 표현한다" 는 거예요.

ReAct 패턴도 결국 Thought → Action → Observation 루프를 노드와 조건부 엣지로 표현한 것입니다. 흐름이 눈에 보이니 디버깅하기 쉽고, 어디서 문제가 생겼는지 바로 파악할 수 있어요.

간단한 에이전트는 create_react_agent로 빠르게 만들고, 복잡한 흐름이 필요하면 StateGraph로 직접 노드와 엣지를 구성하는 방식을 추천드립니다. 😄


 

반응형