基于MIND数据集的 Java + Python + Milvus 的简单推荐系统实战

qq_46285362

813人浏览 · 2025-11-26 16:46:38

qq_46285362 · 2025-11-26 16:46:38 发布

前言

一、系统整体架构设计

在很多业务场景中，我们常常需要为用户推荐他们“可能感兴趣的内容”，比如新闻推荐、商品推荐、内容推荐等。与传统规则推荐相比，基于向量相似度的推荐系统具备更高的智能度与扩展性。

本文将完整介绍一个【简单但足够规范】的推荐系统实现方案，涵盖从系统架构设计到核心代码实现，帮助你快速搭建一个可运行的推荐系统原型。

二、推荐核心流程

该架构具备以下优势：

推荐算法与业务系统解耦
易于扩展和维护
支持模型升级和策略优化

三、Python推荐服务搭建（FastAPI)

为了让 Java 后台可以通过 HTTP 调用推荐算法，我们需要将推荐逻辑封装为一个 Python 服务。本项目采用 FastAPI 框架，具备高性能与自动生成接口文档的优势。

1. 安装依赖

pip install fastapi uvicorn pymilvus numpy

2. 创建服务入口文件app.py

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
from models.recommend_model import RecommendationEngine


app = FastAPI(title="Recommendation Service")
engine = RecommendationEngine()


class RecommendRequest(BaseModel):
    user_id: str
    history: List[str] = []
    clicked: List[str] = []
    top_k: int = 10


@app.post("/api/recommend")
def recommend(req: RecommendRequest):
    recs = engine.recommend(req.history, req.clicked, req.top_k)
    return {
        "user_id": req.user_id,
        "count": len(recs),
        "recommendations": recs
    }

3. 启动服务

在项目根目录运行：

uvicorn app:app --host 0.0.0.0 --port 9000

浏览器访问：

Swagger 文档： http://localhost:9000/docs
ReDoc 文档： http://localhost:9000/redoc

可直接在线测试推荐接口，极大提升调试效率。

四、用户兴趣向量构建

采用加权平均的方式构建用户向量：

用户向量 = 0.6 × 当前点击向量 + 0.4 × 历史向量

核心代码：

def build_user_vector(history_ids, clicked_ids, collection):
    clicked_vec = fetch_vectors(clicked_ids, collection)
    history_vec = fetch_vectors(history_ids, collection)


    if clicked_vec is not None and history_vec is not None:
        return (0.6 * clicked_vec + 0.4 * history_vec).tolist()
    elif clicked_vec is not None:
        return clicked_vec.tolist()
    elif history_vec is not None:
        return history_vec.tolist()
    return None

五、Milvus 相似度推荐

使用内积相似度检索与用户向量最接近的新闻向量：

results = collection.search(
    data=[user_vector],
    anns_field="embedding",
    param={"metric_type": "IP", "params": {"nprobe": 10}},
    limit=top_k
)

这一步构成推荐系统的“召回阶段”。

六、RecommendationEngine全部代码

import numpy as np
from pymilvus import connections, Collection


class RecommendationEngine:
    def __init__(self, host="192.168.74.130", port="19530", collection_name="news_collection"):
        connections.connect(host=host, port=port)
        self.collection = Collection(collection_name)
        self.collection.load()

    def _fetch_vectors(self, news_ids):
        if not news_ids:
            return None

        expr = f"news_id in {news_ids}"
        results = self.collection.query(expr, output_fields=["embedding"])

        if not results:
            return None

        vectors = [r["embedding"] for r in results]
        return np.mean(vectors, axis=0)

    def build_user_vector(self, history_ids, clicked_ids):
        clicked_vec = self._fetch_vectors(clicked_ids)
        history_vec = self._fetch_vectors(history_ids)

        if clicked_vec is not None and history_vec is not None:
            return (0.6 * clicked_vec + 0.4 * history_vec).tolist()
        elif clicked_vec is not None:
            return clicked_vec.tolist()
        elif history_vec is not None:
            return history_vec.tolist()
        else:
            return None

    def recommend(self, history_ids, clicked_ids, top_k=10):
        user_vector = self.build_user_vector(history_ids, clicked_ids)

        if user_vector is None:
            return []

        results = self.collection.search(
            data=[user_vector],
            anns_field="embedding",
            param={"metric_type": "IP", "params": {"nprobe": 10}},
            limit=top_k
        )

        recommendations = []
        for hit in results[0]:
            recommendations.append({
                "news_id": hit.id,
                "score": float(hit.distance)
            })

        return recommendations
recommend_engine = RecommendationEngine()