MGeo模型支持REST API吗？Flask封装服务部署实战教程-平芜编程栈

MGeo模型支持REST API吗？Flask封装服务部署实战教程

1. 为什么需要把MGeo变成API服务？

你刚跑通了MGeo的推理脚本，输入两个地址，它能准确判断相似度——这很酷。但问题来了：

业务系统是Java写的，没法直接调用Python脚本；
前端团队想在网页里实时查地址匹配结果，总不能让用户下载代码再本地运行；
运营同事每天要批量比对上万条地址，手动复制粘贴显然不现实。

这时候，一个能被任何语言、任何平台调用的HTTP接口，就成了刚需。
MGeo本身不带Web服务，但它完全支持封装成REST API——不需要改模型，不重写核心逻辑，只要加一层轻量级的Flask包装，就能把它变成“即插即用”的地址智能匹配服务。

本文就带你从零开始，把本地跑通的MGeo推理能力，变成一个稳定、可调用、带文档的API服务。整个过程不依赖Docker Compose编排、不涉及Kubernetes，单卡4090D环境直连部署，15分钟内完成上线。

2. 环境准备与服务基础结构

2.1 确认当前镜像环境（4090D单卡）

你已通过镜像部署完成初始化，系统中已预装：

CUDA 11.8 + cuDNN 8.6
Python 3.7（py37testmaas环境已就绪）
PyTorch 1.13.1 + Transformers 4.27.2
MGeo模型权重及配套工具包（位于/root/mgeo/）
Jupyter Lab（可通过http://<IP>:8888访问）

提示：所有操作均在容器内终端执行，无需额外安装驱动或CUDA。

2.2 创建服务目录并整理代码

先将原始推理脚本迁移到规范路径，便于后续维护：

mkdir -p /root/mgeo_api cp /root/推理.py /root/mgeo_api/inference.py cp -r /root/mgeo /root/mgeo_api/model/

接着新建核心服务文件：

touch /root/mgeo_api/app.py touch /root/mgeo_api/requirements.txt

我们不追求“大而全”的框架，只保留最精简、最可靠的服务结构：

app.py：Flask主服务，加载模型、定义接口、处理请求
inference.py：原生推理逻辑封装为函数，不耦合Web层
model/：模型权重与配置，保持与原始部署一致
requirements.txt：仅声明必要依赖（Flask + torch + numpy）

2.3 安装轻量依赖（不污染原环境）

激活已有环境，仅安装Flask（其他依赖均已预装）：

conda activate py37testmaas pip install flask==2.2.5 -i https://pypi.tuna.tsinghua.edu.cn/simple/

注意：使用 Flask 2.2.5 是因该版本兼容 Python 3.7 且无 async 默认行为，避免与 MGeo 的同步推理逻辑冲突。

3. 核心代码实现：三步封装API

3.1 改造推理脚本为可复用函数

打开/root/mgeo_api/inference.py，将原脚本中硬编码的输入/输出逻辑抽离为标准函数：

# /root/mgeo_api/inference.py import torch from transformers import AutoTokenizer, AutoModel import numpy as np from sklearn.metrics.pairwise import cosine_similarity def load_mgeo_model(model_path="/root/mgeo_api/model"): """加载MGeo模型与分词器（仅加载一次）""" tokenizer = AutoTokenizer.from_pretrained(model_path) model = AutoModel.from_pretrained(model_path) model.eval() return tokenizer, model def get_address_embedding(tokenizer, model, address: str) -> np.ndarray: """获取单个地址的句向量（768维）""" inputs = tokenizer( address, return_tensors="pt", padding=True, truncation=True, max_length=64 ) with torch.no_grad(): outputs = model(**inputs) # 取[CLS] token的输出作为句子表征 cls_output = outputs.last_hidden_state[:, 0, :].numpy() return cls_output.flatten() def compute_similarity(addr1: str, addr2: str, tokenizer, model) -> float: """计算两个地址的余弦相似度（0~1之间）""" vec1 = get_address_embedding(tokenizer, model, addr1) vec2 = get_address_embedding(tokenizer, model, addr2) sim = cosine_similarity([vec1], [vec2])[0][0] return float(np.clip(sim, 0.0, 1.0))

改动说明：

原始脚本中的print()、input()、路径硬编码全部移除；
模型加载逻辑独立为load_mgeo_model()，确保服务启动时只加载一次；
所有输入输出均为纯Python类型（str → float），无Tensor对象外泄，避免JSON序列化失败。

3.2 编写Flask主服务（app.py）

编辑/root/mgeo_api/app.py，内容如下：

# /root/mgeo_api/app.py from flask import Flask, request, jsonify import os import time from inference import load_mgeo_model, compute_similarity app = Flask(__name__) # 全局变量：模型与分词器（服务启动时加载） print("⏳ 正在加载MGeo模型，请稍候...") tokenizer, model = load_mgeo_model() print(" MGeo模型加载完成，服务就绪！") @app.route("/health", methods=["GET"]) def health_check(): return jsonify({"status": "healthy", "model": "MGeo-chinese-address"}) @app.route("/similarity", methods=["POST"]) def get_address_similarity(): try: data = request.get_json() addr1 = data.get("address1", "").strip() addr2 = data.get("address2", "").strip() if not addr1 or not addr2: return jsonify({"error": "缺少address1或address2字段"}), 400 if len(addr1) > 128 or len(addr2) > 128: return jsonify({"error": "地址长度不能超过128字符"}), 400 start_time = time.time() score = compute_similarity(addr1, addr2, tokenizer, model) latency = round((time.time() - start_time) * 1000, 1) return jsonify({ "address1": addr1, "address2": addr2, "similarity": round(score, 4), "latency_ms": latency }) except Exception as e: return jsonify({"error": f"服务内部错误：{str(e)}"}), 500 if __name__ == "__main__": # 绑定0.0.0.0确保外部可访问，端口设为5000 app.run(host="0.0.0.0", port=5000, debug=False)

