构建AI智能体：智能数据分析机器人：基于Qwen-Agent与Text2SQL的门票分析方案

原创

未闻花名

发布于 2025-12-05 07:57:51

3230

一、系统概述

随着企业数字化进程的加速，数据已成为决策的重要依据，但传统数据分析流程存在诸多痛点。业务人员往往需要依赖专业数据团队进行SQL查询和报表制作，这个过程耗时耗力且响应迟缓，沟通成本高、技术门槛也高。特别是在门票销售这样的实时性要求较高的行业，快速获取数据洞察对业务决策至关重要。

结合昨天我们刚讲过的Qwen-Agent和前期讲过的Text2SQL技术以及Gradio前端展示技术，构建了一个智能化的门票数据分析平台，让业务人员能够通过自然语言直接进行数据查询和分析，大幅降低了技术门槛，提高了决策效率。旨在通过自然语言交互实现数据库查询和可视化分析。

系统结合了大语言模型、数据库操作和数据可视化技术，为用户提供直观的门票销售数据分析体验。该系统不仅解决了传统数据分析流程的痛点，更为企业级智能数据分析应用提供了新的思路和方案。

系统采用分层架构设计，确保各模块之间的松耦合和高内聚。整体架构包含四个核心层次：

用户交互层：基于Gradio构建的Web界面，提供直观的聊天式交互体验。该层负责捕获用户输入、渲染查询结果，并管理用户会话状态。
智能代理层：系统的"大脑"，基于Qwen-Agent框架实现。该层负责自然语言理解、对话管理、工具调用决策和结果整合。
工具执行层：模块化的工具系统，核心包括SQL执行工具和可视化生成工具。该层负责具体的业务逻辑执行和数据处理。
数据服务层：包括MySQL数据库和文件系统，为系统提供数据存储和持久化支持。

二、核心组件

1. Qwen-Agent智能代理

上下文感知：能够理解多轮对话的上下文关系
意图识别：准确识别用户的查询意图和业务需求
工具协调：智能决策何时以及如何调用工具函数
结果整合：将工具执行结果整合成用户友好的格式

2. Text2SQL转换引擎

自然语言理解：解析用户查询的语义结构，识别意图、实体和条件
模式映射：将自然语言中的概念映射到数据库中的表、列和关系
SQL生成：构建符合目标数据库方言的正确语法结构
查询优化：确保生成的SQL在执行时具有良好性能

3. Gradio前端展示

简单易用：界面简单、设计友好，几行代码即可创建功能完整的Web界面
多样化组件：支持文本、图像、音频、视频等多种输入输出格式
即时分享：一键生成可公开访问的链接
框架无关：可与TensorFlow、PyTorch、Scikit-learn等任何机器学习框架配合使用
高度可定制：提供灵活的自定义选项满足不同需求和应用场景

三、系统流程

1. 流程图

2. 流程分解

2.1 基础流程

第一步：用户输入与消息传递 (步骤1-2)

用户输入：用户在Web界面输入自然语言问题，如"查询2023年门票销售情况"
消息捕获：WebUI组件捕获用户输入，格式化为标准消息格式
会话管理：根据对话历史生成唯一的session_id，确保多用户隔离

第二步：智能分析与SQL生成 (步骤3-4)

意图理解：Agent结合system_prompt中的业务知识，理解用户查询意图
上下文整合：如果有历史对话，会结合上下文进行更准确的理解
SQL生成：基于数据表结构和业务逻辑，生成符合规范的SQL查询语句
工具调用决策：判断需要调用exc_sql工具执行数据查询

第三步：数据查询执行 (步骤5-7)

数据库连接：从连接池获取MySQL数据库连接，设置超时和字符集
SQL执行：使用pandas的read_sql方法执行查询，返回DataFrame
结果验证：检查查询结果的有效性，处理空结果或异常情况

第四步：可视化与结果组装 (步骤8-10)

1. 智能可视化：

