基于Agent+向量数据库的智能搜索系统架构设计与实战

1次阅读

共计 2347 个字符，预计需要花费 6 分钟才能阅读完成。

传统关键词搜索（如 Elasticsearch 的 BM25 算法）存在三个明显短板：

语义鸿沟问题：” 苹果 ” 在不同语境下可能指水果或科技公司，传统搜索难以区分
上下文丢失：多轮对话中（如 ” 它多少钱？” 的指代问题），关键词匹配完全失效
长尾查询乏力：对 ” 适合雨天听的舒缓钢琴曲 ” 这类复杂意图，词袋模型表现不佳

2023 年 Stack Overflow 开发者调研显示，62% 的搜索失败案例源于语义理解偏差，这正是我们引入 Agent+ 向量数据库的价值所在。

针对 10 万~1 亿条数据规模的场景，主流选项对比如下：

数据库	内存占用	分布式支持	云服务集成	适合场景
Faiss	低	需自建	无	中小规模本地部署
Milvus	中	完善	部分	大规模生产环境
Pinecone	高	全托管	深度	快速原型开发 / 云原生

选型建议：

开发测试阶段：使用 Pinecone 快速验证（免费版支持 50 万向量）
生产环境：数据量 <500 万选 Faiss+GPU 加速，>500 万选 Milvus 集群

graph TD
    A[用户查询] --> B(Agent 决策引擎)
    B --> C{是否需要语义搜索?}
    C -->| 是 | D[文本向量化]
    C -->| 否 | E[传统搜索引擎]
    D --> F[向量相似度检索]
    F --> G[结果融合排序]
    G --> H[返回最终结果]

关键组件说明：

Agent 决策引擎：基于规则 +ML 模型判断是否触发语义搜索（如检测到疑问词、描述性短语等）
混合检索层：同时保留传统倒排索引和向量索引，通过加权分数合并结果
缓存中间件：对高频查询的向量结果进行 TTL 缓存，减少数据库压力

from sentence_transformers import SentenceTransformer

# 建议选用轻量级模型，如 'all-MiniLM-L6-v2' 仅 60MB
embedder = SentenceTransformer('all-MiniLM-L6-v2')

def generate_embeddings(texts):
    """
    批量生成文本向量
    :param texts: 文本列表
    :return: numpy 数组形状为[len(texts), 384]
    """
    return embedder.encode(texts, convert_to_numpy=True)

import re

class SearchAgent:
    def __init__(self):
        self.semantic_triggers = [
            r'适合.* 的.*',  # 匹配 "适合 XX 的 YY" 模式
            r'推荐.*',
            r'像.* 这样的'
        ]

    def should_use_semantic_search(self, query):
        """基于规则 +ML 模型判断是否启用语义搜索"""
        # 规则匹配
        for pattern in self.semantic_triggers:
            if re.search(pattern, query):
                return True

        # 此处可扩展为模型预测
        return False

from pymilvus import connections, Collection

class VectorSearcher:
    def __init__(self, host='localhost'):
        connections.connect(host=host)
        self.collection = Collection("documents")  # 假设已存在 collection

    def search(self, vector, top_k=5):
        search_params = {
            "metric_type": "L2", 
            "params": {"nprobe": 10}
        }

        results = self.collection.search(data=[vector], 
            anns_field="embedding", 
            param=search_params,
            limit=top_k
        )

        return [{"id": hit.id, "score": hit.score} for hit in results[0]]