deepseek+cherryStudio+MCP构建一个自定义数据库数据分析

一、数据准备

还是常用的那个员工数据库，导入mysql数据库

二、准备MCP服务

开发几个MCP工具，这是成功关键的第一步，可以让deepseek完全按照我们定义的数据库表结构进行数据查询、处理分析。

1、获取所有表信息，这里返回表名、表注释，想要效果好，最好把表的注释写的更详细

@mcp.tool()async def list_tables() -> str:    """List MySQL tables with their comments."""    try:        with get_db_connection() as conn:            with conn.cursor() as cursor:                # Query to get table names and their comments                query = """                SELECT TABLE_NAME, TABLE_COMMENT                FROM INFORMATION_SCHEMA.TABLES                WHERE TABLE_SCHEMA = %s                """                cursor.execute(query, (get_db_config()["database"],))                tables = cursor.fetchall()                logger.info(f"Found tables: {tables}")                # Format the output to include both table name and comment                return "\n".join([f"{table[0]}: {table[1]}" for table in tables])    except Error as e:        logger.error(f"Failed to list tables: {str(e)}")        return f"Failed to list tables: {str(e)}"

2、获取表结构，包括字段名、注释、数据类型及长度等

@mcp.tool()async def get_table_structure(table: str) -> str:    """    Get the structure of a MySQL table.    Args:        table (str): The name of the database table.    Returns:        str: The structure information of the table, or an error message if it fails.    """    # Validate whether the input conforms to MySQL table naming rules    if not validate_table_name(table):        logger.error(f"Invalid table name: {table}")        return f"Invalid table name: {table}"    try:        # Use an asynchronous context manager to manage the database connection        with get_db_connection() as conn:            with conn.cursor() as cursor:                # Use parameterized queries to prevent SQL injection (assuming support)                query = """                SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, COLUMN_COMMENT                FROM INFORMATION_SCHEMA.COLUMNS                WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s                """                cursor.execute(query, (get_db_config()["database"], table))                columns: List[Tuple[str, str, int, str]] = cursor.fetchall()                # Log the table structure information                structure_info = "\n".join([f"{col[0]} {col[1]} {col[2]} {col[3]}" for col in columns])                logger.info(f"Structure of table {table}: {structure_info}")                return structure_info    except Error as e:        # Catch database-related exceptions and log detailed information        logger.error(f"Failed to get structure of table {table}: {str(e)}", exc_info=True)        return f"Failed to get structure of table {table}: {str(e)}"

3、查询表数据，参数是表名

@mcp.tool()async def read_table_data(table: str) -> str:    """Read data from a MySQL table."""    try:        with get_db_connection() as conn:            with conn.cursor() as cursor:                cursor.execute(f"SELECT * FROM {table} LIMIT 100")                columns = [desc[0] for desc in cursor.description]                rows = cursor.fetchall()                result = [",".join(map(str, row)) for row in rows]                logger.info(f"Data from table {table}: {result}")                return "\n".join([",".join(columns)] + result)    except Error as e:        logger.error(f"Failed to read data from table {table}: {str(e)}")        return f"Failed to read data from table {table}: {str(e)}"

4、执行生成的sql语句

@mcp.tool()async def execute_sql(query: str) -> str:    """Execute an SQL query on the MySQL server."""    try:        with get_db_connection() as conn:            with conn.cursor() as cursor:                cursor.execute(query)                # Special handling for SHOW TABLES                if query.strip().upper().startswith("SHOW TABLES"):                    tables = cursor.fetchall()                    result = ["Tables_in_" + get_db_config()["database"]]  # Header                    result.extend([table[0] for table in tables])                    logger.info(f"Result of query {query}: {result}")                    return "\n".join(result)                # Regular SELECT queries                elif query.strip().upper().startswith("SELECT"):                    columns = [desc[0] for desc in cursor.description]                    rows = cursor.fetchall()                    result = [",".join(map(str, row)) for row in rows]                    logger.info(f"Result of query {query}: {result}")                    return "\n".join([",".join(columns)] + result)                # Non-SELECT queries                else:                    conn.commit()                    logger.info(f"Query {query} executed successfully. Rows affected: {cursor.rowcount}")                    return f"Query executed successfully. Rows affected: {cursor.rowcount}"    except Error as e:        logger.error(f"Error executing SQL '{query}': {e}")        return f"Error executing query: {str(e)}"

5、数据图表样例数据工具，这样分析后的数据就可以根据不同的显示要求解析成不同的数据格式，然后展示。

