如何实现多REST API及数据库联合查询并按相关性排序结果？

这问题我之前帮团队解决过类似的，核心难点确实是统一不同数据源的相关性标准——毕竟每个数据库、API的评分逻辑天差地别，有的自带成熟的评分体系，有的完全没给任何相关性参考。给你一套实操的步骤，亲测好用：

1. 先给所有结果做「评分标准化」 #

不管数据源是直接访问的数据库还是REST API，你得把每个结果的相关性转换成0-1区间的统一分数（也可以用0-100，只要统一就行），这是排序的基础：

• 自带评分的数据库（比如Elasticsearch、Solr）：
  这类系统会返回_score之类的原生相关性分，但不同系统的分数范围差异极大（比如ES可能是0-10，有的老数据库可能是0-1000）。你需要做归一化处理，比如用min-max缩放：

  # 假设某数据库的_score范围是min_score到max_score
  normalized_score = (raw_score - min_score) / (max_score - min_score)
  

  如果不知道分数范围，也可以用百分位截断（比如把超过95分位的分数统一设为1），避免极端值影响。
• 无原生评分的数据库/REST API：
  得自己定义评分规则，比如：
  • 关键词出现在标题：加0.5分
  • 关键词出现在内容：加0.3分
  • 文档更新时间在30天内：加0.1分
  • 关键词匹配次数越多，额外加0.1-0.2分
    把这些规则加权求和，最终得到0-1的分数。如果是API返回的结果，你可能需要本地对返回的文本做关键词匹配计算。

2. 并行查询提升效率 #

别串行调用各个数据源，用异步请求同时发起查询（比如Python的aiohttp、JS的Promise.all），等所有结果都返回后再处理排序——这样能把总等待时间从「各数据源耗时之和」降到「最慢数据源的耗时」，用户体验好很多。

3. 合并结果并排序 #

把所有查询结果转换成统一的数据结构，每个结果至少包含：

• 内容主体
• 数据源标识
• 标准化后的相关性分数
• 次要排序字段（比如更新时间、数据源权重）

然后直接按「相关性分数降序」排序，如果分数相同，用次要字段兜底：比如优先展示更新时间更近的，或者你认为更权威的数据源结果。

举个简单的Python伪代码示例：

import asyncio
from aiohttp import ClientSession
from datetime import datetime

# 统一结果结构
class SearchResult:
    def __init__(self, content, source, relevance_score, update_time):
        self.content = content
        self.source = source
        self.relevance_score = relevance_score
        self.update_time = update_time

# 查询直接访问的数据库
async def query_internal_db(query):
    # 模拟从内部数据库获取结果，带原生_score
    raw_results = [
        {"content": "内部数据库结果1", "_score": 8.2, "update_time": "2024-05-20"},
        {"content": "内部数据库结果2", "_score": 5.7, "update_time": "2024-04-10"}
    ]
    # 归一化_score到0-1（假设该数据库_score范围0-10）
    return [
        SearchResult(
            content=res["content"],
            source="内部DB",
            relevance_score=res["_score"] / 10,
            update_time=datetime.fromisoformat(res["update_time"])
        ) for res in raw_results
    ]

# 查询REST API
async def query_external_api(session, query):
    # 模拟API返回结果，无原生评分
    async with session.get(f"https://api.example.com/search?q={query}") as resp:
        raw_results = await resp.json()
        normalized_results = []
        for res in raw_results:
            score = 0
            # 自定义评分规则
            if query.lower() in res["title"].lower():
                score += 0.6
            if query.lower() in res["content"].lower():
                score += 0.3
            # 时间权重
            days_old = (datetime.now() - datetime.fromisoformat(res["update_time"])).days
            score += 0.1 if days_old <= 30 else 0
            normalized_results.append(
                SearchResult(
                    content=res["content"],
                    source="外部API",
                    relevance_score=score,
                    update_time=datetime.fromisoformat(res["update_time"])
                )
            )
        return normalized_results

# 主逻辑：并行查询+合并排序
async def run_unified_search(query):
    async with ClientSession() as session:
        # 并行发起所有查询任务
        tasks = [query_internal_db(query), query_external_api(session, query)]
        all_results = await asyncio.gather(*tasks)
        # 合并所有结果
        merged = []
        for results in all_results:
            merged.extend(results)
        # 排序：先按相关性降序，再按更新时间降序
        sorted_results = sorted(
            merged,
            key=lambda x: (-x.relevance_score, -x.update_time.timestamp())
        )
        return sorted_results

# 调用示例
# asyncio.run(run_unified_search("你的搜索关键词"))

4. 边缘情况处理 #

• 某个数据源返回空结果：直接跳过即可，不影响合并排序。
• 无法归一化的评分：如果某个数据源的评分完全没有参考范围，可以给这类结果设置一个固定的权重系数（比如乘以0.8，降低其优先级），或者手动调整评分规则。
• 缓存优化：如果高频查询多，可以缓存热门关键词的合并排序结果，减少重复计算和API调用。

内容的提问来源于stack exchange，提问作者ragingprog