[LangGraph] 상태 업데이트 및 워크플로우 제어 (3)

랭그래프(LangGraph)를 실무에서 다루기 위해 필수적인 메시지 상태 관리, 그래프 실행 방식, 그리고 노드 간의 다양한 연결 형태(순차, 병렬, 조건 분기, 반복)를 구현하는 방법입니다.

1. 메시지 상태 업데이트와 누적 (State & add_messages)

에이전트 워크플로우에서는 대화 기록이 끊기지 않고 유지되어야 합니다. 단순히 TypedDict만 사용하면 이전 상태가 덮어씌워지지만, Annotated와 add_messages 리듀서를 활용하면 기존 대화 목록에 새 메시지가 계속 누적됩니다.

• HumanMessage: 사용자의 입력 메시지
• AIMessage: 모델(AI)의 출력 메시지

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

# [핵심] add_messages를 통해 기존 메시지 배열에 새 메시지가 자동으로 병합됨
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    extra_field: int

def node(state: State):
    # 이전 메시지를 직접 불러와 합칠 필요 없이, 새 메시지만 반환하면 알아서 누적됨
    new_message = AIMessage("안녕하세요! 무엇을 도와드릴까요?")
    return {"messages": new_message, "extra_field": 10}

graph_builder = StateGraph(State)
graph_builder.add_node("node", node)
graph_builder.set_entry_point("node")
graph = graph_builder.compile()

---

2. 그래프 실행 및 결과 확인 방법

그래프를 컴파일한 후 실행하는 방법은 크게 동기화, 비동기화, 스트리밍으로 나뉩니다.

• invoke: 하나의 요청을 끝까지 처리한 후 최종 결과만 반환합니다.
• ainvoke: invoke의 비동기 버전으로, 여러 요청을 동시에 처리할 때 유용합니다.
• stream: 처리되는 중간 과정을 실시간으로 확인합니다. 모드(stream_mode) 설정에 따라 출력 형태가 달라집니다.
  • values: 각 단계 완료 시점의 전체 상태 값을 출력합니다.
  • updates (기본값): 각 단계에서 새롭게 업데이트된 상태 값만 출력합니다.
  • messages: 생성되는 메시지 단위로 출력합니다.

input_message = {"role": "user", "content": "안녕하세요."}

# 1. 단일 실행 (최종 결과만 확인)
result = graph.invoke({"messages": [input_message]})

# 2. 스트리밍 실행 (업데이트된 내용만 확인)
for chunk in graph.stream({"messages": [input_message]}, stream_mode='updates'):
    for node, value in chunk.items():
        if node:
            print(f"[{node} 실행완료]")
        if "messages" in value:
            print(value['messages'].content)

---

3. 노드와 엣지 연결 (순차 및 병렬)

3-1. 순차적 연결 (Sequence)

작업이 일직선으로 진행되는 가장 기본적인 구조입니다. add_sequence를 사용하면 여러 노드를 한 번에 직렬로 연결할 수 있어 코드가 간결해집니다.

# 하나씩 수동으로 연결하는 대신, 리스트 형태로 전달하여 한 번에 순차 연결
graph_builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
graph_builder.add_edge(START, "step_1")
graph = graph_builder.compile()

3-2. 병렬 연결 (Parallel)

하나의 노드가 끝난 뒤, 여러 노드가 동시에 실행되도록 분기할 수 있습니다.

# 'a' 노드 실행 후 'b'와 'c' 노드가 병렬로 실행됨
graph_builder.add_edge("a", "b")
graph_builder.add_edge("a", "c")
# 'b'와 'c'가 각각 완료되면 'd' 노드로 모임
graph_builder.add_edge("b", "d")
graph_builder.add_edge("c", "d")

---

4. 조건부 라우팅과 반복(Loop) 제어

4-1. 조건부 엣지 (Conditional Edge)

특정 상태 값에 따라 다음에 실행할 노드를 결정합니다. 라우팅 함수를 만들어 add_conditional_edges에 넘겨줍니다.

# 라우팅 함수: state의 값을 확인하여 다음 경로를 리스트로 반환
def route_bc_or_cd(state: State):
    if state["which"] == "cd":
        return ["c", "d"]
    return ["b", "c"]

# 'a' 노드 이후에 route_bc_or_cd 함수의 반환값에 따라 분기
graph_builder.add_conditional_edges("a", route_bc_or_cd, ["b", "c", "d"])

4-2. 재귀와 반복 처리 (Recursion)

에이전트가 특정 목표를 달성할 때까지 노드를 반복(Loop)하도록 설계할 수 있습니다. 이때 무한 루프에 빠지는 것을 방지하기 위해 recursion_limit을 설정하고 에러를 예외 처리합니다.

# 상태 리스트의 길이가 7 미만이면 다시 'b'로 돌아가고, 아니면 종료(END)
def route(state: State):
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END

graph_builder.add_conditional_edges("a", route)

# 무한 루프 방지를 위한 제한 설정 및 예외 처리
from langgraph.errors import GraphRecursionError

try:
    graph.invoke({"aggregate": []}, config={"recursion_limit": 4})
except GraphRecursionError:
    print("Recursion Error: 반복 제한 도달. 강제 종료합니다.")

---

5. 사용자 입력 기반의 동적 제어 (Human-in-the-loop 준비)

LLM을 도입하기 전, 사용자의 콘솔 입력을 받아 조건부로 워크플로우를 제어하는 기본 로직입니다. 사용자가 "반복"이라는 키워드를 입력하면 재시도 노드로 이동하고, 그렇지 않으면 종료하는 흐름입니다.

class State(TypedDict):
    human_messages: Annotated[list[HumanMessage], add_messages]
    ai_messages: Annotated[list[AIMessage], add_messages]
    retry_num : int

# 조건 판별 함수: 마지막 사용자의 입력 메시지에 "반복"이 포함되어 있는지 확인
def route(state: State):
    if "반복" in state["human_messages"][-1].content:
        return "retry"
    else:
        return END

graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges("chatbot", route)
graph_builder.add_edge("retry", "chatbot")
graph = graph_builder.compile()