7. Langchain 캐싱(API 호출 비용 감소)

Langchain은 LLM을 위한 선택적 캐싱 레이어를 제공한다. 

 - 동일한 완료를 여러 번 요청하는 경우 LLM 공급자에 대한 API 호출 횟수를 줄여 비용을 절감할 수 있음

 - LLM 제공업체에 대한 API 호출 횟수를 줄여 애플리케이션의 속도를 높일 수 있다.

 

InMemory Cache

InMemory Cache를 사용해서 동일 질문에 대한 답변을 저장하고, 캐시에 저장된 답변을 반환하는 대신에

휘발성 특징을 가진  InMemory Cache는 프로그램을 종료 즉시  저장된 답변은 사라지게 된다.

 

아래의 사진에서 InMemory Cache를 사용했을 때 비교할 것은 처음 답변과, 두 번째 동일 답변에 대한 출력 시간이다.

1) 처음 답변 - Wall time : 2.74s

2) 두 번째 답변 - Wall time : 3.36ms

 

둘의 차이는 1/1000 차이다. 어마무시한 차이인데, 이전에 내가 RAG를 사용하기 위한 프로젝트에서 Cache를 사용했더라면 API 호출 시간과 비용 절감에 많은 이점을 얻었을텐데 해당 방법을 선택하지 않았던 내 자신이 후회스럽다.

 

SQLite Cache

위 방법의 InMemory 방식 말고 SQLite Cache를 사용하면 프로그램을 닫아도 Cache데이터가 사라지지 않아서 DB 파일에 저장하는 방식으로 유용하게 쓸 수 있다.

Cache 파일이 없다면 Cache 디렉토리를 만들어 답변 저장시키는 코드를 넣어서 사용하면 된다.

물론 목적에 따라 다르겠지만, 프로그램 사용 중에만 캐싱을 시키려면 InMemory Cache를 선택하고

비휘발성으로 DB에 저장하는 방식으로 사용한다면 SQLite Cache를 사용하면 된다.

 

음... 사용 목적에 따라 다르지만 DB 관리를 잘한다면 서비스업 챗봇에 사용하면 굉장히 유용하고 리소스 비용을 아낄 수 있으니 서비스업에서는 부분 사용하면 좋을 것 같다 생각한다.

 

그리고 Batch 단위 실행으로 여러번 답변을 출력할 수 있는 방법이 있다.

from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(
    # 사용할 모델을 "gemini-pro"로 지정합니다.
    model="gemini-1.5-pro-latest",
)

results = llm.batch(
    [
        "대한민국의 수도는?",
        "대한민국의 주요 관광지 5곳을 나열하세요",
    ]
)

for res in results:
    # 각 결과의 내용을 출력합니다.
    print(res.content)
    

llm.invoke[0]

# 출력 : 서울

 

ConversationBufferMemory

대화 내용을 기억하게 해주는 메모리인데, 메시지를 저장해서 변수로 메시지를 추출하게 하는 것이다.

출력 형식은 1. String / 2.객체 형식으로 받아오는 방법이 있다.

때에 따라 다르게 사용하겠지만, 객체 형식으로 받아와야 한다면 두 번째 코드에서 확인하면 좋을 것 같다.

 

둘의 Class 이름 뒤에 파라미터 값인 `return_messages=True`를 넣냐 안 넣냐에 따라 출력 형식이 다르다.

 

1. String 형식

 

from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()
memory

memory.save_context(
    inputs={
        "human": "안녕하세요, 비대면으로 은행 계좌를 개설하고 싶습니다. 어떻게 시작해야 하나요?"
    },
    outputs={
        "ai": "안녕하세요! 계좌 개설을 원하신다니 기쁩니다. 먼저, 본인 인증을 위해 신분증을 준비해 주시겠어요?"
    },
)

# 'history' 키에 저장된 대화 기록을 확인합니다.
print(memory.load_memory_variables({})["history"])

# 출력
Human: 안녕하세요, 비대면으로 은행 계좌를 개설하고 싶습니다. 어떻게 시작해야 하나요?
AI: 안녕하세요! 계좌 개설을 원하신다니 기쁩니다. 먼저, 본인 인증을 위해 신분증을 준비해 주시겠어요?

 

