랭체인 과제 1 중 실습 4번의 “체인에 메모리 추가하기”를 다음의 두 가지 방식으로 진행해 보았습니다.
In memory
Persistent storage : Redis 활용
실습 주요 사항 : 메모리 클래스 사용, 대화의 맥락 유지
**[작용 예제]**
> python add_memory_to_chain.py
> hi im bob
> Hello Bob! How can I assist you today?
> what's my name
> Your name is Bob.(사용자가 ‘Hi, I’m Bob”이라고 입력했을 때, 이 정보를 메모리에 저장하고, 다음 대화에서 “What’s my name ?”이라는 질문에 “Your name is Bob.”이라고 응답. )
[In memory 방식]
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai.chat_models import ChatOpenAI
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.output_parsers import StrOutputParser
model = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages(
[
("system", "You're an assistant."),
MessagesPlaceholder(variable_name="history"),
("human", "{input}"),
]
)
runnable = prompt | model | StrOutputParser()
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
with_message_history = RunnableWithMessageHistory(
runnable,
get_session_history,
input_messages_key="input",
history_messages_key="history",
)
print('--------------------------------------------------')
print(' Assistant Chatbot v.1 ')
print(' If you want to end the conversation, type "bye". ')
print('--------------------------------------------------')
while True :
input_text = input("User : ")
if input_text == "bye":
print("The conversation is over!!")
break
result = with_message_history.invoke(
{"input": input_text},
config={"configurable": {"session_id": "test123"}},
)
print("A I : ", result)
print('---------------------------------')
# print("History : ", store)실행 결과
(프로그램 종료 이후 chat history reset)
[Persistent Storage : Redis(고성능 인-메모리 데이터 저장소)]
코드를 실행하기 전에 다음의 과정을 거쳐 Redis server를 구동 시킵니다.
Redis 라이브러리 설치 : pip install redis
docker 설치 / 구동
Redis server 구동 : docker run -d -p 6379:6379 -p 8001:8001 redis/redis-stack:latest
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai.chat_models import ChatOpenAI
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_message_histories import RedisChatMessageHistory
import redis
# ---------------------------------------------------------------------------------------
# redis server 구동 : docker run -d -p 6379:6379 -p 8001:8001 redis/redis-stack:latest
# redis server 중단 : docker stop [server name]
# redis server 구동 확인 : docker ps
# redis container 접속 : docker exec -it [컨텐이너 이름] sh
# redis cli 실행 : redis-cli
# key 조회 : keys *
# key data type 확인 : type [키 이름]
# list data type 조회 : lrange [리스트 키 이름] 0 -1 // 0 : start index, -1 : end index
# key data 삭제 : del [키 이름]
# ---------------------------------------------------------------------------------------
REDIS_URL = "redis://localhost:6379/0"
model = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages(
[
("system", "You're an assistant."),
MessagesPlaceholder(variable_name="history"),
("human", "{input}"),
]
)
runnable = prompt | model | StrOutputParser()
id_session = "test123"
def get_message_history(session_id: str) -> RedisChatMessageHistory:
return RedisChatMessageHistory(session_id, url=REDIS_URL)
with_message_history = RunnableWithMessageHistory(
runnable,
get_message_history,
input_messages_key="input",
history_messages_key="history",
)
print('--------------------------------------------------')
print(' Assistant Chatbot v.1 ')
print(' If you want to end the conversation, type "bye". ')
print('--------------------------------------------------')
while True :
input_text = input("User : ")
if input_text == "bye":
print("The conversation is over!!")
break
result = with_message_history.invoke(
{"input": input_text},
config={"configurable": {"session_id": id_session}},
)
print("A I : ", result)
print('---------------------------------')
print("History : ")
print(RedisChatMessageHistory("test123", url=REDIS_URL))
# 저장된 채팅 기록 삭제
def delete_chat_history(session_id: str):
# redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)
redis_client = redis.from_url(REDIS_URL)
# Deleting the key associated with session_id
_del = redis_client.delete(session_id)
if _del == 1:
print("session delete success!!")
else:
print("session delete failure!!")
delete_chat_history("message_store:" + id_session)※ 채팅 종료 이후 chat_history 삭제하는 부분 추가.
실행 결과
Reference : Add message history (memory)
#10기랭체인