Langgraph: LLM 기반 애플리케이션의 새로운 뇌 🧠

안녕하세요! 오늘은 LLM 애플리케이션 개발의 게임 체인저, Langgraph에 대해 알아보겠습니다.

Langgraph가 뭔가요? 🤔

여러분이 로봇 부품들을 가지고 복잡한 로봇을 만든다고 상상해 보세요.

• 각 부품은 특정 기능을 수행하지만
• 이 부품들을 어떻게 연결할지, 어떤 순서로 작동시킬지 고민되죠?

Langgraph가 바로 이런 역할을 합니다!

• LLM 기반 애플리케이션의 여러 부품(기능)을
• 그래프 형태로 연결하고 제어해주는 마법 같은 프레임워크 ✨

쉽게 말해, Langgraph는 LangChain이 개발한 상태 관리 및 흐름 제어 프레임워크로, LLM 기반 애플리케이션을 그래프 구조로 구성할 수 있게 해줍니다.

어떻게 동작하나요? 🎬

1. 기본 구성 요소

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

# 1. 상태 타입 정의
class State(TypedDict):
    question: str
    research: str
    answer: str

# 2. 노드 함수 정의
def researcher(state: State) -> State:
    question = state["question"]
    # 연구 로직 구현
    research_result = f"Research results for: {question}"
    return {"research": research_result}

def answerer(state: State) -> State:
    question = state["question"]
    research = state["research"]
    # 답변 생성 로직 구현
    answer = f"Answer based on research: {research}"
    return {"answer": answer}

# 3. 그래프 생성 및 노드 추가
workflow = StateGraph(State)
workflow.add_node("researcher", researcher)
workflow.add_node("answerer", answerer)

# 4. 엣지 추가
workflow.add_edge("researcher", "answerer")
workflow.set_entry_point("researcher")

# 5. 그래프 컴파일 및 실행
app = workflow.compile()
result = app.invoke({"question": "What is Langgraph?"})

2. 주요 구성 요소 자세히 살펴보기

1. 노드(Node): 작업을 수행하는 함수들

   • 마치 회사 내 각 부서와 같습니다
   • 연구팀, 개발팀, 마케팅팀처럼 각각 고유한 역할을 담당

2. 엣지(Edge): 노드 간의 연결

   • 마치 회사 내 업무 흐름과 같습니다
   • "연구팀의 결과물이 개발팀으로 전달된다" 같은 흐름 정의

3. 상태(State): 그래프 내에서 공유되는 데이터

   • 마치 회사 내 공유 문서와 같습니다
   • 모든 팀이 접근하고 업데이트할 수 있는 중앙 정보

4. 조건부 엣지: 상태에 따라 다음 경로 결정

   • 마치 의사결정 회의와 같습니다
   • "예산이 충분하면 개발팀, 부족하면 기획팀으로" 같은 조건부 결정

동작 방식 💫

마치 스마트한 조립 라인처럼 작동합니다!

1. 초기 상태 설정

   라인 시작: 사용자 질문 → 상태에 저장

2. 첫 노드 실행

   연구 노드: "이 질문에 대한 정보를 찾아볼게요"
   → 연구 결과를 상태에 추가

3. 다음 노드로 이동

   답변 노드: "연구 결과를 바탕으로 답변을 만들어볼게요"
   → 최종 답변을 상태에 추가

4. 최종 상태 반환

   완료: 질문, 연구 결과, 답변이 포함된 최종 상태 반환

고급 기능: 더 복잡한 흐름 제어하기 🔄

1. 조건부 분기 처리

def router(state: State) -> Annotated[str, ["route_A", "route_B"]]:
    if state["condition"]:
        return "route_A"
    else:
        return "route_B"

workflow.add_conditional_edges(
    "decision_node",
    router,
    {
        "route_A": "node_A",
        "route_B": "node_B"
    }
)

마치 교차로에서 내비게이션이 "전방에 교통 체증이 있습니다. 우회도로로 안내합니다"라고 알려주는 것과 같습니다!

2. 병렬 처리

# 병렬 처리 설정
parallel_graph = StateGraph(ParallelState)
parallel_graph.add_node("task1", task1)
parallel_graph.add_node("task2", task2)
parallel_graph.add_node("combine", combine_results)

# 병렬 실행 설정
parallel_graph.add_edge("task1", "combine")
parallel_graph.add_edge("task2", "combine")
parallel_graph.set_entry_point(["task1", "task2"])  # 병렬 진입점!

