Memory Center 17｜BM25 如何把关键词匹配变成相关性排序

本文属于 Memory Center：AI 编程上下文与记忆系统工程实践，第 17 篇，位于“检索与证据组合”模块。案例背景统一采用实际开发项目；复现步骤使用脱敏教学样例，不代表特定生产事故。

从实际问题出发

实际开发项目经常按错误码、类名和配置键搜索。这些精确标识对关键词检索十分友好，但简单按出现次数排序容易让冗长文档压过更直接的答案。

从词项到相关性

BM25 将词项稀有程度、词频和文档长度结合起来。词频增长的收益逐渐饱和，常见词权重较低。当前实现为中文构造相邻字片段，并通过词项表限制候选，不直接加载整个语料。

BM25 如何把关键词匹配变成相关性排序：设计链路示意

图中展示的是本篇关注的职责和顺序。正常流程之外，还要结合下面的源码与实验检查失败分支；它不是对某次线上调用的运行时追踪。

有界候选影响结果边界

候选上限控制资源，但也可能影响召回。必须区分数据库匹配行数、唯一候选块数与最终结果数，不能把 LIMIT 理解成完整相关性保证。中文切分和代码标识符策略也需要真实样本评估。

用小语料理解 BM25

可将单词项贡献概括为：IDF(t) × tf × (k1 + 1) / (tf + k1 × (1 - b + b × dl / avgdl))。其中 tf 是词频，dl 是块长度，avgdl 是平均长度；参数和具体 IDF 形式应以实现为准。

文档甲有 100 个词，出现错误码两次；文档乙有 10000 个词，出现三次。简单次数排序偏向乙，长度归一化会抑制这种倾向。再加入一个所有文档都有的通用词，它不应像稀有错误码一样有区分力。此处数字只是教学输入，不是性能测试结果。

关键源码与阅读方法

源码基线为 Memory Center 6e90553（文档标注 5.0.0b24）。定位：memory_center/platform/retrieval.py，函数 search，起始行 54。

以下为源码节选，省略后续分支；不能作为独立脚本执行。

    def search(self, workspace, actor, query, mode='hybrid', limit=10):
        if not query.strip() or len(query) > 1000 or mode not in {'keyword', 'vector', 'hybrid'} or not 1 <= limit <= 50:
            raise CatalogError('QUERY_INVALID', '查询内容、模式或数量无效')
        workspace_id = workspace['workspace_id']
        try:
            index = self.store.get('index', workspace_id)
        except CatalogError:
            return {'state': 'NOT_INDEXED', 'items': [], 'negative_claim_allowed': False}
        if index['config_version'] != workspace['config_version']:
            return {'state': 'STALE', 'items': [], 'negative_claim_allowed': False}
        live = {item['id']: item for _, item in self.knowledge.active(workspace_id)}
        replaced = {item['replaces'] for item in live.values() if item.get('replaces')}

        def visible(row):
            metadata = json.loads(row['metadata'])
            if metadata.get('kind') != 'code':
                current = live.get(row['source_id'])
                if not current or current['revision'] != metadata['source_revision'] or current['id'] in replaced:
                    return False
                if current.get('expires_at') and datetime.fromisoformat(current['expires_at']) < datetime.now(timezone.utc):
                    return False
                if metadata.get('kind') == 'memory' and current['owner'] not in {actor.id, 'admin'}:
                    return False
            return True

        keyword_rows, vector_rows = [], []
        generation = index['generation']
        with self.store.db.connection() as connection:
            if mode != 'vector':
                terms = sorted(set(tokens(query)))[:40]
                if terms:
                    marks = ','.join('?' for _ in terms)
                    # 候选只来自倒排匹配；限定 2000，避免加载整个库。
                    rows = connection.execute(f'''SELECT c.*,t.term,t.frequency FROM chunks c JOIN chunk_terms t ON c.id=t.chunk_id
                        WHERE c.workspace_id=? AND c.generation=? AND t.term IN ({marks}) LIMIT 2000''',
                                              (workspace_id, generation, *terms)).fetchall()

阅读时先找输入由谁提供，再区分纯计算、读取和写入；遇到异常或提前返回，追问调用方会得到什么，以及已有状态是否仍然有效。上面的代码只证明展示范围内的实现，跨模块行为应继续查看调用处和相应测试。

用一个小实验核对设计

建议在独立临时目录和隔离数据中操作，使用虚构的接口、文档与项目名称。

准备长文档和短文档，分别控制关键词密度。
加入罕见错误码。
比较匹配与排名变化。

**要核对的结果：**观察排名原因，记录受候选上限影响的情况，不仅检查第一条是否非空。

可把观察记录为下表；没有执行的项写“未执行”，不能用设计预期冒充实测结果。

核对项	应保留的证据
使用条件	源码版本、配置、输入与项目范围
正常行为	实际返回、来源位置或状态变化
关键失败分支	错误码、可见结果与原状态是否保留
结论范围	已覆盖、未覆盖和需要进一步验证的条件

适用边界与继续阅读

本篇以静态源码与项目设计为依据，给出可执行的核对方法，未据此宣称已经完成新的性能或生产环境测试。不同版本的入口、默认值和底层实现可能变化，应先核对当前合同，再把方法用于自己的实际开发项目。

理解这一节后，可以回到专题总导航，沿同一模块继续阅读；顺序学习时使用专栏内的上一篇和下一篇。涉及具体技术选择，应连同本篇的代价与限制一起采用。

专题目录

BM25 如何把关键词匹配变成相关性排序