← 모든 글

요즘 AI 에이전트 개발: 랭그래프(1)

랭그래프란?

  • 랭체인을 개발한 LangChain.inc에서 개발한 에이전트 오케스트레이션 프레임워크

그래프 자료구조의 이해

  • 기본 개념으로 노드와 엣지로 구성된 자료구조
  • 객체 간의 관계 표현에 효과적
  • 방향성에 따라 방향 그래프, 무방향 그래프로 나뉘며 랭그래프는 방향 그래프 기반이며 작업 흐름이 특정 방향을 가진다.
  • 순환 vs 순환 없는 그래프
  • 대부분 많은 시스템은 순환 없는 그래프 이용
  • 랭그래프는 순환 그래프를 이용

랭그래프의 핵심 개념

상태, 노드, 엣지 3개의 개념

상태

  • 지속적으로 유지되는 데이터로 노드가 실행될 때 읽고, 쓸 수 있음
  • 워크플로 컨텍스트를 관리하는 중앙 저장소 역할
  • TypedDic 또는 Pydantic모델로 정의되어 타입 안정성 보장

노드

  • 기본 실행 단위
  • 특정 작업을 수행하는 함수나 에이전트
  • 현재 상태 입력 받음
  • 특정 작업 수행 (LLM 호출, 데이터 처리, 외부 API 호출)
  • 업데이트 된 상태 반환

엣지

  • 노드 간 연결 정의, 실행 흐름 제어
  • 일반 엣지 : 항상 같은 경로로 진행
  • 조건부 엣지 : 상태에 따라 다른 노드 분기

예시코드

from typing import Dict, Any
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field

# 워크플로우 단계정의
class WorkflowStep:
    GREETING = "GREETING"
    PROCESSING = "PROCESSING"
    
# 그래프 상태 정의
class GraphState(BaseModel):
    name: str = Field(default="", description="사용자 이름")
    greeting: str = Field(default="", description="생성된 인사말")
    processed_message: str = Field(default="", description="처리된 최종 메시지")
    
# 첫 번째 노드 함수
def generate_greeting(state:GraphState) -> Dict[str, Any]:
    name = state.name or "아무개"
    greeting = f"안녕하세요, {name}님!"
    print(f"[generate_greeting] 인사말 생성: {greeting}")
    return {"greeting": greeting}

# 두 번째 노드 함수
def process_message(state: GraphState) -> Dict[str, Any]:
    greeting = state.greeting
    processed_message = f"{greeting} LangGraph에 오신 것을 환영합니다!"
    
    print(f"[process_message] 최종 메시지: {processed_message}")
    
    return {"processed_message" : processed_message}

# 그래프 생성
def create_hello_graphe():
    workflow = StateGraph(GraphState)
    
    # 노드 추가
    workflow.add_node(WorkflowStep.GREETING, generate_greeting)
    workflow.add_node(WorkflowStep.PROCESSING, process_message)
    
    # 시작점 설정
    workflow.add_edge(START, WorkflowStep.GREETING)
    
    #엣지 추가(노드 간 연결)
    workflow.add_edge(WorkflowStep.GREETING, WorkflowStep.PROCESSING)
    workflow.add_edge(WorkflowStep.PROCESSING, END)
    
    # 그래프 컴파일
    app = workflow.compile()
    
    return app

def main():
    print("=== Hello 랭그래프 ===\n")
    app = create_hello_graphe()
    
    initial_state = GraphState(name="모카빵", greeting="", processed_message="")
    print("초기 상태: ", initial_state.model_dump())
    print("\n--- 그래프 실행 시작 ---")
    
    # 그래프 실행
    final_state = app.invoke(initial_state)
    
    print("--- 그래프 실행 종료 ---\n")
    print("최종 상태:", final_state)
    print(f"\n결과 메시지: {final_state['processed_message']}")
    # ASCII로 그래프 출력
    app.get_graph().draw_ascii()
    
    
if __name__ == "__main__":
    main()

조건부 라우팅 적용

  • 그래프의 실행 경로를 동적으로 결정하는 기능
  • 프로그래밍 조건문과 동일하지만 차이점은 LLM이 판단하고 바탕으로 유연한 처리, 유지보수 쉬운 그래프 구조로 표현 가능
import os

from dotenv import load_dotenv
from typing import Dict, Any, Literal
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
import random

load_dotenv()

# 그래프 상태 정의 - 워크플로 전체에서 공유되는 데이터 구조
class EmotionBotState(BaseModel):
    user_message: str = Field(default="", description="사용자 입력 메시지")
    emotion: str = Field(default="", description="분석된 감정")
    response: str = Field(default="", description="최종 응답 메시지")
    
# Langchain LLM 초기화 - 감정 분석에 사용할 AI 모델 설정
llm = ChatOpenAI(model="gpt-5-mini", api_key=os.getenv("OPENAI_API_KEY"))

