(self, session: Session)
| 221 | return True |
| 222 | |
| 223 | def init_messages(self, session: Session): |
| 224 | |
| 225 | self.choose_table_schema(session) |
| 226 | |
| 227 | last_sql_messages: List[dict[str, Any]] = self.generate_sql_logs[-1].messages if len( |
| 228 | self.generate_sql_logs) > 0 else [] |
| 229 | if self.chat_question.regenerate_record_id: |
| 230 | # filter record before regenerate_record_id |
| 231 | _temp_log = next( |
| 232 | filter(lambda obj: obj.pid == self.chat_question.regenerate_record_id, self.generate_sql_logs), None) |
| 233 | last_sql_messages: List[dict[str, Any]] = _temp_log.messages if _temp_log else [] |
| 234 | |
| 235 | # 排除所有的系统提示词 |
| 236 | last_sql_messages = [obj for obj in last_sql_messages if obj.get("sqlbot_system") != True] |
| 237 | |
| 238 | count_limit = self.base_message_round_count_limit |
| 239 | |
| 240 | self.sql_message = [] |
| 241 | # add sys prompt |
| 242 | _system_templates = self.chat_question.sql_sys_question(self.ds.type, self.enable_sql_row_limit) |
| 243 | self.sql_message.append(SystemPromptMessage(content=_system_templates['system'])) |
| 244 | self.sql_message.append(HumanPromptMessage(content=_system_templates['rules'])) |
| 245 | self.sql_message.append( |
| 246 | AIPromptMessage(content='我已掌握所有规则，包括表结构、SQL规范、安全限制和输出格式，我会严格遵守这些规则。')) |
| 247 | self.sql_message.append(HumanPromptMessage(content=_system_templates['schema'])) |
| 248 | self.sql_message.append( |
| 249 | AIPromptMessage(content='我已确认您提供的数据库信息与表结构schema，我生成的SQL不会超出您提供的范围。')) |
| 250 | if _system_templates.get('custom_prompt'): |
| 251 | self.sql_message.append(HumanPromptMessage(content=_system_templates['custom_prompt'])) |
| 252 | self.sql_message.append(AIPromptMessage(content='我已确认您提供的额外信息，我会进行参考。')) |
| 253 | if _system_templates.get('terminologies'): |
| 254 | self.sql_message.append(HumanPromptMessage(content=_system_templates['terminologies'])) |
| 255 | self.sql_message.append(AIPromptMessage(content='我已确认您提供的术语信息，我会进行参考。')) |
| 256 | if _system_templates.get('data_training'): |
| 257 | self.sql_message.append(HumanPromptMessage(content=_system_templates['data_training'])) |
| 258 | self.sql_message.append(AIPromptMessage(content='我已确认您提供的SQL示例，我会进行参考。')) |
| 259 | |
| 260 | if last_sql_messages is not None and len(last_sql_messages) > 0: |
| 261 | last_rounds = get_last_conversation_rounds(last_sql_messages, rounds=count_limit) |
| 262 | |
| 263 | for _msg_dict in last_rounds: |
| 264 | _msg: BaseMessage |
| 265 | if _msg_dict.get('type') == 'human': |
| 266 | _msg = HumanMessage(content=_msg_dict.get('content')) |
| 267 | self.sql_message.append(_msg) |
| 268 | elif _msg_dict.get('type') == 'ai': |
| 269 | _msg = AIMessage(content=_msg_dict.get('content')) |
| 270 | self.sql_message.append(_msg) |
| 271 | |
| 272 | last_chart_messages: List[dict[str, Any]] = self.generate_chart_logs[-1].messages if len( |
| 273 | self.generate_chart_logs) > 0 else [] |
| 274 | if self.chat_question.regenerate_record_id: |
| 275 | # filter record before regenerate_record_id |
| 276 | _temp_log = next( |
| 277 | filter(lambda obj: obj.pid == self.chat_question.regenerate_record_id, self.generate_chart_logs), None) |
| 278 | last_chart_messages: List[dict[str, Any]] = _temp_log.messages if _temp_log else [] |
| 279 | |
| 280 | # 排除所有的系统提示词 |
no test coverage detected