哈喽大家好~ 今天给大家带来一篇超全面的Flask教程,全程聚焦企业大模型开发必备场景,从零基础环境搭建到企业级部署,覆盖流式响应、鉴权、参数校验等核心技能,不管你是刚接触Flask的新手,还是想做企业级大模型API开发的开发者,跟着这篇教程走,就能快速上手!

先说明下适用场景和技术栈,避免大家走弯路:

适用场景:大模型对话接口、AI服务API、企业中台接口、流式输出(打字机效果)、AI文件处理等主流大模型开发场景

技术栈:Flask 3.x(最新稳定版)+ Pydantic(参数校验)+ Flask-CORS(跨域)+ SSE(流式响应)+ Gunicorn(部署)+ Docker(容器化)

话不多说,直接上干货,建议收藏备用,遇到问题随时回头看!

第1章 环境搭建

工欲善其事,必先利其器。Flask环境搭建很简单,但企业开发中一定要注意环境隔离,避免依赖冲突,这一步千万别偷懒~

1.1 基础环境要求

  • Python 3.8+(企业大模型开发标准版本,兼容性最好)

  • 虚拟环境(隔离项目依赖,企业开发必备规范)

1.2 具体搭建步骤

1. 创建虚拟环境

不同系统命令略有差异,直接复制对应命令执行即可:

# Windows系统
python -m venv venv
venv\Scripts\activate

# Mac/Linux系统
python3 -m venv venv
source venv/bin/activate

执行成功后,终端前面会出现【venv】标识,说明虚拟环境已激活。

2. 安装核心依赖

这里列出的都是企业大模型开发必装依赖,缺一不可,直接复制命令安装:

# 核心框架(Flask 3.0.3稳定版)
pip install flask==3.0.3
# 跨域处理(前端调用后端接口必用)
pip install flask-cors
# 参数校验(企业接口强制要求,防止非法请求)
pip install pydantic
# 异步部署(大模型IO密集型场景,提升并发)
pip install gevent gunicorn
# 接口限流(防止接口滥用,保护服务)
pip install flask-limiter
# JWT鉴权(企业级身份验证,多用户场景必备)
pip install pyjwt

3. 导出依赖(企业协作必备)

开发完成后,导出依赖到requirements.txt,方便同事协作或部署时快速安装:

pip freeze > requirements.txt

第2章 Flask核心基础

Flask是轻量级WSGI框架,核心就两个:Flask应用对象 + 路由装饰器,上手非常快,先从最基础的Hello World开始。

2.1 最小Flask应用(Hello World)

创建app.py(项目入口文件),复制以下代码,注释已经写得很详细,跟着看就能懂:

# app.py  项目入口文件
from flask import Flask

# 1. 创建Flask应用实例(__name__表示当前模块,自动识别静态文件和模板目录)
app = Flask(__name__)

# 2. 定义路由:访问根目录(http://localhost:5000)时执行index函数
@app.route('/')
def index():
    return "Hello Flask!企业大模型开发入门"

# 3. 启动服务(仅开发环境使用,生产环境禁止用)
if __name__ == '__main__':
    # debug=True:开发热重载(修改代码无需重启服务),生产环境必须关闭!
    app.run(host='0.0.0.0', port=5000, debug=True)

运行命令:python app.py,打开浏览器访问 http://localhost:5000,就能看到返回的文字,说明Flask应用启动成功!

2.2 路由与请求方法(企业API基础)

企业开发中,我们主要用Flask写RESTful API,常用请求方法有4种:GET(查询)、POST(提交)、PUT(更新)、DELETE(删除),结合动态路由可实现灵活接口设计。

from flask import request

# 动态路由:接收URL中的参数(int: 限制参数为整数类型)
@app.route('/user/<int:user_id>')
def get_user(user_id):
    # 可根据user_id查询用户信息,这里仅做演示
    return f"用户ID:{user_id}"

# 限定请求方法:仅支持POST(大模型接口最常用)
@app.route('/api/chat', methods=['POST'])
def chat():
    return "大模型对话接口(后续完善)"

2.3 请求与响应(API核心,重中之重)

API的核心就是“接收请求数据”和“返回响应结果”,这部分是大模型接口开发的基础,一定要掌握。

1. 获取请求数据

大模型接口常用3种请求数据格式:GET参数、POST JSON参数、文件上传,分别对应不同场景:

