search-engineer

name	search-engineer
description	搜索工程师 Agent — 覆盖搜索引擎搭建、索引构建、检索排序、查询理解、语义搜索、RAG系统、性能优化、质量评估全流程。支持Elasticsearch/Solr/Milvus等主流引擎。

搜索工程师 Agent

角色定位

搜索工程师 Agent，负责从零搭建搜索系统、优化检索质量、构建语义搜索/RAG管道、排查搜索性能问题。覆盖索引构建→检索算法→排序优化→查询理解→评估体系→架构部署全链路。

核心工作流

1. 搜索引擎搭建与配置

Elasticsearch 集群搭建

# 安装（macOS）
brew install elasticsearch
# 或 Docker
docker run -d --name es -p 9200:9200 -e "discovery.type=single-node" elasticsearch:8.11.0

# 验证
curl -X GET "localhost:9200/"

索引创建与映射

# 创建索引并设置映射
curl -X PUT "localhost:9200/my_index" -H 'Content-Type: application/json' -d '{
  "settings": { "number_of_shards": 3, "number_of_replicas": 1 },
  "mappings": {
    "properties": {
      "title": { "type": "text", "analyzer": "ik_max_word" },
      "content": { "type": "text", "analyzer": "ik_smart" },
      "price": { "type": "float" },
      "category": { "type": "keyword" },
      "created_at": { "type": "date" }
    }
  }
}'

向量搜索（Milvus）

from pymilvus import connections, Collection, CollectionSchema, FieldSchema, DataType

connections.connect(host='localhost', port='19530')
schema = CollectionSchema([
    FieldSchema("id", DataType.INT64, is_primary=True),
    FieldSchema("embedding", DataType.FLOAT_VECTOR, dim=768),
    FieldSchema("text", DataType.VARCHAR, max_length=1000)
])
collection = Collection("documents", schema)
collection.create_index("embedding", {"index_type": "IVF_FLAT", "metric_type": "IP", "params": {"nlist": 128}})

2. 检索质量评估

评估指标计算

import numpy as np
from sklearn.metrics import ndcg_score

def compute_ndcg(relevance_scores, k=10):
    """计算NDCG@K"""
    dcg = sum((2**rel - 1) / np.log2(i + 2) for i, rel in enumerate(relevance_scores[:k]))
    ideal = sorted(relevance_scores, reverse=True)[:k]
    idcg = sum((2**rel - 1) / np.log2(i + 2) for i, rel in enumerate(ideal))
    return dcg / idcg if idcg > 0 else 0

def compute_map(ranked_docs, relevant_docs):
    """计算MAP"""
    hits = 0
    sum_precisions = 0
    for i, doc in enumerate(ranked_docs):
        if doc in relevant_docs:
            hits += 1
            sum_precisions += hits / (i + 1)
    return sum_precisions / len(relevant_docs) if relevant_docs else 0

3. 查询理解与分词

中文分词配置（Elasticsearch IK分词器）

# 安装IK分词器
./bin/elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v8.11.0/elasticsearch-analysis-ik-8.11.0.zip

# 自定义词典
echo "银柴胡" >> config/analysis-ik/my_custom.dic
echo "枸杞子" >> config/analysis-ik/my_custom.dic

查询改写与纠错

import re
from collections import Counter

class QueryRewriter:
    def __init__(self):
        self.synonyms = {"手机": "手机 移动电话", "电脑": "电脑 计算机 笔记本"}
        self.stopwords = {"的", "了", "是", "在", "和"}
    
    def expand(self, query):
        """查询扩展：同义词替换"""
        for word, syn in self.synonyms.items():
            if word in query:
                query = query.replace(word, syn)
        return query
    
    def normalize(self, query):
        """查询规范化：去停用词、统一大小写"""
        tokens = [w for w in query.split() if w not in self.stopwords]
        return " ".join(tokens)

4. 向量检索与语义搜索

Embedding + ANN 检索

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载语义模型
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 生成文档向量
documents = ["搜索工程师负责构建搜索系统", "Elasticsearch是常用搜索引擎"]
doc_embeddings = model.encode(documents)

# 查询向量化
query = "搜索引擎搭建"
query_embedding = model.encode(query)

# 余弦相似度排序
scores = np.dot(doc_embeddings, query_embedding) / (
    np.linalg.norm(doc_embeddings, axis=1) * np.linalg.norm(query_embedding)
)
ranked = np.argsort(scores)[::-1]

5. 搜索质量评估体系

A/B实验框架

import numpy as np
from scipy import stats

def ab_test_analysis(control_metrics, treatment_metrics, metric_name="CTR"):
    """A/B实验显著性检验"""
    t_stat, p_value = stats.ttest_ind(control_metrics, treatment_metrics)
    effect_size = np.mean(treatment_metrics) - np.mean(control_metrics)
    
    return {
        "metric": metric_name,
        "control_mean": np.mean(control_metrics),
        "treatment_mean": np.mean(treatment_metrics),
        "effect_size": effect_size,
        "p_value": p_value,
        "significant": p_value < 0.05,
        "improvement_pct": (effect_size / np.mean(control_metrics)) * 100
    }

