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

简介¶

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

您将学习以下内容¶

在 AirBnb 列表数据集上创建 Cortex Search 服务。
创建一个可以查询 Cortex Search 服务的 Streamlit in Snowflake 应用程序。

先决条件¶

要完成本教程，需要满足以下先决条件：

您拥有一个 Snowflake 账户和用户，该用户具有这样的角色：可授予创建数据库、表、虚拟仓库对象、Cortex Search 服务和 Streamlit 应用程序所需的权限。

请参阅 20 分钟学会使用 Snowflake 以获得满足这些要求的说明。

第 1 步：设置¶

获取样本数据¶

您将使用托管在 Huggingface (link removed) 上的示例数据集，并以单个 JSON 文件的形式下载。通过此链接直接从浏览器下载文件：

AirBnB 列表数据集 (link removed)

备注

在非教程设置中，您将自带数据，这些数据可能已经存在于 Snowflake 表中。

创建数据库、表和仓库¶

执行以下语句，创建本教程所需的数据库和虚拟仓库。完成本教程后，您可以删除这些对象。

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;

Copy

请注意以下事项：

CREATE DATABASE 语句创建一个数据库。数据库自动包含一个名为“public”的架构。
CREATE WAREHOUSE 语句创建一个最初暂停的仓库。该语句还设置 AUTO_RESUME = true，这将在您执行需要计算资源的 SQL 语句时自动启动仓库。

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

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

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

选择左侧导航栏上方的 + Create 按钮。
然后选择 Table » From File。
从右上角的下拉列表中选择新创建的仓库作为表格仓库。
将 JSON 数据文件拖放到对话框中。
选择上面创建的数据库，并指定 PUBLIC 架构。
最后，指定创建一个名为 airbnb_listings 的新表，并选择 Next。
在 Load Data into Table 对话框中，进行以下调整。首先，取消选中 image_embeddings、images 和 text_embeddings 列，因为这些列不适用于本教程。其次，将 amenities 字段的数据类型调整为 ARRAY 类型。
完成这些调整后，选择 Load 以继续。
片刻之后，即会看到一个确认页面，显示数据已加载。
选择 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;

Copy

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

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 角色与服务创建步骤中用于创建服务的角色相同。

Sign in to Snowsight.
In the navigation menu, select Projects » Streamlit。
选择 + Streamlit App。
重要：选择应用程序位置的 cortex_search_tutorial_db 数据库和 public 架构。
在 Streamlit in Snowflake 编辑器的左侧窗格中，选择 Packages 并添加 :code:`snowflake`（版本 >= 0.8.0），将软件包安装到应用程序中。

用以下 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 查询获取搜索服务中可用的属性信息，并将其存储在 Streamlit 状态。
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 将搜索结果格式化为显示在“Results”页面中的 Markdown 元素。

第 5 步：清理¶

清理（可选）¶

执行以下 DROP <object> 命令，将系统恢复到教程开始前的状态：

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

Copy

删除数据库会自动移除所有子数据库对象，例如表。

后续步骤¶

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

其他资源¶

此外，您还可以利用以下资源继续学习：