| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570 |
- import os
- import json
- import uuid
- import time
- import logging
- from typing import Dict, Any, Generator, Optional, List
- from flask_cors import CORS
- import requests
- from flask import Flask, request, jsonify, render_template, session, Response, stream_with_context, redirect
- from http import HTTPStatus
- from dotenv import load_dotenv, find_dotenv
- from dashscope import Application
- from datetime import datetime
- # 先加载环境变量,确保后续模块读取到 .env(支持不同工作目录)
- load_dotenv(find_dotenv(), override=False)
- # 导入认证模块
- from auth_middleware import init_auth
- from auth_routes import auth_bp
- # 导入配置和工具
- from config import config
- from utils import (
- sanitize_input, validate_user_id, rate_limit,
- truncate_conversation_history, log_request_info, get_beijing_time,
- generate_session_id
- )
- # 导入数据库模型
- from models import init_database, db, User, Conversation, Message, get_or_create_user
- # 配置日志
- logging.basicConfig(
- level=getattr(logging, config.log_level),
- format=config.log_format
- )
- logger = logging.getLogger(__name__)
- # 创建Flask应用
- app = Flask(__name__)
- CORS(app) # 启用CORS支持
- # 配置session
- app.secret_key = config.secret_key
- app.config['PERMANENT_SESSION_LIFETIME'] = 7 * 24 * 60 * 60 # 7天
- app.config['SESSION_COOKIE_SECURE'] = False # 开发环境设为False
- app.config['SESSION_COOKIE_HTTPONLY'] = True
- app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
- # 配置数据库
- app.config['SQLALCHEMY_DATABASE_URI'] = config.get_database_url()
- app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
- app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
- 'pool_pre_ping': True,
- 'pool_recycle': 300,
- 'pool_timeout': 20,
- 'max_overflow': 0
- }
- # 初始化数据库
- try:
- init_database(app)
- except Exception as e:
- logger.error(f"数据库初始化失败: {str(e)}", exc_info=True)
- raise
- # 初始化认证中间件
- auth_middleware = init_auth(app)
- app.auth_middleware = auth_middleware
- # 注册认证蓝图
- app.register_blueprint(auth_bp)
- # 存储用户当前对话ID的字典(保留用于兼容性)
- user_current_conversations: Dict[str, int] = {}
- @app.after_request
- def add_cache_control_headers(response):
- """
- 防止首页HTML被中间层或浏览器长期缓存,避免出现"旧HTML引用新/旧静态资源"导致白屏。
- 静态资源保持默认缓存策略,仅对HTML页面禁缓存。
- """
- content_type = (response.content_type or "").lower()
- if "text/html" in content_type:
- response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
- response.headers["Pragma"] = "no-cache"
- response.headers["Expires"] = "0"
- return response
- def save_message_to_database(user_id: int, conversation_id: int, role: str, content: str,
- thinking_content: str = None, message_order: int = 1) -> Message:
- """保存消息到数据库"""
- try:
- message = Message(
- conversation_id=conversation_id,
- role=role,
- content=content,
- thinking_content=thinking_content,
- content_type='text',
- message_order=message_order
- )
-
- db.session.add(message)
-
- # 更新对话的消息计数和最后消息时间
- conversation = Conversation.query.get(conversation_id)
- if conversation:
- conversation.message_count += 1
- conversation.last_message_at = get_beijing_time()
- conversation.updated_at = get_beijing_time()
-
- # 一次性提交所有更改
- db.session.commit()
-
- return message
-
- except Exception as e:
- db.session.rollback()
- logger.error(f"保存消息到数据库时发生错误: {str(e)}", exc_info=True)
- raise
- def get_or_create_conversation(user_id: int, title: str = None, conversation_id: int = None, force_new: bool = False) -> Conversation:
- """获取或创建对话"""
- try:
- # 如果强制创建新对话,直接创建
- if force_new:
- if not title:
- title = f"对话 {get_beijing_time().strftime('%Y-%m-%d %H:%M')}"
-
- # 生成新的session_id
- session_id = generate_session_id(user_id)
-
- conversation = Conversation(
- user_id=user_id,
- title=title,
- conversation_type='chat',
- session_id=session_id,
- message_count=0
- )
-
- db.session.add(conversation)
- db.session.commit()
-
- # 更新用户的当前对话ID
- user_current_conversations[str(user_id)] = conversation.conversation_id
-
-
- return conversation
-
- # 如果指定了conversation_id,验证并返回该对话
- if conversation_id:
- conversation = Conversation.query.filter_by(
- conversation_id=conversation_id,
- user_id=user_id,
- is_deleted=False
- ).first()
- if conversation:
- # 如果对话没有session_id,生成一个
- if not conversation.session_id:
- conversation.session_id = generate_session_id(user_id, conversation_id)
- db.session.commit()
-
- # 更新用户的当前对话ID
- user_current_conversations[str(user_id)] = conversation.conversation_id
- return conversation
-
- # 检查用户是否有当前对话映射
- current_conv_id = user_current_conversations.get(str(user_id))
-
- if current_conv_id:
- conversation = Conversation.query.filter_by(
- conversation_id=current_conv_id,
- user_id=user_id,
- is_deleted=False
- ).first()
- if conversation:
- # 如果对话没有session_id,生成一个
- if not conversation.session_id:
- conversation.session_id = generate_session_id(user_id, current_conv_id)
- db.session.commit()
- return conversation
-
- # 创建新对话
- if not title:
- title = f"对话 {get_beijing_time().strftime('%Y-%m-%d %H:%M')}"
-
- # 生成新的session_id
- session_id = generate_session_id(user_id)
-
- conversation = Conversation(
- user_id=user_id,
- title=title,
- conversation_type='chat',
- session_id=session_id,
- message_count=0
- )
-
- db.session.add(conversation)
- db.session.commit()
-
- # 更新用户的当前对话ID
- user_current_conversations[str(user_id)] = conversation.conversation_id
-
- return conversation
-
- except Exception as e:
- db.session.rollback()
- logger.error(f"获取或创建对话时发生错误: {str(e)}", exc_info=True)
- raise
- def switch_to_conversation(user_id: int, conversation_id: int) -> Conversation:
- """切换到指定对话"""
- try:
- # 验证对话是否属于当前用户
- conversation = Conversation.query.filter_by(
- conversation_id=conversation_id,
- user_id=user_id,
- is_deleted=False
- ).first()
-
- if not conversation:
- raise ValueError(f"对话 {conversation_id} 不存在或无权限访问")
-
- # 如果对话没有session_id,生成一个
- if not conversation.session_id:
- conversation.session_id = generate_session_id(user_id, conversation_id)
- db.session.commit()
-
-
- # 更新用户的当前对话ID
- user_current_conversations[str(user_id)] = conversation.conversation_id
-
-
- return conversation
-
- except Exception as e:
- logger.error(f"切换对话时发生错误: {str(e)}", exc_info=True)
- raise
- def generate_conversation_title(user_message: str) -> str:
- """根据用户消息生成对话标题"""
- # 取用户消息的前50个字符作为标题
- title = user_message.strip()[:50]
- if len(user_message) > 50:
- title += "..."
- return title
- @app.route('/')
- def index():
- """渲染聊天界面"""
- # 检查是否是登录成功跳转
- login_success = request.args.get('login_success')
- token = request.args.get('token')
- user_info_json = request.args.get('user_info')
-
- if login_success and token and user_info_json:
- # 登录成功,处理用户信息
- try:
- import urllib.parse
- import json
- user_info = json.loads(urllib.parse.unquote(user_info_json))
-
- # 保存到session
- session['auth_token'] = token
- session['user_info'] = user_info
-
- # 将用户信息传递给模板
- return render_template('index.html', user_info=user_info, jwt_token=token)
- except Exception as e:
- logger.error(f'处理登录成功参数失败: {e}')
- # 如果处理失败,重定向到登录页面
- return redirect('/login')
- else:
- # 正常访问,检查认证状态
- # 优先从session获取
- user_info = session.get('user_info', {})
- jwt_token = session.get('auth_token', '')
-
- # 如果session中没有,尝试从localStorage获取(通过JavaScript)
- if not jwt_token:
- # 返回一个页面,让JavaScript检查localStorage
- return render_template('index.html', user_info={}, jwt_token='')
-
- # 将用户信息传递给模板
- return render_template('index.html', user_info=user_info, jwt_token=jwt_token)
- @app.route('/login')
- def login():
- """登录页面"""
- return render_template('login.html')
- @app.route('/login_error')
- def login_error():
- """登录错误页面"""
- error_type = request.args.get('error', 'unknown')
- error_message = request.args.get('message', '')
- error_code = request.args.get('code', '')
- return render_template('login_error.html',
- error_type=error_type,
- error_message=error_message,
- error_code=error_code)
- @app.route('/api/chat', methods=['POST'])
- @auth_middleware.login_required
- @rate_limit(config.rate_limit_requests, config.rate_limit_window)
- def chat():
- """处理聊天请求并调用阿里云百炼API (非流式响应)"""
- try:
- data = request.json or {}
- message = data.get('message', '')
- user_id = data.get('user_id', str(uuid.uuid4()))
- agent = data.get('agent') or 'kuaishou'
- conversation_id = data.get('conversation_id') # 新增:支持指定对话ID
- session_id = data.get('session_id') # 新增:支持指定session_id
- force_new = data.get('force_new', False) # 新增:强制创建新对话
-
- # 输入验证
- message, error = sanitize_input(message, config.max_message_length)
- if error:
- return jsonify({
- 'success': False,
- 'error': error
- }), HTTPStatus.BAD_REQUEST
-
- user_id, error = validate_user_id(user_id)
- if error:
- return jsonify({
- 'success': False,
- 'error': error
- }), HTTPStatus.BAD_REQUEST
-
- # 从认证中间件获取用户信息
- auth_user_info = getattr(request, 'user_info', {})
- auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID
-
- # 选择对应智能体 APP_ID
- chosen_app_id, appid_env_key = config.get_app_id_for_agent_with_source(agent)
- if not chosen_app_id:
- logger.error(f"未找到对应智能体APP_ID,agent={agent}")
- return jsonify({
- 'success': False,
- 'error': f"服务器未配置该智能体的 APP_ID: {agent}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
-
- # 记录请求信息
- log_request_info(message, user_id, 'POST')
-
- # 获取或创建对话
- conversation = get_or_create_conversation(
- auth_user_id,
- generate_conversation_title(message),
- conversation_id,
- force_new
- )
-
- # 优先使用前端传来的session_id,否则使用对话的session_id
- api_session_id = session_id or conversation.session_id
-
- # 准备API请求参数
- headers = {
- "Authorization": f"Bearer {config.api_key}",
- "Content-Type": "application/json",
- "X-DashScope-SSE": "disable" # 禁用流式输出
- }
-
- # 从数据库获取对话历史
- messages_query = Message.query.filter_by(
- conversation_id=conversation.conversation_id,
- is_deleted=False
- ).order_by(Message.message_order.asc()).all()
-
- # 构建多轮对话历史
- messages = []
- for msg in messages_query:
- messages.append({"role": msg.role, "content": msg.content})
-
- # 添加用户新消息到对话历史
- messages.append({"role": "user", "content": message})
-
- # 截断对话历史以控制内存使用
- messages = truncate_conversation_history(messages, config.max_conversation_history)
-
- payload = {
- "model": "application",
- "parameters": {"app_id": chosen_app_id},
- "input": {"messages": messages} # 发送完整的对话历史
- }
-
- # 如果有会话ID,添加到请求参数中
- if api_session_id:
- payload["parameters"]["session_id"] = api_session_id
-
- # 发送API请求
- response = requests.post(
- config.api_endpoint,
- headers=headers,
- json=payload,
- timeout=60
- )
-
- # 转换为JSON
- response_json = response.json()
-
- if response.status_code != HTTPStatus.OK:
- return jsonify({
- 'success': False,
- 'error': f"错误: {response.status_code} - {response_json.get('message', '未知错误')}",
- 'request_id': response_json.get('request_id', ''),
- 'agent': agent
- }), HTTPStatus.INTERNAL_SERVER_ERROR
-
- # 解析返回结果
- output = response_json.get('output', {})
- choices = output.get('choices', [{}])
- text = choices[0].get('message', {}).get('content', '') if choices else ''
- new_session_id = output.get('session_id', '')
- thoughts_content = output.get('thoughts', [])
-
- # 更新对话的session_id(如果API返回了新的session_id且与当前不同)
- if new_session_id and new_session_id != api_session_id:
- conversation.session_id = new_session_id
- db.session.commit()
-
- elif not api_session_id and new_session_id:
- # 如果对话原本没有session_id,现在有了,更新数据库
- conversation.session_id = new_session_id
- db.session.commit()
-
-
- # 保存到数据库
- try:
- # 获取当前对话的消息数量,用于设置消息顺序
- current_message_count = Message.query.filter_by(
- conversation_id=conversation.conversation_id,
- is_deleted=False
- ).count()
-
- # 保存用户消息
- save_message_to_database(
- auth_user_id,
- conversation.conversation_id,
- 'user',
- message,
- message_order=current_message_count + 1
- )
-
- # 保存AI回复
- if text:
- thinking_text = '\n'.join(thoughts_content) if thoughts_content else None
- save_message_to_database(
- auth_user_id,
- conversation.conversation_id,
- 'assistant',
- text,
- thinking_text,
- message_order=current_message_count + 2
- )
-
- # 更新对话标题(如果是第一条消息)
- if conversation.message_count <= 2: # 用户消息 + AI回复
- new_title = generate_conversation_title(message)
- # 检查是否是默认标题("新对话"或"对话 时间"格式)
- is_default_title = (
- conversation.title == "新对话" or
- conversation.title == f"对话 {conversation.created_at.strftime('%Y-%m-%d %H:%M')}"
- )
- if is_default_title:
- conversation.title = new_title
- db.session.commit()
-
- except Exception as db_error:
- logger.error(f"保存对话到数据库时发生错误: {str(db_error)}", exc_info=True)
- # 不影响主要功能,继续返回结果
-
- return jsonify({
- 'success': True,
- 'message': text,
- 'user_id': user_id,
- 'session_id': new_session_id or api_session_id,
- 'thoughts': thoughts_content,
- 'agent': agent,
- 'conversation_id': conversation.conversation_id
- })
-
- except requests.exceptions.RequestException as e:
- logger.error(f"API请求异常: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"API请求异常: {str(e)}"
- }), HTTPStatus.SERVICE_UNAVAILABLE
- except Exception as e:
- logger.error(f"API调用错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/chat/stream', methods=['GET', 'POST'])
- @auth_middleware.login_required
- @rate_limit(config.rate_limit_requests, config.rate_limit_window)
- def chat_stream():
- """处理聊天请求并以流式方式返回响应"""
- try:
- # 从请求中获取数据
- if request.method == 'GET':
- message = request.args.get('message', '')
- user_id = request.args.get('user_id', str(uuid.uuid4()))
- agent = request.args.get('agent') or 'kuaishou'
- conversation_id = request.args.get('conversation_id') # 支持指定对话ID
- else: # POST
- data = request.json or {}
- message = data.get('message', '')
- user_id = data.get('user_id', str(uuid.uuid4()))
- agent = data.get('agent') or 'kuaishou'
- conversation_id = data.get('conversation_id') # 支持指定对话ID
- session_id = data.get('session_id') # 支持指定session_id
- force_new = data.get('force_new', False) # 强制创建新对话
-
- # 输入验证
- message, error = sanitize_input(message, config.max_message_length)
- if error:
- return Response(
- f"data: {json.dumps({'success': False, 'error': error, 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- user_id, error = validate_user_id(user_id)
- if error:
- return Response(
- f"data: {json.dumps({'success': False, 'error': error, 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 选择对应智能体 APP_ID
- chosen_app_id, appid_env_key = config.get_app_id_for_agent_with_source(agent)
- if not chosen_app_id:
- logger.error(f"未找到对应智能体APP_ID,agent={agent}")
- return Response(
- f"data: {json.dumps({'success': False, 'error': f'服务器未配置该智能体的 APP_ID: {agent}', 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 记录请求信息
- log_request_info(message, user_id, request.method)
-
- # 从认证中间件获取用户信息
- auth_user_info = getattr(request, 'user_info', {})
- auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID
-
- # 获取或创建对话
- conversation = get_or_create_conversation(
- auth_user_id,
- generate_conversation_title(message),
- conversation_id,
- force_new
- )
-
- # 优先使用前端传来的session_id,否则使用对话的session_id
- api_session_id = session_id or conversation.session_id
-
- # 从数据库获取对话历史
- messages_query = Message.query.filter_by(
- conversation_id=conversation.conversation_id,
- is_deleted=False
- ).order_by(Message.message_order.asc()).all()
-
- # 构建多轮对话历史
- messages = []
- for msg in messages_query:
- messages.append({"role": msg.role, "content": msg.content})
-
- # 添加用户新消息到对话历史
- messages.append({"role": "user", "content": message})
-
- # 截断对话历史以控制内存使用
- messages = truncate_conversation_history(messages, config.max_conversation_history)
-
- # 立即保存用户消息到数据库(在流式响应开始前)
- try:
- # 获取当前对话的消息数量,用于设置消息顺序
- current_message_count = Message.query.filter_by(
- conversation_id=conversation.conversation_id,
- is_deleted=False
- ).count()
-
- # 立即保存用户消息
- user_message = save_message_to_database(
- auth_user_id,
- conversation.conversation_id,
- 'user',
- message,
- message_order=current_message_count + 1
- )
-
-
- except Exception as db_error:
- logger.error(f"保存用户消息到数据库时发生错误: {str(db_error)}", exc_info=True)
- return Response(
- f"data: {json.dumps({'success': False, 'error': f'保存用户消息失败: {str(db_error)}', 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 定义流式响应生成器
- def generate() -> Generator[str, None, None]:
- try:
- # 发送初始响应,包含用户消息ID,防止浏览器超时
- initial_data = {
- 'success': True,
- 'chunk': '',
- 'done': False,
- 'agent': agent,
- 'user_message_id': user_message.message_id if user_message else None
- }
- yield f"data: {json.dumps(initial_data)}\n\n"
-
- try:
- # 使用DashScope SDK直接调用 - 多轮对话模式
- params = {
- "api_key": config.api_key,
- "app_id": chosen_app_id,
- "messages": messages, # 发送完整的对话历史
- "stream": True,
- "incremental_output": True,
- "has_thoughts": True
- }
-
- # 使用对话的session_id
- if api_session_id:
- params["session_id"] = api_session_id
-
- response = Application.call(**params)
-
- new_session_id: Optional[str] = None
- accumulated_text: str = ""
- thoughts_content: List[str] = []
-
- # 处理流式响应
- for chunk in response:
- # 检查客户端是否断开连接
- if request.environ.get('werkzeug.server.shutdown') or not request:
- break
-
- if chunk.status_code != HTTPStatus.OK:
- 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"
- return
-
- # 处理思考过程
- if hasattr(chunk.output, 'thoughts') and chunk.output.thoughts:
- for it in chunk.output.thoughts:
- if it.action_type == 'reasoning':
- content = str(it.thought) if not isinstance(it.thought, str) else it.thought
- thoughts_content.append(content)
- yield f"data: {json.dumps({'success': True, 'thought': content, 'done': False, 'agent': agent})}\n\n"
-
- # 处理回答内容
- if hasattr(chunk.output, 'text') and chunk.output.text:
- current_text = str(chunk.output.text)
- accumulated_text += current_text
- yield f"data: {json.dumps({'success': True, 'chunk': current_text, 'done': False, 'agent': agent})}\n\n"
-
- # 检查是否有会话ID
- if hasattr(chunk.output, 'session_id') and chunk.output.session_id:
- new_session_id = chunk.output.session_id
-
- # 处理结束
- if new_session_id and new_session_id != api_session_id:
- conversation.session_id = new_session_id
- db.session.commit()
- elif not api_session_id and new_session_id:
- # 如果对话原本没有session_id,现在有了,更新数据库
- conversation.session_id = new_session_id
- db.session.commit()
-
- # 保存AI回复到数据库
- try:
- if accumulated_text:
- thinking_text = '\n'.join(thoughts_content) if thoughts_content else None
- ai_message = save_message_to_database(
- auth_user_id,
- conversation.conversation_id,
- 'assistant',
- accumulated_text,
- thinking_text,
- message_order=current_message_count + 2
- )
-
- # 更新对话标题(如果是第一条消息)
- if conversation.message_count <= 2: # 用户消息 + AI回复
- new_title = generate_conversation_title(message)
- # 检查是否是默认标题("新对话"或"对话 时间"格式)
- is_default_title = (
- conversation.title == "新对话" or
- conversation.title == f"对话 {conversation.created_at.strftime('%Y-%m-%d %H:%M')}"
- )
- if is_default_title:
- conversation.title = new_title
- db.session.commit()
-
- except Exception as db_error:
- logger.error(f"保存AI回复到数据库时发生错误: {str(db_error)}", exc_info=True)
- # 不影响主要功能,继续返回结果
-
- # 发送完成标记
- final_data = {
- 'success': True,
- 'done': True,
- 'thoughts': thoughts_content,
- 'agent': agent,
- 'session_id': new_session_id or api_session_id,
- 'user_id': user_id,
- 'conversation_id': conversation.conversation_id
- }
-
- yield f"data: {json.dumps(final_data)}\n\n"
-
- except Exception as e:
- logger.error(f"DashScope SDK 调用错误: {str(e)}", exc_info=True)
- yield f"data: {json.dumps({'success': False, 'error': f'DashScope API调用错误: {str(e)}', 'done': True, 'agent': agent})}\n\n"
-
- except Exception as e:
- logger.error(f"流式API调用错误: {str(e)}", exc_info=True)
- yield f"data: {json.dumps({'success': False, 'error': f'发生错误: {str(e)}', 'done': True, 'agent': agent})}\n\n"
-
- # 创建并返回流式响应
- response = Response(stream_with_context(generate()), content_type='text/event-stream')
- response.headers['Cache-Control'] = 'no-cache'
- response.headers['X-Accel-Buffering'] = 'no'
- return response
-
- except Exception as e:
- logger.error(f"请求处理错误: {str(e)}", exc_info=True)
- response = Response(
- f"data: {json.dumps({'success': False, 'error': f'发生错误: {str(e)}', 'done': True})}\n\n",
- content_type='text/event-stream'
- )
- response.headers['Cache-Control'] = 'no-cache'
- return response
- @app.route('/api/reset', methods=['POST'])
- @auth_middleware.login_required
- @rate_limit(10, 60)
- def reset_session():
- """重置用户会话"""
- try:
- data = request.json or {}
- user_id = data.get('user_id')
-
- if not user_id:
- return jsonify({
- 'success': True,
- 'message': '没有提供用户ID,但重置请求已处理'
- })
-
- # 验证用户ID格式
- user_id, error = validate_user_id(user_id)
- if error:
- return jsonify({
- 'success': False,
- 'error': error
- }), HTTPStatus.BAD_REQUEST
-
- # 从认证中间件获取用户信息
- auth_user_info = getattr(request, 'user_info', {})
- auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID
-
- # 清除数据库中的当前对话ID(如果有认证用户信息)
- if auth_user_id and str(auth_user_id) in user_current_conversations:
- del user_current_conversations[str(auth_user_id)]
-
- return jsonify({
- 'success': True,
- 'message': '会话已重置'
- })
- except Exception as e:
- logger.error(f"重置会话时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/status', methods=['GET'])
- @auth_middleware.login_required
- def get_status():
- """获取应用状态信息"""
- try:
- active_users = len(user_current_conversations)
- total_conversations = Conversation.query.filter_by(is_deleted=False).count()
-
- return jsonify({
- 'success': True,
- 'status': 'running',
- 'active_users': active_users,
- 'total_conversations': total_conversations,
- 'app_version': '2.0.0',
- 'python_version': '3.12',
- 'config': {
- 'max_message_length': config.max_message_length,
- 'max_conversation_history': config.max_conversation_history,
- 'rate_limit_requests': config.rate_limit_requests,
- 'rate_limit_window': config.rate_limit_window
- }
- })
- except Exception as e:
- logger.error(f"获取状态时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- # ==================== 历史记录相关API ====================
- @app.route('/api/conversations', methods=['GET'])
- @auth_middleware.login_required
- def get_conversations():
- """获取用户的历史对话列表,按时间分组"""
- try:
- # 从认证中间件获取用户信息
- user_info = getattr(request, 'user_info', {})
- user_id = user_info.get('db_user_id') # 使用数据库用户ID
-
-
- if not user_id:
- logger.warning('获取历史记录失败 - 用户ID为空')
- return jsonify({
- 'success': False,
- 'error': '用户信息获取失败'
- }), HTTPStatus.UNAUTHORIZED
-
- # 获取用户对象
- user = User.query.filter_by(user_id=user_id, is_deleted=False).first()
- if not user:
- return jsonify({
- 'success': False,
- 'error': '用户不存在'
- }), HTTPStatus.NOT_FOUND
-
- # 获取用户的对话列表,按更新时间倒序
- conversations = Conversation.query.filter_by(
- user_id=user_id,
- is_deleted=False
- ).order_by(Conversation.updated_at.desc()).all()
-
- # 按时间分组
- from datetime import datetime, timedelta
- now = get_beijing_time()
- today = now.date()
- yesterday = today - timedelta(days=1)
- week_ago = today - timedelta(days=7)
- month_ago = today - timedelta(days=30)
-
- grouped_conversations = {
- 'today': [],
- 'yesterday': [],
- 'week': [],
- 'month': []
- }
-
- for conv in conversations:
- conv_date = conv.updated_at.date()
- conv_data = {
- 'conversation_id': conv.conversation_id,
- 'title': conv.title,
- 'message_count': conv.message_count,
- 'session_id': conv.session_id,
- 'last_message_at': conv.last_message_at.isoformat() if conv.last_message_at else None,
- 'created_at': conv.created_at.isoformat(),
- 'updated_at': conv.updated_at.isoformat()
- }
-
- if conv_date == today:
- grouped_conversations['today'].append(conv_data)
- elif conv_date == yesterday:
- grouped_conversations['yesterday'].append(conv_data)
- elif conv_date >= week_ago:
- grouped_conversations['week'].append(conv_data)
- elif conv_date >= month_ago:
- grouped_conversations['month'].append(conv_data)
-
- return jsonify({
- 'success': True,
- 'conversations': grouped_conversations
- })
-
- except Exception as e:
- logger.error(f"获取对话列表时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/conversations/<int:conversation_id>', methods=['GET'])
- @auth_middleware.login_required
- def get_conversation_messages(conversation_id):
- """获取指定对话的所有消息"""
- try:
- # 从认证中间件获取用户信息
- user_info = getattr(request, 'user_info', {})
- user_id = user_info.get('db_user_id') # 使用数据库用户ID
-
- if not user_id:
- return jsonify({
- 'success': False,
- 'error': '用户信息获取失败'
- }), HTTPStatus.UNAUTHORIZED
-
- # 验证对话是否属于当前用户
- conversation = Conversation.query.filter_by(
- conversation_id=conversation_id,
- user_id=user_id,
- is_deleted=False
- ).first()
-
- if not conversation:
- return jsonify({
- 'success': False,
- 'error': '对话不存在或无权限访问'
- }), HTTPStatus.NOT_FOUND
-
- # 获取对话的所有消息,按顺序排列
- messages = Message.query.filter_by(
- conversation_id=conversation_id,
- is_deleted=False
- ).order_by(Message.message_order.asc()).all()
-
- messages_data = []
- for msg in messages:
- message_data = {
- 'message_id': msg.message_id,
- 'role': msg.role,
- 'content': msg.content,
- 'thinking_content': msg.thinking_content,
- 'content_type': msg.content_type,
- 'message_order': msg.message_order,
- 'created_at': msg.created_at.isoformat()
- }
- messages_data.append(message_data)
-
- return jsonify({
- 'success': True,
- 'conversation': {
- 'conversation_id': conversation.conversation_id,
- 'title': conversation.title,
- 'conversation_type': conversation.conversation_type,
- 'session_id': conversation.session_id,
- 'message_count': conversation.message_count,
- 'created_at': conversation.created_at.isoformat(),
- 'updated_at': conversation.updated_at.isoformat()
- },
- 'messages': messages_data
- })
-
- except Exception as e:
- logger.error(f"获取对话消息时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/conversations/<int:conversation_id>/rename', methods=['POST'])
- @auth_middleware.login_required
- def rename_conversation(conversation_id):
- """重命名对话"""
- try:
- # 从认证中间件获取用户信息
- user_info = getattr(request, 'user_info', {})
- user_id = user_info.get('db_user_id') # 使用数据库用户ID
-
- if not user_id:
- return jsonify({
- 'success': False,
- 'error': '用户信息获取失败'
- }), HTTPStatus.UNAUTHORIZED
-
- data = request.json or {}
- new_title = data.get('title', '').strip()
-
- if not new_title:
- return jsonify({
- 'success': False,
- 'error': '新标题不能为空'
- }), HTTPStatus.BAD_REQUEST
-
- if len(new_title) > 500:
- return jsonify({
- 'success': False,
- 'error': '标题长度不能超过500个字符'
- }), HTTPStatus.BAD_REQUEST
-
- # 验证对话是否属于当前用户
- conversation = Conversation.query.filter_by(
- conversation_id=conversation_id,
- user_id=user_id,
- is_deleted=False
- ).first()
-
- if not conversation:
- return jsonify({
- 'success': False,
- 'error': '对话不存在或无权限访问'
- }), HTTPStatus.NOT_FOUND
-
- # 更新标题
- conversation.title = new_title
- conversation.updated_at = get_beijing_time()
-
- db.session.commit()
-
- return jsonify({
- 'success': True,
- 'message': '对话标题已更新',
- 'conversation': {
- 'conversation_id': conversation.conversation_id,
- 'title': conversation.title,
- 'updated_at': conversation.updated_at.isoformat()
- }
- })
-
- except Exception as e:
- db.session.rollback()
- logger.error(f"重命名对话时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/conversations/<int:conversation_id>', methods=['DELETE'])
- @auth_middleware.login_required
- def delete_conversation(conversation_id):
- """删除对话(软删除)"""
- try:
- # 从认证中间件获取用户信息
- user_info = getattr(request, 'user_info', {})
- user_id = user_info.get('db_user_id') # 使用数据库用户ID
-
- if not user_id:
- return jsonify({
- 'success': False,
- 'error': '用户信息获取失败'
- }), HTTPStatus.UNAUTHORIZED
-
- # 验证对话是否属于当前用户
- conversation = Conversation.query.filter_by(
- conversation_id=conversation_id,
- user_id=user_id,
- is_deleted=False
- ).first()
-
- if not conversation:
- return jsonify({
- 'success': False,
- 'error': '对话不存在或无权限访问'
- }), HTTPStatus.NOT_FOUND
-
- # 软删除对话
- conversation.is_deleted = True
- conversation.deleted_at = get_beijing_time()
- conversation.updated_at = get_beijing_time()
-
- # 同时软删除该对话的所有消息
- Message.query.filter_by(
- conversation_id=conversation_id,
- is_deleted=False
- ).update({
- 'is_deleted': True,
- 'deleted_at': get_beijing_time()
- })
-
- db.session.commit()
-
- return jsonify({
- 'success': True,
- 'message': '对话已删除'
- })
-
- except Exception as e:
- db.session.rollback()
- logger.error(f"删除对话时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/conversations/new', methods=['POST'])
- @auth_middleware.login_required
- def create_new_conversation():
- """创建新对话"""
- try:
- # 从认证中间件获取用户信息
- user_info = getattr(request, 'user_info', {})
- user_id = user_info.get('db_user_id') # 使用数据库用户ID
-
- if not user_id:
- return jsonify({
- 'success': False,
- 'error': '用户信息获取失败'
- }), HTTPStatus.UNAUTHORIZED
-
- # 获取用户对象
- user = User.query.filter_by(user_id=user_id, is_deleted=False).first()
- if not user:
- return jsonify({
- 'success': False,
- 'error': '用户不存在'
- }), HTTPStatus.NOT_FOUND
-
- data = request.json or {}
- title = data.get('title', '新对话').strip()
-
- if not title:
- title = '新对话'
-
- if len(title) > 500:
- return jsonify({
- 'success': False,
- 'error': '标题长度不能超过500个字符'
- }), HTTPStatus.BAD_REQUEST
-
- # 生成新的session_id
- session_id = generate_session_id(user_id)
-
- # 创建新对话
- conversation = Conversation(
- user_id=user_id,
- title=title,
- conversation_type='chat',
- session_id=session_id,
- message_count=0
- )
-
- db.session.add(conversation)
- db.session.commit()
-
- # 更新用户的当前对话ID
- user_current_conversations[str(user_id)] = conversation.conversation_id
-
-
- return jsonify({
- 'success': True,
- 'message': '新对话已创建',
- 'conversation': {
- 'conversation_id': conversation.conversation_id,
- 'title': conversation.title,
- 'created_at': conversation.created_at.isoformat(),
- 'updated_at': conversation.updated_at.isoformat()
- }
- })
-
- except Exception as e:
- db.session.rollback()
- logger.error(f"创建新对话时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/conversations/clear-current', methods=['POST'])
- @auth_middleware.login_required
- def clear_current_conversation():
- """清除用户的当前对话映射,强制下次创建新对话"""
- try:
- # 从认证中间件获取用户信息
- user_info = getattr(request, 'user_info', {})
- user_id = user_info.get('db_user_id') # 使用数据库用户ID
-
- if not user_id:
- return jsonify({
- 'success': False,
- 'error': '用户信息获取失败'
- }), HTTPStatus.UNAUTHORIZED
-
- # 清除用户的当前对话映射
- if str(user_id) in user_current_conversations:
- del user_current_conversations[str(user_id)]
-
- return jsonify({
- 'success': True,
- 'message': '当前对话映射已清除'
- })
-
- except Exception as e:
- logger.error(f"清除当前对话映射时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/conversations/<int:conversation_id>/switch', methods=['POST'])
- @auth_middleware.login_required
- def switch_conversation(conversation_id):
- """切换到指定对话"""
- try:
- # 从认证中间件获取用户信息
- user_info = getattr(request, 'user_info', {})
- user_id = user_info.get('db_user_id') # 使用数据库用户ID
-
- if not user_id:
- return jsonify({
- 'success': False,
- 'error': '用户信息获取失败'
- }), HTTPStatus.UNAUTHORIZED
-
- # 切换到指定对话
- conversation = switch_to_conversation(user_id, conversation_id)
-
- return jsonify({
- 'success': True,
- 'message': '对话切换成功',
- 'conversation': {
- 'conversation_id': conversation.conversation_id,
- 'title': conversation.title,
- 'session_id': conversation.session_id,
- 'message_count': conversation.message_count,
- 'last_message_at': conversation.last_message_at.isoformat() if conversation.last_message_at else None,
- 'created_at': conversation.created_at.isoformat(),
- 'updated_at': conversation.updated_at.isoformat()
- }
- })
-
- except ValueError as e:
- return jsonify({
- 'success': False,
- 'error': str(e)
- }), HTTPStatus.NOT_FOUND
- except Exception as e:
- logger.error(f"切换对话时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/conversations/<int:conversation_id>/status', methods=['GET'])
- @auth_middleware.login_required
- def get_conversation_status(conversation_id):
- """获取对话的实时状态信息"""
- try:
- # 从认证中间件获取用户信息
- user_info = getattr(request, 'user_info', {})
- user_id = user_info.get('db_user_id') # 使用数据库用户ID
-
- if not user_id:
- return jsonify({
- 'success': False,
- 'error': '用户信息获取失败'
- }), HTTPStatus.UNAUTHORIZED
-
- # 验证对话是否属于当前用户
- conversation = Conversation.query.filter_by(
- conversation_id=conversation_id,
- user_id=user_id,
- is_deleted=False
- ).first()
-
- if not conversation:
- return jsonify({
- 'success': False,
- 'error': '对话不存在或无权限访问'
- }), HTTPStatus.NOT_FOUND
-
- # 获取对话的最新消息
- latest_message = Message.query.filter_by(
- conversation_id=conversation_id,
- is_deleted=False
- ).order_by(Message.message_order.desc()).first()
-
- status_info = {
- 'conversation_id': conversation_id,
- 'title': conversation.title,
- 'message_count': conversation.message_count,
- 'last_message_at': conversation.last_message_at.isoformat() if conversation.last_message_at else None,
- 'updated_at': conversation.updated_at.isoformat(),
- 'has_latest_message': latest_message is not None,
- 'latest_message_role': latest_message.role if latest_message else None,
- '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
- }
-
- return jsonify({
- 'success': True,
- 'status': status_info
- })
-
- except Exception as e:
- logger.error(f"获取对话状态时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
- @app.route('/api/messages/<int:message_id>/edit', methods=['POST'])
- @auth_middleware.login_required
- @rate_limit(config.rate_limit_requests, config.rate_limit_window)
- def edit_message(message_id):
- """编辑用户消息并重新生成AI回复(流式响应)"""
- try:
- # 从请求中获取数据
- data = request.json or {}
- new_content = data.get('content', '')
- agent = data.get('agent') or 'kuaishou'
-
- # 输入验证
- new_content, error = sanitize_input(new_content, config.max_message_length)
- if error:
- # 流式响应中的错误返回
- return Response(
- f"data: {json.dumps({'success': False, 'error': error, 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 从认证中间件获取用户信息
- auth_user_info = getattr(request, 'user_info', {})
- auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID
-
- if not auth_user_id:
- return Response(
- f"data: {json.dumps({'success': False, 'error': '用户信息获取失败', 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 获取要编辑的消息
- message = Message.query.filter_by(
- message_id=message_id,
- is_deleted=False
- ).first()
-
- if not message:
- return Response(
- f"data: {json.dumps({'success': False, 'error': '消息不存在', 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 验证消息是否是用户消息
- if message.role != 'user':
- return Response(
- f"data: {json.dumps({'success': False, 'error': '只能编辑用户消息', 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 获取对话信息
- conversation = Conversation.query.filter_by(
- conversation_id=message.conversation_id,
- user_id=auth_user_id,
- is_deleted=False
- ).first()
-
- if not conversation:
- return Response(
- f"data: {json.dumps({'success': False, 'error': '对话不存在或无权限访问', 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 记录请求信息
- log_request_info(new_content, str(auth_user_id), 'POST')
-
- # 选择对应智能体 APP_ID
- chosen_app_id, appid_env_key = config.get_app_id_for_agent_with_source(agent)
- if not chosen_app_id:
- logger.error(f"未找到对应智能体APP_ID,agent={agent}")
- return Response(
- f"data: {json.dumps({'success': False, 'error': f'服务器未配置该智能体的 APP_ID: {agent}', 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 使用对话的session_id
- api_session_id = conversation.session_id
-
- # ===== 立即软删除当前消息及之后的所有消息(在流式响应开始前) =====
- try:
- # 计算将被删除的消息数(包括当前消息及之后的所有消息)
- deleted_count = Message.query.filter_by(
- conversation_id=conversation.conversation_id,
- is_deleted=False
- ).filter(
- Message.message_order >= message.message_order
- ).count()
-
- # 软删除当前消息及之后的所有消息(包括原来的AI回复等)
- Message.query.filter_by(
- conversation_id=conversation.conversation_id,
- is_deleted=False
- ).filter(
- Message.message_order >= message.message_order
- ).update({
- 'is_deleted': True,
- 'deleted_at': get_beijing_time()
- }, synchronize_session=False)
-
- # 更新消息计数:减去被删除的消息数
- conversation.message_count -= deleted_count
- db.session.commit()
-
- # 保存编辑后的用户消息(使用相同的message_order)
- new_user_message = save_message_to_database(
- auth_user_id,
- conversation.conversation_id,
- 'user',
- new_content,
- None,
- message_order=message.message_order
- )
-
- except Exception as db_error:
- db.session.rollback()
- logger.error(f"软删除消息或保存编辑消息时发生错误: {str(db_error)}", exc_info=True)
- return Response(
- f"data: {json.dumps({'success': False, 'error': f'更新消息失败: {str(db_error)}', 'done': True})}\n\n",
- content_type='text/event-stream'
- )
-
- # 获取该消息之前的所有消息(不包括已删除的消息,现在数据库中已删除的消息不会出现在查询结果中)
- previous_messages = Message.query.filter_by(
- conversation_id=message.conversation_id,
- is_deleted=False
- ).filter(
- Message.message_order < message.message_order
- ).order_by(Message.message_order.asc()).all()
-
- # 构建对话历史(使用编辑后的消息)
- messages = []
- for msg in previous_messages:
- messages.append({"role": msg.role, "content": msg.content})
-
- # 添加编辑后的用户消息(刚保存的)
- messages.append({"role": "user", "content": new_content})
-
- # 截断对话历史以控制内存使用
- messages = truncate_conversation_history(messages, config.max_conversation_history)
-
- # 定义流式响应生成器
- def generate() -> Generator[str, None, None]:
- try:
- # 发送初始响应
- initial_data = {
- 'success': True,
- 'chunk': '',
- 'done': False,
- 'agent': agent,
- 'message_id': message_id
- }
- yield f"data: {json.dumps(initial_data)}\n\n"
-
- try:
- # 使用DashScope SDK直接调用 - 多轮对话模式
- params = {
- "api_key": config.api_key,
- "app_id": chosen_app_id,
- "messages": messages, # 发送完整的对话历史
- "stream": True,
- "incremental_output": True,
- "has_thoughts": True
- }
-
- # 使用对话的session_id
- if api_session_id:
- params["session_id"] = api_session_id
-
- response = Application.call(**params)
-
- new_session_id: Optional[str] = None
- accumulated_text: str = ""
- thoughts_content: List[str] = []
-
- # 处理流式响应
- for chunk in response:
- # 检查客户端是否断开连接
- if request.environ.get('werkzeug.server.shutdown') or not request:
- break
-
- if chunk.status_code != HTTPStatus.OK:
- 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"
- return
-
- # 处理思考过程
- if hasattr(chunk.output, 'thoughts') and chunk.output.thoughts:
- for it in chunk.output.thoughts:
- if it.action_type == 'reasoning':
- content = str(it.thought) if not isinstance(it.thought, str) else it.thought
- thoughts_content.append(content)
- yield f"data: {json.dumps({'success': True, 'thought': content, 'done': False, 'agent': agent})}\n\n"
-
- # 处理回答内容
- if hasattr(chunk.output, 'text') and chunk.output.text:
- current_text = str(chunk.output.text)
- accumulated_text += current_text
- yield f"data: {json.dumps({'success': True, 'chunk': current_text, 'done': False, 'agent': agent})}\n\n"
-
- # 检查是否有会话ID
- if hasattr(chunk.output, 'session_id') and chunk.output.session_id:
- new_session_id = chunk.output.session_id
-
- # 处理结束,保存AI回复到数据库
- try:
- # 更新session_id(如果需要)
- if new_session_id and new_session_id != api_session_id:
- conversation.session_id = new_session_id
- db.session.commit()
-
- # 保存新的AI回复(如果有)
- if accumulated_text:
- thinking_text = '\n'.join(thoughts_content) if thoughts_content else None
- ai_message = save_message_to_database(
- auth_user_id,
- conversation.conversation_id,
- 'assistant',
- accumulated_text,
- thinking_text,
- message_order=message.message_order + 1
- )
- # save_message_to_database 已经提交,message_count 已更新
- else:
- # 如果没有AI回复,更新对话的更新时间
- conversation.updated_at = get_beijing_time()
- conversation.last_message_at = get_beijing_time()
- db.session.commit()
-
- except Exception as db_error:
- db.session.rollback()
- logger.error(f"保存AI回复到数据库时发生错误: {str(db_error)}", exc_info=True)
- # 不影响主要功能,继续返回结果
-
- # 发送完成标记
- final_data = {
- 'success': True,
- 'done': True,
- 'thoughts': thoughts_content,
- 'agent': agent,
- 'session_id': new_session_id or api_session_id,
- 'message_id': message_id,
- 'conversation_id': conversation.conversation_id
- }
-
- yield f"data: {json.dumps(final_data)}\n\n"
-
- except Exception as e:
- logger.error(f"DashScope SDK 调用错误: {str(e)}", exc_info=True)
- yield f"data: {json.dumps({'success': False, 'error': f'DashScope API调用错误: {str(e)}', 'done': True, 'agent': agent})}\n\n"
-
- except Exception as e:
- logger.error(f"流式API调用错误: {str(e)}", exc_info=True)
- yield f"data: {json.dumps({'success': False, 'error': f'发生错误: {str(e)}', 'done': True, 'agent': agent})}\n\n"
-
- # 创建并返回流式响应
- response = Response(stream_with_context(generate()), content_type='text/event-stream')
- response.headers['Cache-Control'] = 'no-cache'
- response.headers['X-Accel-Buffering'] = 'no'
- return response
-
- except Exception as e:
- logger.error(f"编辑消息时发生错误: {str(e)}", exc_info=True)
- return jsonify({
- 'success': False,
- 'error': f"发生错误: {str(e)}"
- }), HTTPStatus.INTERNAL_SERVER_ERROR
-
- # 健康检查端点
- @app.route('/health', methods=['GET'])
- def health_check():
- """健康检查端点"""
- return jsonify({
- 'status': 'ok',
- 'app_version': '2.0.0',
- 'python_version': '3.12',
- 'timestamp': time.time()
- })
- if __name__ == '__main__':
- app.run(host='0.0.0.0', port=config.port, debug=config.debug)
|