인턴 일기 [04화]
jo._.on_
·
2025. 5. 29.
"인턴 일기 [03화]에서 이어집니다."
안녕하세요 인턴 jo._.on_ 입니다. LinkBrain 팀에 합류한지도 어느덧 6개월이 다 되어가고 있습니다. 어쩜 이렇게 시간이 빠른지…😖😖 그래도 마지막까지 블로그에 소홀해져선 안되겠죠!! LangGraph를 사용한 그래프 설계도에 익숙해지고 있는 요즘, LinkBrain의 내부 파이프라인을 하나하나 직접 뜯어보는 재미가 쏠쏠합니다. 이번엔 그 중에서도 LangGraph의 특히 중요한 핵심 모듈 StateGraph와 Conditional_Edge라는 개념에 대한 [기술편]을 정리해보려고 해요.
그럼 이번 인턴 일기 04화 [기술편]도 시작해보겠습니다.
1️⃣ StateGraph란?
StateGraph는 LLM과 상호작용하는 우리의 주요 소통 창구로, 전체 플로우를 구성하는 세 가지 주요 요소로 이루어져 있어요.
바로 State (상태), Nodes (작업), 그리고 Edges (연결)입니다.
출처 : LangGraph - 'StateGraph'
LangGraph에서 StateGraph는 LLM 기반 워크플로우를 구성하는 핵심 구조라고 할 수 있습니다. 전체 애플리케이션의 상태를 하나의 그래프로 정의하고, 흐름에 따라 다양한 작업을 수행할 수 있게 돕는 역할을 하죠. 쉽게 말해, LangGraph에서 상태와 로직, 그리고 흐름 제어를 모두 담당하는 공유 데이터 구조라고 보면 됩니다.
StateGraph는 다음과 같은 세 가지 주요 구성요소로 이루어져 있습니다.
요소 | 정의 | 역할 |
State | 현재 애플리케이션의 상태를 나타내는 공유 데이터 구조 | 전체 컨텍스트를 유지하고, 노드 간 정보를 전달하는 역할을 합니다. 일반적으로 |
Nodes | 실제 작업을 수행하는 Python 함수들 | 특정 기능을 실행하며 상태를 업데이트합니다. 입력값으로 상태를 받고, 수정된 상태를 반환하는 방식이죠. |
Edges | 노드 간의 흐름을 연결하는 요소 | 그래프의 전반적인 동선을 결정하고, 조건에 따라 분기하는 흐름도 만들 수 있습니다. 다음에 어떤 노드를 실행할지 정해주는 역할이죠. |
✅ State (상태)
State는 말 그대로 현재 프로세스의 상황을 표현하는 데이터 구조입니다. LLM 프로세스가 어떤 입력을 받고 어떤 작업을 수행했는지를 기록하고, 이 정보를 모든 노드에서 공유할 수 있게 하죠.
State는 말 그대로 지금 LLM이 마주한 상황을 담고 있는 '중앙 정보 저장소'라고 할 수 있습니다. 공장의 예시를 든다면 생산 현황, 재고 상태, 작업자 배치 등 운영에 필요한 모든 정보가 이곳에 저장되고 공유되는 셈이죠.
예제 코드)
from langgraph.graph import StateGraph from typing import TypedDict, List, Annotated import operator class State(TypedDict): input: str all_actions: Annotated[List[str], operator.add] graph = StateGraph(State)
✅ Nodes (노드란?)
노드는 각각의 작업을 담당하는 작은 함수 단위입니다. 엔진을 조립하는 곳, 차체를 도색하는 곳, 타이어를 설치하는 곳, 품질을 검사하는 곳처럼 각자 맡은 기능이 있는 공장 내 별개의 작업장이라고 보면 됩니다.
예제 코드)
graph.add_node("model", model) graph.add_node("tools", tool_executor) # from langgraph.graph import END => 종료 시점을 알리기 위한 END 스테이션
그럼 각 스테이션(a.k.a 작업장)은 자신의 고유한 작업을 수행하고 각 단계에서 공정을 거친 생산품은 다음 노드로 계속해서 이동해야 되겠죠??
✅ Edges (엣지)
그 때 엣지는 노드 간의 흐름을 연결해주는 역할을 합니다. 타이어 조립 작업이 끝나고 문을 다는 공정으로 넘어가는 식이죠. 이전 노드에서 다음 노드로 넘어가게 하고 싶을 때, 두 노드를 add_edge()라는 도구로 연결할 수 있습니다.
예제 코드)
# 일반 엣지 graph.add_edge("tools", "model") # 조건부 엣지 graph.add_conditional_edge( "model", should_continue, { "end": END, "continue": "tools" } )
마찬가지로 공장의 예시를 든다면, 한 스테이션(작업장)의 작업이 끝난 뒤 다음 스테이션으로 제품이나 정보를 전달하는 파이프라인. 즉 노드와 노드를 연결하는 일종의 컨테이너 벨트로 생각할 수 있겠네요.
추가로, 조금 더 똑똑한 흐름을 만들고 싶을 때는, 조건에 따라 흐름을 나누는 조건부 엣지도 쓸 수 있습니다. 이건 뒤에서 좀 더 자세히 살펴볼게요!!
✅ Compile (컴파일)
이렇게 정의된 상태, 노드, 엣지들을 종합하여 하나의 실행 가능한 그래프를 완성하는 단계입니다.
앞선 공장의 예시를 이어서 설명하자면, 단일 작업의 스테이션(작업장) 간 연결과 관리(상태) 시스템의 처음부터 끝을 이어서 전체 공장을 완성시키는 역할입니다.
예제 코드)
graph = GraphState(State) app = graph.compile()
이 compile 명령을 통해 그래프 전체가 유기적으로 작동하게 되죠.
그럼 여기서 우린 조건부로 동작하는 Conditional Edge 에 대해 좀 더 자세히 살펴보겠습니다!
2️⃣ 조건부 엣지 (Conditional Edge)
이번에 가장 흥미로웠던 개념은 바로 조건부 엣지였습니다. 그래프 흐름을 그냥 고정된 순서대로 흘려보내는 게 아니라, 상태에 따라 다음 노드를 선택할 수 있는 구조죠. 그럼 일반 엣지와 비교부터 해볼까요?
🔸엔트리 포인트 (Entry Point)
사용자 입력이 도착했을 때, 가장 먼저 호출되는 노드를 지정하는 역할로, START 노드에서 node_a로 이동하는 엣지를 먼저 만들어줍니다.
from langgraph.graph import START # 초기 엔트리 포인트 : START graph.add_edge(START, "node_a")
🔸 일반 엣지 (Normal Edges)
그리고 node_a가 만들어졌다면 node_b로 이어지는 엣지도 만들어줄 수 있겠죠.
# "node_a" -> "node_b"로 이동하는 엣지 graph.add_edge("node_a", "node_b")
여기서 일반 엣지의 특징은 다음과 같습니다.
하나의 노드는 여러 개의 출력(Out-going) 엣지를 가질 수 있다.
만약 하나의 노드가 여러 개의 출력 엣지를 가진 경우, 그 모든 대상 노드(destination nodes)가 병렬(parallel)로 실행되며, 이는 다음 슈퍼스텝(superstep)의 일부가 된다.
즉, "엣지는 그래프의 흐름을 결정하는 중요한 요소이며, 병렬 실행을 통해 효율적인 데이터 흐름을 만들 수 있다."는 점이겠네요. 그렇다면 조건부 엣지(Conditional Edge)는 어떤 특징을 갖고 있을까를 살펴보죠.
📌 Conditional Edge (조건부 엣지)
조건부 엣지는 특정 함수를 실행한 결과에 따라 다음에 이동할 노드를 유동적으로 선택할 수 있게 해주는 구조입니다.
일반 엣지가 단순한 A → B 이동이라면, 조건부 엣지는 “상황에 따라 B로 갈 수도 있고, C로도 갈 수 있다”는 식의 분기 흐름을 만들어준다고 볼 수 있어요.
📌 조건부 엔트리 포인트 (Conditional Entry Point)
조금 더 유연하게 시작 노드를 지정하고 싶을 때는, 앞의 일반 엣지에서처럼 conditional 엔트리 포인트에도 조건부 흐름을 적용할 수 있습니다. 즉, 사용자 입력이 처음 들어왔을 때, 어떤 노드에서 실행을 시작할지 조건에 따라 결정할 수 있다는 이야기죠.
# "node_a" -> "node_b"로 이동하는 엣지 graph.add_conditional_edges(START, "node_b")
3️⃣ 직접 만들어본 간단한 LangGraph 파이프라인
최근 LangChain LinkedIn과 Substack에서 소개된 LangGraph 파이프라인 예제를 직접 구현해보며, 조건부 엣지를 어떻게 활용할 수 있는지 하나씩 짚어봤습니다.
간단 파이프라인 구현 예시
https://diamantai.substack.com - langgraph 파이프라인 예시
이제 실제 사례를 통해 조건부 엣지가 얼마나 유용한지 알아보겠습니다. 이번 예시는 뉴스 기사 수집 및 요약 파이프라인에서 출발했어요.
🤔 문제 상황
기사의 길이가 너무 짧은 경우, 요약(summarization)을 굳이 실행할 필요가 없다는 점이 고민이었습니다. 하지만 일반 엣지(add_edge)만 사용할 경우, 기사의 길이와 상관없이 요약 노드가 항상 실행되고 있었어요.
그 결과,
불필요한 연산 발생
의미 없는 요약 결과 생성 이라는 문제가 반복됐습니다.
😛 해결 방법: 조건부 엣지 추가
그래서 아래와 같은 조건을 기준으로 요약 단계를 분기시켜보았습니다.
from langgraph.graph import StateGraph, END, START from langgraph.prebuilt import tools_condition # === 상태(State) 관리 클래스 정의 === class State: def __init__(self): self.article_texts = [] # 기사 본문 리스트 (각 항목은 문자열) self.selected_urls = [] # 선택된 URL 목록 # === 그래프 생성 === graph = StateGraph(State) # === 일반 노드 정의 === 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 # === 노드 추가 === graph.add_node("generate_newsapi_params", generate_newsapi_params) graph.add_node("retrieve_articles_metadata", retrieve_articles_metadata) graph.add_node("retrieve_articles_text", retrieve_articles_text) graph.add_node("select_top_urls", select_top_urls) graph.add_node("summarize_articles_parallel", summarize_articles_parallel) graph.add_node("format_results", format_results) # === 일반 엣지 추가 === graph.add_edge(START, "generate_newsapi_params") graph.add_edge("generate_newsapi_params", "retrieve_articles_metadata") graph.add_edge("retrieve_articles_metadata", "retrieve_articles_text") graph.add_edge("retrieve_articles_text", "select_top_urls") # === 조건부 함수 정의 === def article_length_check(state): """ 전체 기사 텍스트의 길이가 3000자 이상인지 확인 """ total_length = sum(len(text) for text in state.article_texts) print(f"Total article text length: {total_length}") return total_length >= 3000 # === 조건부 엣지 추가 === graph.add_conditional_edges( "select_top_urls", tools_condition(article_length_check), {"true": "summarize_articles_parallel", "false": END} ) # === 후속 엣지 추가 === graph.add_edge("summarize_articles_parallel", "format_results") graph.add_edge("format_results", END) # === 실행 예제 === print("\n[실행 예제 - 총 기사 길이 3500자]") test_state_1 = State() test_state_1.article_texts = [ "뉴스기사1 내용입니다." * 100, # 약 1500자 "뉴스기사2 내용입니다." * 100, # 약 1500자 "뉴스기사3 내용입니다." * 50 # 약 500자 ] graph.run(test_state_1) print("\n[실행 예제 - 총 기사 길이 2000자]") test_state_2 = State() test_state_2.article_texts = [ "짧은기사" * 100, # 약 800자 "짧은기사" * 100, # 약 800자 "짧은기사" * 50 # 약 400자 ] graph.run(test_state_2)
🔸 graph (State) 흐름
select_top_urls 노드가 실행된 후, article_length_check(state)를 통해 기사의 길이를 확인하고
조건(state.article_length >= 3000 or state.article_count <3000) 에 따라 요약 노드(summarize_articles_parallel -> True 경로)로 이동할지, 종료 노드(END 0-> False 경로 → 종료)로 이동할지 결정됩니다!
🔸Conditional Edge의 주요 기능
조건 (article_length_check(state)) | 이동 노드 |
True (article_count >= 3000) | summarize_articles_parallel |
False (article_count < 3000) | END (그래프 종료) |
결국 조건부 엣지는 다음과 같은 이유에서 사용한다고 볼 수 있습니다.
Conditional Edge가 필요한 이유
조건부 엣지는 단순히 흐름을 분기하는 기능 이상으로, 다음과 같은 효과를 만들어냅니다:
✅ 유연성: 입력이나 상황에 따라 흐름을 유동적으로 제어할 수 있어요.
✅ 효율성: 꼭 필요한 경우에만 노드를 실행하므로, 불필요한 연산을 피할 수 있죠.
✅ 복잡한 워크플로우 구현: 다단계 의사결정이나 반복 로직 등을 훨씬 깔끔하게 설계할 수 있습니다.
✅ 사용자 경험 개선: 사용자 요청이나 조건에 맞게 자연스럽게 반응하는 시스템을 만들 수 있어요.
이번 인턴 일기에서는 LinkBrain 프로젝트를 진행하며 익히게 된 LangGraph의 주요 개념, 특히 StateGraph를 활용한 상태 관리 구조와 Conditional Edge를 통한 분기 흐름 제어 방식에 대해 정리해보았습니다.
특정 조건에 따라 실행 경로를 유연하게 제어할 수 있다는 점은, 실제 서비스 설계에서 매우 중요한 요소라고 느꼈어요. 단순한 흐름을 넘어서 다양한 사용자 상황에 맞춘 동적인 처리를 가능하게 해주기 때문이죠.
아직 LangGraph Studio와 같은 GUI 툴은 Windows 환경에서 제대로 지원되지 않아 아쉬움이 있지만, 그래도 하나씩 코드를 작성해보며 점점 이 프레임워크에 익숙해지고 있습니다. 오히려 로컬 웹서버에서 동작할 수 있는 방안을 찾아서 잘 사용하고 있기도 하고요. 새로운 개념을 배울 때마다 흥미롭고, 그걸 실습으로 이어갈 수 있다는 점에서 정말 보람도 큽니다 😄😄
그럼 이번 인턴 일기는 여기서 마칩니다!! 다음 5화에서 봐요!
(인턴 일기 5화에서 계속됩니다...)
3️⃣ Reference
https://medium.com/ai-agents/langgraph-for-beginners-part-3-conditional-edges-16a3aaad9f31
https://langchain-ai.github.io/langgraph/concepts/low_level/#conditional-edges
https://wikidocs.net/261579
"인턴 일기 [03화]에서 이어집니다."
안녕하세요 인턴 jo._.on_ 입니다. LinkBrain 팀에 합류한지도 어느덧 6개월이 다 되어가고 있습니다. 어쩜 이렇게 시간이 빠른지…😖😖 그래도 마지막까지 블로그에 소홀해져선 안되겠죠!! LangGraph를 사용한 그래프 설계도에 익숙해지고 있는 요즘, LinkBrain의 내부 파이프라인을 하나하나 직접 뜯어보는 재미가 쏠쏠합니다. 이번엔 그 중에서도 LangGraph의 특히 중요한 핵심 모듈 StateGraph와 Conditional_Edge라는 개념에 대한 [기술편]을 정리해보려고 해요.
그럼 이번 인턴 일기 04화 [기술편]도 시작해보겠습니다.
1️⃣ StateGraph란?
StateGraph는 LLM과 상호작용하는 우리의 주요 소통 창구로, 전체 플로우를 구성하는 세 가지 주요 요소로 이루어져 있어요.
바로 State (상태), Nodes (작업), 그리고 Edges (연결)입니다.
출처 : LangGraph - 'StateGraph'
LangGraph에서 StateGraph는 LLM 기반 워크플로우를 구성하는 핵심 구조라고 할 수 있습니다. 전체 애플리케이션의 상태를 하나의 그래프로 정의하고, 흐름에 따라 다양한 작업을 수행할 수 있게 돕는 역할을 하죠. 쉽게 말해, LangGraph에서 상태와 로직, 그리고 흐름 제어를 모두 담당하는 공유 데이터 구조라고 보면 됩니다.
StateGraph는 다음과 같은 세 가지 주요 구성요소로 이루어져 있습니다.
요소 | 정의 | 역할 |
State | 현재 애플리케이션의 상태를 나타내는 공유 데이터 구조 | 전체 컨텍스트를 유지하고, 노드 간 정보를 전달하는 역할을 합니다. 일반적으로 |
Nodes | 실제 작업을 수행하는 Python 함수들 | 특정 기능을 실행하며 상태를 업데이트합니다. 입력값으로 상태를 받고, 수정된 상태를 반환하는 방식이죠. |
Edges | 노드 간의 흐름을 연결하는 요소 | 그래프의 전반적인 동선을 결정하고, 조건에 따라 분기하는 흐름도 만들 수 있습니다. 다음에 어떤 노드를 실행할지 정해주는 역할이죠. |
✅ State (상태)
State는 말 그대로 현재 프로세스의 상황을 표현하는 데이터 구조입니다. LLM 프로세스가 어떤 입력을 받고 어떤 작업을 수행했는지를 기록하고, 이 정보를 모든 노드에서 공유할 수 있게 하죠.
State는 말 그대로 지금 LLM이 마주한 상황을 담고 있는 '중앙 정보 저장소'라고 할 수 있습니다. 공장의 예시를 든다면 생산 현황, 재고 상태, 작업자 배치 등 운영에 필요한 모든 정보가 이곳에 저장되고 공유되는 셈이죠.
예제 코드)
from langgraph.graph import StateGraph from typing import TypedDict, List, Annotated import operator class State(TypedDict): input: str all_actions: Annotated[List[str], operator.add] graph = StateGraph(State)
✅ Nodes (노드란?)
노드는 각각의 작업을 담당하는 작은 함수 단위입니다. 엔진을 조립하는 곳, 차체를 도색하는 곳, 타이어를 설치하는 곳, 품질을 검사하는 곳처럼 각자 맡은 기능이 있는 공장 내 별개의 작업장이라고 보면 됩니다.
예제 코드)
graph.add_node("model", model) graph.add_node("tools", tool_executor) # from langgraph.graph import END => 종료 시점을 알리기 위한 END 스테이션
그럼 각 스테이션(a.k.a 작업장)은 자신의 고유한 작업을 수행하고 각 단계에서 공정을 거친 생산품은 다음 노드로 계속해서 이동해야 되겠죠??
✅ Edges (엣지)
그 때 엣지는 노드 간의 흐름을 연결해주는 역할을 합니다. 타이어 조립 작업이 끝나고 문을 다는 공정으로 넘어가는 식이죠. 이전 노드에서 다음 노드로 넘어가게 하고 싶을 때, 두 노드를 add_edge()라는 도구로 연결할 수 있습니다.
예제 코드)
# 일반 엣지 graph.add_edge("tools", "model") # 조건부 엣지 graph.add_conditional_edge( "model", should_continue, { "end": END, "continue": "tools" } )
마찬가지로 공장의 예시를 든다면, 한 스테이션(작업장)의 작업이 끝난 뒤 다음 스테이션으로 제품이나 정보를 전달하는 파이프라인. 즉 노드와 노드를 연결하는 일종의 컨테이너 벨트로 생각할 수 있겠네요.
추가로, 조금 더 똑똑한 흐름을 만들고 싶을 때는, 조건에 따라 흐름을 나누는 조건부 엣지도 쓸 수 있습니다. 이건 뒤에서 좀 더 자세히 살펴볼게요!!
✅ Compile (컴파일)
이렇게 정의된 상태, 노드, 엣지들을 종합하여 하나의 실행 가능한 그래프를 완성하는 단계입니다.
앞선 공장의 예시를 이어서 설명하자면, 단일 작업의 스테이션(작업장) 간 연결과 관리(상태) 시스템의 처음부터 끝을 이어서 전체 공장을 완성시키는 역할입니다.
예제 코드)
graph = GraphState(State) app = graph.compile()
이 compile 명령을 통해 그래프 전체가 유기적으로 작동하게 되죠.
그럼 여기서 우린 조건부로 동작하는 Conditional Edge 에 대해 좀 더 자세히 살펴보겠습니다!
2️⃣ 조건부 엣지 (Conditional Edge)
이번에 가장 흥미로웠던 개념은 바로 조건부 엣지였습니다. 그래프 흐름을 그냥 고정된 순서대로 흘려보내는 게 아니라, 상태에 따라 다음 노드를 선택할 수 있는 구조죠. 그럼 일반 엣지와 비교부터 해볼까요?
🔸엔트리 포인트 (Entry Point)
사용자 입력이 도착했을 때, 가장 먼저 호출되는 노드를 지정하는 역할로, START 노드에서 node_a로 이동하는 엣지를 먼저 만들어줍니다.
from langgraph.graph import START # 초기 엔트리 포인트 : START graph.add_edge(START, "node_a")
🔸 일반 엣지 (Normal Edges)
그리고 node_a가 만들어졌다면 node_b로 이어지는 엣지도 만들어줄 수 있겠죠.
# "node_a" -> "node_b"로 이동하는 엣지 graph.add_edge("node_a", "node_b")
여기서 일반 엣지의 특징은 다음과 같습니다.
하나의 노드는 여러 개의 출력(Out-going) 엣지를 가질 수 있다.
만약 하나의 노드가 여러 개의 출력 엣지를 가진 경우, 그 모든 대상 노드(destination nodes)가 병렬(parallel)로 실행되며, 이는 다음 슈퍼스텝(superstep)의 일부가 된다.
즉, "엣지는 그래프의 흐름을 결정하는 중요한 요소이며, 병렬 실행을 통해 효율적인 데이터 흐름을 만들 수 있다."는 점이겠네요. 그렇다면 조건부 엣지(Conditional Edge)는 어떤 특징을 갖고 있을까를 살펴보죠.
📌 Conditional Edge (조건부 엣지)
조건부 엣지는 특정 함수를 실행한 결과에 따라 다음에 이동할 노드를 유동적으로 선택할 수 있게 해주는 구조입니다.
일반 엣지가 단순한 A → B 이동이라면, 조건부 엣지는 “상황에 따라 B로 갈 수도 있고, C로도 갈 수 있다”는 식의 분기 흐름을 만들어준다고 볼 수 있어요.
📌 조건부 엔트리 포인트 (Conditional Entry Point)
조금 더 유연하게 시작 노드를 지정하고 싶을 때는, 앞의 일반 엣지에서처럼 conditional 엔트리 포인트에도 조건부 흐름을 적용할 수 있습니다. 즉, 사용자 입력이 처음 들어왔을 때, 어떤 노드에서 실행을 시작할지 조건에 따라 결정할 수 있다는 이야기죠.
# "node_a" -> "node_b"로 이동하는 엣지 graph.add_conditional_edges(START, "node_b")
3️⃣ 직접 만들어본 간단한 LangGraph 파이프라인
최근 LangChain LinkedIn과 Substack에서 소개된 LangGraph 파이프라인 예제를 직접 구현해보며, 조건부 엣지를 어떻게 활용할 수 있는지 하나씩 짚어봤습니다.
간단 파이프라인 구현 예시
https://diamantai.substack.com - langgraph 파이프라인 예시
이제 실제 사례를 통해 조건부 엣지가 얼마나 유용한지 알아보겠습니다. 이번 예시는 뉴스 기사 수집 및 요약 파이프라인에서 출발했어요.
🤔 문제 상황
기사의 길이가 너무 짧은 경우, 요약(summarization)을 굳이 실행할 필요가 없다는 점이 고민이었습니다. 하지만 일반 엣지(add_edge)만 사용할 경우, 기사의 길이와 상관없이 요약 노드가 항상 실행되고 있었어요.
그 결과,
불필요한 연산 발생
의미 없는 요약 결과 생성 이라는 문제가 반복됐습니다.
😛 해결 방법: 조건부 엣지 추가
그래서 아래와 같은 조건을 기준으로 요약 단계를 분기시켜보았습니다.
from langgraph.graph import StateGraph, END, START from langgraph.prebuilt import tools_condition # === 상태(State) 관리 클래스 정의 === class State: def __init__(self): self.article_texts = [] # 기사 본문 리스트 (각 항목은 문자열) self.selected_urls = [] # 선택된 URL 목록 # === 그래프 생성 === graph = StateGraph(State) # === 일반 노드 정의 === 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 # === 노드 추가 === graph.add_node("generate_newsapi_params", generate_newsapi_params) graph.add_node("retrieve_articles_metadata", retrieve_articles_metadata) graph.add_node("retrieve_articles_text", retrieve_articles_text) graph.add_node("select_top_urls", select_top_urls) graph.add_node("summarize_articles_parallel", summarize_articles_parallel) graph.add_node("format_results", format_results) # === 일반 엣지 추가 === graph.add_edge(START, "generate_newsapi_params") graph.add_edge("generate_newsapi_params", "retrieve_articles_metadata") graph.add_edge("retrieve_articles_metadata", "retrieve_articles_text") graph.add_edge("retrieve_articles_text", "select_top_urls") # === 조건부 함수 정의 === def article_length_check(state): """ 전체 기사 텍스트의 길이가 3000자 이상인지 확인 """ total_length = sum(len(text) for text in state.article_texts) print(f"Total article text length: {total_length}") return total_length >= 3000 # === 조건부 엣지 추가 === graph.add_conditional_edges( "select_top_urls", tools_condition(article_length_check), {"true": "summarize_articles_parallel", "false": END} ) # === 후속 엣지 추가 === graph.add_edge("summarize_articles_parallel", "format_results") graph.add_edge("format_results", END) # === 실행 예제 === print("\n[실행 예제 - 총 기사 길이 3500자]") test_state_1 = State() test_state_1.article_texts = [ "뉴스기사1 내용입니다." * 100, # 약 1500자 "뉴스기사2 내용입니다." * 100, # 약 1500자 "뉴스기사3 내용입니다." * 50 # 약 500자 ] graph.run(test_state_1) print("\n[실행 예제 - 총 기사 길이 2000자]") test_state_2 = State() test_state_2.article_texts = [ "짧은기사" * 100, # 약 800자 "짧은기사" * 100, # 약 800자 "짧은기사" * 50 # 약 400자 ] graph.run(test_state_2)
🔸 graph (State) 흐름
select_top_urls 노드가 실행된 후, article_length_check(state)를 통해 기사의 길이를 확인하고
조건(state.article_length >= 3000 or state.article_count <3000) 에 따라 요약 노드(summarize_articles_parallel -> True 경로)로 이동할지, 종료 노드(END 0-> False 경로 → 종료)로 이동할지 결정됩니다!
🔸Conditional Edge의 주요 기능
조건 (article_length_check(state)) | 이동 노드 |
True (article_count >= 3000) | summarize_articles_parallel |
False (article_count < 3000) | END (그래프 종료) |
결국 조건부 엣지는 다음과 같은 이유에서 사용한다고 볼 수 있습니다.
Conditional Edge가 필요한 이유
조건부 엣지는 단순히 흐름을 분기하는 기능 이상으로, 다음과 같은 효과를 만들어냅니다:
✅ 유연성: 입력이나 상황에 따라 흐름을 유동적으로 제어할 수 있어요.
✅ 효율성: 꼭 필요한 경우에만 노드를 실행하므로, 불필요한 연산을 피할 수 있죠.
✅ 복잡한 워크플로우 구현: 다단계 의사결정이나 반복 로직 등을 훨씬 깔끔하게 설계할 수 있습니다.
✅ 사용자 경험 개선: 사용자 요청이나 조건에 맞게 자연스럽게 반응하는 시스템을 만들 수 있어요.
이번 인턴 일기에서는 LinkBrain 프로젝트를 진행하며 익히게 된 LangGraph의 주요 개념, 특히 StateGraph를 활용한 상태 관리 구조와 Conditional Edge를 통한 분기 흐름 제어 방식에 대해 정리해보았습니다.
특정 조건에 따라 실행 경로를 유연하게 제어할 수 있다는 점은, 실제 서비스 설계에서 매우 중요한 요소라고 느꼈어요. 단순한 흐름을 넘어서 다양한 사용자 상황에 맞춘 동적인 처리를 가능하게 해주기 때문이죠.
아직 LangGraph Studio와 같은 GUI 툴은 Windows 환경에서 제대로 지원되지 않아 아쉬움이 있지만, 그래도 하나씩 코드를 작성해보며 점점 이 프레임워크에 익숙해지고 있습니다. 오히려 로컬 웹서버에서 동작할 수 있는 방안을 찾아서 잘 사용하고 있기도 하고요. 새로운 개념을 배울 때마다 흥미롭고, 그걸 실습으로 이어갈 수 있다는 점에서 정말 보람도 큽니다 😄😄
그럼 이번 인턴 일기는 여기서 마칩니다!! 다음 5화에서 봐요!
(인턴 일기 5화에서 계속됩니다...)
3️⃣ Reference
https://medium.com/ai-agents/langgraph-for-beginners-part-3-conditional-edges-16a3aaad9f31
https://langchain-ai.github.io/langgraph/concepts/low_level/#conditional-edges
https://wikidocs.net/261579