自动分析数据类型（分类变量vs数值变量）
根据数据特征选择图表类型（普通柱状图vs堆积柱状图）
处理中文显示和格式美化
文件保存：将生成的图表保存为PNG文件，使用时间戳确保文件名唯一

2. 结果组装：

将DataFrame转换为Markdown表格格式
生成图片的Markdown引用链接
组合表格和图片为完整响应

第五步：结果返回与显示 (步骤11-12)

结果传递：工具结果逐级返回给Agent和WebUI
界面渲染：WebUI将Markdown内容渲染为美观的显示格式
用户交互：用户可以看到结构化数据和可视化图表，可进行后续交互

2.2 自然语言到SQL的转换流程

意图解析：理解用户查询的业务背景和具体需求
条件提取：识别时间范围、筛选条件、分组维度等关键要素
SQL构造：根据数据库schema构建符合语法的SQL语句
优化验证：检查SQL的合理性和执行效率

2.3 Agent创建过程

配置LLM参数：指定模型版本、超时设置、重试策略
实例化Assistant：传入名称、描述、系统提示词等
注册工具函数：将'exc_sql'工具绑定到Agent
返回初始化完成的bot对象

系统运行界面：

查询过程中生成的图示：

此处也可导入echarts的组件，达到动态炫酷的展示效果！

四、核心代码分解

1. Assistant智能代理

llm_cfg = {
        'model': 'qwen-turbo',
        'timeout': 30,
        'retry_count': 3,
    }
bot = Assistant(
    llm=llm_cfg,
    name='门票助手',
    description='门票查询与订单分析',
    system_message=system_prompt,
    function_list=['exc_sql'],
)

使用qwen-turbo模型作为语言理解核心
通过system_prompt定义专业领域知识
函数调用机制实现工具扩展

2. 系统提示词设计