关键设计点：

/health接口用于K8s探针或运维监控，返回轻量健康状态；
/similarity接口严格校验输入，拒绝空值和超长地址，避免模型OOM；
自动记录每次请求耗时（毫秒级），方便后续性能评估；
debug=False确保生产环境不暴露调试信息；
host="0.0.0.0"是关键，否则外部无法访问容器内服务。

3.3 启动服务并验证可用性

在终端中执行：

cd /root/mgeo_api conda activate py37testmaas python app.py

你会看到类似输出：

⏳ 正在加载MGeo模型，请稍候... MGeo模型加载完成，服务就绪！ * Running on http://0.0.0.0:5000

此时服务已在后台运行。新开一个终端窗口，用curl测试：

curl -X POST http://localhost:5000/similarity \ -H "Content-Type: application/json" \ -d '{"address1":"北京市朝阳区建国路8号", "address2":"北京朝阳建国路8号"}'

预期返回：

{ "address1": "北京市朝阳区建国路8号", "address2": "北京朝阳建国路8号", "similarity": 0.9234, "latency_ms": 326.7 }

成功！你已拥有一个开箱即用的地址相似度API服务。

4. 实战调用：三种常见接入方式

4.1 Python客户端调用（推荐给数据同学）

新建client_demo.py：

import requests url = "http://localhost:5000/similarity" data = { "address1": "上海市浦东新区张江路123号", "address2": "上海浦东张江路123号" } res = requests.post(url, json=data, timeout=10) print(res.json()) # 输出：{'address1': '...', 'address2': '...', 'similarity': 0.9412, 'latency_ms': 312.5}

4.2 JavaScript前端调用（适配网页表单）

// 前端HTML中可直接使用fetch async function checkAddressMatch() { const res = await fetch('http://<你的服务器IP>:5000/similarity', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ address1: document.getElementById('addr1').value, address2: document.getElementById('addr2').value }) }); const result = await res.json(); document.getElementById('score').innerText = result.similarity; }

注意：若部署在公网，需配置Nginx反向代理并启用CORS，或由后端统一转发请求。

4.3 批量处理脚本（运营/ETL场景）

# batch_check.py import pandas as pd import requests import time df = pd.read_csv("address_pairs.csv") # 两列：addr_a, addr_b results = [] for _, row in df.iterrows(): payload = {"address1": row["addr_a"], "address2": row["addr_b"]} try: r = requests.post("http://localhost:5000/similarity", json=payload, timeout=5) res = r.json() results.append({ "addr_a": row["addr_a"], "addr_b": row["addr_b"], "similarity": res.get("similarity", 0.0), "status": "success" }) except Exception as e: results.append({ "addr_a": row["addr_a"], "addr_b": row["addr_b"], "similarity": 0.0, "status": f"error: {e}" }) time.sleep(0.1) # 避免瞬时压垮服务 pd.DataFrame(results).to_csv("batch_results.csv", index=False)

小技巧：单卡4090D实测，该服务可持续支撑12~15 QPS（地址长度≤64字），满足中小规模业务需求。

5. 生产就绪增强建议

5.1 加入请求限流（防误刷）

在app.py顶部添加：

from flask_limiter import Limiter from flask_limiter.util import get_remote_address limiter = Limiter( app, key_func=get_remote_address, default_limits=["200 per day", "50 per hour"] )

并在接口装饰器中启用：

@app.route("/similarity", methods=["POST"]) @limiter.limit("30 per minute") def get_address_similarity(): # ...原有逻辑

安装依赖：pip install Flask-Limiter==3.5.0

5.2 日志记录与错误追踪

在app.py开头添加日志配置：

import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/root/mgeo_api/api.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__)

在接口中记录关键事件：

logger.info(f" Matched: '{addr1}' ↔ '{addr2}' → {score:.4f}")

5.3 模型热更新支持（进阶）

如需不重启服务更新模型，可将load_mgeo_model()改为带时间戳缓存的函数，并监听/model/update接口触发重载。此功能适用于A/B测试或多版本灰度发布场景，本文暂不展开。

6. 总结：你已掌握MGeo服务化的完整链路

6.1 本文你实际完成了什么？

在4090D单卡镜像中，零修改模型代码，仅通过封装完成服务化；
将原始Jupyter推理脚本，重构为可导入、可测试、可复用的模块化函数；
用不到50行Flask代码，搭建出生产可用的REST API，含健康检查、输入校验、耗时统计；
验证了三种主流调用方式：Python脚本、前端页面、批量处理，覆盖技术栈盲区；
获得了可立即落地的增强方案：限流、日志、错误码体系，让服务真正“扛得住”。

6.2 下一步你可以做什么？

把这个API注册到公司内部API网关，供多个业务线复用；
结合地址解析服务（如高德逆地理编码），构建“地址清洗→标准化→相似度匹配”全链路；
将/similarity接口扩展为/match，支持一对多批量比对（传入1个地址+100个候选地址）；
用Nginx做反向代理+HTTPS，对外提供安全、稳定的域名访问入口。

MGeo不是只能跑在Jupyter里的玩具模型——它是你手边一块可即插即用的“地址智能模块”。今天封装的这层Flask，就是让它走出实验室、走进真实业务的第一步。

获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

MGeo模型支持REST API吗？Flask封装服务部署实战教程