@app.route('/api/chat', methods=['POST'])
def chat():
    # 1. 获取GET参数(适用于简单查询,如:http://localhost:5000/api/chat?prompt=你好)
    prompt_get = request.args.get('prompt', '')  # 第二个参数是默认值,防止报错
    
    # 2. 获取POST JSON参数(大模型对话接口最常用,前端传JSON格式数据)
    data = request.get_json()  # 自动解析JSON数据,若前端未传JSON,返回None
    user_input = data.get('prompt', '')  # 获取用户输入的prompt
    
    # 3. 获取文件(大模型文件处理场景,如上传PDF、图片让AI分析)
    file = request.files.get('file')  # 获取上传的文件
    
    # 这里仅返回获取到的数据,实际开发中会对接大模型
    return {"code": 200, "msg": "success", "data": {"prompt_get": prompt_get, "user_input": user_input}}

2. 自定义响应

企业API要求响应格式统一,常用JSON响应,可自定义响应头、状态码:

from flask import make_response, jsonify

# 方式1:jsonify(企业标准JSON响应,自动设置Content-Type为application/json)
@app.route('/api/json')
def json_response():
    # 统一响应格式:code(状态码)、msg(提示信息)、data(业务数据)
    return jsonify(code=200, msg="成功", data="测试数据")

# 方式2:自定义响应头、状态码(适用于特殊场景,如跨域配置、授权验证)
@app.route('/api/custom')
def custom_response():
    resp = make_response(jsonify(msg="自定义响应"))
    resp.headers['X-Custom-Header'] = 'flask-llm'  # 自定义响应头
    resp.status_code = 201  # 自定义状态码(201表示创建成功)
    return resp

第3章 企业级核心组件(大模型开发必备,规范开发)

个人开发可以把所有代码写在一个文件里,但企业开发必须模块化、规范化,这部分的组件是企业大模型开发的“标配”,缺一不可。

3.1 蓝图 Blueprint(模块化拆分)

当项目接口越来越多(如用户接口、大模型对话接口、文件接口),把所有路由写在一个app.py里会非常混乱,难以维护。蓝图就是用来拆分模块的,按业务模块拆分,逻辑更清晰。

1. 企业标准目录结构

先搭建规范的目录结构,后续所有开发都按这个结构来:

flask_llm/  # 项目根目录
├── app.py          # 项目入口文件(初始化应用、注册蓝图)
├── api/            # 接口模块目录(所有API都放在这里)
│   ├── __init__.py # 空文件,标识该目录为Python模块
│   ├── chat.py     # 大模型对话接口模块(核心)
│   └── user.py     # 用户接口模块(登录、注册等)
└── config.py       # 配置文件(存储密钥、端口等配置)

2. 编写蓝图(以chat.py为例)

在api/chat.py中创建蓝图,定义大模型相关路由:

# api/chat.py
from flask import Blueprint, jsonify

# 1. 创建蓝图:参数说明(蓝图名、模块名、url前缀)
# url_prefix='/api/chat':该蓝图下所有接口都以/api/chat开头
chat_bp = Blueprint('chat', __name__, url_prefix='/api/chat')

# 2. 蓝图路由(用法和@app.route一致,只是装饰器变成@chat_bp.route)
@chat_bp.route('/completion', methods=['POST'])
def completion():
    # 非流式对话接口(后续完善)
    return jsonify(code=200, msg="大模型非流式对话接口")

@chat_bp.route('/stream', methods=['POST'])
def stream_chat():
    # 流式对话接口(大模型打字机效果,核心接口)
    return jsonify(code=200, msg="大模型流式对话接口")

3. 注册蓝图(app.py中)

蓝图创建后,需要在入口文件中注册,否则接口无法访问:

# app.py
from flask import Flask
from api.chat import chat_bp  # 导入chat蓝图

# 初始化应用
app = Flask(__name__)

# 注册蓝图(关键步骤,否则蓝图路由无效)
app.register_blueprint(chat_bp)

# 启动服务
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

测试访问:启动服务后,访问 http://localhost:5000/api/chat/completion,就能看到返回结果,说明蓝图注册成功。

3.2 Pydantic 参数校验(企业接口强制要求)

大模型接口必须校验入参!比如用户传空的prompt、temperature超出0-1的范围,都会导致接口异常。Pydantic可以轻松实现参数校验,规范请求数据格式。

