nomic-embed-text-v2-moe实战教程：使用FastAPI封装Gradio服务供生产调用

1. 环境准备与模型部署

在开始之前，我们需要先准备好运行环境并部署模型。这个部分会带你一步步完成基础环境的搭建。

1.1 系统要求与依赖安装

首先确保你的系统满足以下基本要求：

• Python 3.8 或更高版本
• 至少 8GB 内存（推荐 16GB）
• 足够的存储空间用于模型文件

安装必要的Python依赖包：

pip install fastapi uvicorn gradio httpx numpy pandas
pip install ollama  # 用于模型部署和管理

1.2 使用Ollama部署嵌入模型

Ollama是一个强大的模型部署工具，可以简化模型的下载和运行过程。使用以下命令部署nomic-embed-text-v2-moe模型：

# 拉取并运行模型
ollama pull nomic-embed-text-v2-moe
ollama run nomic-embed-text-v2-moe

部署成功后，你应该能看到模型正常运行的信息。默认情况下，Ollama会在本地11434端口提供服务。

2. 基础概念与模型特性

在开始编码之前，我们先了解一下这个嵌入模型的核心特性，这样能更好地理解后续的实现。

nomic-embed-text-v2-moe是一个多语言混合专家（MoE）文本嵌入模型，具有以下突出特点：

• 多语言支持：能够处理约100种不同的语言文本
• 高性能检索：在多语言检索任务上达到先进水平
• 灵活嵌入维度：支持Matryoshka嵌入，可以根据需要调整维度大小
• 完全开源：模型权重、训练代码和数据全部开放

与同类模型相比，它在多语言场景下表现优异，特别是在BEIR和MIRACL基准测试中都有不错的表现。

3. 使用Gradio构建前端界面

Gradio是一个快速构建机器学习演示界面的工具，我们先创建一个简单的前端来测试模型功能。

3.1 创建基础推理界面

创建一个简单的Gradio应用来测试文本嵌入功能：

import gradio as gr
import requests
import json

def get_embedding(text):
    """调用Ollama服务获取文本嵌入向量"""
    try:
        response = requests.post(
            "http://localhost:11434/api/embeddings",
            json={"model": "nomic-embed-text-v2-moe", "prompt": text}
        )
        if response.status_code == 200:
            result = response.json()
            return result.get('embedding', [])
        else:
            return f"Error: {response.status_code}"
    except Exception as e:
        return f"Exception: {str(e)}"

def similarity_check(text1, text2):
    """计算两个文本的相似度"""
    emb1 = get_embedding(text1)
    emb2 = get_embedding(text2)
    
    if isinstance(emb1, list) and isinstance(emb2, list):
        # 计算余弦相似度
        import numpy as np
        dot_product = np.dot(emb1, emb2)
        norm1 = np.linalg.norm(emb1)
        norm2 = np.linalg.norm(emb2)
        similarity = dot_product / (norm1 * norm2)
        return f"相似度: {similarity:.4f}"
    else:
        return "计算失败，请检查输入文本"

# 创建Gradio界面
with gr.Blocks() as demo:
    gr.Markdown("# nomic-embed-text-v2-moe 文本嵌入演示")
    
    with gr.Row():
        with gr.Column():
            text1 = gr.Textbox(label="文本1", lines=2, placeholder="输入第一段文本...")
            text2 = gr.Textbox(label="文本2", lines=2, placeholder="输入第二段文本...")
            btn = gr.Button("计算相似度")
        
        with gr.Column():
            output = gr.Textbox(label="相似度结果", interactive=False)
    
    btn.click(similarity_check, inputs=[text1, text2], outputs=output)

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)

3.2 测试Gradio界面

运行上面的代码后，打开浏览器访问 http://localhost:7860，你应该能看到一个简单的界面。输入两段文本，点击"计算相似度"按钮，就能看到它们之间的相似度得分。

例如尝试输入：

• 文本1: "我喜欢编程"
• 文本2: "编程是我的爱好"

你会得到一个0.8以上的相似度分数，说明模型能够很好地理解中文文本的语义。

4. 使用FastAPI封装生产服务

虽然Gradio界面很方便，但在生产环境中我们需要更稳定、高效的API服务。接下来我们用FastAPI来封装模型推理功能。

4.1 创建FastAPI应用

首先创建一个完整的FastAPI应用，提供文本嵌入和相似度计算接口：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import numpy as np
from typing import List
import uvicorn

app = FastAPI(title="nomic-embed-text-v2-moe API服务", version="1.0.0")

class EmbeddingRequest(BaseModel):
    text: str

class SimilarityRequest(BaseModel):
    text1: str
    text2: str

class EmbeddingResponse(BaseModel):
    embedding: List[float]
    dimension: int

class SimilarityResponse(BaseModel):
    similarity: float
    status: str = "success"

