Chainlit集成LlamaIndex并使用通义千问实现和数据库交互的网页对话应用（text2sql）

前言

我在之前的几篇文章中写了如何使用Chainlit集成Langchain并使用通义千问实现和数据库交互的网页对话应用，但是发现Langchain的几种和数据库交互的组件都不够让我满意，虽然已经满足了大部分场景的需求，但是问题还是很多，比如问题和数据库好不相关的时候，程序生成错误的sql,导致报错，sql智能体交互响应太慢等等，最近使用了LlamaIndex中的sql交互组件NLSQLTableQueryEngine，发现YYDS。就是我一直寻找的东西。既满足了响应速度又足够智能不会报错，下面教大家如何使用Chainlit集成LlamaIndex实现网页和数据库交互的应用。

LlamaIndex官方地址 https://docs.llamaindex.ai/en/stable/

快速上手

创建一个文件，例如“chainlit_chat”

mkdir chainlit_chat

进入 chainlit_chat文件夹下，执行命令创建python 虚拟环境空间(需要提前安装好python sdk。 Chainlit 需要python>=3.8。,具体操作，由于文章长度问题就不在叙述，自行百度)，命令如下：

python -m venv .venv

这一步是避免python第三方库冲突，省事版可以跳过
.venv是创建的虚拟空间文件夹可以自定义

接下来激活你创建虚拟空间，命令如下：

#linux or mac
source .venv/bin/activate
#windows
.venv\Scripts\activate

在项目根目录下创建`requirements.txt`，内容如下：

chainlit
llama-index-core
llama-index-llms-dashscope
llama-index-embeddings-dashscope
llama-index-retrievers-bm25~=0.3.0

执行以下命令安装依赖：

pip install -r .\requirements.txt

安装后，项目根目录下会多出.chainlit 和.files文件夹和chainlit.md文件

代码创建

只使用通义千问的DashScope模型服务灵积的接口

在项目根目录下创建`.env`环境变量，配置如下：

DASHSCOPE_API_KEY="sk-api_key"

DASHSCOPE_API_KEY 是阿里dashscope的服务的APIkey，代码中使用DashScope的sdk实现，所以不需要配置base_url。默认就是阿里的base_url。
阿里模型接口地址 https://dashscope.console.aliyun.com/model

在项目根目录下创建app.py文件，代码如下：

此代码使用摘要索引和向量索引，利用RetrieverQueryEngine 路由检索器，根据问题分类提示，选择摘要索引和向量索引进行索引。

import os
import time

import chainlit as cl
from llama_index.core import SQLDatabase, Settings, PromptTemplate
from llama_index.core.indices.struct_store import NLSQLTableQueryEngine
from llama_index.core.prompts import PromptType
from llama_index.embeddings.dashscope import DashScopeEmbedding, DashScopeTextEmbeddingModels, \
    DashScopeTextEmbeddingType
from llama_index.llms.dashscope import DashScopeGenerationModels, DashScope
from sqlalchemy import create_engine

Settings.llm = DashScope(
    model_name=DashScopeGenerationModels.QWEN_TURBO, api_key=os.environ["DASHSCOPE_API_KEY"], max_tokens=512
)
Settings.embed_model = DashScopeEmbedding(
    model_name=DashScopeTextEmbeddingModels.TEXT_EMBEDDING_V2,
    text_type=DashScopeTextEmbeddingType.TEXT_TYPE_DOCUMENT,
)

engine = create_engine("postgresql+psycopg2://username:passward@ip:5432/dbname")

tables = ["blade_user", "blade_role", "blade_menu", "exam_course", "exam_data"]
# 准备数据
sql_database = SQLDatabase(engine)

# 创建大模型
llm = DashScope(
    model_name=DashScopeGenerationModels.QWEN_MAX, api_key=os.environ["DASHSCOPE_API_KEY"]
)

