고급 Agent를 위한 Langgraph 개념 이해 #2 - Node

2. Node

Node는 State 정보를 활용해서, State 정보를 변경하거나 action을 취하는 그래프내의 함수이다. 

함수내에서는 LLM 모델을 실행해서 질문에 대한 정답을 찾거나 , Tool 등을 이용해서 Action을 취하거나, 또는 RAG를 통해서 질문에 필요한 추가 컨택스트 정보를 검색해와서 State에 저장해서 다음 Node에서 LLM이 답변에 활용할 수 있도록 한다. 

 

예를 들어 web_search,chat 이라는 Node가 있을때, web_search Node에서는 Google 검색을 이용하여, 질문에 필요한 추가 정보를 검색해서 State에 저장한후, 다음 chat 노드에서 해당 State에 저장된 Context 정보를 리턴하는 식으로 구현이 된다. 

 

Node는 Python function으로 정의되며, 입력 parameter는 State 타입을 사용하며, 리턴은 State 정보의 스키마를 따르는 Dictionary나 TypedDict를 데이터 형으로 사용한다. 

아래 그림은 노드를 Pseudo 코드로 간단하게 구현해 놓은 예제이다.  

class State(TypedDict):
   question: str
   answer: str

def basic_node(state: State) -> dict:
# 1. State에서 처리에 필요한 Input을 받음
question = state.get("question") 
# 2. Input을 통해서 LLM을 호출
answer_from_llm = llm.invoke(question) 
# 3. LLM에서 받은 Output을 State에 업데이트하여 다음 Node로 리턴
return {"answer":answer_from_llm}

• State는 질문(question)과 답변(answer) 두개의 필드를 가지고 있다.
• basic_node 노드는 State정보를 받은 후에, State 에서 질문(question)을 이용하여 LLM을 호출하고, 호출된 결과를 State의 answer에 넣기 위해서 리턴으로 Dictionary 타입으로 “answer”를 키에 LLM으로 부터 받은 응답인 answer_from_llm을 넣어서 리턴한다. 

Config

Node에는 추가적으로, 두번째 인자를 통해서 config 정보를 전달할 수 있다. config 는 Runnable Config 타입으로, 사용하고자 하는 LLM 이름이나, 사용자 정보, 로깅/모니터링을 위한 request id 등의 메타 정보를 저장해서 Node로 전달할 수 있다. State 정보와 유사하게 전역변수의 역할을 하지만, State 정보는 Node가 답변을 내는데 필요한 정보라고 하면, Config 정보는 답변 생성에는 사용하지 않지만 Node의 행동을 정의하기 위한 추가적인 메타 정보라고 생각하면 된다. 

 

아래 예제는 Langgraph 공식홈페이지에 나와있는 Config 사용법에 대한 가이드 문서이다. 

from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, StateGraph, START
from typing_extensions import TypedDict

# 1. Specify config schema
class ConfigSchema(TypedDict):
    my_runtime_value: str

# 2. Define a graph that accesses the config in a node
class State(TypedDict):
    my_state_value: str

def node(state: State, config: RunnableConfig):
    if config["configurable"]["my_runtime_value"] == "a":
        return {"my_state_value": 1}
    elif config["configurable"]["my_runtime_value"] == "b":
        return {"my_state_value": 2}
    else:
        raise ValueError("Unknown values.")

builder = StateGraph(State, config_schema=ConfigSchema)
builder.add_node(node)
builder.add_edge(START, "node")
builder.add_edge("node", END)

graph = builder.compile()

# 3. Pass in configuration at runtime:
print(graph.invoke({}, {"configurable": {"my_runtime_value": "a"}}))
print(graph.invoke({}, {"configurable": {"my_runtime_value": "b"}}))