마치 여러 요리사가 동시에 다른 요리를 준비한 후, 최종적으로 하나의 코스 요리로 완성하는 것과 같습니다!

3. 순환 처리 (반복 작업)

def should_continue(state: State) -> Annotated[str, ["continue", "stop"]]:
    if state["iterations"] < state["max_iterations"]:
        return "continue"
    else:
        return "stop"

workflow.add_conditional_edges(
    "process",
    should_continue,
    {
        "continue": "process",  # 자기 자신으로 돌아가는 엣지
        "stop": "final_node"
    }
)

마치 요리사가 "맛이 완벽해질 때까지 간을 보고 조절하는" 과정과 같습니다!

실제 사용 사례 📱

1. ReAct 에이전트 구현

class AgentState(TypedDict):
    input: str
    thought: str
    action: str
    observation: str
    answer: str

def think(state: AgentState) -> AgentState:
    input_text = state["input"]
    # LLM을 사용하여 생각 생성
    thought = llm.invoke(f"Think about: {input_text}")
    return {"thought": thought}

def act(state: AgentState) -> AgentState:
    thought = state["thought"]
    # LLM을 사용하여 행동 결정
    action = llm.invoke(f"Based on {thought}, what action should I take?")
    return {"action": action}

def observe(state: AgentState) -> AgentState:
    action = state["action"]
    # 외부 환경과 상호작용
    observation = perform_action(action)
    return {"observation": observation}

def should_continue(state: AgentState) -> str:
    # 종료 조건 검사
    if "final answer" in state["thought"].lower():
        return "finish"
    else:
        return "continue"

# 그래프 설정
agent_graph = StateGraph(AgentState)
agent_graph.add_node("think", think)
agent_graph.add_node("act", act)
agent_graph.add_node("observe", observe)
agent_graph.add_node("answer", generate_answer)

# 조건부 엣지 설정
agent_graph.add_edge("think", "act")
agent_graph.add_edge("act", "observe")
agent_graph.add_conditional_edges(
    "observe", 
    should_continue,
    {
        "continue": "think",
        "finish": "answer"
    }
)

2. 다중 에이전트 협업 시스템

여러 전문가 에이전트가 협력하여 복잡한 문제 해결:

# 각 에이전트 노드
def researcher_agent(state):
    # 연구 역할 수행
    return {"research_data": research_results}

def writer_agent(state):
    # 글쓰기 역할 수행
    return {"draft": written_content}

def editor_agent(state):
    # 편집 역할 수행
    return {"edited_content": edited_version}

def coordinator(state):
    # 다음 단계 결정
    if state["needs_more_research"]:
        return "research"
    elif state["needs_revision"]:
        return "write"
    elif state["needs_editing"]:
        return "edit"
    else:
        return "complete"

# 다중 에이전트 그래프 설정
multi_agent = StateGraph(MultiAgentState)
multi_agent.add_node("research", researcher_agent)
multi_agent.add_node("write", writer_agent)
multi_agent.add_node("edit", editor_agent)

# 조정자 노드를 통한 복잡한 흐름 제어
multi_agent.add_conditional_edges(
    "research", 
    coordinator,
    {
        "research": "research",
        "write": "write",
        "edit": "edit",
        "complete": "output"
    }
)

Langgraph vs LangChain 비교 🔍

특징	Langgraph	LangChain
주요 특징	그래프 기반 워크플로우 관리	컴포넌트 기반 LLM 통합
상태 관리	명시적 상태 객체	암시적 체인 컨텍스트
흐름 제어	조건부 분기, 사이클 지원	선형적 체인 중심
복잡성	높음 (그래프 설계 필요)	중간 (체인 연결 중심)
사용 사례	복잡한 에이전트, 다단계 워크플로우	간단한 LLM 파이프라인

LangChain이 레고 블록이라면, Langgraph는 레고 블록으로 만든 로봇에 인공지능 두뇌를 넣어주는 것과 같습니다!

장점은? 🌟

1. 상태 관리의 명확성

   • 각 단계의 입출력이 명확하게 정의됨
   • 마치 깔끔한 회계장부처럼 모든 데이터 흐름이 추적 가능

2. 유연한 흐름 제어

   • 조건부 분기, 루프, 병렬 처리 등 복잡한 패턴 지원
   • 마치 고급 교통 통제 시스템처럼 데이터 흐름을 정교하게 제어