CUSTOM_TEXT_TO_SQL_TMPL = (
    "You are a {dialect} expert. Given an input question, first create a syntactically correct {dialect} "
    "query to run, then look at the results of the query and return the answer. "
    "You can order the results by a relevant column to return the most "
    "interesting examples in the database.\n\n"
    "Never query for all the columns from a specific table, only ask for a "
    "few relevant columns given the question.\n\n"
    "Pay attention to use only the column names that you can see in the schema "
    "description. "
    "Be careful to not query for columns that do not exist. "
    "Pay attention to which column is in which table. "
    "Also, qualify column names with the table name when needed. "
    "If there is a table whose column name is marked with is_deleted, the default value is is_deleted=0"
    "You are required to use the following format, each taking one line:\n\n"
    "Question: Question here\n"
    "SQLQuery: SQL Query to run\n"
    "SQLResult: Result of the SQLQuery\n"
    "Answer: Final answer here\n\n"
    "Only use tables listed below.\n"
    "{schema}\n\n"
    "Question: {query_str}\n"
    "SQLQuery: "
)

CUSTOM_TEXT_TO_SQL_PROMPT = PromptTemplate(
    CUSTOM_TEXT_TO_SQL_TMPL,
    prompt_type=PromptType.TEXT_TO_SQL,
)

# 构建查询引擎
query_engine = NLSQLTableQueryEngine(
    sql_database=sql_database,
    text_to_sql_prompt=CUSTOM_TEXT_TO_SQL_PROMPT,
    tables=tables,
    llm=llm,
    streaming=True,
    verbose=True
)


@cl.on_message
async def on_message(message: cl.Message):
    start_time = time.time()
    msg = cl.Message(content="", author="Assistant")
    res = await query_engine.aquery(message.content)
    async for token in res.response_gen:
        await msg.stream_token(token)
    print(f"代码执行时间: {time.time() - start_time} 秒")
    await msg.send()

默认使用open ai的sdk，这里我换成国内阿里云的DashScope。
官方教程中没有设置自定义，文本转sql提示词示例，这里我设置了自定义提示词，因为数据库表中有is_deleted是否删除字段，默认提示词生成sql，是查询所有数据没有筛选被删除的数据，每次提问必须加上类似正常数据的提示词，才能给我想要的。我的复制默认的文本转sql的提示词，增加了一句If there is a table whose column name is marked with is_deleted, the default value is is_deleted=0，默认返回的就是筛选后正常的数据。
create_engine底层数据库连接是SQLAlchemy ，SQLAlchemy 支持的数据库，都可以使用。SQLAlchemy 官网地址
- SQLAlchemy 是一个非常灵活的 Python 库，用于数据库操作，并且它支持多种数据库系统。根据多个资料来源，SQLAlchemy 支持的主要数据库系统包括但不限于：PostgreSQL、MySQL、SQLite、Oracle、SQL Server、DB2、Firebird、Sybase 等。

代码解读

这段代码是一个使用了chainlit框架的Python脚本，它结合了自然语言处理（NLP）与SQL查询生成技术，旨在从数据库中以自然语言形式提出问题，并返回相应的查询结果。以下是该段代码的功能分析：

环境配置：
- 导入必要的库和模块，如os用于访问环境变量，time用于计时，chainlit用于构建交互式应用，llama_index用于处理自然语言到SQL的转换，以及sqlalchemy用于数据库连接。
设置LLM和Embedding Model：
- 使用DashScope作为语言模型（LLM）和嵌入模型（Embedding Model），具体使用的是Qwen Turbo模型进行文本生成，并使用TEXT_EMBEDDING_V2模型进行文本嵌入。这些模型是通过API密钥从DashScope服务获取的。
数据库连接：
- 使用sqlalchemy创建了一个数据库引擎来连接到一个PostgreSQL数据库实例。数据库信息包括用户名、密码、IP地址、端口和数据库名。
定义SQL查询模板：
- 定义了一个自定义的文本到SQL查询的提示模板（Prompt Template）。这个模板指导LLM如何将自然语言问题转换为SQL查询，并如何解释查询结果以提供最终答案。模板包含了一些规则，比如如何选择相关列，避免查询不存在的列等。
构建查询引擎：
- 基于上述配置创建了一个NLSQLTableQueryEngine实例，它负责接收自然语言问题，将其转换成SQL查询语句，并执行查询后返回结果。
消息处理函数：
- 使用chainlit框架中的on_message装饰器定义了一个异步函数on_message，这个函数会在接收到用户消息时触发。它调用查询引擎处理消息，并通过流式传输的方式将结果发送给用户。