HY-MT1.5-1.8B部署监控：API调用日志与QPS统计实战配置

1. 引言：为什么需要监控翻译模型部署

当你把HY-MT1.5-1.8B这样的高性能翻译模型部署到生产环境后，最关心的问题是什么？是翻译质量？是响应速度？还是系统稳定性？

实际上，这些都需要通过监控来保障。没有监控的模型部署就像开车没有仪表盘——你不知道当前速度多少，不知道油量还剩多少，更不知道发动机是否正常运转。

HY-MT1.5-1.8B作为腾讯混元开源的轻量级多语翻译模型，虽然本身性能出色（手机端1GB内存可跑、速度0.18秒），但在实际部署中，你仍然需要：

• 实时了解API调用情况，知道哪些语言对最常用
• 监控QPS（每秒查询数），确保系统不会过载
• 统计响应时间，保证用户体验
• 及时发现异常调用和错误请求

本文将手把手教你如何为HY-MT1.5-1.8B部署添加完整的监控体系，让你对模型的运行状况了如指掌。

2. 环境准备与基础部署

2.1 安装HY-MT1.5-1.8B

首先，我们需要部署模型本身。HY-MT1.5-1.8B支持多种部署方式，这里以Hugging Face Transformers为例：

# 安装依赖
pip install transformers torch

# 或者使用ModelScope（国内镜像）
# pip install modelscope transformers

2.2 基础模型加载代码

from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
import torch

# 加载模型和分词器
model_name = "Hunyuan/HY-MT1.5-1.8B"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSeq2SeqLM.from_pretrained(
    model_name,
    torch_dtype=torch.float16,  # 使用半精度减少内存占用
    device_map="auto"           # 自动分配设备
)

def translate_text(text, src_lang="zh", tgt_lang="en"):
    """基础翻译函数"""
    # 添加语言标签
    input_text = f"<{src_lang}2{tgt_lang}> {text}"
    
    # 编码输入
    inputs = tokenizer(input_text, return_tensors="pt", truncation=True, max_length=512)
    
    # 生成翻译
    with torch.no_grad():
        outputs = model.generate(
            inputs.input_ids,
            max_length=512,
            num_beams=5,
            early_stopping=True
        )
    
    # 解码输出
    translated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
    return translated_text

这个基础版本已经可以工作，但没有任何监控功能。接下来我们逐步添加监控能力。

3. API调用日志记录实战

3.1 添加基础日志记录

首先，我们为翻译函数添加详细的日志记录：

import logging
import json
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('translation_api.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger("HY-MT-API")

def translate_with_logging(text, src_lang="zh", tgt_lang="en", user_id="anonymous"):
    """带日志记录的翻译函数"""
    start_time = datetime.now()
    
    try:
        # 记录请求开始
        logger.info(f"Translation request - User: {user_id}, Src: {src_lang}, Tgt: {tgt_lang}, Text: {text[:100]}...")
        
        # 执行翻译
        translated_text = translate_text(text, src_lang, tgt_lang)
        
        # 计算耗时
        processing_time = (datetime.now() - start_time).total_seconds()
        
        # 记录成功日志
        logger.info(f"Translation success - Time: {processing_time:.3f}s, "
                   f"Chars: {len(text)}->{len(translated_text)}")
        
        # 返回结果
        return {
            "success": True,
            "translated_text": translated_text,
            "processing_time": processing_time,
            "source_chars": len(text),
            "target_chars": len(translated_text)
        }
        
    except Exception as e:
        # 记录错误日志
        processing_time = (datetime.now() - start_time).total_seconds()
        logger.error(f"Translation failed - Error: {str(e)}, Time: {processing_time:.3f}s")
        
        return {
            "success": False,
            "error": str(e),
            "processing_time": processing_time
        }

3.2 结构化日志记录

为了便于后续分析，我们使用JSON格式的结构化日志：

import json

def log_structured_event(event_type, data):
    """记录结构化日志"""
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "event_type": event_type,
        "data": data
    }
    
    with open("structured_translation_logs.jsonl", "a") as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")

