feat: add chart reasoning content

This commit is contained in:
ulleo
2025-07-10 19:13:57 +08:00
parent e69caadd09
commit c909cb78d8
12 changed files with 592 additions and 185 deletions
+23 -41
View File
@@ -1,13 +1,14 @@
import json
from pydantic import BaseModel
from typing import Optional, Dict, Any, Type
from abc import ABC, abstractmethod
from langchain_core.language_models import BaseLLM as LangchainBaseLLM
from langchain_openai import ChatOpenAI
from typing import Optional, Dict, Any, Type
from langchain.chat_models.base import BaseChatModel
from pydantic import BaseModel
from sqlmodel import Session, select
from common.core.db import engine
from apps.ai_model.openai.llm import BaseChatOpenAI
from apps.system.models.system_model import AiModelDetail
from common.core.db import engine
# from langchain_community.llms import Tongyi, VLLM
@@ -20,76 +21,57 @@ class LLMConfig(BaseModel):
api_key: Optional[str] = None
api_base_url: Optional[str] = None
additional_params: Dict[str, Any] = {}
class BaseLLM(ABC):
"""Abstract base class for large language models"""
def __init__(self, config: LLMConfig):
self.config = config
self._llm = self._init_llm()
@abstractmethod
def _init_llm(self) -> LangchainBaseLLM:
def _init_llm(self) -> BaseChatModel:
"""Initialize specific large language model instance"""
pass
@property
def llm(self) -> LangchainBaseLLM:
def llm(self) -> BaseChatModel:
"""Return the langchain LLM instance"""
return self._llm
class OpenAILLM(BaseLLM):
def _init_llm(self) -> LangchainBaseLLM:
return ChatOpenAI(
def _init_llm(self) -> BaseChatModel:
return BaseChatOpenAI(
model=self.config.model_name,
api_key=self.config.api_key,
base_url=self.config.api_base_url,
stream_usage=True,
**self.config.additional_params
**self.config.additional_params,
extra_body={"enable_thinking": True},
)
def generate(self, prompt: str) -> str:
return self.llm.invoke(prompt)
""" class TongyiLLM(BaseLLM):
def _init_llm(self) -> LangchainBaseLLM:
return Tongyi(
model_name=self.config.model_name,
dashscope_api_key=self.config.api_key,
**self.config.additional_params
)
def generate(self, prompt: str) -> str:
return self.llm.invoke(prompt)
class VLLMLLM(BaseLLM):
def _init_llm(self) -> LangchainBaseLLM:
return VLLM(
model=self.config.model_name,
**self.config.additional_params
)
def generate(self, prompt: str) -> str:
return self.llm.invoke(prompt) """
class LLMFactory:
"""Large Language Model Factory Class"""
_llm_types: Dict[str, Type[BaseLLM]] = {
"openai": OpenAILLM,
"tongyi": OpenAILLM,
"vllm": OpenAILLM
}
@classmethod
def create_llm(cls, config: LLMConfig) -> BaseLLM:
llm_class = cls._llm_types.get(config.model_type)
if not llm_class:
raise ValueError(f"Unsupported LLM type: {config.model_type}")
return llm_class(config)
@classmethod
def register_llm(cls, model_type: str, llm_class: Type[BaseLLM]):
"""Register new model type"""
@@ -107,6 +89,7 @@ class LLMFactory:
)
return config """
def get_default_config() -> LLMConfig:
with Session(engine) as session:
db_model = session.exec(
@@ -130,6 +113,5 @@ def get_default_config() -> LLMConfig:
model_name=db_model.base_model,
api_key=db_model.api_key,
api_base_url=db_model.api_domain,
additional_params=additional_params
additional_params=additional_params,
)
+167 -1
View File
@@ -1 +1,167 @@
# todo
from typing import Dict, Optional, Any, Iterator, cast, Mapping
from langchain_core.language_models import LanguageModelInput
from langchain_core.messages import BaseMessage, BaseMessageChunk, HumanMessageChunk, AIMessageChunk, \
SystemMessageChunk, FunctionMessageChunk, ChatMessageChunk
from langchain_core.messages.ai import UsageMetadata
from langchain_core.messages.tool import tool_call_chunk, ToolMessageChunk
from langchain_core.outputs import ChatGenerationChunk
from langchain_core.runnables import RunnableConfig, ensure_config
from langchain_openai import ChatOpenAI
from langchain_openai.chat_models.base import _create_usage_metadata
def _convert_delta_to_message_chunk(
_dict: Mapping[str, Any], default_class: type[BaseMessageChunk]
) -> BaseMessageChunk:
id_ = _dict.get("id")
role = cast(str, _dict.get("role"))
content = cast(str, _dict.get("content") or "")
additional_kwargs: dict = {}
if 'reasoning_content' in _dict:
additional_kwargs['reasoning_content'] = _dict.get('reasoning_content')
if _dict.get("function_call"):
function_call = dict(_dict["function_call"])
if "name" in function_call and function_call["name"] is None:
function_call["name"] = ""
additional_kwargs["function_call"] = function_call
tool_call_chunks = []
if raw_tool_calls := _dict.get("tool_calls"):
additional_kwargs["tool_calls"] = raw_tool_calls
try:
tool_call_chunks = [
tool_call_chunk(
name=rtc["function"].get("name"),
args=rtc["function"].get("arguments"),
id=rtc.get("id"),
index=rtc["index"],
)
for rtc in raw_tool_calls
]
except KeyError:
pass
if role == "user" or default_class == HumanMessageChunk:
return HumanMessageChunk(content=content, id=id_)
elif role == "assistant" or default_class == AIMessageChunk:
return AIMessageChunk(
content=content,
additional_kwargs=additional_kwargs,
id=id_,
tool_call_chunks=tool_call_chunks, # type: ignore[arg-type]
)
elif role in ("system", "developer") or default_class == SystemMessageChunk:
if role == "developer":
additional_kwargs = {"__openai_role__": "developer"}
else:
additional_kwargs = {}
return SystemMessageChunk(
content=content, id=id_, additional_kwargs=additional_kwargs
)
elif role == "function" or default_class == FunctionMessageChunk:
return FunctionMessageChunk(content=content, name=_dict["name"], id=id_)
elif role == "tool" or default_class == ToolMessageChunk:
return ToolMessageChunk(
content=content, tool_call_id=_dict["tool_call_id"], id=id_
)
elif role or default_class == ChatMessageChunk:
return ChatMessageChunk(content=content, role=role, id=id_)
else:
return default_class(content=content, id=id_)
class BaseChatOpenAI(ChatOpenAI):
usage_metadata: dict = {}
# custom_get_token_ids = custom_get_token_ids
def get_last_generation_info(self) -> Optional[Dict[str, Any]]:
return self.usage_metadata
def _stream(self, *args: Any, **kwargs: Any) -> Iterator[ChatGenerationChunk]:
kwargs['stream_usage'] = True
for chunk in super()._stream(*args, **kwargs):
if chunk.message.usage_metadata is not None:
self.usage_metadata = chunk.message.usage_metadata
yield chunk
def _convert_chunk_to_generation_chunk(
self,
chunk: dict,
default_chunk_class: type,
base_generation_info: Optional[dict],
) -> Optional[ChatGenerationChunk]:
if chunk.get("type") == "content.delta": # from beta.chat.completions.stream
return None
token_usage = chunk.get("usage")
choices = (
chunk.get("choices", [])
# from beta.chat.completions.stream
or chunk.get("chunk", {}).get("choices", [])
)
usage_metadata: Optional[UsageMetadata] = (
_create_usage_metadata(token_usage) if token_usage and token_usage.get("prompt_tokens") else None
)
if len(choices) == 0:
# logprobs is implicitly None
generation_chunk = ChatGenerationChunk(
message=default_chunk_class(content="", usage_metadata=usage_metadata)
)
return generation_chunk
choice = choices[0]
if choice["delta"] is None:
return None
message_chunk = _convert_delta_to_message_chunk(
choice["delta"], default_chunk_class
)
generation_info = {**base_generation_info} if base_generation_info else {}
if finish_reason := choice.get("finish_reason"):
generation_info["finish_reason"] = finish_reason
if model_name := chunk.get("model"):
generation_info["model_name"] = model_name
if system_fingerprint := chunk.get("system_fingerprint"):
generation_info["system_fingerprint"] = system_fingerprint
logprobs = choice.get("logprobs")
if logprobs:
generation_info["logprobs"] = logprobs
if usage_metadata and isinstance(message_chunk, AIMessageChunk):
message_chunk.usage_metadata = usage_metadata
generation_chunk = ChatGenerationChunk(
message=message_chunk, generation_info=generation_info or None
)
return generation_chunk
def invoke(
self,
input: LanguageModelInput,
config: Optional[RunnableConfig] = None,
*,
stop: Optional[list[str]] = None,
**kwargs: Any,
) -> BaseMessage:
config = ensure_config(config)
chat_result = cast(
"ChatGeneration",
self.generate_prompt(
[self._convert_input(input)],
stop=stop,
callbacks=config.get("callbacks"),
tags=config.get("tags"),
metadata=config.get("metadata"),
run_name=config.get("run_name"),
run_id=config.pop("run_id", None),
**kwargs,
).generations[0][0],
).message
self.usage_metadata = chat_result.response_metadata[
'token_usage'] if 'token_usage' in chat_result.response_metadata else chat_result.usage_metadata
return chat_result
+30 -5
View File
@@ -4,7 +4,7 @@ from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from apps.chat.curd.chat import list_chats, get_chat_with_records, create_chat, rename_chat, \
delete_chat
delete_chat, get_chat_chart_data, get_chat_predict_data
from apps.chat.models.chat_model import CreateChat, ChatRecord, RenameChat, ChatQuestion
from apps.chat.task.llm import LLMService, run_task, run_analysis_or_predict_task, run_recommend_questions_task
from common.core.deps import SessionDep, CurrentUser
@@ -28,6 +28,28 @@ async def get_chat(session: SessionDep, current_user: CurrentUser, chart_id: int
)
@router.get("/record/get/{chart_record_id}/data")
async def chat_record_data(session: SessionDep, chart_record_id: int):
try:
return get_chat_chart_data(chart_record_id=chart_record_id, session=session)
except Exception as e:
raise HTTPException(
status_code=500,
detail=str(e)
)
@router.get("/record/get/{chart_record_id}/predict_data")
async def chat_predict_data(session: SessionDep, chart_record_id: int):
try:
return get_chat_predict_data(chart_record_id=chart_record_id, session=session)
except Exception as e:
raise HTTPException(
status_code=500,
detail=str(e)
)
@router.post("/rename")
async def rename(session: SessionDep, chat: RenameChat):
try:
@@ -61,7 +83,7 @@ async def start_chat(session: SessionDep, current_user: CurrentUser, create_chat
)
@router.get("/recommend_questions/{chat_record_id}")
@router.post("/recommend_questions/{chat_record_id}")
async def recommend_questions(session: SessionDep, current_user: CurrentUser, chat_record_id: int):
try:
record = session.query(ChatRecord).get(chat_record_id)
@@ -74,11 +96,14 @@ async def recommend_questions(session: SessionDep, current_user: CurrentUser, ch
llm_service = LLMService(session, current_user, request_question)
llm_service.set_record(record)
return run_recommend_questions_task(llm_service)
except Exception as e:
traceback.print_exc()
return '[]'
raise HTTPException(
status_code=500,
detail=str(e)
)
return StreamingResponse(run_recommend_questions_task(llm_service), media_type="text/event-stream")
@router.post("/question")
+69 -10
View File
@@ -1,7 +1,8 @@
import datetime
from typing import List
from sqlalchemy import and_, distinct
import orjson
from sqlalchemy import and_
from sqlalchemy.orm import load_only
from apps.chat.models.chat_model import Chat, ChatRecord, CreateChat, ChatInfo, RenameChat, ChatQuestion
@@ -42,6 +43,26 @@ def delete_chat(session, chart_id) -> str:
return f'Chat with id {chart_id} has been deleted'
def get_chat_chart_data(session: SessionDep, chart_record_id: int):
res = session.query(ChatRecord).options(load_only(ChatRecord.data)).get(chart_record_id)
if res:
try:
return orjson.loads(res.data)
except Exception:
pass
return {}
def get_chat_predict_data(session: SessionDep, chart_record_id: int):
res = session.query(ChatRecord).options(load_only(ChatRecord.predict_data)).get(chart_record_id)
if res:
try:
return orjson.loads(res.data)
except Exception:
pass
return ''
def get_chat_with_records(session: SessionDep, chart_id: int, current_user: CurrentUser) -> ChatInfo:
chat = session.query(Chat).filter(Chat.id == chart_id).first()
if not chat:
@@ -61,16 +82,54 @@ 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.datasource_select_answer,
ChatRecord.recommended_question, ChatRecord.first_chat,
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()
chat_info.records = record_list
result = list(map(format_record, record_list))
chat_info.records = result
return chat_info
def format_record(record: ChatRecord):
_dict = record.model_dump()
if record.sql_answer and record.sql_answer.strip() != '' and record.sql_answer.strip()[0] == '{' and \
record.sql_answer.strip()[-1] == '}':
_obj = orjson.loads(record.sql_answer)
_dict['sql_answer'] = _obj.get('reasoning_content')
if record.chart_answer and record.chart_answer.strip() != '' and record.chart_answer.strip()[0] == '{' and \
record.chart_answer.strip()[-1] == '}':
_obj = orjson.loads(record.chart_answer)
_dict['chart_answer'] = _obj.get('reasoning_content')
if record.analysis and record.analysis.strip() != '' and record.analysis.strip()[0] == '{' and \
record.analysis.strip()[-1] == '}':
_obj = orjson.loads(record.analysis)
_dict['analysis_thinking'] = _obj.get('reasoning_content')
_dict['analysis'] = _obj.get('content')
if record.predict and record.predict.strip() != '' and record.predict.strip()[0] == '{' and record.predict.strip()[
-1] == '}':
_obj = orjson.loads(record.predict)
_dict['predict'] = _obj.get('reasoning_content')
_dict['predict_content'] = _obj.get('content')
if record.data and record.data.strip() != '':
try:
_obj = orjson.loads(record.data)
_dict['data'] = _obj
except Exception:
pass
if record.predict_data and record.predict_data.strip() != '':
try:
_obj = orjson.loads(record.predict_data)
_dict['predict_data'] = _obj
except Exception:
pass
return _dict
def list_records(session: SessionDep, chart_id: int, current_user: CurrentUser) -> List[ChatRecord]:
record_list = session.query(ChatRecord).filter(
and_(Chat.create_by == current_user.id, ChatRecord.chat_id == chart_id)).order_by(ChatRecord.create_time).all()
@@ -251,18 +310,18 @@ def save_full_select_datasource_message_and_answer(session: SessionDep, record_i
return result
def save_full_recommend_question_message_and_answer(session: SessionDep, record_id: int, answer: str,
full_message: str) -> ChatRecord:
def save_full_recommend_question_message_and_answer(session: SessionDep, record_id: int, answer: dict = None,
full_message: str = '[]') -> ChatRecord:
if not record_id:
raise Exception("Record id cannot be None")
record = session.query(ChatRecord).filter(ChatRecord.id == record_id).first()
record.full_recommended_question_message = full_message
record.recommended_question_answer = answer
record.recommended_question_answer = orjson.dumps(answer).decode()
json_str = '[]'
if answer and answer != '':
if answer and answer.get('content') and answer.get('content') != '':
try:
json_str = extract_nested_json(answer)
json_str = extract_nested_json(answer.get('content'))
if not json_str:
json_str = '[]'
@@ -339,7 +398,7 @@ def save_chart(session: SessionDep, record_id: int, chart: str) -> ChatRecord:
return result
def save_predict_data(session: SessionDep, record_id: int, data: str) -> ChatRecord:
def save_predict_data(session: SessionDep, record_id: int, data: str = '') -> ChatRecord:
if not record_id:
raise Exception("Record id cannot be None")
record = session.query(ChatRecord).filter(ChatRecord.id == record_id).first()
@@ -414,6 +473,6 @@ def get_old_questions(session: SessionDep, datasource: int):
if not datasource:
return []
records = session.query(ChatRecord.question, ChatRecord.create_time).filter(ChatRecord.datasource == datasource,
ChatRecord.question != None).order_by(
ChatRecord.question != None).order_by(
ChatRecord.create_time.desc()).limit(20).all()
return records
+2 -2
View File
@@ -2,7 +2,7 @@ from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel
from sqlalchemy import Column, Text, BigInteger, DateTime, Integer, Identity, Boolean
from sqlalchemy import Column, Text, BigInteger, DateTime, Identity, Boolean
from sqlmodel import SQLModel, Field
from apps.template.generate_analysis.generator import get_analysis_template
@@ -86,7 +86,7 @@ class ChatInfo(BaseModel):
engine_type: str = ''
datasource_name: str = ''
datasource_exists: bool = True
records: List[ChatRecord] = []
records: List[ChatRecord | dict] = []
class AiModelQuestion(BaseModel):
+133 -75
View File
@@ -7,9 +7,9 @@ import numpy as np
import orjson
import pandas as pd
import requests
from langchain.chat_models.base import BaseChatModel
from langchain_community.utilities import SQLDatabase
from langchain_core.language_models import BaseLLM
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage, AIMessageChunk
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage
from sqlalchemy import select
from sqlalchemy.orm import load_only
@@ -37,7 +37,7 @@ class LLMService:
chat_question: ChatQuestion
record: ChatRecord
config: LLMConfig
llm: BaseLLM
llm: BaseChatModel
sql_message: List[Union[BaseMessage, dict[str, Any]]] = []
chart_message: List[Union[BaseMessage, dict[str, Any]]] = []
history_records: List[ChatRecord] = []
@@ -195,23 +195,27 @@ class LLMService:
'content': msg.content} for msg
in
analysis_msg]).decode())
full_thinking_text = ''
full_analysis_text = ''
res = self.llm.stream(analysis_msg)
for chunk in res:
print(chunk)
if isinstance(chunk, dict):
full_analysis_text += chunk['content']
yield chunk['content']
continue
if isinstance(chunk, AIMessageChunk):
full_analysis_text += chunk.content
yield chunk.content
continue
reasoning_content_chunk = ''
if 'reasoning_content' in chunk.additional_kwargs:
reasoning_content_chunk = chunk.additional_kwargs.get('reasoning_content', '')
# else:
# reasoning_content_chunk = chunk.get('reasoning_content')
if reasoning_content_chunk is None:
reasoning_content_chunk = ''
full_thinking_text += reasoning_content_chunk
full_analysis_text += chunk.content
yield {'content': chunk.content, 'reasoning_content': reasoning_content_chunk}
analysis_msg.append(AIMessage(full_analysis_text))
self.record = save_full_analysis_message_and_answer(session=self.session, record_id=self.record.id,
answer=full_analysis_text,
answer=orjson.dumps({'content': full_analysis_text,
'reasoning_content': full_thinking_text}).decode(),
full_message=orjson.dumps(history_msg +
[{'type': msg.type,
'content': msg.content} for msg
@@ -238,23 +242,28 @@ class LLMService:
'content': msg.content} for msg
in
predict_msg]).decode())
full_thinking_text = ''
full_predict_text = ''
res = self.llm.stream(predict_msg)
for chunk in res:
print(chunk)
if isinstance(chunk, dict):
full_predict_text += chunk['content']
yield chunk['content']
continue
if isinstance(chunk, AIMessageChunk):
full_predict_text += chunk.content
yield chunk.content
continue
reasoning_content_chunk = ''
if 'reasoning_content' in chunk.additional_kwargs:
reasoning_content_chunk = chunk.additional_kwargs.get('reasoning_content', '')
# else:
# reasoning_content_chunk = chunk.get('reasoning_content')
if reasoning_content_chunk is None:
reasoning_content_chunk = ''
full_thinking_text += reasoning_content_chunk
full_predict_text += chunk.content
yield {'content': chunk.content, 'reasoning_content': reasoning_content_chunk}
predict_msg.append(AIMessage(full_predict_text))
self.record = save_full_predict_message_and_answer(session=self.session, record_id=self.record.id,
answer=full_predict_text, data='',
answer=orjson.dumps({'content': full_predict_text,
'reasoning_content': full_thinking_text}).decode(),
data='',
full_message=orjson.dumps(history_msg +
[{'type': msg.type,
'content': msg.content} for msg
@@ -269,38 +278,44 @@ class LLMService:
guess_msg: List[Union[BaseMessage, dict[str, Any]]] = []
guess_msg.append(SystemMessage(content=self.chat_question.guess_sys_question()))
# todo old questions
old_questions = list(map(lambda q: q[0].strip(), get_old_questions(self.session, self.record.datasource)))
guess_msg.append(HumanMessage(content=self.chat_question.guess_user_question(orjson.dumps(old_questions).decode())))
guess_msg.append(
HumanMessage(content=self.chat_question.guess_user_question(orjson.dumps(old_questions).decode())))
self.record = save_full_recommend_question_message_and_answer(session=self.session, record_id=self.record.id,
answer='',
full_message=orjson.dumps([{'type': msg.type,
'content': msg.content}
for msg
in
guess_msg]).decode())
full_thinking_text = ''
full_guess_text = ''
res = self.llm.stream(guess_msg)
for chunk in res:
print(chunk)
if isinstance(chunk, dict):
full_guess_text += chunk['content']
continue
if isinstance(chunk, AIMessageChunk):
full_guess_text += chunk.content
continue
reasoning_content_chunk = ''
if 'reasoning_content' in chunk.additional_kwargs:
reasoning_content_chunk = chunk.additional_kwargs.get('reasoning_content', '')
# else:
# reasoning_content_chunk = chunk.get('reasoning_content')
if reasoning_content_chunk is None:
reasoning_content_chunk = ''
full_thinking_text += reasoning_content_chunk
full_guess_text += chunk.content
yield {'content': chunk.content, 'reasoning_content': reasoning_content_chunk}
guess_msg.append(AIMessage(full_guess_text))
self.record = save_full_recommend_question_message_and_answer(session=self.session, record_id=self.record.id,
answer=full_guess_text,
answer={'content': full_guess_text,
'reasoning_content': full_thinking_text},
full_message=orjson.dumps([{'type': msg.type,
'content': msg.content}
for msg
in
guess_msg]).decode())
return self.record.recommended_question
yield {'recommended_question': self.record.recommended_question}
def select_datasource(self):
datasource_msg: List[Union[BaseMessage, dict[str, Any]]] = []
@@ -325,17 +340,22 @@ class LLMService:
for msg
in
datasource_msg]).decode())
full_thinking_text = ''
full_text = ''
res = self.llm.stream(datasource_msg)
for chunk in res:
if isinstance(chunk, dict):
full_text += chunk['content']
yield chunk['content']
continue
if isinstance(chunk, AIMessageChunk):
full_text += chunk.content
yield chunk.content
continue
print(chunk)
reasoning_content_chunk = ''
if 'reasoning_content' in chunk.additional_kwargs:
reasoning_content_chunk = chunk.additional_kwargs.get('reasoning_content', '')
# else:
# reasoning_content_chunk = chunk.get('reasoning_content')
if reasoning_content_chunk is None:
reasoning_content_chunk = ''
full_thinking_text += reasoning_content_chunk
full_text += chunk.content
yield {'content': chunk.content, 'reasoning_content': reasoning_content_chunk}
datasource_msg.append(AIMessage(full_text))
json_str = extract_nested_json(full_text)
@@ -374,7 +394,9 @@ class LLMService:
_error = e
self.record = save_full_select_datasource_message_and_answer(session=self.session, record_id=self.record.id,
answer=full_text, datasource=_datasource,
answer=orjson.dumps({'content': full_text,
'reasoning_content': full_thinking_text}).decode(),
datasource=_datasource,
engine_type=_engine_type,
full_message=orjson.dumps(history_msg +
[{'type': msg.type,
@@ -394,22 +416,27 @@ class LLMService:
full_message=orjson.dumps(
[{'type': msg.type, 'content': msg.content} for msg in
self.sql_message]).decode())
full_thinking_text = ''
full_sql_text = ''
res = self.llm.stream(self.sql_message)
for chunk in res:
print(chunk)
if isinstance(chunk, dict):
full_sql_text += chunk['content']
yield chunk['content']
continue
if isinstance(chunk, AIMessageChunk):
full_sql_text += chunk.content
yield chunk.content
continue
reasoning_content_chunk = ''
if 'reasoning_content' in chunk.additional_kwargs:
reasoning_content_chunk = chunk.additional_kwargs.get('reasoning_content', '')
# else:
# reasoning_content_chunk = chunk.get('reasoning_content')
if reasoning_content_chunk is None:
reasoning_content_chunk = ''
full_thinking_text += reasoning_content_chunk
full_sql_text += chunk.content
yield {'content': chunk.content, 'reasoning_content': reasoning_content_chunk}
self.sql_message.append(AIMessage(full_sql_text))
self.record = save_full_sql_message_and_answer(session=self.session, record_id=self.record.id,
answer=full_sql_text,
answer=orjson.dumps({'content': full_sql_text,
'reasoning_content': full_thinking_text}).decode(),
full_message=orjson.dumps(
[{'type': msg.type, 'content': msg.content} for msg in
self.sql_message]).decode())
@@ -421,21 +448,27 @@ class LLMService:
full_message=orjson.dumps(
[{'type': msg.type, 'content': msg.content} for msg in
self.chart_message]).decode())
full_thinking_text = ''
full_chart_text = ''
res = self.llm.stream(self.chart_message)
for chunk in res:
if isinstance(chunk, dict):
full_chart_text += chunk['content']
yield chunk['content']
continue
if isinstance(chunk, AIMessageChunk):
full_chart_text += chunk.content
yield chunk.content
continue
print(chunk)
reasoning_content_chunk = ''
if 'reasoning_content' in chunk.additional_kwargs:
reasoning_content_chunk = chunk.additional_kwargs.get('reasoning_content', '')
# else:
# reasoning_content_chunk = chunk.get('reasoning_content')
if reasoning_content_chunk is None:
reasoning_content_chunk = ''
full_thinking_text += reasoning_content_chunk
full_chart_text += chunk.content
yield {'content': chunk.content, 'reasoning_content': reasoning_content_chunk}
self.chart_message.append(AIMessage(full_chart_text))
self.record = save_full_chart_message_and_answer(session=self.session, record_id=self.record.id,
answer=full_chart_text,
answer=orjson.dumps({'content': full_chart_text,
'reasoning_content': full_thinking_text}).decode(),
full_message=orjson.dumps(
[{'type': msg.type, 'content': msg.content} for msg in
self.chart_message]).decode())
@@ -492,7 +525,7 @@ class LLMService:
return chart
def check_save_predict_data(self, res: str) -> Dict[str, Any]:
def check_save_predict_data(self, res: str) -> bool:
json_str = extract_nested_json(res)
@@ -501,7 +534,10 @@ class LLMService:
save_predict_data(session=self.session, record_id=self.record.id, data=json_str)
return json_str
if json_str == '':
return False
return True
def save_error(self, message: str):
return save_error_message(session=self.session, record_id=self.record.id, message=message)
@@ -566,7 +602,9 @@ def run_task(llm_service: LLMService, session: SessionDep, in_chat: bool = True)
for chunk in ds_res:
print(chunk)
if in_chat:
yield orjson.dumps({'content': chunk, 'type': 'datasource-result'}).decode() + '\n\n'
yield orjson.dumps(
{'content': chunk.get('content'), 'reasoning_content': chunk.get('reasoning_content'),
'type': 'datasource-result'}).decode() + '\n\n'
if in_chat:
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'
@@ -577,9 +615,11 @@ def run_task(llm_service: LLMService, session: SessionDep, in_chat: bool = True)
sql_res = llm_service.generate_sql()
full_sql_text = ''
for chunk in sql_res:
full_sql_text += chunk
full_sql_text += chunk.get('content')
if in_chat:
yield orjson.dumps({'content': chunk, 'type': 'sql-result'}).decode() + '\n\n'
yield orjson.dumps(
{'content': chunk.get('content'), 'reasoning_content': chunk.get('reasoning_content'),
'type': 'sql-result'}).decode() + '\n\n'
if in_chat:
yield orjson.dumps({'type': 'info', 'msg': 'sql generated'}).decode() + '\n\n'
@@ -596,15 +636,17 @@ def run_task(llm_service: LLMService, session: SessionDep, in_chat: bool = True)
result = llm_service.execute_sql(sql=sql)
llm_service.save_sql_data(data_obj=result)
if in_chat:
yield orjson.dumps({'content': orjson.dumps(result).decode(), 'type': 'sql-data'}).decode() + '\n\n'
yield orjson.dumps({'content': 'execute-success', 'type': 'sql-data'}).decode() + '\n\n'
# generate chart
chart_res = llm_service.generate_chart()
full_chart_text = ''
for chunk in chart_res:
full_chart_text += chunk
full_chart_text += chunk.get('content')
if in_chat:
yield orjson.dumps({'content': chunk, 'type': 'chart-result'}).decode() + '\n\n'
yield orjson.dumps(
{'content': chunk.get('content'), 'reasoning_content': chunk.get('reasoning_content'),
'type': 'chart-result'}).decode() + '\n\n'
if in_chat:
yield orjson.dumps({'type': 'info', 'msg': 'chart generated'}).decode() + '\n\n'
@@ -667,7 +709,9 @@ def run_analysis_or_predict_task(llm_service: LLMService, action_type: str):
# generate analysis
analysis_res = llm_service.generate_analysis()
for chunk in analysis_res:
yield orjson.dumps({'content': chunk, 'type': 'analysis-result'}).decode() + '\n\n'
yield orjson.dumps(
{'content': chunk.get('content'), 'reasoning_content': chunk.get('reasoning_content'),
'type': 'analysis-result'}).decode() + '\n\n'
yield orjson.dumps({'type': 'info', 'msg': 'analysis generated'}).decode() + '\n\n'
yield orjson.dumps({'type': 'analysis_finish'}).decode() + '\n\n'
@@ -677,12 +721,17 @@ def run_analysis_or_predict_task(llm_service: LLMService, action_type: str):
analysis_res = llm_service.generate_predict()
full_text = ''
for chunk in analysis_res:
yield orjson.dumps({'content': chunk, 'type': 'predict-result'}).decode() + '\n\n'
full_text += chunk
yield orjson.dumps(
{'content': chunk.get('content'), 'reasoning_content': chunk.get('reasoning_content'),
'type': 'predict-result'}).decode() + '\n\n'
full_text += chunk.get('content')
yield orjson.dumps({'type': 'info', 'msg': 'predict generated'}).decode() + '\n\n'
_data = llm_service.check_save_predict_data(res=full_text)
yield orjson.dumps({'type': 'predict', 'content': _data}).decode() + '\n\n'
if _data:
yield orjson.dumps({'type': 'predict-success'}).decode() + '\n\n'
else:
yield orjson.dumps({'type': 'predict-failed'}).decode() + '\n\n'
yield orjson.dumps({'type': 'predict_finish'}).decode() + '\n\n'
@@ -694,7 +743,16 @@ def run_analysis_or_predict_task(llm_service: LLMService, action_type: str):
def run_recommend_questions_task(llm_service: LLMService):
return llm_service.generate_recommend_questions_task()
res = llm_service.generate_recommend_questions_task()
for chunk in res:
if chunk.get('recommended_question'):
yield orjson.dumps(
{'content': chunk.get('recommended_question'), 'type': 'recommended_question'}).decode() + '\n\n'
else:
yield orjson.dumps(
{'content': chunk.get('content'), 'reasoning_content': chunk.get('reasoning_content'),
'type': 'recommended_question_result'}).decode() + '\n\n'
def request_picture(chat_id: int, record_id: int, chart: dict, data: dict):
+10 -20
View File
@@ -23,13 +23,9 @@ template:
- 如果问题是图表展示相关,可参考的图表类型为表格(table)、柱状图(column)、条形图(bar)、折线图(line)或饼图(pie),返回的JSON:
{{"success":true,"sql":"生成的SQL语句","chart-type":"选择的图表类型(table/column/bar/line/pie)"}}
### 请遵守以下输出要求:
- 请逐步推理后给出答案
- 推理过程的步骤中不需要输出任何JSON,仅在最终结果内输出符合要求的JSON
步骤1: [思考内容]
步骤2: [思考内容]
……
最终答案: [结果]
### 响应, 请直接返回JSON结果:
```json
user: |
### 表结构:
{schema}
@@ -37,7 +33,7 @@ template:
### 问题:
{question}
### 请使用 i18n: {lang} 对应的语言输出你的推理思考步骤、内容和结果
### 请使用 i18n: {lang} 对应的语言输出你的结果
### 其他规则:
{rule}
@@ -69,17 +65,9 @@ template:
问题是:每个商品分类的平均价格
则生成的 JSON 可以是: {{"type":"table", "title": "每个商品分类的平均价格", "columns": [{{"name":"商品分类","value":"category"}}, {{"name":"平均价格","value":"average_price"}}]}}
### 响应:
根据您的指示,这是我生成的与 问题 和 sql 匹配的 JSON:
### 响应, 请直接返回JSON结果:
```json
### 请遵守以下输出要求:
- 请逐步推理后给出答案
- 推理过程的步骤中不需要输出任何JSON,仅在最终结果内输出符合要求的JSON
步骤1: [思考内容]
步骤2: [思考内容]
……
最终答案: [结果]
user: |
### SQL:
{sql}
@@ -87,7 +75,7 @@ template:
### 问题:
{question}
### 请使用 i18n: {lang} 对应的语言输出你的推理思考步骤、内容和结果
### 请使用 i18n: {lang} 对应的语言输出你的结果
### 其他规则:
{rule}
@@ -108,8 +96,9 @@ template:
- 最多返回4个你推测出的结果
- 若无法推测,则返回空数据JSON:
[]
- 不需要思考过程,或者尽量精简思考过程
### 响应:
### 响应, 请直接返回JSON结果:
```json
user: |
@@ -164,6 +153,7 @@ template:
- 如果没有符合要求的数据源,则返回:{{"fail":"没有找到匹配的数据源"}}
- 不需要思考过程,请直接返回JSON结果
### 响应, 请直接返回JSON结果:
```json
user: |
### 请使用 i18n: {lang} 对应的语言输出你的结果
+23 -7
View File
@@ -34,12 +34,14 @@ export class ChatRecord {
question?: string
sql_answer?: string
sql?: string
data?: string
data?: string | any
chart_answer?: string
chart?: string
analysis?: string
analysis_thinking?: string
predict?: string
predict_data?: string
predict_content?: string
predict_data?: string | any
finish?: boolean = false
error?: string
run_time: number = 0
@@ -55,12 +57,14 @@ export class ChatRecord {
question: string,
sql_answer: string | undefined,
sql: string | undefined,
data: string | undefined,
data: string | any | undefined,
chart_answer: string | undefined,
chart: string | undefined,
analysis: string | undefined,
analysis_thinking: string | undefined,
predict: string | undefined,
predict_data: string | undefined,
predict_content: string | undefined,
predict_data: string | any | undefined,
finish: boolean,
error: string | undefined,
run_time: number,
@@ -75,12 +79,14 @@ export class ChatRecord {
question?: string,
sql_answer?: string,
sql?: string,
data?: string,
data?: string | any,
chart_answer?: string,
chart?: string,
analysis?: string,
analysis_thinking?: string,
predict?: string,
predict_data?: string,
predict_content?: string,
predict_data?: string | any,
finish?: boolean,
error?: string,
run_time?: number,
@@ -98,7 +104,9 @@ export class ChatRecord {
this.chart_answer = chart_answer
this.chart = chart
this.analysis = analysis
this.analysis_thinking = analysis_thinking
this.predict = predict
this.predict_content = predict_content
this.predict_data = predict_data
this.finish = !!finish
this.error = error
@@ -219,7 +227,9 @@ const toChatRecord = (data?: any): ChatRecord | undefined => {
data.chart_answer,
data.chart,
data.analysis,
data.analysis_thinking,
data.predict,
data.predict_content,
data.predict_data,
data.finish,
data.error,
@@ -273,6 +283,12 @@ export const chatApi = {
get: (id: number): Promise<ChatInfo> => {
return request.get(`/chat/get/${id}`)
},
get_chart_data: (record_id: number): Promise<any> => {
return request.get(`/chat/record/get/${record_id}/data`)
},
get_chart_predict_data: (record_id: number): Promise<any> => {
return request.get(`/chat/record/get/${record_id}/predict_data`)
},
startChat: (data: any): Promise<ChatInfo> => {
return request.post('/chat/start', data)
},
@@ -289,6 +305,6 @@ export const chatApi = {
return request.fetchStream(`/chat/record/${record_id}/predict`, {})
},
recommendQuestions: (record_id: number | undefined) => {
return request.get(`/chat/recommend_questions/${record_id}`)
return request.fetchStream(`/chat/recommend_questions/${record_id}`, {})
},
}
+1 -1
View File
@@ -38,7 +38,7 @@ export const supplierList: Array<{
model_config: {
0: {
api_domain: 'https://api.deepseek.com',
model_options: ['deepseek-chat'],
model_options: ['deepseek-chat', 'deepseek-reasoner'],
},
},
},
+5 -1
View File
@@ -49,7 +49,11 @@ const dataObject = computed<{
data: Array<{ [key: string]: any }>
}>(() => {
if (props.message?.record?.data) {
return JSON.parse(props.message.record.data)
if (typeof props.message?.record?.data === 'string') {
return JSON.parse(props.message.record.data)
} else {
return props.message.record.data
}
}
return {}
})
@@ -9,7 +9,7 @@ const props = defineProps<{
id?: number | string
chartType: ChartTypes
message: ChatMessage
data: string
data: string | any
}>()
const dataObject = computed<{
@@ -17,23 +17,32 @@ const dataObject = computed<{
data: Array<{ [key: string]: any }>
}>(() => {
if (props.message?.record?.data) {
return JSON.parse(props.message.record.data)
if (typeof props.message?.record?.data === 'string') {
return JSON.parse(props.message.record.data)
} else {
return props.message.record.data
}
}
return {}
})
const _data = computed(() => {
let _list = []
if (
props.data &&
props.data.length > 0 &&
props.data.trim().startsWith('[') &&
props.data.trim().endsWith(']')
) {
try {
_list = JSON.parse(props.data)
} catch (e) {
console.error(e)
if (props.data && typeof props.data === 'string') {
if (
props.data.length > 0 &&
props.data.trim().startsWith('[') &&
props.data.trim().endsWith(']')
) {
try {
_list = JSON.parse(props.data)
} catch (e) {
console.error(e)
}
}
} else {
if (props.data.length > 0) {
_list = props.data
}
}
+108 -10
View File
@@ -79,6 +79,7 @@
/>
<template v-if="message.record?.predict || isPredictTyping">
<MdComponent :message="message.record?.predict" />
<MdComponent :message="message.record?.predict_content" />
<PredictChartBlock
:id="message.record?.id + '-predict'"
:data="message.record?.predict_data ?? '[]'"
@@ -279,8 +280,8 @@ function onChatCreated(chat: ChatInfo) {
}
}
function getRecommendQuestions(record_id?: number) {
chatApi.recommendQuestions(record_id).then((res) => {
async function getRecommendQuestions(record_id?: number) {
/*chatApi.recommendQuestions(record_id).then((res) => {
if (res && res.length > 0 && startsWith(res.trim(), '[') && endsWith(res.trim(), ']')) {
if (currentChat.value?.records) {
for (let record of currentChat.value.records) {
@@ -290,7 +291,74 @@ function getRecommendQuestions(record_id?: number) {
}
}
}
})
})*/
const response = await chatApi.recommendQuestions(record_id)
const reader = response.body.getReader()
const decoder = new TextDecoder()
while (true) {
const { done, value } = await reader.read()
if (done) {
isTyping.value = false
break
}
const chunk = decoder.decode(value)
let _list = [chunk]
const lines = chunk.trim().split('}\n\n{')
if (lines.length > 1) {
_list = []
for (let line of lines) {
if (!line.trim().startsWith('{')) {
line = '{' + line.trim()
}
if (!line.trim().endsWith('}')) {
line = line.trim() + '}'
}
_list.push(line)
}
}
for (const str of _list) {
let data
try {
data = JSON.parse(str)
} catch (err) {
console.error('JSON string:', str)
throw err
}
if (data.code && data.code !== 200) {
ElMessage({
message: data.msg,
type: 'error',
showClose: true,
})
isTyping.value = false
return
}
switch (data.type) {
case 'recommended_question':
if (
data.content &&
data.content.length > 0 &&
startsWith(data.content.trim(), '[') &&
endsWith(data.content.trim(), ']')
) {
if (currentChat.value?.records) {
for (let record of currentChat.value.records) {
if (record.id === record_id) {
record.recommended_question = data.content
}
}
}
}
}
}
}
}
onMounted(() => {
@@ -398,17 +466,18 @@ const sendMessage = async () => {
isTyping.value = false
break
case 'sql-result':
sql_answer += data.content
sql_answer += data.reasoning_content
currentChat.value.records[currentChat.value.records.length - 1].sql_answer = sql_answer
break
case 'sql':
currentChat.value.records[currentChat.value.records.length - 1].sql = data.content
break
case 'sql-data':
currentChat.value.records[currentChat.value.records.length - 1].data = data.content
//currentChat.value.records[currentChat.value.records.length - 1].data = data.content
getChatData(currentChat.value.records[currentChat.value.records.length - 1].id)
break
case 'chart-result':
chart_answer += data.content
chart_answer += data.reasoning_content
currentChat.value.records[currentChat.value.records.length - 1].chart_answer =
chart_answer
break
@@ -417,7 +486,7 @@ const sendMessage = async () => {
break
case 'finish':
isTyping.value = false
getRecommendQuestions(currentRecord.id)
await getRecommendQuestions(currentRecord.id)
break
}
await nextTick()
@@ -438,6 +507,26 @@ const sendMessage = async () => {
}
}
function getChatData(recordId) {
chatApi.get_chart_data(recordId).then((response) => {
currentChat.value.records.forEach((record) => {
if (record.id === recordId) {
record.data = response
}
})
})
}
function getChatPredictData(recordId) {
chatApi.get_chart_predict_data(recordId).then((response) => {
currentChat.value.records.forEach((record) => {
if (record.id === recordId) {
record.predict_data = response
}
})
})
}
async function clickAnalysis(id?: number) {
let _index = -1
const currentRecord = find(currentChat.value.records, (value, index) => {
@@ -459,6 +548,7 @@ async function clickAnalysis(id?: number) {
const decoder = new TextDecoder()
let analysis_answer = ''
let analysis_answer_thinking = ''
while (true) {
const { done, value } = await reader.read()
@@ -514,7 +604,9 @@ async function clickAnalysis(id?: number) {
throw Error(data.content)
case 'analysis-result':
analysis_answer += data.content
analysis_answer_thinking += data.reasoning_content
currentChat.value.records[_index].analysis = analysis_answer
currentChat.value.records[_index].analysis_thinking = analysis_answer_thinking
break
case 'analysis_finish':
isAnalysisTyping.value = false
@@ -556,6 +648,7 @@ async function clickPredict(id?: number) {
const decoder = new TextDecoder()
let predict_answer = ''
let predict_content = ''
while (true) {
const { done, value } = await reader.read()
@@ -609,11 +702,16 @@ async function clickPredict(id?: number) {
case 'error':
throw Error(data.content)
case 'predict-result':
predict_answer += data.content
predict_answer += data.reasoning_content
predict_content += data.content
currentChat.value.records[_index].predict = predict_answer
currentChat.value.records[_index].predict_content = predict_content
break
case 'predict':
currentChat.value.records[_index].predict_data = data.content
case 'predict-failed':
break
case 'predict-success':
//currentChat.value.records[_index].predict_data = data.content
getChatPredictData(currentChat.value.records[_index].id)
break
case 'predict_finish':
isPredictTyping.value = false