LLM与SQL结合：ReAct模式实现高效数据统计分析-AI智能范式网

LLM与SQL结合：ReAct模式实现高效数据统计分析

元宿six·2026-09-06·阅读 21 GEO检测（免费）

LLMSQLReAct模式

|难度：普通|阅读约 16 分钟

1. 项目背景与核心思路

在数据处理领域，文本数据的去重清洗只是第一步。当我们面对已经清洗好的结构化数据时，如何高效地进行统计分析是一个常见需求。传统做法是直接编写SQL查询或使用统计软件，但对于非技术人员来说存在门槛。而大语言模型（LLM）虽然能理解自然语言，但在直接执行统计任务时存在明显缺陷：

结果不一致性：相同问题多次查询可能得到不同答案
注意力漂移：处理长数据时会出现信息遗漏
数值幻觉：可能生成看似合理但实际错误的统计结果

ReAct（Reasoning and Acting）模式提供了一种创新解决方案：让LLM专注于它擅长的自然语言理解和逻辑推理，而将具体的统计计算交给专业工具执行。这种分工协作的模式既保留了自然语言交互的便利性，又确保了统计结果的精确性。

关键设计原则：让每个组件做自己最擅长的事。LLM作为"大脑"负责理解意图和规划步骤，SQL引擎作为"手"负责精确执行计算任务。

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. 技术实现详解

2.1 系统架构设计

整个系统的工作流程可以分为三个核心环节：

自然语言理解层：LLM解析用户问题，识别统计需求
工具调用层：LLM生成对应的SQL查询语句
结果整合层：LLM将SQL执行结果转化为自然语言回答

mermaid复制graph TD
    A[用户提问] --> B(LLM理解问题)
    B --> C{是否需要统计}
    C -->|是| D[生成SQL查询]
    C -->|否| E[直接回答]
    D --> F[执行SQL工具]
    F --> G[返回JSON结果]
    G --> H(LLM组织回答)
    H --> I[输出最终答案]

2.2 核心组件实现

2.2.1 数据库连接工具

python复制import sqlite3
import pandas as pd

def execute_sql(query: str) -> str:
    """执行安全的只读SQL查询，返回JSON格式结果"""
    try:
        conn = sqlite3.connect("example.db")
        cursor = conn.cursor()
        
        # 安全校验：确保是SELECT查询
        if not query.strip().upper().startswith("SELECT"):
            raise ValueError("只允许执行SELECT查询")
            
        df = pd.read_sql_query(query, conn)
        return df.to_json(orient="records")
    except Exception as e:
        return f"错误：{str(e)}"
    finally:
        conn.close()

这个工具函数有几个关键设计点：

使用上下文管理器确保连接关闭
添加查询类型校验防止数据修改
返回标准JSON格式便于后续处理
捕获所有异常并提供友好错误信息

2.2.2 工具描述定义

为了让LLM理解如何使用这个工具，需要提供清晰的元数据描述：

