자습서 2: Cortex Search를 사용하여 간단한 채팅 애플리케이션 개발¶

소개¶

이 자습서에서는 Cortex Search와 COMPLETE (SNOWFLAKE.CORTEX) 함수를 사용하여 Snowflake에서 RAG(Retrieval-Augmented Generation) 챗봇을 설정하는 방법을 설명합니다.

1단계: 설정¶

CREATE DATABASE IF NOT EXISTS cortex_search_tutorial_db;

CREATE OR REPLACE WAREHOUSE cortex_search_tutorial_wh WITH
    WAREHOUSE_SIZE='X-SMALL'
    AUTO_SUSPEND = 120
    AUTO_RESUME = TRUE
    INITIALLY_SUSPENDED=TRUE;

USE WAREHOUSE cortex_search_tutorial_wh;

2단계: Snowflake에 데이터 로드¶

먼저 Kaggle에서 다운로드한 파일을 저장할 스테이지를 생성합니다. 이 스테이지에는 장부 데이터 세트가 보관됩니다.

CREATE OR REPLACE STAGE books_data_stage
    DIRECTORY = (ENABLE = TRUE)
    ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE');

이제 데이터 세트를 업로드합니다. 데이터 세트는 Snowsight 에서 업로드하거나 SQL을 사용하여 업로드할 수 있습니다. Snowsight 에서 업로드하려면:

1. Snowsight 에 로그인합니다.

2. 왼쪽 탐색 메뉴에서 Data 를 선택합니다.

3. cortex_search_tutorial_db 데이터베이스를 선택합니다.

4. 스키마 public 을 선택합니다.

5. Stages 를 선택하고 books_data_stage 를 선택합니다.

6. 오른쪽 상단에서 + Files 버튼을 선택합니다.

7. UI에 파일을 끌어서 놓거나 Browse 를 선택하여 대화 창에서 파일을 선택합니다.

8. Upload 를 선택하여 BooksDatasetClean.csv 파일을 업로드합니다.

9. 파일 오른쪽에 있는 세 개의 점을 선택하고 Load into table 을 선택합니다.

10. 테이블 이름을 BOOKS_DATASET_RAW 로 지정하고 Next 를 선택합니다.

    

11. 데이터 로드 대화 상자의 왼쪽 패널에 있는 Header 메뉴에서 First line contains header 를 선택합니다.

12. 그런 다음 Load 를 선택합니다.

2단계: 청킹 UDF 만들기¶

검색 모델은 소규모 텍스트 청크에서 가장 잘 작동하므로 긴 문서를 Cortex Search에 제공하면 성능이 저하됩니다. 그런 다음 Python UDF를 생성하여 텍스트를 청크합니다. SQL 편집기로 돌아가서 다음을 실행합니다.

CREATE OR REPLACE FUNCTION cortex_search_tutorial_db.public.books_chunk(
    description string, title string, authors string, category string, publisher string
)
    returns table (chunk string, title string, authors string, category string, publisher string)
    language python
    runtime_version = '3.9'
    handler = 'text_chunker'
    packages = ('snowflake-snowpark-python','langchain')
    as
$$
from langchain.text_splitter import RecursiveCharacterTextSplitter
import copy
from typing import Optional

class text_chunker:

    def process(self, description: Optional[str], title: str, authors: str, category: str, publisher: str):
        if description == None:
            description = "" # handle null values

        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size = 2000,
            chunk_overlap  = 300,
            length_function = len
        )
        chunks = text_splitter.split_text(description)
        for chunk in chunks:
            yield (title + "\n" + authors + "\n" + chunk, title, authors, category, publisher) # always chunk with title
$$;

4단계: 청크 테이블 만들기¶

기록에서 추출한 텍스트 청크를 저장할 테이블을 생성합니다. 청크에 제목과 화자를 포함하여 컨텍스트를 제공합니다.

CREATE TABLE cortex_search_tutorial_db.public.book_description_chunks AS (
    SELECT
        books.*,
        t.CHUNK as CHUNK
    FROM cortex_search_tutorial_db.public.books_dataset_raw books,
        TABLE(cortex_search_tutorial_db.public.books_chunk(books.description, books.title, books.authors, books.category, books.publisher)) t
);

테이블 내용을 확인합니다.

SELECT chunk, * FROM book_description_chunks LIMIT 10;

5단계: Cortex Search Service 만들기¶

book_description_chunks 의 청크를 검색할 수 있도록 테이블에 Cortex Search Service를 생성합니다.

CREATE CORTEX SEARCH SERVICE cortex_search_tutorial_db.public.books_dataset_service
    ON CHUNK
    WAREHOUSE = cortex_search_tutorial_wh
    TARGET_LAG = '1 hour'
    AS (
        SELECT *
        FROM cortex_search_tutorial_db.public.book_description_chunks
    );

6단계: Streamlit 앱 생성¶

