[AI Study] LangGraph로 상태 기반 Agent 만들기

[AI Study] LangGraph로 상태 기반 Agent 만들기

복잡한 AI 워크플로우를 그래프로 관리하는 방법

---

🎯 만들 것

"LangGraph 고객 지원 시스템"

기능:

1. 상태(State) 기반 데이터 관리
2. 노드(Node)로 각 단계 정의
3. 조건부 분기 (Conditional Edge)
4. 자동 워크플로우 실행
5. 투명한 실행 과정

---

🛠️ 기술 스택

• Python 3.11+
• LangGraph - 상태 기반 워크플로우
• LangChain - AI 프레임워크
• OpenAI API - GPT-4o-mini

---

💡 LangGraph란?

Multi-Agent vs LangGraph:

구분	Multi-Agent	LangGraph
상태 관리	수동	자동
흐름 제어	if문	그래프
복잡한 로직	어려움	쉬움
시각화	불가	가능
반복 처리	복잡	간단

LangGraph 구조:

[START]
   ↓
[분류 노드] ← State: {question, category, answer}
   ↓
조건부 분기
 / | \
🔧💳📞
 \ | /
   ↓
[END]

---

📦 Step 1. 패키지 설치

1-1 가상환경 활성화

source venv/bin/activate

1-2 패키지 설치

pip install langgraph langchain langchain-openai python-dotenv

---

📂 Step 2. 프로젝트 구조

agent/
└── langgraph/
    ├── __init__.py
    ├── main.py              # 실행 파일
    ├── state.py             # State 정의
    ├── graph.py             # 그래프 구성
    └── node/
        ├── __init__.py
        ├── classify.py      # 분류 노드
        ├── technical.py     # 기술 노드
        ├── payment.py       # 결제 노드
        └── general.py       # 일반 노드

---

💻 Step 3. 핵심 개념 구현

3-1. State 정의 (state.py)

State = 데이터 저장소

"""LangGraph State 정의"""
from typing import TypedDict


class AgentState(TypedDict):
    """Agent의 상태"""
    question: str          # 사용자 질문
    category: str          # 분류 결과
    answer: str            # 최종 답변
    messages: list         # 대화 기록

→ 노드 간 데이터 전달

3-2. Node 정의 (node/classify.py)

Node = 각 처리 단계

"""분류 노드"""
from langchain_openai import ChatOpenAI
from state import AgentState


def classify_node(state: AgentState) -> AgentState:
    """질문을 카테고리로 분류"""
    print(f"\n🏢 분류 중: {state['question']}")
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    
    prompt = f"""다음 질문을 분류하세요:
- technical: 기술 문제
- payment: 결제 문제  
- general: 일반 문의

질문: {state['question']}

한 단어로만 답하세요: technical, payment, general"""
    
    response = llm.invoke(prompt)
    category = response.content.strip().lower()
    
    # 유효성 검사
    if category not in ['technical', 'payment', 'general']:
        category = 'general'
    
    state["category"] = category
    print(f"   ✅ 분류: {category}")
    
    return state

→ State를 받아서 처리하고 반환

3-3. 전문 노드 (node/technical.py)

"""기술 지원 노드"""
from langchain_openai import ChatOpenAI
from state import AgentState


def technical_node(state: AgentState) -> AgentState:
    """기술 문제 해결"""
    print(f"\n🔧 기술 지원 처리 중...")
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    
    prompt = f"""당신은 기술 지원 전문가입니다.

질문: {state['question']}

친절하게 단계별로 해결 방법을 알려주세요."""
    
    response = llm.invoke(prompt)
    state["answer"] = response.content
    
    print(f"   ✅ 답변 완료")
    return state

---

🎮 Step 4. 그래프 구성 (graph.py)

핵심: 노드 연결 + 조건 분기

"""LangGraph 구성"""
from langgraph.graph import StateGraph, END
from state import AgentState
from node import (
    classify_node,
    technical_node,
    payment_node,
    general_node
)


def route_question(state: AgentState) -> str:
    """카테고리에 따라 다음 노드 결정"""
    category = state.get("category", "general")
    
    if category == "technical":
        return "technical"
    elif category == "payment":
        return "payment"
    else:
        return "general"


def create_graph():
    """LangGraph 생성 및 컴파일"""
    
    # 그래프 생성
    workflow = StateGraph(AgentState)
    
    # 노드 추가
    workflow.add_node("classify", classify_node)
    workflow.add_node("technical", technical_node)
    workflow.add_node("payment", payment_node)
    workflow.add_node("general", general_node)
    
    # 시작점 설정
    workflow.set_entry_point("classify")
    
    # 조건부 분기
    workflow.add_conditional_edges(
        "classify",                # 출발 노드
        route_question,            # 분기 함수
        {
            "technical": "technical",
            "payment": "payment",
            "general": "general"
        }
    )
    
    # 종료 연결
    workflow.add_edge("technical", END)
    workflow.add_edge("payment", END)
    workflow.add_edge("general", END)
    
    # 컴파일
    return workflow.compile()

---

🚀 Step 5. 실행 (main.py)

"""LangGraph 고객 지원 시스템 - 메인"""
import os
import sys
# 현재 디렉토리를 경로에 추가
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from dotenv import load_dotenv
from graph import create_graph

load_dotenv()