system_prompt = """我是门票助手，以下是关于门票订单表相关的字段，我可能会编写对应的SQL，对数据进行查询
-- 门票订单表
CREATE TABLE tkt_orders (
    order_time DATETIME,             -- 订单日期
    ......
    quantity INT                     -- 商品数量
);
一日门票，对应多种SKU：
Universal Studios Beijing One-Day Dated Ticket-Standard
Universal Studios Beijing One-Day Dated Ticket-Child
Universal Studios Beijing One-Day Dated Ticket-Senior

设计原则：明确的角色定义、数据结构说明、SQL查询模式示例、输出格式规范
领域专业化：系统提示词不仅定义了助手角色，更包含了详细的数据表结构说明和业务逻辑，使模型能够理解门票业务的特殊性。
查询模式预置：通过提供常见的SQL查询模板，如一日门票、二日门票的统计方式，引导模型生成符合业务需求的查询语句。
输出规范约束：明确要求原样输出工具返回内容，避免模型过度"聪明"地总结或简化，确保用户获得完整的数据信息。

3. SQLAlchemy引擎配置

engine = create_engine(
    f'mysql+mysqlconnector://root:Aa123456!@localhost:3306/{database}?charset=utf8mb4',
    connect_args={'connect_timeout': 10}, pool_size=10, max_overflow=20
)

连接参数详解：

连接字符串格式：数据库类型+驱动://用户名:密码@主机:端口/数据库名
连接池配置：pool_size控制连接数，max_overflow控制超额连接
字符集设置：utf8mb4支持完整Unicode字符
并发支持：pool_size=10确保系统能同时处理多个查询请求
弹性扩展：max_overflow=20在高峰期提供额外连接缓冲
故障隔离：超时设置防止单次查询阻塞整个系统

4. 数据查询执行

df = pd.read_sql(sql_input, engine)

使用pandas的read_sql方法直接执行SQL
自动将结果转换为DataFrame格式
支持复杂的SQL查询语句

5. 可视化函数架构

def generate_chart_png(df_sql, save_path):
    # 数据类型识别与处理
    object_columns = df_sql.select_dtypes(include='O').columns.tolist()
    num_columns = df_sql.select_dtypes(exclude='O').columns.tolist()
    
    # 智能图表类型选择
    if len(object_columns) > 0:
        # 堆积柱状图逻辑
        pivot_df = df_sql.pivot_table(...)
    else:
        # 普通柱状图逻辑
        bottom = np.zeros(len(df_sql))

可视化策略：

自动识别数据类型：系统能够区分分类变量和数值变量，这是选择合适图表类型的基础
智能图表选择：根据数据特征选择最佳可视化方式
- 当数据包含多个分类维度时，自动选择堆积柱状图展示复合关系
- 单一维度数据使用普通柱状图，保持图表简洁性
透视表自动生成：通过pivot_table实现数据重组，满足复杂多维度的可视化需求

6. 中文显示解决方案

plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'SimSun', 'Arial Unicode MS']
plt.rcParams['axes.unicode_minus'] = False

设置中文字体优先级列表
解决负号显示异常问题
确保图表中文字符正常渲染

7. 函数的调用机制

functions_desc = [
    {
        "name": "exc_sql",
        "description": "对于生成的SQL，进行SQL查询",
        # 参数定义确保SQL语句的正确传递
    }
]

能力边界清晰：语言模型负责理解意图和生成SQL，专业工具负责执行和可视化
安全性保障：通过参数校验确保输入的SQL符合预期格式
可扩展性强：新的分析功能可以通过添加工具函数快速集成

8. @register_tool装饰器

@register_tool('exc_sql')
class ExcSQLTool(BaseTool):
    description = '对于生成的SQL，进行SQL查询，并自动可视化'
    parameters = [{'name': 'sql_input', 'type': 'string', 'required': True}]

使用装饰器注册工具名称
继承BaseTool基类
定义工具描述和参数规范

9. 工具调用接口

def call(self, params: str, **kwargs) -> str:
    args = json.loads(params)
    sql_input = args['sql_input']

JSON格式参数解析
异常处理与错误返回
统一的返回格式

10. 使用建议与默认提示

chatbot_config = {
    'prompt.suggestions': [
        '2023年4、5、6月一日门票，二日门票的销量多少？帮我按照周进行统计',
        '2023年7月的不同省份的入园人数统计',
        '帮我查看2023年10月1-7日销售渠道订单金额排名',
    ]
}

预置典型查询问题
降低用户学习成本
引导用户使用模式

11. 异常处理机制

try:
    df = pd.read_sql(sql_input, engine)
    # 正常处理逻辑
except Exception as e:
    return f"SQL执行或可视化出错: {str(e)}"

数据库连接异常捕获
SQL执行错误处理
可视化过程异常管理

12. 性能优化

dashscope.timeout = 30  # API调用超时设置
connect_args={'connect_timeout': 10}  # 数据库连接超时
pool_size=10, max_overflow=20  # 连接池配置

合理的超时设置
数据库连接池管理
资源释放与清理

五：优化与不足

针对SQL生成可能出现的错误，系统建立了多级校验机制。包括语法检查、执行验证和错误反馈循环，确保生成SQL的准确性和安全性。
通过异步处理、连接池管理和缓存策略优化系统性能。数据库查询使用连接池复用连接，图表生成结果进行缓存，避免重复计算。
采用会话隔离和资源池化技术支持多用户并发访问。每个会话有独立的数据空间，关键资源通过池化管理，确保系统稳定运行。

六、总结

本项目成功构建了一个基于Qwen-Agent和Text2SQL的智能门票数据分析系统。通过自然语言交互大幅降低了数据分析的技术门槛，提高了业务决策效率，自动化报表生成，减少人工数据处理工作量，让非技术人员也能深度参与数据分析过程。

系统展示了大语言模型在企业级应用中的巨大潜力。基于提示工程的Text2SQL方案为类似项目提供了新的技术思路，避免了专门模型训练的复杂性。下一步将扩展系统分析能力，支持预测分析和异常检测等高级功能。同时探索更多应用场景，将这一技术方案推广到其他行业领域。

附录：完整实例代码

import os
import asyncio
from typing import Optional
import dashscope
from qwen_agent.agents import Assistant
from qwen_agent.gui import WebUI
import pandas as pd
from sqlalchemy import create_engine
from qwen_agent.tools.base import BaseTool, register_tool
import matplotlib.pyplot as plt
import io
import base64
import time
import numpy as np

# 解决中文显示问题
plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'SimSun', 'Arial Unicode MS']  # 优先使用的中文字体
plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示问题

# 定义资源文件根目录
ROOT_RESOURCE = os.path.join(os.path.dirname(__file__), 'resource')

# 配置 DashScope
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY', '')  # 从环境变量获取 API Key
dashscope.timeout = 30  # 设置超时时间为 30 秒

# ====== 门票助手 system prompt 和函数描述 ======
system_prompt = """我是门票助手，以下是关于门票订单表相关的字段，我可能会编写对应的SQL，对数据进行查询
-- 门票订单表
CREATE TABLE tkt_orders (
    order_time DATETIME,             -- 订单日期
    account_id INT,                  -- 预定用户ID
    gov_id VARCHAR(18),              -- 商品使用人ID（身份证号）
    gender VARCHAR(10),              -- 使用人性别
    age INT,                         -- 年龄
    province VARCHAR(30),           -- 使用人省份
    SKU VARCHAR(100),                -- 商品SKU名
    product_serial_no VARCHAR(30),  -- 商品ID
    eco_main_order_id VARCHAR(20),  -- 订单ID
    sales_channel VARCHAR(20),      -- 销售渠道
    status VARCHAR(30),             -- 商品状态
    order_value DECIMAL(10,2),       -- 订单金额
    quantity INT                     -- 商品数量
);
一日门票，对应多种SKU：
Universal Studios Beijing One-Day Dated Ticket-Standard
Universal Studios Beijing One-Day Dated Ticket-Child
Universal Studios Beijing One-Day Dated Ticket-Senior
二日门票，对应多种SKU：
USB 1.5-Day Dated Ticket Standard
USB 1.5-Day Dated Ticket Discounted
一日门票、二日门票查询
SUM(CASE WHEN SKU LIKE 'Universal Studios Beijing One-Day%' THEN quantity ELSE 0 END) AS one_day_ticket_sales,
SUM(CASE WHEN SKU LIKE 'USB%' THEN quantity ELSE 0 END) AS two_day_ticket_sales
我将回答用户关于门票相关的问题