Python SDK로 서비스를 쿼리(snowflake Python 패키지 사용)할 수 있습니다. 이 자습서에서는 Streamlit in Snowflake 애플리케이션에서 Python SDK를 사용하는 방법을 보여줍니다.

먼저 서비스 생성 단계에서 전역 Snowsight UI 역할이 서비스를 생성하는 데 사용된 역할과 동일한지 확인합니다.

1. Snowsight 에 로그인합니다.

2. 왼쪽 탐색 메뉴에서 Projects » Streamlit 를 선택합니다.

3. + Streamlit App 를 선택합니다.

4. 중요: 앱 위치에 대한 cortex_search_tutorial_db 데이터베이스와 public 스키마를 선택합니다.

5. Streamlit in Snowflake 편집기의 왼쪽 창에서 Packages 를 선택하고 snowflake (버전 >= 0.8.0)를 추가하여 애플리케이션에 패키지를 설치합니다.

6. 예제 애플리케이션 코드를 다음 Streamlit 앱으로 바꿉니다.

   import streamlit as st
   from snowflake.core import Root # requires snowflake>=0.8.0
   from snowflake.snowpark.context import get_active_session
   
   MODELS = [
       "mistral-large",
       "snowflake-arctic",
       "llama3-70b",
       "llama3-8b",
   ]
   
   def init_messages():
       """
       Initialize the session state for chat messages. If the session state indicates that the
       conversation should be cleared or if the "messages" key is not in the session state,
       initialize it as an empty list.
       """
       if st.session_state.clear_conversation or "messages" not in st.session_state:
           st.session_state.messages = []
   
   def init_service_metadata():
       """
       Initialize the session state for cortex search service metadata. Query the available
       cortex search services from the Snowflake session and store their names and search
       columns in the session state.
       """
       if "service_metadata" not in st.session_state:
           services = session.sql("SHOW CORTEX SEARCH SERVICES;").collect()
           service_metadata = []
           if services:
               for s in services:
                   svc_name = s["name"]
                   svc_search_col = session.sql(
                       f"DESC CORTEX SEARCH SERVICE {svc_name};"
                   ).collect()[0]["search_column"]
                   service_metadata.append(
                       {"name": svc_name, "search_column": svc_search_col}
                   )
   
           st.session_state.service_metadata = service_metadata
   
   def init_config_options():
       """
       Initialize the configuration options in the Streamlit sidebar. Allow the user to select
       a cortex search service, clear the conversation, toggle debug mode, and toggle the use of
       chat history. Also provide advanced options to select a model, the number of context chunks,
       and the number of chat messages to use in the chat history.
       """
       st.sidebar.selectbox(
           "Select cortex search service:",
           [s["name"] for s in st.session_state.service_metadata],
           key="selected_cortex_search_service",
       )
   
       st.sidebar.button("Clear conversation", key="clear_conversation")
       st.sidebar.toggle("Debug", key="debug", value=False)
       st.sidebar.toggle("Use chat history", key="use_chat_history", value=True)
   
       with st.sidebar.expander("Advanced options"):
           st.selectbox("Select model:", MODELS, key="model_name")
           st.number_input(
               "Select number of context chunks",
               value=5,
               key="num_retrieved_chunks",
               min_value=1,
               max_value=10,
           )
           st.number_input(
               "Select number of messages to use in chat history",
               value=5,
               key="num_chat_messages",
               min_value=1,
               max_value=10,
           )
   
       st.sidebar.expander("Session State").write(st.session_state)
   
   def query_cortex_search_service(query):
       """
       Query the selected cortex search service with the given query and retrieve context documents.
       Display the retrieved context documents in the sidebar if debug mode is enabled. Return the
       context documents as a string.
   
       Args:
           query (str): The query to search the cortex search service with.
   
       Returns:
           str: The concatenated string of context documents.
       """
       db, schema = session.get_current_database(), session.get_current_schema()
   
       cortex_search_service = (
           root.databases[db]
           .schemas[schema]
           .cortex_search_services[st.session_state.selected_cortex_search_service]
       )
   
       context_documents = cortex_search_service.search(
           query, columns=[], limit=st.session_state.num_retrieved_chunks
       )
       results = context_documents.results
   
       service_metadata = st.session_state.service_metadata
       search_col = [s["search_column"] for s in service_metadata
                       if s["name"] == st.session_state.selected_cortex_search_service][0]
   
       context_str = ""
       for i, r in enumerate(results):
           context_str += f"Context document {i+1}: {r[search_col]} \n" + "\n"
   
       if st.session_state.debug:
           st.sidebar.text_area("Context documents", context_str, height=500)
   
       return context_str
   
   def get_chat_history():
       """
       Retrieve the chat history from the session state limited to the number of messages specified
       by the user in the sidebar options.
   
       Returns:
           list: The list of chat messages from the session state.
       """
       start_index = max(
           0, len(st.session_state.messages) - st.session_state.num_chat_messages
       )
       return st.session_state.messages[start_index : len(st.session_state.messages) - 1]
   
   def complete(model, prompt):
       """
       Generate a completion for the given prompt using the specified model.
   
       Args:
           model (str): The name of the model to use for completion.
           prompt (str): The prompt to generate a completion for.
   
       Returns:
           str: The generated completion.
       """
       return session.sql("SELECT snowflake.cortex.complete(?,?)", (model, prompt)).collect()[0][0]
   
   def make_chat_history_summary(chat_history, question):
       """
       Generate a summary of the chat history combined with the current question to extend the query
       context. Use the language model to generate this summary.
   
       Args:
           chat_history (str): The chat history to include in the summary.
           question (str): The current user question to extend with the chat history.
   
       Returns:
           str: The generated summary of the chat history and question.
       """
       prompt = f"""
           [INST]
           Based on the chat history below and the question, generate a query that extend the question
           with the chat history provided. The query should be in natural language.
           Answer with only the query. Do not add any explanation.
   
           <chat_history>
           {chat_history}
           </chat_history>
           <question>
           {question}
           </question>
           [/INST]
       """
   
       summary = complete(st.session_state.model_name, prompt)
   
       if st.session_state.debug:
           st.sidebar.text_area(
               "Chat history summary", summary.replace("$", "\$"), height=150
           )
   
       return summary
   
   def create_prompt(user_question):
       """
       Create a prompt for the language model by combining the user question with context retrieved
       from the cortex search service and chat history (if enabled). Format the prompt according to
       the expected input format of the model.
   
       Args:
           user_question (str): The user's question to generate a prompt for.
   
       Returns:
           str: The generated prompt for the language model.
       """
       if st.session_state.use_chat_history:
           chat_history = get_chat_history()
           if chat_history != []:
               question_summary = make_chat_history_summary(chat_history, user_question)
               prompt_context = query_cortex_search_service(question_summary)
           else:
               prompt_context = query_cortex_search_service(user_question)
       else:
           prompt_context = query_cortex_search_service(user_question)
           chat_history = ""
   
       prompt = f"""
               [INST]
               You are a helpful AI chat assistant with RAG capabilities. When a user asks you a question,
               you will also be given context provided between <context> and </context> tags. Use that context
               with the user's chat history provided in the between <chat_history> and </chat_history> tags
               to provide a summary that addresses the user's question. Ensure the answer is coherent, concise,
               and directly relevant to the user's question.
   
               If the user asks a generic question which cannot be answered with the given context or chat_history,
               just say "I don't know the answer to that question.
   
               Don't saying things like "according to the provided context".
   
               <chat_history>
               {chat_history}
               </chat_history>
               <context>
               {prompt_context}
               </context>
               <question>
               {user_question}
               </question>
               [/INST]
               Answer:
           """
       return prompt
   
   def main():
       st.title(f":speech_balloon: Chatbot with Snowflake Cortex")
   
       init_service_metadata()
       init_config_options()
       init_messages()
   
       icons = {"assistant": "❄️", "user": "👤"}
   
       # Display chat messages from history on app rerun
       for message in st.session_state.messages:
           with st.chat_message(message["role"], avatar=icons[message["role"]]):
               st.markdown(message["content"])
   
       disable_chat = (
           "service_metadata" not in st.session_state
           or len(st.session_state.service_metadata) == 0
       )
       if question := st.chat_input("Ask a question...", disabled=disable_chat):
           # Add user message to chat history
           st.session_state.messages.append({"role": "user", "content": question})
           # Display user message in chat message container
           with st.chat_message("user", avatar=icons["user"]):
               st.markdown(question.replace("$", "\$"))
   
           # Display assistant response in chat message container
           with st.chat_message("assistant", avatar=icons["assistant"]):
               message_placeholder = st.empty()
               question = question.replace("'", "")
               with st.spinner("Thinking..."):
                   generated_response = complete(
                       st.session_state.model_name, create_prompt(question)
                   )
                   message_placeholder.markdown(generated_response)
   
           st.session_state.messages.append(
               {"role": "assistant", "content": generated_response}
           )
   
   if __name__ == "__main__":
       session = get_active_session()
       root = Root(session)
       main()
   

   Copy

7단계: 앱 사용해 보기¶

텍스트 상자에 질문을 입력하여 새 앱을 사용해 봅니다. 시도해 볼 수 있는 몇 가지 샘플 쿼리는 다음과 같습니다.

• I like Harry Potter. Can you recommend more books I will like?

• Can you recommend me books on Greek Mythology?

7단계: 정리¶

DROP DATABASE IF EXISTS cortex_search_tutorial_db;
DROP WAREHOUSE IF EXISTS cortex_search_tutorial_wh;

다음 단계¶

축하합니다! Snowflake에서 텍스트 데이터를 기반으로 간단한 검색 앱을 성공적으로 생성했습니다. 자습서 3 으로 넘어가서 일련의 PDF 파일에서 Cortex Search를 사용하여 AI 챗봇을 구축하는 방법을 살펴볼 수 있습니다.