Chroma와 Ollama로 코드 분석, Redis로 대화 기록 저장

from langchain_text_splitters import Language
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OllamaEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders.generic import GenericLoader
from langchain_community.document_loaders.parsers import LanguageParser


loader = GenericLoader.from_filesystem(
    path="/Users/ekwak/libft/libft",
    glob="**/*",
    suffixes=[".c", ".h"],
    parser=LanguageParser(language=Language.CPP, parser_threshold=500),
)
documents = loader.load()

c_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.CPP, chunk_size=500, chunk_overlap=50
)
texts = c_splitter.split_documents(documents)

db = Chroma.from_documents(texts, OllamaEmbeddings(model="nomic-embed-text"), persist_directory="./chroma_db");

from dotenv import load_dotenv
import os

load_dotenv()

# LANGCHAIN_TRACING_V2 환경 변수를 "true"로 설정합니다.
os.environ["LANGCHAIN_TRACING_V2"] = "true"
# LANGCHAIN_PROJECT 설정
os.environ["LANGCHAIN_PROJECT"] = "RunnableWithMessageHistory"


from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OllamaEmbeddings

db = Chroma(persist_directory="./chroma_db", embedding_function=OllamaEmbeddings(model="nomic-embed-text"))

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_community.chat_models import ChatOllama
from langchain_community.chat_message_histories import RedisChatMessageHistory

llm = ChatOllama(model_name="llama2:70b")

contextualize_q_system_prompt = """Given a chat history and the latest user question \
which might reference context in the chat history, formulate a standalone question \
which can be understood without the chat history. Do NOT answer the question, \
just reformulate it if needed and otherwise return it as is."""
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", contextualize_q_system_prompt),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)

retriever = db.as_retriever(
    search_type="mmr",  # Also test "similarity"
    search_kwargs={"k": 8},
)
history_aware_retriever = create_history_aware_retriever(
    llm, retriever, prompt
)

qa_system_prompt = """You are an assistant for question-answering tasks. \
Use the following pieces of retrieved context to answer the question. \
If you don't know the answer, just say that you don't know. \

{context}"""
qa_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", qa_system_prompt),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)
question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)

rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

REDIS_URL = "redis://localhost:6379/0"

def get_message_history(session_id: str) -> RedisChatMessageHistory:
    # 세션 ID를 기반으로 RedisChatMessageHistory 객체를 반환합니다.
    return RedisChatMessageHistory(session_id, url=REDIS_URL)


with_message_history = RunnableWithMessageHistory(
    rag_chain,  # 실행 가능한 객체
    get_message_history,  # 메시지 기록을 가져오는 함수
    input_messages_key="input",  # 입력 메시지의 키
    history_messages_key="history",  # 기록 메시지의 키
    output_messages_key="answer",
)


while(True):
    text = input("What can i help you?")
    if text == "/bye":
        break
    result = with_message_history.invoke(
        {"input": text},
        config={
            "configurable": {"session_id": "1"}
        },  # constructs a key "abc123" in `store`.
    )["answer"]
    print(result)

docker run -d -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

from dotenv import load_dotenv
import os

load_dotenv()

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "RunnableWithMessageHistory"

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OllamaEmbeddings

db = Chroma(persist_directory="./chroma_db", embedding_function=OllamaEmbeddings(model="nomic-embed-text"))

contextualize_q_system_prompt = """Given a chat history and the latest user question \
which might reference context in the chat history, formulate a standalone question \
which can be understood without the chat history. Do NOT answer the question, \
just reformulate it if needed and otherwise return it as is."""
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", contextualize_q_system_prompt),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)

"채팅 기록과 채팅 기록의 맥락을 참조할 수 있는 최신 사용자 질문이 주어지면, 채팅 기록 없이도 이해할 수 있는 독립형 질문을 만들어 보세요. 질문에 답변하지 말고 필요한 경우 재구성하고 그렇지 않으면 그대로 반환합니다."

qa_system_prompt = """You are an assistant for question-answering tasks. \
Use the following pieces of retrieved context to answer the question. \
If you don't know the answer, just say that you don't know. \

{context}"""
qa_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", qa_system_prompt),
        MessagesPlaceholder("history"),
        ("human", "{input}"),
    ]
)

"귀하는 질문 답변 작업의 보조자입니다. 검색된 다음 문맥을 사용하여 질문에 답하세요. 답을 모른다면 모른다고 말하세요. "

rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

REDIS_URL = "redis://localhost:6379/0"

def get_message_history(session_id: str) -> RedisChatMessageHistory:
    # 세션 ID를 기반으로 RedisChatMessageHistory 객체를 반환합니다.
    return RedisChatMessageHistory(session_id, url=REDIS_URL)


with_message_history = RunnableWithMessageHistory(
    rag_chain,  # 실행 가능한 객체
    get_message_history,  # 메시지 기록을 가져오는 함수
    input_messages_key="input",  # 입력 메시지의 키
    history_messages_key="history",  # 기록 메시지의 키
    output_messages_key="answer",
)