2. 객체 형식

memory = ConversationBufferMemory(return_messages=True)

memory.save_context(
    inputs={
        "human": "안녕하세요, 비대면으로 은행 계좌를 개설하고 싶습니다. 어떻게 시작해야 하나요?"
    },
    outputs={
        "ai": "안녕하세요! 계좌 개설을 원하신다니 기쁩니다. 먼저, 본인 인증을 위해 신분증을 준비해 주시겠어요?"
    },
)

# 출력
[HumanMessage(content='안녕하세요, 비대면으로 은행 계좌를 개설하고 싶습니다. 어떻게 시작해야 하나요?', additional_kwargs={}, response_metadata={}),
 AIMessage(content='안녕하세요! 계좌 개설을 원하신다니 기쁩니다. 먼저, 본인 인증을 위해 신분증을 준비해 주시겠어요?', additional_kwargs={}, response_metadata={}),

 

추후, Lnagchain.chains import ConversationChain은 사용 불가할 예정이다. Langchain이 1.0으로 업데이트 되면서부터는 공식적으로 사용 못하게 된다. Conversation으로 Chain생성을 하기 보다는 RunnableWithMessageHistory를 사용해서 Chain형성을 하면 될 것 같다.

 

현재 25.03.14일 기준 0.3v이 최신버전이니, 1.0버전까지는 여유가 많은 것 같다.

 

ConversationChain 사용은 아래 코드처럼 memory로 선언한 변수에 ConvsationBufferMemory()

from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain

# LLM 모델을 생성합니다.
llm = ChatOpenAI(temperature=0, model_name="gpt-4o")

# ConversationChain을 생성합니다.
conversation = ConversationChain(
    # ConversationBufferMemory를 사용합니다.
    llm=llm,
    memory=ConversationBufferMemory(),
)

 

이렇게 하면 이전에 GPT와 대화했던 내용들을 ConversationBufferMemory를 통해 계속 가지고 있어서 손쉬운 문장들을 만들어낼 수 있다.

 

원래는 모든 대화 내용은 save_context라는 메서드로 전부 새로운 내용과 답변을 수작업해서 저장 해야 하는데, 그런 거 없이 ConversationChain을 사용하면 이전 대화내용 기록 Chain을 만들 수 있다. 항상 생각했는데, 엄청 나게 많은 수작업을 어떻게 할까 생각했고 변수로 로컬에 저장해서 대화 내용을 불러오는 것도 생각했지만, 가장 좋은 건 모듈로 이렇게 저장해서 대화 내용을 축적시키는 방법이 가장 깔끔하고 정신 건강에도 좋은 것 같다. 로컬환경으로 돌린다면 다르겠지만.

 

하지만, 장점이 있다면 단점도 있기 마련인데 BufferMemory는 가용 가능한 만큼 대화 내용을 저장하지만 GPT-4o의 입력토큰인 12만 토큰을 넘어선다면 대화 내용을 불러올 때 오류가 발생할 수 있다.

그래서 이 메모리 용량 관리가 필요한데 그건! 바로 아래에 있다.

 

ConversationBufferWindowMemory

여기서 봐야할 건, ConversationBufferWindowMemory(k=2, return_messages=True)에서 k는 대화 내용의 개수를 뜻하는데 k의 개수를 조정하면 저장되는 내용 관리가 용이할 것이고. WindowMemory는 맨 마지막 대화 내용인 Memory.save_context(input={"human":}, outputs={"ai":}) 대화 뭉텅이 개수를 뜻한다.

처음에 나눴던 대화 내용이 아닌, 마지막의 두 개 내용들이다.

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(k=2, return_messages=True)

memory.save_context(
    inputs={
        "human": "안녕하세요, 비대면으로 은행 계좌를 개설하고 싶습니다. 어떻게 시작해야 하나요?"
    },
    outputs={
        "ai": "안녕하세요! 계좌 개설을 원하신다니 기쁩니다. 먼저, 본인 인증을 위해 신분증을 준비해 주시겠어요?"
    },
)
memory.save_context(
    inputs={"human": "네, 신분증을 준비했습니다. 이제 무엇을 해야 하나요?"},
    outputs={
        "ai": "감사합니다. 신분증 앞뒤를 명확하게 촬영하여 업로드해 주세요. 이후 본인 인증 절차를 진행하겠습니다."
    },
)
memory.save_context(
    inputs={"human": "정보를 모두 입력했습니다. 다음 단계는 무엇인가요?"},
    outputs={
        "ai": "입력해 주신 정보를 확인했습니다. 계좌 개설 절차가 거의 끝났습니다. 마지막으로 이용 약관에 동의해 주시고, 계좌 개설을 최종 확인해 주세요."
    },
)
memory.save_context(
    inputs={"human": "모든 절차를 완료했습니다. 계좌가 개설된 건가요?"},
    outputs={
        "ai": "네, 계좌 개설이 완료되었습니다. 고객님의 계좌 번호와 관련 정보는 등록하신 이메일로 발송되었습니다. 추가적인 도움이 필요하시면 언제든지 문의해 주세요. 감사합니다!"
    },
)

 

음... WindowMemory를 활용하는 방법은 가장 최신으로 필요한 내용만을 필두로 저장을 할 때 가장 많이 사용되는 건데

즉, 쓸 데 없는 내용들이 저장되는 것보다 더 신빙성 있는 정보나 최신 정보를 갖는 대화 내용이 있어야 대화의 흐름이 정확하게 이어져서 생성 AI를 잘 사용할 수 있을 것으로 보이므로 WindowMemory는 요긴하게 사용할 수 있을 것 같다.

서비스 론칭할 때 어느정도의 Window Size인 k가 적합할지 테스트 해보면서 사용하면 좋을 것 같다.

 

ConversationTokenBufferMemory

위에서 써놓은 ConversationBufferWindowMemory과 다르게 TokenBuffer는 Chunking된 문자 즉, Token 단위로 대화 내용을 제한할 수 있다. Window는 memory.save_context의 문장별로 대화 내용을 제한 했지만

Token은 최하단 마지막 대화 내용을 기준으로 토큰 값을 제한해서 대화 내용을 저장하는 것이다.

from langchain.memory import ConversationTokenBufferMemory
from langchain_openai import ChatOpenAI


# LLM 모델 생성
llm = ChatOpenAI(model_name="gpt-4o")

# 메모리 설정
memory = ConversationTokenBufferMemory(
    llm=llm, max_token_limit=300, return_messages=True  # 최대 토큰 길이를 50개로 제한
)

memory.save_context(
    inputs={"human": "연결은 어떻게 하나요?"},
    outputs={
        "ai": "매뉴얼의 5페이지를 참조해 주세요. 케이블 연결에 관한 상세한 지침이 있습니다. 이 과정에서 어려움이 있으시면 추가적으로 도와드리겠습니다."
    },
)
memory.save_context(
    inputs={"human": "설치가 완료되면 어떻게 해야 하나요?"},
    outputs={
        "ai": "설치가 완료되면, 전원을 켜고 초기 구동 테스트를 진행해 주시기 바랍니다. 테스트 절차는 매뉴얼의 10페이지에 설명되어 있습니다. 만약 기계에 이상이 있거나 추가적인 지원이 필요하시면 언제든지 연락 주시기 바랍니다."
    },
)
memory.save_context(
    inputs={"human": "감사합니다, 도움이 많이 되었어요!"},
    outputs={
        "ai": "언제든지 도와드릴 준비가 되어 있습니다. 추가적인 질문이나 지원이 필요하시면 언제든지 문의해 주세요. 좋은 하루 되세요!"
    },
)

# 대화내용을 확인합니다.
memory.load_memory_variables({})["history"]

# 출력
[HumanMessage(content='연결은 어떻게 하나요?', additional_kwargs={}, response_metadata={}),
 AIMessage(content='매뉴얼의 5페이지를 참조해 주세요. 케이블 연결에 관한 상세한 지침이 있습니다. 이 과정에서 어려움이 있으시면 추가적으로 도와드리겠습니다.', additional_kwargs={}, response_metadata={}),
 HumanMessage(content='설치가 완료되면 어떻게 해야 하나요?', additional_kwargs={}, response_metadata={}),
 AIMessage(content='설치가 완료되면, 전원을 켜고 초기 구동 테스트를 진행해 주시기 바랍니다. 테스트 절차는 매뉴얼의 10페이지에 설명되어 있습니다. 만약 기계에 이상이 있거나 추가적인 지원이 필요하시면 언제든지 연락 주시기 바랍니다.', additional_kwargs={}, response_metadata={}),
 HumanMessage(content='감사합니다, 도움이 많이 되었어요!', additional_kwargs={}, response_metadata={}),
 AIMessage(content='언제든지 도와드릴 준비가 되어 있습니다. 추가적인 질문이나 지원이 필요하시면 언제든지 문의해 주세요. 좋은 하루 되세요!', additional_kwargs={}, response_metadata={})]

 

ConversationSummaryBufferMemory

SummaryBufferMemory는 `max_token_limit = 200`으로 지정해서 원본을 200토큰까지만 그대로 가져오고 난 후,

이후의 대화를 요약해서 출력해달라는 변수 값이다.

 

"이전의 대화 내용을 요약해줘"라고 input에 넣으면 최근 대화 내용 중 원본 내용의 n개 토큰 만큼 유지를 시켜주기 때문에 Entity Memory보다 좀 더 요약을 잘해주는 것을 볼 수 있다.

from langchain_openai import ChatOpenAI
from langchain.memory import ConversationSummaryBufferMemory

llm = ChatOpenAI()

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=200,  # 요약의 기준이 되는 토큰 길이를 설정합니다.
    return_messages=True,
)

 

