M2LOrder情感分析API文档实战：Swagger交互式调试+curl命令速查表

M2LOrder情感分析API文档实战Swagger交互式调试curl命令速查表1. 引言为什么你需要这份实战指南如果你正在寻找一个开箱即用的情感分析服务或者想快速集成情绪识别功能到你的应用中那么M2LOrder可能就是你需要的解决方案。但面对97个不同大小的模型、两种访问方式API和WebUI、以及一堆技术文档从哪里开始上手最有效率这就是我写这篇文章的原因。我花了几天时间深入测试M2LOrder的每个功能从最简单的文本分析到批量处理从WebUI的图形化操作到API的编程调用。我发现虽然官方文档很全面但对于实际使用来说有两个痛点特别明显API文档分散各种curl命令散落在文档各处使用时需要来回翻找调试效率低每次测试API都要手动构造请求费时费力所以我把所有实用的API命令整理成了速查表并且详细展示了如何用Swagger文档进行交互式调试。无论你是前端开发者想快速集成情感分析还是后端工程师需要批量处理用户反馈这篇文章都能帮你节省大量时间。2. M2LOrder快速入门5分钟启动服务在深入API细节之前我们先确保服务能正常运行。M2LOrder提供了三种启动方式我推荐使用第一种因为它最简单。2.1 三种启动方式对比启动方式命令适合场景优点缺点启动脚本cd /root/m2lorder ./start.sh快速测试、开发环境一键启动最简单后台运行管理稍弱Supervisorcd /root/m2lorder supervisord -c supervisor/supervisord.conf生产环境、长期运行进程监控、自动重启配置稍复杂手动启动分别启动API和WebUI调试、自定义配置最灵活可实时查看输出需要两个终端窗口2.2 我推荐的启动流程对于大多数用户我建议这样操作# 1. 进入项目目录 cd /root/m2lorder # 2. 使用启动脚本最简单 ./start.sh # 3. 验证服务是否启动成功 # 检查API服务 curl http://localhost:8001/health # 检查WebUI服务在浏览器中访问 # http://你的服务器IP:7861如果一切正常你会看到类似这样的健康检查响应{ status: healthy, service: m2lorder-api, timestamp: 2026-01-31T10:29:09.870785, task: emotion-recognition }2.3 访问地址一览启动成功后你可以通过以下地址访问不同服务服务类型访问地址用途说明WebUI界面http://你的服务器IP:7861图形化操作适合非技术人员API服务http://你的服务器IP:8001程序调用接口API文档http://你的服务器IP:8001/docs重点交互式调试界面备用文档http://你的服务器IP:8001/redoc另一种格式的API文档小贴士如果你在云服务器上部署记得在安全组中开放8001和7861端口。本地测试的话直接用localhost或127.0.0.1即可。3. WebUI图形界面零代码情感分析对于不熟悉编程的用户或者只是想快速测试一下效果WebUI是最佳选择。它的界面设计得很直观基本上看一眼就知道怎么用。3.1 界面布局与功能打开WebUI后你会看到这样一个布局左侧面板模型选择 右侧面板分析区域 ├─ 模型列表下拉框 ├─ 单文本输入框 ├─ 刷新模型按钮 ├─ 批量输入文本框 └─ 当前模型信息 └─ 分析结果展示区3.2 四步完成情感分析让我带你走一遍完整流程第一步选择模型点击左侧模型列表下拉框你会看到97个模型选项是的真的很多对于初次使用我建议选A001它只有3MB响应最快第二步输入文本在右侧输入文本框中输入你想分析的句子比如我今天收到了期待已久的礼物真是太开心了第三步开始分析点击那个大大的 开始分析按钮等待1-3秒取决于模型大小第四步查看结果你会看到类似这样的输出情感happy (开心) 置信度0.92 模型A001不同的情感会用不同颜色高亮显示3.3 情感类型与颜色对应M2LOrder支持6种基本情感每种都有对应的颜色标识情感类型英文标识颜色中文含义典型场景开心happy绿色 (#4CAF50)高兴、愉悦收到好消息、达成目标悲伤sad蓝色 (#2196F3)难过、失落失去重要事物、遭遇挫折愤怒angry红色 (#F44336)生气、不满受到不公对待、计划被打乱中性neutral灰色 (#9E9E9E)平静、无情绪事实陈述、客观描述兴奋excited橙色 (#FF9800)激动、期待即将旅行、重大活动前焦虑anxious紫色 (#9C27B0)担心、紧张考试前、重要决策时3.4 批量分析功能如果你有多条文本需要分析不用一条条输入。WebUI提供了批量处理功能在批量输入框中每行输入一条文本点击 批量分析按钮系统会以表格形式展示所有结果批量输入示例今天天气真好心情特别愉快 工作压力太大了有点喘不过气 这个决定让我很不安 完成了项目很有成就感输出结果文本情感置信度今天天气真好心情特别愉快happy0.88工作压力太大了有点喘不过气anxious0.76这个决定让我很不安anxious0.81完成了项目很有成就感excited0.793.5 模型选择策略面对97个模型怎么选我根据测试经验总结了一些建议按需求选择模型测试/演示选A001-A0123-4MB响应最快日常使用选A021-A0317-8MB平衡精度和速度高精度分析选A204-A236619MB系列但响应较慢特定场景A2xx系列可能针对特定角色/场景优化模型大小与性能关系# 简单来说模型越大精度越高但速度越慢 小模型3-8MB: 响应时间 1秒适合实时应用 中模型15-113MB: 响应时间 1-3秒平衡选择 大模型114-771MB: 响应时间 3-10秒适合离线分析 超大模型619MB-1.9GB: 响应时间 10秒专业场景4. API深度解析Swagger交互式调试实战现在进入重头戏——API接口。这是开发者最关心的部分因为只有通过API你才能把情感分析集成到自己的应用中。4.1 为什么Swagger是你的最佳调试工具在传统开发中测试API通常是这样打开终端输入curl命令查看返回结果如果有错修改命令重试反复循环...有了Swagger整个过程变得可视化打开浏览器访问http://你的IP:8001/docs点击要测试的API端点在网页上填写参数点击Execute按钮直接看到请求和响应4.2 访问Swagger文档确保API服务运行后在浏览器中输入http://你的服务器IP:8001/docs你会看到一个类似这样的界面FastAPI - Swagger UI ├─ /health [GET] 健康检查 ├─ /models [GET] 获取所有模型 ├─ /models/{model_id} [GET] 获取模型详情 ├─ /predict [POST] 单条情感预测 ├─ /predict/batch [POST] 批量情感预测 └─ /stats [GET] 统计信息每个端点旁边都有Try it out按钮这就是交互式调试的入口。4.3 六步掌握Swagger调试让我用一个实际例子演示如何使用Swagger调试/predict接口第一步找到目标接口在Swagger页面中找到/predict端点点击它展开详细面板第二步点击Try it out你会看到参数输入框从灰色变成可编辑状态第三步填写请求参数{ model_id: A001, input_data: Im really excited about the upcoming project! }model_id: 选择你要使用的模型IDinput_data: 输入要分析的文本第四步点击ExecuteSwagger会自动构造HTTP请求并发送你可以在页面上看到完整的curl命令第五步查看响应结果响应状态码200表示成功响应体JSON格式的情感分析结果响应头信息第六步复制代码Swagger提供了多种语言的代码示例你可以直接复制Python、JavaScript等代码到你的项目中4.4 实际调试示例分析一段复杂文本让我们分析一段包含混合情感的文本输入文本虽然项目最终成功了但过程中的压力让我好几次想要放弃。现在回想起来既感到欣慰又有些后怕。 **Swagger调试过程** 1. 在input_data中粘贴上述文本 2. model_id选择A001 3. 点击Execute **返回结果** json { model_id: A001, emotion: anxious, confidence: 0.72, timestamp: 20250601000001, metadata: { model_version: 0, model_size_mb: 3.0 } }结果解读模型识别为焦虑anxious置信度0.72属于中等置信度这反映了文本中压力、想要放弃、后怕等词汇的情感倾向4.5 Swagger的高级功能除了基本调试Swagger还有一些实用功能1. 参数说明查看每个参数都有详细说明包括数据类型、是否必填、示例值等2. 响应模式预览可以提前看到响应的JSON结构了解每个字段的含义3. 认证测试如果API需要支持Bearer Token、API Key等多种认证方式可以在UI中配置认证信息4. 批量测试虽然界面只显示单次请求但你可以快速修改参数进行多次测试5. 导出为OpenAPI规范点击Export可以下载完整的API规范用于导入到其他API测试工具5. curl命令速查表所有API一键调用虽然Swagger很方便但有时候你需要在脚本中调用API或者进行自动化测试。这时候curl命令就派上用场了。我把所有常用的API调用整理成了速查表你可以直接复制使用。5.1 基础信息获取健康检查- 确认服务是否正常运行curl http://localhost:8001/health获取所有模型- 查看可用的97个模型curl http://localhost:8001/models获取模型详情- 查看特定模型的详细信息curl http://localhost:8001/models/A001统计信息- 查看服务状态和模型统计curl http://localhost:8001/stats5.2 情感分析API单条文本分析- 最常用的功能curl -X POST http://localhost:8001/predict \ -H Content-Type: application/json \ -d { model_id: A001, input_data: 今天是个好日子心想的事儿都能成 }批量文本分析- 一次分析多条文本curl -X POST http://localhost:8001/predict/batch \ -H Content-Type: application/json \ -d { model_id: A001, inputs: [ 我中奖了太开心了, 项目失败了很难过, 明天要开会有点紧张, 今天的工作完成了 ] }5.3 高级用法示例使用变量简化命令# 设置变量 API_URLhttp://localhost:8001 MODEL_IDA001 TEXT这个产品太好用了我非常满意 # 调用API curl -X POST ${API_URL}/predict \ -H Content-Type: application/json \ -d {\model_id\:\${MODEL_ID}\,\input_data\:\${TEXT}\}从文件读取文本# 创建文本文件 echo {model_id:A001,input_data:测试文本} request.json # 从文件发送请求 curl -X POST http://localhost:8001/predict \ -H Content-Type: application/json \ -d request.json批量处理文本文件# 假设有一个文本文件每行一条待分析文本 cat texts.txt EOF 客户反馈产品很好用 客户投诉物流太慢了 客户建议希望增加新功能 客户表扬客服态度很好 EOF # 转换为JSON数组并发送 TEXTS$(cat texts.txt | jq -R -s split(\n) | map(select(. ! ))) curl -X POST http://localhost:8001/predict/batch \ -H Content-Type: application/json \ -d {\model_id\:\A001\,\inputs\:${TEXTS}}5.4 错误处理与调试查看详细响应包括响应头curl -v -X POST http://localhost:8001/predict \ -H Content-Type: application/json \ -d {model_id:A001,input_data:测试}只显示HTTP状态码curl -s -o /dev/null -w %{http_code} \ http://localhost:8001/health超时设置大模型可能需要更长时间curl -m 30 -X POST http://localhost:8001/predict \ -H Content-Type: application/json \ -d {model_id:A265,input_data:长文本测试...}5.5 常用组合命令健康检查状态监控# 一键检查所有服务状态 echo API健康检查 curl -s http://localhost:8001/health | jq .status echo 模型统计 curl -s http://localhost:8001/stats | jq .total_files, .total_size_mb echo 可用模型数 curl -s http://localhost:8001/models | jq length批量测试不同模型# 测试多个模型对同一文本的分析结果 TEXT这个结果让我既惊喜又担忧 for MODEL in A001 A002 A003 A004; do echo 测试模型: $MODEL curl -s -X POST http://localhost:8001/predict \ -H Content-Type: application/json \ -d {\model_id\:\$MODEL\,\input_data\:\$TEXT\} | \ jq .emotion, .confidence echo --- done性能测试脚本#!/bin/bash # 测试API响应时间 API_URLhttp://localhost:8001/predict TEST_TEXT这是一个测试文本用于评估情感分析API的性能表现 echo 开始性能测试... echo 测试文本: $TEST_TEXT echo for MODEL in A001 A021 A204; do echo 测试模型: $MODEL # 测量响应时间 START_TIME$(date %s%N) curl -s -X POST $API_URL \ -H Content-Type: application/json \ -d {\model_id\:\$MODEL\,\input_data\:\$TEST_TEXT\} /dev/null END_TIME$(date %s%N) DURATION$((($END_TIME - $START_TIME)/1000000)) echo 响应时间: ${DURATION}ms echo --- done6. 实战应用场景从理论到落地了解了基本用法后我们来看看M2LOrder在实际项目中能做什么。我整理了几个常见的应用场景和对应的实现方案。6.1 场景一用户反馈情感分析需求电商平台需要自动分析用户评价的情感倾向。解决方案import requests import json from typing import List, Dict class FeedbackAnalyzer: def __init__(self, api_url: str http://localhost:8001): self.api_url api_url self.model_id A001 # 使用轻量级模型保证速度 def analyze_single_feedback(self, feedback: str) - Dict: 分析单条用户反馈 response requests.post( f{self.api_url}/predict, json{ model_id: self.model_id, input_data: feedback } ) return response.json() def analyze_batch_feedbacks(self, feedbacks: List[str]) - Dict: 批量分析用户反馈 response requests.post( f{self.api_url}/predict/batch, json{ model_id: self.model_id, inputs: feedbacks } ) return response.json() def generate_report(self, feedbacks: List[str]) - Dict: 生成情感分析报告 results self.analyze_batch_feedbacks(feedbacks) # 统计情感分布 emotion_stats {} for pred in results.get(predictions, []): emotion pred.get(emotion, unknown) emotion_stats[emotion] emotion_stats.get(emotion, 0) 1 # 计算总体情感倾向 total len(feedbacks) positive_count emotion_stats.get(happy, 0) emotion_stats.get(excited, 0) negative_count emotion_stats.get(sad, 0) emotion_stats.get(angry, 0) emotion_stats.get(anxious, 0) neutral_count emotion_stats.get(neutral, 0) return { total_feedbacks: total, emotion_distribution: emotion_stats, positive_rate: positive_count / total if total 0 else 0, negative_rate: negative_count / total if total 0 else 0, neutral_rate: neutral_count / total if total 0 else 0, details: results.get(predictions, []) } # 使用示例 if __name__ __main__: analyzer FeedbackAnalyzer() # 模拟用户反馈 feedbacks [ 商品质量很好物流也快非常满意, 包装破损了有点失望, 客服态度不错解决问题很快, 价格有点高考虑一下, 已经第二次购买了一如既往的好 ] report analyzer.generate_report(feedbacks) print(json.dumps(report, indent2, ensure_asciiFalse))6.2 场景二社交媒体情感监控需求监控品牌在社交媒体上的情感口碑。解决方案import requests import schedule import time from datetime import datetime import sqlite3 class SocialMediaMonitor: def __init__(self): self.api_url http://localhost:8001 self.db_conn sqlite3.connect(social_media.db) self._init_database() def _init_database(self): 初始化数据库 cursor self.db_conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS sentiment_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT, content TEXT, emotion TEXT, confidence REAL, timestamp DATETIME ) ) self.db_conn.commit() def fetch_posts(self, platform: str, keyword: str) - List[str]: 模拟从社交媒体获取帖子实际项目中替换为真实API # 这里简化为返回模拟数据 mock_posts [ f刚买了{keyword}体验超棒, f{keyword}的售后服务太差了, f推荐大家试试{keyword}真的不错, f{keyword}的价格能不能再优惠点, f对{keyword}有点失望和宣传的不一样 ] return mock_posts def analyze_and_store(self, platform: str, keyword: str): 分析并存储情感数据 posts self.fetch_posts(platform, keyword) if not posts: return # 批量分析 response requests.post( f{self.api_url}/predict/batch, json{ model_id: A001, inputs: posts } ) if response.status_code 200: results response.json() cursor self.db_conn.cursor() for i, pred in enumerate(results.get(predictions, [])): cursor.execute( INSERT INTO sentiment_records (platform, content, emotion, confidence, timestamp) VALUES (?, ?, ?, ?, ?) , ( platform, posts[i], pred.get(emotion), pred.get(confidence), datetime.now().isoformat() )) self.db_conn.commit() print(f[{datetime.now()}] 已分析并存储 {len(posts)} 条{platform}帖子) def start_monitoring(self): 启动监控任务 # 每30分钟监控一次微博 schedule.every(30).minutes.do( self.analyze_and_store, weibo, 你的品牌 ) # 每1小时监控一次知乎 schedule.every(1).hours.do( self.analyze_and_store, zhihu, 你的品牌 ) print(社交媒体情感监控已启动...) while True: schedule.run_pending() time.sleep(60) # 使用示例 if __name__ __main__: monitor SocialMediaMonitor() monitor.start_monitoring()6.3 场景三客服对话情感分析需求实时分析客服对话中的用户情绪帮助客服人员更好地响应。解决方案from flask import Flask, request, jsonify import requests from collections import deque import threading import time app Flask(__name__) class ConversationAnalyzer: def __init__(self, api_url: str http://localhost:8001): self.api_url api_url self.conversation_history {} # 用户ID - 对话历史 self.emotion_trends {} # 用户ID - 情绪趋势 def analyze_message(self, user_id: str, message: str) - Dict: 分析单条消息的情感 # 调用M2LOrder API response requests.post( f{self.api_url}/predict, json{ model_id: A001, input_data: message }, timeout2 # 设置超时保证实时性 ) if response.status_code 200: result response.json() emotion result.get(emotion) confidence result.get(confidence) # 更新对话历史 if user_id not in self.conversation_history: self.conversation_history[user_id] deque(maxlen10) self.conversation_history[user_id].append({ message: message, emotion: emotion, confidence: confidence, timestamp: time.time() }) # 分析情绪趋势 trend self._analyze_emotion_trend(user_id) return { current_emotion: emotion, confidence: confidence, emotion_trend: trend, suggested_response: self._get_suggested_response(emotion, trend) } return {error: 分析失败} def _analyze_emotion_trend(self, user_id: str) - str: 分析用户情绪趋势 if user_id not in self.conversation_history: return neutral history list(self.conversation_history[user_id]) if len(history) 2: return stable # 简单趋势分析最近的情绪变化 recent_emotions [item[emotion] for item in history[-3:]] if len(set(recent_emotions)) 1: return stable # 情绪稳定 elif recent_emotions[-1] in [happy, excited]: return improving # 情绪改善 elif recent_emotions[-1] in [sad, angry, anxious]: return worsening # 情绪恶化 else: return fluctuating # 情绪波动 def _get_suggested_response(self, emotion: str, trend: str) - str: 根据情绪和趋势提供回复建议 suggestions { happy: { stable: 用户心情愉快可以继续当前话题, improving: 用户情绪正在变好可以询问更多需求, worsening: 用户情绪转好但需关注是否稳定, fluctuating: 用户情绪积极但有波动需温和沟通 }, sad: { stable: 用户持续悲伤需要更多关怀, improving: 用户情绪有所好转给予鼓励, worsening: 用户情绪恶化需要紧急关注, fluctuating: 用户情绪低落且不稳定需耐心安抚 }, angry: { stable: 用户持续愤怒需要冷静处理, improving: 用户愤怒情绪缓解继续安抚, worsening: 用户更加愤怒需要升级处理, fluctuating: 用户愤怒情绪波动需谨慎应对 } } return suggestions.get(emotion, {}).get(trend, 根据对话内容正常回应) # 初始化分析器 analyzer ConversationAnalyzer() app.route(/analyze, methods[POST]) def analyze_conversation(): 分析客服对话接口 data request.json user_id data.get(user_id) message data.get(message) if not user_id or not message: return jsonify({error: 缺少参数}), 400 result analyzer.analyze_message(user_id, message) return jsonify(result) app.route(/history/user_id, methods[GET]) def get_conversation_history(user_id): 获取用户对话历史 if user_id in analyzer.conversation_history: history list(analyzer.conversation_history[user_id]) return jsonify({history: history}) return jsonify({history: []}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)7. 性能优化与最佳实践在实际使用中你可能会遇到性能问题或者需要优化使用体验。这里我分享一些实战经验。7.1 模型选择策略根据我的测试不同场景下应该选择不同的模型实时应用场景响应时间 1秒# 推荐模型A001-A012 (3-4MB) FAST_MODELS [A001, A002, A003, A004, A005] # 使用策略轮询或根据负载选择 def get_fast_model(): 获取可用的快速模型 available_models [A001, A002, A003] # 可以在这里添加负载均衡逻辑 return available_models[0]批量处理场景可以接受较慢响应# 推荐模型A204-A236 (619MB系列) ACCURATE_MODELS [A204, A205, A206] # 使用策略夜间批量处理 def batch_process_at_night(texts: List[str]): 夜间批量处理使用高精度模型 # 选择精度最高的可用模型 model_id A204 # 或根据实际情况选择 results [] batch_size 10 # 分批处理避免内存溢出 for i in range(0, len(texts), batch_size): batch texts[i:ibatch_size] response requests.post( f{API_URL}/predict/batch, json{model_id: model_id, inputs: batch}, timeout30 # 大模型需要更长时间 ) if response.status_code 200: results.extend(response.json().get(predictions, [])) return results7.2 错误处理与重试机制网络请求可能会失败需要完善的错误处理import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry import time class RobustAPIClient: def __init__(self, base_url: str, max_retries: int 3): self.base_url base_url self.session requests.Session() # 配置重试策略 retry_strategy Retry( totalmax_retries, backoff_factor1, status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET, POST] ) adapter HTTPAdapter(max_retriesretry_strategy) self.session.mount(http://, adapter) self.session.mount(https://, adapter) def predict_with_retry(self, text: str, model_id: str A001) - Dict: 带重试的预测请求 for attempt in range(3): try: response self.session.post( f{self.base_url}/predict, json{ model_id: model_id, input_data: text }, timeout5 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: if attempt 2: # 最后一次尝试 raise wait_time 2 ** attempt # 指数退避 print(f请求失败{wait_time}秒后重试... 错误: {e}) time.sleep(wait_time) return {error: 请求失败} def safe_predict(self, text: str, fallback_model: str A001) - Dict: 安全预测自动降级 models_to_try [A204, A021, A001] # 从高精度到低精度 for model_id in models_to_try: try: result self.predict_with_retry(text, model_id) if result.get(confidence, 0) 0.5: # 置信度阈值 return result except: continue # 尝试下一个模型 # 所有模型都失败返回中性结果 return { model_id: fallback, emotion: neutral, confidence: 0.0, timestamp: time.time() }7.3 缓存策略优化频繁调用相同文本的分析可以加入缓存import redis import hashlib import json from functools import lru_cache class CachedAnalyzer: def __init__(self, api_url: str, redis_host: str localhost): self.api_url api_url self.redis_client redis.Redis( hostredis_host, port6379, db0, decode_responsesTrue ) self.cache_ttl 3600 # 缓存1小时 def _get_cache_key(self, text: str, model_id: str) - str: 生成缓存键 content f{model_id}:{text} return hashlib.md5(content.encode()).hexdigest() lru_cache(maxsize1000) def predict_with_cache(self, text: str, model_id: str A001) - Dict: 带缓存的预测 cache_key self._get_cache_key(text, model_id) # 尝试从Redis获取缓存 cached_result self.redis_client.get(cache_key) if cached_result: return json.loads(cached_result) # 调用API response requests.post( f{self.api_url}/predict, json{model_id: model_id, input_data: text} ) if response.status_code 200: result response.json() # 缓存结果 self.redis_client.setex( cache_key, self.cache_ttl, json.dumps(result) ) return result return {error: 请求失败} def batch_predict_with_cache(self, texts: List[str], model_id: str A001) - List[Dict]: 批量预测带缓存 results [] uncached_texts [] uncached_indices [] # 检查缓存 for i, text in enumerate(texts): cache_key self._get_cache_key(text, model_id) cached self.redis_client.get(cache_key) if cached: results.append(json.loads(cached)) else: uncached_texts.append(text) uncached_indices.append(i) results.append(None) # 占位 # 批量请求未缓存的内容 if uncached_texts: response requests.post( f{self.api_url}/predict/batch, json{model_id: model_id, inputs: uncached_texts} ) if response.status_code 200: batch_results response.json().get(predictions, []) # 更新结果和缓存 for idx, result in zip(uncached_indices, batch_results): results[idx] result # 缓存新结果 cache_key self._get_cache_key(uncached_texts[idx], model_id) self.redis_client.setex( cache_key, self.cache_ttl, json.dumps(result) ) return results7.4 监控与日志生产环境需要完善的监控import logging from prometheus_client import Counter, Histogram, start_http_server import time # 定义监控指标 REQUEST_COUNT Counter(m2lorder_requests_total, Total API requests) REQUEST_LATENCY Histogram(m2lorder_request_latency_seconds, Request latency) ERROR_COUNT Counter(m2lorder_errors_total, Total errors) # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(m2lorder_client.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) class MonitoredAPIClient: def __init__(self, api_url: str): self.api_url api_url def predict_with_monitoring(self, text: str, model_id: str) - Dict: 带监控的预测请求 REQUEST_COUNT.inc() start_time time.time() try: response requests.post( f{self.api_url}/predict, json{model_id: model_id, input_data: text}, timeout10 ) latency time.time() - start_time REQUEST_LATENCY.observe(latency) if response.status_code 200: logger.info(f预测成功: model{model_id}, latency{latency:.3f}s) return response.json() else: ERROR_COUNT.inc() logger.error(f预测失败: status{response.status_code}) return {error: fHTTP {response.status_code}} except Exception as e: ERROR_COUNT.inc() logger.error(f请求异常: {str(e)}) return {error: str(e)} # 启动监控服务器在单独线程中 def start_monitoring_server(port: int 8000): 启动Prometheus监控服务器 start_http_server(port) logger.info(f监控服务器已启动端口: {port}) # 使用示例 if __name__ __main__: # 启动监控 import threading monitor_thread threading.Thread(targetstart_monitoring_server) monitor_thread.daemon True monitor_thread.start() # 使用带监控的客户端 client MonitoredAPIClient(http://localhost:8001) # 测试请求 result client.predict_with_monitoring(测试文本, A001) print(result)8. 总结从入门到精通的完整路径通过这篇文章我们从最基础的M2LOrder服务启动一直讲到了高级的生产环境部署和优化。让我们回顾一下关键要点8.1 核心要点回顾快速启动三种启动方式中./start.sh最适合初学者Supervisor适合生产环境。模型选择97个模型看似复杂但记住这个简单原则测试/演示用A001-A0123-4MB最快日常使用选A021-A0317-8MB平衡高精度需求用A204-A236619MB最准API使用两个核心接口要掌握/predict- 单条文本分析/predict/batch- 批量文本分析调试技巧Swagger文档是你的最佳调试伙伴可视化操作比curl命令更直观。8.2 实战建议根据不同的使用场景我建议个人学习/测试用WebUI快速体验用Swagger熟悉API从A001模型开始逐步尝试其他模型项目集成先用curl命令测试连通性编写封装类处理错误和重试根据性能需求选择合适的模型添加缓存提升响应速度生产部署使用Supervisor管理进程配置监控和告警实现负载均衡如果需要定期备份重要数据8.3 常见问题解决如果你遇到问题可以按这个顺序排查服务无法启动检查端口是否被占用查看日志文件API调用失败确认服务地址和端口检查网络连接响应时间慢换用更小的模型或优化网络配置分析结果不准尝试不同的模型或预处理输入文本8.4 下一步学习方向掌握了基础用法后你可以进一步探索模型微调如果有标注数据可以训练自己的情感分析模型多语言支持测试M2LOrder对其他语言的情感分析效果集成其他服务将情感分析接入客服系统、社交媒体监控等性能优化研究模型压缩、量化等技术提升推理速度情感分析是一个很有价值的AI应用场景M2LOrder提供了一个很好的起点。无论你是想快速验证想法还是构建正式的产品功能这套工具都能帮到你。记住最好的学习方式就是动手实践。从简单的curl命令开始逐步构建完整的应用。如果在使用过程中遇到问题欢迎在评论区交流讨论。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。