• Config를 사용하는 방법은 간단하다.먼저 Config에 들어갈 정보를 Schema 형태로 정의해야 한다. class ConfigSchema(TypedDict):에서 TypedDict 타입으로 ConfigSchema를 정의하였고, 필드로는 my_runtime_value: str 를 정의하였다. 
•  Node 정의시 두번째 인자로 config : RunnableConfig를 넘기도록 하면, 해당 Node내에서 Config에 저장된 정보를 access할 수 있다. def node(state: State, config: RunnableConfig) 에서 Node를 정의하는 부분을 보면 두번째 인자로, config를 넘긴것을 볼 수 있다. 
• Config 정보를 access하기 위해서는 예제의 config["configurable"]["my_runtime_value"] 와 같이, config["configurable"]["키 이름”] 식으로 해당 값을 접근할 수 있다. 
• 마지막으로 Config 정보를 전달하기 위해서는 그래프 생성시에, builder = StateGraph(State, config_schema=ConfigSchema)에서 인자로 config_schema에 앞에서 정의한 ConfigSchema를 인자로 넘긴다. 
• Config 정보를 전달하는 방법은 Graph를 실행할때, print(graph.invoke({}, {"configurable": {"my_runtime_value": "a"}})) 처럼, invoke나 stream 메서드의 두번째 인자로 "configurable" 키로, Config값들을 Dictionary 형태로 인자로 넘긴다. 

Caching

Node에서 또 유용하게 사용할 수 있는 기능은 Cache 기능이다.

아래 그림과 같이 일반적으로 Node가 호출되면, State에 있는 정보로, Node는 LLM 모델을 호출하거나 외부 Tool을 호출하여 다음 Node로 전달한다. 

<그림. 일반적으로 Node에서 LLM 을 호출하는 구조>

 

그런데, 만약에 State에 있는 정보가 이전에 사용했던 정보와 동일하거나, 유사하다면 다시 LLM을 호출하는 것은 비용적으로 낭비가 될 수 있다. 그래서 Node는 State 정보를 키로 이용하여 응답을 캐쉬에 저장하고, 같은 키가 들어왔을 경우, 캐슁해놓은 응답을 리턴하도록 한다.  

이 캐슁 기법은 로직을 수행하는 시간을 줄여서 응답 시간을 높일뿐만 아니라, LLM 등 외부 호출을 줄임으로서 LLM API 비용등을 절약할 수 있다. 

<그림. 캐쉬를 적용하는 구조>

코드를 통해서 구현 방법을 살펴보자.

먼저 Key와 Value를 저장할 캐쉬를 생성해야 한다. 이 예제에서는 In-memory를 캐슁으로 사용하기 위해서 MemorySaver를 생성하였다. 간단한 예제이기 때문에, In-memory 캐슁을 사용하였지만, Redis나 외부 RDBMS등도 캐슁 저장소로 사용할 수 있다. 

# 인메모리 캐쉬를 위해 MemorySaver를 사용합니다.

memory = MemorySaver()

다음으로 캐슁 키 함수를 정의해야 한다. State는 Dictionary 형태로, 여러개의 필드를 포함하고 있기 때문에, State를 하나의 단일 키로 표현해줄 수 있는 방법이 필요한데, 이 캐슁 키 함수가 이 역할을 한다. 질문에 대한 Embedding 값을 사용해도 되고, State 필드내의 특정 값을 사용해도 된다. 여기서는 State내의 “question” 문자열을 키로 사용하도록 하였다. 

# 캐쉬 함수 정의 
def get_cache_key(state: ChatState, config: dict) -> str:
    return state["question"]

이 리턴되는 키에 따라서 캐쉬에서 해당 키에 맵핑 되는 값이 있으면, Node의 내용을 수행하지 않고 저장되어 있는 State값을 리턴하게 된다.  캐슁을 사용하는 Node는 일반적으로 정의하는 함수 기반의 Node를 사용하면 안되고, 정의된 함수 기반의 Node를 NodeBuilder에 .cache 메서드를 이용하여 캐쉬키 생성 함수를 지정해서 생성해야 한다. 

cached_node_builder = NodeBuilder(generate_answer).cache(get_cache_key)

 

그리고 마지막으로, 그래프를 Building할때 checkpointer=memory로 지정해서, Graph에서 State정보를 저장할 수 있는 컴포넌트를 지정한다. 

graph = builder.compile(checkpointer=memory)

 

아래는 앞에서 설명한 코드의 풀 코드이다. 

import os
from typing import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.pregel import NodeBuilder
os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"

class ChatState(TypedDict):
    question: str
    answer: str
llm = ChatOpenAI(model="gpt-4o-mini")
# 인메모리 캐쉬를 위해 MemorySaver를 사용합니다.
memory = MemorySaver()

# 노드 정의
def generate_answer(state: ChatState) -> dict:
    """LLM을 호출하여 답변을 생성하는 노드 함수"""
    print("\n>>>>> LLM 노드 실행! (이 메시지는 캐쉬되지 않았을 때만 보입니다) <<<<<")
    question = state["question"]
    response = llm.invoke(question)
    return {"answer": response.content}

# 캐쉬 함수 정의 
def get_cache_key(state: ChatState, config: dict) -> str:
    return state["question"]

# 노드 빌더를 이용하여 Cache 기능을 갖는 노드를 정의함
cached_node_builder = NodeBuilder(generate_answer).cache(get_cache_key)
# StateGraph 빌더를 생성합니다.
builder = StateGraph(ChatState)
# NodeBuilder로 생성한 캐쉬 적용 노드를 그래프에 추가합니다.
builder.add_node("generate_answer_node", cached_node_builder.build())
builder.add_edge(START, "generate_answer_node")
builder.add_edge("generate_answer_node", END)

# Checkpointer(캐쉬)를 연결하여 그래프를 컴파일합니다.
graph = builder.compile(checkpointer=memory)

# ---  그래프 실행 및 캐싱 확인 ---
# 대화 스레드를 식별하기 위한 설정
config = {"configurable": {"thread_id": "chat-thread-1"}}
question = "LangGraph에서 노드 캐싱은 어떻게 하나요?"
initial_state = {"question": question, "answer": ""}
# 첫 번째 호출: LLM 노드가 실행됩니다.
print("--- 첫 번째 질문 ---")
print(f"질문: {question}")
final_state = graph.invoke(initial_state, config)
print(f"답변: {final_state['answer']}")
# 두 번째 호출: 동일한 질문이므로 캐쉬된 결과를 반환합니다.
print("\n\n--- 두 번째 질문 (동일한 내용) ---")
print(f"질문: {question}")
# 'LLM 노드 실행!' 메시지가 출력되지 않아야 합니다.
cached_state = graph.invoke(initial_state, config)
print(f"답변: {cached_state['answer']}")
# 세 번째 호출: 다른 질문이므로 LLM 노드가 다시 실행됩니다.
print("\n\n--- 세 번째 질문 (다른 내용) ---")
new_question = "LangChain의 장점은 무엇인가요?"
new_initial_state = {"question": new_question, "answer": ""}
print(f"질문: {new_question}")
new_final_state = graph.invoke(new_initial_state, config)
print(f"답변: {new_final_state['answer']}")

 

앞의 코드에서는 Cache의 동작원리에 대한 이해를 돕기 위해서 간단하게 Key를 질문 문자열을 키로 사용했지만, 실제 애플리케이션에서는 질문이 글자하나 안틀리고 정확하게 일치하기는 어렵다. 그래서 일반적으로 문장을 키로 사용할 경우에는 임베딩을 이용해서, 문장간의 유사도가 일정 수준 (Threshold)이면 같은 문맥의 문장으로 취급해서 캐쉬에 저장된 State 정보를 사용하도록 할 수 있다. 

 

Langgraph의 캐쉬는 cache key function에서 리턴된 키값이 일치하면, 캐쉬에 저장되어 있다고 판단하기 때문에, 캐쉬에 저장된 문장과, 새롭게 질의된 문장이 서로 유사하면, 캐쉬에 저장된 문장을 키로 리턴하게 되면, Langgraph는 이미 저장된 캐쉬키가 리턴되었기 때문에 캐쉬에서 저장된 State 정보를 리턴하게 된다. 

이를 구현하기 위해서는 캐쉬에 저장된 키와 새롭게 질의된 문장이 서로 유사한지를 판단해야 하는데, 유사도 검색 알고리즘을 이용한다. 

별도의 벡터 데이터베이스를 설정한 후에, 캐쉬에 저장되는 키 값에 대한 임베딩 벡터를 벡터데이터 베이스에 저장하고, 새롭게 질의된 문장의 임베딩 값을 계산해서, 벡터데이터베이스에 기 저장된 문장의 임베딩값과 비교한다. 

 

아래 코드는 벡터데이터베이스에서 새롭게 질의된 문장과 유사한 문장을 검색하는 코드이다. 

search_results = vector_store.similarity_search_with_score(question, k=1)

만약에 검색된 결과가 SIMLARITY_THRESHOLD (기준값) 이상이면, 기존에 저장된 질문을 캐쉬 값으로 리턴한다. 

    if search_results and search_results[0][1] < SIMILARITY_THRESHOLD:
        # 캐쉬 히트: 기존에 저장된 원본 질문을 키로 반환
        cached_question = search_results[0][0].page_content
        print(f"\n>>>>> 시맨틱 캐쉬 히트! 기존 키 '{cached_question}'를 재사용합니다. <<<<<")
        return cached_question

만약에, 저장되어 있지 않은 키로 판명이 되면, 벡터데이터베이스에 새로운 질문을 키로 새로 등록하고, 새로운 질문을 키값으로 리턴한다. (새로운 질문은 캐쉬에 들어가 있지 않기 때문에, 새로운 질문을 키로 리턴하면 캐쉬는 Hit 되지 않고, 다음을 위해서 그 키와 값을 캐쉬에 등록하게 된다.)

        vector_store.add_documents([Document(page_content=question)])
        return question

아래는 앞에서 설명한 내용을 FAISS 벡터데이터 베이스 기반으로 재구현한 cache key function이다. 

# Embedding API 정의
embeddings = OpenAIEmbeddings()
# FAISS 벡터 저장소를 전역으로 생성하여 초기화 
vector_store = FAISS.from_documents(
    [Document(page_content="initial_dummy_document")], embeddings
)
# ---  캐쉬 키 결정자 및 노드 함수 정의 ---
def get_cache_key(state: ChatState, config: dict) -> str:
    """
    유사도 검색을 수행하여 사용할 캐쉬 키를 결정합니다.
    - 유사한 질문이 있으면: 기존 질문을 키로 반환 (캐쉬 히트 유도)
    - 없으면: 새 질문을 저장하고, 새 질문을 키로 반환 (캐쉬 미스)
    """
    question = state["question"]
    # 유사도 검색 (L2 거리가 0.1 미만이면 매우 유사하다고 판단)
    search_results = vector_store.similarity_search_with_score(question, k=1)
    SIMILARITY_THRESHOLD = 0.1
    if search_results and search_results[0][1] < SIMILARITY_THRESHOLD:
        # 캐쉬 히트: 기존에 저장된 원본 질문을 키로 반환
        cached_question = search_results[0][0].page_content
        print(f"\n>>>>> 시맨틱 캐쉬 히트! 기존 키 '{cached_question}'를 재사용합니다. <<<<<")
        return cached_question
    else:
        # 캐쉬 미스: 새 질문을 벡터 저장소에 등록하고, 새 질문을 키로 사용
        print(f"\n>>>>> 시맨틱 캐쉬 미스! '{question}'을(를) 새 캐쉬 키로 등록합니다. <<<<<")
        vector_store.add_documents([Document(page_content=question)])
        return question

 

START와 END Node

Start와 End Node는 특수목적 노드로, 그래프가 어디서 부터 시작될건지와, 그래프가 어디서 끝날것인지, 시작점과 끝점을 정의하는 노드이다.

from langgraph.graph import START
from langgraph.graph import END

graph.add_edge(START, "node_a")
graph.add_edge("node_a", END)