VectorstoreRectroeverMemory 

벡터 스토어 메모리는 저장하고 호출 될 때마다 가장 눈에 띄는 상위 K개의 문서를 쿼리하는 클래스 메모리이다.

대화내용의 순서를 명식적으로 추적하지 않는다는 점에서 다른 메모리 클래스와 조금은 다를 수 있다.

 

OpenAIEmbedding size는 1536개 차원을 기준으로 벡터 값을 가지기 때문에 사이즈 값은 `1536`으로 지정한다.

import faiss
from langchain_openai import OpenAIEmbeddings
from langchain.docstore import InMemoryDocstore
from langchain.vectorstores import FAISS


# 임베딩 모델을 정의합니다.
embeddings_model = OpenAIEmbeddings()

# Vector Store 를 초기화 합니다.
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings_model, index, InMemoryDocstore({}), {})

 

검색기인 retriever의 함수 (serach_kwargs = {"k":1})에서 k는 과거의 대화 턴 갯수를 의미하고 1은 1개를 의미한다.

그럼 k:1이 된다면 이전 대화에서 상위 대화 1개를 리턴한다.

 

하지만 값을 넉넉하게 받기 위해서는 내가 만든 대화 내용에 따라 값은 달라지므로 적당한 값은 k:5 형식으로 하는 게 좋다.

