教程 2：使用 Cortex Search 构建简单的聊天应用程序¶

简介¶

本教程介绍如何使用 Cortex Search 和:doc:/sql-reference/functions/complete-snowflake-cortex 函数在 Snowflake 中设置检索增强生成 (RAG) 聊天机器人。

第 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。

第 3 步：创建一个分块 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 服务¶

在表上创建 Cortex Search 服务，允许您仔细搜索 book_description_chunks 中的文本块：

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 并添加 :code:`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 中创建了一个简单的文本数据搜索应用程序。您可继续阅读 :doc:` 教程 3 </user-guide/snowflake-cortex/cortex-search/tutorials/cortex-search-tutorial-3-chat-advanced>`，了解如何使用 Cortex Search 从一组 PDF 文件构建一个 AI 聊天机器人。