스타트업 인턴 일기 [04화]

스타트업 인턴 일기 [03화]에서 이어집니다..
https://jhklee-coder.tistory.com/93

오늘은 LangGraph AI 프레임워크의 핵심 툴 가운데 StateGraph라는 개념과 Conditional_edge 의 개념에 대해서 다뤄보려 한다. 

LangGraph - 'StateGraph'

 

 LinkBrain 프로젝트의 내부 LangGraph 파이프라인에서 동작하는 StateGraph는 LLM과의 핵심 소통 도구로써 크게 3가지의 핵심 내부 요소들로 이루어져있다.

 

 현 상태를 나타내는 States, 특정 로직을 담당하는 Nodes, 그리고 Node들끼리의 연결을 담당하는 Edge다. 간단히 말해 그래프 형태의 데이터 구조를 띄고 있는 일종의 전체 애플리케이션 상태를 나타내는 공유 데이터 structure라고 할 수 있다. 

 

1️⃣ StateGraph?

요소	정의	역할
State	애플리케이션의 현재 상태를 나타내는 공유 데이터 구조	전체 워크플로우의 컨텍스트 유지; 노드 간 정보 공유; TypedDict 또는 Pydantic BaseModel로 정의
Nodes	실제 작업을 수행하는 Python 함수들	특정 로직 수행 (예: LLM 호출, 도구 사용); 상태 업데이트; 입력으로 현재 상태를 받고, 업데이트된 상태 반환
Edges	노드 간의 연결을 정의하는 요소	워크플로우의 흐름 제어; 조건부 라우팅 가능; 다음에 실행할 노드 결정

 

✅ State (상태): 통합 애플리케이션 내 LLM 프로세스가 마주한 현재 상황을 나타내는 중앙 관제 시스템.

예제 코드)

from langgraph.graph import StateGraph
from typing import TypedDict, List, Annotated
import Operator

class State(TypedDict): 
    # State를 통한 현재 상태 정의
    input: str
    all_actions: Annotated[List[str], operator.add]

graph = StateGraph(State)

Ex) 공장 : 생산 라인의 진행 상황, 재고 현황, 직원 배치 등 공장 운영에 필요한 모든 정보가 실시간으로 업데이트된다. 즉 그래프 내에서의 상태가 모든 직원들에게 공유되게끔 하는 상태 관리 시스템.

✅ Nodes (노드): 특정 작업을 위한 단일 스테이션.

예제 코드)

graph.add_node("model", model)
graph.add_node("tools", tool_executor)

# from langgraph.graph import END => 종료 시점을 알리기 위한 END 스테이션

Ex) 공장 : 엔진 조립 스테이션, 차체 도장 스테이션, 품질 검사 스테이션 등이 있을 수 있다. 각 스테이션은 자신의 고유한 작업을 수행하고 각 단계에서 공정을 거친 생산품은 다음 노드로 계속해서 이동하게 된다.

✅ Edges (엣지): 단일 작업 스테이션 간의 연결 파이프라인이자 정보 전달 시스템.

예제 코드)

# 일반 엣지 
graph.add_edge("tools", "model")

# 조건부 엣지
graph.add_conditional_edge(
    "model",
    should_continue,
    {
        "end": END,
        "continue": "tools"
    }
)

Ex) 공장 : 한 스테이션의 작업이 끝난 뒤 다음 스테이션으로 제품이나 정보를 전달하는 파이프라인. 즉 노드와 노드를 연결하는 개념으로 생각할 수 있다. 

 

✅ Compile (컴파일) : 단일 작업 스테이션 연결과 상태 시스템의 처음부터 끝까지 이어서 Graph 구조로 완성시키는 역할. 

예제 코드)

graph = GraphState(State)

app = graph.compile()

 

 

여기서 우린 조건부로 동작하는 Conditional Edge 에 대해 좀 더 자세히 살펴볼 것이다!! 

 

 

2️⃣ 조건부 엣지 (Conditional Edge) 

📌 Edges (엣지)