VectorStoreretriever memory는 좋은 게 과거 대화 내용이 뒤죽박죽해서 찾기 엄청 힘들 때, 관련 Key값을 input으로 넣어 그 유사도를 검색해서 가져오기 때문에 이미 지난 과거 대화 내용에서 검색해서 찾기가 굉장히 유용하다.

 

ps : 음.... 생각을 해보면 나는 대화를 뒤죽박죽 형식으로 하니깐 VectorStore에 저장해서 유사도 검색으로 찾는 이 메모리를 사용하고싶다.

from langchain.memory import VectorStoreRetrieverMemory

# 벡터 조회가 여전히 의미적으로 관련성 있는 정보를 반환한다는 것을 보여주기 위해서입니다.
retriever = vectorstore.as_retriever(search_kwargs={"k": 1})
memory = VectorStoreRetrieverMemory(retriever=retriever)

# 임의의 대화를 저장합니다.
memory.save_context(
    inputs={
        "human": "안녕하세요, 오늘 면접에 참석해주셔서 감사합니다. 자기소개 부탁드립니다."
    },
    outputs={
        "ai": "안녕하세요. 저는 컴퓨터 과학을 전공한 신입 개발자입니다. 대학에서는 주로 자바와 파이썬을 사용했으며, 최근에는 웹 개발 프로젝트에 참여하여 실제 사용자를 위한 서비스를 개발하는 경험을 했습니다."
    },
)
memory.save_context(
    inputs={"human": "프로젝트에서 어떤 역할을 맡았나요?"},
    outputs={
        "ai": "제가 맡은 역할은 백엔드 개발자였습니다. 사용자 데이터 처리와 서버 로직 개발을 담당했으며, RESTful API를 구현하여 프론트엔드와의 통신을 담당했습니다. 또한, 데이터베이스 설계에도 참여했습니다."
    },
)
memory.save_context(
    inputs={
        "human": "팀 프로젝트에서 어려움을 겪었던 경험이 있다면 어떻게 해결했나요?"
    },
    outputs={
        "ai": "프로젝트 초기에 의사소통 문제로 몇 가지 어려움이 있었습니다. 이를 해결하기 위해 저희 팀은 정기적인 미팅을 갖고 각자의 진행 상황을 공유했습니다. 또한, 문제가 발생했을 때는 적극적으로 의견을 나누고, 합리적인 해결책을 찾기 위해 노력했습니다."
    },
)
memory.save_context(
    inputs={"human": "개발자로서 자신의 강점은 무엇이라고 생각하나요?"},
    outputs={
        "ai": "제 강점은 빠른 학습 능력과 문제 해결 능력입니다. 새로운 기술이나 도구를 빠르게 습득할 수 있으며, 복잡한 문제에 직면했을 때 창의적인 해결책을 제시할 수 있습니다. 또한, 팀워크를 중시하며 동료들과 협력하는 것을 중요하게 생각합니다."
    },
)

 

 

