46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
import os
|
||
import json
|
||
from typing import List
|
||
import httpx
|
||
import requests
|
||
|
||
|
||
def get_embeddings(embedding_texts: List[str]):
|
||
base_url=os.getenv("OPENAI_API_EMBEDDING_BASE","http://172.18.30.122:40060/v1")
|
||
embedding_base_url = os.path.join(base_url, "embeddings")
|
||
with httpx.Client(timeout=60) as client:
|
||
response = client.post(
|
||
embedding_base_url,
|
||
json={"model": "bge-m3", "input": embedding_texts},
|
||
headers={"Authorization": os.getenv("OPENAI_API_KEY","EMPTY")},
|
||
)
|
||
embedding_list = response.json()["data"][0]["embedding"]
|
||
return embedding_list
|
||
# 1. 组装请求体（与 SearchRequest 一一对应）
|
||
|
||
|
||
payload = { # 若不提供 query，可改为传 query_vector
|
||
"top_k": 3,
|
||
"database_name": "XIAN",
|
||
"collection_name": "b9d47d5a6d66cbe2cd44a8ea4ad021fc",
|
||
"query_vector": get_embeddings(["和尚"]) # 如果本地已算好向量，可直接把 list[float] 写进来
|
||
}
|
||
|
||
# 2. 发送 POST 请求
|
||
url = "http://172.18.30.122:9079/search_3D"
|
||
headers = {"Content-Type": "application/json"}
|
||
|
||
resp = requests.post(url, data=json.dumps(payload), headers=headers, timeout=30)
|
||
|
||
# 3. 处理返回
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
if data.get("code") == 200:
|
||
print("检索成功：")
|
||
for item in data.get("data", []):
|
||
print("score:", item["score"])
|
||
print("text :", item["text"][:200], "...")
|
||
else:
|
||
print("业务逻辑异常：", data.get("message"))
|
||
else:
|
||
print("HTTP 异常：", resp.status_code, resp.text) |