우선 기본적인 Normal 형태의 엣지에 대해 먼저 살펴보자. 엣지(Edge)는 그래프의 논리가 어떻게 라우팅되는지(경로가 결정되는지)와 그래프가 언제 멈추는지를 정의한다. 이는 에이전트(agent)의 작동 방식과 서로 다른 노드(node) 간의 통신 방식에서 굉장히 중요한 요소로, 엣지의 주요 요소는 다음과 같다. 

🔸 일반 엣지 (Normal Edges)

한 노드에서 다음 노드로 직접 이동하는 엣지

# "node_a" -> "node_b"로 이동하는 엣지

graph.add_edge("node_a", "node_b")

 

🔸엔트리 포인트 (Entry Point)

사용자 입력이 도착했을 때, 가장 먼저 호출되는 노드를 지정하는 역할.

from langgraph.graph import START

# 초기 엔트리 포인트 : START
graph.add_edge(START, "node_a")

여기서 일반 엣지의 특징은 다음과 같다.  

• 하나의 노드는 여러 개의 출력(Out-going) 엣지를 가질 수 있다.
• 만약 하나의 노드가 여러 개의 출력 엣지를 가진 경우, 그 모든 대상 노드(destination nodes)가 병렬(parallel)로 실행되며, 이는 다음 슈퍼스텝(superstep)의 일부가 된다.

 

즉, 엣지는 그래프의 흐름을 결정하는 중요한 요소이며, 병렬 실행을 통해 효율적인 데이터 흐름을 만들 수 있다. 그렇다면 조건부 엣지(Conditional Edge)는 어떤 특징을 갖고 있을까?

 

 

📌 Conditional Edge (조건부 엣지)

특정 함수(노드)를 호출하여 다음으로 이동할 노드(들)를 결정하는 엣지

# "node_a" -> 특정 조건에 따라 routing_function으로 이동하느 조건부 엣지

graph.add_conditional_edges("node_a", routing_function)

🔸 조건부 엔트리 포인트 (Conditional Entry Point)

특정 함수를 호출하여 사용자 입력이 도착했을 때 가장 먼저 호출할 노드(들)를 결정하는 방식.

from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)

 

설명 없이 일방적인 코드 나열만 하면 이해가 다소 어려울 것 같아 최근 LangChain LinkedIn에서 공개한 LangGraph 파이프라인을 직접 구현하면서 설명을 이어보려 한다!

 

간단 파이프라인 구현 예시

https://diamantai.substack.com/ langgraph 파이프라인 예시

 

해당 LangGraph 파이프라인에서 conditional_edge 를 한번 추가해보자. 우리는 해당 파이프라인에서 불필요한 실행을 방지하고, 실행 흐름을 동적으로 변경하기 위해 article_count_check 라는 조건부 엣지를 추가할 것이다. 

 

✅ 기사 개수가 너무 적을 때 요약 단계 실행 불필요

• 기사 개수가 5개 이상일 때만 요약(summarization)을 수행하는 것이 적절하다는 결론.
• 하지만 일반적인 엣지(add_edge())를 사용하면 기사 개수가 부족해도 불필요한 요약 단계가 실행됨.
• article_count < 5인 경우에는 요약할 데이터가 부족하므로 요약 단계를 생략해야 함.

✅ 조건부 엣지가 없으면?

• select_top_urls → summarize_articles_parallel → format_results로 항상 고정된 경로를 실행해야 함.
• 하지만 기사 개수가 충분하지 않다면 굳이 summarize_articles_parallel을 실행할 필요 없음.

from langgraph.graph import StateGraph, END, START
from langgraph.prebuilt import tools_condition, ToolNode

# === 상태(State) 관리 클래스 정의 ===
class State:
    def __init__(self):
        self.article_count = 0  # 기사 개수
        self.selected_urls = []  # 선택된 URL 목록

# === 그래프 생성 ===
state = StateGraph(State)

# === 일반 노드 (ToolNode) 정의 ===
def generate_newsapi_params(state):
    print("Executing: generate_newsapi_params")
    return state

