（免费）flask调用讯飞星火AI，实现websocket

本文章可借鉴学习，不可直接盗用

接入ai要获取ID，Secret，Key，和接口地址，由于我们服务接口类型是websocket，所以要获取相应的接口地址。（千万不要复制粘贴到http的了）

还要获取domain，在操作文档中找到对应模型的admain

安装库pip install flask websocket-client werkzeug

具体看这篇文章

Django接入免费的 AI 大模型——讯飞星火（2025年4月最新！！！）_星火大模型 api password-CSDN博客

# coding: utf-8
import _thread as thread
import base64
import datetime
import hashlib
import hmac
import json
import ssl
from urllib.parse import urlparse, urlencode
from wsgiref.handlers import format_date_time
from time import mktime
import websocket

class Ws_Param:
    def __init__(self, APPID, APIKey, APISecret, gpt_url):
        self.APPID = APPID
        self.APIKey = APIKey
        self.APISecret = APISecret
        self.host = urlparse(gpt_url).netloc
        self.path = urlparse(gpt_url).path
        self.gpt_url = gpt_url

    def create_url(self):
        now = datetime.datetime.now()
        date = format_date_time(mktime(now.timetuple()))
        signature_origin = f"host: {self.host}\ndate: {date}\nGET {self.path} HTTP/1.1"
        signature_sha = hmac.new(
            self.APISecret.encode("utf-8"),
            signature_origin.encode("utf-8"),
            digestmod=hashlib.sha256
        ).digest()
        signature_sha_base64 = base64.b64encode(signature_sha).decode("utf-8")
        authorization_origin = (
            f'api_key="{self.APIKey}", algorithm="hmac-sha256", '
            f'headers="host date request-line", signature="{signature_sha_base64}"'
        )
        authorization = base64.b64encode(authorization_origin.encode("utf-8")).decode("utf-8")
        url_params = {
            "authorization": authorization,
            "date": date,
            "host": self.host
        }
        return f"{self.gpt_url}?{urlencode(url_params)}"

def on_error(ws, error):
    print("### 连接错误:", error)

def on_close(ws, close_status_code, close_msg):
    print("### 连接关闭 ###")

def on_open(ws):
    thread.start_new_thread(run, (ws,))

def run(ws):
    data = json.dumps(gen_params(ws.appid, ws.query, ws.domain))
    ws.send(data)

def on_message(ws, message):
    try:
        data = json.loads(message)
        code = data['header']['code']
        if code != 0:
            print(f"请求错误: {code}, 错误信息: {data['header']['message']}")
        else:
            choices = data["payload"]["choices"]
            content = choices["text"][0]["content"]
            print(content, end='')
            if choices["status"] == 2:
                print("\n#### 会话结束")
                ws.close()
    except Exception as e:
        print("### 消息解析异常:", e)

def gen_params(appid, query, domain):
    return {
        "header": {"app_id": appid, "uid": "1234"},
        "parameter": {"chat": {"domain": domain, "temperature": 0.5, "max_tokens": 4096}},
        "payload": {"message": {"text": [{"role": "user", "content": query}]}}
    }

def main(appid, api_secret, api_key, gpt_url, domain, query):
    # 验证密钥格式（APPID长度8，APIKey和APISecret长度32）
    if not (len(appid) == 8 and len(api_key) == 32 and len(api_secret) == 32):
        print("错误: API密钥格式不正确")
        return

    ws_param = Ws_Param(appid, api_key, api_secret, gpt_url)
    ws_url = ws_param.create_url()

    ws = websocket.WebSocketApp(
        ws_url,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.appid = appid
    ws.query = query
    ws.domain = domain
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})

if __name__ == "__main__":
    # 根据截图中的 Websocket 认证信息修正参数（Spark4.0 Ultra 配置）
    main(
        appid="",                  # 截图中显示的 APPID
        api_secret="",  # 截图中 APISecret（注意核对隐藏字符）
        api_key="",      # 截图中 APIKey
        gpt_url="",  # Websocket 地址需与控制台一致
        domain="4.0Ultra",                 # Spark4.0 对应 domain
        query="你叫什么名字"
    )

注意这里要替换成你的密钥，接口地址和domain

运行结果如下

前端templates\index.html

