RagFlow_Plugs/test/test_fastapi_hybrid_search.py

#!/usr/bin/env python3
"""
测试混合检索FastAPI服务
"""

import requests
import json

# 测试数据
test_data = {
    "text_query": "测试",
    "image": "https://example.com/image.jpg",
    "topn": 2
}

# 发送POST请求
def test_hybrid_search():
    url = "http://localhost:18001/hybrid_search"
    headers = {
        "Content-Type": "application/json"
    }
    
    print("开始测试混合检索FastAPI服务...")
    print(f"请求URL: {url}")
    print(f"请求数据: {json.dumps(test_data, indent=2, ensure_ascii=False)}")
    
    try:
        # 发送POST请求
        response = requests.post(url, headers=headers, json=test_data, timeout=10)
        
        # 打印响应结果
        print(f"\n响应状态码: {response.status_code}")
        print(f"响应头: {dict(response.headers)}")
        print(f"响应内容: {response.text}")
        
        if response.status_code == 200:
            # 解析JSON响应
            response_data = response.json()
            print(f"\n解析后的响应数据: {json.dumps(response_data, indent=2, ensure_ascii=False)}")
            print("测试成功!")
        else:
            print(f"\n测试失败，状态码: {response.status_code}")
    except Exception as e:
        print(f"\n测试失败，请求异常: {str(e)}")

# 测试健康检查接口
def test_health_check():
    url = "http://localhost:18001/health"
    
    print("\n开始测试健康检查接口...")
    print(f"请求URL: {url}")
    
    try:
        # 发送GET请求
        response = requests.get(url, timeout=5)
        
        # 打印响应结果
        print(f"\n响应状态码: {response.status_code}")
        print(f"响应头: {dict(response.headers)}")
        print(f"响应内容: {response.text}")
        
        if response.status_code == 200:
            # 解析JSON响应
            response_data = response.json()
            print(f"\n解析后的响应数据: {json.dumps(response_data, indent=2, ensure_ascii=False)}")
            print("健康检查测试成功!")
        else:
            print(f"\n健康检查测试失败，状态码: {response.status_code}")
    except Exception as e:
        print(f"\n健康检查测试失败，请求异常: {str(e)}")

if __name__ == "__main__":
    test_hybrid_search()
    test_health_check()