book_rag_server/src/api/search_infinity.py

# Infinity搜索API服务

from fastapi import FastAPI, HTTPException
from typing import List, Dict, Any, Optional
from src.api.db.services.infinity_search_service import InfinitySearchService
from src.utils.infinity import get_client


# 创建FastAPI应用

app = FastAPI(
    title="Infinity Search API",
    description="基于Infinity向量数据库的搜索API服务",
    version="1.0.0"
)

# 请求模型
from pydantic import BaseModel

class SearchRequest(BaseModel):
    """搜索请求模型"""
    search_query: Dict[str, Any]

# 1. 普通搜索接口
@app.post("/text", response_model=Dict[str, Any])
def search(request: SearchRequest):
    """
    普通搜索接口
    
    - **table_name**: 表名
    - **output_fields**: 要返回的字段列表
    - **query**: 查询条件，包含field、query和topn字段
    - **database_name**: 数据库名称（可选，默认使用客户端配置的数据库）
    """
    try:
        search_service = InfinitySearchService(infinity_client=get_client())
        result = search_service.search(request.search_query)
        return {"success": True, "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"搜索失败: {str(e)}")

# 2. 向量搜索接口
@app.post("/vector", response_model=Dict[str, Any])
def vector_search(request: SearchRequest):
    """
    向量搜索接口
    
    - **table_name**: 表名
    - **output_fields**: 要返回的字段列表
    - **query**: 查询条件，包含vector_field、query_vector和topn字段
    - **database_name**: 数据库名称（可选，默认使用客户端配置的数据库）
    """
    try:
        search_service = InfinitySearchService(infinity_client=get_client())
        result = search_service.vector_search(request.search_query)
        return {"success": True, "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"向量搜索失败: {str(e)}")

# 3. 混合搜索接口
@app.post("/hybrid", response_model=Dict[str, Any])
def hybrid_search(request: SearchRequest):
    """
    混合搜索接口
    
    - **table_name**: 表名
    - **output_fields**: 要返回的字段列表
    - **query**: 查询条件，包含vector_field、query_vector、field、query、topn和fusion_weight字段
    - **database_name**: 数据库名称（可选，默认使用客户端配置的数据库）
    """
    try:
        search_service = InfinitySearchService(infinity_client=get_client())
        result = search_service.hybrid_search(request.search_query)
        return {"success": True, "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"混合搜索失败: {str(e)}")