# 修改翻译函数，添加结构化日志
def translate_with_structured_logging(text, src_lang="zh", tgt_lang="en", user_id="anonymous"):
    start_time = datetime.now()
    
    # 记录请求日志
    log_structured_event("translation_request", {
        "user_id": user_id,
        "src_lang": src_lang,
        "tgt_lang": tgt_lang,
        "text_length": len(text),
        "text_preview": text[:100]
    })
    
    try:
        translated_text = translate_text(text, src_lang, tgt_lang)
        processing_time = (datetime.now() - start_time).total_seconds()
        
        # 记录成功日志
        log_structured_event("translation_success", {
            "user_id": user_id,
            "src_lang": src_lang,
            "tgt_lang": tgt_lang,
            "processing_time": processing_time,
            "source_chars": len(text),
            "target_chars": len(translated_text),
            "chars_per_second": len(text) / processing_time if processing_time > 0 else 0
        })
        
        return translated_text
        
    except Exception as e:
        processing_time = (datetime.now() - start_time).total_seconds()
        
        # 记录错误日志
        log_structured_event("translation_error", {
            "user_id": user_id,
            "src_lang": src_lang,
            "tgt_lang": tgt_lang,
            "error_type": type(e).__name__,
            "error_message": str(e),
            "processing_time": processing_time
        })
        
        raise e

4. QPS统计与性能监控

4.1 实时QPS计算

要实现QPS统计，我们需要跟踪请求频率：

from collections import deque
import time

class QPSMonitor:
    """QPS监控器"""
    
    def __init__(self, window_size=60):
        self.request_times = deque()
        self.window_size = window_size  # 统计窗口大小（秒）
    
    def record_request(self):
        """记录一个请求"""
        current_time = time.time()
        self.request_times.append(current_time)
        
        # 移除过期的记录
        while self.request_times and self.request_times[0] < current_time - self.window_size:
            self.request_times.popleft()
    
    def get_current_qps(self):
        """获取当前QPS"""
        if not self.request_times:
            return 0
        
        current_time = time.time()
        # 计算窗口内的请求数
        valid_requests = sum(1 for t in self.request_times 
                           if t >= current_time - self.window_size)
        
        return valid_requests / self.window_size
    
    def get_stats(self):
        """获取统计信息"""
        return {
            "current_qps": self.get_current_qps(),
            "total_requests_1min": len(self.request_times),
            "window_size": self.window_size
        }

# 全局QPS监控器
qps_monitor = QPSMonitor()

# 集成QPS监控的翻译函数
def translate_with_monitoring(text, src_lang="zh", tgt_lang="en", user_id="anonymous"):
    # 记录QPS
    qps_monitor.record_request()
    
    start_time = time.time()
    try:
        result = translate_with_structured_logging(text, src_lang, tgt_lang, user_id)
        
        # 记录性能指标
        processing_time = time.time() - start_time
        log_structured_event("performance_metrics", {
            "processing_time": processing_time,
            "text_length": len(text),
            "qps_at_request": qps_monitor.get_current_qps(),
            "timestamp": datetime.now().isoformat()
        })
        
        return result
        
    except Exception as e:
        # 记录错误时的性能指标
        processing_time = time.time() - start_time
        log_structured_event("performance_error", {
            "processing_time": processing_time,
            "error": str(e),
            "qps_at_request": qps_monitor.get_current_qps()
        })
        raise e

4.2 实时监控仪表板

我们可以创建一个简单的实时监控页面：

from flask import Flask, jsonify
import threading

app = Flask(__name__)

# 全局统计变量
stats = {
    "total_requests": 0,
    "successful_requests": 0,
    "failed_requests": 0,
    "avg_processing_time": 0,
    "last_updated": datetime.now().isoformat()
}

@app.route('/api/stats')
def get_stats():
    """获取实时统计信息"""
    current_stats = {
        **stats,
        **qps_monitor.get_stats(),
        "uptime": (datetime.now() - app_start_time).total_seconds()
    }
    return jsonify(current_stats)

@app.route('/api/health')
def health_check():
    """健康检查端点"""
    return jsonify({
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "model_loaded": model is not None
    })

def update_stats_loop():
    """后台统计更新循环"""
    while True:
        time.sleep(5)  # 每5秒更新一次
        # 这里可以添加从日志中聚合统计信息的逻辑
        stats["last_updated"] = datetime.now().isoformat()

# 启动后台线程
stats_thread = threading.Thread(target=update_stats_loop, daemon=True)
stats_thread.start()

app_start_time = datetime.now()

5. 高级监控功能实现

5.1 多维度统计聚合

为了更深入的分析，我们可以实现多维度统计：

import pandas as pd
from collections import defaultdict

