메시지 요약 기능을 갖춘 챗봇
복습
그래프 상태 스키마와 리듀서를 커스터마이징하는 방법을 다뤘습니다.
또한 그래프 상태에서 메시지를 트리밍하거나 필터링하는 여러 방법을 보여드렸습니다.
목표
이제 한 단계 더 나아가보겠습니다!
메시지를 단순히 트리밍하거나 필터링하는 대신, LLM을 사용하여 대화의 실행 요약을 생성하는 방법을 보여드리겠습니다.
이를 통해 트리밍이나 필터링으로 단순히 제거하는 것이 아니라, 전체 대화의 압축된 표현을 유지할 수 있습니다.
이 요약 기능을 간단한 챗봇에 통합할 것입니다.
그리고 해당 챗봇에 메모리를 장착하여 높은 토큰 비용/지연 시간을 발생시키지 않으면서 장기 실행 대화를 지원할 것입니다.
%%capture --no-stderr
%pip install --quiet -U langchain_core langgraph langchain_openaifrom dotenv import load_dotenv
load_dotenv("../.env", override=True)True
import os
import getpass
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")tracing을 위해 LangSmith를 사용하겠습니다.
langchain-academy 프로젝트에 로깅할 것입니다.
_set_env("LANGSMITH_API_KEY")
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "langchain-academy"from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o", temperature=0)이전과 마찬가지로 MessagesState를 사용하겠습니다.
내장된 messages 키 외에도 이제 커스텀 키(summary)를 포함할 것입니다.
from langgraph.graph import MessagesState
class State(MessagesState):
summary: str요약이 존재하는 경우 이를 프롬프트에 통합하는 LLM을 호출하는 노드를 정의하겠습니다.
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage
# 모델 호출 로직 정의
def call_model(state: State):
# 요약이 존재하면 가져오기
summary = state.get("summary", "")
# 요약이 있다면 추가합니다
if summary:
# 시스템 메시지에 요약 추가
system_message = f"이전 대화 요약: {summary}"
# 요약문을 최신 메시지에 추가하십시오
messages = [SystemMessage(content=system_message)] + state["messages"]
else:
messages = state["messages"]
response = model.invoke(messages)
return {"messages": response}요약을 생성하는 노드를 정의하겠습니다.
여기서는 요약을 생성한 후 상태를 필터링하기 위해 RemoveMessage를 사용할 것입니다.
def summarize_conversation(state: State):
# 먼저, 기존 요약문을 가져옵니다.
summary = state.get("summary", "")
# 요약 프롬프트 생성
if summary:
# 요약본이 이미 존재합니다
summary_message = (
f"지금까지의 대화 요약은 다음과 같습니다: {summary}\n\n"
"위의 새로운 메시지를 고려하여 요약을 확장하십시오:"
)
else:
summary_message = "위의 대화 내용을 요약하세요:"
# 우리의 기록에 프롬프트를 추가하세요
messages = state["messages"] + [HumanMessage(content=summary_message)]
response = model.invoke(messages)
# 가장 최근의 2개 메시지를 제외한 모든 메시지를 삭제하세요
delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
return {
"summary": response.content,
"messages": delete_messages,
}대화 길이를 기반으로 요약을 생성할지 여부를 결정하는 조건부 엣지를 추가하겠습니다.
from langgraph.graph import END
from typing_extensions import Literal
# 대화를 종료할지 요약할지 결정하십시오
def should_continue(state: State) -> Literal["summarize_conversation", END]:
"""Return the next node to execute."""
messages = state["messages"]
# 메시지가 여섯 개 이상일 경우 대화를 요약합니다
if len(messages) > 6:
return "summarize_conversation"
# 그렇지 않으면 그냥 종료합니다.
return END메모리 추가하기
상태는 단일 그래프 실행에 일시적이라는 것을 기억하세요.
이는 중단이 있는 다회차 대화를 수행하는 능력을 제한합니다.
모듈 1의 끝부분에서 소개했듯이, 이를 해결하기 위해 지속성(persistence)을 사용할 수 있습니다!
LangGraph는 체크포인터를 사용하여 각 단계 후 그래프 상태를 자동으로 저장할 수 있습니다.
이 내장된 지속성 레이어는 메모리를 제공하여 LangGraph가 마지막 상태 업데이트부터 다시 시작할 수 있도록 합니다.
이전에 보여드렸듯이, 가장 사용하기 쉬운 것 중 하나는 그래프 상태를 위한 인메모리 키-값 저장소인 MemorySaver입니다.
체크포인터로 그래프를 컴파일하기만 하면 그래프에 메모리가 생깁니다!
from IPython.display import Image, display
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
# Define a new graph
workflow = StateGraph(State)
workflow.add_node("conversation", call_model)
workflow.add_node(summarize_conversation)
# Set the entrypoint as conversation
workflow.add_edge(START, "conversation")
workflow.add_conditional_edges("conversation", should_continue)
workflow.add_edge("summarize_conversation", END)
# Compile
memory = MemorySaver()
graph = workflow.compile(checkpointer=memory)
# display(Image(graph.get_graph().draw_mermaid_png()))
스레드
체크포인터는 각 단계의 상태를 체크포인트로 저장합니다.
이렇게 저장된 체크포인트들은 대화의 스레드로 그룹화될 수 있습니다.
Slack을 비유로 생각해보세요: 서로 다른 채널이 서로 다른 대화를 담습니다.
스레드는 Slack 채널과 같아서, 그룹화된 상태 모음(예: 대화)을 포착합니다.
아래에서는 configurable을 사용하여 스레드 ID를 설정합니다.
# Create a thread
config = {"configurable": {"thread_id": "1"}}
# Start conversation
input_message = HumanMessage(content="안녕하세요! 저는 랜스입니다.")
output = graph.invoke({"messages": [input_message]}, config)
for m in output["messages"][-1:]:
m.pretty_print()
input_message = HumanMessage(content="내 이름이 뭐지?")
output = graph.invoke({"messages": [input_message]}, config)
for m in output["messages"][-1:]:
m.pretty_print()
input_message = HumanMessage(content="저는 49ers를 좋아해요!")
output = graph.invoke({"messages": [input_message]}, config)
for m in output["messages"][-1:]:
m.pretty_print()==================================[1m Ai Message [0m==================================
안녕하세요, 랜스님! 만나서 반갑습니다. 어떻게 도와드릴까요?
==================================[1m Ai Message [0m==================================
당신의 이름은 랜스라고 하셨습니다. 다른 질문이나 도움이 필요하시면 말씀해 주세요!
==================================[1m Ai Message [0m==================================
샌프란시스코 49ers를 좋아하시는군요! 49ers는 NFL에서 매우 인기 있는 팀 중 하나로, 특히 그들의 역사적인 성공과 전설적인 선수들로 유명합니다. 팀이나 선수에 대해 더 이야기하고 싶으신 게 있나요?
graph.get_state(config).values.get("summary", "")''
이제 아직 상태의 요약이 없는데, 아직 6개 이하의 메시지를 가지고 있기 때문입니다.
이는 should_continue에서 설정되었습니다.
# 메시지가 6개를 초과하면 대화를 요약합니다
if len(messages) > 6:
return "summarize_conversation"
스레드가 있기 때문에 대화를 이어갈 수 있습니다.
스레드 ID가 있는 config를 사용하면 이전에 기록된 상태에서 계속 진행할 수 있습니다!
input_message = HumanMessage(
content="닉 보사 좋아하는데, 그 선수 수비수 중 최고 연봉자 아니야?"
)
output = graph.invoke({"messages": [input_message]}, config)
for m in output["messages"][-1:]:
m.pretty_print()==================================[1m Ai Message [0m==================================
네, 닉 보사는 2023년 기준으로 NFL 수비수 중 최고 연봉을 받는 선수입니다. 그는 샌프란시스코 49ers와의 계약 연장을 통해 이 기록을 세웠습니다. 보사는 뛰어난 수비 능력과 경기력으로 팀에 큰 기여를 하고 있으며, 많은 팬들에게 사랑받고 있습니다. 그의 활약을 지켜보는 것은 정말 흥미진진하죠!
graph.get_state(config).values.get("summary", "")'사용자는 자신을 랜스라고 소개하며 샌프란시스코 49ers를 좋아한다고 말했습니다. 특히 닉 보사를 좋아하며, 그가 NFL 수비수 중 최고 연봉자라는 사실을 언급했습니다.'
LangSmith
Let’s review the trace!
https://smith.langchain.com/public/f8468b91-a5cd-4573-b703-6afa9d374981/r