LangGraph 실습

대분류

인공지능/데이터

프레임워크

소분류

LangChain/RAG

유형

LangChain

부유형

Application LangChain

최종 편집 일시

2024/10/31 08:09

생성 일시

2024/10/31 05:51

14 more properties

기본 동작 구조

상태(State) 정의 

노드(Node) 정의

그래프 생성

노드 추가

엣지 추가

그래프 컴파일

그래프 시각화(선택)

그래프 실행

.env

OPENAI_API_KEY='sk-...'

LANGCHAIN_TRACING_V2='ture'
LANGCHAIN_ENDPOINT="https://api.smith.langchain.com"
LANGCHAIN_PROJECT="GRAPH TUTORIAL"
LANGCHAIN_API_KEY="lsv2_..."
Plain Text
복사

베이직 OpenAI Graph v1

from dotenv import load_dotenv

# .env 파일에 등록된 변수(데이터)를 os 환경변수에 적용
load_dotenv()
Python
복사

State

•

add_messages

기존 메시지가 있으면, update / 없으면, add

from typing import Annotated  
# 타입 힌트를 위한 Annotated

from typing_extensions import TypedDict  
# TypedDict를 사용하기 위해 typing_extensions에서 임포트
from langgraph.graph.message import add_messages  
# 메시지 추가를 위한 add_messages 함수를 임포트

class State(TypedDict):
    messages: Annotated[list, add_messages]  
    # messages 키는 add_messages가 적용된 리스트 타입
Python
복사

Node

•

모델 설정

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
Python
복사

•

챗봇 상태 머신 지정

def chatbot(state:State) -> State:
    return {
        "messages": [
            llm.invoke(
                state["messages"]
            )
        ]
    }
Python
복사

Create Graph

from langgraph.graph import StateGraph

# graph 객체 생성(선언)
# simple_graph 상 데이터 전달에 사용할 객체 -> State 클래스 
simple_graph = StateGraph(State)
Python
복사

Add Node

# openai_chatbot -> 나중에 엣지가 노드들을 식별하는데 사용하는 이름 
simple_graph.add_node(
    "openai_chatbot", chatbot
)

# 실행결과 : 메모리에 설정된다.
<langgraph.graph.state.StateGraph at 0x289d7da7260>
Python
복사

Start Node

# 그래프에 정의된(add_node) Node들 중에서 
# 시작으로 사용할 Node 선언(정의) 
simple_graph.set_entry_point(
    "openai_chatbot" 
)

# 실행결과 : 동일한 메모리에 설정된다.
<langgraph.graph.state.StateGraph at 0x289d7da7260>
Python
복사

End Node

# 그래프에 정의된(add_node) Node들 중에서 
# 끝으로 사용할 Node 선언(정의) 
simple_graph.set_finish_point(
    "openai_chatbot"
)

# 실행결과 : 동일한 메모리에 설정된다.
<langgraph.graph.state.StateGraph at 0x289d7da7260>
Python
복사

Compile Graph

# compile()를 통해서 객체화(메모리에 올리감)
# simple_graph에 포함된 Node들이 모두 메모리에 올라감
compiled_graph = simple_graph.compile()
Python
복사

Display Graph

from IPython.display import Image, display  
# IPython에서 이미지와 디스플레이 기능을 임포트한다

try:
    display(
        Image(
            compiled_graph.get_graph().draw_mermaid_png()  
            # 그래프를 Mermaid 형식의 PNG 이미지로 그린다
        )
    )
except:
    pass  # 예외가 발생해도 아무 동작도 하지 않는다
Python
복사

Run Graph

•

CallbackHandler

◦

Langsmith를 통해 Graph 로깅을 도와주는 객체

from langfuse.callback import CallbackHandler

handler = CallbackHandler()
Python
복사

from langchain_core.messages import HumanMessage

for chunk in compiled_graph.stream(
    # Start Node에게 전달할 Input data
    {
      "messages":[
          HumanMessage(content="대한민국의 수도는?")
      ]
    }, 
    # 로깅 처리할 설정 
    config={
        "callbacks":[
            handler
        ]
    }
):
    print(chunk)
    
