当前位置：首页 > news >正文

Gemma-3-270m实战落地：为制造业MES系统添加自然语言工单查询入口

news 2026/8/28 8:22:43

Gemma-3-270m实战落地：为制造业MES系统添加自然语言工单查询入口

1. 引言：让MES系统听懂人话

想象一下这个场景：车间主任老张站在生产线旁，想快速了解"上个月华为订单P20型号还有多少未完成"，传统MES系统需要他进入层层菜单，选择时间范围、客户名称、产品型号、状态筛选...等找到结果时，半小时过去了。

这就是制造业MES系统的普遍痛点——功能强大但操作复杂，非技术人员难以快速获取所需信息。现在，有了Gemma-3-270m这个轻量级AI模型，我们可以为现有MES系统添加一个自然语言查询入口，让任何一线人员都能用最自然的方式获取工单信息。

本文将手把手带你实现这个功能，不需要深度学习专家，只需要基本的Python编程能力，就能让你们的MES系统"听懂人话"。

2. 环境准备与快速部署

2.1 系统要求与安装

Gemma-3-270m对硬件要求很友好，以下是最低配置：

CPU：4核以上（Intel i5或同等性能）
内存：8GB RAM
存储：2GB可用空间
系统：Linux/Windows/macOS均可

安装Ollama只需一行命令：

# Linux/macOS curl -fsSL https://ollama.ai/install.sh | sh # Windows winget install Ollama.Ollama

2.2 部署Gemma-3-270m模型

安装完成后，拉取Gemma-3-270m模型：

ollama pull gemma3:270m

这个270M的模型大小适中，下载很快，一般几分钟就能完成。部署成功后，你可以测试一下：

ollama run gemma3:270m "你好，请自我介绍"

如果看到模型回复，说明部署成功！

3. MES系统自然语言查询实现

3.1 整体架构设计

我们的目标不是重建MES系统，而是在现有系统上加一个智能查询层：

自然语言输入 → Gemma理解意图 → 转换为SQL查询 → MES数据库执行 → 结果返回

这种设计的好处是：

不改变现有MES系统结构
部署简单，风险低
维护方便，出了问题可以快速切换回传统方式

3.2 核心代码实现

首先安装必要的Python库：

pip install ollama python-dotenv sqlalchemy

创建主程序文件mes_nl_query.py：

import ollama import re from datetime import datetime, timedelta class MESQueryAssistant: def __init__(self): self.model_name = "gemma3:270m" def parse_natural_language(self, query): """将自然语言转换为结构化查询条件""" prompt = f""" 你是一个MES系统查询助手，请将用户的自然语言查询转换为结构化条件。 用户查询：{query} 请以JSON格式返回，包含以下字段： - customer: 客户名称（如华为、苹果） - product_model: 产品型号（如P20、iPhone15） - date_range: 时间范围（如last_week, last_month, specific_date） - status: 工单状态（如pending, completed, all） - quantity_type: 查询类型（如remaining, completed, total） 示例输出： {{"customer": "华为", "product_model": "P20", "date_range": "last_month", "status": "pending", "quantity_type": "remaining"}} """ response = ollama.chat(model=self.model_name, messages=[ {'role': 'user', 'content': prompt} ]) # 提取JSON格式的响应 result = self._extract_json(response['message']['content']) return result def _extract_json(self, text): """从文本中提取JSON内容""" import json try: # 尝试查找JSON对象 json_match = re.search(r'\{.*\}', text, re.DOTALL) if json_match: return json.loads(json_match.group()) return {} except: return {} def generate_sql_query(self, conditions): """根据条件生成SQL查询""" # 这里是示例，实际需要根据你的数据库结构调整 base_query = "SELECT COUNT(*) as count, SUM(quantity) as total FROM work_orders WHERE 1=1" if conditions.get('customer'): base_query += f" AND customer_name = '{conditions['customer']}'" if conditions.get('product_model'): base_query += f" AND product_model = '{conditions['product_model']}'" # 处理时间范围 if conditions.get('date_range') == 'last_month': first_day = (datetime.now().replace(day=1) - timedelta(days=1)).replace(day=1) last_day = datetime.now().replace(day=1) - timedelta(days=1) base_query += f" AND create_date BETWEEN '{first_day}' AND '{last_day}'" # 处理状态 status_map = {'pending': 'pending', 'completed': 'completed'} if conditions.get('status') in status_map: base_query += f" AND status = '{status_map[conditions['status']]}'" return base_query # 使用示例 assistant = MESQueryAssistant() user_query = "查一下上个月华为P20还有多少未完成工单" conditions = assistant.parse_natural_language(user_query) print("解析结果:", conditions) sql_query = assistant.generate_sql_query(conditions) print("生成SQL:", sql_query)

