11.24 速率限制

docker-compose.yml

@@ -1,15 +1,57 @@
 version: '3.8'
+
 services:
-  flask_app:
-    build: .
+  # Flask Web 服务
+  web:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    command: python flask_app/start_up.py
     ports:
       - "5000:5000"
-    environment:
-      - PYTHONPATH=/flask_project
-      - DASHSCOPE_API_KEY=sk-f7ad8ad193064cf482588f7064e75183
-      - DASHSCOPE_WORKSPACE_ID=llm-mo38469hdfwtervi
-      - ALIBABA_CLOUD_ACCESS_KEY_ID=LTAI5tRWhjktXyY5MovoiNuF
-      - ALIBABA_CLOUD_ACCESS_KEY_SECRET=88oyw7LniqV8i0SnOuSFS5lprfrPtw
     volumes:
       - .:/flask_project
-    command: python flask_app/main/start_up.py
+    environment:
+      FLASK_ENV: development
+      DASHSCOPE_API_KEY: ${DASHSCOPE_API_KEY}
+      DASHSCOPE_WORKSPACE_ID: ${DASHSCOPE_WORKSPACE_ID}
+      ALIBABA_CLOUD_ACCESS_KEY_ID: ${ALIBABA_CLOUD_ACCESS_KEY_ID}
+      ALIBABA_CLOUD_ACCESS_KEY_SECRET: ${ALIBABA_CLOUD_ACCESS_KEY_SECRET}
+      CELERY_BROKER_URL: redis://redis:6379/0
+      CELERY_RESULT_BACKEND: redis://redis:6379/0
+    depends_on:
+      - redis
+    networks:
+      - flask_network
+
+  # Celery Worker 服务
+  worker:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    command: celery -A flask_app.celery_app.celery_app worker --loglevel=info
+    volumes:
+      - .:/flask_project
+    environment:
+      DASHSCOPE_API_KEY: ${DASHSCOPE_API_KEY}
+      DASHSCOPE_WORKSPACE_ID: ${DASHSCOPE_WORKSPACE_ID}
+      ALIBABA_CLOUD_ACCESS_KEY_ID: ${ALIBABA_CLOUD_ACCESS_KEY_ID}
+      ALIBABA_CLOUD_ACCESS_KEY_SECRET: ${ALIBABA_CLOUD_ACCESS_KEY_SECRET}
+      CELERY_BROKER_URL: redis://redis:6379/0
+      CELERY_RESULT_BACKEND: redis://redis:6379/0
+    depends_on:
+      - redis
+    networks:
+      - flask_network
+
+  # Redis 服务
+  redis:
+    image: redis:7.0
+    ports:
+      - "6379:6379"
+    networks:
+      - flask_network
+
+networks:
+  flask_network:
+    driver: bridge

flask_app/celery_app.py

@@ -0,0 +1,20 @@
+import os
+from celery import Celery
+
+# 创建 Celery 实例
+celery_app = Celery(
+    'tasks',
+    broker=os.getenv('CELERY_BROKER_URL', 'redis://redis:6379/0'),  # 从环境变量加载 Broker 地址
+    backend=os.getenv('CELERY_RESULT_BACKEND', 'redis://redis:6379/0')  # 从环境变量加载 Backend 地址
+)
+
+# 设置任务队列的速率限制（全局每分钟 250 次）
+celery_app.conf.task_annotations = {
+    'tasks.process_qianwen_long': {'rate_limit': '250/m'}
+}
+
+# 可选：加载更多配置项（从环境变量或直接配置）
+celery_app.conf.update(
+    timezone='UTC',  # 设置时区（可根据需要修改）
+    enable_utc=True   # 确保 UTC 时间生效
+)

flask_app/routes/routes.py

@@ -0,0 +1,45 @@
+# flask_project/flask_app/routes.py
+from flask import request, jsonify
+from flask_app.task import process_qianwen_long
+
+def register_routes(app):
+    @app.route('/api/qianwen_long', methods=['POST'])
+    def qianwen_long_endpoint():
+        data = request.json
+        file_id = data.get('file_id')
+        user_query = data.get('user_query')
+
+        if not file_id or not user_query:
+            return jsonify({'error': 'file_id 和 user_query 是必需的'}), 400
+
+        try:
+            # 发送任务到 Celery 队列
+            task = process_qianwen_long.delay(file_id, user_query)
+            return jsonify({'task_id': task.id}), 202
+        except Exception as e:
+            return jsonify({'error': str(e)}), 500
+
+    @app.route('/api/task_status/<task_id>', methods=['GET'])
+    def get_task_status(task_id):
+        task = process_qianwen_long.AsyncResult(task_id)
+        if task.state == 'PENDING':
+            response = {
+                'state': task.state,
+                'status': '等待中...'
+            }
+        elif task.state == 'SUCCESS':
+            response = {
+                'state': task.state,
+                'result': task.result
+            }
+        elif task.state == 'FAILURE':
+            response = {
+                'state': task.state,
+                'status': str(task.info),  # 错误信息
+            }
+        else:
+            response = {
+                'state': task.state,
+                'status': task.info,
+            }
+        return jsonify(response)

flask_app/start_up.py

@@ -1,15 +1,15 @@
 # flask_app/start_up.py
 
 import logging
-from flask import Flask, request
-
+from flask import Flask, request, jsonify
 
 from flask_app.logger_setup import CSTFormatter, create_logger
 from flask_app.routes.get_deviation import get_deviation_bp
 from flask_app.routes.little_zbparse import little_zbparse_bp
 from flask_app.routes.upload import upload_bp
+from flask_app.routes.routes import register_routes
 from flask_app.routes.test_zbparse import test_zbparse_bp
-
+from flask_app.task import process_qianwen_long
 def create_app():
     app = Flask(__name__)
 
@@ -43,7 +43,7 @@ def create_app():
     app.register_blueprint(little_zbparse_bp)
     app.register_blueprint(upload_bp)
     app.register_blueprint(test_zbparse_bp)
-
+    register_routes(app)
     return app
 
 #TODO:培训要求、总体要求、进度要求、'建设要求'到技术要求中，归类到其他要求中

flask_app/task.py

@@ -0,0 +1,11 @@
+from celery_app import celery_app
+from flask_app.general.通义千问long import qianwen_long
+
+
+@celery_app.task(name='tasks.process_qianwen_long')
+def process_qianwen_long(self, file_id, user_query):
+    try:
+        return qianwen_long(file_id, user_query)
+    except Exception as exc:
+        # 如果发生异常，重试任务
+        raise self.retry(exc=exc, countdown=60)  # 60 秒后重试

requirements.txt

@@ -14,4 +14,6 @@ dashscope==1.19.2
 PyMuPDF==1.24.1
 openai==1.33.0
 pathlib==1.0.1
-alibabacloud_bailian20231229==1.7.0
+alibabacloud_bailian20231229==1.7.0
+celery==5.2.3
+redis==4.1.0