class AdvancedMetrics:
    """高级指标统计"""
    
    def __init__(self):
        self.lang_pair_stats = defaultdict(lambda: {
            "count": 0,
            "total_chars": 0,
            "total_time": 0,
            "errors": 0
        })
        
        self.user_stats = defaultdict(lambda: {
            "requests": 0,
            "total_chars": 0
        })
    
    def record_translation(self, src_lang, tgt_lang, user_id, chars, processing_time, success=True):
        """记录翻译指标"""
        lang_pair = f"{src_lang}-{tgt_lang}"
        
        # 更新语言对统计
        self.lang_pair_stats[lang_pair]["count"] += 1
        self.lang_pair_stats[lang_pair]["total_chars"] += chars
        self.lang_pair_stats[lang_pair]["total_time"] += processing_time
        if not success:
            self.lang_pair_stats[lang_pair]["errors"] += 1
        
        # 更新用户统计
        self.user_stats[user_id]["requests"] += 1
        self.user_stats[user_id]["total_chars"] += chars
    
    def get_lang_pair_stats(self):
        """获取语言对统计"""
        stats = []
        for lang_pair, data in self.lang_pair_stats.items():
            if data["count"] > 0:
                stats.append({
                    "lang_pair": lang_pair,
                    "request_count": data["count"],
                    "total_chars": data["total_chars"],
                    "avg_chars_per_request": data["total_chars"] / data["count"],
                    "avg_processing_time": data["total_time"] / data["count"] if data["count"] > 0 else 0,
                    "error_rate": data["errors"] / data["count"] if data["count"] > 0 else 0
                })
        return sorted(stats, key=lambda x: x["request_count"], reverse=True)
    
    def get_user_stats(self):
        """获取用户统计"""
        return [
            {"user_id": uid, **data} 
            for uid, data in self.user_stats.items()
        ]

# 初始化高级统计
advanced_metrics = AdvancedMetrics()

# 集成高级统计的翻译函数
def translate_with_advanced_metrics(text, src_lang="zh", tgt_lang="en", user_id="anonymous"):
    start_time = time.time()
    
    try:
        result = translate_with_monitoring(text, src_lang, tgt_lang, user_id)
        processing_time = time.time() - start_time
        
        # 记录高级指标
        advanced_metrics.record_translation(
            src_lang, tgt_lang, user_id, 
            len(text), processing_time, True
        )
        
        return result
        
    except Exception as e:
        processing_time = time.time() - start_time
        advanced_metrics.record_translation(
            src_lang, tgt_lang, user_id,
            len(text), processing_time, False
        )
        raise e

5.2 异常检测与告警

class AnomalyDetector:
    """异常检测器"""
    
    def __init__(self):
        self.processing_times = []
        self.error_rates = []
        self.max_window_size = 1000
    
    def check_anomalies(self, processing_time, error_occurred=False):
        """检查异常"""
        # 记录历史数据
        self.processing_times.append(processing_time)
        self.error_rates.append(1 if error_occurred else 0)
        
        # 保持窗口大小
        if len(self.processing_times) > self.max_window_size:
            self.processing_times = self.processing_times[-self.max_window_size:]
            self.error_rates = self.error_rates[-self.max_window_size:]
        
        # 检测处理时间异常
        if len(self.processing_times) > 10:
            recent_avg = sum(self.processing_times[-10:]) / 10
            historical_avg = sum(self.processing_times) / len(self.processing_times)
            
            if processing_time > recent_avg * 2:  # 超过近期平均2倍
                logger.warning(f"Processing time anomaly detected: {processing_time:.3f}s "
                              f"(recent avg: {recent_avg:.3f}s)")
        
        # 检测错误率异常
        if len(self.error_rates) > 20:
            recent_error_rate = sum(self.error_rates[-20:]) / 20
            if recent_error_rate > 0.1:  # 错误率超过10%
                logger.error(f"High error rate detected: {recent_error_rate:.1%}")
                
        # 检测QPS异常
        current_qps = qps_monitor.get_current_qps()
        if current_qps > 50:  # QPS超过50
            logger.warning(f"High QPS detected: {current_qps:.1f}")

# 初始化异常检测器
anomaly_detector = AnomalyDetector()

6. 完整部署示例与测试

6.1 完整监控集成

现在让我们把所有组件整合到一起：