<!-- templates/index.html -->
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>AI 智能助手</title>
    <link href="https://cdn.bootcdn.net/ajax/libs/font-awesome/6.4.0/css/all.min.css" rel="stylesheet">
    <style>
        :root {
            --primary-color: #4a90e2;
            --secondary-color: #f5f5f5;
            --success-color: #34d399;
            --error-color: #ef4444;
        }

        * {
            box-sizing: border-box;
            margin: 0;
            padding: 0;
        }

        body {
            font-family: 'Segoe UI', system-ui, -apple-system, sans-serif;
            background: #f0f2f5;
            min-height: 100vh;
        }

        .container {
            max-width: 1200px;
            margin: 0 auto;
            padding: 20px;
        }

        .chat-container {
            background: white;
            border-radius: 12px;
            box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
            height: 80vh;
            display: flex;
            flex-direction: column;
        }

        #chat-box {
            flex: 1;
            overflow-y: auto;
            padding: 20px;
            scroll-behavior: smooth;
        }

        .message {
            display: flex;
            gap: 12px;
            margin-bottom: 16px;
        }

        .message.user {
            flex-direction: row-reverse;
        }

        .avatar {
            width: 40px;
            height: 40px;
            border-radius: 50%;
            background: var(--secondary-color);
            display: flex;
            align-items: center;
            justify-content: center;
            flex-shrink: 0;
        }

        .message-content {
            max-width: 70%;
            padding: 12px 16px;
            border-radius: 12px;
            position: relative;
        }

        .user .message-content {
            background: var(--primary-color);
            color: white;
            border-radius: 12px 12px 0 12px;
        }

        .bot .message-content {
            background: var(--secondary-color);
            border-radius: 12px 12px 12px 0;
        }

        .file-upload {
            background: #f8fafc;
            border: 2px dashed #cbd5e1;
            border-radius: 8px;
            padding: 16px;
            text-align: center;
            margin: 16px 0;
            cursor: pointer;
            transition: all 0.3s ease;
        }

        .file-upload:hover {
            border-color: var(--primary-color);
            background: rgba(74, 144, 226, 0.05);
        }

        .input-area {
            border-top: 1px solid #eee;
            padding: 16px;
            display: flex;
            gap: 12px;
            background: white;
        }

        #user-input {
            flex: 1;
            padding: 12px;
            border: 1px solid #e2e8f0;
            border-radius: 8px;
            font-size: 16px;
            transition: all 0.3s ease;
        }

        #user-input:focus {
            outline: none;
            border-color: var(--primary-color);
            box-shadow: 0 0 0 3px rgba(74, 144, 226, 0.2);
        }

        button {
            padding: 12px 24px;
            background: var(--primary-color);
            color: white;
            border: none;
            border-radius: 8px;
            cursor: pointer;
            transition: all 0.2s ease;
            display: flex;
            align-items: center;
            gap: 8px;
        }

        button:hover {
            background: #357abd;
            transform: translateY(-1px);
        }

        button:disabled {
            background: #94a3b8;
            cursor: not-allowed;
        }

        .file-item {
            display: flex;
            align-items: center;
            gap: 8px;
            padding: 8px;
            background: white;
            border-radius: 8px;
            margin: 8px 0;
        }

        .file-icon {
            font-size: 20px;
        }

        .progress-bar {
            height: 4px;
            background: #e2e8f0;
            border-radius: 2px;
            overflow: hidden;
            margin-top: 8px;
        }

        .progress {
            width: 0%;
            height: 100%;
            background: var(--primary-color);
            transition: width 0.3s ease;
        }

        @media (max-width: 768px) {
            .container {
                padding: 10px;
            }

            .message-content {
                max-width: 85%;
            }
        }
    </style>
