如何使用LLM智能进行数据库操作，例如：ChatGPT-4

大模型开发/技术交流

LLM
Prompt
文心大模型

1月5日272看过

AIGC出现后持续改变着我们的方方面面，今天就改变传统软件开发思维模式的一些应用实践给大家分享一些经验。

首先说说我们以前操作数据库的流程：

1. 设计数据库及相关表
2. 按照业务逻辑编写相关的SQL
3. 通过程序驱动这些SQL获取数据集
4. 对这些数据集进行处理
5. 按照需求展现在前端

LLM出现后就大大简化了上述的过程：

1. 设计数据库，当然这个部分也可以交给LLM，但是效率不高，不建议采用
2. 把数据库的结构描述喂给LLM，让他理解数据库的结构和字段含义，这里要对每个字段进行必要的描述（数据库设计中的备注）
3. 对数据的操作用自然语言描述并向LLM提问，让LLM按照数据需求生成相应的SQL
4. 执行SQL得到数据集，同时让LLM对这些输出的数据进行需要的格式化输出

大家可以看到几个原先都是依靠编程实现的过程完全可以被LLM替代，下面就给出相关的代码来实现上述过程。我这里举一个对物联网数据处理的简单例子，这里使用了LangChain和ChatGPT-4来实现：

这里涉及2张表：

1. 设备信息表iot_devices
2. 上行数据表iot_upload_datas

第一步：两张表设计好后将表结构的描述导出并喂给LLM

  
  
  
  
  
  
   
   
   
   
   
   
    @tool
   
   
   
   
   
   
   
   
   
   
   
   
    def iot_device_info(key_str: str):
   
   
   
   
   
   
   
   
   
   
   
   
        """物联网SQL生成器"""
   
   
   
   
   
   
   
   
   
   
   
   
        logger.info("参数：%s", key_str)
   
   
   
   
   
   
   
   
   
   
   
   
        # 模型定义
   
   
   
   
   
   
   
   
   
   
   
   
        llm = ChatOpenAI(model="gpt-4-1106-preview")
   
   
   
   
   
   
   
   
   
   
   
   
        # 系统提示词模板
   
   
   
   
   
   
   
   
   
   
   
   
        template = """
   
   
   
   
   
   
   
   
   
   
   
   
            请作为sql工程师按照MySQL的如下表结构描述，按照用户的需求描述生成SQL语句，结果必须以json格式输出，SQL语句赋给query_sql：
   
   
   
   
   
   
   
   
   
   
   
   
            1.表iot_devices的字段描述为
   
   
   
   
   
   
   
   
   
   
   
   
                `id` int NOT NULL AUTO_INCREMENT COMMENT '序号',
   
   
   
   
   
   
   
   
   
   
   
   
                `device_id` varchar(255) NOT NULL COMMENT '设备编号',
   
   
   
   
   
   
   
   
   
   
   
   
                `device_name` varchar(255) NOT NULL COMMENT '设备名称',
   
   
   
   
   
   
   
   
   
   
   
   
                `position` json DEFAULT NULL COMMENT '所在位置经纬度',
   
   
   
   
   
   
   
   
   
   
   
   
                `status` int DEFAULT NULL COMMENT '设备状态',
   
   
   
   
   
   
   
   
   
   
   
   
            2.表iot_upload_datas的字段描述为
   
   
   
   
   
   
   
   
   
   
   
   
                `id` int NOT NULL AUTO_INCREMENT COMMENT '序号',
   
   
   
   
   
   
   
   
   
   
   
   
                `device_id` varchar(255) NOT NULL COMMENT '设备编号',
   
   
   
   
   
   
   
   
   
   
   
   
                `upload_data` json NOT NULL COMMENT '上行的数据',
   
   
   
   
   
   
   
   
   
   
   
   
                `created_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '上行数据时间',
   
   
   
   
   
   
   
   
   
   
   
   
            """
   
   
   
   
   
   
   
   
   
   
   
   
        system_message_prompt = SystemMessagePromptTemplate.from_template(template)
   
   
   
   
   
   
   
   
   
   
   
   
        # 人为输入的提示词模板
   
   
   
   
   
   
   
   
   
   
   
   
        human_template = "{message}"
   
   
   
   
   
   
   
   
   
   
   
   
        human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)
   
   
   
   
   
   
   
   
   
   
   
   
        # 整合系统提示词模板，自然人提示词模板
   
   
   
   
   
   
   
   
   
   
   
   
        chat_prompt = ChatPromptTemplate.from_messages(
   
   
   
   
   
   
   
   
   
   
   
   
            [system_message_prompt, human_message_prompt]
   
   
   
   
   
   
   
   
   
   
   
   
        )
   
   
   
   
   
   
   
   
   
   
   
   
        chain = LLMChain(llm=llm, prompt=chat_prompt, verbose=True)
   
   
   
   
   
   
   
   
   
   
   
   
        # AI生成SQL
   
   
   
   
   
   
   
   
   
   
   
   
        result = chain.run(message=key_str)
   
   
   
   
   
   
   
   
   
   
   
   
        return result