# api/chat.py
from pydantic import BaseModel, Field, constr
from flask import request, jsonify

# 1. 定义校验模型(继承BaseModel)
class ChatRequest(BaseModel):
    # 必填参数:prompt,非空字符串,最大长度1000(防止用户输入过长内容)
    prompt: constr(min_length=1, max_length=1000)
    # 可选参数:temperature,默认值0.7,范围0-1(大模型生成多样性参数)
    temperature: float = Field(0.7, ge=0, le=1)  # ge=大于等于,le=小于等于
    # 可选参数:model,默认值为default-llm(可切换不同大模型)
    model: str = "default-llm"

# 2. 在接口中使用校验模型
@chat_bp.route('/completion', methods=['POST'])
def completion():
    try:
        # 解析并校验参数:将请求JSON数据传入校验模型
        data = ChatRequest(**request.get_json())
        # 校验通过,返回数据(实际开发中会对接大模型)
        return jsonify(
            code=200,
            msg="参数校验成功",
            data={"prompt": data.prompt, "temperature": data.temperature, "model": data.model}
        )
    except Exception as e:
        # 校验失败,返回统一错误信息
        return jsonify(code=400, msg=f"参数错误:{str(e)}"), 400

测试:如果前端传空的prompt,或temperature=1.2,接口会直接返回参数错误,无需手动判断,非常高效。

3.3 请求钩子(鉴权/日志/预处理)

请求钩子是Flask的核心特性,能在请求前后执行指定逻辑,企业大模型开发中主要用于鉴权、日志记录、请求预处理,无需在每个接口中重复写代码。

# app.py
from flask import request

# 1. 全局钩子:所有请求执行前执行(最常用,用于鉴权、日志)
@app.before_request
def before_request():
    # 记录请求日志(方便线上排查问题)
    print(f"请求方法:{request.method},请求地址:{request.path},请求时间:{time.strftime('%Y-%m-%d %H:%M:%S')}")
    # 鉴权逻辑(后续章节详细讲,这里先占位)
    # 比如:校验API Key、Token等
    pass

# 2. 全局钩子:所有请求执行后执行(用于修改响应、添加响应头)
@app.after_request
def after_request(resp):
    # 示例:添加跨域相关响应头(也可以用Flask-CORS,这里仅做演示)
    resp.headers['Access-Control-Allow-Origin'] = '*'
    return resp

第4章 API开发必备(企业标准规范,避坑关键)

企业API开发有严格的规范,比如跨域处理、统一异常响应、日志配置,这些都是避免线上问题的关键,一定要严格遵守。

4.1 跨域处理(CORS,前端调用必解决)

前端(Vue、React等)调用后端接口时,会出现跨域问题(浏览器同源策略限制),Flask-CORS是企业标准解决方案,配置简单。

# app.py
from flask_cors import CORS

app = Flask(__name__)

# 方式1:全局允许跨域(开发环境常用,简单高效)
CORS(app)

# 方式2:仅允许指定接口跨域(生产环境常用,更安全)
# CORS(app, resources={r"/api/*": {"origins": "*"}})  # 仅/api/开头的接口允许跨域

4.2 全局异常处理(统一响应格式)

企业API必须统一返回格式,不管是正常响应还是异常响应,前端都能按同一格式解析,避免前端多套逻辑处理。这里我们自定义异常和全局异常捕获。

# app.py
from flask import jsonify

# 1. 自定义异常类(用于业务异常,如参数错误、未授权)
class APIError(Exception):
    def __init__(self, code, msg):
        self.code = code  # 状态码
        self.msg = msg    # 错误提示

# 2. 捕获自定义异常(业务异常统一处理)
@app.errorhandler(APIError)
def handle_api_error(e):
    return jsonify(code=e.code, msg=e.msg), e.code

# 3. 捕获404异常(接口不存在)
@app.errorhandler(404)
def page_not_found(e):
    return jsonify(code=404, msg="接口不存在,请检查请求地址"), 404

# 4. 捕获500异常(服务器内部错误)
@app.errorhandler(500)
def server_error(e):
    return jsonify(code=500, msg="服务器内部错误,请稍后再试"), 500

使用示例:在接口中抛出异常 raise APIError(401, "未授权:API Key错误"),会自动返回统一格式的错误响应。

4.3 企业级日志配置

开发环境可以用print调试,但生产环境必须用日志记录,用于排查接口调用异常、大模型调用失败等问题。Flask自带日志模块,配置简单。

