[GenAI] LangGraph 튜토리얼

LangGraph란?

LangGraph는 LangChain을 기반으로 하는 Graph 기반 워크플로우 라이브러리

• 복잡한 LLM 흐름을 다룰 때 유용함.
• DAG(Directed Acyclic Graph) 기반으로 구성되어 LLM 애플리케이션을 모듈화 가능.
• 상태(State) 관리가 가능하여 대화형 애플리케이션에 적합.

LangGraph는 생성형 AI 애플리케이션을 구축하기 위한 프레임워크로, 특히 복잡한 추론 과정을 관리하는 데 도움을 주는 도구

LLM(대규모 언어 모델)을 활용한 애플리케이션의 흐름을 상태 기계(state machine)로 모델링하고 관리할 수 있게 해줌

LangGraph의 핵심 개념

1. 상태 기계(State Machine) 기반:
   • 애플리케이션의 흐름을 명확한 상태와 상태 간 전환으로 정의
   • 각 상태는 특정 작업이나 처리를 담당
2. 노드(Node)와 엣지(Edge):
   • 노드: 특정 기능을 수행하는 단위 (예: 사용자 입력 처리, LLM 호출, 데이터베이스 쿼리 등)
   • 엣지: 노드 간의 연결로, 흐름의 방향을 정의
3. 상태(State):
   • 애플리케이션이 처리 중인 데이터를 담고 있는 컨테이너
   • 메시지 기록, 중간 결과, 컨텍스트 정보 등을 포함

정리하면,

1. 상태(State) - 그래프가 관리하는 데이터 (예: 대화 메시지)
2. 노드(Node) - 상태를 처리하는 함수들 (예: 챗봇, 도구 실행기)
3. 엣지(Edge) - 노드 간의 연결, 데이터 흐름 경로

• 상태는 노드 간에 전달되는 데이터
• 각 노드는 상태를 받아서 처리한 후 업데이트된 상태를 반환
• 엣지는 이 데이터가 어떤 노드로 흘러가야 할지 결정

LangGraph 코드 작성 예제

1. 라이브러리 임포트

# 기본 라이브러리
import os
from typing import Annotated, Dict, List, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# LLM 연결 라이브러리
from langchain_aws import ChatBedrock  # AWS Bedrock 사용 시

# 필요한 도구 가져오기
from langchain_community.tools.tavily_search import TavilySearchResults  # 검색 도구

 

2. 상태 (State) 정의

: 상태는 그래프가 추적하는 "데이터"임

class State(TypedDict):
    # 메시지 목록을 저장 (add_messages는 메시지를 덮어쓰지 않고 추가하라는 힌트)
    messages: Annotated[list, add_messages]
    # 필요하다면 추가 상태 정보를 저장할 수 있음
    # context: dict  # 예: 컨텍스트 정보
    # memory: list   # 예: 기억해야 할 데이터

 

3. LLM 및 도구 설정

# OpenAI 사용 예시
llm = ChatOpenAI(model="gpt-4o")

AWS Bedrock 사용 예시
import boto3
bedrock_client = boto3.client('bedrock-runtime', region_name='us-east-1')
llm = ChatBedrock(
    client=bedrock_client,
    model_id="anthropic.claude-3-sonnet-20240229-v1:0"
 )

# 도구 설정
search_tool = TavilySearchResults(max_results=3)
tools = [search_tool]

# 도구를 LLM에 연결
llm_with_tools = llm.bind_tools(tools)

 

4. 그래프 빌더 생성 (상태 초기화)

# 상태 클래스로 그래프 빌더 초기화
graph_builder = StateGraph(State)

 

5. 노드(Node) 함수 정의

: 노드는 상태를 입력받아 처리하고, 새로운 상태를 반환

# 기본 챗봇 노드 함수
def chatbot(state: State):
    """사용자 메시지에 응답하는 챗봇 노드"""
    # state["messages"]에서 대화 내용을 가져와 LLM으로 처리
    response = llm.invoke(state["messages"])
    # 새 메시지를 포함한 상태 반환
    return {"messages": [response]}

