citu_app.py 47 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138
  1. # 给dataops 对话助手返回结果
  2. from vanna.flask import VannaFlaskApp
  3. from core.vanna_llm_factory import create_vanna_instance
  4. from flask import request, jsonify
  5. import pandas as pd
  6. import common.result as result
  7. from datetime import datetime, timedelta
  8. from common.session_aware_cache import WebSessionAwareMemoryCache
  9. from app_config import API_MAX_RETURN_ROWS, DISPLAY_SUMMARY_THINKING
  10. import re
  11. # 设置默认的最大返回行数
  12. DEFAULT_MAX_RETURN_ROWS = 200
  13. MAX_RETURN_ROWS = API_MAX_RETURN_ROWS if API_MAX_RETURN_ROWS is not None else DEFAULT_MAX_RETURN_ROWS
  14. vn = create_vanna_instance()
  15. # 创建带时间戳的缓存
  16. timestamped_cache = WebSessionAwareMemoryCache()
  17. # 实例化 VannaFlaskApp,使用自定义缓存
  18. app = VannaFlaskApp(
  19. vn,
  20. cache=timestamped_cache, # 使用带时间戳的缓存
  21. title="辞图智能数据问答平台",
  22. logo = "https://www.citupro.com/img/logo-black-2.png",
  23. subtitle="让 AI 为你写 SQL",
  24. chart=False,
  25. allow_llm_to_see_data=True,
  26. ask_results_correct=True,
  27. followup_questions=True,
  28. debug=True
  29. )
  30. def _remove_thinking_content(text: str) -> str:
  31. """
  32. 移除文本中的 <think></think> 标签及其内容
  33. 复用自 base_llm_chat.py 中的同名方法
  34. Args:
  35. text (str): 包含可能的 thinking 标签的文本
  36. Returns:
  37. str: 移除 thinking 内容后的文本
  38. """
  39. if not text:
  40. return text
  41. # 移除 <think>...</think> 标签及其内容(支持多行)
  42. # 使用 re.DOTALL 标志使 . 匹配包括换行符在内的任何字符
  43. cleaned_text = re.sub(r'<think>.*?</think>\s*', '', text, flags=re.DOTALL | re.IGNORECASE)
  44. # 移除可能的多余空行
  45. cleaned_text = re.sub(r'\n\s*\n\s*\n', '\n\n', cleaned_text)
  46. # 去除开头和结尾的空白字符
  47. cleaned_text = cleaned_text.strip()
  48. return cleaned_text
  49. # 修改ask接口,支持前端传递session_id
  50. @app.flask_app.route('/api/v0/ask', methods=['POST'])
  51. def ask_full():
  52. req = request.get_json(force=True)
  53. question = req.get("question", None)
  54. browser_session_id = req.get("session_id", None) # 前端传递的会话ID
  55. if not question:
  56. return jsonify(result.failed(message="未提供问题", code=400)), 400
  57. # 如果使用WebSessionAwareMemoryCache
  58. if hasattr(app.cache, 'generate_id_with_browser_session') and browser_session_id:
  59. # 这里需要修改vanna的ask方法来支持传递session_id
  60. # 或者预先调用generate_id来建立会话关联
  61. conversation_id = app.cache.generate_id_with_browser_session(
  62. question=question,
  63. browser_session_id=browser_session_id
  64. )
  65. try:
  66. sql, df, _ = vn.ask(
  67. question=question,
  68. print_results=False,
  69. visualize=False,
  70. allow_llm_to_see_data=True
  71. )
  72. # 关键:检查是否有LLM解释性文本(无法生成SQL的情况)
  73. if sql is None and hasattr(vn, 'last_llm_explanation') and vn.last_llm_explanation:
  74. # 根据 DISPLAY_SUMMARY_THINKING 参数决定是否移除 thinking 内容
  75. explanation_message = vn.last_llm_explanation
  76. if not DISPLAY_SUMMARY_THINKING:
  77. explanation_message = _remove_thinking_content(explanation_message)
  78. print(f"[DEBUG] 隐藏thinking内容 - 原始长度: {len(vn.last_llm_explanation)}, 处理后长度: {len(explanation_message)}")
  79. # 在解释性文本末尾添加提示语
  80. explanation_message = explanation_message + "请尝试提问其它问题。"
  81. # 使用 result.failed 返回,success为false,但在message中包含LLM友好的解释
  82. return jsonify(result.failed(
  83. message=explanation_message, # 处理后的解释性文本
  84. code=400, # 业务逻辑错误,使用400
  85. data={
  86. "sql": None,
  87. "rows": [],
  88. "columns": [],
  89. "summary": None,
  90. "conversation_id": conversation_id if 'conversation_id' in locals() else None,
  91. "session_id": browser_session_id
  92. }
  93. )), 200 # HTTP状态码仍为200,因为请求本身成功处理了
  94. # 如果sql为None但没有解释性文本,返回通用错误
  95. if sql is None:
  96. return jsonify(result.failed(
  97. message="无法生成SQL查询,请检查问题描述或数据表结构",
  98. code=400,
  99. data={
  100. "sql": None,
  101. "rows": [],
  102. "columns": [],
  103. "summary": None,
  104. "conversation_id": conversation_id if 'conversation_id' in locals() else None,
  105. "session_id": browser_session_id
  106. }
  107. )), 200
  108. # 正常SQL流程
  109. rows, columns = [], []
  110. summary = None
  111. if isinstance(df, pd.DataFrame) and not df.empty:
  112. rows = df.head(MAX_RETURN_ROWS).to_dict(orient="records")
  113. columns = list(df.columns)
  114. # 生成数据摘要
  115. try:
  116. summary = vn.generate_summary(question=question, df=df)
  117. print(f"[INFO] 成功生成摘要: {summary}")
  118. except Exception as e:
  119. print(f"[WARNING] 生成摘要失败: {str(e)}")
  120. summary = None
  121. return jsonify(result.success(data={
  122. "sql": sql,
  123. "rows": rows,
  124. "columns": columns,
  125. "summary": summary, # 添加摘要到返回结果
  126. "conversation_id": conversation_id if 'conversation_id' in locals() else None,
  127. "session_id": browser_session_id
  128. }))
  129. except Exception as e:
  130. print(f"[ERROR] ask_full执行失败: {str(e)}")
  131. # 即使发生异常,也检查是否有业务层面的解释
  132. if hasattr(vn, 'last_llm_explanation') and vn.last_llm_explanation:
  133. # 根据 DISPLAY_SUMMARY_THINKING 参数决定是否移除 thinking 内容
  134. explanation_message = vn.last_llm_explanation
  135. if not DISPLAY_SUMMARY_THINKING:
  136. explanation_message = _remove_thinking_content(explanation_message)
  137. print(f"[DEBUG] 异常处理中隐藏thinking内容 - 原始长度: {len(vn.last_llm_explanation)}, 处理后长度: {len(explanation_message)}")
  138. # 在解释性文本末尾添加提示语
  139. explanation_message = explanation_message + "请尝试提问其它问题。"
  140. return jsonify(result.failed(
  141. message=explanation_message,
  142. code=400,
  143. data={
  144. "sql": None,
  145. "rows": [],
  146. "columns": [],
  147. "summary": None,
  148. "conversation_id": conversation_id if 'conversation_id' in locals() else None,
  149. "session_id": browser_session_id
  150. }
  151. )), 200
  152. else:
  153. # 技术错误,使用500错误码
  154. return jsonify(result.failed(
  155. message=f"查询处理失败: {str(e)}",
  156. code=500
  157. )), 500
  158. @app.flask_app.route('/api/v0/citu_run_sql', methods=['POST'])
  159. def citu_run_sql():
  160. req = request.get_json(force=True)
  161. sql = req.get('sql')
  162. if not sql:
  163. return jsonify(result.failed(message="未提供SQL查询", code=400)), 400
  164. try:
  165. df = vn.run_sql(sql)
  166. rows, columns = [], []
  167. if isinstance(df, pd.DataFrame) and not df.empty:
  168. rows = df.head(MAX_RETURN_ROWS).to_dict(orient="records")
  169. columns = list(df.columns)
  170. return jsonify(result.success(data={
  171. "sql": sql,
  172. "rows": rows,
  173. "columns": columns
  174. }))
  175. except Exception as e:
  176. print(f"[ERROR] citu_run_sql执行失败: {str(e)}")
  177. return jsonify(result.failed(
  178. message=f"SQL执行失败: {str(e)}",
  179. code=500
  180. )), 500
  181. @app.flask_app.route('/api/v0/ask_cached', methods=['POST'])
  182. def ask_cached():
  183. """
  184. 带缓存功能的智能查询接口
  185. 支持会话管理和结果缓存,提高查询效率
  186. """
  187. req = request.get_json(force=True)
  188. question = req.get("question", None)
  189. browser_session_id = req.get("session_id", None)
  190. if not question:
  191. return jsonify(result.failed(message="未提供问题", code=400)), 400
  192. try:
  193. # 生成conversation_id
  194. # 调试:查看generate_id的实际行为
  195. print(f"[DEBUG] 输入问题: '{question}'")
  196. conversation_id = app.cache.generate_id(question=question)
  197. print(f"[DEBUG] 生成的conversation_id: {conversation_id}")
  198. # 再次用相同问题测试
  199. conversation_id2 = app.cache.generate_id(question=question)
  200. print(f"[DEBUG] 再次生成的conversation_id: {conversation_id2}")
  201. print(f"[DEBUG] 两次ID是否相同: {conversation_id == conversation_id2}")
  202. # 检查缓存
  203. cached_sql = app.cache.get(id=conversation_id, field="sql")
  204. if cached_sql is not None:
  205. # 缓存命中
  206. print(f"[CACHE HIT] 使用缓存结果: {conversation_id}")
  207. sql = cached_sql
  208. df = app.cache.get(id=conversation_id, field="df")
  209. summary = app.cache.get(id=conversation_id, field="summary")
  210. else:
  211. # 缓存未命中,执行新查询
  212. print(f"[CACHE MISS] 执行新查询: {conversation_id}")
  213. sql, df, _ = vn.ask(
  214. question=question,
  215. print_results=False,
  216. visualize=False,
  217. allow_llm_to_see_data=True
  218. )
  219. # 检查是否有LLM解释性文本(无法生成SQL的情况)
  220. if sql is None and hasattr(vn, 'last_llm_explanation') and vn.last_llm_explanation:
  221. # 根据 DISPLAY_SUMMARY_THINKING 参数决定是否移除 thinking 内容
  222. explanation_message = vn.last_llm_explanation
  223. if not DISPLAY_SUMMARY_THINKING:
  224. explanation_message = _remove_thinking_content(explanation_message)
  225. print(f"[DEBUG] ask_cached中隐藏thinking内容 - 原始长度: {len(vn.last_llm_explanation)}, 处理后长度: {len(explanation_message)}")
  226. # 在解释性文本末尾添加提示语
  227. explanation_message = explanation_message + "请尝试用其它方式提问。"
  228. return jsonify(result.failed(
  229. message=explanation_message,
  230. code=400,
  231. data={
  232. "sql": None,
  233. "rows": [],
  234. "columns": [],
  235. "summary": None,
  236. "conversation_id": conversation_id,
  237. "session_id": browser_session_id,
  238. "cached": False
  239. }
  240. )), 200
  241. # 如果sql为None但没有解释性文本,返回通用错误
  242. if sql is None:
  243. return jsonify(result.failed(
  244. message="无法生成SQL查询,请检查问题描述或数据表结构",
  245. code=400,
  246. data={
  247. "sql": None,
  248. "rows": [],
  249. "columns": [],
  250. "summary": None,
  251. "conversation_id": conversation_id,
  252. "session_id": browser_session_id,
  253. "cached": False
  254. }
  255. )), 200
  256. # 缓存结果
  257. app.cache.set(id=conversation_id, field="question", value=question)
  258. app.cache.set(id=conversation_id, field="sql", value=sql)
  259. app.cache.set(id=conversation_id, field="df", value=df)
  260. # 生成并缓存摘要
  261. summary = None
  262. if isinstance(df, pd.DataFrame) and not df.empty:
  263. try:
  264. summary = vn.generate_summary(question=question, df=df)
  265. print(f"[INFO] 成功生成摘要: {summary}")
  266. except Exception as e:
  267. print(f"[WARNING] 生成摘要失败: {str(e)}")
  268. summary = None
  269. app.cache.set(id=conversation_id, field="summary", value=summary)
  270. # 处理返回数据
  271. rows, columns = [], []
  272. if isinstance(df, pd.DataFrame) and not df.empty:
  273. rows = df.head(MAX_RETURN_ROWS).to_dict(orient="records")
  274. columns = list(df.columns)
  275. return jsonify(result.success(data={
  276. "sql": sql,
  277. "rows": rows,
  278. "columns": columns,
  279. "summary": summary,
  280. "conversation_id": conversation_id,
  281. "session_id": browser_session_id,
  282. "cached": cached_sql is not None # 标识是否来自缓存
  283. }))
  284. except Exception as e:
  285. print(f"[ERROR] ask_cached执行失败: {str(e)}")
  286. return jsonify(result.failed(
  287. message=f"查询处理失败: {str(e)}",
  288. code=500
  289. )), 500
  290. @app.flask_app.route('/api/v0/citu_train_question_sql', methods=['POST'])
  291. def citu_train_question_sql():
  292. """
  293. 训练问题-SQL对接口
  294. 此API将接收的question/sql pair写入到training库中,用于训练和改进AI模型。
  295. 支持仅传入SQL或同时传入问题和SQL进行训练。
  296. Args:
  297. question (str, optional): 用户问题
  298. sql (str, required): 对应的SQL查询语句
  299. Returns:
  300. JSON: 包含训练ID和成功消息的响应
  301. """
  302. try:
  303. req = request.get_json(force=True)
  304. question = req.get('question')
  305. sql = req.get('sql')
  306. if not sql:
  307. return jsonify(result.failed(
  308. message="'sql' are required",
  309. code=400
  310. )), 400
  311. # 正确的调用方式:同时传递question和sql
  312. if question:
  313. training_id = vn.train(question=question, sql=sql)
  314. print(f"训练成功,训练ID为:{training_id},问题:{question},SQL:{sql}")
  315. else:
  316. training_id = vn.train(sql=sql)
  317. print(f"训练成功,训练ID为:{training_id},SQL:{sql}")
  318. return jsonify(result.success(data={
  319. "training_id": training_id,
  320. "message": "Question-SQL pair trained successfully"
  321. }))
  322. except Exception as e:
  323. return jsonify(result.failed(
  324. message=f"Training failed: {str(e)}",
  325. code=500
  326. )), 500
  327. # ==================== 日常管理API ====================
  328. @app.flask_app.route('/api/v0/cache_overview', methods=['GET'])
  329. def cache_overview():
  330. """日常管理:轻量概览 - 合并原cache_inspect的核心功能"""
  331. try:
  332. cache = app.cache
  333. result_data = {
  334. 'overview_summary': {
  335. 'total_conversations': 0,
  336. 'total_sessions': 0,
  337. 'query_time': datetime.now().isoformat()
  338. },
  339. 'recent_conversations': [], # 最近的对话
  340. 'session_summary': [] # 会话摘要
  341. }
  342. if hasattr(cache, 'cache') and isinstance(cache.cache, dict):
  343. result_data['overview_summary']['total_conversations'] = len(cache.cache)
  344. # 获取会话信息
  345. if hasattr(cache, 'get_all_sessions'):
  346. all_sessions = cache.get_all_sessions()
  347. result_data['overview_summary']['total_sessions'] = len(all_sessions)
  348. # 会话摘要(按最近活动排序)
  349. session_list = []
  350. for session_id, session_data in all_sessions.items():
  351. session_summary = {
  352. 'session_id': session_id,
  353. 'start_time': session_data['start_time'].isoformat(),
  354. 'conversation_count': session_data.get('conversation_count', 0),
  355. 'duration_seconds': session_data.get('session_duration_seconds', 0),
  356. 'last_activity': session_data.get('last_activity', session_data['start_time']).isoformat(),
  357. 'is_active': (datetime.now() - session_data.get('last_activity', session_data['start_time'])).total_seconds() < 1800 # 30分钟内活跃
  358. }
  359. session_list.append(session_summary)
  360. # 按最后活动时间排序
  361. session_list.sort(key=lambda x: x['last_activity'], reverse=True)
  362. result_data['session_summary'] = session_list
  363. # 最近的对话(最多显示10个)
  364. conversation_list = []
  365. for conversation_id, conversation_data in cache.cache.items():
  366. conversation_start_time = cache.conversation_start_times.get(conversation_id)
  367. conversation_info = {
  368. 'conversation_id': conversation_id,
  369. 'conversation_start_time': conversation_start_time.isoformat() if conversation_start_time else None,
  370. 'session_id': cache.conversation_to_session.get(conversation_id),
  371. 'has_question': 'question' in conversation_data,
  372. 'has_sql': 'sql' in conversation_data,
  373. 'has_data': 'df' in conversation_data and conversation_data['df'] is not None,
  374. 'question_preview': conversation_data.get('question', '')[:80] + '...' if len(conversation_data.get('question', '')) > 80 else conversation_data.get('question', ''),
  375. }
  376. # 计算对话持续时间
  377. if conversation_start_time:
  378. duration = datetime.now() - conversation_start_time
  379. conversation_info['conversation_duration_seconds'] = duration.total_seconds()
  380. conversation_list.append(conversation_info)
  381. # 按对话开始时间排序,显示最新的10个
  382. conversation_list.sort(key=lambda x: x['conversation_start_time'] or '', reverse=True)
  383. result_data['recent_conversations'] = conversation_list[:10]
  384. return jsonify(result.success(data=result_data))
  385. except Exception as e:
  386. return jsonify(result.failed(
  387. message=f"获取缓存概览失败: {str(e)}",
  388. code=500
  389. )), 500
  390. @app.flask_app.route('/api/v0/cache_stats', methods=['GET'])
  391. def cache_stats():
  392. """日常管理:统计信息 - 合并原session_stats和cache_stats功能"""
  393. try:
  394. cache = app.cache
  395. current_time = datetime.now()
  396. stats = {
  397. 'basic_stats': {
  398. 'total_sessions': len(getattr(cache, 'session_info', {})),
  399. 'total_conversations': len(getattr(cache, 'cache', {})),
  400. 'active_sessions': 0, # 最近30分钟有活动
  401. 'average_conversations_per_session': 0
  402. },
  403. 'time_distribution': {
  404. 'sessions': {
  405. 'last_1_hour': 0,
  406. 'last_6_hours': 0,
  407. 'last_24_hours': 0,
  408. 'last_7_days': 0,
  409. 'older': 0
  410. },
  411. 'conversations': {
  412. 'last_1_hour': 0,
  413. 'last_6_hours': 0,
  414. 'last_24_hours': 0,
  415. 'last_7_days': 0,
  416. 'older': 0
  417. }
  418. },
  419. 'session_details': [],
  420. 'time_ranges': {
  421. 'oldest_session': None,
  422. 'newest_session': None,
  423. 'oldest_conversation': None,
  424. 'newest_conversation': None
  425. }
  426. }
  427. # 会话统计
  428. if hasattr(cache, 'session_info'):
  429. session_times = []
  430. total_conversations = 0
  431. for session_id, session_data in cache.session_info.items():
  432. start_time = session_data['start_time']
  433. session_times.append(start_time)
  434. conversation_count = len(session_data.get('conversations', []))
  435. total_conversations += conversation_count
  436. # 检查活跃状态
  437. last_activity = session_data.get('last_activity', session_data['start_time'])
  438. if (current_time - last_activity).total_seconds() < 1800:
  439. stats['basic_stats']['active_sessions'] += 1
  440. # 时间分布统计
  441. age_hours = (current_time - start_time).total_seconds() / 3600
  442. if age_hours <= 1:
  443. stats['time_distribution']['sessions']['last_1_hour'] += 1
  444. elif age_hours <= 6:
  445. stats['time_distribution']['sessions']['last_6_hours'] += 1
  446. elif age_hours <= 24:
  447. stats['time_distribution']['sessions']['last_24_hours'] += 1
  448. elif age_hours <= 168: # 7 days
  449. stats['time_distribution']['sessions']['last_7_days'] += 1
  450. else:
  451. stats['time_distribution']['sessions']['older'] += 1
  452. # 会话详细信息
  453. session_duration = current_time - start_time
  454. stats['session_details'].append({
  455. 'session_id': session_id,
  456. 'start_time': start_time.isoformat(),
  457. 'last_activity': last_activity.isoformat(),
  458. 'conversation_count': conversation_count,
  459. 'duration_seconds': session_duration.total_seconds(),
  460. 'duration_formatted': str(session_duration),
  461. 'is_active': (current_time - last_activity).total_seconds() < 1800,
  462. 'browser_session_id': session_data.get('browser_session_id')
  463. })
  464. # 计算平均值
  465. if len(cache.session_info) > 0:
  466. stats['basic_stats']['average_conversations_per_session'] = total_conversations / len(cache.session_info)
  467. # 时间范围
  468. if session_times:
  469. stats['time_ranges']['oldest_session'] = min(session_times).isoformat()
  470. stats['time_ranges']['newest_session'] = max(session_times).isoformat()
  471. # 对话统计
  472. if hasattr(cache, 'conversation_start_times'):
  473. conversation_times = []
  474. for conv_time in cache.conversation_start_times.values():
  475. conversation_times.append(conv_time)
  476. age_hours = (current_time - conv_time).total_seconds() / 3600
  477. if age_hours <= 1:
  478. stats['time_distribution']['conversations']['last_1_hour'] += 1
  479. elif age_hours <= 6:
  480. stats['time_distribution']['conversations']['last_6_hours'] += 1
  481. elif age_hours <= 24:
  482. stats['time_distribution']['conversations']['last_24_hours'] += 1
  483. elif age_hours <= 168:
  484. stats['time_distribution']['conversations']['last_7_days'] += 1
  485. else:
  486. stats['time_distribution']['conversations']['older'] += 1
  487. if conversation_times:
  488. stats['time_ranges']['oldest_conversation'] = min(conversation_times).isoformat()
  489. stats['time_ranges']['newest_conversation'] = max(conversation_times).isoformat()
  490. # 按最近活动排序会话详情
  491. stats['session_details'].sort(key=lambda x: x['last_activity'], reverse=True)
  492. return jsonify(result.success(data=stats))
  493. except Exception as e:
  494. return jsonify(result.failed(
  495. message=f"获取缓存统计失败: {str(e)}",
  496. code=500
  497. )), 500
  498. # ==================== 高级功能API ====================
  499. @app.flask_app.route('/api/v0/cache_export', methods=['GET'])
  500. def cache_export():
  501. """高级功能:完整导出 - 保持原cache_raw_export的完整功能"""
  502. try:
  503. cache = app.cache
  504. # 验证缓存的实际结构
  505. if not hasattr(cache, 'cache'):
  506. return jsonify(result.failed(message="缓存对象没有cache属性", code=500)), 500
  507. if not isinstance(cache.cache, dict):
  508. return jsonify(result.failed(message="缓存不是字典类型", code=500)), 500
  509. # 定义JSON序列化辅助函数
  510. def make_json_serializable(obj):
  511. """将对象转换为JSON可序列化的格式"""
  512. if obj is None:
  513. return None
  514. elif isinstance(obj, (str, int, float, bool)):
  515. return obj
  516. elif isinstance(obj, (list, tuple)):
  517. return [make_json_serializable(item) for item in obj]
  518. elif isinstance(obj, dict):
  519. return {str(k): make_json_serializable(v) for k, v in obj.items()}
  520. elif hasattr(obj, 'isoformat'): # datetime objects
  521. return obj.isoformat()
  522. elif hasattr(obj, 'item'): # numpy scalars
  523. return obj.item()
  524. elif hasattr(obj, 'tolist'): # numpy arrays
  525. return obj.tolist()
  526. elif hasattr(obj, '__dict__'): # pandas dtypes and other objects
  527. return str(obj)
  528. else:
  529. return str(obj)
  530. # 获取完整的原始缓存数据
  531. raw_cache = cache.cache
  532. # 获取会话和对话时间信息
  533. conversation_times = getattr(cache, 'conversation_start_times', {})
  534. session_info = getattr(cache, 'session_info', {})
  535. conversation_to_session = getattr(cache, 'conversation_to_session', {})
  536. export_data = {
  537. 'export_metadata': {
  538. 'export_time': datetime.now().isoformat(),
  539. 'total_conversations': len(raw_cache),
  540. 'total_sessions': len(session_info),
  541. 'cache_type': type(cache).__name__,
  542. 'cache_object_info': str(cache),
  543. 'has_session_times': bool(session_info),
  544. 'has_conversation_times': bool(conversation_times)
  545. },
  546. 'session_info': {
  547. session_id: {
  548. 'start_time': session_data['start_time'].isoformat(),
  549. 'last_activity': session_data.get('last_activity', session_data['start_time']).isoformat(),
  550. 'conversations': session_data['conversations'],
  551. 'conversation_count': len(session_data['conversations']),
  552. 'browser_session_id': session_data.get('browser_session_id'),
  553. 'user_info': session_data.get('user_info', {})
  554. }
  555. for session_id, session_data in session_info.items()
  556. },
  557. 'conversation_times': {
  558. conversation_id: start_time.isoformat()
  559. for conversation_id, start_time in conversation_times.items()
  560. },
  561. 'conversation_to_session_mapping': conversation_to_session,
  562. 'conversations': {}
  563. }
  564. # 处理每个对话的完整数据
  565. for conversation_id, conversation_data in raw_cache.items():
  566. # 获取时间信息
  567. conversation_start_time = conversation_times.get(conversation_id)
  568. session_id = conversation_to_session.get(conversation_id)
  569. session_start_time = None
  570. if session_id and session_id in session_info:
  571. session_start_time = session_info[session_id]['start_time']
  572. processed_conversation = {
  573. 'conversation_id': conversation_id,
  574. 'conversation_start_time': conversation_start_time.isoformat() if conversation_start_time else None,
  575. 'session_id': session_id,
  576. 'session_start_time': session_start_time.isoformat() if session_start_time else None,
  577. 'field_count': len(conversation_data),
  578. 'fields': {}
  579. }
  580. # 添加时间计算
  581. if conversation_start_time:
  582. conversation_duration = datetime.now() - conversation_start_time
  583. processed_conversation['conversation_duration_seconds'] = conversation_duration.total_seconds()
  584. processed_conversation['conversation_duration_formatted'] = str(conversation_duration)
  585. if session_start_time:
  586. session_duration = datetime.now() - session_start_time
  587. processed_conversation['session_duration_seconds'] = session_duration.total_seconds()
  588. processed_conversation['session_duration_formatted'] = str(session_duration)
  589. # 处理每个字段,确保JSON序列化安全
  590. for field_name, field_value in conversation_data.items():
  591. field_info = {
  592. 'field_name': field_name,
  593. 'data_type': type(field_value).__name__,
  594. 'is_none': field_value is None
  595. }
  596. try:
  597. if field_value is None:
  598. field_info['value'] = None
  599. elif field_name in ['conversation_start_time', 'session_start_time']:
  600. # 处理时间字段
  601. field_info['content'] = make_json_serializable(field_value)
  602. elif field_name == 'df' and field_value is not None:
  603. # DataFrame的安全处理
  604. if hasattr(field_value, 'to_dict'):
  605. # 安全地处理dtypes
  606. try:
  607. dtypes_dict = {}
  608. for col, dtype in field_value.dtypes.items():
  609. dtypes_dict[col] = str(dtype)
  610. except Exception:
  611. dtypes_dict = {"error": "无法序列化dtypes"}
  612. # 安全地处理内存使用
  613. try:
  614. memory_usage = field_value.memory_usage(deep=True)
  615. memory_dict = {}
  616. for idx, usage in memory_usage.items():
  617. memory_dict[str(idx)] = int(usage) if hasattr(usage, 'item') else int(usage)
  618. except Exception:
  619. memory_dict = {"error": "无法获取内存使用信息"}
  620. field_info.update({
  621. 'dataframe_info': {
  622. 'shape': list(field_value.shape),
  623. 'columns': list(field_value.columns),
  624. 'dtypes': dtypes_dict,
  625. 'index_info': {
  626. 'type': type(field_value.index).__name__,
  627. 'length': len(field_value.index)
  628. }
  629. },
  630. 'data': make_json_serializable(field_value.to_dict('records')),
  631. 'memory_usage': memory_dict
  632. })
  633. else:
  634. field_info['value'] = str(field_value)
  635. field_info['note'] = 'not_standard_dataframe'
  636. elif field_name == 'fig_json':
  637. # 图表JSON数据处理
  638. if isinstance(field_value, str):
  639. try:
  640. import json
  641. parsed_fig = json.loads(field_value)
  642. field_info.update({
  643. 'json_valid': True,
  644. 'json_size_bytes': len(field_value),
  645. 'plotly_structure': {
  646. 'has_data': 'data' in parsed_fig,
  647. 'has_layout': 'layout' in parsed_fig,
  648. 'data_traces_count': len(parsed_fig.get('data', [])),
  649. },
  650. 'raw_json': field_value
  651. })
  652. except json.JSONDecodeError:
  653. field_info.update({
  654. 'json_valid': False,
  655. 'raw_content': str(field_value)
  656. })
  657. else:
  658. field_info['value'] = make_json_serializable(field_value)
  659. elif field_name == 'followup_questions':
  660. # 后续问题列表
  661. field_info.update({
  662. 'content': make_json_serializable(field_value)
  663. })
  664. elif field_name in ['question', 'sql', 'summary']:
  665. # 文本字段
  666. if isinstance(field_value, str):
  667. field_info.update({
  668. 'text_length': len(field_value),
  669. 'content': field_value
  670. })
  671. else:
  672. field_info['value'] = make_json_serializable(field_value)
  673. else:
  674. # 未知字段的安全处理
  675. field_info['content'] = make_json_serializable(field_value)
  676. except Exception as e:
  677. field_info.update({
  678. 'processing_error': str(e),
  679. 'fallback_value': str(field_value)[:500] + '...' if len(str(field_value)) > 500 else str(field_value)
  680. })
  681. processed_conversation['fields'][field_name] = field_info
  682. export_data['conversations'][conversation_id] = processed_conversation
  683. # 添加缓存统计信息
  684. field_frequency = {}
  685. data_types_found = set()
  686. total_dataframes = 0
  687. total_questions = 0
  688. for conv_data in export_data['conversations'].values():
  689. for field_name, field_info in conv_data['fields'].items():
  690. field_frequency[field_name] = field_frequency.get(field_name, 0) + 1
  691. data_types_found.add(field_info['data_type'])
  692. if field_name == 'df' and not field_info['is_none']:
  693. total_dataframes += 1
  694. if field_name == 'question' and not field_info['is_none']:
  695. total_questions += 1
  696. export_data['cache_statistics'] = {
  697. 'field_frequency': field_frequency,
  698. 'data_types_found': list(data_types_found),
  699. 'total_dataframes': total_dataframes,
  700. 'total_questions': total_questions,
  701. 'has_session_timing': 'session_start_time' in field_frequency,
  702. 'has_conversation_timing': 'conversation_start_time' in field_frequency
  703. }
  704. return jsonify(result.success(data=export_data))
  705. except Exception as e:
  706. import traceback
  707. error_details = {
  708. 'error_message': str(e),
  709. 'error_type': type(e).__name__,
  710. 'traceback': traceback.format_exc()
  711. }
  712. return jsonify(result.failed(
  713. message=f"导出缓存失败: {str(e)}",
  714. code=500,
  715. data=error_details
  716. )), 500
  717. # ==================== 清理功能API ====================
  718. @app.flask_app.route('/api/v0/cache_preview_cleanup', methods=['POST'])
  719. def cache_preview_cleanup():
  720. """清理功能:预览删除操作 - 保持原功能"""
  721. try:
  722. req = request.get_json(force=True)
  723. # 时间条件 - 支持三种方式
  724. older_than_hours = req.get('older_than_hours')
  725. older_than_days = req.get('older_than_days')
  726. before_timestamp = req.get('before_timestamp') # YYYY-MM-DD HH:MM:SS 格式
  727. cache = app.cache
  728. # 计算截止时间
  729. cutoff_time = None
  730. time_condition = None
  731. if older_than_hours:
  732. cutoff_time = datetime.now() - timedelta(hours=older_than_hours)
  733. time_condition = f"older_than_hours: {older_than_hours}"
  734. elif older_than_days:
  735. cutoff_time = datetime.now() - timedelta(days=older_than_days)
  736. time_condition = f"older_than_days: {older_than_days}"
  737. elif before_timestamp:
  738. try:
  739. # 支持 YYYY-MM-DD HH:MM:SS 格式
  740. cutoff_time = datetime.strptime(before_timestamp, '%Y-%m-%d %H:%M:%S')
  741. time_condition = f"before_timestamp: {before_timestamp}"
  742. except ValueError:
  743. return jsonify(result.failed(
  744. message="before_timestamp格式错误,请使用 YYYY-MM-DD HH:MM:SS 格式",
  745. code=400
  746. )), 400
  747. else:
  748. return jsonify(result.failed(
  749. message="必须提供时间条件:older_than_hours, older_than_days 或 before_timestamp (YYYY-MM-DD HH:MM:SS)",
  750. code=400
  751. )), 400
  752. preview = {
  753. 'time_condition': time_condition,
  754. 'cutoff_time': cutoff_time.isoformat(),
  755. 'will_be_removed': {
  756. 'sessions': []
  757. },
  758. 'will_be_kept': {
  759. 'sessions_count': 0,
  760. 'conversations_count': 0
  761. },
  762. 'summary': {
  763. 'sessions_to_remove': 0,
  764. 'conversations_to_remove': 0,
  765. 'sessions_to_keep': 0,
  766. 'conversations_to_keep': 0
  767. }
  768. }
  769. # 预览按session删除
  770. sessions_to_remove_count = 0
  771. conversations_to_remove_count = 0
  772. for session_id, session_data in cache.session_info.items():
  773. session_preview = {
  774. 'session_id': session_id,
  775. 'start_time': session_data['start_time'].isoformat(),
  776. 'conversation_count': len(session_data['conversations']),
  777. 'conversations': []
  778. }
  779. # 添加conversation详情
  780. for conv_id in session_data['conversations']:
  781. if conv_id in cache.cache:
  782. conv_data = cache.cache[conv_id]
  783. session_preview['conversations'].append({
  784. 'conversation_id': conv_id,
  785. 'question': conv_data.get('question', '')[:50] + '...' if conv_data.get('question') else '',
  786. 'start_time': cache.conversation_start_times.get(conv_id, '').isoformat() if cache.conversation_start_times.get(conv_id) else ''
  787. })
  788. if session_data['start_time'] < cutoff_time:
  789. preview['will_be_removed']['sessions'].append(session_preview)
  790. sessions_to_remove_count += 1
  791. conversations_to_remove_count += len(session_data['conversations'])
  792. else:
  793. preview['will_be_kept']['sessions_count'] += 1
  794. preview['will_be_kept']['conversations_count'] += len(session_data['conversations'])
  795. # 更新摘要统计
  796. preview['summary'] = {
  797. 'sessions_to_remove': sessions_to_remove_count,
  798. 'conversations_to_remove': conversations_to_remove_count,
  799. 'sessions_to_keep': preview['will_be_kept']['sessions_count'],
  800. 'conversations_to_keep': preview['will_be_kept']['conversations_count']
  801. }
  802. return jsonify(result.success(data=preview))
  803. except Exception as e:
  804. return jsonify(result.failed(
  805. message=f"预览清理操作失败: {str(e)}",
  806. code=500
  807. )), 500
  808. @app.flask_app.route('/api/v0/cache_cleanup', methods=['POST'])
  809. def cache_cleanup():
  810. """清理功能:实际删除缓存 - 保持原功能"""
  811. try:
  812. req = request.get_json(force=True)
  813. # 时间条件 - 支持三种方式
  814. older_than_hours = req.get('older_than_hours')
  815. older_than_days = req.get('older_than_days')
  816. before_timestamp = req.get('before_timestamp') # YYYY-MM-DD HH:MM:SS 格式
  817. cache = app.cache
  818. if not hasattr(cache, 'session_info'):
  819. return jsonify(result.failed(
  820. message="缓存不支持会话功能",
  821. code=400
  822. )), 400
  823. # 计算截止时间
  824. cutoff_time = None
  825. time_condition = None
  826. if older_than_hours:
  827. cutoff_time = datetime.now() - timedelta(hours=older_than_hours)
  828. time_condition = f"older_than_hours: {older_than_hours}"
  829. elif older_than_days:
  830. cutoff_time = datetime.now() - timedelta(days=older_than_days)
  831. time_condition = f"older_than_days: {older_than_days}"
  832. elif before_timestamp:
  833. try:
  834. # 支持 YYYY-MM-DD HH:MM:SS 格式
  835. cutoff_time = datetime.strptime(before_timestamp, '%Y-%m-%d %H:%M:%S')
  836. time_condition = f"before_timestamp: {before_timestamp}"
  837. except ValueError:
  838. return jsonify(result.failed(
  839. message="before_timestamp格式错误,请使用 YYYY-MM-DD HH:MM:SS 格式",
  840. code=400
  841. )), 400
  842. else:
  843. return jsonify(result.failed(
  844. message="必须提供时间条件:older_than_hours, older_than_days 或 before_timestamp (YYYY-MM-DD HH:MM:SS)",
  845. code=400
  846. )), 400
  847. cleanup_stats = {
  848. 'time_condition': time_condition,
  849. 'cutoff_time': cutoff_time.isoformat(),
  850. 'sessions_removed': 0,
  851. 'conversations_removed': 0,
  852. 'sessions_kept': 0,
  853. 'conversations_kept': 0,
  854. 'removed_session_ids': [],
  855. 'removed_conversation_ids': []
  856. }
  857. # 按session删除
  858. sessions_to_remove = []
  859. for session_id, session_data in cache.session_info.items():
  860. if session_data['start_time'] < cutoff_time:
  861. sessions_to_remove.append(session_id)
  862. # 删除符合条件的sessions及其所有conversations
  863. for session_id in sessions_to_remove:
  864. session_data = cache.session_info[session_id]
  865. conversations_in_session = session_data['conversations'].copy()
  866. # 删除session中的所有conversations
  867. for conv_id in conversations_in_session:
  868. if conv_id in cache.cache:
  869. del cache.cache[conv_id]
  870. cleanup_stats['conversations_removed'] += 1
  871. cleanup_stats['removed_conversation_ids'].append(conv_id)
  872. # 清理conversation相关的时间记录
  873. if hasattr(cache, 'conversation_start_times') and conv_id in cache.conversation_start_times:
  874. del cache.conversation_start_times[conv_id]
  875. if hasattr(cache, 'conversation_to_session') and conv_id in cache.conversation_to_session:
  876. del cache.conversation_to_session[conv_id]
  877. # 删除session记录
  878. del cache.session_info[session_id]
  879. cleanup_stats['sessions_removed'] += 1
  880. cleanup_stats['removed_session_ids'].append(session_id)
  881. # 统计保留的sessions和conversations
  882. cleanup_stats['sessions_kept'] = len(cache.session_info)
  883. cleanup_stats['conversations_kept'] = len(cache.cache)
  884. return jsonify(result.success(data=cleanup_stats))
  885. except Exception as e:
  886. return jsonify(result.failed(
  887. message=f"清理缓存失败: {str(e)}",
  888. code=500
  889. )), 500
  890. @app.flask_app.route('/api/v0/training_error_question_sql', methods=['POST'])
  891. def training_error_question_sql():
  892. """
  893. 存储错误的question-sql对到error_sql集合中
  894. 此API将接收的错误question/sql pair写入到error_sql集合中,用于记录和分析错误的SQL查询。
  895. Args:
  896. question (str, required): 用户问题
  897. sql (str, required): 对应的错误SQL查询语句
  898. Returns:
  899. JSON: 包含训练ID和成功消息的响应
  900. """
  901. try:
  902. data = request.get_json()
  903. question = data.get('question')
  904. sql = data.get('sql')
  905. print(f"[DEBUG] 接收到错误SQL训练请求: question={question}, sql={sql}")
  906. if not question or not sql:
  907. return jsonify(result.failed(
  908. message="question和sql参数都是必需的",
  909. code=400
  910. )), 400
  911. # 使用vn实例的train_error_sql方法存储错误SQL
  912. id = vn.train_error_sql(question=question, sql=sql)
  913. print(f"[INFO] 成功存储错误SQL,ID: {id}")
  914. return jsonify(result.success(data={
  915. "id": id,
  916. "message": "错误SQL对已成功存储到error_sql集合"
  917. }))
  918. except Exception as e:
  919. print(f"[ERROR] 存储错误SQL失败: {str(e)}")
  920. return jsonify(result.failed(
  921. message=f"存储错误SQL失败: {str(e)}",
  922. code=500
  923. )), 500
  924. # 前端JavaScript示例 - 如何维持会话
  925. """
  926. // 前端需要维护一个会话ID
  927. class ChatSession {
  928. constructor() {
  929. // 从localStorage获取或创建新的会话ID
  930. this.sessionId = localStorage.getItem('chat_session_id') || this.generateSessionId();
  931. localStorage.setItem('chat_session_id', this.sessionId);
  932. }
  933. generateSessionId() {
  934. return 'session_' + Date.now() + '_' + Math.random().toString(36).substr(2, 9);
  935. }
  936. async askQuestion(question) {
  937. const response = await fetch('/api/v0/ask', {
  938. method: 'POST',
  939. headers: {
  940. 'Content-Type': 'application/json',
  941. },
  942. body: JSON.stringify({
  943. question: question,
  944. session_id: this.sessionId // 关键:传递会话ID
  945. })
  946. });
  947. return await response.json();
  948. }
  949. // 开始新会话
  950. startNewSession() {
  951. this.sessionId = this.generateSessionId();
  952. localStorage.setItem('chat_session_id', this.sessionId);
  953. }
  954. }
  955. // 使用示例
  956. const chatSession = new ChatSession();
  957. chatSession.askQuestion("各年龄段客户的流失率如何?");
  958. """
  959. print("正在启动Flask应用: http://localhost:8084")
  960. app.run(host="0.0.0.0", port=8084, debug=True)