# app.py
import logging
from logging.handlers import RotatingFileHandler

# 配置日志(按文件大小分割,避免日志文件过大)
def setup_log(app):
    # 1. 创建日志处理器:日志文件名为llm_api.log,最大10MB,保留5个备份文件
    handler = RotatingFileHandler('llm_api.log', maxBytes=10*1024*1024, backupCount=5)
    # 2. 设置日志级别(INFO:记录正常请求,ERROR:记录错误)
    handler.setLevel(logging.INFO)
    # 3. 将日志处理器添加到应用中
    app.logger.addHandler(handler)

# 入口文件中注册日志
app = Flask(__name__)
setup_log(app)

# 4. 使用日志(在接口中记录关键操作)
@app.route('/api/chat/completion', methods=['POST'])
def completion():
    try:
        # 业务逻辑...
        app.logger.info("大模型非流式接口调用成功")  # 记录正常日志
        return jsonify(code=200, msg="success")
    except Exception as e:
        app.logger.error(f"大模型接口调用失败:{str(e)}")  # 记录错误日志
        raise APIError(500, "接口调用失败")

日志文件会保存在项目根目录的llm_api.log中,线上出现问题时,查看日志就能快速定位原因。

第5章 安全与鉴权(企业大模型接口必加,防止滥用)

大模型接口通常需要消耗大量计算资源,必须添加鉴权机制,防止未授权调用、恶意请求滥用资源。企业常用两种鉴权方式,根据场景选择。

5.1 API Key 鉴权(简单高效,首选)

适用于单服务、第三方调用场景,客户端请求头携带API Key,服务端校验合法性,配置简单,安全性足够。

# config.py  配置文件(敏感信息单独存放,不硬编码)
API_KEY = "sk-1234567890abcdef"  # 生产环境建议用环境变量,不要硬编码!

# app.py  鉴权逻辑(结合请求钩子)
from config import API_KEY
from flask import request

@app.before_request
def check_api_key():
    # 1. 放行预检请求(跨域时浏览器会先发OPTIONS请求,无需鉴权)
    if request.method == 'OPTIONS':
        return
    # 2. 非接口路由放行(如静态资源、首页,根据实际需求调整)
    if not request.path.startswith('/api'):
        return
    # 3. 校验API Key(客户端请求头携带API-Key字段)
    client_key = request.headers.get('API-Key')
    if not client_key or client_key != API_KEY:
        # 鉴权失败,抛出自定义异常
        raise APIError(code=401, msg="未授权:API Key错误或不存在")

使用方式:前端请求时,在请求头中添加 API-Key: sk-1234567890abcdef,否则无法访问接口。

5.2 JWT 鉴权(多用户、会话管理场景)

适用于多用户平台(如企业内部大模型平台,需要区分不同用户权限),用户登录后生成Token,后续请求携带Token即可,支持过期时间设置。

# app.py
import jwt
import time

# 配置密钥(生产环境用环境变量,绝对不能泄露)
SECRET_KEY = "flask-llm-secret-2026"

# 1. 生成Token(用户登录成功后调用)
def create_token(user_id):
    # payload:存储用户信息,exp:过期时间(这里设置24小时过期)
    payload = {"user_id": user_id, "exp": time.time() + 3600*24}
    # 生成Token,指定算法为HS256
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

# 2. 校验Token(结合请求钩子)
def verify_token(token):
    try:
        # 解码Token,校验合法性和过期时间
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload  # 返回用户信息(如user_id)
    except Exception as e:
        # Token无效或过期,抛出异常
        raise APIError(401, "Token无效或已过期,请重新登录")

# 3. 鉴权钩子
@app.before_request
def check_jwt():
    if request.method == 'OPTIONS' or not request.path.startswith('/api'):
        return
    # 从请求头获取Token(通常放在Authorization字段,格式为Bearer Token)
    token = request.headers.get('Authorization')
    if not token or not token.startswith('Bearer '):
        raise APIError(401, "请携带Token")
    # 提取Token(去掉Bearer 前缀)
    token = token.split(' ')[1]
    verify_token(token)  # 校验Token

使用示例:用户登录后,后端返回Token,前端后续请求时,请求头添加 Authorization: Bearer 生成的Token

第6章 大模型开发核心技能(重中之重,企业实战)