3.3 数据库连接与查询执行

添加数据库连接逻辑（以MySQL为例）：

import mysql.connector from mysql.connector import Error def execute_query(sql_query): """执行SQL查询并返回结果""" try: connection = mysql.connector.connect( host='your_mes_db_host', database='your_mes_database', user='your_username', password='your_password' ) if connection.is_connected(): cursor = connection.cursor() cursor.execute(sql_query) results = cursor.fetchall() return results except Error as e: print(f"数据库错误: {e}") return None finally: if connection.is_connected(): cursor.close() connection.close() # 整合到主类中 class MESQueryAssistant: # ... 之前的代码 ... def query_mes(self, natural_language_query): """完整的查询流程""" print(f"用户查询: {natural_language_query}") # 解析自然语言 conditions = self.parse_natural_language(natural_language_query) print(f"解析条件: {conditions}") # 生成SQL sql = self.generate_sql_query(conditions) print(f"生成SQL: {sql}") # 执行查询 results = execute_query(sql) # 格式化结果 return self.format_results(results, conditions) def format_results(self, results, conditions): """将查询结果格式化为自然语言回复""" if not results: return "抱歉，没有找到相关工单信息。" count, total = results[0] response_prompt = f""" 根据以下查询结果生成友好的中文回复： 查询条件：{conditions} 找到工单数量：{count} 总数量：{total} 请用简洁明了的中文回复用户，不要透露技术细节。 """ response = ollama.chat(model=self.model_name, messages=[ {'role': 'user', 'content': response_prompt} ]) return response['message']['content']

4. 实际应用演示

4.1 测试不同查询场景

让我们测试几个典型的车间查询场景：

# 测试用例 test_queries = [ "今天有多少苹果订单要完成？", "查看上周小米手机的生产进度", "华为P30还有多少没做完？", "这个月完成了多少OPPO订单？" ] assistant = MESQueryAssistant() for query in test_queries: print(f"\n=== 测试查询: {query} ===") result = assistant.query_mes(query) print(f"回复: {result}") print("=" * 50)

4.2 Web接口集成

为了让车间人员方便使用，我们可以添加一个简单的Web界面：

from flask import Flask, request, jsonify app = Flask(__name__) assistant = MESQueryAssistant() @app.route('/api/mes-query', methods=['POST']) def mes_query(): data = request.json query_text = data.get('query', '') if not query_text: return jsonify({'error': '请输入查询内容'}) try: result = assistant.query_mes(query_text) return jsonify({'result': result}) except Exception as e: return jsonify({'error': str(e)}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)

这样，车间人员可以通过手机或电脑访问这个接口，用自然语言查询工单信息。

5. 部署与优化建议

5.1 生产环境部署

对于生产环境，建议：

使用GPU加速：如果查询量大，可以考虑使用带GPU的服务器
添加缓存机制：对常见查询结果进行缓存，提高响应速度
设置超时限制：避免复杂查询占用过多资源
添加权限控制：不同岗位的人员只能查询权限内的信息

5.2 性能优化技巧

# 添加查询缓存 from functools import lru_cache class MESQueryAssistant: @lru_cache(maxsize=100) def parse_natural_language(self, query): # 原有的解析逻辑 pass # 批量处理查询 def batch_query(self, queries): """批量处理多个查询，提高效率""" results = [] for query in queries: results.append(self.query_mes(query)) return results