def main():
    print("="*60)
    print("🤖 LangGraph 고객 지원 시스템")
    print("="*60)
    
    # 그래프 생성
    app = create_graph()
    
    # 예제 질문들
    examples = [
        "앱이 자꾸 꺼져요",
        "환불 받고 싶어요",
        "영업시간이 어떻게 되나요?"
    ]
    
    print("\n💡 예제 질문:")
    for i, ex in enumerate(examples, 1):
        print(f"   {i}. {ex}")
    
    print("\n" + "="*60)
    
    while True:
        try:
            user_input = input("\n질문 (숫자 또는 직접 입력, exit=종료): ").strip()
            
            if not user_input:
                continue
            
            if user_input.lower() == "exit":
                print("\n👋 시스템 종료!")
                break
            
            # 숫자면 예제 선택
            if user_input.isdigit() and 1 <= int(user_input) <= len(examples):
                question = examples[int(user_input) - 1]
            else:
                question = user_input
            
            # 초기 상태
            initial_state = {
                "question": question,
                "category": "",
                "answer": "",
                "messages": []
            }
            
            # 그래프 실행
            print("\n" + "="*60)
            result = app.invoke(initial_state)
            
            # 결과 출력
            print("\n" + "="*60)
            print("💬 최종 답변:")
            print("="*60)
            print(result["answer"])
            print("="*60)
            
        except KeyboardInterrupt:
            print("\n\n👋 시스템 종료!")
            break


if __name__ == "__main__":
    main()

---

🧪 Step 6. 테스트

6-1 프로그램 실행

python agent/langgraph/main.py

6-2 예제 테스트

============================================================
🤖 LangGraph 고객 지원 시스템
============================================================

💡 예제 질문:
   1. 앱이 자꾸 꺼져요
   2. 환불 받고 싶어요
   3. 영업시간이 어떻게 되나요?

============================================================

질문: 1

============================================================

🏢 분류 중: 앱이 자꾸 꺼져요
   ✅ 분류: technical

🔧 기술 지원 처리 중...
   ✅ 답변 완료

============================================================
💬 최종 답변:
============================================================
앱이 계속 종료되는 문제를 해결하기 위해 다음 단계를 시도해보세요:

1. 앱 완전 종료 후 재시작
2. 캐시 삭제 (설정 > 앱 관리 > 캐시 삭제)
3. 앱 업데이트 확인
4. 기기 재시작
5. 앱 재설치

문제가 지속되면 support@example.com으로 연락주세요!
============================================================

---

📊 LangGraph 작동 원리

1️⃣ 초기 State 생성:
{
    "question": "앱이 꺼져요",
    "category": "",
    "answer": ""
}

2️⃣ classify 노드 실행:
→ State 업데이트: category = "technical"

3️⃣ route_question 함수 실행:
→ "technical" 반환

4️⃣ technical 노드로 이동:
→ State 업데이트: answer = "해결 방법..."

5️⃣ END:
→ 최종 State 반환

---

🎯 핵심 개념 복습

1. State (상태)

class AgentState(TypedDict):
    question: str
    category: str
    answer: str

→ 노드 간 데이터 저장소

2. Node (노드)

def classify_node(state: AgentState) -> AgentState:
    # 처리
    state["category"] = "technical"
    return state

→ 각 처리 단계

3. Conditional Edge (조건 분기)

workflow.add_conditional_edges(
    "classify",
    route_question,
    {
        "technical": "technical",
        "payment": "payment"
    }
)

→ if문처럼 분기

4. 그래프 컴파일

app = workflow.compile()
result = app.invoke(initial_state)

→ 실행 가능한 앱으로 변환

---

💡 LangGraph의 장점

장점	설명
상태 관리	State가 자동으로 노드 간 전달
명확한 흐름	그래프로 워크플로우 표현
조건 분기	복잡한 if-else 로직을 간단하게
반복 처리	루프 구조 쉽게 구현
시각화	그래프 구조 확인 가능

---

🔥 고급 기능 미리보기

1. 반복 처리 (Loop)

# 조건이 만족될 때까지 반복
workflow.add_edge("search", "analyze")
workflow.add_conditional_edges(
    "analyze",
    should_continue,  # True면 다시 search
    {
        True: "search",
        False: END
    }
)

2. 병렬 처리 (Parallel)

# 여러 노드 동시 실행
workflow.add_node("web_search", web_search_node)
workflow.add_node("doc_search", doc_search_node)

# 둘 다 실행 후 합치기
workflow.add_edge("start", "web_search")
workflow.add_edge("start", "doc_search")
workflow.add_edge("web_search", "combine")
workflow.add_edge("doc_search", "combine")

3. Human-in-the-Loop

# 사용자 확인 대기
workflow.add_node("draft", create_draft)
workflow.add_node("review", human_review)  # 여기서 멈춤
workflow.add_node("send", send_email)

---

✅ 완료 체크리스트

• LangGraph 개념 이해
• State 정의
• Node 구현
• 그래프 구성
• 조건부 분기 설정
• 그래프 컴파일 및 실행
• 테스트 완료

---

🚀 다음 단계

다음은 멀티모달 AI를 배웁니다!

멀티모달 = 텍스트 + 이미지 + 음성

LangGraph:
- 텍스트 기반 워크플로우
- 상태 관리

멀티모달:
- 이미지 분석
- 음성 인식/생성
- 문서 OCR

---

📚 참고 자료

• LangGraph 공식 문서
• LangChain 문서
• LangGraph GitHub

---