SQLAlchemy

이전 글에서 다뤘던 InMemory(휘발성)과 SQLiteMemory(비휘발성)에 대해서 이야기 했는데

SQLAlchemy라는 코드를 어떻게 사용하는지 적어보려고한다.

 

Structures Query Languqge(SQL)은 프로그래밍에 사용되는 도메인 특화 언어로, 관계형 데이터베이스 관리 시스템

(RDBMS)에서 데이터를 관리하거나 관계형 데이터 스트림 관리 시스템에서 스트림 처리를 위해 설계되었다.

 

1. session_id: 사용자 이름, 이메일, 채팅 ID 등과 같은 세션의 고유 식별자

2. connection : 데이터베이스 연결을 지정하는 문자열이고 문자열은 SQLAlchemy의 Create_engine 함수에 전달됨

from langchain_community.chat_message_histories import SQLChatMessageHistory

# SQLChatMessageHistory 객체를 생성하고 세션 ID와 데이터베이스 연결 파일을 설정
chat_message_history = SQLChatMessageHistory(
    session_id="sql_history2", connection="sqlite:///sqlite.db"
)

# 사용자 메시지를 추가합니다.
chat_message_history.add_user_message(
    "안녕? 만나서 반가워. 내 이름은 네티야. 나는 랭체인 개발자야~~~"
)
# AI 메시지를 추가합니다.
chat_message_history.add_ai_message("안녕 네티, 만나서 반가워. 나도 잘 부탁해!")

# 채팅 메시지 기록의 메시지들
chat_message_history.messages

# 출력

[HumanMessage(content='안녕? 만나서 반가워. 내 이름은 테디야. 나는 랭체인 개발자야. 앞으로 잘 부탁해!', additional_kwargs={}, response_metadata={}),
 AIMessage(content='안녕 테디, 만나서 반가워. 나도 잘 부탁해!', additional_kwargs={}, response_metadata={}),
 HumanMessage(content='안녕? 만나서 반가워. 내 이름은 테디야. 나는 랭체인 개발자야. 앞으로 잘 부탁해!', additional_kwargs={}, response_metadata={}),
 AIMessage(content='안녕 테디, 만나서 반가워. 나도 잘 부탁해!', additional_kwargs={}, response_metadata={})]

 

 

여기서 session ID인 사용자 이름을 여러개로 바꿔서 sqlite.db에 저장하면 column 값에서 ID(사용자 이름)을 기준으로

유저 대화 내용을 구분하는 방법이 있다. 별도의 sqlite.db라는 파일 이름으로 seesion_id와 user_message등을 저장하는 방식이니 SQLite_db viewer를 설치해서 보면 확인할 수 있다.

 

다음은 Langchain의 Documents Loader에 대해서 작성하려고한다.