这里的提示词很重要，一定明确告知LLM你的目的是让AI理解数据库的表结构，并按照需求描述生成SQL

第二步：将生成的SQL引入到数据库执行并生成数据集

  
  
  
  
  
  
   
   
   
   
   
   
    @tool
   
   
   
   
   
   
   
   
   
   
   
   
    def iot_query(query_sql: str):
   
   
   
   
   
   
   
   
   
   
   
   
        """物联网SQL查询"""
   
   
   
   
   
   
   
   
   
   
   
   
        logger.info("参数：%s", query_sql)
   
   
   
   
   
   
   
   
   
   
   
   
    
   
   
   
   
   
   
   
   
   
   
   
   
        return_data=None
   
   
   
   
   
   
   
   
   
   
   
   
        connection = pool.get_connection()
   
   
   
   
   
   
   
   
   
   
   
   
        try:
   
   
   
   
   
   
   
   
   
   
   
   
            with connection.cursor() as cursor:
   
   
   
   
   
   
   
   
   
   
   
   
                # 执行查询操作
   
   
   
   
   
   
   
   
   
   
   
   
                cursor.execute(query_sql)
   
   
   
   
   
   
   
   
   
   
   
   
                result = cursor.fetchall()
   
   
   
   
   
   
   
   
   
   
   
   
                return_data = result
   
   
   
   
   
   
   
   
   
   
   
   
        except Exception as e:
   
   
   
   
   
   
   
   
   
   
   
   
            logger.error("物联网查询异常：%s", e)
   
   
   
   
   
   
   
   
   
   
   
   
            return_data = str(e)
   
   
   
   
   
   
   
   
   
   
   
   
        finally:
   
   
   
   
   
   
   
   
   
   
   
   
            # 关闭数据库连接
   
   
   
   
   
   
   
   
   
   
   
   
            connection.close()
   
   
   
   
   
   
   
   
   
   
   
   
        return return_data

这个部分其实是非常通用的，直接将完整的SQL注入就可以，当然这里要考虑数据库的安全性，本例就不详细说明

第三步：使用LangChain驱动这些Tool来实现完整事务链

  
  
  
  
  
  
   
   
   
   
   
   
    # 复杂任务调度
   
   
   
   
   
   
   
   
   
   
   
   
    def task_controlor(question: str):
   
   
   
   
   
   
   
   
   
   
   
   
        logger.info("提交到LangChain的问题：%s", question)
   
   
   
   
   
   
   
   
   
   
   
   
        # 定义工具集
   
   
   
   
   
   
   
   
   
   
   
   
        tools = [
   
   
   
   
   
   
   
   
   
   
   
   
            Tool(name="iot_device_info", func=iot_device_info, description="物联网SQL生成器"),
   
   
   
   
   
   
   
   
   
   
   
   
            Tool(name="iot_query", func=iot_query, description="物联网SQL查询"),
   
   
   
   
   
   
   
   
   
   
   
   
        ]
   
   
   
   
   
   
   
   
   
   
   
   
    
   
   
   
   
   
   
   
   
   
   
   
   
        system_message = SystemMessage(
   
   
   
   
   
   
   
   
   
   
   
   
            content="你是一个AI助手，完成下面的工作，如果失败就再尝试一次，如果没有顺利完成所有任务就不要输出结果："
   
   
   
   
   
   
   
   
   
   
   
   
        )
   
   
   
   
   
   
   
   
   
   
   
   
        prompt = OpenAIFunctionsAgent.create_prompt(
   
   
   
   
   
   
   
   
   
   
   
   
            system_message=system_message,
   
   
   
   
   
   
   
   
   
   
   
   
        )
   
   
   
   
   
   
   
   
   
   
   
   
        # 用openAI大模型进行任务分解
   
   
   
   
   
   
   
   
   
   
   
   
        agent = OpenAIFunctionsAgent(
   
   
   
   
   
   
   
   
   
   
   
   
            llm=ChatOpenAI(model="gpt-4-1106-preview"), tools=tools, prompt=prompt
   
   
   
   
   
   
   
   
   
   
   
   
        )
   
   
   
   
   
   
   
   
   
   
   
   
        agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
   
   
   
   
   
   
   
   
   
   
   
   
        answer = ""
   
   
   
   
   
   
   
   
   
   
   
   
        r = agent_executor.run(f"先用物联网SQL生成器按照[{question}]要求生成SQL语句，再让物联网SQL查询执行，把查询结果用表格的方式输出")
   
   
   
   
   
   
   
   
   
   
   
   
        answer += f"{r}\n"
   
   
   
   
   
   
   
   
   
   
   
   
        return answer