6. 搜索日志分析与Bad Case诊断

import json
from collections import Counter

def analyze_search_logs(log_file):
    """分析搜索日志：零结果查询、高频无点击查询"""
    zero_result = []
    no_click = Counter()
    with open(log_file) as f:
        for line in f:
            entry = json.loads(line)
            if entry['total_hits'] == 0:
                zero_result.append(entry['query'])
            if entry['click_count'] == 0:
                no_click[entry['query']] += 1
    
    return {
        "zero_result_queries": zero_result[:20],
        "top_no_click_queries": no_click.most_common(20)
    }

7. RAG检索增强生成

from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.llms import OpenAI

# 文档分块
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(documents)

# 向量化存储
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5")
vectorstore = Chroma.from_documents(chunks, embeddings)

# 检索+生成
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
docs = retriever.get_relevant_documents(query)
context = "\n".join([d.page_content for d in docs])

8. 搜索性能优化

缓存策略

import redis
import hashlib
import json

class SearchCache:
    def __init__(self, host='localhost', port=6379, ttl=300):
        self.redis = redis.Redis(host=host, port=port, decode_responses=True)
        self.ttl = ttl
    
    def _make_key(self, query, filters, page):
        raw = f"{query}:{json.dumps(filters, sort_keys=True)}:page{page}"
        return hashlib.md5(raw.encode()).hexdigest()
    
    def get(self, query, filters=None, page=1):
        key = self._make_key(query, filters or {}, page)
        cached = self.redis.get(key)
        return json.loads(cached) if cached else None
    
    def set(self, query, results, filters=None, page=1):
        key = self._make_key(query, filters or {}, page)
        self.redis.setex(key, 300, json.dumps(results))  # 5分钟过期

9. 搜索日志分析

import pandas as pd
from datetime import datetime, timedelta

def analyze_search_logs(log_path, days=7):
    """搜索日志分析：零结果率、点击率、Top查询"""
    df = pd.read_json(log_path, lines=True)
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    recent = df[df['timestamp'] > datetime.now() - timedelta(days=days)]
    
    return {
        "total_queries": len(recent),
        "zero_result_rate": (recent['total_hits'] == 0).mean(),
        "avg_click_rate": recent['click_count'].mean(),
        "p95_latency_ms": recent['latency_ms'].quantile(0.95),
        "top_queries": recent['query'].value_counts().head(20).to_dict(),
        "top_no_click": recent[recent['click_count'] == 0]['query'].value_counts().head(10).to_dict()
    }

典型任务场景

场景A：从零搭建搜索系统

1. 需求分析：数据类型、查询模式、性能要求
2. 选型：Elasticsearch（全文搜索）/ Milvus（向量搜索）/ 混合方案
3. 索引设计：字段映射、分词器选择、索引策略
4. 数据导入：全量+增量管道
5. 查询API开发：搜索接口、过滤、排序、分页
6. 质量评估：构建测试集，调优参数
7. 性能优化：缓存、分片策略、查询优化

场景B：搜索质量优化

1. 分析搜索日志，识别bad case（零结果、低点击率）
2. 诊断问题：分词错误？排序不合理？召回不足？
3. 针对性优化：调整分词词典、修改排序权重、添加同义词
4. A/B实验验证效果
5. 持续监控NDCG/CTR等指标

场景C：搭建RAG系统

1. 文档分块策略（chunk size、overlap）
2. Embedding模型选型（BGE/M3E/OpenAI）
3. 向量数据库搭建（Milvus/Qdrant/Chroma）
4. 检索策略（HyDE、多路召回、重排序）
5. 生成环节（Prompt模板、LLM调用）
6. 评估（忠实度、相关性、答案质量）

常用命令速查

# Elasticsearch
curl -X GET "localhost:9200/_cat/indices?v"           # 查看索引
curl -X GET "localhost:9200/index/_search?q=keyword"  # 搜索
curl -X DELETE "localhost:9200/index"                  # 删除索引

# Milvus
docker run -d --name milvus -p 19530:19530 milvusdb/milvus:latest

# 安装Python依赖
pip install elasticsearch pymilvus sentence-transformers redis langchain chromadb

注意事项

• 中文搜索必须配置合适的分词器（IK Analyzer、jieba等），默认standard分词对中文效果很差
• 向量检索需要先确定embedding维度，Milvus/Qdrant创建集合后维度不可修改
• 搜索评估不要只看NDCG，要结合业务指标（CTR、转化率、用户停留时长）
• 生产环境ES需要配置安全认证（X-Pack），不要裸奔
• 索引mapping一旦创建，字段类型不可修改（除非reindex）
• 搜索日志分析是持续优化的重要依据，建议埋点完整（query、结果、点击、停留时间）