如何在Rasa聊天助手展示MySQL数据并按product_id查询全列信息

1. 准备依赖 #

首先安装MySQL Python驱动：

pip install mysql-connector-python

确保你的Rasa环境已正确配置（已安装Rasa及相关依赖）。

2. 封装数据库连接与查询函数 #

创建db_utils.py文件，封装数据库连接和产品查询逻辑：

import mysql.connector
from mysql.connector import Error

def create_db_connection():
    try:
        connection = mysql.connector.connect(
            host='你的数据库IP/域名',
            database='目标数据库名',
            user='数据库用户名',
            password='数据库密码'
        )
        if connection.is_connected():
            return connection
    except Error as e:
        print(f"数据库连接失败: {e}")
        return None

def get_product_by_id(product_id):
    connection = create_db_connection()
    if not connection:
        return None
    try:
        cursor = connection.cursor(dictionary=True)
        # 假设产品表名为products，根据实际表名调整
        query = "SELECT * FROM products WHERE product_id = %s"
        cursor.execute(query, (product_id,))
        product_info = cursor.fetchone()
        return product_info
    except Error as e:
        print(f"查询执行失败: {e}")
        return None
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()

3. 编写自定义动作处理查询 #

在actions.py中实现处理产品查询的自定义动作：

from typing import Any, Text, Dict, List
from rasa_sdk import Action, Tracker
from rasa_sdk.executor import CollectingDispatcher
from db_utils import get_product_by_id

class ActionFetchProductDetails(Action):
    def name(self) -> Text:
        return "action_fetch_product_details"

    def run(self, dispatcher: CollectingDispatcher,
            tracker: Tracker,
            domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
        
        # 提取用户输入的product_id槽位
        product_id = tracker.get_slot("product_id")
        
        # 若未获取到product_id，提示用户输入
        if not product_id:
            dispatcher.utter_message(text="请提供要查询的product_id")
            return []
        
        # 查询数据库
        product_data = get_product_by_id(product_id)
        
        # 构造回复消息
        if product_data:
            response_text = "产品详情如下:\n"
            for column, value in product_data.items():
                response_text += f"- {column}: {value}\n"
            dispatcher.utter_message(text=response_text)
        else:
            dispatcher.utter_message(text=f"未找到product_id为{product_id}的产品信息")
        
        return []

4. 配置Rasa域文件（domain.yml） #

更新domain.yml，添加意图、槽位和实体定义：

intents:
  - query_product:
      triggers: action_fetch_product_details

slots:
  product_id:
    type: text
    influence_conversation: false
    mappings:
      - type: from_entity
        entity: product_id

entities:
  - product_id

responses:
  utter_ask_product_id:
    - text: "请告诉我你要查询的product_id"

5. 添加NLU训练数据（nlu.yml） #

在nlu.yml中添加用户查询的示例语句，帮助Rasa识别意图和实体：

nlu:
- intent: query_product
  examples: |
    - 查询product_id为1001的产品
    - 帮我查product_id是2002的详情
    - product_id 3003的信息是什么
    - 查ID为4004的产品数据
    - 我想知道product_id=5005的产品情况

6. 训练并运行Rasa #

1. 训练Rasa模型：

rasa train

2. 启动动作服务器（自定义动作需要单独运行）：

rasa run actions

3. 启动聊天界面测试：

rasa shell

测试时，用户输入类似“查询product_id为1001的产品”，机器人会自动查询MySQL数据库并返回对应产品的所有列信息。

内容的提问来源于stack exchange，提问作者Bhaskar Gupta