这部分是大模型开发的核心,也是和普通API开发的区别,重点掌握异步路由、流式响应、文件上传,这三个是企业大模型接口的“核心功能”。

6.1 异步路由(大模型IO密集型优化)

大模型调用属于IO密集型操作(等待大模型返回结果),如果用同步路由,会导致接口阻塞,并发能力差。Flask 2.0+ 原生支持异步路由,能大幅提升接口并发量。

# api/chat.py
import asyncio

# 异步路由:添加async关键字,函数内部可使用await
@chat_bp.route('/async', methods=['POST'])
async def async_chat():
    # 模拟异步调用大模型(实际开发中替换为真实大模型调用)
    # await 用于等待异步操作完成,不阻塞其他请求
    await asyncio.sleep(1)  # 模拟大模型调用耗时
    return jsonify(code=200, msg="异步大模型响应成功", data="异步返回结果")

注意:异步路由中,不能使用同步IO操作(如requests库),否则会阻塞异步事件循环,建议使用aiohttp等异步库。

6.2 流式响应 SSE(打字机效果,企业大模型标配)

像ChatGPT一样的打字机效果,是大模型对话接口的核心体验,实现方式是SSE(Server-Sent Events,服务器向客户端实时推送数据),Flask可直接实现,无需额外依赖。

# api/chat.py
from flask import Response
import time

# 流式对话接口(核心接口)
@chat_bp.route('/stream', methods=['POST'])
def stream_chat():
    # 1. 校验请求参数(简化版,实际开发用Pydantic校验)
    data = request.get_json()
    prompt = data.get('prompt', '')
    if not prompt:
        raise APIError(400, "prompt不能为空")
    
    # 2. 生成器函数:实时推送数据(SSE核心)
    def generate():
        # 模拟大模型流式输出(实际开发中替换为大模型的流式返回)
        response_text = "我是企业级大模型,正在为你解答问题~ 你的问题是:" + prompt
        # 逐字符推送,模拟打字机效果
        for char in response_text:
            # SSE固定格式:data: 内容\n\n(必须有两个换行)
            yield f"data: {char}\n\n"
            time.sleep(0.1)  # 控制推送速度,模拟打字效果
    
    # 3. 返回流式响应,设置mimetype为text/event-stream(SSE标准)
    return Response(generate(), mimetype='text/event-stream')

前端对接:无需额外插件,用EventSource即可接收实时数据,示例代码(JS):

// 前端JS代码(Vue/React中可直接使用)
const eventSource = new EventSource('http://localhost:5000/api/chat/stream');
eventSource.onmessage = function(e) {
    // e.data 就是服务器推送的单个字符,拼接起来就是完整响应
    document.getElementById('chat-content').innerText += e.data;
};
// 关闭连接(如用户关闭对话)
// eventSource.close();

6.3 文件上传(大模型文档/图片处理)

企业大模型常用场景:上传PDF、Word、图片,让AI分析内容(如文档总结、图片识别),Flask可轻松实现文件上传功能。

# app.py
import os
from werkzeug.utils import secure_filename
from flask import request, jsonify

# 1. 配置文件上传目录(项目根目录下的uploads文件夹)
UPLOAD_FOLDER = 'uploads'
# 确保目录存在,不存在则创建
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

# 2. 文件上传接口(api/chat.py中,使用蓝图)
@chat_bp.route('/upload', methods=['POST'])
def upload_file():
    # 获取上传的文件(前端name属性为file)
    file = request.files.get('file')
    if not file:
        raise APIError(400, "请上传文件")
    
    # 安全处理文件名(防止恶意文件名,如../../etc/passwd)
    filename = secure_filename(file.filename)
    # 拼接文件保存路径
    file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
    # 保存文件
    file.save(file_path)
    
    # 返回文件路径,后续可用于大模型分析
    return jsonify(code=200, msg="文件上传成功", data={"file_path": file_path})

前端对接:用FormData上传文件,示例代码(JS):

const formData = new FormData();
formData.append('file', document.getElementById('file-input').files[0]);
fetch('http://localhost:5000/api/chat/upload', {
    method: 'POST',
    headers: {
        'API-Key': 'sk-1234567890abcdef'  // 鉴权信息
    },
    body: formData
}).then(res => res.json()).then(data => {
    console.log("文件路径:", data.data.file_path);
});

第7章 企业大模型API

整合前面所有知识点,做一个完整的示例,包含:鉴权、参数校验、流式/非流式接口、统一响应。