class HYMTMonitor:
    """HY-MT模型监控完整集成"""
    
    def __init__(self):
        self.qps_monitor = QPSMonitor()
        self.advanced_metrics = AdvancedMetrics()
        self.anomaly_detector = AnomalyDetector()
        self.total_requests = 0
        
    def translate(self, text, src_lang="zh", tgt_lang="en", user_id="anonymous"):
        """完整的监控翻译函数"""
        self.total_requests += 1
        self.qps_monitor.record_request()
        
        start_time = time.time()
        
        # 记录请求开始
        log_structured_event("request_start", {
            "request_id": self.total_requests,
            "user_id": user_id,
            "src_lang": src_lang,
            "tgt_lang": tgt_lang,
            "text_length": len(text),
            "timestamp": datetime.now().isoformat()
        })
        
        try:
            # 执行翻译
            translated_text = translate_text(text, src_lang, tgt_lang)
            processing_time = time.time() - start_time
            
            # 记录各种指标
            self.advanced_metrics.record_translation(
                src_lang, tgt_lang, user_id, len(text), processing_time, True
            )
            
            self.anomaly_detector.check_anomalies(processing_time, False)
            
            # 记录成功日志
            log_structured_event("request_success", {
                "request_id": self.total_requests,
                "processing_time": processing_time,
                "qps": self.qps_monitor.get_current_qps(),
                "translated_length": len(translated_text)
            })
            
            return translated_text
            
        except Exception as e:
            processing_time = time.time() - start_time
            
            # 记录错误指标
            self.advanced_metrics.record_translation(
                src_lang, tgt_lang, user_id, len(text), processing_time, False
            )
            
            self.anomaly_detector.check_anomalies(processing_time, True)
            
            # 记录错误日志
            log_structured_event("request_error", {
                "request_id": self.total_requests,
                "processing_time": processing_time,
                "error": str(e),
                "qps": self.qps_monitor.get_current_qps()
            })
            
            raise e
    
    def get_comprehensive_stats(self):
        """获取综合统计信息"""
        return {
            "total_requests": self.total_requests,
            "qps_stats": self.qps_monitor.get_stats(),
            "lang_pair_stats": self.advanced_metrics.get_lang_pair_stats()[:5],  # 前5个最常用语言对
            "top_users": self.advanced_metrics.get_user_stats()[:5]  # 前5个最活跃用户
        }

# 初始化监控器
monitor = HYMTMonitor()

6.2 测试监控系统

让我们测试一下完整的监控系统：

# 测试代码
def test_monitoring_system():
    """测试监控系统"""
    test_cases = [
        ("你好，世界", "zh", "en", "test_user_1"),
        ("Hello world", "en", "zh", "test_user_2"),
        ("这是一段较长的文本，用于测试监控系统是否能正确处理各种长度的输入", "zh", "en", "test_user_1"),
    ]
    
    print("开始测试监控系统...")
    
    for i, (text, src, tgt, user) in enumerate(test_cases):
        try:
            print(f"\n测试 {i+1}: {src} -> {tgt}")
            result = monitor.translate(text, src, tgt, user)
            print(f"翻译结果: {result}")
            
            # 稍微延迟，模拟真实请求间隔
            time.sleep(0.1)
            
        except Exception as e:
            print(f"翻译失败: {e}")
    
    # 显示统计信息
    print("\n=== 测试统计 ===")
    stats = monitor.get_comprehensive_stats()
    print(f"总请求数: {stats['total_requests']}")
    print(f"当前QPS: {stats['qps_stats']['current_qps']:.2f}")
    
    print("\n语言对统计:")
    for lang_stat in stats['lang_pair_stats']:
        print(f"  {lang_stat['lang_pair']}: {lang_stat['request_count']} 次请求")

# 运行测试
if __name__ == "__main__":
    test_monitoring_system()

7. 总结与最佳实践

通过本文的实战配置，我们为HY-MT1.5-1.8B翻译模型构建了完整的监控体系，包括：

7.1 核心监控能力

1. API调用日志：详细记录每个翻译请求的详细信息
2. QPS实时统计：监控系统负载和请求频率
3. 性能指标收集：跟踪处理时间、字符数等关键指标
4. 多维度统计：按语言对、用户等维度聚合数据
5. 异常检测：自动检测性能异常和高错误率

7.2 部署建议

1. 日志管理：对于生产环境，建议使用ELK栈或类似工具管理日志
2. 监控仪表板：可以考虑使用Grafana等工具创建可视化监控面板
3. 告警集成：将异常检测与邮件、短信等告警系统集成
4. 数据持久化：将统计信息保存到数据库中进行长期分析
5. 性能优化：根据监控数据调整模型配置和硬件资源

7.3 扩展思路

1. 质量监控：可以添加翻译质量评估指标
2. 成本统计：跟踪API调用成本（如果部署在云上）
3. 用户行为分析：分析用户使用模式和偏好
4. A/B测试：基于监控数据开展模型版本对比测试

HY-MT1.5-1.8B作为一个高效的翻译模型，配合完善的监控系统，可以在生产环境中发挥最大价值。通过实时监控和数据分析，你不仅能确保系统稳定运行，还能不断优化用户体验和服务质量。

---

获取更多AI镜像

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