app.py 61 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570
  1. import os
  2. import json
  3. import uuid
  4. import time
  5. import logging
  6. from typing import Dict, Any, Generator, Optional, List
  7. from flask_cors import CORS
  8. import requests
  9. from flask import Flask, request, jsonify, render_template, session, Response, stream_with_context, redirect
  10. from http import HTTPStatus
  11. from dotenv import load_dotenv, find_dotenv
  12. from dashscope import Application
  13. from datetime import datetime
  14. # 先加载环境变量,确保后续模块读取到 .env(支持不同工作目录)
  15. load_dotenv(find_dotenv(), override=False)
  16. # 导入认证模块
  17. from auth_middleware import init_auth
  18. from auth_routes import auth_bp
  19. # 导入配置和工具
  20. from config import config
  21. from utils import (
  22. sanitize_input, validate_user_id, rate_limit,
  23. truncate_conversation_history, log_request_info, get_beijing_time,
  24. generate_session_id
  25. )
  26. # 导入数据库模型
  27. from models import init_database, db, User, Conversation, Message, get_or_create_user
  28. # 配置日志
  29. logging.basicConfig(
  30. level=getattr(logging, config.log_level),
  31. format=config.log_format
  32. )
  33. logger = logging.getLogger(__name__)
  34. # 创建Flask应用
  35. app = Flask(__name__)
  36. CORS(app) # 启用CORS支持
  37. # 配置session
  38. app.secret_key = config.secret_key
  39. app.config['PERMANENT_SESSION_LIFETIME'] = 7 * 24 * 60 * 60 # 7天
  40. app.config['SESSION_COOKIE_SECURE'] = False # 开发环境设为False
  41. app.config['SESSION_COOKIE_HTTPONLY'] = True
  42. app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
  43. # 配置数据库
  44. app.config['SQLALCHEMY_DATABASE_URI'] = config.get_database_url()
  45. app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
  46. app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
  47. 'pool_pre_ping': True,
  48. 'pool_recycle': 300,
  49. 'pool_timeout': 20,
  50. 'max_overflow': 0
  51. }
  52. # 初始化数据库
  53. try:
  54. init_database(app)
  55. except Exception as e:
  56. logger.error(f"数据库初始化失败: {str(e)}", exc_info=True)
  57. raise
  58. # 初始化认证中间件
  59. auth_middleware = init_auth(app)
  60. app.auth_middleware = auth_middleware
  61. # 注册认证蓝图
  62. app.register_blueprint(auth_bp)
  63. # 存储用户当前对话ID的字典(保留用于兼容性)
  64. user_current_conversations: Dict[str, int] = {}
  65. @app.after_request
  66. def add_cache_control_headers(response):
  67. """
  68. 防止首页HTML被中间层或浏览器长期缓存,避免出现"旧HTML引用新/旧静态资源"导致白屏。
  69. 静态资源保持默认缓存策略,仅对HTML页面禁缓存。
  70. """
  71. content_type = (response.content_type or "").lower()
  72. if "text/html" in content_type:
  73. response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
  74. response.headers["Pragma"] = "no-cache"
  75. response.headers["Expires"] = "0"
  76. return response
  77. def save_message_to_database(user_id: int, conversation_id: int, role: str, content: str,
  78. thinking_content: str = None, message_order: int = 1) -> Message:
  79. """保存消息到数据库"""
  80. try:
  81. message = Message(
  82. conversation_id=conversation_id,
  83. role=role,
  84. content=content,
  85. thinking_content=thinking_content,
  86. content_type='text',
  87. message_order=message_order
  88. )
  89. db.session.add(message)
  90. # 更新对话的消息计数和最后消息时间
  91. conversation = Conversation.query.get(conversation_id)
  92. if conversation:
  93. conversation.message_count += 1
  94. conversation.last_message_at = get_beijing_time()
  95. conversation.updated_at = get_beijing_time()
  96. # 一次性提交所有更改
  97. db.session.commit()
  98. return message
  99. except Exception as e:
  100. db.session.rollback()
  101. logger.error(f"保存消息到数据库时发生错误: {str(e)}", exc_info=True)
  102. raise
  103. def get_or_create_conversation(user_id: int, title: str = None, conversation_id: int = None, force_new: bool = False) -> Conversation:
  104. """获取或创建对话"""
  105. try:
  106. # 如果强制创建新对话,直接创建
  107. if force_new:
  108. if not title:
  109. title = f"对话 {get_beijing_time().strftime('%Y-%m-%d %H:%M')}"
  110. # 生成新的session_id
  111. session_id = generate_session_id(user_id)
  112. conversation = Conversation(
  113. user_id=user_id,
  114. title=title,
  115. conversation_type='chat',
  116. session_id=session_id,
  117. message_count=0
  118. )
  119. db.session.add(conversation)
  120. db.session.commit()
  121. # 更新用户的当前对话ID
  122. user_current_conversations[str(user_id)] = conversation.conversation_id
  123. return conversation
  124. # 如果指定了conversation_id,验证并返回该对话
  125. if conversation_id:
  126. conversation = Conversation.query.filter_by(
  127. conversation_id=conversation_id,
  128. user_id=user_id,
  129. is_deleted=False
  130. ).first()
  131. if conversation:
  132. # 如果对话没有session_id,生成一个
  133. if not conversation.session_id:
  134. conversation.session_id = generate_session_id(user_id, conversation_id)
  135. db.session.commit()
  136. # 更新用户的当前对话ID
  137. user_current_conversations[str(user_id)] = conversation.conversation_id
  138. return conversation
  139. # 检查用户是否有当前对话映射
  140. current_conv_id = user_current_conversations.get(str(user_id))
  141. if current_conv_id:
  142. conversation = Conversation.query.filter_by(
  143. conversation_id=current_conv_id,
  144. user_id=user_id,
  145. is_deleted=False
  146. ).first()
  147. if conversation:
  148. # 如果对话没有session_id,生成一个
  149. if not conversation.session_id:
  150. conversation.session_id = generate_session_id(user_id, current_conv_id)
  151. db.session.commit()
  152. return conversation
  153. # 创建新对话
  154. if not title:
  155. title = f"对话 {get_beijing_time().strftime('%Y-%m-%d %H:%M')}"
  156. # 生成新的session_id
  157. session_id = generate_session_id(user_id)
  158. conversation = Conversation(
  159. user_id=user_id,
  160. title=title,
  161. conversation_type='chat',
  162. session_id=session_id,
  163. message_count=0
  164. )
  165. db.session.add(conversation)
  166. db.session.commit()
  167. # 更新用户的当前对话ID
  168. user_current_conversations[str(user_id)] = conversation.conversation_id
  169. return conversation
  170. except Exception as e:
  171. db.session.rollback()
  172. logger.error(f"获取或创建对话时发生错误: {str(e)}", exc_info=True)
  173. raise
  174. def switch_to_conversation(user_id: int, conversation_id: int) -> Conversation:
  175. """切换到指定对话"""
  176. try:
  177. # 验证对话是否属于当前用户
  178. conversation = Conversation.query.filter_by(
  179. conversation_id=conversation_id,
  180. user_id=user_id,
  181. is_deleted=False
  182. ).first()
  183. if not conversation:
  184. raise ValueError(f"对话 {conversation_id} 不存在或无权限访问")
  185. # 如果对话没有session_id,生成一个
  186. if not conversation.session_id:
  187. conversation.session_id = generate_session_id(user_id, conversation_id)
  188. db.session.commit()
  189. # 更新用户的当前对话ID
  190. user_current_conversations[str(user_id)] = conversation.conversation_id
  191. return conversation
  192. except Exception as e:
  193. logger.error(f"切换对话时发生错误: {str(e)}", exc_info=True)
  194. raise
  195. def generate_conversation_title(user_message: str) -> str:
  196. """根据用户消息生成对话标题"""
  197. # 取用户消息的前50个字符作为标题
  198. title = user_message.strip()[:50]
  199. if len(user_message) > 50:
  200. title += "..."
  201. return title
  202. @app.route('/')
  203. def index():
  204. """渲染聊天界面"""
  205. # 检查是否是登录成功跳转
  206. login_success = request.args.get('login_success')
  207. token = request.args.get('token')
  208. user_info_json = request.args.get('user_info')
  209. if login_success and token and user_info_json:
  210. # 登录成功,处理用户信息
  211. try:
  212. import urllib.parse
  213. import json
  214. user_info = json.loads(urllib.parse.unquote(user_info_json))
  215. # 保存到session
  216. session['auth_token'] = token
  217. session['user_info'] = user_info
  218. # 将用户信息传递给模板
  219. return render_template('index.html', user_info=user_info, jwt_token=token)
  220. except Exception as e:
  221. logger.error(f'处理登录成功参数失败: {e}')
  222. # 如果处理失败,重定向到登录页面
  223. return redirect('/login')
  224. else:
  225. # 正常访问,检查认证状态
  226. # 优先从session获取
  227. user_info = session.get('user_info', {})
  228. jwt_token = session.get('auth_token', '')
  229. # 如果session中没有,尝试从localStorage获取(通过JavaScript)
  230. if not jwt_token:
  231. # 返回一个页面,让JavaScript检查localStorage
  232. return render_template('index.html', user_info={}, jwt_token='')
  233. # 将用户信息传递给模板
  234. return render_template('index.html', user_info=user_info, jwt_token=jwt_token)
  235. @app.route('/login')
  236. def login():
  237. """登录页面"""
  238. return render_template('login.html')
  239. @app.route('/login_error')
  240. def login_error():
  241. """登录错误页面"""
  242. error_type = request.args.get('error', 'unknown')
  243. error_message = request.args.get('message', '')
  244. error_code = request.args.get('code', '')
  245. return render_template('login_error.html',
  246. error_type=error_type,
  247. error_message=error_message,
  248. error_code=error_code)
  249. @app.route('/api/chat', methods=['POST'])
  250. @auth_middleware.login_required
  251. @rate_limit(config.rate_limit_requests, config.rate_limit_window)
  252. def chat():
  253. """处理聊天请求并调用阿里云百炼API (非流式响应)"""
  254. try:
  255. data = request.json or {}
  256. message = data.get('message', '')
  257. user_id = data.get('user_id', str(uuid.uuid4()))
  258. agent = data.get('agent') or 'kuaishou'
  259. conversation_id = data.get('conversation_id') # 新增:支持指定对话ID
  260. session_id = data.get('session_id') # 新增:支持指定session_id
  261. force_new = data.get('force_new', False) # 新增:强制创建新对话
  262. # 输入验证
  263. message, error = sanitize_input(message, config.max_message_length)
  264. if error:
  265. return jsonify({
  266. 'success': False,
  267. 'error': error
  268. }), HTTPStatus.BAD_REQUEST
  269. user_id, error = validate_user_id(user_id)
  270. if error:
  271. return jsonify({
  272. 'success': False,
  273. 'error': error
  274. }), HTTPStatus.BAD_REQUEST
  275. # 从认证中间件获取用户信息
  276. auth_user_info = getattr(request, 'user_info', {})
  277. auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID
  278. # 选择对应智能体 APP_ID
  279. chosen_app_id, appid_env_key = config.get_app_id_for_agent_with_source(agent)
  280. if not chosen_app_id:
  281. logger.error(f"未找到对应智能体APP_ID,agent={agent}")
  282. return jsonify({
  283. 'success': False,
  284. 'error': f"服务器未配置该智能体的 APP_ID: {agent}"
  285. }), HTTPStatus.INTERNAL_SERVER_ERROR
  286. # 记录请求信息
  287. log_request_info(message, user_id, 'POST')
  288. # 获取或创建对话
  289. conversation = get_or_create_conversation(
  290. auth_user_id,
  291. generate_conversation_title(message),
  292. conversation_id,
  293. force_new
  294. )
  295. # 优先使用前端传来的session_id,否则使用对话的session_id
  296. api_session_id = session_id or conversation.session_id
  297. # 准备API请求参数
  298. headers = {
  299. "Authorization": f"Bearer {config.api_key}",
  300. "Content-Type": "application/json",
  301. "X-DashScope-SSE": "disable" # 禁用流式输出
  302. }
  303. # 从数据库获取对话历史
  304. messages_query = Message.query.filter_by(
  305. conversation_id=conversation.conversation_id,
  306. is_deleted=False
  307. ).order_by(Message.message_order.asc()).all()
  308. # 构建多轮对话历史
  309. messages = []
  310. for msg in messages_query:
  311. messages.append({"role": msg.role, "content": msg.content})
  312. # 添加用户新消息到对话历史
  313. messages.append({"role": "user", "content": message})
  314. # 截断对话历史以控制内存使用
  315. messages = truncate_conversation_history(messages, config.max_conversation_history)
  316. payload = {
  317. "model": "application",
  318. "parameters": {"app_id": chosen_app_id},
  319. "input": {"messages": messages} # 发送完整的对话历史
  320. }
  321. # 如果有会话ID,添加到请求参数中
  322. if api_session_id:
  323. payload["parameters"]["session_id"] = api_session_id
  324. # 发送API请求
  325. response = requests.post(
  326. config.api_endpoint,
  327. headers=headers,
  328. json=payload,
  329. timeout=60
  330. )
  331. # 转换为JSON
  332. response_json = response.json()
  333. if response.status_code != HTTPStatus.OK:
  334. return jsonify({
  335. 'success': False,
  336. 'error': f"错误: {response.status_code} - {response_json.get('message', '未知错误')}",
  337. 'request_id': response_json.get('request_id', ''),
  338. 'agent': agent
  339. }), HTTPStatus.INTERNAL_SERVER_ERROR
  340. # 解析返回结果
  341. output = response_json.get('output', {})
  342. choices = output.get('choices', [{}])
  343. text = choices[0].get('message', {}).get('content', '') if choices else ''
  344. new_session_id = output.get('session_id', '')
  345. thoughts_content = output.get('thoughts', [])
  346. # 更新对话的session_id(如果API返回了新的session_id且与当前不同)
  347. if new_session_id and new_session_id != api_session_id:
  348. conversation.session_id = new_session_id
  349. db.session.commit()
  350. elif not api_session_id and new_session_id:
  351. # 如果对话原本没有session_id,现在有了,更新数据库
  352. conversation.session_id = new_session_id
  353. db.session.commit()
  354. # 保存到数据库
  355. try:
  356. # 获取当前对话的消息数量,用于设置消息顺序
  357. current_message_count = Message.query.filter_by(
  358. conversation_id=conversation.conversation_id,
  359. is_deleted=False
  360. ).count()
  361. # 保存用户消息
  362. save_message_to_database(
  363. auth_user_id,
  364. conversation.conversation_id,
  365. 'user',
  366. message,
  367. message_order=current_message_count + 1
  368. )
  369. # 保存AI回复
  370. if text:
  371. thinking_text = '\n'.join(thoughts_content) if thoughts_content else None
  372. save_message_to_database(
  373. auth_user_id,
  374. conversation.conversation_id,
  375. 'assistant',
  376. text,
  377. thinking_text,
  378. message_order=current_message_count + 2
  379. )
  380. # 更新对话标题(如果是第一条消息)
  381. if conversation.message_count <= 2: # 用户消息 + AI回复
  382. new_title = generate_conversation_title(message)
  383. # 检查是否是默认标题("新对话"或"对话 时间"格式)
  384. is_default_title = (
  385. conversation.title == "新对话" or
  386. conversation.title == f"对话 {conversation.created_at.strftime('%Y-%m-%d %H:%M')}"
  387. )
  388. if is_default_title:
  389. conversation.title = new_title
  390. db.session.commit()
  391. except Exception as db_error:
  392. logger.error(f"保存对话到数据库时发生错误: {str(db_error)}", exc_info=True)
  393. # 不影响主要功能,继续返回结果
  394. return jsonify({
  395. 'success': True,
  396. 'message': text,
  397. 'user_id': user_id,
  398. 'session_id': new_session_id or api_session_id,
  399. 'thoughts': thoughts_content,
  400. 'agent': agent,
  401. 'conversation_id': conversation.conversation_id
  402. })
  403. except requests.exceptions.RequestException as e:
  404. logger.error(f"API请求异常: {str(e)}", exc_info=True)
  405. return jsonify({
  406. 'success': False,
  407. 'error': f"API请求异常: {str(e)}"
  408. }), HTTPStatus.SERVICE_UNAVAILABLE
  409. except Exception as e:
  410. logger.error(f"API调用错误: {str(e)}", exc_info=True)
  411. return jsonify({
  412. 'success': False,
  413. 'error': f"发生错误: {str(e)}"
  414. }), HTTPStatus.INTERNAL_SERVER_ERROR
  415. @app.route('/api/chat/stream', methods=['GET', 'POST'])
  416. @auth_middleware.login_required
  417. @rate_limit(config.rate_limit_requests, config.rate_limit_window)
  418. def chat_stream():
  419. """处理聊天请求并以流式方式返回响应"""
  420. try:
  421. # 从请求中获取数据
  422. if request.method == 'GET':
  423. message = request.args.get('message', '')
  424. user_id = request.args.get('user_id', str(uuid.uuid4()))
  425. agent = request.args.get('agent') or 'kuaishou'
  426. conversation_id = request.args.get('conversation_id') # 支持指定对话ID
  427. else: # POST
  428. data = request.json or {}
  429. message = data.get('message', '')
  430. user_id = data.get('user_id', str(uuid.uuid4()))
  431. agent = data.get('agent') or 'kuaishou'
  432. conversation_id = data.get('conversation_id') # 支持指定对话ID
  433. session_id = data.get('session_id') # 支持指定session_id
  434. force_new = data.get('force_new', False) # 强制创建新对话
  435. # 输入验证
  436. message, error = sanitize_input(message, config.max_message_length)
  437. if error:
  438. return Response(
  439. f"data: {json.dumps({'success': False, 'error': error, 'done': True})}\n\n",
  440. content_type='text/event-stream'
  441. )
  442. user_id, error = validate_user_id(user_id)
  443. if error:
  444. return Response(
  445. f"data: {json.dumps({'success': False, 'error': error, 'done': True})}\n\n",
  446. content_type='text/event-stream'
  447. )
  448. # 选择对应智能体 APP_ID
  449. chosen_app_id, appid_env_key = config.get_app_id_for_agent_with_source(agent)
  450. if not chosen_app_id:
  451. logger.error(f"未找到对应智能体APP_ID,agent={agent}")
  452. return Response(
  453. f"data: {json.dumps({'success': False, 'error': f'服务器未配置该智能体的 APP_ID: {agent}', 'done': True})}\n\n",
  454. content_type='text/event-stream'
  455. )
  456. # 记录请求信息
  457. log_request_info(message, user_id, request.method)
  458. # 从认证中间件获取用户信息
  459. auth_user_info = getattr(request, 'user_info', {})
  460. auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID
  461. # 获取或创建对话
  462. conversation = get_or_create_conversation(
  463. auth_user_id,
  464. generate_conversation_title(message),
  465. conversation_id,
  466. force_new
  467. )
  468. # 优先使用前端传来的session_id,否则使用对话的session_id
  469. api_session_id = session_id or conversation.session_id
  470. # 从数据库获取对话历史
  471. messages_query = Message.query.filter_by(
  472. conversation_id=conversation.conversation_id,
  473. is_deleted=False
  474. ).order_by(Message.message_order.asc()).all()
  475. # 构建多轮对话历史
  476. messages = []
  477. for msg in messages_query:
  478. messages.append({"role": msg.role, "content": msg.content})
  479. # 添加用户新消息到对话历史
  480. messages.append({"role": "user", "content": message})
  481. # 截断对话历史以控制内存使用
  482. messages = truncate_conversation_history(messages, config.max_conversation_history)
  483. # 立即保存用户消息到数据库(在流式响应开始前)
  484. try:
  485. # 获取当前对话的消息数量,用于设置消息顺序
  486. current_message_count = Message.query.filter_by(
  487. conversation_id=conversation.conversation_id,
  488. is_deleted=False
  489. ).count()
  490. # 立即保存用户消息
  491. user_message = save_message_to_database(
  492. auth_user_id,
  493. conversation.conversation_id,
  494. 'user',
  495. message,
  496. message_order=current_message_count + 1
  497. )
  498. except Exception as db_error:
  499. logger.error(f"保存用户消息到数据库时发生错误: {str(db_error)}", exc_info=True)
  500. return Response(
  501. f"data: {json.dumps({'success': False, 'error': f'保存用户消息失败: {str(db_error)}', 'done': True})}\n\n",
  502. content_type='text/event-stream'
  503. )
  504. # 定义流式响应生成器
  505. def generate() -> Generator[str, None, None]:
  506. try:
  507. # 发送初始响应,包含用户消息ID,防止浏览器超时
  508. initial_data = {
  509. 'success': True,
  510. 'chunk': '',
  511. 'done': False,
  512. 'agent': agent,
  513. 'user_message_id': user_message.message_id if user_message else None
  514. }
  515. yield f"data: {json.dumps(initial_data)}\n\n"
  516. try:
  517. # 使用DashScope SDK直接调用 - 多轮对话模式
  518. params = {
  519. "api_key": config.api_key,
  520. "app_id": chosen_app_id,
  521. "messages": messages, # 发送完整的对话历史
  522. "stream": True,
  523. "incremental_output": True,
  524. "has_thoughts": True
  525. }
  526. # 使用对话的session_id
  527. if api_session_id:
  528. params["session_id"] = api_session_id
  529. response = Application.call(**params)
  530. new_session_id: Optional[str] = None
  531. accumulated_text: str = ""
  532. thoughts_content: List[str] = []
  533. # 处理流式响应
  534. for chunk in response:
  535. # 检查客户端是否断开连接
  536. if request.environ.get('werkzeug.server.shutdown') or not request:
  537. break
  538. if chunk.status_code != HTTPStatus.OK:
  539. yield f"data: {json.dumps({'success': False, 'error': f'错误: {chunk.status_code} - {chunk.message}', 'request_id': chunk.request_id, 'done': True, 'agent': agent})}\n\n"
  540. return
  541. # 处理思考过程
  542. if hasattr(chunk.output, 'thoughts') and chunk.output.thoughts:
  543. for it in chunk.output.thoughts:
  544. if it.action_type == 'reasoning':
  545. content = str(it.thought) if not isinstance(it.thought, str) else it.thought
  546. thoughts_content.append(content)
  547. yield f"data: {json.dumps({'success': True, 'thought': content, 'done': False, 'agent': agent})}\n\n"
  548. # 处理回答内容
  549. if hasattr(chunk.output, 'text') and chunk.output.text:
  550. current_text = str(chunk.output.text)
  551. accumulated_text += current_text
  552. yield f"data: {json.dumps({'success': True, 'chunk': current_text, 'done': False, 'agent': agent})}\n\n"
  553. # 检查是否有会话ID
  554. if hasattr(chunk.output, 'session_id') and chunk.output.session_id:
  555. new_session_id = chunk.output.session_id
  556. # 处理结束
  557. if new_session_id and new_session_id != api_session_id:
  558. conversation.session_id = new_session_id
  559. db.session.commit()
  560. elif not api_session_id and new_session_id:
  561. # 如果对话原本没有session_id,现在有了,更新数据库
  562. conversation.session_id = new_session_id
  563. db.session.commit()
  564. # 保存AI回复到数据库
  565. try:
  566. if accumulated_text:
  567. thinking_text = '\n'.join(thoughts_content) if thoughts_content else None
  568. ai_message = save_message_to_database(
  569. auth_user_id,
  570. conversation.conversation_id,
  571. 'assistant',
  572. accumulated_text,
  573. thinking_text,
  574. message_order=current_message_count + 2
  575. )
  576. # 更新对话标题(如果是第一条消息)
  577. if conversation.message_count <= 2: # 用户消息 + AI回复
  578. new_title = generate_conversation_title(message)
  579. # 检查是否是默认标题("新对话"或"对话 时间"格式)
  580. is_default_title = (
  581. conversation.title == "新对话" or
  582. conversation.title == f"对话 {conversation.created_at.strftime('%Y-%m-%d %H:%M')}"
  583. )
  584. if is_default_title:
  585. conversation.title = new_title
  586. db.session.commit()
  587. except Exception as db_error:
  588. logger.error(f"保存AI回复到数据库时发生错误: {str(db_error)}", exc_info=True)
  589. # 不影响主要功能,继续返回结果
  590. # 发送完成标记
  591. final_data = {
  592. 'success': True,
  593. 'done': True,
  594. 'thoughts': thoughts_content,
  595. 'agent': agent,
  596. 'session_id': new_session_id or api_session_id,
  597. 'user_id': user_id,
  598. 'conversation_id': conversation.conversation_id
  599. }
  600. yield f"data: {json.dumps(final_data)}\n\n"
  601. except Exception as e:
  602. logger.error(f"DashScope SDK 调用错误: {str(e)}", exc_info=True)
  603. yield f"data: {json.dumps({'success': False, 'error': f'DashScope API调用错误: {str(e)}', 'done': True, 'agent': agent})}\n\n"
  604. except Exception as e:
  605. logger.error(f"流式API调用错误: {str(e)}", exc_info=True)
  606. yield f"data: {json.dumps({'success': False, 'error': f'发生错误: {str(e)}', 'done': True, 'agent': agent})}\n\n"
  607. # 创建并返回流式响应
  608. response = Response(stream_with_context(generate()), content_type='text/event-stream')
  609. response.headers['Cache-Control'] = 'no-cache'
  610. response.headers['X-Accel-Buffering'] = 'no'
  611. return response
  612. except Exception as e:
  613. logger.error(f"请求处理错误: {str(e)}", exc_info=True)
  614. response = Response(
  615. f"data: {json.dumps({'success': False, 'error': f'发生错误: {str(e)}', 'done': True})}\n\n",
  616. content_type='text/event-stream'
  617. )
  618. response.headers['Cache-Control'] = 'no-cache'
  619. return response
  620. @app.route('/api/reset', methods=['POST'])
  621. @auth_middleware.login_required
  622. @rate_limit(10, 60)
  623. def reset_session():
  624. """重置用户会话"""
  625. try:
  626. data = request.json or {}
  627. user_id = data.get('user_id')
  628. if not user_id:
  629. return jsonify({
  630. 'success': True,
  631. 'message': '没有提供用户ID,但重置请求已处理'
  632. })
  633. # 验证用户ID格式
  634. user_id, error = validate_user_id(user_id)
  635. if error:
  636. return jsonify({
  637. 'success': False,
  638. 'error': error
  639. }), HTTPStatus.BAD_REQUEST
  640. # 从认证中间件获取用户信息
  641. auth_user_info = getattr(request, 'user_info', {})
  642. auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID
  643. # 清除数据库中的当前对话ID(如果有认证用户信息)
  644. if auth_user_id and str(auth_user_id) in user_current_conversations:
  645. del user_current_conversations[str(auth_user_id)]
  646. return jsonify({
  647. 'success': True,
  648. 'message': '会话已重置'
  649. })
  650. except Exception as e:
  651. logger.error(f"重置会话时发生错误: {str(e)}", exc_info=True)
  652. return jsonify({
  653. 'success': False,
  654. 'error': f"发生错误: {str(e)}"
  655. }), HTTPStatus.INTERNAL_SERVER_ERROR
  656. @app.route('/api/status', methods=['GET'])
  657. @auth_middleware.login_required
  658. def get_status():
  659. """获取应用状态信息"""
  660. try:
  661. active_users = len(user_current_conversations)
  662. total_conversations = Conversation.query.filter_by(is_deleted=False).count()
  663. return jsonify({
  664. 'success': True,
  665. 'status': 'running',
  666. 'active_users': active_users,
  667. 'total_conversations': total_conversations,
  668. 'app_version': '2.0.0',
  669. 'python_version': '3.12',
  670. 'config': {
  671. 'max_message_length': config.max_message_length,
  672. 'max_conversation_history': config.max_conversation_history,
  673. 'rate_limit_requests': config.rate_limit_requests,
  674. 'rate_limit_window': config.rate_limit_window
  675. }
  676. })
  677. except Exception as e:
  678. logger.error(f"获取状态时发生错误: {str(e)}", exc_info=True)
  679. return jsonify({
  680. 'success': False,
  681. 'error': f"发生错误: {str(e)}"
  682. }), HTTPStatus.INTERNAL_SERVER_ERROR
  683. # ==================== 历史记录相关API ====================
  684. @app.route('/api/conversations', methods=['GET'])
  685. @auth_middleware.login_required
  686. def get_conversations():
  687. """获取用户的历史对话列表,按时间分组"""
  688. try:
  689. # 从认证中间件获取用户信息
  690. user_info = getattr(request, 'user_info', {})
  691. user_id = user_info.get('db_user_id') # 使用数据库用户ID
  692. if not user_id:
  693. logger.warning('获取历史记录失败 - 用户ID为空')
  694. return jsonify({
  695. 'success': False,
  696. 'error': '用户信息获取失败'
  697. }), HTTPStatus.UNAUTHORIZED
  698. # 获取用户对象
  699. user = User.query.filter_by(user_id=user_id, is_deleted=False).first()
  700. if not user:
  701. return jsonify({
  702. 'success': False,
  703. 'error': '用户不存在'
  704. }), HTTPStatus.NOT_FOUND
  705. # 获取用户的对话列表,按更新时间倒序
  706. conversations = Conversation.query.filter_by(
  707. user_id=user_id,
  708. is_deleted=False
  709. ).order_by(Conversation.updated_at.desc()).all()
  710. # 按时间分组
  711. from datetime import datetime, timedelta
  712. now = get_beijing_time()
  713. today = now.date()
  714. yesterday = today - timedelta(days=1)
  715. week_ago = today - timedelta(days=7)
  716. month_ago = today - timedelta(days=30)
  717. grouped_conversations = {
  718. 'today': [],
  719. 'yesterday': [],
  720. 'week': [],
  721. 'month': []
  722. }
  723. for conv in conversations:
  724. conv_date = conv.updated_at.date()
  725. conv_data = {
  726. 'conversation_id': conv.conversation_id,
  727. 'title': conv.title,
  728. 'message_count': conv.message_count,
  729. 'session_id': conv.session_id,
  730. 'last_message_at': conv.last_message_at.isoformat() if conv.last_message_at else None,
  731. 'created_at': conv.created_at.isoformat(),
  732. 'updated_at': conv.updated_at.isoformat()
  733. }
  734. if conv_date == today:
  735. grouped_conversations['today'].append(conv_data)
  736. elif conv_date == yesterday:
  737. grouped_conversations['yesterday'].append(conv_data)
  738. elif conv_date >= week_ago:
  739. grouped_conversations['week'].append(conv_data)
  740. elif conv_date >= month_ago:
  741. grouped_conversations['month'].append(conv_data)
  742. return jsonify({
  743. 'success': True,
  744. 'conversations': grouped_conversations
  745. })
  746. except Exception as e:
  747. logger.error(f"获取对话列表时发生错误: {str(e)}", exc_info=True)
  748. return jsonify({
  749. 'success': False,
  750. 'error': f"发生错误: {str(e)}"
  751. }), HTTPStatus.INTERNAL_SERVER_ERROR
  752. @app.route('/api/conversations/<int:conversation_id>', methods=['GET'])
  753. @auth_middleware.login_required
  754. def get_conversation_messages(conversation_id):
  755. """获取指定对话的所有消息"""
  756. try:
  757. # 从认证中间件获取用户信息
  758. user_info = getattr(request, 'user_info', {})
  759. user_id = user_info.get('db_user_id') # 使用数据库用户ID
  760. if not user_id:
  761. return jsonify({
  762. 'success': False,
  763. 'error': '用户信息获取失败'
  764. }), HTTPStatus.UNAUTHORIZED
  765. # 验证对话是否属于当前用户
  766. conversation = Conversation.query.filter_by(
  767. conversation_id=conversation_id,
  768. user_id=user_id,
  769. is_deleted=False
  770. ).first()
  771. if not conversation:
  772. return jsonify({
  773. 'success': False,
  774. 'error': '对话不存在或无权限访问'
  775. }), HTTPStatus.NOT_FOUND
  776. # 获取对话的所有消息,按顺序排列
  777. messages = Message.query.filter_by(
  778. conversation_id=conversation_id,
  779. is_deleted=False
  780. ).order_by(Message.message_order.asc()).all()
  781. messages_data = []
  782. for msg in messages:
  783. message_data = {
  784. 'message_id': msg.message_id,
  785. 'role': msg.role,
  786. 'content': msg.content,
  787. 'thinking_content': msg.thinking_content,
  788. 'content_type': msg.content_type,
  789. 'message_order': msg.message_order,
  790. 'created_at': msg.created_at.isoformat()
  791. }
  792. messages_data.append(message_data)
  793. return jsonify({
  794. 'success': True,
  795. 'conversation': {
  796. 'conversation_id': conversation.conversation_id,
  797. 'title': conversation.title,
  798. 'conversation_type': conversation.conversation_type,
  799. 'session_id': conversation.session_id,
  800. 'message_count': conversation.message_count,
  801. 'created_at': conversation.created_at.isoformat(),
  802. 'updated_at': conversation.updated_at.isoformat()
  803. },
  804. 'messages': messages_data
  805. })
  806. except Exception as e:
  807. logger.error(f"获取对话消息时发生错误: {str(e)}", exc_info=True)
  808. return jsonify({
  809. 'success': False,
  810. 'error': f"发生错误: {str(e)}"
  811. }), HTTPStatus.INTERNAL_SERVER_ERROR
  812. @app.route('/api/conversations/<int:conversation_id>/rename', methods=['POST'])
  813. @auth_middleware.login_required
  814. def rename_conversation(conversation_id):
  815. """重命名对话"""
  816. try:
  817. # 从认证中间件获取用户信息
  818. user_info = getattr(request, 'user_info', {})
  819. user_id = user_info.get('db_user_id') # 使用数据库用户ID
  820. if not user_id:
  821. return jsonify({
  822. 'success': False,
  823. 'error': '用户信息获取失败'
  824. }), HTTPStatus.UNAUTHORIZED
  825. data = request.json or {}
  826. new_title = data.get('title', '').strip()
  827. if not new_title:
  828. return jsonify({
  829. 'success': False,
  830. 'error': '新标题不能为空'
  831. }), HTTPStatus.BAD_REQUEST
  832. if len(new_title) > 500:
  833. return jsonify({
  834. 'success': False,
  835. 'error': '标题长度不能超过500个字符'
  836. }), HTTPStatus.BAD_REQUEST
  837. # 验证对话是否属于当前用户
  838. conversation = Conversation.query.filter_by(
  839. conversation_id=conversation_id,
  840. user_id=user_id,
  841. is_deleted=False
  842. ).first()
  843. if not conversation:
  844. return jsonify({
  845. 'success': False,
  846. 'error': '对话不存在或无权限访问'
  847. }), HTTPStatus.NOT_FOUND
  848. # 更新标题
  849. conversation.title = new_title
  850. conversation.updated_at = get_beijing_time()
  851. db.session.commit()
  852. return jsonify({
  853. 'success': True,
  854. 'message': '对话标题已更新',
  855. 'conversation': {
  856. 'conversation_id': conversation.conversation_id,
  857. 'title': conversation.title,
  858. 'updated_at': conversation.updated_at.isoformat()
  859. }
  860. })
  861. except Exception as e:
  862. db.session.rollback()
  863. logger.error(f"重命名对话时发生错误: {str(e)}", exc_info=True)
  864. return jsonify({
  865. 'success': False,
  866. 'error': f"发生错误: {str(e)}"
  867. }), HTTPStatus.INTERNAL_SERVER_ERROR
  868. @app.route('/api/conversations/<int:conversation_id>', methods=['DELETE'])
  869. @auth_middleware.login_required
  870. def delete_conversation(conversation_id):
  871. """删除对话(软删除)"""
  872. try:
  873. # 从认证中间件获取用户信息
  874. user_info = getattr(request, 'user_info', {})
  875. user_id = user_info.get('db_user_id') # 使用数据库用户ID
  876. if not user_id:
  877. return jsonify({
  878. 'success': False,
  879. 'error': '用户信息获取失败'
  880. }), HTTPStatus.UNAUTHORIZED
  881. # 验证对话是否属于当前用户
  882. conversation = Conversation.query.filter_by(
  883. conversation_id=conversation_id,
  884. user_id=user_id,
  885. is_deleted=False
  886. ).first()
  887. if not conversation:
  888. return jsonify({
  889. 'success': False,
  890. 'error': '对话不存在或无权限访问'
  891. }), HTTPStatus.NOT_FOUND
  892. # 软删除对话
  893. conversation.is_deleted = True
  894. conversation.deleted_at = get_beijing_time()
  895. conversation.updated_at = get_beijing_time()
  896. # 同时软删除该对话的所有消息
  897. Message.query.filter_by(
  898. conversation_id=conversation_id,
  899. is_deleted=False
  900. ).update({
  901. 'is_deleted': True,
  902. 'deleted_at': get_beijing_time()
  903. })
  904. db.session.commit()
  905. return jsonify({
  906. 'success': True,
  907. 'message': '对话已删除'
  908. })
  909. except Exception as e:
  910. db.session.rollback()
  911. logger.error(f"删除对话时发生错误: {str(e)}", exc_info=True)
  912. return jsonify({
  913. 'success': False,
  914. 'error': f"发生错误: {str(e)}"
  915. }), HTTPStatus.INTERNAL_SERVER_ERROR
  916. @app.route('/api/conversations/new', methods=['POST'])
  917. @auth_middleware.login_required
  918. def create_new_conversation():
  919. """创建新对话"""
  920. try:
  921. # 从认证中间件获取用户信息
  922. user_info = getattr(request, 'user_info', {})
  923. user_id = user_info.get('db_user_id') # 使用数据库用户ID
  924. if not user_id:
  925. return jsonify({
  926. 'success': False,
  927. 'error': '用户信息获取失败'
  928. }), HTTPStatus.UNAUTHORIZED
  929. # 获取用户对象
  930. user = User.query.filter_by(user_id=user_id, is_deleted=False).first()
  931. if not user:
  932. return jsonify({
  933. 'success': False,
  934. 'error': '用户不存在'
  935. }), HTTPStatus.NOT_FOUND
  936. data = request.json or {}
  937. title = data.get('title', '新对话').strip()
  938. if not title:
  939. title = '新对话'
  940. if len(title) > 500:
  941. return jsonify({
  942. 'success': False,
  943. 'error': '标题长度不能超过500个字符'
  944. }), HTTPStatus.BAD_REQUEST
  945. # 生成新的session_id
  946. session_id = generate_session_id(user_id)
  947. # 创建新对话
  948. conversation = Conversation(
  949. user_id=user_id,
  950. title=title,
  951. conversation_type='chat',
  952. session_id=session_id,
  953. message_count=0
  954. )
  955. db.session.add(conversation)
  956. db.session.commit()
  957. # 更新用户的当前对话ID
  958. user_current_conversations[str(user_id)] = conversation.conversation_id
  959. return jsonify({
  960. 'success': True,
  961. 'message': '新对话已创建',
  962. 'conversation': {
  963. 'conversation_id': conversation.conversation_id,
  964. 'title': conversation.title,
  965. 'created_at': conversation.created_at.isoformat(),
  966. 'updated_at': conversation.updated_at.isoformat()
  967. }
  968. })
  969. except Exception as e:
  970. db.session.rollback()
  971. logger.error(f"创建新对话时发生错误: {str(e)}", exc_info=True)
  972. return jsonify({
  973. 'success': False,
  974. 'error': f"发生错误: {str(e)}"
  975. }), HTTPStatus.INTERNAL_SERVER_ERROR
  976. @app.route('/api/conversations/clear-current', methods=['POST'])
  977. @auth_middleware.login_required
  978. def clear_current_conversation():
  979. """清除用户的当前对话映射,强制下次创建新对话"""
  980. try:
  981. # 从认证中间件获取用户信息
  982. user_info = getattr(request, 'user_info', {})
  983. user_id = user_info.get('db_user_id') # 使用数据库用户ID
  984. if not user_id:
  985. return jsonify({
  986. 'success': False,
  987. 'error': '用户信息获取失败'
  988. }), HTTPStatus.UNAUTHORIZED
  989. # 清除用户的当前对话映射
  990. if str(user_id) in user_current_conversations:
  991. del user_current_conversations[str(user_id)]
  992. return jsonify({
  993. 'success': True,
  994. 'message': '当前对话映射已清除'
  995. })
  996. except Exception as e:
  997. logger.error(f"清除当前对话映射时发生错误: {str(e)}", exc_info=True)
  998. return jsonify({
  999. 'success': False,
  1000. 'error': f"发生错误: {str(e)}"
  1001. }), HTTPStatus.INTERNAL_SERVER_ERROR
  1002. @app.route('/api/conversations/<int:conversation_id>/switch', methods=['POST'])
  1003. @auth_middleware.login_required
  1004. def switch_conversation(conversation_id):
  1005. """切换到指定对话"""
  1006. try:
  1007. # 从认证中间件获取用户信息
  1008. user_info = getattr(request, 'user_info', {})
  1009. user_id = user_info.get('db_user_id') # 使用数据库用户ID
  1010. if not user_id:
  1011. return jsonify({
  1012. 'success': False,
  1013. 'error': '用户信息获取失败'
  1014. }), HTTPStatus.UNAUTHORIZED
  1015. # 切换到指定对话
  1016. conversation = switch_to_conversation(user_id, conversation_id)
  1017. return jsonify({
  1018. 'success': True,
  1019. 'message': '对话切换成功',
  1020. 'conversation': {
  1021. 'conversation_id': conversation.conversation_id,
  1022. 'title': conversation.title,
  1023. 'session_id': conversation.session_id,
  1024. 'message_count': conversation.message_count,
  1025. 'last_message_at': conversation.last_message_at.isoformat() if conversation.last_message_at else None,
  1026. 'created_at': conversation.created_at.isoformat(),
  1027. 'updated_at': conversation.updated_at.isoformat()
  1028. }
  1029. })
  1030. except ValueError as e:
  1031. return jsonify({
  1032. 'success': False,
  1033. 'error': str(e)
  1034. }), HTTPStatus.NOT_FOUND
  1035. except Exception as e:
  1036. logger.error(f"切换对话时发生错误: {str(e)}", exc_info=True)
  1037. return jsonify({
  1038. 'success': False,
  1039. 'error': f"发生错误: {str(e)}"
  1040. }), HTTPStatus.INTERNAL_SERVER_ERROR
  1041. @app.route('/api/conversations/<int:conversation_id>/status', methods=['GET'])
  1042. @auth_middleware.login_required
  1043. def get_conversation_status(conversation_id):
  1044. """获取对话的实时状态信息"""
  1045. try:
  1046. # 从认证中间件获取用户信息
  1047. user_info = getattr(request, 'user_info', {})
  1048. user_id = user_info.get('db_user_id') # 使用数据库用户ID
  1049. if not user_id:
  1050. return jsonify({
  1051. 'success': False,
  1052. 'error': '用户信息获取失败'
  1053. }), HTTPStatus.UNAUTHORIZED
  1054. # 验证对话是否属于当前用户
  1055. conversation = Conversation.query.filter_by(
  1056. conversation_id=conversation_id,
  1057. user_id=user_id,
  1058. is_deleted=False
  1059. ).first()
  1060. if not conversation:
  1061. return jsonify({
  1062. 'success': False,
  1063. 'error': '对话不存在或无权限访问'
  1064. }), HTTPStatus.NOT_FOUND
  1065. # 获取对话的最新消息
  1066. latest_message = Message.query.filter_by(
  1067. conversation_id=conversation_id,
  1068. is_deleted=False
  1069. ).order_by(Message.message_order.desc()).first()
  1070. status_info = {
  1071. 'conversation_id': conversation_id,
  1072. 'title': conversation.title,
  1073. 'message_count': conversation.message_count,
  1074. 'last_message_at': conversation.last_message_at.isoformat() if conversation.last_message_at else None,
  1075. 'updated_at': conversation.updated_at.isoformat(),
  1076. 'has_latest_message': latest_message is not None,
  1077. 'latest_message_role': latest_message.role if latest_message else None,
  1078. 'latest_message_preview': latest_message.content[:100] + '...' if latest_message and len(latest_message.content) > 100 else latest_message.content if latest_message else None
  1079. }
  1080. return jsonify({
  1081. 'success': True,
  1082. 'status': status_info
  1083. })
  1084. except Exception as e:
  1085. logger.error(f"获取对话状态时发生错误: {str(e)}", exc_info=True)
  1086. return jsonify({
  1087. 'success': False,
  1088. 'error': f"发生错误: {str(e)}"
  1089. }), HTTPStatus.INTERNAL_SERVER_ERROR
  1090. @app.route('/api/messages/<int:message_id>/edit', methods=['POST'])
  1091. @auth_middleware.login_required
  1092. @rate_limit(config.rate_limit_requests, config.rate_limit_window)
  1093. def edit_message(message_id):
  1094. """编辑用户消息并重新生成AI回复(流式响应)"""
  1095. try:
  1096. # 从请求中获取数据
  1097. data = request.json or {}
  1098. new_content = data.get('content', '')
  1099. agent = data.get('agent') or 'kuaishou'
  1100. # 输入验证
  1101. new_content, error = sanitize_input(new_content, config.max_message_length)
  1102. if error:
  1103. # 流式响应中的错误返回
  1104. return Response(
  1105. f"data: {json.dumps({'success': False, 'error': error, 'done': True})}\n\n",
  1106. content_type='text/event-stream'
  1107. )
  1108. # 从认证中间件获取用户信息
  1109. auth_user_info = getattr(request, 'user_info', {})
  1110. auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID
  1111. if not auth_user_id:
  1112. return Response(
  1113. f"data: {json.dumps({'success': False, 'error': '用户信息获取失败', 'done': True})}\n\n",
  1114. content_type='text/event-stream'
  1115. )
  1116. # 获取要编辑的消息
  1117. message = Message.query.filter_by(
  1118. message_id=message_id,
  1119. is_deleted=False
  1120. ).first()
  1121. if not message:
  1122. return Response(
  1123. f"data: {json.dumps({'success': False, 'error': '消息不存在', 'done': True})}\n\n",
  1124. content_type='text/event-stream'
  1125. )
  1126. # 验证消息是否是用户消息
  1127. if message.role != 'user':
  1128. return Response(
  1129. f"data: {json.dumps({'success': False, 'error': '只能编辑用户消息', 'done': True})}\n\n",
  1130. content_type='text/event-stream'
  1131. )
  1132. # 获取对话信息
  1133. conversation = Conversation.query.filter_by(
  1134. conversation_id=message.conversation_id,
  1135. user_id=auth_user_id,
  1136. is_deleted=False
  1137. ).first()
  1138. if not conversation:
  1139. return Response(
  1140. f"data: {json.dumps({'success': False, 'error': '对话不存在或无权限访问', 'done': True})}\n\n",
  1141. content_type='text/event-stream'
  1142. )
  1143. # 记录请求信息
  1144. log_request_info(new_content, str(auth_user_id), 'POST')
  1145. # 选择对应智能体 APP_ID
  1146. chosen_app_id, appid_env_key = config.get_app_id_for_agent_with_source(agent)
  1147. if not chosen_app_id:
  1148. logger.error(f"未找到对应智能体APP_ID,agent={agent}")
  1149. return Response(
  1150. f"data: {json.dumps({'success': False, 'error': f'服务器未配置该智能体的 APP_ID: {agent}', 'done': True})}\n\n",
  1151. content_type='text/event-stream'
  1152. )
  1153. # 使用对话的session_id
  1154. api_session_id = conversation.session_id
  1155. # ===== 立即软删除当前消息及之后的所有消息(在流式响应开始前) =====
  1156. try:
  1157. # 计算将被删除的消息数(包括当前消息及之后的所有消息)
  1158. deleted_count = Message.query.filter_by(
  1159. conversation_id=conversation.conversation_id,
  1160. is_deleted=False
  1161. ).filter(
  1162. Message.message_order >= message.message_order
  1163. ).count()
  1164. # 软删除当前消息及之后的所有消息(包括原来的AI回复等)
  1165. Message.query.filter_by(
  1166. conversation_id=conversation.conversation_id,
  1167. is_deleted=False
  1168. ).filter(
  1169. Message.message_order >= message.message_order
  1170. ).update({
  1171. 'is_deleted': True,
  1172. 'deleted_at': get_beijing_time()
  1173. }, synchronize_session=False)
  1174. # 更新消息计数:减去被删除的消息数
  1175. conversation.message_count -= deleted_count
  1176. db.session.commit()
  1177. # 保存编辑后的用户消息(使用相同的message_order)
  1178. new_user_message = save_message_to_database(
  1179. auth_user_id,
  1180. conversation.conversation_id,
  1181. 'user',
  1182. new_content,
  1183. None,
  1184. message_order=message.message_order
  1185. )
  1186. except Exception as db_error:
  1187. db.session.rollback()
  1188. logger.error(f"软删除消息或保存编辑消息时发生错误: {str(db_error)}", exc_info=True)
  1189. return Response(
  1190. f"data: {json.dumps({'success': False, 'error': f'更新消息失败: {str(db_error)}', 'done': True})}\n\n",
  1191. content_type='text/event-stream'
  1192. )
  1193. # 获取该消息之前的所有消息(不包括已删除的消息,现在数据库中已删除的消息不会出现在查询结果中)
  1194. previous_messages = Message.query.filter_by(
  1195. conversation_id=message.conversation_id,
  1196. is_deleted=False
  1197. ).filter(
  1198. Message.message_order < message.message_order
  1199. ).order_by(Message.message_order.asc()).all()
  1200. # 构建对话历史(使用编辑后的消息)
  1201. messages = []
  1202. for msg in previous_messages:
  1203. messages.append({"role": msg.role, "content": msg.content})
  1204. # 添加编辑后的用户消息(刚保存的)
  1205. messages.append({"role": "user", "content": new_content})
  1206. # 截断对话历史以控制内存使用
  1207. messages = truncate_conversation_history(messages, config.max_conversation_history)
  1208. # 定义流式响应生成器
  1209. def generate() -> Generator[str, None, None]:
  1210. try:
  1211. # 发送初始响应
  1212. initial_data = {
  1213. 'success': True,
  1214. 'chunk': '',
  1215. 'done': False,
  1216. 'agent': agent,
  1217. 'message_id': message_id
  1218. }
  1219. yield f"data: {json.dumps(initial_data)}\n\n"
  1220. try:
  1221. # 使用DashScope SDK直接调用 - 多轮对话模式
  1222. params = {
  1223. "api_key": config.api_key,
  1224. "app_id": chosen_app_id,
  1225. "messages": messages, # 发送完整的对话历史
  1226. "stream": True,
  1227. "incremental_output": True,
  1228. "has_thoughts": True
  1229. }
  1230. # 使用对话的session_id
  1231. if api_session_id:
  1232. params["session_id"] = api_session_id
  1233. response = Application.call(**params)
  1234. new_session_id: Optional[str] = None
  1235. accumulated_text: str = ""
  1236. thoughts_content: List[str] = []
  1237. # 处理流式响应
  1238. for chunk in response:
  1239. # 检查客户端是否断开连接
  1240. if request.environ.get('werkzeug.server.shutdown') or not request:
  1241. break
  1242. if chunk.status_code != HTTPStatus.OK:
  1243. yield f"data: {json.dumps({'success': False, 'error': f'错误: {chunk.status_code} - {chunk.message}', 'request_id': chunk.request_id, 'done': True, 'agent': agent})}\n\n"
  1244. return
  1245. # 处理思考过程
  1246. if hasattr(chunk.output, 'thoughts') and chunk.output.thoughts:
  1247. for it in chunk.output.thoughts:
  1248. if it.action_type == 'reasoning':
  1249. content = str(it.thought) if not isinstance(it.thought, str) else it.thought
  1250. thoughts_content.append(content)
  1251. yield f"data: {json.dumps({'success': True, 'thought': content, 'done': False, 'agent': agent})}\n\n"
  1252. # 处理回答内容
  1253. if hasattr(chunk.output, 'text') and chunk.output.text:
  1254. current_text = str(chunk.output.text)
  1255. accumulated_text += current_text
  1256. yield f"data: {json.dumps({'success': True, 'chunk': current_text, 'done': False, 'agent': agent})}\n\n"
  1257. # 检查是否有会话ID
  1258. if hasattr(chunk.output, 'session_id') and chunk.output.session_id:
  1259. new_session_id = chunk.output.session_id
  1260. # 处理结束,保存AI回复到数据库
  1261. try:
  1262. # 更新session_id(如果需要)
  1263. if new_session_id and new_session_id != api_session_id:
  1264. conversation.session_id = new_session_id
  1265. db.session.commit()
  1266. # 保存新的AI回复(如果有)
  1267. if accumulated_text:
  1268. thinking_text = '\n'.join(thoughts_content) if thoughts_content else None
  1269. ai_message = save_message_to_database(
  1270. auth_user_id,
  1271. conversation.conversation_id,
  1272. 'assistant',
  1273. accumulated_text,
  1274. thinking_text,
  1275. message_order=message.message_order + 1
  1276. )
  1277. # save_message_to_database 已经提交,message_count 已更新
  1278. else:
  1279. # 如果没有AI回复,更新对话的更新时间
  1280. conversation.updated_at = get_beijing_time()
  1281. conversation.last_message_at = get_beijing_time()
  1282. db.session.commit()
  1283. except Exception as db_error:
  1284. db.session.rollback()
  1285. logger.error(f"保存AI回复到数据库时发生错误: {str(db_error)}", exc_info=True)
  1286. # 不影响主要功能,继续返回结果
  1287. # 发送完成标记
  1288. final_data = {
  1289. 'success': True,
  1290. 'done': True,
  1291. 'thoughts': thoughts_content,
  1292. 'agent': agent,
  1293. 'session_id': new_session_id or api_session_id,
  1294. 'message_id': message_id,
  1295. 'conversation_id': conversation.conversation_id
  1296. }
  1297. yield f"data: {json.dumps(final_data)}\n\n"
  1298. except Exception as e:
  1299. logger.error(f"DashScope SDK 调用错误: {str(e)}", exc_info=True)
  1300. yield f"data: {json.dumps({'success': False, 'error': f'DashScope API调用错误: {str(e)}', 'done': True, 'agent': agent})}\n\n"
  1301. except Exception as e:
  1302. logger.error(f"流式API调用错误: {str(e)}", exc_info=True)
  1303. yield f"data: {json.dumps({'success': False, 'error': f'发生错误: {str(e)}', 'done': True, 'agent': agent})}\n\n"
  1304. # 创建并返回流式响应
  1305. response = Response(stream_with_context(generate()), content_type='text/event-stream')
  1306. response.headers['Cache-Control'] = 'no-cache'
  1307. response.headers['X-Accel-Buffering'] = 'no'
  1308. return response
  1309. except Exception as e:
  1310. logger.error(f"编辑消息时发生错误: {str(e)}", exc_info=True)
  1311. return jsonify({
  1312. 'success': False,
  1313. 'error': f"发生错误: {str(e)}"
  1314. }), HTTPStatus.INTERNAL_SERVER_ERROR
  1315. # 健康检查端点
  1316. @app.route('/health', methods=['GET'])
  1317. def health_check():
  1318. """健康检查端点"""
  1319. return jsonify({
  1320. 'status': 'ok',
  1321. 'app_version': '2.0.0',
  1322. 'python_version': '3.12',
  1323. 'timestamp': time.time()
  1324. })
  1325. if __name__ == '__main__':
  1326. app.run(host='0.0.0.0', port=config.port, debug=config.debug)