教程 1：使用 Cortex Search 构建简单的搜索应用程序¶

简介¶

本教程介绍如何开始使用 Cortex Search 创建一个简单的搜索应用程序。

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

第 2 步：将数据加载到 Snowflake¶

在创建搜索服务之前，必须将示例数据加载到 Snowflake 中。

您可在 Snowsight 中或使用 SQL 上传数据集。要在 Snowsight 中上传，请执行以下操作：

1. 选择左侧导航栏上方的 + Create 按钮。

2. 然后选择 Table » From File。

   

3. 从右上角的下拉列表中选择新创建的仓库作为表格仓库。

4. 将 JSON 数据文件拖放到对话框中。

5. 选择上面创建的数据库，并指定 PUBLIC 架构。

6. 最后，指定创建一个名为 airbnb_listings 的新表，并选择 Next。

   

7. 在 Load Data into Table 对话框中，进行以下调整。首先，取消选中 image_embeddings、images 和 text_embeddings 列，因为这些列不适用于本教程。其次，将 amenities 字段的数据类型调整为 ARRAY 类型。

   

   

8. 完成这些调整后，选择 Load 以继续。

9. 片刻之后，即会看到一个确认页面，显示数据已加载。

   

10. 选择 Query Data，打开一个新的 Snowsight 工作表，您将在下一步中使用它。

第 3 步：创建搜索服务¶

运行以下 SQL 命令，在新表上创建搜索服务。

CREATE OR REPLACE CORTEX SEARCH SERVICE cortex_search_tutorial_db.public.airbnb_svc
ON listing_text
ATTRIBUTES room_type, amenities
WAREHOUSE = cortex_search_tutorial_wh
TARGET_LAG = '1 hour'
AS
    SELECT
        room_type,
        amenities,
        price,
        cancellation_policy,
        ('Summary\n\n' || summary || '\n\n\nDescription\n\n' || description || '\n\n\nSpace\n\n' || space) as listing_text
    FROM
    cortex_search_tutorial_db.public.airbnb_listings;

让我们来分析一下此命令的参数：

• ON 参数用于指定查询要搜索的列。在这种情况下，它是 listing_text，在源查询中作为基表中若干文本列的连接生成。

• ATTRIBUTES 参数指定可以筛选搜索结果的列。向 listing_text 列发出查询时，该示例会对 room_type 和 amenities 进行筛选。

• WAREHOUSE 和 TARGET_LAG 参数分别指定用户提供的仓库和搜索服务所需的新鲜度。本示例指定使用 cortex_search_tutorial_wh 仓库创建索引和执行刷新，并将服务保持在源表 AIRBNB_LISTINGS 之后不超过 '1 hour'。

• AS 字段定义服务的源表。此示例将原始表中的几个文本列连接到搜索列 listing_text 中，以便查询可以搜索多个字段。

