搜索技术：Elasticsearch不够用了之后

作者: 醉月思发布日期: 2021-11-23 分类: 技术实践

这次做搜索系统改造，从数据库搜索到 Elasticsearch，再到向量搜索，。

最初的问题

数据库搜索

-- LIKE 搜索（慢）
SELECT * FROM products 
WHERE name LIKE '%laptop%' 
  OR description LIKE '%laptop%';

-- 正则表达式（更慢）
SELECT * FROM products 
WHERE name REGEXP 'laptop|computer|notebook';

-- 问题：
-- 1. 性能差
-- 2. 功能有限
-- 3. 无法排序相关性
-- 4. 无法处理同义词

Elasticsearch

基础索引

from elasticsearch import Elasticsearch

# 连接 Elasticsearch
es = Elasticsearch('http://localhost:9200')

# 创建索引
index_name = 'products'
if not es.indices.exists(index=index_name):
    es.indices.create(
        index=index_name,
        body={
            'mappings': {
                'properties': {
                    'name': {
                        'type': 'text',
                        'fields': {
                            'keyword': {
                                'type': 'keyword'
                            }
                        }
                    },
                    'description': {
                        'type': 'text',
                        'analyzer': 'english'
                    },
                    'price': {
                        'type': 'float'
                    },
                    'category': {
                        'type': 'keyword'
                    },
                    'tags': {
                        'type': 'keyword'
                    }
                }
            }
        }
    )

# 索引文档
def index_product(product):
    es.index(
        index=index_name,
        id=product['id'],
        body=product
    )

# 批量索引
def bulk_index_products(products):
    actions = []
    for product in products:
        actions.append({
            '_index': index_name,
            '_id': product['id'],
            '_source': product
        })
    
    from elasticsearch.helpers import bulk
    bulk(es, actions)

全文搜索

# 基础搜索
def search_products(query):
    response = es.search(
        index=index_name,
        body={
            'query': {
                'multi_match': {
                    'query': query,
                    'fields': ['name', 'description'],
                    'type': 'best_fields'
                }
            },
            'size': 10
        }
    )
    
    return [hit['_source'] for hit in response['hits']['hits']]

# 高级搜索
def advanced_search(query, filters=None):
    must = []
    
    # 全文搜索
    must.append({
        'multi_match': {
            'query': query,
            'fields': ['name^2', 'description'],  # name 字段权重更高
            'type': 'best_fields'
        }
    })
    
    # 过滤条件
    if filters:
        filter_query = []
        
        if 'category' in filters:
            filter_query.append({
                'term': {'category': filters['category']}
            })
        
        if 'price_range' in filters:
            filter_query.append({
                'range': {
                    'price': {
                        'gte': filters['price_range']['min'],
                        'lte': filters['price_range']['max']
                    }
                }
            })
        
        if filter_query:
            must.append({'bool': {'filter': filter_query}})
    
    response = es.search(
        index=index_name,
        body={
            'query': {
                'bool': {
                    'must': must
                }
            },
            'size': 20
        }
    )
    
    return [hit['_source'] for hit in response['hits']['hits']]

分词和同义词

# 配置分析器
def setup_analyzer():
    # 更新索引设置
    es.indices.put_settings(
        index=index_name,
        body={
            'settings': {
                'analysis': {
                    'filter': {
                        'my_synonym_filter': {
                            'type': 'synonym',
                            'synonyms': [
                                'laptop,notebook,computer',
                                'phone,mobile,smartphone'
                            ]
                        }
                    },
                    'analyzer': {
                        'my_analyzer': {
                            'tokenizer': 'standard',
                            'filter': [
                                'lowercase',
                                'my_synonym_filter'
                            ]
                        }
                    }
                }
            }
        }
    )

# 使用自定义分析器
es.indices.create(
    index=index_name,
    body={
        'mappings': {
            'properties': {
                'name': {
                    'type': 'text',
                    'analyzer': 'my_analyzer'
                }
            }
        }
    }
)

向量搜索

文本向量化

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载预训练模型
model = SentenceTransformer('all-MiniLM-L6-v2')

# 文本向量化
def text_to_vector(text):
    return model.encode(text)

# 创建文档向量
def create_document_vectors():
    documents = [
        {'id': 1, 'text': 'Laptop with high performance'},
        {'id': 2, 'text': 'Smartphone with great camera'},
        {'id': 3, 'text': 'Notebook for business use'}
    ]
    
    vectors = []
    for doc in documents:
        vector = text_to_vector(doc['text'])
        vectors.append({
            'id': doc['id'],
            'vector': vector,
            'text': doc['text']
        })
    
    return vectors

# 向量搜索
def vector_search(query, documents, top_k=5):
    query_vector = text_to_vector(query)
    
    # 计算相似度
    similarities = []
    for doc in documents:
        similarity = np.dot(query_vector, doc['vector']) / (
            np.linalg.norm(query_vector) * np.linalg.norm(doc['vector'])
        )
        similarities.append({
            'document': doc,
            'similarity': similarity
        })
    
    # 排序并返回结果
    similarities.sort(key=lambda x: x['similarity'], reverse=True)
    return similarities[:top_k]

向量数据库

import chromadb

# 创建向量数据库
chroma_client = chromadb.Client()

# 创建集合
collection = chroma_client.create_collection(name="documents")

# 添加文档
def add_documents_to_collection(documents):
    for doc in documents:
        collection.add(
            documents=[doc['text']],
            ids=[str(doc['id'])],
            metadatas=[{'category': doc.get('category', 'unknown')}]
        )

# 向量搜索
def search_in_collection(query, n_results=5):
    results = collection.query(
        query_texts=[query],
        n_results=n_results
    )
    
    return results['documents'][0]