# 도구를 사용하는 챗봇 노드 함수 (도구 사용 시)
def chatbot_with_tools(state: State):
    """도구를 사용할 수 있는 챗봇 노드"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

 

6. 도구 실행 함수 정의

import json
from langchain_core.messages import ToolMessage

class ToolNode:
    """LLM이 요청한 도구를 실행하는 노드"""
    
    def __init__(self, tools: list):
        # 도구 이름으로 접근할 수 있는 딕셔너리 생성
        self.tools_by_name = {tool.name: tool for tool in tools}
    
    def __call__(self, state: State):
        """노드가 호출될 때 실행되는 함수"""
        # 마지막 메시지 가져오기
        last_message = state["messages"][-1]
        
        # 도구 호출 요청 확인
        tool_outputs = []
        for tool_call in last_message.tool_calls:
            # 도구 이름과 인자 추출
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            
            # 도구 실행
            tool_result = self.tools_by_name[tool_name].invoke(tool_args)
            
            # 결과를 ToolMessage로 변환
            tool_output = ToolMessage(
                content=json.dumps(tool_result),
                name=tool_name,
                tool_call_id=tool_call["id"]
            )
            tool_outputs.append(tool_output)
        
        # 도구 실행 결과를 포함한 상태 반환
        return {"messages": tool_outputs}

# 도구 노드 인스턴스 생성
tool_node = ToolNode(tools=tools)

 

7. 라우팅 함수 정의

def route_to_tool_or_end(state: State):
    """도구 사용이 필요한지 결정하는 라우팅 함수"""
    # 마지막 메시지 확인
    last_message = state["messages"][-1]
    
    # 도구 호출 요청이 있는지 확인
    if hasattr(last_message, "tool_calls") and len(last_message.tool_calls) > 0:
        # 도구 호출이 필요하면 'tool' 노드로 라우팅
        return "tool"
    
    # 도구 호출이 필요 없으면 종료
    return END

 

8. 노드와 엣지 추가

# 노드 추가
graph_builder.add_node("chatbot", chatbot_with_tools)
graph_builder.add_node("tool", tool_node)

# 시작점에서 챗봇으로 연결
graph_builder.add_edge(START, "chatbot")

# 조건부 엣지 추가: 챗봇 -> (도구 필요 여부에 따라) -> 도구 또는 종료
graph_builder.add_conditional_edges(
    "chatbot",  # 출발 노드
    route_to_tool_or_end,  # 라우팅 함수
    {
        "tool": "tool",  # "tool" 반환 시 tool 노드로
        END: END  # END 반환 시 그래프 종료
    }
)

# 도구 실행 후 다시 챗봇으로 돌아가는 엣지
graph_builder.add_edge("tool", "chatbot")

 

9. 그래프 컴파일 및 실행

# 그래프 컴파일
graph = graph_builder.compile()

# 그래프 실행 함수
def process_message(user_input: str):
    """사용자 입력을 처리하고 결과 반환"""
    # 사용자 메시지를 상태에 포함
    initial_state = {"messages": [{"role": "user", "content": user_input}]}
    
    # 그래프 실행
    for event in graph.stream(initial_state):
        for value in event.values():
            # 각 노드의 결과 확인
            if "messages" in value and value["messages"]:
                last_message = value["messages"][-1]
                if hasattr(last_message, "content") and last_message.content:
                    # AI 응답 출력
                    if last_message.type == "ai":
                        print(f"AI: {last_message.content}")

 

10. 대화 인터페이스 만들기

def chat_interface():
    """간단한 대화 인터페이스"""
    print("AI 어시스턴트와 대화를 시작합니다. 종료하려면 'exit'를 입력하세요.")
    
    while True:
        user_input = input("사용자: ")
        if user_input.lower() in ["exit", "quit", "종료"]:
            print("대화를 종료합니다.")
            break
        
        process_message(user_input)

# 대화 시작
if __name__ == "__main__":
    chat_interface()