feat: auto select datasource

This commit is contained in:
ulleo
2025-06-26 19:06:00 +08:00
parent a0b9c081e1
commit 60e9b7a3b4
12 changed files with 316 additions and 50 deletions
@@ -0,0 +1,71 @@
"""016_modify_chat
Revision ID: 031148da1d81
Revises: 02d84523a979
Create Date: 2025-06-26 17:00:07.054531
"""
from alembic import op
import sqlalchemy as sa
import sqlmodel.sql.sqltypes
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = '031148da1d81'
down_revision = '02d84523a979'
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column('chat', 'datasource',
existing_type=sa.INTEGER(),
nullable=True)
op.add_column('chat_record', sa.Column('ai_modal_id', sa.Integer(), nullable=True))
op.add_column('chat_record', sa.Column('first_chat', sa.Boolean(), nullable=True))
op.add_column('chat_record', sa.Column('recommended_question_answer', sa.Text(), nullable=True))
op.add_column('chat_record', sa.Column('recommended_question', sa.Text(), nullable=True))
op.add_column('chat_record', sa.Column('datasource_select_answer', sa.Text(), nullable=True))
op.add_column('chat_record', sa.Column('token_sql', sa.Integer(), nullable=True))
op.add_column('chat_record', sa.Column('token_chart', sa.Integer(), nullable=True))
op.add_column('chat_record', sa.Column('token_analysis', sa.Integer(), nullable=True))
op.add_column('chat_record', sa.Column('token_predict', sa.Integer(), nullable=True))
op.add_column('chat_record', sa.Column('full_recommended_question_message', sa.Text(), nullable=True))
op.add_column('chat_record', sa.Column('token_recommended_question', sa.Integer(), nullable=True))
op.add_column('chat_record', sa.Column('full_select_datasource_message', sa.Text(), nullable=True))
op.add_column('chat_record', sa.Column('token_select_datasource_question', sa.Integer(), nullable=True))
op.alter_column('chat_record', 'chat_id',
existing_type=sa.INTEGER(),
nullable=False)
op.alter_column('chat_record', 'datasource',
existing_type=sa.INTEGER(),
nullable=True)
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column('chat_record', 'datasource',
existing_type=sa.INTEGER(),
nullable=False)
op.alter_column('chat_record', 'chat_id',
existing_type=sa.INTEGER(),
nullable=True)
op.drop_column('chat_record', 'token_select_datasource_question')
op.drop_column('chat_record', 'full_select_datasource_message')
op.drop_column('chat_record', 'token_recommended_question')
op.drop_column('chat_record', 'full_recommended_question_message')
op.drop_column('chat_record', 'token_predict')
op.drop_column('chat_record', 'token_analysis')
op.drop_column('chat_record', 'token_chart')
op.drop_column('chat_record', 'token_sql')
op.drop_column('chat_record', 'datasource_select_answer')
op.drop_column('chat_record', 'recommended_question')
op.drop_column('chat_record', 'recommended_question_answer')
op.drop_column('chat_record', 'first_chat')
op.drop_column('chat_record', 'ai_modal_id')
op.alter_column('chat', 'datasource',
existing_type=sa.INTEGER(),
nullable=False)
# ### end Alembic commands ###
+29 -14
View File
@@ -72,7 +72,7 @@ async def start_chat(session: SessionDep, current_user: CurrentUser, create_chat
async def mcp_question(session: SessionDep, token: str, request_question: ChatQuestion):
user = await get_current_user(session, token)
# return await stream_sql(session, user, request_question)
return {"content":"""步骤1: 确定需要查询的字段。
return {"content": """步骤1: 确定需要查询的字段。
我们需要统计上海的订单总数,因此需要从"城市"字段中筛选出值为"上海"的记录,并使用COUNT函数计算这些记录的数量。
步骤2: 确定筛选条件。
@@ -106,16 +106,17 @@ async def stream_sql(session: SessionDep, current_user: CurrentUser, request_que
status_code=400,
detail=f"Chat with id {request_question.chat_id} not found"
)
ds: CoreDatasource | None = None
if chat.datasource:
# Get available datasource
ds = session.query(CoreDatasource).filter(CoreDatasource.id == chat.datasource).first()
if not ds:
raise HTTPException(
status_code=500,
detail="No available datasource configuration found"
)
# Get available datasource
ds = session.query(CoreDatasource).filter(CoreDatasource.id == chat.datasource).first()
if not ds:
raise HTTPException(
status_code=500,
detail="No available datasource configuration found"
)
request_question.engine = ds.type_name if ds.type != 'excel' else 'PostgreSQL'
request_question.engine = ds.type_name if ds.type != 'excel' else 'PostgreSQL'
# Get available AI model
aimodel = session.exec(select(AiModelDetail).where(
@@ -128,14 +129,18 @@ async def stream_sql(session: SessionDep, current_user: CurrentUser, request_que
detail="No available AI model configuration found"
)
history_records: List[ChatRecord] = list_records(session=session, current_user=current_user,
chart_id=request_question.chat_id)
history_records: List[ChatRecord] = list(filter(lambda r: True if r.first_chat != True else False,
list_records(session=session, current_user=current_user,
chart_id=request_question.chat_id)))
# get schema
request_question.db_schema = get_table_schema(session=session, ds=ds)
if ds:
request_question.db_schema = get_table_schema(session=session, ds=ds)
db_user = get_user_info(session=session, user_id=current_user.id)
request_question.lang = db_user.language
llm_service = LLMService(request_question, aimodel, history_records, CoreDatasource(**ds.model_dump()))
llm_service = LLMService(request_question, aimodel, history_records,
CoreDatasource(**ds.model_dump()) if ds else None)
llm_service.init_record(session=session, current_user=current_user)
@@ -144,6 +149,16 @@ async def stream_sql(session: SessionDep, current_user: CurrentUser, request_que
# return id
yield orjson.dumps({'type': 'id', 'id': llm_service.get_record().id}).decode() + '\n\n'
# select datasource if datasource is none
if not ds:
ds_res = llm_service.select_datasource(session=session)
for chunk in ds_res:
yield orjson.dumps({'content': chunk, 'type': 'datasource-result'}).decode() + '\n\n'
yield orjson.dumps({'id': llm_service.ds.id, 'datasource_name': llm_service.ds.name,
'engine_type': llm_service.ds.type_name, 'type': 'datasource'}).decode() + '\n\n'
llm_service.chat_question.db_schema = get_table_schema(session=session, ds=llm_service.ds)
# generate sql
sql_res = llm_service.generate_sql(session=session)
full_sql_text = ''
+32 -11
View File
@@ -60,6 +60,8 @@ def get_chat_with_records(session: SessionDep, chart_id: int, current_user: Curr
load_only(ChatRecord.id, ChatRecord.chat_id, ChatRecord.create_time, ChatRecord.finish_time,
ChatRecord.question, ChatRecord.sql_answer, ChatRecord.sql, ChatRecord.data,
ChatRecord.chart_answer, ChatRecord.chart, ChatRecord.analysis, ChatRecord.predict,
ChatRecord.datasource_select_answer, ChatRecord.recommended_question_answer,
ChatRecord.recommended_question,
ChatRecord.predict_data, ChatRecord.finish, ChatRecord.error, ChatRecord.run_time)).filter(
and_(Chat.create_by == current_user.id, ChatRecord.chat_id == chart_id)).order_by(ChatRecord.create_time).all()
@@ -74,7 +76,8 @@ def list_records(session: SessionDep, chart_id: int, current_user: CurrentUser)
return record_list
def create_chat(session: SessionDep, current_user: CurrentUser, create_chat_obj: CreateChat, require_datasource: bool = True) -> ChatInfo:
def create_chat(session: SessionDep, current_user: CurrentUser, create_chat_obj: CreateChat,
require_datasource: bool = True) -> ChatInfo:
if not create_chat_obj.datasource and require_datasource:
raise Exception("Datasource cannot be None")
@@ -84,7 +87,7 @@ def create_chat(session: SessionDep, current_user: CurrentUser, create_chat_obj:
chat = Chat(create_time=datetime.datetime.now(),
create_by=current_user.id,
brief=create_chat_obj.question.strip()[:20])
ds: CoreDatasource = None
ds: CoreDatasource | None = None
if create_chat_obj.datasource:
chat.datasource = create_chat_obj.datasource
ds = session.query(CoreDatasource).filter(CoreDatasource.id == create_chat_obj.datasource).first()
@@ -102,15 +105,9 @@ def create_chat(session: SessionDep, current_user: CurrentUser, create_chat_obj:
chat_info.id = chat.id
session.commit()
if not create_chat_obj.datasource:
# use AI to get ds
if not ds:
raise Exception(f"Datasource with id {create_chat_obj.datasource} not found")
chat_info.datasource_exists = True
chat_info.datasource_name = ds.name
if ds:
chat_info.datasource_exists = True
chat_info.datasource_name = ds.name
return chat_info
@@ -205,6 +202,30 @@ def save_full_predict_message_and_answer(session: SessionDep, record_id: int, an
return result
def save_full_select_datasource_message_and_answer(session: SessionDep, record_id: int, answer: str,
full_message: str, datasource: int = None,
engine_type: str = None) -> ChatRecord:
if not record_id:
raise Exception("Record id cannot be None")
record = session.query(ChatRecord).filter(ChatRecord.id == record_id).first()
record.full_select_datasource_message = full_message
record.datasource_select_answer = answer
if datasource:
record.datasource = datasource
record.engine_type = engine_type
result = ChatRecord(**record.model_dump())
session.add(record)
session.flush()
session.refresh(record)
session.commit()
return result
def save_sql(session: SessionDep, record_id: int, sql: str) -> ChatRecord:
if not record_id:
raise Exception("Record id cannot be None")
+31 -3
View File
@@ -7,8 +7,10 @@ from sqlmodel import SQLModel, Field
from apps.template.generate_analysis.generator import get_analysis_template
from apps.template.generate_chart.generator import get_chart_template
from apps.template.generate_guess_question.generator import get_guess_question_template
from apps.template.generate_predict.generator import get_predict_template
from apps.template.generate_sql.generator import get_sql_template
from apps.template.select_datasource.generator import get_datasource_template
class Chat(SQLModel, table=True):
@@ -25,11 +27,13 @@ class Chat(SQLModel, table=True):
class ChatRecord(SQLModel, table=True):
__tablename__ = "chat_record"
id: Optional[int] = Field(sa_column=Column(Integer, Identity(always=True), primary_key=True))
chat_id: int = Field(sa_column=Column(Integer))
chat_id: int = Field(sa_column=Column(Integer, nullable=False))
ai_modal_id: Optional[int] = Field(sa_column=Column(Integer))
first_chat: bool = Field(sa_column=Column(Boolean, nullable=True, default=False))
create_time: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=True))
finish_time: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=True))
create_by: int = Field(sa_column=Column(BigInteger, nullable=True))
datasource: int = Field(sa_column=Column(Integer, nullable=False))
datasource: int = Field(sa_column=Column(Integer, nullable=True))
engine_type: str = Field(max_length=64)
question: str = Field(sa_column=Column(Text, nullable=True))
sql_answer: str = Field(sa_column=Column(Text, nullable=True))
@@ -41,10 +45,21 @@ class ChatRecord(SQLModel, table=True):
analysis: str = Field(sa_column=Column(Text, nullable=True))
predict: str = Field(sa_column=Column(Text, nullable=True))
predict_data: str = Field(sa_column=Column(Text, nullable=True))
recommended_question_answer: str = Field(sa_column=Column(Text, nullable=True))
recommended_question: str = Field(sa_column=Column(Text, nullable=True))
datasource_select_answer: str = Field(sa_column=Column(Text, nullable=True))
full_sql_message: str = Field(sa_column=Column(Text, nullable=True))
token_sql: int = Field(default=0, nullable=True)
full_chart_message: str = Field(sa_column=Column(Text, nullable=True))
token_chart: int = Field(default=0, nullable=True)
full_analysis_message: str = Field(sa_column=Column(Text, nullable=True))
token_analysis: int = Field(default=0, nullable=True)
full_predict_message: str = Field(sa_column=Column(Text, nullable=True))
token_predict: int = Field(default=0, nullable=True)
full_recommended_question_message: str = Field(sa_column=Column(Text, nullable=True))
token_recommended_question: int = Field(default=0, nullable=True)
full_select_datasource_message: str = Field(sa_column=Column(Text, nullable=True))
token_select_datasource_question: int = Field(default=0, nullable=True)
finish: bool = Field(sa_column=Column(Boolean, nullable=True, default=False))
error: str = Field(sa_column=Column(Text, nullable=True))
run_time: float = Field(default=0)
@@ -67,7 +82,7 @@ class ChatInfo(BaseModel):
create_by: int = None
brief: str = ''
chat_type: str = "chat"
datasource: int = None
datasource: Optional[int] = None
engine_type: str = ''
datasource_name: str = ''
datasource_exists: bool = True
@@ -108,6 +123,19 @@ class AiModelQuestion(BaseModel):
def predict_user_question(self):
return get_predict_template()['user'].format(fields=self.fields, data=self.data, lang=self.lang)
def datasource_sys_question(self):
return get_datasource_template()['system']
def datasource_user_question(self, datasource_list: str = "[]"):
return get_datasource_template()['user'].format(question=self.question, data=datasource_list, lang=self.lang)
def datasource_guess_sys_question(self):
return get_guess_question_template()['system']
def datasource_guess_user_question(self, old_questions: str = "[]"):
return get_guess_question_template()['user'].format(question=self.question, schema=self.db_schema,
old_questions=old_questions, lang=self.lang)
class ChatQuestion(AiModelQuestion):
question: str
+99 -4
View File
@@ -6,12 +6,15 @@ import orjson
from langchain_community.utilities import SQLDatabase
from langchain_core.language_models import BaseLLM
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage, AIMessageChunk
from sqlalchemy import select
from sqlalchemy.orm import load_only
from apps.ai_model.model_factory import LLMConfig, LLMFactory, get_llm_config
from apps.chat.curd.chat import save_question, save_full_sql_message, save_full_sql_message_and_answer, save_sql, \
save_error_message, save_sql_exec_data, save_full_chart_message, save_full_chart_message_and_answer, save_chart, \
finish_record, save_full_analysis_message_and_answer, save_full_predict_message_and_answer, save_predict_data
from apps.chat.models.chat_model import ChatQuestion, ChatRecord
finish_record, save_full_analysis_message_and_answer, save_full_predict_message_and_answer, save_predict_data, \
save_full_select_datasource_message_and_answer
from apps.chat.models.chat_model import ChatQuestion, ChatRecord, Chat
from apps.datasource.models.datasource import CoreDatasource
from apps.db.db import exec_sql
from apps.system.models.system_model import AiModelDetail
@@ -30,6 +33,7 @@ class LLMService:
llm: BaseLLM
sql_message: List[Union[BaseMessage, dict[str, Any]]] = []
chart_message: List[Union[BaseMessage, dict[str, Any]]] = []
history_records: List[ChatRecord] = []
def __init__(self, chat_question: ChatQuestion, aimodel: AiModelDetail, history_records: List[ChatRecord] = [],
ds: CoreDatasource = None):
@@ -41,17 +45,22 @@ class LLMService:
llm_instance = LLMFactory.create_llm(self.config)
self.llm = llm_instance.llm
self.history_records = history_records
self.init_messages()
def init_messages(self):
# self.agent_executor = create_react_agent(self.llm)
last_sql_messages = list(
filter(lambda r: True if r.full_sql_message is not None and r.full_sql_message.strip() != '' else False,
history_records))
self.history_records))
last_sql_message_str = "[]" if last_sql_messages is None or len(last_sql_messages) == 0 else last_sql_messages[
-1].full_sql_message
last_chart_messages = list(
filter(
lambda r: True if r.full_chart_message is not None and r.full_chart_message.strip() != '' else False,
history_records))
self.history_records))
last_chart_message_str = "[]" if last_chart_messages is None or len(last_chart_messages) == 0 else \
last_chart_messages[-1].full_chart_message
@@ -216,6 +225,92 @@ class LLMService:
in
predict_msg]).decode())
def select_datasource(self, session: SessionDep):
datasource_msg: List[Union[BaseMessage, dict[str, Any]]] = []
datasource_msg.append(SystemMessage(self.chat_question.datasource_sys_question()))
_ds_list = session.exec(select(CoreDatasource).options(
load_only(CoreDatasource.id, CoreDatasource.name, CoreDatasource.description))).all()
_ds_list_dict = []
for _ds in _ds_list:
_ds_list_dict.append({'id': _ds[0].id, 'name': _ds[0].name, 'description': _ds[0].description})
datasource_msg.append(
HumanMessage(self.chat_question.datasource_user_question(orjson.dumps(_ds_list_dict).decode())))
history_msg = []
if self.record.full_select_datasource_message and self.record.full_select_datasource_message.strip() != '':
history_msg = orjson.loads(self.record.full_select_datasource_message)
self.record = save_full_select_datasource_message_and_answer(session=session, record_id=self.record.id,
answer='',
full_message=orjson.dumps(history_msg +
[{'type': msg.type,
'content': msg.content}
for msg
in
datasource_msg]).decode())
full_text = ''
res = self.llm.stream(datasource_msg)
for chunk in res:
print(chunk)
if isinstance(chunk, dict):
full_text += chunk['content']
yield chunk['content']
continue
if isinstance(chunk, AIMessageChunk):
full_text += chunk.content
yield chunk.content
continue
datasource_msg.append(AIMessage(full_text))
json_str = extract_nested_json(full_text)
_error: Exception | None = None
_datasource: int | None = None
_engine_type: str | None = None
try:
data = orjson.loads(json_str)
if data['id'] and data['id'] != 0:
_datasource = data['id']
_ds = session.query(CoreDatasource).filter(CoreDatasource.id == _datasource).first()
if not _ds:
_datasource = None
raise Exception(f"Datasource configuration with id {_datasource} not found")
self.ds = CoreDatasource(**_ds.model_dump())
self.chat_question.engine = _ds.type_name if _ds.type != 'excel' else 'PostgreSQL'
_engine_type = self.chat_question.engine
# save chat
_chat = session.query(Chat).filter(Chat.id == self.record.chat_id).first()
_chat.datasource = _datasource
_chat.engine_type = _ds.type_name
session.add(_chat)
session.flush()
session.refresh(_chat)
session.commit()
elif data['fail']:
raise Exception(data['fail'])
else:
raise Exception('No available datasource configuration found')
except Exception as e:
_error = e
self.record = save_full_select_datasource_message_and_answer(session=session, record_id=self.record.id,
answer=full_text, datasource=_datasource,
engine_type=_engine_type,
full_message=orjson.dumps(history_msg +
[{'type': msg.type,
'content': msg.content}
for msg
in
datasource_msg]).decode())
self.init_messages()
if _error:
raise _error
def generate_sql(self, session: SessionDep):
# append current question
self.sql_message.append(HumanMessage(self.chat_question.sql_user_question()))
@@ -0,0 +1,6 @@
from apps.template.template import get_base_template
def get_guess_question_template():
template = get_base_template()
return template['template']['guess']
@@ -0,0 +1,6 @@
from apps.template.template import get_base_template
def get_datasource_template():
template = get_base_template()
return template['template']['datasource']
+38 -17
View File
@@ -10,9 +10,9 @@ template:
- SQL查询的字段名不要自动翻译,别名必须为英文。
- 计算占比,百分比类型字段,保留两位小数,以%结尾。
- 生成SQL时,必须避免关键字冲突:
- 如{engine}是 PostgreSQL、Oracle,则在schema、表名、字段名、别名外层加双引号;
- 如{engine}是 MySQL,则在表名、字段名、别名外层加反引号;
- 如{engine}是 Microsoft SQL Server,则在schema、表名、字段名、别名外层加方括号。
- 如数据库引擎是 PostgreSQL、Oracle,则在schema、表名、字段名、别名外层加双引号;
- 如数据库引擎是 MySQL,则在表名、字段名、别名外层加反引号;
- 如数据库引擎是 Microsoft SQL Server,则在schema、表名、字段名、别名外层加方括号。
- 生成的SQL使用JSON格式返回:
{{"success":true,"sql":"生成的SQL语句"}}
- 问题与生成SQL无关时,直接回答:
@@ -94,30 +94,31 @@ template:
guess:
system: |
### 说明:
您的任务是根据给定的表结构和问题,推测用户接下来可能提问的1-4个问题。
您的任务是根据给定的表结构,用户问题以及以往用户提问,推测用户接下来可能提问的1-4个问题。
请遵循以下规则:
- 推测的问题不能与用户问题重复。
- 推测的问题必须与给出的表结构相关。
- 推测的问题不能与当前用户问题重复
- 推测的问题必须与给出的表结构相关
- 若有以往用户提问列表,则根据以往用户提问列表,推测用户最频繁提问的问题,加入到你生成的推测问题中
- 生成的推测问题使用JSON格式返回:
{["推测问题1", "推测问题2", "推测问题3", "推测问题4"]}
["推测问题1", "推测问题2", "推测问题3", "推测问题4"]
- 最多返回4个你推测出的结果
- 若无法推测,则返回空数据JSON:
[]
### 响应:
根据您的指示,这是我生成的与 表结构 和 问题 相关的 JSON:
```json
### 请遵守以下输出要求:
请逐步推理后给出答案
推理过程中不需要输出JSON,仅在最终结果内输出符合要求的JSON
步骤1: [思考内容]
步骤2: [思考内容]
……
最终答案: [结果]
user: |
### 请使用 i18n: {lang} 对应的语言输出你的结果
### 表结构:
{schema}
### 问题:
### 当前问题:
{question}
### 以往提问:
{old_questions}
analysis:
system: |
### 说明:
@@ -148,3 +149,23 @@ template:
### 数据:
{data}
datasource:
system: |
### 说明:
你是一个数据分析师,你需要根据用户的提问,以及提供的数据源列表(格式为JSON数组:[{{"id": 数据源ID1,"name":"数据源名称1","description":"数据源描述1"}},{{"id": 数据源ID2,"name":"数据源名称2","description":"数据源描述2"}}]),根据名称和描述找出最符合用户提问的数据源,这个数据源后续将被用来进行数据的分析
### 要求:
- 以JSON格式返回你找到的符合提问的数据源ID,格式为:{{"id": 符合要求的数据源ID}}
- 如果匹配到多个数据源,则只需要返回其中一个即可
- 如果没有符合要求的数据源,则返回:{{"fail":"没有找到匹配的数据源"}}
- 不需要思考过程,请直接返回JSON结果
```json
user: |
### 请使用 i18n: {lang} 对应的语言输出你的结果
### 数据源列表:
{data}
### 问题:
{question}
@@ -63,7 +63,7 @@ defineExpose({
<template>
<div v-if="message?.record">
<div v-if="message.record.chart" class="chart-base-container">
<div>
<div v-if="data?.length > 0">
<ChartComponent
v-if="message.record.id"
:id="id ?? 'default_chat_id'"
@@ -76,6 +76,7 @@ defineExpose({
:data="data"
/>
</div>
<el-empty v-else description="No Data" />
</div>
</div>
</template>
+2
View File
@@ -409,6 +409,8 @@ const sendMessage = async () => {
currentRecord.error = currentRecord.error + 'Error:' + error
console.error('Error:', error)
isTyping.value = false
} finally {
loading.value = false
}
}