第 4 步：创建 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 python packages
   import streamlit as st
   from snowflake.core import Root
   from snowflake.snowpark.context import get_active_session
   
   # Constants
   DB = "cortex_search_tutorial_db"
   SCHEMA = "public"
   SERVICE = "airbnb_svc"
   BASE_TABLE = "cortex_search_tutorial_db.public.airbnb_listings"
   ARRAY_ATTRIBUTES = {"AMENITIES"}
   
   
   def get_column_specification():
       """
       Returns the name of the search column and a list of the names of the attribute columns
       for the provided cortex search service
       """
       session = get_active_session()
       search_service_result = session.sql(f"DESC CORTEX SEARCH SERVICE {DB}.{SCHEMA}.{SERVICE}").collect()[0]
       st.session_state.attribute_columns = search_service_result.attribute_columns.split(",")
       st.session_state.search_column = search_service_result.search_column
       st.session_state.columns = search_service_result.columns.split(",")
   
   def init_layout():
       st.title("Cortex AI Search")
       st.markdown(f"Querying service: `{DB}.{SCHEMA}.{SERVICE}`".replace('"', ''))
   
   def query_cortex_search_service(query, filter={}):
       """
       Queries the cortex search service in the session state and returns a list of results
       """
       session = get_active_session()
       cortex_search_service = (
           Root(session)
           .databases[DB]
           .schemas[SCHEMA]
           .cortex_search_services[SERVICE]
       )
       context_documents = cortex_search_service.search(
           query,
           columns=st.session_state.columns,
           filter=filter,
           limit=st.session_state.limit)
       return context_documents.results
   
   @st.cache_data
   def distinct_values_for_attribute(col_name, is_array_attribute=False):
       session = get_active_session()
       if is_array_attribute:
           values = session.sql(f'''
           SELECT DISTINCT value FROM {BASE_TABLE},
           LATERAL FLATTEN(input => {col_name})
           ''').collect()
       else:
           values = session.sql(f"SELECT DISTINCT {col_name} AS VALUE FROM {BASE_TABLE}").collect()
       return [ x["VALUE"].replace('"', "") for x in values ]
   
   def init_search_input():
       st.session_state.query = st.text_input("Query")
   
   def init_limit_input():
       st.session_state.limit = st.number_input("Limit", min_value=1, value=5)
   
   def init_attribute_selection():
       st.session_state.attributes = {}
       for col in st.session_state.attribute_columns:
           is_multiselect = col in ARRAY_ATTRIBUTES
           st.session_state.attributes[col] = st.multiselect(
               col,
               distinct_values_for_attribute(col, is_array_attribute=is_multiselect)
           )
   
   def display_search_results(results):
       """
       Display the search results in the UI
       """
       st.subheader("Search results")
       for i, result in enumerate(results):
           result = dict(result)
           container = st.expander(f"[Result {i+1}]", expanded=True)
   
           # Add the result text.
           container.markdown(result[st.session_state.search_column])
   
           # Add the attributes.
           for column, column_value in sorted(result.items()):
               if column == st.session_state.search_column:
                   continue
               container.markdown(f"**{column}**: {column_value}")
   
   def create_filter_object(attributes):
       """
       Create a filter object for the search query
       """
       and_clauses = []
       for column, column_values in attributes.items():
           if len(column_values) == 0:
               continue
           if column in ARRAY_ATTRIBUTES:
               for attr_value in column_values:
                   and_clauses.append({"@contains": { column: attr_value }})
           else:
               or_clauses = [{"@eq": {column: attr_value}} for attr_value in column_values]
               and_clauses.append({"@or": or_clauses })
   
       return {"@and": and_clauses} if and_clauses else {}
   
   
   def main():
       init_layout()
       get_column_specification()
       init_attribute_selection()
       init_limit_input()
       init_search_input()
   
       if not st.session_state.query:
           return
       results = query_cortex_search_service(
           st.session_state.query,
           filter = create_filter_object(st.session_state.attributes)
       )
       display_search_results(results)
   
   
   if __name__ == "__main__":
       st.set_page_config(page_title="Cortex AI Search and Summary", layout="wide")
       main()
   

   Copy

下面简要介绍上述 Streamlit-in-Snowflake 代码中的主要组成部分：

• get_column_specification 使用 DESCRIBE SQL 查询获取搜索服务中可用的属性信息，并将其存储在 Stremalit 状态。

• init_layout 设置页面的标头和引言。

• query_cortex_search_service 通过 Python 客户端库处理对 Cortex Search 服务的查询。

• create_filter_object 将 Streamlit 表单中的选定筛选器属性处理为合适的对象，以供 Python 库用于查询 Cortex Search。

• distinct_values_for_attribute 决定每个可筛选属性的可能值，以填充下拉菜单。

• init_search_input、init_limit_input、init_attribute_selection 初始化搜索查询输入、结果数量限制和属性筛选器。

• display_search_results 将搜索结果格式化为显示在结果页面中的 Markdown 元素。

第 5 步：清理¶

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

后续步骤¶

恭喜！您已成功在 Snowflake 中创建了一个简单的文本数据搜索应用程序。您可以继续阅读 教程 2，了解如何层叠 CortexLLM 函数，使用 Cortex Search 构建 AI 聊天机器人。