@mcp.tool()def generate_chart_data(chart_type):    """    根据图表类型生成不同格式的图表数据。    参数:    - chart_type (str): 图表类型，可选值为 'bar', 'line', 'pie' 或 'scatter'。    - data (list): 图表数据，格式为 [{'name': '数据名称', 'value': '数据值'}, ...]。    返回:    - dict: 生成的图表数据，格式为 {'title': '图表标题', 'data': [{'name': '数据名称', 'value': '数据值'}, ...]}。    ## 图表配置模板    ### 柱状图示例    ```echarts    {      "title": {        "text": "部门人数分布",        "left": "center"      },      "tooltip": {        "trigger": "axis"      },      "xAxis": {        "type": "category",        "data": ["开发部", "市场部", "财务部"]      },      "yAxis": {"type": "value"},      "series": [{        "name": "人数",        "type": "bar",        "data": [15, 8, 5],        "itemStyle": {"color": "#5470C6"}      }]    }    ```    ### 饼图示例    ```echarts    {      "title": {        "text": "性别比例分析",        "left": "center"      },      "tooltip": {"trigger": "item"},      "legend": {        "orient": "vertical",        "left": "left",        "data": ["男性", "女性"]      },      "series": [{        "name": "比例",        "type": "pie",        "radius": "50%",        "data": [          {"value": 65, "name": "男性"},          {"value": 35, "name": "女性"}        ]      }]    }    ```    ### 折线图示例    ```echarts    {      "title": {        "text": "月度入职趋势",        "left": "center"      },      "tooltip": {"trigger": "axis"},      "xAxis": {        "type": "category",        "data": ["1月","2月","3月","4月"]      },      "yAxis": {"type": "value"},      "series": [{        "name": "入职人数",        "type": "line",        "smooth": true,        "data": [12, 18, 9, 15],        "itemStyle": {"color": "#91CC75"}      }]    }    ```    """    if chart_type == 'line':        return {            "title": {                "text": "月度入职趋势",                "left": "center"            },            "tooltip": {"trigger": "axis"},            "xAxis": {                "type": "category",                "data": ["1月", "2月", "3月", "4月"]            },            "yAxis": {"type": "value"},            "series": [{                "name": "入职人数",                "type": "line",                "smooth": True,                "data": [12, 18, 9, 15],                "itemStyle": {"color": "#91CC75"}            }]        }    elif chart_type == 'pie':        return {            "title": {                "text": "性别比例分析",                "left": "center"            },            "tooltip": {"trigger": "item"},            "legend": {                "orient": "vertical",                "left": "left",                "data": ["男性", "女性"]            },            "series": [{                "name": "比例",                "type": "pie",                "radius": "50%",                "data": [                    {"value": 65, "name": "男性"},                    {"value": 35, "name": "女性"}                ]            }]        }    elif chart_type == 'bar':        return {            "title": {                "text": "部门人数分布",                "left": "center"            },            "tooltip": {                "trigger": "axis"            },            "xAxis": {                "type": "category",                "data": ["开发部", "市场部", "财务部"]            },            "yAxis": {"type": "value"},            "series": [{                "name": "人数",                "type": "bar",                "data": [15, 8, 5],                "itemStyle": {"color": "#5470C6"}            }]        }    else:        raise ValueError('Invalid chart type')

三、写提示词

完善的提示词是最终成功的关键，这里可以先写一个大概的提示词，然后使用deepseek反向优化一下。

初稿

你是一个DBA及数据分析师，精通mysql数据库，擅长根据用户的提问写出高性能 的SQL语句，并对查询结果进行数据分析，按用户的要求进行展示，如果用户没有要求展示方式，请使用表格展示。以下是一些限制1、请通过MCP服务查询出数据表及字段注释2、分析用户的提问，解析后从上面获取的表及字段注释中找出所用到的表及字段。3、生成高性能的SQL语句，满足mysql8的语法要求4、使用explain先分析生成的sql可用性及性能5、对不可用的SQL重新生成，对低性能的sql进行优化6、重复3、4和5三条确定最终sql语句7、执行sql语句，最大结果不要超过100

然后我使用gemma3:27b对以上的提示词进行了优化，最终提示词如下：

你是一个DBA及数据分析师，精通mysql数据库，擅长根据用户的提问写出高性能 的SQL语句，并对查询结果进行数据分析，按用户的要求进行展示，如果用户没有要求展示方式，请使用表格展示。请严格按照以下步骤执行：1. **获取元数据：** 使用MCP服务查询 {{database_name}} 数据库中所有数据表及其字段注释。将这些信息存储起来，作为后续步骤的参考。2. **问题解析：** 仔细分析用户提出的问题 {{user_question}}。理解问题的意图，并识别问题中涉及的关键信息和需要查询的字段。3. **表字段映射：**  根据问题解析的结果，从步骤1获取的表字段注释中，找出与问题相关的表和字段。记录下这些表和字段的名称及其注释，以便后续构建SQL语句。4. **SQL语句生成（初稿）：** 基于步骤3的结果，编写一个初步的SQL语句，以满足用户的问题。确保SQL语句符合MySQL 8的语法要求。5. **SQL语句分析（EXPLAIN）：** 使用 `EXPLAIN` 命令分析生成的SQL语句的执行计划。评估SQL语句的可用性和性能。关注 `type`、`possible_keys`、`key`、`rows` 等关键指标。6. **SQL语句优化/重写：**   * **如果 `EXPLAIN` 分析结果表明SQL语句不可用（例如，语法错误、表不存在等）：**  修复错误，重新生成SQL语句，并返回步骤5。   * **如果 `EXPLAIN` 分析结果表明SQL语句性能较低（例如，全表扫描、使用了大量临时表等）：**  优化SQL语句，例如：      * 添加合适的索引。      * 优化JOIN顺序。      * 避免使用 `SELECT *`。      * 尽量减少子查询。      * 使用 `WHERE` 子句过滤数据。      * 考虑使用 `LIMIT` 限制结果集大小。      然后返回步骤5。7. **循环优化：** 重复步骤5和步骤6，最多经过三次尝试先出一个最好的Sql语句执行。8. **SQL语句执行：** 执行最终确定的SQL语句。9. **结果处理：**   * 如果用户指定了展示方式 ，如echarts的饼图、柱状图、折线图等

可以参照上面的写法再继续进行优化。