{'openai_chatbot': {'messages': [AIMessage(content='대한민국의 수도는 서울입니다.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 13, 'total_tokens': 21, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_9b78b61c52', 'finish_reason': 'stop', 'logprobs': None}, id='run-7f6281ec-2849-4dfb-9a21-e4cf50b84a0e-0', usage_metadata={'input_tokens': 13, 'output_tokens': 8, 'total_tokens': 21, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}}
Python
복사

HumanMessage : 사람이 모델에 전달하는 메시지

Combined Node - Graph v2

State

from typing_extensions import TypedDict
Python
복사

class State(TypedDict):
    """
    {
      "input":"Hello",
      "result":5
    }
    """
    input: str    # 사용자 input data
    result: int   # Node의 result data
Python
복사

Node

def len_str(state:State) -> State:
    # hello 
    input = state["input"]
    # 5
    result = len(input)
    return {
        "result": result
    }
Python
복사

def add_one(state:State) -> State:
    # 5
    input = state["result"]
    # 6
    result = input + 1
    return {
        "result": result
    }
Python
복사

Create Graph

from langgraph.graph import StateGraph

# graph 객체 생성(선언)
# simple_graph 상 데이터 전달에 사용할 객체 -> State 클래스 
simple_graph = StateGraph(State)
Python
복사

Add Node

simple_graph.add_node(
    "len_str", len_str
)

simple_graph.add_node(
    "add_one", add_one
)

# 이번엔 다른 메모리
<langgraph.graph.state.StateGraph at 0x289f99bb8c0>
Python
복사

Add Edge

from langgraph.graph import START, END 
Python
복사

Start Node

•

start -> len_str

simple_graph.add_edge(
    START, "len_str"
)
Python
복사

Combined Nodes

•

len_str → add_one

simple_graph.add_edge(
    "len_str", "add_one"
)
Python
복사

End Node

•

add_one → end

simple_graph.add_edge(
    "add_one", END
)
Python
복사

Compile Graph

compiled_graph = simple_graph.compile()
Python
복사

Display Graph

from IPython.display import Image, display

try:
    display(
        Image(
            compiled_graph.get_graph().draw_mermaid_png()
        )
    )
except:
    pass 
Python
복사

Run Graph

for chunk in compiled_graph.stream(
    # Start Node에게 전달할 Input data
    {
      "input":"Hello World"
    }
):
    print(chunk)
    
    
len_str: Hello World
{'len_str': {'result': 11}}
add_one: 11
{'add_one': {'result': 12}}
Python
복사

Conditional Edge - Graph v3

State

from typing_extensions import TypedDict, Optional
Python
복사

class State(TypedDict):
    input:  Optional[str] = None
    node_output: Optional[int] = None 
    is_stop: Optional[bool] = False #기존과는 다르게 컨디션 변수가 삽입
Python
복사

Node

def len_str(state:State) -> State:
    input = state["input"]
    result = len(input)
    return {
        **state, # state에 저장된 데이터를 다음 Node 전달 
        "node_output": result
    }
Python
복사

def add_one(state:State) -> State:
    input = state["node_ouput"]
    is_stop = state["is_stop"]
    
    result = input + 1
    if result > 10:
        is_stop = True

    return {
        **state, # state에 저장된 데이터를 다음 Node 전달 
        "node_output": result,
        "is_stop": is_stop
    }
Python
복사

def add_two(state:State) -> State:
    input = state["node_ouput"]
    result = input + 2
    return {
        **state, # state에 저장된 데이터를 다음 Node 전달 
        "node_output": result
    }
Python
복사

Create Graph

from langgraph.graph import StateGraph

# graph 객체 생성(선언)
# simple_graph 상 데이터 전달에 사용할 객체 -> State 클래스 
simple_graph = StateGraph(State)
Python
복사

Add Nodes

simple_graph.add_node(
    "len_str", len_str
)

simple_graph.add_node(
    "add_one", add_one
)

simple_graph.add_node(
    "add_two", add_two
)
Python
복사

Add Edge

from langgraph.graph import START, END 

simple_graph.add_edge(
    START , "len_str"
)

simple_graph.add_edge(
    "len_str", "add_one"
)
Python
복사

Add Conditional Edge

def is_stop(state: State) -> str:
    is_stop = state["is_stop"]
    if is_stop:
        return "go_stop" 
    else:
        return "go_to_add_two_fnc"  
Python
복사

Compile Graph

compiled_graph = simple_graph.compile()
Python
복사

Display Graph

from IPython.display import Image, display

try:
    display(
        Image(
            compiled_graph.get_graph().draw_mermaid_png()
        )
    )
except:
    pass 
Python
복사

Run Graph

for chunk in compiled_graph.stream(
    # Start Node에게 전달할 Input data
    {
      "input":"Hello", "is_stop":False
    }
):
    print(chunk)
Python
복사

{'len_str': {'input': 'Hello', 'node_output': 5, 'is_stop': False}}
{'add_one': {'input': 'Hello', 'node_output': 6, 'is_stop': False}}
{'add_two': {'input': 'Hello', 'node_output': 8, 'is_stop': False}}
{'add_one': {'input': 'Hello', 'node_output': 9, 'is_stop': False}}
{'add_two': {'input': 'Hello', 'node_output': 11, 'is_stop': False}}
{'add_one': {'input': 'Hello', 'node_output': 12, 'is_stop': True}}
Python
복사

•

해당 과정을 보면 add_one에서 node_output가 10이 넘는지 안넘는지 관측자 역할을 하여 is_stop 변수를 조작하는 것을 알 수 있다.