배경 및 목적
이번 13기 AI 스터디에 앞서, 1DAY 1랭그래프, 즉 하루에 1개의 랭그래프 개념을 설명하는 부분을 기획하게 됐습니다.
-> 랭그래프 스터디에 관심이 있다면 신청해주세요!
스터디 신청링크
이 프로젝트의 목표는 랭그래프를 쉽게 입문하실 수 있게 상세한 개념부터 코드까지 함께 공부해보실 수 있게 정리해서 드릴 예정입니다.
저는 랭그래프를 처음들어봐요!, 1일차 글을 보고 싶다면!?
-> 링크로 접속하기!
참고 자료
활용 툴
파이썬, 코랩, 랭그래프, 랭체인
메시지 요약 및 외부 DB 메모리를 갖춘 LangGraph 챗봇 구축하기
1. 핵심 요약
LangGraph와 LLM을 활용하여 대화의 실시간 요약을 생성하고, 이를 시스템 메시지에 통합하여 효율성을 높입니다.
MemorySaver체크포인터를 사용해 대화 상태를 SQLite와 같은 외부 데이터베이스에 영구 저장하여 장기 대화를 지원합니다.메시지 수에 따라 자동으로 대화를 요약하는 조건부 로직과 스레드 ID를 통해 유연한 대화 흐름을 관리합니다.
2. 핵심 기능
1. 대화 요약 통합:
- LangGraph와 대형 언어 모델(LLM)을 활용하여 대화의 실시간 요약을 생성합니다.
- 요약된 정보를 시스템 메시지로 포함시켜, 대화의 핵심 내용을 유지하면서 토큰 비용과 응답 지연 시간을 절감합니다.
2. 외부 DB 메모리 도입:
- MemorySaver와 같은 체크포인터를 사용하여 대화 상태를 SQLite와 같은 외부 데이터베이스에 영구적으로 저장합니다.
- 이를 통해 노트북 커널, 혹은 서버를 재시작하더라도 이전 대화 상태를 복원할 수 있어, 장기적인 대화 유지가 가능합니다.
3. 조건부 로직 및 요약 트리거:
- 대화 메시지 수가 일정 기준(예: 6개)을 초과하면 자동으로 대화를 요약하는 조건부 에지를 설정합니다.
- RemoveMessage를 사용하여 요약 후 불필요한 메시지를 상태에서 제거함으로써, 상태 관리를 효율적으로 수행합니다.
3. 구현 단계
1. 환경 설정 및 라이브러리 설치:
```python
%%capture --no-stderr
%pip install --quiet -U langgraph-checkpoint-sqlite langchain_core langgraph langchain_openai
```
2. API 키 설정:
```python
import os, getpass
def setenv(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
setenv("OPENAI_API_KEY")
```
3. SQLite 체크포인터 설정:
- 인메모리 SQLite 데이터베이스를 사용하거나, 파일 경로를 지정하여 영구적인 저장소를 생성합니다.
```python
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
# 파일 경로를 지정하여 SQLite 데이터베이스 연결
db_path = "state_db/example.db"
conn = sqlite3.connect(db_path, check_same_thread=False)
# SqliteSaver 체크포인터 초기화
memory = SqliteSaver(conn)
```
4. 챗봇 로직 정의:
- LLM을 호출하여 메시지에 요약을 통합하고, 대화 길이에 따라 요약을 생성 합니다.
```python
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage
from langgraph.graph import END, MessagesState
model = ChatOpenAI(model="gpt-4o", temperature=0)
class State(MessagesState):
summary: str
def call_model(state: State):
summary = state.get("summary", "")
if summary:
system_message = f"Summary of conversation earlier: {summary}"
messages = [SystemMessage(content=system_message)] + state["messages"]
else:
messages = state["messages"]
response = model.invoke(messages)
return {"messages": response}
def summarize_conversation(state: State):
summary = state.get("summary", "")
if summary:
summary_message = (
f"This is summary of the conversation to date: {summary}\n\n"
"Extend the summary by taking into account the new messages above:"
)
else:
summary_message = "Create a summary of the conversation above:"
messages = state["messages"] + [HumanMessage(content=summary_message)]
response = model.invoke(messages)
delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
return {"summary": response.content, "messages": delete_messages}
```
5. 그래프 빌드 및 메모리 통합:
```python
from langgraph.graph import StateGraph, START
from IPython.display import Image, display
def should_continue(state: State):
return "summarize_conversation" if len(state["messages"]) > 6 else END
# 그래프 빌더 초기화 및 노드 추가
builder = StateGraph(State)
builder.add_node("conversation", call_model)
builder.add_node("summarize_conversation", summarize_conversation)
# 에지 추가
builder.add_edge(START, "conversation")
builder.add_conditional_edges("conversation", should_continue)
builder.add_edge("summarize_conversation", END)
# 그래프 컴파일 및 메모리 체크포인터 통합
graph = builder.compile(checkpointer=memory)
# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))
```
6. 대화 시작 및 스레드 관리:
```python
config = {"configurable": {"thread_id": "1"}}
# 초기 대화
input_message = HumanMessage(content="hi! I'm Lance")
output = graph.invoke({"messages": [input_message]}, config)
for m in output['messages'][-1:]:
m.pretty_print()
# 이어지는 대화
input_message = HumanMessage(content="what's my name?")
output = graph.invoke({"messages": [input_message]}, config)
for m in output['messages'][-1:]:
m.pretty_print()
input_message = HumanMessage(content="i like the 49ers!")
output = graph.invoke({"messages": [input_message]}, config)
for m in output['messages'][-1:]:
m.pretty_print()
# 요약 트리거 전 상태 확인
graph_state = graph.get_state(config)
print(graph_state)
```
4. 결론
LangGraph를 이용한 메시지 요약과 외부 DB 메모리 통합으로 챗봇의 대화 관리 능력이 크게 향상됩니다.
이를 통해 장기적이고 복잡한 대화를 효율적으로 처리하며, 토큰 비용과 응답 지연 시간을 절감할 수 있습니다.
외부 데이터베이스를 통한 지속성 도입과 유연한 상태 스키마 정의로 다양한 대화 시나리오에 맞춘 챗봇 커스터마이징이 가능합니다.
랭그래프, AI 에이전트 구현에 관심이 생겼나요?
->지금바로 랭그래프 스터디에 관심이 있다면 신청해주세요!
지피터스 13기 스터디 신청링크
