[LangGraph] Message

📌 LangGraph - Message

1️⃣ Message 사용

• LangGraph는 메시지 목록 기반의 채팅 모델 인터페이스를 활용

• HumanMessage와 AIMessage 등 다양한 메시지 타입을 지원

• 그래프 상태에서 대화 기록은 메시지 객체 리스트로 저장되며, 이를 통해 효율적인 대화 관리를 가능

• reducer 함수를 통해 상태 업데이트 시 메시지 목록이 어떻게 갱신될지 정의할 수 있음

---

1. operator.add

• 메시지 목록에 새로운 메시지를 간단히 추가하는 기본적인 reducer 함수

  1. messages 키가 메시지 리스트를 저장
  2. add reducer가 새 메시지를 기존 리스트에 추가
  3. 모든 종류의 메시지(HumanMessage, AIMessage 등)가 허용됨

• 주의사항:

  • operator.add는 단순히 리스트를 연결
  • 중복 메시지도 추가됨
  • 메시지 삭제나 수정은 불가능

from langchain_core.messages import AnyMessage
from operator import add
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI

# 상태 정의
class GraphState(TypedDict):
    messages: Annotated[list[AnyMessage], add]

# LLM 인스턴스 생성
llm = ChatOpenAI(model="gpt-4o-mini")

# chatbot 노드 함수 정의
def chatbot(state: GraphState) -> GraphState:
    # LLM을 사용하여 챗봇 메시지 생성
    return {"messages": [llm.invoke(state["messages"])]}

그래프 정의 및 실행

# Workflow Graph
builder = StateGraph(GraphState)

builder.add_node("chatbot", chatbot)

builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

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

# 초기 상태
initial_state = {"messages": [("user", "안녕하세요!")]}

# 그래프 실행
for event in graph.stream(initial_state, stream_mode="values"):
    pprint(event['messages'])
    print("-"*100)

- 출력

[('user', '안녕하세요!')]
----------------------------------------------------------------------------------------------------
[('user', '안녕하세요!'),
 AIMessage(content='안녕하세요! 어떻게 도와드릴까요?', additional_kwargs={'refusal': None}, response_metadata={'token_usage': ... })]
----------------------------------------------------------------------------------------------------

---

2. add_messages

• 메시지 ID를 기반으로 기존 메시지를 업데이트하거나 새 메시지를 추가하는 고급 관리 기능을 제공

  • 새 메시지는 기존 목록에 추가
  • 기존 메시지 업데이트도 올바르게 처리 (메시지 ID를 추적)

• 기존 메시지의 중복 추가를 방지

from typing import Annotated
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

# add_messages 사용 상태 정의
class GraphState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

# LLM 인스턴스 생성
llm = ChatOpenAI(model="gpt-4o-mini")

# chatbot 노드 함수 정의
def chatbot(state: GraphState) -> GraphState:
    # LLM을 사용하여 챗봇 메시지 생성
    return {"messages": [llm.invoke(state["messages"])]}

# Workflow Graph
builder = StateGraph(GraphState)

builder.add_node("chatbot", chatbot)

builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

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

# 초기 상태
initial_state = {"messages": [("user", "안녕하세요!")]}

# 그래프 실행
for event in graph.stream(initial_state, stream_mode="values"):
    pprint(event['messages'])
    print("-"*100)

- 출력

