最近我獲得了一項任務，要為我們的目錄實現AI驅動的語義搜尋，而我們已經在使用Snowflake，因此決定使用Cortex Search Service來實現這個功能.
如果你還不知道Cortex Search是什麼，你可以快速查看這個連結以了解概覽基本上，Cortex Search 允許您直接在 Snowflake 中已儲存的数据上建立低延遲的語義和全文搜尋
在這篇文章中，我將分享我整合 Cortex Search Service 到我們的 Python 後端應用的經驗，包括我們遇到的陷阱
保持搜尋欄位專注
TheSEARCH_TEXT 欄位，或更精確地說是 Cortex Search 的「搜尋欄位」，是 Cortex Search 索引並用於搜尋回傳的欄位.
我最初犯的一個錯誤是試圖將幾乎所有可用的欄位放入 SEARCH_TEXT 欄位.
初始時的邏輯看起來合理：Cortex Search 看到的欄位越多，它擁有的上下文就越多。但在實際操作中，這可能會讓可搜尋的文本變得雜亂。
例如，像 IDs、狀態旗標、租戶/公司 IDs、貨幣 IDs 和內部分類 IDs 等欄位通常對於語義搜尋沒有意義：
'id:' || COALESCE(id::STRING, ''),
'warehouse_id:' || COALESCE(warehouse_id::STRING, ''),
'project_id:' || COALESCE(project_id::STRING, ''),
'is_active:' || COALESCE(is_active::STRING, ''),
'status:' || COALESCE(status, ''),
'currency_id:' || COALESCE(currency_id::STRING, '')
這些欄位通常不是使用者搜尋的內容。它們應該被顯示為ATTRIBUTES並用於過濾。我會在下一節中講解。
這個SEARCH_TEXT該欄位應專注於描述項目實際意義的字段：
CONCAT_WS(
' ',
title,
brand,
origin,
category_name,
subcategory_name,
short_description
) AS SEARCH_TEXT
我們需要包含對終端用戶搜尋查詢有用的欄位。
暴露可過濾欄位為ATTRIBUTES
Cortex 搜索服務配置中的另一個小但重要的細節是ATTRIBUTES 屬性.
The ON 欄位是可搜尋的文本欄位。這是 Cortex Search 用來匹配使用者查詢的方式。但如果你也需要根據元資料篩選結果，例如組織、狀態、分類、品牌、區域或可用性，這些欄位必須在服務建立時加入 ATTRIBUTES。
屬性並非主要的搜尋文字。它們是與搜尋結果一起回傳的欄位，可供篩選或顯示。Snowflake 文件中，Cortex Search 篩選功能會作用於在 CREATE CORTEX SEARCH SERVICE 命令中指定的 ATTRIBUTES 欄位上。
CREATE OR REPLACE CORTEX SEARCH SERVICE DEMO_DB.SEARCH.PRODUCT_SEARCH_SERVICE ON SEARCH_TEXT
ATTRIBUTES (
CITY,
COUNTRY,
CURRENCY,
IS_ACTIVE
)
WAREHOUSE = WH_SEARCH_DEMO
TARGET_LAG = '1 hour'
AS
SELECT
ID,
CITY,
COUNTRY,
CURRENCY,
IS_ACTIVE
FROM DEMO_DB.SEARCH.PRODUCT_SEARCH_SOURCE;
要記住的重點：你後來想要過濾的每一個欄位都必須作為屬性可用。此外，Snowflake指出ATTRIBUTES中的欄位必須包含在用於建立服務的來源查詢中。
接下來我們可以在python程式中應用這些過濾器：
from typing import Any, Dict, List
COUNTRY_ATTRIBUTE = "COUNTRY"
CITY_ATTRIBUTE = "CITY"
IS_ACTIVE_ATTRIBUTE = "IS_ACTIVE"
filters: List[Dict[str, Any]] = [
{
"@or": [
{
"@eq": {
COUNTRY_ATTRIBUTE: "UK",
}
},
{
"@eq": {
CITY_ATTRIBUTE: "London",
}
},
],
},
{
"@eq": {
IS_ACTIVE_ATTRIBUTE: True,
}
},
]
response = search_service.search(
query=query,
columns=[
"ID",
"COUNTRY",
"CITY",
"IS_ACTIVE",
],
filter={
"@and": filters,
},
limit=20,
)
嘗試讓過濾器選擇性強且容易理解。好的過濾器應該在Cortex Search排序和返回結果之前減少搜索空間。如果過濾器有效載荷變得太大或太複雜，這可能是一個信號，表示搜索來源表格應該調整，而不是將太多應用程式邏輯移入搜索查詢中.
注意TARGET_LAG
還有一個重要的爭論在建立或替換CORTEX搜尋服務指令是TARGET_LAG.
簡單來說，TARGET_LAG控制您的CORTEX搜尋索引與來源表的更新程度
例如：
TARGET_LAG = '10 minutes'
這不代表每一個新資料列會立即變得可搜尋。Cortex Search 還需要先更新其內部索引。所以當你在來源表中插入或更新資料列時，這些變更只會在下次更新發生後才會出現在搜尋結果中。
如果你使用受管理的嵌入，這點尤其重要。Cortex Search 需要處理更新後的來源數據，創建或更新嵌入，並刷新搜索索引，才能使用戶透過語義搜索找到這些記錄.
所以如果 TARGET_LAG 太長，你的搜索結果可能會變得過時。一個新發布的項目可能已經存在於你的來源表中，但用戶可能需要一段時間才能在搜索中找到它。
同時，將TARGET_LAG設定得太低也未必是最佳方案。更頻繁的更新可能意味著背景中需要更多 Snowflake 工作，這可能會增加信用使用量。Snowflake 也提到，如果目標延遲設定得太低，索引可能會比需要時更頻繁地刷新。
因此，正確的值取決於您的搜尋結果實際需要多新。
CREATE OR REPLACE CORTEX SEARCH SERVICE DEMO_DB.SEARCH.PRODUCT_SEARCH_SERVICE
ON SEARCH_TEXT
ATTRIBUTES
(
ITEM_ID,
STATUS,
IS_ACTIVE,
ACCOUNT_ID
)
WAREHOUSE = WH_SEARCH_DEMO
TARGET_LAG = '30 minutes'
AS
SELECT
ITEM_ID,
SEARCH_TEXT,
STATUS,
IS_ACTIVE,
ACCOUNT_ID
FROM DEMO_DB.SEARCH.PRODUCT_SEARCH_SOURCE;
若為面向用戶的目錄搜尋，新發表的物品應快速出現，那麼15-20分鐘或許是合理的。
若為內部知識庫、文件搜尋或任何不常變更的數據，1小時甚至更長時間也完全沒問題。
選擇適合您應用場景的嵌入模型
Cortex Search 在向量搜尋階段使用嵌入模型。簡單來說，這個模型將您的搜尋欄位和用戶查詢轉換為向量，讓 Cortex Search 可以找到語義上相似的記錄，而不只是包含相同關鍵字的記錄。Snowflake 允許在建立 Cortex Search 服務時，透過 EMBEDDING_MODEL 參數選擇模型。
CREATE OR REPLACE CORTEX SEARCH SERVICE DEMO_DB.SEARCH.PRODUCT_SEARCH_SERVICE
ON SEARCH_TEXT
ATTRIBUTES
(
ITEM_ID,
STATUS,
IS_ACTIVE,
ACCOUNT_ID
)
WAREHOUSE = WH_SEARCH_DEMO
TARGET_LAG = '30 minutes'
EMBEDDING_MODEL = 'snowflake-arctic-embed-m-v1.5' # custom EMBEDDING_MODEL
AS
SELECT
ITEM_ID,
SEARCH_TEXT,
STATUS,
IS_ACTIVE,
ACCOUNT_ID
FROM DEMO_DB.SEARCH.PRODUCT_SEARCH_SOURCE;
Snowflake 列表snowflake-arctic-embed-m-v1.5為預設的 Cortex 搜尋嵌入模型。它有 768 個輸出維度，一個 512 個 token 的上下文視窗，並支援英文語言。Snowflake 也將其描述為在可用的 Cortex 搜尋模型中索引速度最快的選項。這使它成為英文專用目錄或索引速度至關重要的內部搜尋的優佳起點。
若您的目錄是多語言的，僅支援英文的預設模型可能不適用於您。此時，請檢查支援多語言的模型，例如snowflake-arctic-embed-l-v2.0、snowflake-arctic-embed-l-v2.0-8k或voyage-multilingual-2。如需查詢支援的模型、維度、上下文視窗及語言支援的完整列表，請參考官方的Snowflake embedding models table。
Snowflake的CREATE CORTEX SEARCH SERVICE 文件顯示 EMBEDDING_MODEL 為服務定義的一部分，更改它通常意味著重新創建服務，而不只是調整一個運行時參數。因此，值得早期測試這一點，特別是如果您的產品有多語言數據或用戶使用不同語言搜尋.
Cortex 連接預熱
我們整合Cortex Search的主要原因是為了減少搜尋查詢延遲，因為我們有大量的物品。
第一個瓶頸來自連接處理。我使用了Snowflake Python library，而每個新的搜尋請求都需要打開一個新的連接。僅僅是連接設置就花費了約1–1.5秒，而實際的搜尋查詢只花了500–600毫秒。所以在優化搜尋本身之前，我專注於從請求路徑中移除連接設置的成本。
此次優化是將 Snowflake 連接設定移出請求路徑。我新增了一個 warmup() 方法，它會解析一次 Cortex Search 服務，並在工作者進程層級緩存 Snowflake 連接、Root 物件和服務參考。這個緩存受鎖定保護，因此即使同時有多個請求到達，初始化也能保持安全。
def warmup(self) -> None:
self._get_service()
def _get_service(self):
cache_key = (self.database, self.schema, self.service_name)
service = self.__class__._service_cache.get(cache_key)
if service:
return service
with self.__class__._lock:
service = self.__class__._service_cache.get(cache_key)
if service:
return service
root = self._get_root()
service = (
root.databases[self.database]
.schemas[self.schema]
.cortex_search_services[self.service_name]
)
self.__class__._service_cache[cache_key] = service
return service
def _get_root(self):
root = self.__class__._root
connection = self.__class__._connection
if root and connection and not connection.is_closed():
return root
with self.__class__._lock:
root = self.__class__._root
connection = self.__class__._connection
if root and connection and not connection.is_closed():
return root
if connection and not connection.is_closed():
try:
connection.close()
except Exception:
pass
self.__class__._connection = None
self.__class__._root = None
self.__class__._service_cache = {}
try:
connection = snowflake.connector.connect(**self._connection_parameters)
except Exception as exc:
raise CortexSearchCatalogServiceError(
f"Failed to create Snowflake connection for Cortex search: {exc}"
)
self.__class__._connection = connection
root = Root(connection)
self.__class__._root = root
return root
接著我從 Gunicorn 的 post_fork 鈎子呼叫了這個暖身方法。因為 Gunicorn 工作程式是獨立的進程，每個工作進程在分叉後都必須建立自己的 Snowflake 連接。透過這個變更，連接是在工作進程啟動時建立，而不是在第一次搜尋請求時建立，這消除了用戶端路徑上的 1–1.5 秒連接設定成本。
def post_fork(server, worker):
# Warm the Snowflake Cortex client in each worker process so the first user
# search does not pay the connection/session setup cost on the request path.
try:
if not _cortex_warmup_is_configured():
server.log.debug(
"Skipping Cortex warmup for worker pid=%s because Snowflake settings are incomplete.",
worker.pid,
)
return
from apps.db.cortex_search_services.cortex_search_catalog_service import CortexSearchCatalogService
CortexSearchCatalogService().warmup()
server.log.info("Cortex warmup completed for worker pid=%s", worker.pid)
except Exception:
server.log.exception("Cortex warmup failed for worker pid=%s", worker.pid)
調整Cortex搜尋分數權重
接下來重要的優化不是關於延遲，而是關於結果品質。
Cortex Search 使用混合排序。它可以結合語義/向量相似度、關鍵字/文字匹配，以及神經重新排序。Snowflake 透過 scoring_config.weights 來暴露這個功能，其中向量、文字和重新排序器控制每個評分組成部分的相對貢獻。預設情況下，這些組成部分具有相等的權重，但你可以根據你的使用情況，每個查詢進行調整。
例如，在我們的情況下，純關鍵字匹配有時會因為它包含匹配的詞語而將錯誤的項目排名更高。一個真實的例子是「手套保護」的標誌被排名在實際手套之上。文本匹配很強，但在語義上它並不是用戶預期的產品.
為了解決這個問題，我增加了相對於文本分數的向量分數權重：
# Tune these weights to adjust Cortex ranking before results reach application code.
# vectors = semantic similarity, texts = keyword match, reranker = neural reranker.
# Raise vectors relative to texts to prevent keyword-heavy but semantically irrelevant
# items from outranking semantically correct ones.
DEFAULT_SCORING_CONFIG = {
"weights": {
"vectors": 2,
"texts": 1,
"reranker": 1,
}
}
這是一個重要的參數，因為它直接改變了Cortex Search在結果傳遞到您的應用程式代碼之前如何排序結果。如果您的搜尋更接近傳統的全文搜尋，您可能想要增加text的weight。如果您的用戶透過意義、同義詞、描述或自然語言進行搜尋，增加vector的weight可以產生更好的結果。
另一個值得測試的參數是重新排序器。Cortex Search 預設使用語義重新排序來提高相關性，但重新排序也可能增加查詢延遲。Snowflake 允許在延遲比額外的品質改進更重要的情況下禁用重新排序。
從 Cortex Search 僅僅返回您需要的数据
在您將 Cortex Search 接入生產代碼之前，有一個實際的限制值得了解，那就是有效載荷大小.
Snowflake 文檔對 Cortex Search 查詢的響應大小進行了限制：REST API 和 Python API 的響應有效載荷不得超過 10 MB
這意味著您需要謹慎處理查詢的雙方：您發送到 Cortex Search 的內容以及您要求它返回的內容。
對於過濾器，盡量只傳送真正需要的內容。過於複雜的嵌套是生產環境中靜默錯誤的最短途徑。如果你在相同字段上有大型的OR條件，在可能的情况下，優先選擇更緊湊的操作符。例如，而不是建立一個長的or列表：
filter={
"@or": [
{"@eq": {"STATUS": "published"}},
{"@eq": {"STATUS": "scheduled"}},
{"@eq": {"STATUS": "archived"}},
]
}
盡量使用usein 如果符合您的情况：
filter={
"@in": {
"STATUS": ["published", "scheduled", "archived"]
}
}
對你的數據模型也是如此。如果每個搜尋請求都需要大量的過濾邏輯，這可能是一個信號，表示用於搜尋的模型設計得還不夠好。有時候，準備一個更乾淨的搜尋來源表格或視圖，其中包含更容易進行過濾的字段，而不是將過多的應用程式邏輯推入 Cortex Search 請求，這是值得考慮的。
回應方面，我比較喜歡保持Cortex Search結果欄位為小型。在大多數應用程式搜尋流程中，Cortex Search不需要回傳完整的物件有效負載。它只需要回傳內部ID，以及一些用於排序或除錯的輕量級欄位即可。
response = search_service.search(
query=query,
columns=[
"ID",
],
filter=filter,
limit=50,
)
然後應用程式可以取得這些 ID，並從主要應用程式資料庫中擷取完整記錄：
item_ids = [row["ID"] for row in response.results]
items = (
CatalogItem.objects
.filter(id__in=item_ids)
.select_related("brand", "category")
.prefetch_related("tags")
)
這讓 Cortex Search 聚焦於它最擅長的事：尋找相關記錄。您的應用程式資料庫仍然負責載入所有領域物件，並且您可以節省查詢成本。
結論
Cortex Search 是一個強大的選項，當您已經在 Snowflake 中有數據，並且需要低延遲的語義和全文搜索，而不需要從頭開始建立一個獨立的搜索流程時。
但重要的是，它並不是完全「設定後就忘記」。質量和性能很大程度上取決於您如何配置該服務以及您如何從後端應用程式中使用它。
我經驗中最主要的教訓是：
- 在第一個用戶請求之前預熱 Snowflake 連接；
- 將搜尋欄位專注於用戶實際搜尋的字段；
- 根據您的數據和語言需求選擇嵌入模型；
- 根據您的應用場景是否需要更多語義或關鍵字匹配來調整評分權重；
- 如果您需要過濾，請將必要的欄位添加到 ATTRIBUTES。
- 根據搜尋結果需要的新鮮度來設定 TARGET_LAG；
- 僅從 Cortex Search 回傳您需要的數據，並從您的主要應用程式資料庫載入完整物件；
總體來說，Cortex Search 對應用程式層級的搜尋非常有效，特別是目錄搜尋、文件搜尋，以及其他已儲存在 Snowflake 中的數據。