def get_embedding_vector(text: str) -> List[float]:
    """调用Ollama获取文本嵌入向量"""
    try:
        response = requests.post(
            "http://localhost:11434/api/embeddings",
            json={"model": "nomic-embed-text-v2-moe", "prompt": text},
            timeout=30
        )
        response.raise_for_status()
        return response.json().get('embedding', [])
    except requests.exceptions.RequestException as e:
        raise HTTPException(status_code=500, detail=f"模型服务调用失败: {str(e)}")

@app.post("/embed", response_model=EmbeddingResponse)
async def get_embedding(request: EmbeddingRequest):
    """获取单个文本的嵌入向量"""
    embedding = get_embedding_vector(request.text)
    return EmbeddingResponse(embedding=embedding, dimension=len(embedding))

@app.post("/batch_embed", response_model=List[EmbeddingResponse])
async def get_batch_embedding(requests: List[EmbeddingRequest]):
    """批量获取文本嵌入向量"""
    results = []
    for req in requests:
        embedding = get_embedding_vector(req.text)
        results.append(EmbeddingResponse(embedding=embedding, dimension=len(embedding)))
    return results

@app.post("/similarity", response_model=SimilarityResponse)
async def calculate_similarity(request: SimilarityRequest):
    """计算两个文本的相似度"""
    emb1 = get_embedding_vector(request.text1)
    emb2 = get_embedding_vector(request.text2)
    
    # 计算余弦相似度
    dot_product = np.dot(emb1, emb2)
    norm1 = np.linalg.norm(emb1)
    norm2 = np.linalg.norm(emb2)
    
    if norm1 == 0 or norm2 == 0:
        raise HTTPException(status_code=400, detail="嵌入向量长度为零")
    
    similarity = dot_product / (norm1 * norm2)
    return SimilarityResponse(similarity=float(similarity))

@app.get("/health")
async def health_check():
    """健康检查接口"""
    try:
        # 简单测试模型服务是否正常
        test_text = "健康检查"
        embedding = get_embedding_vector(test_text)
        if len(embedding) > 0:
            return {"status": "healthy", "model": "nomic-embed-text-v2-moe"}
        else:
            return {"status": "unhealthy", "reason": "空嵌入向量"}
    except Exception as e:
        return {"status": "unhealthy", "reason": str(e)}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

4.2 测试FastAPI服务

启动FastAPI服务后，你可以通过以下方式测试API：

# 启动服务
python app.py

# 测试单个文本嵌入
curl -X POST "http://localhost:8000/embed" \
     -H "Content-Type: application/json" \
     -d '{"text": "这是一个测试文本"}'

# 测试相似度计算
curl -X POST "http://localhost:8000/similarity" \
     -H "Content-Type: application/json" \
     -d '{"text1": "我喜欢编程", "text2": "编程很有趣"}'

服务会返回JSON格式的响应，包含嵌入向量或相似度分数。

5. 整合Gradio与FastAPI

现在我们将Gradio前端与FastAPI后端整合，创建一个既方便演示又适合生产使用的完整系统。

5.1 创建生产级Gradio界面

创建一个改进的Gradio界面，直接调用我们的FastAPI服务：

import gradio as gr
import requests
import json

API_BASE = "http://localhost:8000"

def get_embedding_via_api(text):
    """通过FastAPI获取文本嵌入"""
    try:
        response = requests.post(
            f"{API_BASE}/embed",
            json={"text": text},
            timeout=30
        )
        if response.status_code == 200:
            return response.json()["embedding"]
        else:
            return f"API错误: {response.status_code}"
    except Exception as e:
        return f"异常: {str(e)}"

def similarity_via_api(text1, text2):
    """通过FastAPI计算相似度"""
    try:
        response = requests.post(
            f"{API_BASE}/similarity",
            json={"text1": text1, "text2": text2},
            timeout=30
        )
        if response.status_code == 200:
            result = response.json()
            return f"相似度: {result['similarity']:.4f}"
        else:
            return f"计算失败: {response.status_code}"
    except Exception as e:
        return f"请求异常: {str(e)}"

def batch_process_texts(texts):
    """批量处理文本"""
    texts_list = [t.strip() for t in texts.split("\n") if t.strip()]
    requests_data = [{"text": text} for text in texts_list]
    
    try:
        response = requests.post(
            f"{API_BASE}/batch_embed",
            json=requests_data,
            timeout=60
        )
        if response.status_code == 200:
            results = response.json()
            output = "批量处理结果:\n\n"
            for i, result in enumerate(results):
                output += f"文本 {i+1}: 维度={result['dimension']}, 前5个值={result['embedding'][:5]}\n"
            return output
        else:
            return f"批量处理失败: {response.status_code}"
    except Exception as e:
        return f"批量处理异常: {str(e)}"