# LLM 기반 감정 분석 노드 - 첫번째 단계
def analyze_emotion(state: EmotionBotState) -> Dict[str, Any]:
    message = state.user_message
    print(f"LLM 감정 분석 중: '{message}'")
    
    messages = [
        SystemMessage(
            content="당신은 감정 분석 전문가입니다. 사용자의 메시지를 분석하여 'positive', 'negative', 'neutral'중 하나로 감정을 분류해주세요. 답변은 반드시 하나의 단어만 출력하세요."
        ),
        HumanMessage(
            content=f"다음 메시지의 감정을 분석해주세요: '{message}'"
        )
    ]
    
    response = llm.invoke(messages)
    emotion = response.content.strip().lower()
    
    if emotion not in ["positive", "negative", "neutral"]:
        emotion = "neutral"
        
    print(f"LLM 감정 분석 결과 : {emotion}")
    return {"emotion": emotion}

# 긍정적 응답 생성
def generate_positive_response(state: EmotionBotState) -> Dict[str, Any]:
    responses = ["정말 좋은 소식이네요!", "기분이 좋으시군요!", "멋지네요!"]
    
    return {"response": random.choice(responses)}

# 부정적 응답 생성
def generate_negative_response(state: EmotionBotState) -> Dict[str, Any]:
    responses = [
        "힘든 시간이시군요. 괜찮아요.",
        "마음이 아프시겠어요.",
        "더 좋은 날이 올 거예요."
    ]
    
    return {"response": random.choice(responses)}

# 중립적 응답 생성
def generate_neutural_response(state: EmotionBotState) -> Dict[str, Any]:
    responses = [
        "감사해요! 더 자세히 말씀해주세요.",
        "이해했어요. 다른 도움이 필요하면 말씀하세요!",
        "흥미로운 주제네요!"
    ]
    
    return {"response": random.choice(responses)}

# 조건부 라우팅 함수 - 감정 분석 결과에 따라 다음 노드 결정
def route_by_emotion(
    state: EmotionBotState
) -> Literal['positive_response', 'negative_response', 'neutral_response']:
    emotion = state.emotion
    print(f"라우팅: {emotion}")
    
    if emotion == "positive":
        return "positive_response"
    elif emotion == "negative":
        return "negative_response"
    else:
        return "neutral_response"
    
    
# 그래프 생성 함수 - 전체 워크플로 구성
def create_emotion_bot_graph():
    workflow = StateGraph(EmotionBotState)
    
    # 노드 추가 - 처리 단계 그래프 등록
    workflow.add_node("analyze_emotion", analyze_emotion)
    workflow.add_node("positive_response", generate_positive_response)
    workflow.add_node("negative_response", generate_negative_response)
    workflow.add_node("neutral_response", generate_neutural_response)
    
    # 시작 엣ㅅ지 설정
    workflow.add_edge(START, "analyze_emotion")
    workflow.add_conditional_edges("analyze_emotion", 
                                   route_by_emotion,
                                   {
                                       "positive_response": "positive_response",
                                       "negative_response": "negative_response",
                                       "neutral_response":"neutral_response"
                                   })
    
    # 종료 엣지 설정
    workflow.add_edge("positive_response", END)
    workflow.add_edge("negative_response", END)
    workflow.add_edge("neutral_response", END)
    
    return workflow.compile()

def main():
    print("=== 감정 분석 챗봇 테스트 ===\n")
    app = create_emotion_bot_graph()
    
    test_cases = [
        "오늘 정말 기분이 좋아요!",
        "너무 슬프고 힘들어요...",
        "날씨가 어떤가요?",
    ]
    
    for i, message in enumerate(test_cases, 1):
        print(f"테스트 {i}: '{message}'")
        state = EmotionBotState(user_message=message)
        result = app.invoke(state)
        print(f"응답: {result['response']}\n")
        
    # 그래프 시각화
    mermaid_png = app.get_graph().draw_mermaid_png()
    with open(".02_conditional_routing.png", "wb") as f:
        f.write(mermaid_png)
        
if __name__ == "__main__":
    main()

이번 글의 핵심 정리

  1. 랭그래프는 랭체인에서 만든 오케스트레이션 에이전트이다!
  2. 그래프 자료구조에서 객체간의 관계를 표현하는 과정의 노드와 엣지가 중요하다.
  3. 상태 그리고 노드와 엣지
  4. 상태 : 워크플로우의 중앙저장소 역할이며 전체적인 상태를 관리한다. 노드의 작업 이후 상태가 변경 될 수 있다.
  5. 노드 : 각 별개의 작업을 수행하는 작업 단위이다. (LLM, API호출, 도구 등.)
  6. 엣지 : 노드들의 작업들을 연결해주는 작업을 진행한다. 때에 따라서 조건부 엣지를 이용하여 라우팅의 경로를 변경할 수 있다.

TIL 느낀점

  • 인프런 강의로 들었던 부분을 다시 책으로 공부하면서 복습하는 느낌이 들어 되려 이해되는 부분이 많았다.
  • 직접 예시코드로 실행해보면서 재밌다는 것을 느꼈던 시간이었고, 특히 조건부 엣지 부분에서 llm을 이용하여 결과를 직접적으로 받아 분기 처리를 하는 것에서 서브 에이전트들도 다룬다면 유용하게 이용할 수 있겠다고 느꼈다.