# 使用示例
documents = [
    {'id': 1, 'text': 'Laptop with high performance', 'category': 'electronics'},
    {'id': 2, 'text': 'Smartphone with great camera', 'category': 'electronics'},
    {'id': 3, 'text': 'Notebook for business use', 'category': 'electronics'}
]

add_documents_to_collection(documents)

results = search_in_collection('high performance computer')
print(results)

混合搜索

结合全文搜索和向量搜索

def hybrid_search(query, filters=None):
    # 全文搜索
    fulltext_results = advanced_search(query, filters)
    
    # 向量搜索
    vector_results = search_in_collection(query, n_results=20)
    
    # 合并结果
    combined = {}
    
    # 添加全文搜索结果（权重 0.6）
    for i, result in enumerate(fulltext_results):
        score = (len(fulltext_results) - i) / len(fulltext_results) * 0.6
        
        if result['id'] not in combined:
            combined[result['id']] = {
                'document': result,
                'score': score
            }
        else:
            combined[result['id']]['score'] += score
    
    # 添加向量搜索结果（权重 0.4）
    for i, result in enumerate(vector_results):
        # 假设返回的是文档文本，需要查找完整文档
        # 这里简化处理
        score = (len(vector_results) - i) / len(vector_results) * 0.4
        
        if result['id'] not in combined:
            combined[result['id']] = {
                'document': result,
                'score': score
            }
        else:
            combined[result['id']]['score'] += score
    
    # 排序并返回结果
    sorted_results = sorted(
        combined.values(),
        key=lambda x: x['score'],
        reverse=True
    )
    
    return [result['document'] for result in sorted_results[:10]]

踩过的坑

坑一：分词不准确

分词不准确，导致搜索结果不理想。

解决：配置合适的分词器。

# 使用中文分词器
es.indices.create(
    index=index_name,
    body={
        'settings': {
            'analysis': {
                'analyzer': {
                    'my_analyzer': {
                        'tokenizer': 'ik_max_word',
                        'filter': ['lowercase']
                    }
                }
            }
        },
        'mappings': {
            'properties': {
                'name': {
                    'type': 'text',
                    'analyzer': 'my_analyzer'
                }
            }
        }
    }
)

坑二：相关性排序不理想

搜索结果相关性排序不理想。

解决：调整字段权重和使用相关性评分。

response = es.search(
    index=index_name,
    body={
        'query': {
            'multi_match': {
                'query': query,
                'fields': [
                    'name^3',     # name 字段权重最高
                    'description^2',
                    'tags^1'     # tags 字段权重最低
                ],
                'type': 'best_fields',
                'tie_breaker': 0.3
            }
        },
        'size': 10
    }
)

坑三：向量搜索性能差

向量搜索性能差，特别是大量数据。

解决：使用专门的向量数据库。

# 使用 Faiss 加速向量搜索
import faiss
import numpy as np

# 创建索引
d = 384  # 向量维度
index = faiss.IndexFlatL2(d)

# 添加向量
vectors = np.array([doc['vector'] for doc in documents])
index.add(vectors)

# 搜索
def faiss_search(query_vector, k=10):
    distances, indices = index.search(np.array([query_vector]), k)
    
    results = []
    for i, idx in enumerate(indices[0]):
        results.append({
            'document': documents[idx],
            'distance': distances[0][i]
        })
    
    return results

写在最后

搜索技术这东西，不只是技术，是用户体验。

解决了：

搜索性能
搜索质量
智能检索

带来了：

复杂度增加
存储成本
维护成本

选型之前先评估：

数据规模
搜索需求
团队能力
预算

Elasticsearch 适合：

传统全文搜索
结构化数据搜索
复杂的过滤和聚合

向量搜索适合：

语义搜索
相似度搜索
推荐系统

混合搜索适合：

需要关键词和语义搜索
高质量的搜索结果

不是所有场景都需要复杂的搜索技术，有时候简单的数据库搜索就够用。

这次搜索系统改造花了一个月，从数据库搜索到 Elasticsearch，再到向量搜索。改造完成后，搜索准确率提升了 40%，搜索响应时间从 500ms 降到 50ms。

版权声明: 本文首发于指尖魔法屋-搜索技术：Elasticsearch不够用了之后（https://blog.thinkmoon.cn/post/89-search-technology-elasticsearch-vector-search-practice/）转载或引用必须申明原指尖魔法屋来源及源地址！

利用 Cursor ACP 模式搭建飞书助手机器人，无需任何大模型接口

用 lark-cli 收飞书消息，Python 桥接 Cursor ACP（agent acp），项目代码里不集成任何大模型 API；适合已有 Cursor 订阅、想在本机跑 Agent 的场景。

📁 技术实践

📅 2026-07-07

用 GolemBot 把 Codex 接到飞书机器人

这篇文章记录一种少写代码的方案：用 GolemBot 接飞书机器人，用 Codex 做 Agent，再通过飞书 CLI Skills 读取文档、需求清单和 bug 清单。

📁 技术实践

📅 2026-06-09

调试 OpenAI Responses API 的 Web Search 踩坑记录

本文记录一次在投资研究 Agent 中调试 OpenAI Responses API Web Search 工具的过程，重点梳理请求格式、流式 SSE 解析、工具名兼容性和结构化数据源设计。

📁 技术实践

📅 2026-04-30

搜索技术：Elasticsearch不够用了之后

最初的问题

数据库搜索

Elasticsearch

基础索引

全文搜索

分词和同义词

向量搜索

文本向量化

向量数据库

混合搜索

结合全文搜索和向量搜索

踩过的坑

坑一：分词不准确

坑二：相关性排序不理想

坑三：向量搜索性能差

写在最后

评论

相关阅读