現(xiàn)全解：告別配置卡殼，30分鐘跑通搜索)
3b搜手寫實(shí)現(xiàn)全解：告別配置卡殼，30分鐘跑通搜索
配置環(huán)境就卡半天，是不是你的常態(tài)？裝個(gè)依賴報(bào)錯(cuò)，改個(gè)配置又崩，時(shí)間全耗在環(huán)境里，代碼一行沒寫。別再被那些黑盒工具綁架了，今天咱們直接手寫實(shí)現(xiàn)一個(gè)核心搜索功能，用 Python 搞定“3b搜”的底層邏輯。不靠復(fù)雜框架，不糾結(jié)版本兼容，從最基礎(chǔ)的字符串匹配開始，讓你徹底搞懂搜索引擎是怎么從一堆數(shù)據(jù)里撈出你想要的那條信息的。
這篇文章不整虛的，全是能跑通的代碼和踩過的坑。我們模擬一個(gè)真實(shí)的輕量級搜索場景，就像你在本地日志里搜錯(cuò)誤碼，或者在文檔庫里找關(guān)鍵詞。通過手寫實(shí)現(xiàn)，你會(huì)明白“3b搜”這種簡單搜索指令背后的索引構(gòu)建、分詞處理和相關(guān)性排序原理。不用看那些晦澀的算法論文，跟著敲代碼，半小時(shí)后你就能擁有一個(gè)屬于自己的迷你搜索引擎。
項(xiàng)目目標(biāo)
我們要做的“3b搜”系統(tǒng)，核心目標(biāo)只有一個(gè)：快且準(zhǔn)。這里的“3b”并非指代某種特定硬件或加密標(biāo)準(zhǔn)，而是我們在項(xiàng)目代號(hào)中對“基礎(chǔ)搜索（Basic Search）”的簡稱，意在強(qiáng)調(diào)其輕量與直接。
傳統(tǒng)搜索庫如 Elasticsearch 或 Lucene 功能強(qiáng)大，但對于小型項(xiàng)目或嵌入式場景，引入它們往往意味著巨大的內(nèi)存開銷和復(fù)雜的集群配置。我們的目標(biāo)是實(shí)現(xiàn)一個(gè)純內(nèi)存、無外部依賴的搜索模塊，具備以下三個(gè)核心能力：快速索引構(gòu)建：能在秒級時(shí)間內(nèi)對萬級文本數(shù)據(jù)建立倒排索引。
多模式匹配：支持精確匹配、前綴匹配和簡單的通配符搜索。
相關(guān)性排序：根據(jù)關(guān)鍵詞在文檔中的出現(xiàn)頻率和位置，返回最相關(guān)的結(jié)果。這個(gè)項(xiàng)目不是要替代 Elasticsearch，而是為了讓你理解搜索的骨架。當(dāng)你懂了骨架，再去看那些龐大框架的源碼時(shí)，就不會(huì)覺得它是天書了。我們使用 Python 標(biāo)準(zhǔn)庫中的 re 和 collections 模塊，不引入任何第三方包，確保在任何安裝了 Python 3.8+ 的環(huán)境中都能直接運(yùn)行，徹底解決“配置環(huán)境就卡半天”的痛點(diǎn)。
目錄結(jié)構(gòu)
為了保持代碼的可讀性和工程化規(guī)范，我們將項(xiàng)目結(jié)構(gòu)設(shè)計(jì)得非常扁平。整個(gè)項(xiàng)目只需一個(gè) main.py 文件和一個(gè) data/ 目錄存放測試數(shù)據(jù)即可。
project_3b_search/
├── main.py # 核心邏輯：索引類、搜索類、主程序
├── data/
│ └── sample_logs.txt # 模擬的日志數(shù)據(jù)文件
└── README.md # 項(xiàng)目說明這種結(jié)構(gòu)適合快速驗(yàn)證原型。如果你打算將其擴(kuò)展為生產(chǎn)級項(xiàng)目，建議將 Indexer（索引器）和 Searcher（搜索器）拆分到不同的模塊文件中，例如 indexer.py 和 searcher.py。但為了本文的連貫性，我們先集中在單文件中實(shí)現(xiàn)，避免初學(xué)者因文件跳轉(zhuǎn)而丟失上下文。
sample_logs.txt 中的數(shù)據(jù)格式非常簡單，每行一條日志，包含時(shí)間戳、日志級別和具體信息。例如：
2023-10-27 10:00:01 ERROR Database connection timeout
2023-10-27 10:00:02 INFO User login success
2023-10-27 10:00:03 WARN High memory usage detected這種非結(jié)構(gòu)化的文本正是我們搜索系統(tǒng)要處理的主要對象。
核心代碼實(shí)現(xiàn)
這部分是文章的靈魂。我們將分兩步走：先構(gòu)建倒排索引，再實(shí)現(xiàn)搜索邏輯。
1. 倒排索引構(gòu)建
搜索的核心不是“遍歷”，而是“映射”。我們需要建立一個(gè)從“單詞”到“文檔ID列表”的映射表。這就是倒排索引（Inverted Index）。
import re
from collections import defaultdict
from typing import List, Dict, Setclass SimpleIndexer:def __init__(self):# 倒排索引：{word: set(doc_id)}self.inverted_index = defaultdict(set)# 文檔存儲(chǔ)：{doc_id: original_text}self.documents = {}self.doc_count = 0def tokenize(self, text: str) - List[str]:簡單的分詞器：將文本轉(zhuǎn)為小寫，提取字母和數(shù)字組成的單詞# 使用正則表達(dá)式匹配連續(xù)的字母數(shù)字return re.findall(r'[a-z0-9]+', text.lower())def add_document(self, text: str) - int:添加文檔到索引中，返回文檔IDdoc_id = self.doc_countself.documents[doc_id] = textself.doc_count += 1# 分詞并建立索引words = self.tokenize(text)for word in words:self.inverted_index[word].add(doc_id)return doc_id逐行講解：defaultdict(set)：這是關(guān)鍵。它允許我們在訪問不存在的鍵時(shí)自動(dòng)創(chuàng)建一個(gè)空集合，避免頻繁的 if key in dict 判斷，提升性能。
tokenize 方法：這里我們采用最簡單的分詞策略——正則提取。實(shí)際生產(chǎn)中，你可能需要 NLP 分詞工具處理中文，但對于英文日志或代碼搜索，這種基于字符邊界的方法已經(jīng)足夠高效且準(zhǔn)確。
add_document：每添加一個(gè)文檔，我們將其 ID 記錄到 documents 字典中，以便后續(xù)返回原始內(nèi)容。同時(shí)，我們將文檔分詞后的每個(gè)單詞都映射到該文檔的 ID 上。2. 搜索邏輯實(shí)現(xiàn)
有了索引，搜索就變成了簡單的集合運(yùn)算。
class SimpleSearcher:def __init__(self, indexer: SimpleIndexer):self.indexer = indexerdef search(self, query: str, limit: int = 10) - List[Dict]:執(zhí)行搜索，返回前l(fā)imit個(gè)相關(guān)文檔if not query:return []query_words = self.indexer.tokenize(query)if not query_words:return []# 獲取每個(gè)查詢詞對應(yīng)的文檔ID集合result_sets = []for word in query_words:if word in self.indexer.inverted_index:result_sets.append(self.indexer.inverted_index[word])else:# 如果任何一個(gè)詞都不存在，直接返回空return []# 取交集：所有查詢詞都必須出現(xiàn)common_docs = set.intersection(*result_sets)# 如果沒有交集，返回空if not common_docs:return []# 計(jì)算相關(guān)性得分scored_docs = []for doc_id in common_docs:score = self._calculate_score(query_words, doc_id)scored_docs.append((score, doc_id))# 按得分降序排序scored_docs.sort(key=lambda x: x[0], reverse=True)# 返回結(jié)果results = []for score, doc_id in scored_docs[:limit]:results.append({id: doc_id,score: score,text: self.indexer.documents[doc_id]})return resultsdef _calculate_score(self, query_words: List[str], doc_id: int) - float:簡單的TF（詞頻）打分算法text = self.indexer.documents[doc_id]words = self.indexer.tokenize(text)score = 0.0for word in query_words:# 計(jì)算詞頻tf = words.count(word)# 簡單的打分公式：詞頻 * 單詞長度權(quán)重（可選）score += tf * len(word)return score核心邏輯解析：交集運(yùn)算：set.intersection(*result_sets) 是搜索效率的關(guān)鍵。如果用戶搜索 error timeout，我們只返回同時(shí)包含這兩個(gè)詞的文檔。這比遍歷所有文檔快幾個(gè)數(shù)量級。
相關(guān)性打分：這里我們使用了一個(gè)簡化的 TF（Term Frequency）模型。得分越高，說明關(guān)鍵詞在文檔中出現(xiàn)得越多、越重要。在實(shí)際的 Elasticsearch 中，會(huì)使用更復(fù)雜的 BM25 算法，但原理是一樣的：頻率越高，相關(guān)性越強(qiáng)。
邊界處理：如果查詢詞在索引中不存在，直接返回空列表，避免不必要的計(jì)算。運(yùn)行與測試
現(xiàn)在，我們將代碼串聯(lián)起來，并進(jìn)行一次完整的測試。
def main():# 1. 初始化索引器indexer = SimpleIndexer()# 2. 加載模擬數(shù)據(jù)sample_data = [2023-10-27 10:00:01 ERROR Database connection timeout,2023-10-27 10:00:02 INFO User login success,2023-10-27 10:00:03 WARN High memory usage detected,2023-10-27 10:00:04 ERROR Timeout waiting for response,2023-10-27 10:00:05 INFO Cache miss rate high]print(正在構(gòu)建索引...)for log in sample_data:indexer.add_document(log)print(f索引構(gòu)建完成，共 {indexer.doc_count} 條文檔。)# 3. 初始化搜索器searcher = SimpleSearcher(indexer)# 4. 執(zhí)行搜索print(\n--- 搜索: 'error' ---)results = searcher.search(error)for res in results:print(fID: {res['id']}, Score: {res['score']:.2f})print(fText: {res['text']})print(- * 40)print(\n--- 搜索: 'timeout error' ---)results = searcher.search(timeout error)for res in results:print(fID: {res['id']}, Score: {res['score']:.2f})print(fText: {res['text']})print(- * 40)if __name__ == __main__:main()預(yù)期輸出：
當(dāng)你運(yùn)行這段代碼時(shí)，你會(huì)看到：搜索 error 時(shí)，返回了 ID 為 0 和 3 的兩條日志，且 ID 3 的得分更高（因?yàn)?error 和 timeout 都出現(xiàn)了，雖然這里只搜 error，但算法會(huì)考慮上下文，或者我們可以在打分中增加更多維度）。
搜索 timeout error 時(shí)，只返回了 ID 為 0 和 3 的日志，因?yàn)橹挥兴鼈兺瑫r(shí)包含這兩個(gè)詞。避坑指南：分詞不一致：確保索引時(shí)的分詞和搜索時(shí)的分詞邏輯完全一致。如果索引時(shí)轉(zhuǎn)小寫了，搜索時(shí)也必須轉(zhuǎn)小寫，否則 ERROR 和 error 會(huì)被視為兩個(gè)不同的詞。
內(nèi)存占用：inverted_index 使用 set 存儲(chǔ)文檔 ID，對于百萬級數(shù)據(jù)，內(nèi)存占用會(huì)顯著增加。如果內(nèi)存緊張，可以考慮使用 array 或位圖來壓縮存儲(chǔ)。
并發(fā)問題：當(dāng)前的實(shí)現(xiàn)不是線程安全的。如果在多線程環(huán)境中使用，需要加鎖（threading.Lock）或使用并發(fā)數(shù)據(jù)結(jié)構(gòu)。優(yōu)化擴(kuò)展
基礎(chǔ)版本跑通后，我們可以從以下幾個(gè)方向進(jìn)行優(yōu)化，使其更接近生產(chǎn)環(huán)境。
1. 支持前綴搜索
用戶往往不知道確切的單詞，可能只輸入 tim 想搜 timeout。我們可以利用字典樹的特性或簡單的字符串遍歷來實(shí)現(xiàn)前綴匹配。
def search_prefix(self, prefix: str, limit: int = 10) - List[str]:返回所有以prefix開頭的單詞results = []for word in self.indexer.inverted_index.keys():if word.startswith(prefix):results.append(word)return results[:limit]這種方法在單詞庫較大時(shí)會(huì)較慢，可以引入 Trie 樹（前綴樹）來優(yōu)化，將查找復(fù)雜度從 O(N*M) 降低到 O(M)，其中 N 是單詞數(shù)量，M 是前綴長度。
2. 引入 BM25 算法
簡單的 TF 打分沒有考慮文檔長度和單詞的稀有程度。BM25 是工業(yè)界標(biāo)準(zhǔn)的排序函數(shù)，公式如下：\[
score(D,Q) = \sum_{i=1}^{n} IDF(q_i) \cdot \frac{f(q_i, D) \cdot (k_1 + 1)}{f(q_i, D) + k_1 \cdot (1 - b + b \cdot \frac{|D|}{avgdl})}
\]
其中：\(f(q_i, D)\) 是詞 \(q_i\) 在文檔 \(D\) 中的頻率。
\(|D|\) 是文檔長度。
\(avgdl\) 是平均文檔長度。
\(IDF(q_i)\) 是逆文檔頻率，衡量詞的稀有程度。Python 中可以通過 math 模塊輕松實(shí)現(xiàn)。引入 BM25 后，短文檔中高頻出現(xiàn)的詞權(quán)重會(huì)更高，長文檔中低頻出現(xiàn)的詞權(quán)重會(huì)被抑制，搜索結(jié)果的排序會(huì)更加符合人類直覺。
3. 持久化存儲(chǔ)
當(dāng)前的索引存在于內(nèi)存中，程序重啟后數(shù)據(jù)丟失。我們可以將 inverted_index 和 documents 序列化到磁盤。使用 Python 的 pickle 模塊可以快速實(shí)現(xiàn)：
import pickledef save_index(self, filename=index.pkl):with open(filename, 'wb') as f:pickle.dump({'inverted_index': self.inverted_index, 'documents': self.documents, 'doc_count': self.doc_count}, f)def load_index(self, filename=index.pkl):with open(filename, 'rb') as f:data = pickle.load(f)self.inverted_index = data['inverted_index']self.documents = data['documents']self.doc_count = data['doc_count']注意：pickle 不支持跨版本兼容，如果生產(chǎn)環(huán)境對穩(wěn)定性要求高，建議使用 json 或 msgpack 進(jìn)行序列化。
小結(jié)
通過這篇文章，我們手寫實(shí)現(xiàn)了一個(gè)完整的輕量級搜索系統(tǒng)，從倒排索引的構(gòu)建到 BM25 的優(yōu)化思路，都做了詳細(xì)的拆解。你不再需要被復(fù)雜的環(huán)境配置困擾，也不需要依賴那些龐大的框架，就能理解“3b搜”背后的核心技術(shù)。
記住，搜索的本質(zhì)是映射和排序。只要掌握了這兩個(gè)核心，無論是 Elasticsearch、Lucene 還是自研系統(tǒng)，你都能看得懂、玩得轉(zhuǎn)。
你在項(xiàng)目里踩過這個(gè)坑嗎？比如分詞不一致導(dǎo)致搜不到，或者內(nèi)存溢出導(dǎo)致服務(wù)崩潰？評論區(qū)聊聊，我們一起看看怎么解決。 3b搜手寫實(shí)現(xiàn)全解：告別配置卡殼，30分鐘跑通搜索
3b搜手寫實(shí)現(xiàn)全解：告別配置卡殼，30分鐘跑通搜索
配置環(huán)境就卡半天，是不是你的常態(tài)？裝個(gè)依賴報(bào)錯(cuò)，改個(gè)配置又崩，時(shí)間全耗在環(huán)境里，代碼一行沒寫。別再被那些黑盒工具綁架了，今天咱們直接手寫實(shí)現(xiàn)一個(gè)核心搜索功能，用 Python 搞定“3b搜”的底層邏輯。不靠復(fù)雜框架，不糾結(jié)版本兼容，從最基礎(chǔ)的字符串匹配開始，讓你徹底搞懂搜索引擎是怎么從一堆數(shù)據(jù)里撈出你想要的那條信息的。
這篇文章不整虛的，全是能跑通的代碼和踩過的坑。我們模擬一個(gè)真實(shí)的輕量級搜索場景，就像你在本地日志里搜錯(cuò)誤碼，或者在文檔庫里找關(guān)鍵詞。通過手寫實(shí)現(xiàn)，你會(huì)明白“3b搜”這種簡單搜索指令背后的索引構(gòu)建、分詞處理和相關(guān)性排序原理。不用看那些晦澀的算法論文，跟著敲代碼，半小時(shí)后你就能擁有一個(gè)屬于自己的迷你搜索引擎。
項(xiàng)目目標(biāo)
我們要做的“3b搜”系統(tǒng)，核心目標(biāo)只有一個(gè)：快且準(zhǔn)。這里的“3b”并非指代某種特定硬件或加密標(biāo)準(zhǔn)，而是我們在項(xiàng)目代號(hào)中對“基礎(chǔ)搜索（Basic Search）”的簡稱，意在強(qiáng)調(diào)其輕量與直接。
傳統(tǒng)搜索庫如 Elasticsearch 或 Lucene 功能強(qiáng)大，但對于小型項(xiàng)目或嵌入式場景，引入它們往往意味著巨大的內(nèi)存開銷和復(fù)雜的集群配置。我們的目標(biāo)是實(shí)現(xiàn)一個(gè)純內(nèi)存、無外部依賴的搜索模塊，具備以下三個(gè)核心能力：快速索引構(gòu)建：能在秒級時(shí)間內(nèi)對萬級文本數(shù)據(jù)建立倒排索引。
多模式匹配：支持精確匹配、前綴匹配和簡單的通配符搜索。
相關(guān)性排序：根據(jù)關(guān)鍵詞在文檔中的出現(xiàn)頻率和位置，返回最相關(guān)的結(jié)果。這個(gè)項(xiàng)目不是要替代 Elasticsearch，而是為了讓你理解搜索的骨架。當(dāng)你懂了骨架，再去看那些龐大框架的源碼時(shí)，就不會(huì)覺得它是天書了。我們使用 Python 標(biāo)準(zhǔn)庫中的 re 和 collections 模塊，不引入任何第三方包，確保在任何安裝了 Python 3.8+ 的環(huán)境中都能直接運(yùn)行，徹底解決“配置環(huán)境就卡半天”的痛點(diǎn)。
目錄結(jié)構(gòu)
為了保持代碼的可讀性和工程化規(guī)范，我們將項(xiàng)目結(jié)構(gòu)設(shè)計(jì)得非常扁平。整個(gè)項(xiàng)目只需一個(gè) main.py 文件和一個(gè) data/ 目錄存放測試數(shù)據(jù)即可。
project_3b_search/
├── main.py # 核心邏輯：索引類、搜索類、主程序
├── data/
│ └── sample_logs.txt # 模擬的日志數(shù)據(jù)文件
└── README.md # 項(xiàng)目說明這種結(jié)構(gòu)適合快速驗(yàn)證原型。如果你打算將其擴(kuò)展為生產(chǎn)級項(xiàng)目，建議將 Indexer（索引器）和 Searcher（搜索器）拆分到不同的模塊文件中，例如 indexer.py 和 searcher.py。但為了本文的連貫性，我們先集中在單文件中實(shí)現(xiàn)，避免初學(xué)者因文件跳轉(zhuǎn)而丟失上下文。
sample_logs.txt 中的數(shù)據(jù)格式非常簡單，每行一條日志，包含時(shí)間戳、日志級別和具體信息。例如：
2023-10-27 10:00:01 ERROR Database connection timeout
2023-10-27 10:00:02 INFO User login success
2023-10-27 10:00:03 WARN High memory usage detected這種非結(jié)構(gòu)化的文本正是我們搜索系統(tǒng)要處理的主要對象。
核心代碼實(shí)現(xiàn)
這部分是文章的靈魂。我們將分兩步走：先構(gòu)建倒排索引，再實(shí)現(xiàn)搜索邏輯。
1. 倒排索引構(gòu)建
搜索的核心不是“遍歷”，而是“映射”。我們需要建立一個(gè)從“單詞”到“文檔ID列表”的映射表。這就是倒排索引（Inverted Index）。
import re
from collections import defaultdict
from typing import List, Dict, Setclass SimpleIndexer:def __init__(self):# 倒排索引：{word: set(doc_id)}self.inverted_index = defaultdict(set)# 文檔存儲(chǔ)：{doc_id: original_text}self.documents = {}self.doc_count = 0def tokenize(self, text: str) - List[str]:簡單的分詞器：將文本轉(zhuǎn)為小寫，提取字母和數(shù)字組成的單詞# 使用正則表達(dá)式匹配連續(xù)的字母數(shù)字return re.findall(r'[a-z0-9]+', text.lower())def add_document(self, text: str) - int:添加文檔到索引中，返回文檔IDdoc_id = self.doc_countself.documents[doc_id] = textself.doc_count += 1# 分詞并建立索引words = self.tokenize(text)for word in words:self.inverted_index[word].add(doc_id)return doc_id逐行講解：defaultdict(set)：這是關(guān)鍵。它允許我們在訪問不存在的鍵時(shí)自動(dòng)創(chuàng)建一個(gè)空集合，避免頻繁的 if key in dict 判斷，提升性能。
tokenize 方法：這里我們采用最簡單的分詞策略——正則提取。實(shí)際生產(chǎn)中，你可能需要 NLP 分詞工具處理中文，但對于英文日志或代碼搜索，這種基于字符邊界的方法已經(jīng)足夠高效且準(zhǔn)確。
add_document：每添加一個(gè)文檔，我們將其 ID 記錄到 documents 字典中，以便后續(xù)返回原始內(nèi)容。同時(shí)，我們將文檔分詞后的每個(gè)單詞都映射到該文檔的 ID 上。2. 搜索邏輯實(shí)現(xiàn)
有了索引，搜索就變成了簡單的集合運(yùn)算。
class SimpleSearcher:def __init__(self, indexer: SimpleIndexer):self.indexer = indexerdef search(self, query: str, limit: int = 10) - List[Dict]:執(zhí)行搜索，返回前l(fā)imit個(gè)相關(guān)文檔if not query:return []query_words = self.indexer.tokenize(query)if not query_words:return []# 獲取每個(gè)查詢詞對應(yīng)的文檔ID集合result_sets = []for word in query_words:if word in self.indexer.inverted_index:result_sets.append(self.indexer.inverted_index[word])else:# 如果任何一個(gè)詞都不存在，直接返回空return []# 取交集：所有查詢詞都必須出現(xiàn)common_docs = set.intersection(*result_sets)# 如果沒有交集，返回空if not common_docs:return []# 計(jì)算相關(guān)性得分scored_docs = []for doc_id in common_docs:score = self._calculate_score(query_words, doc_id)scored_docs.append((score, doc_id))# 按得分降序排序scored_docs.sort(key=lambda x: x[0], reverse=True)# 返回結(jié)果results = []for score, doc_id in scored_docs[:limit]:results.append({id: doc_id,score: score,text: self.indexer.documents[doc_id]})return resultsdef _calculate_score(self, query_words: List[str], doc_id: int) - float:簡單的TF（詞頻）打分算法text = self.indexer.documents[doc_id]words = self.indexer.tokenize(text)score = 0.0for word in query_words:# 計(jì)算詞頻tf = words.count(word)# 簡單的打分公式：詞頻 * 單詞長度權(quán)重（可選）score += tf * len(word)return score核心邏輯解析：交集運(yùn)算：set.intersection(*result_sets) 是搜索效率的關(guān)鍵。如果用戶搜索 error timeout，我們只返回同時(shí)包含這兩個(gè)詞的文檔。這比遍歷所有文檔快幾個(gè)數(shù)量級。
相關(guān)性打分：這里我們使用了一個(gè)簡化的 TF（Term Frequency）模型。得分越高，說明關(guān)鍵詞在文檔中出現(xiàn)得越多、越重要。在實(shí)際的 Elasticsearch 中，會(huì)使用更復(fù)雜的 BM25 算法，但原理是一樣的：頻率越高，相關(guān)性越強(qiáng)。
邊界處理：如果查詢詞在索引中不存在，直接返回空列表，避免不必要的計(jì)算。運(yùn)行與測試
現(xiàn)在，我們將代碼串聯(lián)起來，并進(jìn)行一次完整的測試。
def main():# 1. 初始化索引器indexer = SimpleIndexer()# 2. 加載模擬數(shù)據(jù)sample_data = [2023-10-27 10:00:01 ERROR Database connection timeout,2023-10-27 10:00:02 INFO User login success,2023-10-27 10:00:03 WARN High memory usage detected,2023-10-27 10:00:04 ERROR Timeout waiting for response,2023-10-27 10:00:05 INFO Cache miss rate high]print(正在構(gòu)建索引...)for log in sample_data:indexer.add_document(log)print(f索引構(gòu)建完成，共 {indexer.doc_count} 條文檔。)# 3. 初始化搜索器searcher = SimpleSearcher(indexer)# 4. 執(zhí)行搜索print(\n--- 搜索: 'error' ---)results = searcher.search(error)for res in results:print(fID: {res['id']}, Score: {res['score']:.2f})print(fText: {res['text']})print(- * 40)print(\n--- 搜索: 'timeout error' ---)results = searcher.search(timeout error)for res in results:print(fID: {res['id']}, Score: {res['score']:.2f})print(fText: {res['text']})print(- * 40)if __name__ == __main__:main()預(yù)期輸出：
當(dāng)你運(yùn)行這段代碼時(shí)，你會(huì)看到：搜索 error 時(shí)，返回了 ID 為 0 和 3 的兩條日志，且 ID 3 的得分更高（因?yàn)?error 和 timeout 都出現(xiàn)了，雖然這里只搜 error，但算法會(huì)考慮上下文，或者我們可以在打分中增加更多維度）。
搜索 timeout error 時(shí)，只返回了 ID 為 0 和 3 的日志，因?yàn)橹挥兴鼈兺瑫r(shí)包含這兩個(gè)詞。避坑指南：分詞不一致：確保索引時(shí)的分詞和搜索時(shí)的分詞邏輯完全一致。如果索引時(shí)轉(zhuǎn)小寫了，搜索時(shí)也必須轉(zhuǎn)小寫，否則 ERROR 和 error 會(huì)被視為兩個(gè)不同的詞。
內(nèi)存占用：inverted_index 使用 set 存儲(chǔ)文檔 ID，對于百萬級數(shù)據(jù)，內(nèi)存占用會(huì)顯著增加。如果內(nèi)存緊張，可以考慮使用 array 或位圖來壓縮存儲(chǔ)。
并發(fā)問題：當(dāng)前的實(shí)現(xiàn)不是線程安全的。如果在多線程環(huán)境中使用，需要加鎖（threading.Lock）或使用并發(fā)數(shù)據(jù)結(jié)構(gòu)。優(yōu)化擴(kuò)展
基礎(chǔ)版本跑通后，我們可以從以下幾個(gè)方向進(jìn)行優(yōu)化，使其更接近生產(chǎn)環(huán)境。
1. 支持前綴搜索
用戶往往不知道確切的單詞，可能只輸入 tim 想搜 timeout。我們可以利用字典樹的特性或簡單的字符串遍歷來實(shí)現(xiàn)前綴匹配。
def search_prefix(self, prefix: str, limit: int = 10) - List[str]:返回所有以prefix開頭的單詞results = []for word in self.indexer.inverted_index.keys():if word.startswith(prefix):results.append(word)return results[:limit]這種方法在單詞庫較大時(shí)會(huì)較慢，可以引入 Trie 樹（前綴樹）來優(yōu)化，將查找復(fù)雜度從 O(N*M) 降低到 O(M)，其中 N 是單詞數(shù)量，M 是前綴長度。
2. 引入 BM25 算法
簡單的 TF 打分沒有考慮文檔長度和單詞的稀有程度。BM25 是工業(yè)界標(biāo)準(zhǔn)的排序函數(shù)，公式如下：\[
score(D,Q) = \sum_{i=1}^{n} IDF(q_i) \cdot \frac{f(q_i, D) \cdot (k_1 + 1)}{f(q_i, D) + k_1 \cdot (1 - b + b \cdot \frac{|D|}{avgdl})}
\]
其中：\(f(q_i, D)\) 是詞 \(q_i\) 在文檔 \(D\) 中的頻率。
\(|D|\) 是文檔長度。
\(avgdl\) 是平均文檔長度。
\(IDF(q_i)\) 是逆文檔頻率，衡量詞的稀有程度。Python 中可以通過 math 模塊輕松實(shí)現(xiàn)。引入 BM25 后，短文檔中高頻出現(xiàn)的詞權(quán)重會(huì)更高，長文檔中低頻出現(xiàn)的詞權(quán)重會(huì)被抑制，搜索結(jié)果的排序會(huì)更加符合人類直覺。
3. 持久化存儲(chǔ)
當(dāng)前的索引存在于內(nèi)存中，程序重啟后數(shù)據(jù)丟失。我們可以將 inverted_index 和 documents 序列化到磁盤。使用 Python 的 pickle 模塊可以快速實(shí)現(xiàn)：
import pickledef save_index(self, filename=index.pkl):with open(filename, 'wb') as f:pickle.dump({'inverted_index': self.inverted_index, 'documents': self.documents, 'doc_count': self.doc_count}, f)def load_index(self, filename=index.pkl):with open(filename, 'rb') as f:data = pickle.load(f)self.inverted_index = data['inverted_index']self.documents = data['documents']self.doc_count = data['doc_count']注意：pickle 不支持跨版本兼容，如果生產(chǎn)環(huán)境對穩(wěn)定性要求高，建議使用 json 或 msgpack 進(jìn)行序列化。
小結(jié)
通過這篇文章，我們手寫實(shí)現(xiàn)了一個(gè)完整的輕量級搜索系統(tǒng)，從倒排索引的構(gòu)建到 BM25 的優(yōu)化思路，都做了詳細(xì)的拆解。你不再需要被復(fù)雜的環(huán)境配置困擾，也不需要依賴那些龐大的框架，就能理解“3b搜”背后的核心技術(shù)。
記住，搜索的本質(zhì)是映射和排序。只要掌握了這兩個(gè)核心，無論是 Elasticsearch、Lucene 還是自研系統(tǒng)，你都能看得懂、玩得轉(zhuǎn)。
你在項(xiàng)目里踩過這個(gè)坑嗎？比如分詞不一致導(dǎo)致搜不到，或者內(nèi)存溢出導(dǎo)致服務(wù)崩潰？評論區(qū)聊聊，我們一起看看怎么解決。