</head>
<body>
    <div class="container">
        <h1 style="margin-bottom: 20px; color: #1e293b;">AI 智能助手 <i class="fas fa-robot"></i></h1>

        <div class="chat-container">
            <div id="chat-box">
                <div class="message bot">
                    <div class="avatar"><i class="fas fa-robot"></i></div>
                    <div class="message-content">您好！我是智能助手，可以回答各种问题，也可以帮您分析上传的文件</div>
                </div>
            </div>

            <div class="input-area">
                <label class="file-upload">
                    <input type="file" id="file-input" hidden multiple>
                    <i class="fas fa-cloud-upload-alt"></i> 点击上传文件
                </label>

                <input type="text" id="user-input" placeholder="输入消息...">
                <button onclick="sendMessage()" id="send-btn">
                    <i class="fas fa-paper-plane"></i> 发送
                </button>
            </div>
        </div>
    </div>

    <script>
        let isSending = false

        // 文件上传处理
        document.getElementById('file-input').addEventListener('change', async function(e) {
            const files = e.target.files
            for (let file of files) {
                await uploadFile(file)
            }
            e.target.value = ''
        })

        async function uploadFile(file) {
            const formData = new FormData()
            formData.append('file', file)

            try {
                showFileItem(file)
                const response = await fetch('/upload', {
                    method: 'POST',
                    body: formData
                })

                const data = await response.json()
                if (response.ok) {
                    updateFileStatus(file.name, 'success', data.url)
                } else {
                    updateFileStatus(file.name, 'error', data.error)
                }
            } catch (error) {
                updateFileStatus(file.name, 'error', '上传失败')
            }
        }

        function showFileItem(file) {
            const chatBox = document.getElementById('chat-box')
            const itemId = `file-${Date.now()}`

            const html = `
                <div class="file-item" id="${itemId}">
                    <i class="fas fa-file-alt file-icon"></i>
                    <div style="flex:1">
                        <div>${file.name}</div>
                        <div class="progress-bar"><div class="progress"></div></div>
                    </div>
                </div>
            `
            chatBox.insertAdjacentHTML('beforeend', html)
            chatBox.scrollTop = chatBox.scrollHeight
        }

        function updateFileStatus(filename, status, message) {
            const items = document.querySelectorAll('.file-item')
            const item = Array.from(items).find(el =>
                el.textContent.includes(filename))

            if (!item) return

            if (status === 'success') {
                item.innerHTML = `
                    <i class="fas fa-check-circle" style="color: var(--success-color)"></i>
                    <div style="flex:1">
                        <div>${filename}</div>
                        <a href="${message}" target="_blank" style="color: var(--primary-color); text-decoration: none">
                            查看文件 <i class="fas fa-external-link-alt"></i>
                        </a>
                    </div>
                `
            } else {
                item.innerHTML = `
                    <i class="fas fa-times-circle" style="color: var(--error-color)"></i>
                    <div style="flex:1">
                        <div>${filename}</div>
                        <div style="color: var(--error-color)">${message}</div>
                    </div>
                `
            }
        }

        // 消息处理
        async function sendMessage() {
            if (isSending) return

            const input = document.getElementById('user-input')
            const message = input.value.trim()
            if (!message) return

            isSending = true
            input.disabled = true
            document.getElementById('send-btn').disabled = true

            try {
                appendMessage(message, 'user')
                input.value = ''

                const response = await fetch('/chat', {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    body: JSON.stringify({ message })
                })

                const data = await response.json()
                if (response.ok) {
                    appendMessage(data.response, 'bot')
                } else {
                    appendMessage(`错误：${data.error}`, 'bot')
                }
            } catch (error) {
                appendMessage('网络请求失败', 'bot')
            } finally {
                isSending = false
                input.disabled = false
                document.getElementById('send-btn').disabled = false
                input.focus()
            }
        }

        function appendMessage(message, sender) {
            const chatBox = document.getElementById('chat-box')
            const isUser = sender === 'user'

            const messageEl = document.createElement('div')
            messageEl.className = `message ${isUser ? 'user' : 'bot'}`
            messageEl.innerHTML = `
                <div class="avatar">
                    ${isUser ? '<i class="fas fa-user"></i>' : '<i class="fas fa-robot"></i>'}
                </div>
                <div class="message-content">${message}</div>
            `

            chatBox.appendChild(messageEl)
            chatBox.scrollTop = chatBox.scrollHeight
        }

        // 回车发送
        document.getElementById('user-input').addEventListener('keypress', (e) => {
            if (e.key === 'Enter' && !e.shiftKey) {
                e.preventDefault()
                sendMessage()
            }
        })
    </script>
</body>
</html>

后端flask自命名

# app.py
import os
from flask import Flask, render_template, request, jsonify, send_from_directory
from werkzeug.utils import secure_filename
import _thread as thread
import base64
import datetime
import hashlib
import hmac
import json
import ssl
from urllib.parse import urlparse, urlencode
from wsgiref.handlers import format_date_time
from time import mktime
import websocket