3. 모듈성과 재사용성

   • 각 노드는 독립적으로 개발 및 테스트 가능
   • 마치 레고 블록처럼 조합하고 재사용 가능

4. 디버깅 용이성

   • 각 노드와 상태 변화를 명확하게 추적 가능
   • 문제가 발생한 정확한 지점을 파악하기 쉬움

5. 확장성

   • 간단한 프로토타입에서 복잡한 시스템으로 쉽게 확장 가능
   • 새로운 기능을 추가하거나 기존 기능을 수정하기 용이

주의할 점 ⚠️

1. 학습 곡선이 가파름

   • 그래프 기반 사고방식과 타입 시스템 이해 필요
   • 간단한 작업에는 과도한 복잡성을 추가할 수 있음

2. 상태 설계의 중요성

   • 잘못된 상태 설계는 전체 시스템 성능에 영향
   • 상태가 너무 크거나 복잡하면 메모리 및 성능 이슈 발생

3. 오류 처리 전략 필수

   • 노드 실패 시 전체 그래프 처리 방식 고려 필요
   • 장애 복구 메커니즘 설계가 필수적

4. 버전 호환성 이슈

   • LangChain 생태계와의 버전 호환성 주의
   • 빠른 발전으로 인한 API 변경 가능성

5. 테스트 복잡성

   • 모든 가능한 경로와 상태 조합 테스트 필요
   • 특히 조건부 분기가 많은 경우 테스트 케이스 설계가 복잡

실제 개발 시 TIP 💡

1. 작게 시작하기

   # 최소한의 그래프로 시작
   mini_graph = StateGraph(SimpleState)
   mini_graph.add_node("single_node", simple_function)
   mini_graph.set_entry_point("single_node")

2. 점진적으로 확장하기

   # 노드 하나씩 추가하며 테스트
   mini_graph.add_node("second_node", another_function)
   mini_graph.add_edge("single_node", "second_node")

3. 상태 디버깅 지원 추가

   def debug_state(state: State) -> State:
       print(f"Current state: {state}")
       return {}  # 상태 변경 없음
   
   workflow.add_node("debug", debug_state)
   # 디버깅이 필요한 노드 사이에 추가
   workflow.add_edge("node_a", "debug")
   workflow.add_edge("debug", "node_b")

4. 재사용 가능한 서브그래프 설계

   # 독립적인 서브그래프 생성
   sub_graph = StateGraph(SubState)
   # ... 서브그래프 설정 ...
   
   # 메인 그래프에 통합
   main_graph.add_node("sub_process", sub_graph)

마치며 🎁

Langgraph는 LLM 애플리케이션 개발의 새로운 패러다임을 제시합니다. 복잡한 AI 워크플로우를 명확하고 유지보수 가능한 방식으로 구현할 수 있게 해주며, 특히 에이전트 시스템과 다단계 추론 프로세스 개발에 큰 힘을 발휘합니다.

처음에는 학습 곡선이 있지만, 익숙해지면 강력한 AI 애플리케이션을 구축할 수 있는 도구가 될 것입니다. 작은 프로젝트에서 시작하여 점진적으로 Langgraph의 고급 기능을 탐색해보세요!

---

궁금하신 점 있으시다면 댓글로 남겨주세요! 😊

주의할 점

1. 상태 관리의 복잡성: 대규모 애플리케이션에서는 상태 객체가 매우 커질 수 있으며, 이로 인한 메모리 및 성능 이슈가 발생할 수 있습니다.

2. 학습 곡선: 그래프 기반 프로그래밍 패러다임을 이해하는 데 시간이 필요하며, 특히 타입 시스템과 함께 사용할 때 더욱 그렇습니다.

3. 디버깅 어려움: 복잡한 그래프에서는 오류 추적이 어려울 수 있으며, 특히 조건부 분기와 순환이 많은 경우 더욱 그렇습니다.

4. 버전 호환성: LangChain 생태계가 빠르게 발전하면서 API 변경이 자주 있을 수 있어 코드 유지보수에 주의가 필요합니다.

5. 테스트 복잡성: 모든 가능한 그래프 경로를 테스트하는 것은 도전적인 작업이며, 철저한 테스트 전략이 필요합니다.

참고 자료

• LangGraph 공식 문서
• LangChain 블로그 - LangGraph 소개
• LangGraph Studio - 에이전트 IDE
• LangGraph 멀티 에이전트 워크플로우