python复制tools = [
    {
        "type": "function",
        "function": {
            "name": "execute_sql",
            "description": "在SQLite数据库上执行安全的只读查询。表'records_dedup'包含字段：name(文本), company(文本), date(日期), ext_amount(实数)",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "符合SQLite语法的SELECT语句"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

描述中特别强调了：

数据库类型(SQLite)
主要表结构
允许的查询类型(只读)
参数要求和格式

2.3 ReAct Agent实现

完整的Agent实现包含多轮交互能力：

python复制def react_agent(user_question: str, max_steps=3):
    messages = [
        {"role": "system", "content": "你是一个数据分析助手。必须通过execute_sql工具获取准确数据，禁止自行计算数值。"},
        {"role": "user", "content": user_question}
    ]
    
    for step in range(max_steps):
        response = client.chat.completions.create(
            model=model_name,
            messages=messages,
            tools=tools,
            tool_choice="auto",
            temperature=0  # 保持确定性输出
        )
        
        assistant_msg = response.choices[0].message
        messages.append(assistant_msg)
        
        if not assistant_msg.tool_calls:
            return assistant_msg.content
            
        for tool_call in assistant_msg.tool_calls:
            if tool_call.function.name == "execute_sql":
                sql = json.loads(tool_call.function.arguments)["query"]
                result = execute_sql(sql)
                messages.append({
                    "role": "tool",
                    "content": result,
                    "tool_call_id": tool_call.id
                })
    
    return "查询过于复杂，请简化您的问题。"

关键控制逻辑：

限制最大交互轮次防止无限循环
保持temperature=0确保SQL生成的确定性
严格区分工具调用和最终回答阶段
维护完整的对话上下文

3. 典型应用场景

3.1 基础统计查询

用户问题："各公司的平均交易金额是多少？"

Agent处理过程：

生成SQL：SELECT company, AVG(ext_amount) as avg_amount FROM records_dedup GROUP BY company
执行查询获取原始数据
组织回答："ABC科技的平均交易金额为50,000元，XYZ有限公司为1,200元"

3.2 多步骤对比分析

用户问题："比较北京和上海地区的总交易额"

Agent处理流程：

生成第一个SQL：SELECT SUM(ext_amount) FROM records_dedup WHERE ext_place='北京'
生成第二个SQL：SELECT SUM(ext_amount) FROM records_dedup WHERE ext_place='上海'
计算结果差异
输出："北京地区总交易额50,000元，上海地区1,200元，北京高出48,800元"

3.3 复杂条件查询

用户问题："列出2023年第二季度交易金额超过1000元的记录"

生成SQL示例：

sql复制SELECT name, company, ext_amount 
FROM records_dedup 
WHERE date BETWEEN '2023-04-01' AND '2023-06-30' 
AND CAST(ext_amount AS REAL) > 1000

4. 性能优化与实践经验

4.1 SQL生成优化策略

Schema提示增强：

python复制system_prompt = """
数据库包含表records_dedup，结构如下：
- name: 姓名(TEXT)
- company: 公司(TEXT) 
- date: 日期(TEXT, 格式YYYY-MM-DD)
- ext_amount: 金额(TEXT, 如'50000 元')
查询时需注意：
1. 金额比较需使用CAST(ext_amount AS REAL)
2. 日期比较需确保格式一致
"""

查询模板预置：

python复制QUERY_TEMPLATES = {
    "sum_by_company": "SELECT company, SUM(CAST(ext_amount AS REAL)) as total FROM records_dedup GROUP BY company",
    "filter_by_date": "SELECT * FROM records_dedup WHERE date BETWEEN '{start}' AND '{end}'"
}

4.2 常见问题排查指南

问题现象	可能原因	解决方案
SQL语法错误	LLM生成的SQL不符合方言	在system prompt中明确数据库类型
空结果返回	字段名大小写不匹配	提供完整的字段名列表和示例
数值比较错误	ext_amount包含单位	使用CAST(ext_amount AS REAL)转换
性能低下	未加索引的大表查询	提示LLM添加WHERE条件限制范围

4.3 安全性最佳实践

输入校验：

python复制def validate_sql(query: str) -> bool:
    forbidden = ["INSERT", "UPDATE", "DELETE", "DROP", ";"]
    return all(cmd not in query.upper() for cmd in forbidden)

权限控制：

使用只读数据库用户
设置行级权限限制
启用查询日志审计

资源限制：

python复制conn.set_limits(
    sqlite3.SQLITE_LIMIT_ATTACHED, 0,
    sqlite3.SQLITE_LIMIT_COMPOUND_SELECT, 5
)

5. 扩展应用方向

5.1 多数据源整合

通过扩展工具集支持不同数据源：

python复制tools.append({
    "name": "query_elasticsearch",
    "description": "在Elasticsearch中执行查询",
    # 参数定义...
})

5.2 可视化增强

让LLM生成可视化指令：

python复制def generate_plotly(code: str):
    """执行Plotly代码生成图表"""
    # 安全沙箱执行...

5.3 自动化报告

结合模板引擎自动生成分析报告：

python复制from jinja2 import Template

report_template = Template("""
# {{title}}

## 主要发现
{% for item in findings %}
- {{item}}
{% endfor %}

## 详细数据
{{table|safe}}
""")

在实际项目中，这种ReAct架构已经帮助我们将复杂统计问题的处理时间从小时级缩短到分钟级，同时保证了结果的准确性。一个特别有用的技巧是在system prompt中提供3-5个典型查询示例，这可以显著提高LLM生成正确SQL的概率。