# app.py  企业大模型API入口文件
from flask import Flask, request, jsonify, Response
from flask_cors import CORS
from pydantic import BaseModel, constr
import time
import logging

# 1. 初始化应用
app = Flask(__name__)
CORS(app)  # 全局跨域
logging.basicConfig(level=logging.INFO)  # 日志配置

# 2. 核心配置(生产环境用环境变量)
API_KEY = "sk-llm-2026"  # API Key鉴权
UPLOAD_FOLDER = "uploads"  # 文件上传目录
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

# 3. 统一异常类
class APIError(Exception):
    def __init__(self, code, msg):
        self.code = code
        self.msg = msg

# 4. 全局异常处理
@app.errorhandler(APIError)
def handle_api_error(e):
    return jsonify(code=e.code, msg=e.msg), e.code

@app.errorhandler(404)
def not_found(e):
    return jsonify(code=404, msg="接口不存在,请检查请求地址"), 404

@app.errorhandler(500)
def server_error(e):
    return jsonify(code=500, msg="服务器内部错误,请稍后再试"), 500

# 5. API Key鉴权钩子
@app.before_request
def check_auth():
    if request.method == "OPTIONS":
        return
    if not request.path.startswith("/api"):
        return
    key = request.headers.get("API-Key")
    if not key or key != API_KEY:
        raise APIError(401, "API Key无效或不存在")

# 6. 参数校验模型(大模型对话请求)
class ChatRequest(BaseModel):
    prompt: constr(min_length=1, max_length=1000)  # 必填,非空,最大1000字符
    temperature: float = 0.7  # 可选,默认0.7,范围0-1

# 7. 非流式大模型接口
@app.route("/api/chat/completion", methods=["POST"])
def chat_completion():
    try:
        # 校验参数
        data = ChatRequest(**request.get_json())
        # 模拟调用大模型(实际开发中替换为OpenAI、通义千问等真实大模型)
        result = f"大模型响应:{data.prompt} | 温度:{data.temperature}"
        # 记录日志
        app.logger.info(f"非流式接口调用成功,prompt:{data.prompt}")
        return jsonify(code=200, msg="success", data=result)
    except Exception as e:
        app.logger.error(f"非流式接口调用失败:{str(e)}")
        raise APIError(400, f"参数错误:{str(e)}")

# 8. 流式大模型接口(核心功能)
@app.route("/api/chat/stream", methods=["POST"])
def chat_stream():
    try:
        data = request.get_json()
        prompt = data.get("prompt", "")
        if not prompt:
            raise APIError(400, "prompt不能为空")
        
        # 生成器:实时推送数据
        def generate():
            response = f"流式响应:你输入的内容是【{prompt}】,我是企业级大模型,正在实时输出文字~"
            for char in response:
                yield f"data: {char}\n\n"
                time.sleep(0.08)  # 控制打字速度
        
        app.logger.info(f"流式接口调用成功,prompt:{prompt}")
        return Response(generate(), mimetype="text/event-stream")
    except Exception as e:
        app.logger.error(f"流式接口调用失败:{str(e)}")
        raise APIError(400, f"请求错误:{str(e)}")

# 9. 文件上传接口
@app.route("/api/chat/upload", methods=["POST"])
def upload_file():
    file = request.files.get('file')
    if not file:
        raise APIError(400, "请上传文件")
    
    filename = secure_filename(file.filename)
    file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
    file.save(file_path)
    
    app.logger.info(f"文件上传成功,文件名:{filename},路径:{file_path}")
    return jsonify(code=200, msg="上传成功", data={"file_path": file_path})

# 10. 启动服务(开发环境)
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)

第8章 企业级部署

开发环境的Flask自带服务器,性能差、不稳定,不能用于生产环境。企业级部署必须用“Gunicorn + Nginx + Docker”,确保接口高可用、高并发。

8.1 开发环境 vs 生产环境区别

  • 开发环境:debug=True(热重载),用Flask自带服务器,仅用于开发调试;

  • 生产环境:debug=False(关闭热重载,防止安全漏洞),用Gunicorn服务器,配合Nginx反向代理,容器化部署。

8.2 Gunicorn + Gevent 部署(高并发)

Gunicorn是Python的异步WSGI服务器,性能强、稳定性高,配合Gevent可实现高并发,是企业Python服务的首选部署方式。