每当 exc_sql 工具返回 markdown 表格和图片时，你必须原样输出工具返回的全部内容（包括图片 markdown），不要只总结表格，也不要省略图片。这样用户才能直接看到表格和图片。
"""

functions_desc = [
    {
        "name": "exc_sql",
        "description": "对于生成的SQL，进行SQL查询",
        "parameters": {
            "type": "object",
            "properties": {
                "sql_input": {
                    "type": "string",
                    "description": "生成的SQL语句",
                }
            },
            "required": ["sql_input"],
        },
    },
]

# ====== 会话隔离 DataFrame 存储 ======
# 用于存储每个会话的 DataFrame，避免多用户数据串扰
_last_df_dict = {}

def get_session_id(kwargs):
    """根据 kwargs 获取当前会话的唯一 session_id，这里用 messages 的 id"""
    messages = kwargs.get('messages')
    if messages is not None:
        return id(messages)
    return None

# ====== exc_sql 工具类实现 ======
@register_tool('exc_sql')
class ExcSQLTool(BaseTool):
    """
    SQL查询工具，执行传入的SQL语句并返回结果，并自动进行可视化。
    """
    description = '对于生成的SQL，进行SQL查询，并自动可视化'
    parameters = [{
        'name': 'sql_input',
        'type': 'string',
        'description': '生成的SQL语句',
        'required': True
    }]

    def call(self, params: str, **kwargs) -> str:
        import json
        import matplotlib.pyplot as plt
        import io, os, time
        import numpy as np
        args = json.loads(params)
        sql_input = args['sql_input']
        database = args.get('database', 'world')
        engine = create_engine(
            f'mysql+mysqlconnector://root:Aa123456!@localhost:3306/{database}?charset=utf8mb4',
            connect_args={'connect_timeout': 10}, pool_size=10, max_overflow=20
        )
        try:
            df = pd.read_sql(sql_input, engine)
            md = df.head(10).to_markdown(index=False)
            # 自动创建目录
            save_dir = os.path.join(os.path.dirname(__file__), 'image_show')
            os.makedirs(save_dir, exist_ok=True)
            filename = f'bar_{int(time.time()*1000)}.png'
            save_path = os.path.join(save_dir, filename)
            # 生成图表
            generate_chart_png(df, save_path)
            img_path = os.path.join('image_show', filename)
            img_md = f'![柱状图]({img_path})'
            return f"{md}\n\n{img_md}"
        except Exception as e:
            return f"SQL执行或可视化出错: {str(e)}"

# ========== 通用可视化函数 ========== 
def generate_chart_png(df_sql, save_path):
    columns = df_sql.columns
    x = np.arange(len(df_sql))
    # 获取object类型
    object_columns = df_sql.select_dtypes(include='O').columns.tolist()
    if columns[0] in object_columns:
        object_columns.remove(columns[0])
    num_columns = df_sql.select_dtypes(exclude='O').columns.tolist()
    if len(object_columns) > 0:
        # 对数据进行透视，以便为每个日期和销售渠道创建堆积柱状图
        pivot_df = df_sql.pivot_table(index=columns[0], columns=object_columns, 
                                      values=num_columns, 
                                      fill_value=0)
        # 绘制堆积柱状图
        fig, ax = plt.subplots(figsize=(10, 6))
        # 为每个销售渠道和票类型创建柱状图
        bottoms = None
        for col in pivot_df.columns:
            ax.bar(pivot_df.index, pivot_df[col], bottom=bottoms, label=str(col))
            if bottoms is None:
                bottoms = pivot_df[col].copy()
            else:
                bottoms += pivot_df[col]
    else:
        print('进入到else...')
        bottom = np.zeros(len(df_sql))
        for column in columns[1:]:
            plt.bar(x, df_sql[column], bottom=bottom, label=column)
            bottom += df_sql[column]
        plt.xticks(x, df_sql[columns[0]])
    plt.legend()
    plt.title("销售统计")
    plt.xlabel(columns[0])
    plt.ylabel("门票数量")
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig(save_path)
    plt.close()

# ====== 初始化门票助手服务 ======
def init_agent_service():
    """初始化门票助手服务"""
    llm_cfg = {
        'model': 'qwen-turbo-2025-04-28',
        'timeout': 30,
        'retry_count': 3,
    }
    try:
        bot = Assistant(
            llm=llm_cfg,
            name='门票助手',
            description='门票查询与订单分析',
            system_message=system_prompt,
            function_list=['exc_sql'],  # 移除绘图工具
        )
        print("助手初始化成功！")
        return bot
    except Exception as e:
        print(f"助手初始化失败: {str(e)}")
        raise


def app_gui():
    """图形界面模式，提供 Web 图形界面"""
    try:
        print("正在启动 Web 界面...")
        # 初始化助手
        bot = init_agent_service()
        # 配置聊天界面，列举3个典型门票查询问题
        chatbot_config = {
            'prompt.suggestions': [
                '2023年4、5、6月一日门票，二日门票的销量多少？帮我按照周进行统计',
                '2023年7月的不同省份的入园人数统计',
                '帮我查看2023年10月1-7日销售渠道订单金额排名',
            ]
        }
        print("Web 界面准备就绪，正在启动服务...")
        # 启动 Web 界面
        WebUI(
            bot,
            chatbot_config=chatbot_config
        ).run()
    except Exception as e:
        print(f"启动 Web 界面失败: {str(e)}")
        print("请检查网络连接和 API Key 配置")


if __name__ == '__main__':
    # 运行模式选择
    app_gui()          # 图形界面模式（默认）

数据库结构参考：

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

第四期热点征文-大模型技术

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

登录后参与评论

0 条评论

热度