def retrieve_articles_metadata(state):
    print("Executing: retrieve_articles_metadata")
    return state

def retrieve_articles_text(state):
    print("Executing: retrieve_articles_text")
    return state

def select_top_urls(state):
    print("Executing: select_top_urls")
    return state

def summarize_articles_parallel(state):
    print("Executing: summarize_articles_parallel")
    return state

def format_results(state):
    print("Executing: format_results")
    return state

# === 노드 추가 ===
state.add_node("generate_newsapi_params", generate_newsapi_params)
state.add_node("retrieve_articles_metadata", retrieve_articles_metadata)
state.add_node("retrieve_articles_text", retrieve_articles_text)
state.add_node("select_top_urls", select_top_urls)
state.add_node("summarize_articles_parallel", summarize_articles_parallel)
state.add_node("format_results", format_results)

# === 일반 엣지 추가 ===
state.add_edge(START, "generate_newsapi_params")
state.add_edge("generate_newsapi_params", "retrieve_articles_metadata")
state.add_edge("retrieve_articles_metadata", "retrieve_articles_text")
state.add_edge("retrieve_articles_text", "select_top_urls")

# === 조건부 엣지 정의 & 추가 ===
def article_count_check(state):
    """ 기사 개수가 5개 이상인지 검사하는 함수 """
    return state.article_count >= 5

workflow.add_conditional_edges(
    "select_top_urls",
    tools_condition(article_count_check),
    {"true": "summarize_articles_parallel", "false": END}  # True → summarize, False → 종료
)

# === 후속 엣지 추가 ===
state.add_edge("summarize_articles_parallel", "format_results")
state.add_edge("format_results", END)

# === 실행 예제 ===
print("\n[실행 예제 - 기사 개수가 5개 이상]")
new_state = State()
new_state.article_count = 5  # 기사 개수 5개 설정
state.run(new_state)

print("\n[실행 예제 - 기사 개수가 4개일 때]")
new_state = State()
new_state.article_count = 4  # 기사 개수 4개 설정
state.run(new_state)

🔸 State 흐름

• select_top_urls 노드 실행 후, article_count_check(state) 함수를 실행
• state.article_count >= 5이면 "summarize_articles_parallel"로 이동 (True 경로)
• state.article_count < 5이면 END로 이동 (False 경로 → 종료)

🔸Conditional Edge의 주요 기능

조건 (article_count_check(state))	이동 노드
True (article_count >= 5)	summarize_articles_parallel
False (article_count < 5)	END (그래프 종료)

 

결국 조건부 라우팅, 조건부 엣지는 다음과 같은 이유에서 사용한다고 볼 수 있다. 

 

✅ 유연성: 다양한 사용자 입력에 대응하여 적절한 처리 경로를 선택.

✅ 효율성: 필요한 경우에만 특정 노드나 기능을 실행하여 리소스를 절약.

✅ 복잡한 워크플로우: 다단계 의사결정 프로세스를 구현.

✅ 사용자 경험 개선: 상황에 따라 다른 응답을 제공하여 더 자연스러운 대화가 가능.

 

 

 

 

 이번 인턴 일기는 LinkBrain 프로젝트 진행 가운데 공부했던 LangGraph의 핵심 모듈에 대해 다루어보았다. 그 가운데서도 특히 StateGraph를 통해 프로세스의 현 State와 metadata를 관리하는 LangGraph의 방식, 그리고 조건부 라우팅을 가능하게 하는 Conditional Edge에 대해 소개했다. 

 

 아쉽게도 LangGraph Studio와 같은 GUI 는 windows에서 지원하지 않아 내부 프로세스에 대한 Graph 구조를 동적으로 확인하는 것이 조금 어렵긴 하지만, 다른 패키지들을 활용하여 조금씩 프레임워크에 익숙해지고  있다. (새로운 개념이 등장할 때마다 넘넘 재밌다!!!)

 

 

(인턴 일기 5화에서 계속됩니다...)

3️⃣ Reference

LangGraph for Beginners, Part 3: Conditional Edges

LangGraph Glossary

1-3. StateGraph 이해하기