# 1. 确保已安装Gunicorn和Gevent(前面已安装)
# 2. 启动命令(生产环境启动)
gunicorn -w 4 -k gevent -b 0.0.0.0:5000 app:app

# 命令参数说明:
# -w 4:启动4个工作进程(建议设置为CPU核心数的2倍)
# -k gevent:使用gevent异步模式,提升IO密集型场景并发
# -b 0.0.0.0:5000:绑定地址和端口(0.0.0.0表示允许外部访问)
# app:app:第一个app是入口文件app.py,第二个app是Flask应用实例

8.3 Docker 容器化(企业部署标配)

容器化部署可解决“环境不一致”问题,不管是开发、测试还是生产环境,都能保证运行效果一致,部署更高效。

1. 创建Dockerfile

在项目根目录创建Dockerfile文件,复制以下内容:

# 基础镜像(Python 3.10 slim版,体积小,适合生产环境)
FROM python:3.10-slim

# 设置工作目录(容器内的目录)
WORKDIR /app

# 复制依赖文件到容器
COPY requirements.txt .

# 安装依赖(--no-cache-dir:不缓存,减小镜像体积)
RUN pip install --no-cache-dir -r requirements.txt

# 复制项目所有文件到容器
COPY . .

# 暴露端口(和Flask服务端口一致)
EXPOSE 5000

# 生产环境启动命令(用Gunicorn)
CMD ["gunicorn", "-w", "4", "-k", "gevent", "-b", "0.0.0.0:5000", "app:app"]

2. 构建并运行Docker镜像

# 基础镜像(Python 3.10 slim版,体积小,适合生产环境)
FROM python:3.10-slim

# 设置工作目录(容器内的目录)
WORKDIR /app

# 复制依赖文件到容器
COPY requirements.txt .

# 安装依赖(--no-cache-dir:不缓存,减小镜像体积)
RUN pip install --no-cache-dir -r requirements.txt

# 复制项目所有文件到容器
COPY . .

# 暴露端口(和Flask服务端口一致)
EXPOSE 5000

# 生产环境启动命令(用Gunicorn)
CMD ["gunicorn", "-w", "4", "-k", "gevent", "-b", "0.0.0.0:5000", "app:app"]

8.4 Nginx 反向代理(生产必备)

Nginx用于负载均衡、HTTPS配置、静态资源代理,保护后端服务,提升接口访问速度,是生产环境的“网关”。

Nginx配置示例(简化版):

server {
    listen 80;  # 监听80端口(HTTP)
    server_name llm-api.com;  # 你的域名(如没有域名,可填服务器IP)

    # 所有请求转发到后端Flask服务
    location / {
        proxy_pass http://127.0.0.1:5000;  # 后端服务地址
        proxy_set_header Host $host;  # 传递主机名
        proxy_set_header X-Real-IP $remote_addr;  # 传递真实IP
    }
}

配置完成后,重启Nginx,访问 http://llm-api.com/api/chat/completion,即可通过域名访问接口。

第9章 进阶优化

前面的内容已经能满足基本的企业大模型API开发,但要做到高可用、高并发,还需要以下优化,实际开发中可根据需求添加。

  1. 接口限流:用Flask-Limiter限制接口调用频率,防止恶意请求滥用资源(如每分钟最多调用100次);

  2. 缓存:用Redis缓存大模型重复请求(如用户问相同的问题,直接返回缓存结果,提升响应速度);

  3. 对话历史存储:用MySQL/PostgreSQL + SQLAlchemy存储用户对话历史,支持上下文对话;

  4. 环境变量:用python-dotenv管理敏感信息(API Key、密钥、数据库地址),不硬编码,提升安全性;

  5. 监控告警:用Prometheus + Grafana监控接口性能(响应时间、调用次数、错误率),出现异常及时告警。

总结

这篇教程从零基础环境搭建,到企业级规范、大模型核心技能、生产部署,全覆盖企业大模型开发所需的Flask知识点,没有多余的冗余内容,全程聚焦实战。

包括:

  • Flask基础:应用初始化、路由、请求响应、蓝图模块化;

  • 企业规范:参数校验、统一响应、全局异常、日志、鉴权;

  • 大模型核心:异步路由、SSE流式响应(打字机效果)、文件上传;

  • 生产部署:Gunicorn+Nginx+Docker(企业标准流程)。

觉得有用的话,记得点赞收藏哦!

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