# 创建增强的Gradio界面
with gr.Blocks(title="nomic-embed-text-v2-moe 生产服务") as demo:
    gr.Markdown("""
    # 🚀 nomic-embed-text-v2-moe 生产服务界面
    
    本界面提供文本嵌入和相似度计算功能，后端基于FastAPI生产服务。
    """)
    
    with gr.Tab("单文本嵌入"):
        with gr.Row():
            single_text = gr.Textbox(label="输入文本", lines=3, placeholder="请输入要嵌入的文本...")
            single_btn = gr.Button("生成嵌入")
        single_output = gr.Textbox(label="嵌入结果", interactive=False, lines=6)
        
        single_btn.click(get_embedding_via_api, inputs=single_text, outputs=single_output)
    
    with gr.Tab("文本相似度"):
        with gr.Row():
            with gr.Column():
                text1 = gr.Textbox(label="文本1", lines=2)
                text2 = gr.Textbox(label="文本2", lines=2)
            sim_btn = gr.Button("计算相似度")
        sim_output = gr.Textbox(label="相似度结果", interactive=False)
        
        sim_btn.click(similarity_via_api, inputs=[text1, text2], outputs=sim_output)
    
    with gr.Tab("批量处理"):
        batch_texts = gr.Textbox(label="批量文本", lines=6, 
                               placeholder="每行一个文本，支持批量处理...")
        batch_btn = gr.Button("批量处理")
        batch_output = gr.Textbox(label="处理结果", interactive=False, lines=10)
        
        batch_btn.click(batch_process_texts, inputs=batch_texts, outputs=batch_output)
    
    with gr.Tab("API文档"):
        gr.Markdown("""
        ## API接口文档
        
        ### 基础信息
        - 基础URL: `http://localhost:8000`
        - 健康检查: `GET /health`
        
        ### 主要接口
        1. **单文本嵌入**
           - 端点: `POST /embed`
           - 参数: `{"text": "你的文本"}`
           - 返回: 嵌入向量和维度
        
        2. **批量文本嵌入**
           - 端点: `POST /batch_embed`
           - 参数: `[{"text": "文本1"}, {"text": "文本2"}]`
           - 返回: 批量嵌入结果
        
        3. **文本相似度**
           - 端点: `POST /similarity`
           - 参数: `{"text1": "文本1", "text2": "文本2"}`
           - 返回: 相似度分数
        """)

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860, share=False)

5.2 部署与运行说明

要运行完整的生产系统，你需要按顺序启动两个服务：

1. 首先启动Ollama模型服务：

ollama run nomic-embed-text-v2-moe

2. 然后启动FastAPI后端服务：

python app.py

3. 最后启动Gradio前端界面：

python gradio_app.py

现在你可以通过以下方式访问系统：

• FastAPI后端: http://localhost:8000
• Gradio前端: http://localhost:7860
• API文档: http://localhost:8000/docs

6. 实用技巧与进阶功能

在实际使用过程中，这里有一些实用技巧可以帮助你获得更好的效果。

6.1 性能优化建议

对于生产环境，可以考虑以下优化措施：

# 添加缓存机制减少重复计算
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_embedding(text: str) -> List[float]:
    """带缓存的嵌入获取函数"""
    return get_embedding_vector(text)

# 添加请求限流保护服务
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/embed")
@limiter.limit("10/minute")  # 每分钟10次请求
async def get_embedding_limited(request: EmbeddingRequest):
    # 原有逻辑
    pass

6.2 错误处理与监控

增强服务的稳定性和可观测性：

# 添加详细的日志记录
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request, call_next):
    logger.info(f"请求: {request.method} {request.url}")
    response = await call_next(request)
    logger.info(f"响应: {response.status_code}")
    return response

# 添加Prometheus监控指标
from prometheus_fastapi_instrumentator import Instrumentator
Instrumentator().instrument(app).expose(app)

7. 常见问题解答

在实际部署和使用过程中，可能会遇到一些常见问题。

问题1: Ollama服务连接失败

• 检查Ollama是否正常运行：ollama list
• 确认模型已正确下载：ollama pull nomic-embed-text-v2-moe
• 验证服务端口：默认是11434

问题2: 内存不足

• 模型需要较多内存，建议16GB以上
• 可以调整Ollama的GPU设置来减少内存使用

问题3: 请求超时

• 长文本处理可能需要更多时间
• 调整FastAPI和请求的超时设置

问题4: 批量处理性能

• 对于大量文本，建议使用批量接口
• 考虑添加队列系统处理大量请求

8. 总结

通过本教程，我们完成了从模型部署到生产服务的完整流程。你现在拥有：

1. 本地运行的nomic-embed-text-v2-moe模型：通过Ollama轻松管理
2. 高效的FastAPI后端服务：提供稳定的生产级API接口
3. 友好的Gradio前端界面：方便测试和演示功能
4. 完整的文档和示例：快速上手和使用

这种架构的优势在于：

• 前后端分离：前端专注于用户体验，后端保证稳定性
• 易于扩展：可以轻松添加新的功能接口
• 生产就绪：包含错误处理、监控、限流等生产环境特性
• 灵活部署：可以容器化部署到各种环境

现在你可以将这个系统集成到自己的应用中，享受多语言文本嵌入的强大能力。无论是构建搜索引擎、推荐系统，还是内容分析工具，这个基础架构都能为你提供可靠的技术支持。

---

获取更多AI镜像

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