ql-link/linkrag/code-annotator

class SparseVectorService:
    """稀疏向量服务层。
    编排稀疏向量编码与产出规整，按 provider 复用本地或远程 BGE-M3 编码器，
    向上游提供与 dense 召回对仗的稀疏向量化能力，不直接维护 MySQL/Qdrant 状态。
    """

    async def vectorize_chunk(self, request: SparseChunkVectorizationRequest) -> SparseVector:
        """对单个 chunk 原文生成稀疏向量。
        Args:
            request: 待编码 chunk（含 chunk_id、content、bucket_id 等定位字段）
        Returns:
            indices 升序、values 一一对应的稀疏向量
        Raises:
            SparseVectorEncodingError: 编码失败或返回结构异常时抛出
            SparseVectorOutputError: 清洗后稀疏维度为空时抛出
        """

        # 现场过滤：只处理 dense 已成功且 sparse 尚未成功的 chunk（幂等、避免重复写）
        sparse_chunks = [
            c for c in chunks
            if c.dense_vector_status == CHUNK_STATUS_INDEXED
            and c.sparse_vector_status != SPARSE_VECTOR_STATUS_INDEXED
        ]
        # 复用同一套 lexical weights 清洗，保证本地/远程 provider 产出口径一致
        vectors = await self._encoder.aencode([c.content for c in sparse_chunks])

class BGEM3HttpSparseVectorEncoder:
    """调用远程 bge-m3-server 生成 sparse lexical weights 的编码器。
    与本地 BGEM3SparseVectorEncoder 实现同一 SparseVectorEncoderProtocol，
    上层编排无感切换；本类只负责 HTTP 调用与输出规整，不处理 MySQL/Qdrant 状态。
    """
    async def aencode(self, texts: Sequence[str]) -> list[SparseVector]:
        """调用远程 /encode 接口，把一批文本编码为稀疏向量。
        Args:
            texts: 待编码的 chunk 原文，返回向量与其一一同序。
        Returns:
            与输入同序的稀疏向量列表；输入为空时返回空列表。
        Raises:
            SparseVectorEncodingError: HTTP 调用失败或响应结构异常时抛出。
        """
        if not texts:
            return []
        # 只取 sparse，关闭 dense/colbert，降低远程计算与网络开销
        payload = {"texts": list(texts), "return_dense": False, "return_sparse": True}
        data = await self._post_encode(payload)
        # 远程返回的 sparse 必须与输入数量严格对齐，否则后续与 chunk 配对会错位
        sparse = data.get("sparse")
        if not isinstance(sparse, list) or len(sparse) != len(texts):
            raise SparseVectorEncodingError("bge-m3-server sparse 结构或数量不匹配")
        # 复用与本地推理同一套清洗规则，保证两种 provider 产出口径一致
        return [normalize_lexical_weights(w, top_k=self._top_k) for w in sparse]