app = Flask(__name__)
app.config.update({
    'UPLOAD_FOLDER': 'uploads',
    'MAX_CONTENT_LENGTH': 100 * 1024 * 1024,  # 100MB
    'ALLOWED_EXTENSIONS': {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif', 'zip', 'exe', 'apk'}
})

# 配置信息（生产环境建议使用环境变量）
CONFIG = {
    "APPID": "ba110a39",
    "API_SECRET": "NWQ5Nzk1OGY5MjkxNGQyMjMyOGFkNjgy",
    "API_KEY": "63f807b3f65c6c98a249c167403901f2",
    "GPT_URL": "wss://spark-api.xf-yun.com/v4.0/chat",
    "DOMAIN": "4.0Ultra"
}

os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)


class WsParam:
    def __init__(self, appid, api_key, api_secret, gpt_url):
        self.APPID = appid
        self.APIKey = api_key
        self.APISecret = api_secret
        self.host = urlparse(gpt_url).netloc
        self.path = urlparse(gpt_url).path
        self.gpt_url = gpt_url

    def create_url(self):
        now = datetime.datetime.now()
        date = format_date_time(mktime(now.timetuple()))
        signature_origin = f"host: {self.host}\ndate: {date}\nGET {self.path} HTTP/1.1"
        signature_sha = hmac.new(
            self.APISecret.encode("utf-8"),
            signature_origin.encode("utf-8"),
            digestmod=hashlib.sha256
        ).digest()
        signature_sha_base64 = base64.b64encode(signature_sha).decode("utf-8")
        authorization_origin = (
            f'api_key="{self.APIKey}", algorithm="hmac-sha256", '
            f'headers="host date request-line", signature="{signature_sha_base64}"'
        )
        authorization = base64.b64encode(authorization_origin.encode("utf-8")).decode("utf-8")
        return f"{self.gpt_url}?{urlencode({'authorization': authorization, 'date': date, 'host': self.host})}"


def allowed_file(filename):
    return '.' in filename and \
        filename.rsplit('.', 1)[1].lower() in app.config['ALLOWED_EXTENSIONS']


def get_spark_response(query):
    response_text = []

    def on_message(ws, message):
        data = json.loads(message)
        if data["header"]["code"] != 0:
            raise Exception(f'请求错误: {data["header"]["code"]}, {data["header"]["message"]}')
        else:
            choices = data["payload"]["choices"]
            text = choices["text"][0]["content"]
            response_text.append(text)
            if choices["status"] == 2:
                ws.close()

    def on_error(ws, error):
        raise Exception(f"WebSocket错误: {error}")

    def on_close(ws, *args):
        pass

    def on_open(ws):
        data = json.dumps({
            "header": {"app_id": CONFIG["APPID"], "uid": "1234"},
            "parameter": {"chat": {"domain": CONFIG["DOMAIN"], "temperature": 0.5, "max_tokens": 4096}},
            "payload": {"message": {"text": [{"role": "user", "content": query}]}}
        })
        ws.send(data)

    ws_param = WsParam(CONFIG["APPID"], CONFIG["API_KEY"], CONFIG["API_SECRET"], CONFIG["GPT_URL"])
    ws_url = ws_param.create_url()

    ws = websocket.WebSocketApp(
        ws_url,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
    return "".join(response_text)


@app.route('/')
def index():
    return render_template('index.html')


@app.route('/chat', methods=['POST'])
def chat():
    try:
        user_input = request.json.get('message')
        if not user_input:
            return jsonify({"error": "空消息"}), 400

        ai_response = get_spark_response(user_input)
        return jsonify({"response": ai_response})

    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route('/upload', methods=['POST'])
def upload_file():
    if 'file' not in request.files:
        return jsonify({"error": "没有选择文件"}), 400
    file = request.files['file']
    if file.filename == '':
        return jsonify({"error": "没有选择文件"}), 400
    if file and allowed_file(file.filename):
        filename = secure_filename(file.filename)
        file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
        return jsonify({
            "filename": filename,
            "url": f"/uploads/{filename}"
        })
    return jsonify({"error": "文件类型不允许"}), 400


@app.route('/uploads/<filename>')
def uploaded_file(filename):
    return send_from_directory(app.config['UPLOAD_FOLDER'], filename)


if __name__ == '__main__':
    app.run(debug=True)

运行结果（未添加结束对话功能和加载进度，待自行开发）