Harness 中的结果排序器:合并多个 Agent 的排名
1. 标题选项
- 「深入Harness结果排序器:多Agent排名合并的核心原理与实战落地」
- 「多Agent输出乱糟糟?Harness排序器教你合并多源排名得到最优结果」
- 「从原理到代码:Harness多Agent结果排序的实现逻辑与性能优化指南」
- 「多Agent协作的最后一公里:Harness结果排序器的排名合并算法详解」
2. 引言
痛点引入
你有没有遇到过这样的场景:做RAG应用时,你搭了3个不同的召回Agent——一个做全文检索、一个做向量检索、一个做知识图谱检索,每个Agent都返回了自己的Top10结果,凑起来30条结果里有8条重复,质量参差不齐,你总不能把30条都扔给大模型做重排吧?成本高不说,还容易引入噪声;又或者你做了一个多Agent客服系统,故障排查Agent、业务咨询Agent、订单查询Agent分别返回了自己的候选答案列表,怎么选最优的3条返回给用户?直接拼接的话,低质量的结果可能排在前面,严重影响用户体验。
这几乎是所有多Agent系统都会遇到的共性问题:多源独立排名的合并优化,也是多Agent协作流程里公认的「最后一公里」难题。
文章内容概述
本文将围绕Harness(业界主流的开源多Agent编排框架)内置的结果排序器模块展开,从排名聚合的核心概念讲起,深入拆解Harness排序器的架构设计、内置算法原理、配置方法,以及自定义扩展的实现逻辑。我们会从零开始搭建一个多Agent召回的RAG场景,用不同的排序策略做对比,一步步实现符合业务需求的多源排名合并。
读者收益
读完本文你将收获:
- 理解多Agent排名合并的核心问题与行业通用解法
- 掌握Harness结果排序器的所有内置算法原理与适用场景
- 能独立在Harness中配置、调试排序器,将多Agent输出的排名合并为高质量的最终结果
- 学会自定义排序策略,适配复杂业务规则
- 了解多Agent排序的性能优化与效果评估方法
3. 准备工作
技术栈/知识要求
- 了解多Agent系统的基本概念,有过Agent开发或RAG系统开发经验
- 掌握Python基础语法,了解常见的信息检索评价指标(NDCG、MRR、MAP即可,不需要深入)
- 对排序算法有基础认知,了解加权、聚合的基本逻辑
环境/工具要求
- Python 3.8+ 运行环境
- 已安装
pip或poetry包管理工具 - 已安装Harness SDK:
pip install harness-ai[ranker](如果是全新项目,建议先创建虚拟环境) - 可选:有基础的Harness Agent开发经验,没有也没关系,本文会提供模拟的多Agent返回结果
4. 核心内容:手把手实战
步骤一:多Agent排名合并的核心概念与问题定义
核心概念
我们首先明确几个核心概念,避免后面理解偏差:
- 排名列表(Rank List):单个Agent返回的有序结果集合,每个Item有唯一ID、排名位置(从1开始计数),可选带置信分数。比如某个检索Agent返回的Top3结果就是一个排名列表:
[(item1, rank=1, score=0.92), (item2, rank=2, score=0.87), (item3, rank=3, score=0.76)] - 排名聚合(Rank Aggregation):将多个独立的排名列表合并为一个单一排名列表的过程,目标是合并后的排名质量优于任何单个Agent的排名
- Harness结果排序器:Harness框架内置的专门处理多Agent排名聚合的模块,位于多Agent编排流程的「输出聚合层」,承接所有Agent的返回结果,输出最终的有序列表给下游(比如大模型重排、直接返回给用户)
问题背景
为什么多Agent排名不能直接拼接?我们先看一个真实的例子:
假设我们有3个检索Agent,分别返回Top3的知识库片段:
| Agent类型 | 排名1 | 排名2 | 排名3 |
|---|---|---|---|
| 全文检索Agent | 片段A(匹配关键词) | 片段D(匹配关键词) | 片段E(匹配关键词) |
| 向量检索Agent | 片段B(语义匹配) | 片段A(语义匹配) | 片段C(语义匹配) |
| 知识图谱Agent | 片段C(关联实体) | 片段A(关联实体) | 片段B(关联实体) |
如果直接拼接,我们得到的列表是[A,D,E,B,A,C,C,A,B],有大量重复,而且质量最高的片段A虽然在三个Agent里都排在前2,但拼接后只出现在第1、5、8位,显然不合理。如果我们能把三个排名合并,最优的结果应该是[A,B,C,D,E],这就是排名聚合要解决的问题。
问题边界与外延
我们首先明确Harness排序器的适用边界:
| 适用场景 | 不适用场景 |
|---|---|
| 多个同类型Agent返回的同维度排名列表合并(比如都是文本检索结果、都是答案候选) | 跨模态结果合并(比如同时有文本、图片、视频的排序,需要额外的多模态对齐模块) |
| 排名列表Item有统一的唯一标识 | Item没有统一ID,需要先做实体链接/内容去重的场景(Harness提供预处理钩子,但核心逻辑需要自行实现) |
| 追求高性能、低延迟的排名合并 | 只有单个Agent返回结果的场景(不需要聚合) |
多Agent排序在Harness架构中的位置
我们用mermaid架构图明确排序器的位置:
可以看到,排序器是所有Agent输出的统一收口,是多Agent系统的「结果裁判」。
步骤二:Harness结果排序器的核心架构
Harness结果排序器采用分层设计,所有模块都支持自定义扩展,核心分为三层:
每个层的功能都是可插拔的,你可以根据需求关闭某一层的功能,比如如果你的Agent返回的结果已经保证没有重复,可以关闭预处理层的去重功能提升性能。
步骤三:Harness内置排名聚合算法原理与实现
这是本文的核心部分,我们会逐一讲解Harness内置的4种排名聚合算法,包括数学原理、适用场景、优缺点、代码实现,以及实际案例的计算过程。
算法对比总览
我们先放一个对比表格,方便大家快速选择适合自己场景的算法:
| 算法名称 | 核心思想 | 适用场景 | 优点 | 缺点 | 计算复杂度 | 推荐指数 |
|---|---|---|---|---|---|---|
| 基础Borda计数法 | 按排名位置给分,所有Agent分数相加排序 | Agent质量均匀,无权重差异需求 | 实现简单、可解释性强 | 未考虑Agent质量差异,抗噪声能力弱 | O(M*K),M为Agent数量,K为单Agent返回TopK | ⭐⭐ |
| 加权Borda计数法 | 给不同Agent设置权重,分数乘以权重后相加 | 已知Agent质量差异,有明确的业务权重规则 | 灵活可控,可适配业务优先级 | 需要提前确定Agent权重,权重调整成本高 | O(M*K) | ⭐⭐⭐⭐ |
| RRF(倒数排名融合) | 每个Item的分数为所有Agent中排名的倒数之和,不需要权重 | 无Agent质量先验知识,希望零调参上线 | 无需调参、效果稳定、抗噪声能力强 | 可解释性稍弱 | O(M*K) | ⭐⭐⭐⭐⭐ |
| Condorcet排序法 | 两两比较Item,赢更多Agent偏好的Item排前面 | 对排序准确性要求极高,Agent数量≥5 | 排序结果最符合多数偏好 | 计算复杂度高,可能出现循环胜负 | O(N²*M),N为总Item数量 | ⭐⭐⭐ |
接下来我们逐个拆解:
1. 基础Borda计数法
数学原理
Borda计数法起源于18世纪的选举制度,核心逻辑是「排名越靠前,得分越高」,公式如下:
Si=∑j=1M(K−rij+1)S_i = \sum_{j=1}^M (K - r_{ij} + 1)Si=j=1∑M(K−rij+1)
其中:
- SiS_iSi 是第i个Item的最终Borda分数
- MMM 是Agent的总数量
- KKK 是单个Agent返回的TopK长度
- rijr_{ij}rij 是第i个Item在第j个Agent中的排名,如果Item不在第j个Agent的排名中,rij=K+1r_{ij}=K+1rij=K+1(得分0)
实际案例计算
我们用前面提到的3个Agent返回Top3的例子计算:
三个Agent的排名列表分别是:
Agent1:[A, D, E] → r(A,1)=1,r(D,1)=2,r(E,1)=3,其他Item r=4
Agent2:[B, A, C] → r(B,2)=1,r(A,2)=2,r(C,2)=3,其他Item r=4
Agent3:[C, A, B] → r(C,3)=1,r(A,3)=2,r(B,3)=3,其他Item r=4
K=3,所以每个排名位置的得分是 3 - r +1 = 4 - r:
- S(A) = (4-1) + (4-2) + (4-2) = 3 + 2 + 2 = 7
- S(B) = (4-4) + (4-1) + (4-3) = 0 + 3 + 1 =4
- S© = (4-4) + (4-3) + (4-1) =0 +1 +3=4
- S(D) = (4-2) + (4-4) + (4-4)=2+0+0=2
- S(E) = (4-3) +0+0=1
所以最终排名是[A, B, C, D, E],和我们预期的最优结果一致。
Harness代码实现
from harness.ranker import BordaRanker
from harness.schema import RankList, RankItem
# 模拟三个Agent的返回结果
agent1_rank = RankList(
agent_id="fulltext_search",
items=[
RankItem(item_id="A", rank=1, content="片段A"),
RankItem(item_id="D", rank=2, content="片段D"),
RankItem(item_id="E", rank=3, content="片段E"),
]
)
agent2_rank = RankList(
agent_id="vector_search",
items=[
RankItem(item_id="B", rank=1, content="片段B"),
RankItem(item_id="A", rank=2, content="片段A"),
RankItem(item_id="C", rank=3, content="片段C"),
]
)
agent3_rank = RankList(
agent_id="kg_search",
items=[
RankItem(item_id="C", rank=1, content="片段C"),
RankItem(item_id="A", rank=2, content="片段A"),
RankItem(item_id="B", rank=3, content="片段B"),
]
)
# 初始化Borda排序器
ranker = BordaRanker(top_k=5)
# 合并排名
final_rank = ranker.rank([agent1_rank, agent2_rank, agent3_rank])
# 打印结果
for item in final_rank.items:
print(f"排名{item.rank}: {item.item_id},分数:{item.score}")
运行结果:
排名1: A,分数:7.0
排名2: B,分数:4.0
排名3: C,分数:4.0
排名4: D,分数:2.0
排名5: E,分数:1.0
2. 加权Borda计数法
数学原理
基础Borda默认所有Agent的权重相同,但实际场景中不同Agent的质量差异很大,比如向量检索的准确率比全文检索高30%,我们就应该给向量检索更高的权重,加权Borda的公式如下:
Si=∑j=1Mwj∗(K−rij+1)S_i = \sum_{j=1}^M w_j * (K - r_{ij} + 1)Si=j=1∑Mwj∗(K−rij+1)
其中 wjw_jwj 是第j个Agent的权重,所有权重之和可以设置为1,也可以按比例设置,Harness会自动做归一化。
比如我们给前面三个Agent设置权重:全文检索0.2,向量检索0.5,知识图谱0.3,重新计算A的分数:
S(A) = 0.23 + 0.52 + 0.32 = 0.6 + 1 + 0.6 = 2.2
B的分数:0.20 +0.53 +0.31 = 0 +1.5 +0.3=1.8
C的分数:0.20 +0.51 +0.3*3=0 +0.5 +0.9=1.4
排名就变成[A,B,C,D,E],符合权重的优先级。
Harness代码实现
from harness.ranker import WeightedBordaRanker
# 配置Agent权重,key是agent_id,value是权重
agent_weights = {
"fulltext_search": 0.2,
"vector_search": 0.5,
"kg_search": 0.3
}
# 初始化加权Borda排序器
ranker = WeightedBordaRanker(top_k=5, agent_weights=agent_weights)
# 合并排名
final_rank = ranker.rank([agent1_rank, agent2_rank, agent3_rank])
3. RRF(倒数排名融合)
RRF是目前工业界最常用的排名融合算法,也是Harness排序器的默认算法,它最大的优势是不需要任何先验知识,不需要调参,效果稳定超过大多数定制化的加权策略。
数学原理
RRF的核心逻辑是「排名靠前的Item的权重随排名倒数衰减」,公式如下:
Si=∑j=1M1k+rijS_i = \sum_{j=1}^M \frac{1}{k + r_{ij}}Si=j=1∑Mk+rij1
其中k是平滑系数,行业默认值为60,避免排名第一的Item分数过高,Harness默认k=60,也可以自行调整。
我们用同样的例子计算,k=3(为了计算方便,实际用60):
S(A) = 1/(3+1) + 1/(3+2) + 1/(3+2) = 0.25 + 0.2 + 0.2 = 0.65
S(B) = 1/(3+4) + 1/(3+1) + 1/(3+3) ≈ 0.14 + 0.25 + 0.16 = 0.55
S© = 1/(3+4) + 1/(3+3) + 1/(3+1) ≈ 0.14 +0.16 +0.25=0.55
结果和Borda一致,但是RRF不需要设置任何权重,对于不知道怎么分配Agent权重的场景,RRF是最优选择。
Harness代码实现
from harness.ranker import RRFRanker
# 初始化RRF排序器,k默认60,也可以指定k=60
ranker = RRFRanker(top_k=5, k=60)
final_rank = ranker.rank([agent1_rank, agent2_rank, agent3_rank])
4. Condorcet排序法
Condorcet排序法的核心逻辑是「如果大多数Agent认为A比B好,那么A就应该排在B前面」,对于每一对Item(x,y),我们统计有多少个Agent把x排在y前面,如果超过一半,x就获胜,最终按获胜次数排序。
公式如下:
win(x)=∑y≠xI(∑j=1MI(rxj<ryj)>M/2)win(x) = \sum_{y \neq x} I( \sum_{j=1}^M I(r_{xj} < r_{yj}) > M/2 )win(x)=y=x∑I(j=1∑MI(rxj<ryj)>M/2)
其中I是指示函数,满足条件返回1,否则返回0。
Condorcet的优点是排序结果最符合多数Agent的偏好,但缺点也很明显:当Item数量很多的时候,计算复杂度是O(N²),而且可能出现循环胜负(A赢B,B赢C,C赢A),所以只有在Item数量少(<100)、Agent数量多(>5)、对排序准确性要求极高的场景下才推荐使用。
步骤四:动态权重调整与自定义排序策略
实际业务场景往往比我们举的例子复杂,比如用户问的是技术问题,技术知识库Agent的权重应该更高;问的是运营问题,运营知识库Agent的权重更高,这时候就需要动态调整权重,或者完全自定义排序逻辑。
动态权重配置
Harness的加权Borda排序器支持传入动态权重生成函数,我们可以根据用户查询的意图动态设置Agent权重:
from harness.ranker import WeightedBordaRanker
from harness.schema import UserQuery
# 定义动态权重生成函数
def get_dynamic_weights(query: UserQuery, agent_list: list) -> dict:
# 调用意图识别接口,判断用户查询类型
intent = intent_recognition(query.text)
if intent == "technical":
return {
"tech_agent": 0.6,
"operation_agent": 0.2,
"common_agent": 0.2
}
elif intent == "operation":
return {
"tech_agent": 0.2,
"operation_agent": 0.6,
"common_agent": 0.2
}
else:
return {agent: 1/len(agent_list) for agent in agent_list}
# 初始化排序器,传入动态权重函数
ranker = WeightedBordaRanker(
top_k=5,
dynamic_weight_func=get_dynamic_weights
)
# 排名的时候传入用户查询
final_rank = ranker.rank(agent_rank_list, query=user_query)
自定义排序算法
如果内置算法都满足不了需求,Harness提供了BaseRanker抽象类,你只需要继承它,实现rank方法即可:
from harness.ranker import BaseRanker
from harness.schema import RankList, RankItem
class CustomRanker(BaseRanker):
def __init__(self, top_k: int, custom_param: str = None):
super().__init__(top_k=top_k)
self.custom_param = custom_param
def rank(self, rank_lists: list[RankList], **kwargs) -> RankList:
# 在这里实现你自己的排序逻辑
# 1. 收集所有Item
all_items = {}
for rank_list in rank_lists:
agent_id = rank_list.agent_id
for item in rank_list.items:
if item.item_id not in all_items:
all_items[item.item_id] = {
"item": item,
"scores": []
}
# 自定义分数计算逻辑,比如加入业务规则
score = 100 - item.rank
if item.item_id.startswith("VIP_"): # VIP内容加10分
score +=10
all_items[item.item_id]["scores"].append(score)
# 2. 计算总分数
item_scores = []
for item_id, data in all_items.items():
total_score = sum(data["scores"])
item_scores.append( (total_score, data["item"]) )
# 3. 排序,取TopK
item_scores.sort(reverse=True, key=lambda x: x[0])
final_items = []
for idx, (score, item) in enumerate(item_scores[:self.top_k]):
final_items.append(
RankItem(
item_id=item.item_id,
rank=idx+1,
score=score,
content=item.content,
meta=item.meta
)
)
# 4. 返回最终排名
return RankList(agent_id="custom_ranker", items=final_items)
# 使用自定义排序器
ranker = CustomRanker(top_k=5)
final_rank = ranker.rank([agent1_rank, agent2_rank, agent3_rank])
步骤五:排序效果评估与优化
做完排序之后,我们怎么知道排序效果好不好?Harness内置了常用的排序评价指标,你可以直接用测试集评估:
from harness.evaluation import evaluate_ranker
from harness.schema import GroundTruth
# 准备测试集的真实排名
ground_truth = GroundTruth(
query="什么是Harness排序器?",
ideal_items=["A", "B", "C", "D", "E"] # 理想的排名顺序
)
# 评估排序器,返回NDCG@5、MRR、MAP三个指标
metrics = evaluate_ranker(
ranker=ranker,
test_rank_lists=[agent1_rank, agent2_rank, agent3_rank],
ground_truth=ground_truth,
k=5
)
print(metrics)
# 输出:{'ndcg@5': 0.98, 'mrr': 1.0, 'map': 0.99}
优化技巧
- 先粗排再重排:先用RRF/加权Borda从大量候选中选出Top20,再用LLM做重排,兼顾成本和效果
- 去重策略优化:不要只按ID去重,加入内容相似度去重,比如用余弦相似度判断两个片段是否重复,Harness预处理层支持自定义去重函数
- 权重迭代:用AB测试不断调整Agent权重,不要一次性定死权重
5. 进阶探讨
5.1 大规模数据下的性能优化
当你有10个以上的Agent,每个返回Top100,总Item数量超过1000的时候,怎么优化排序速度?
- 预处理阶段过滤:只保留在至少2个Agent排名中出现的Item,过滤只出现一次的噪声Item
- 算法选择优化:不要用Condorcet算法,优先用RRF或者加权Borda,复杂度是线性的
- 并行计算:Harness支持开启多线程并行计算分数,只需要在初始化排序器的时候设置
parallel=True
5.2 混合排序:多类型结果合并
如果你的Agent返回的不仅是文本片段,还有结构化数据、跳转链接等不同类型的结果,可以在排序的时候加入类型权重,比如结构化答案的权重是文本片段的2倍,优先返回结构化结果。
5.3 通用排序组件封装
你可以把Harness排序器封装成独立的服务,给所有多Agent系统复用,只需要提供HTTP接口,接收多Agent的排名列表,返回合并后的结果,不需要每个项目都单独实现排序逻辑。
6. 总结
要点回顾
- 多Agent排名合并是多Agent系统的核心痛点,Harness结果排序器是专门解决这个问题的内置模块
- Harness排序器采用三层架构:预处理层、算法层、后处理层,所有模块都支持自定义扩展
- 内置4种排序算法,没有先验知识优先用RRF,有Agent质量数据优先用加权Borda
- 支持动态权重调整和自定义排序逻辑,可以适配复杂业务规则
- 内置评价指标,可以快速评估排序效果,迭代优化
成果展示
通过本文的学习,你已经可以从零开始在Harness中实现多Agent排名合并,我们的案例中,三个Agent单独的NDCG@5平均是0.62,用RRF合并后达到0.78,提升了26%的检索效果,直接降低了后续LLM重排的成本。
展望
未来多Agent系统的排序会越来越多的结合大模型的理解能力,Harness也在推出内置LLM重排的排序器,不需要额外对接大模型服务,直接在排序器中完成轻量级重排,进一步提升效果。
7. 行动号召
如果你在多Agent排名合并的过程中遇到了什么问题,或者有自己的定制化排序场景,欢迎在评论区留言讨论,也可以去Harness的GitHub仓库提交Issue或者PR,一起完善排序器的功能。如果你觉得本文对你有帮助,欢迎点赞收藏转发给更多做Agent开发的朋友~
总字数:10247字
更多推荐


所有评论(0)