[HumanMessage(content='안녕하세요!', additional_kwargs={}, response_metadata={}, id='38c96e72-6b38-4644-ad2b-8a962331ad61')]
----------------------------------------------------------------------------------------------------
[HumanMessage(content='안녕하세요!', additional_kwargs={}, response_metadata={}, id='38c96e72-6b38-4644-ad2b-8a962331ad61'),
 AIMessage(content='안녕하세요! 어떻게 도와드릴까요?', additional_kwargs={'refusal': None}, response_metadata={'token_usage': ... )]
----------------------------------------------------------------------------------------------------

---

3. MessageState

• MessagesState 는 메시지 관리를 위해 미리 정의된 상태 타입

• 이 상태는 add_messages reducer를 기본으로 사용하여 메시지 업데이트를 자동으로 처리

• AnyMessage 객체 리스트를 포함하는 단일 messages 키로 구성되어 있어 구조가 단순함

from langgraph.graph import MessagesState

# messages 키를 가진 상태 생성 (messages 키는 기본 제공)
class GraphState(MessagesState):  # MessagesState 상속
    ... 
    # 추가적인 필드 정의 가능
    # custom_field: str

# LLM 인스턴스 생성
llm = ChatOpenAI(model="gpt-4o-mini")

# chatbot 노드 함수 정의
def chatbot(state: GraphState) -> GraphState:
    # LLM을 사용하여 챗봇 메시지 생성
    return {"messages": [llm.invoke(state["messages"])]}

# Workflow Graph
builder = StateGraph(GraphState)

builder.add_node("chatbot", chatbot)

builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

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

# 초기 상태
initial_state = {"messages": [("user", "안녕하세요!")]}

# 그래프 실행
for event in graph.stream(initial_state, stream_mode="values"):
    pprint(event['messages'])
    print("-"*100)

- 출력

[HumanMessage(content='안녕하세요!', additional_kwargs={}, response_metadata={}, id='ed15fe8f-ae5b-4e73-890f-a922a884b7d4')]
----------------------------------------------------------------------------------------------------
[HumanMessage(content='안녕하세요!', additional_kwargs={}, response_metadata={}, id='ed15fe8f-ae5b-4e73-890f-a922a884b7d4'),
 AIMessage(content='안녕하세요! 어떻게 도와드릴까요?', additional_kwargs={'refusal': None}, response_metadata={'token_usage': ... )]
----------------------------------------------------------------------------------------------------

---

4. MessageState 커스텀 필드 추가

• MessagesState를 상속받아 추가 필드를 포함하는 새로운 상태 타입을 정의할 수 있음

• 기존 messages 키의 add_messages reducer 기능을 그대로 유지

from typing import Optional
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_openai import ChatOpenAI

# MessagesState를 상속하여 커스텀 필드 추가
class GraphState(MessagesState):
    # 사용자의 감정 상태를 추적하는 필드 추가
    emotion: Optional[str] 

# LLM 인스턴스 생성
llm = ChatOpenAI(model="gpt-4o-mini")

# 감정 분석을 위한 프롬프트 템플릿
EMOTION_PROMPT = """
사용자의 메시지를 분석하여 감정 상태를 파악해주세요.
가능한 감정 상태: 행복, 슬픔, 화남, 중립

사용자 메시지: {message}

감정 상태만 한 단어로 답변해주세요.
"""

# 감정 분석 노드
def analyze_emotion(state: GraphState) -> GraphState:
    # 가장 최근 사용자 메시지 가져오기
    last_message = state["messages"][-1].content
    
    # 감정 분석 실행
    emotion_analysis = llm.invoke(EMOTION_PROMPT.format(message=last_message))
    
    # 상태 업데이트
    return {
        "emotion": emotion_analysis.content.strip()
    }

# 챗봇 응답 노드
def chatbot(state: GraphState) -> GraphState:
    # 현재 감정 상태를 고려한 시스템 메시지 생성
    system_message = f"""
    사용자의 현재 감정 상태는 {state['emotion']}입니다.
    이를 고려하여 공감적이고 적절한 응답을 해주세요.
    """
    
    # 기존 메시지에 시스템 메시지 추가
    messages = [{"role": "system", "content": system_message}] + state["messages"]
    
    # LLM 응답 생성
    response = llm.invoke(messages)
    
    return {"messages": [response]}

# Workflow Graph 구성
builder = StateGraph(GraphState)

# 노드 추가
builder.add_node("analyze_emotion", analyze_emotion)
builder.add_node("chatbot", chatbot)

# 엣지 추가
builder.add_edge(START, "analyze_emotion")
builder.add_edge("analyze_emotion", "chatbot")
builder.add_edge("chatbot", END)

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

# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))

- 출력

---

그래프 실행

# 초기 상태
initial_state = {
    "messages": [{"role": "user", "content": "오늘 정말 힘든 하루였어요..."}]
}

# 그래프 실행
for event in graph.stream(initial_state, stream_mode="values"):
    if "emotion" in event:
        print(f"감정 상태: {event['emotion']}")
    if "messages" in event:
        print("메시지:")
        for msg in event["messages"]:
            print(f"{msg.type}: {msg.content}")
    print("-"*100)

- 출력

메시지:
human: 오늘 정말 힘든 하루였어요...
----------------------------------------------------------------------------------------------------
감정 상태: 슬픔
메시지:
human: 오늘 정말 힘든 하루였어요...
----------------------------------------------------------------------------------------------------
감정 상태: 슬픔
메시지:
human: 오늘 정말 힘든 하루였어요...
ai: 정말 힘든 하루셨군요. 그런 날은 누구에게나 있을 수 있죠. 마음이 무겁고 슬플 때, 누군가와 이야기하는 것이 큰 도움이 되기도 해요. 어떤 일들이 있었는지 나눠보고 싶으세요? 듣고 싶어요.
----------------------------------------------------------------------------------------------------