代码收藏家技术教程 2025-02-21

Python连接通义千问大模型

项目源码地址：Python连接通义千问大模型项目源码

第一步：我们先创建一个数据库用来存放用户和大模型API密钥

第二步：创建一个名为users的表

create table users(
	user_name varchar(15),
	user_password varchar(20),
	api_key varchar(100),
	cishu int,
	moxing varchar(15)
	);

列名	数据类型	作用
user_name	字符串	用户名
user_password	字符串	用户密码
api_key	字符串	API密钥
cishu	整形	调用次数
moxing	字符串	模型名

第三步：我们去阿里云百炼获取API密钥

大模型服务平台百炼_企业级大模型开发平台_AI应用构建_人工智能与机器学习-阿里云

这个就是我们的API密钥
切记！！！一定不要对外泄露
切记！！！一定不要对外泄露
切记！！！一定不要对外泄露

第四步：在MySQL表里创建一个用户

api_key字段填写我们刚才复制的那一串密钥
moxing字段填写模型名称，具体查看模型列表模型列表_大模型服务平台百炼(Model Studio)-阿里云帮助中心

第五步：在Python代码里填写数据库信息

然后就可以直接运行了~

源代码如下：

from tkinter import Tk,Label,Entry,Button,messagebox
from pymysql import Connection
import tkinter as tk
from tkinter import scrolledtext
from openai import OpenAI
import time

def get_completion(prompt, max_retries=3):
    api_key = key1
    if not api_key:
        raise ValueError("Please set the OPENAI_API_KEY environment variable.")

    client = OpenAI(
        api_key=api_key,
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    )
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=moxing,
                messages=[{'role': 'system', 'content': 'You are a helpful assistant.'},
                          {'role': 'user', 'content': prompt}],
                stream=True,
            )
            return response
        except Exception as e:
            if attempt < max_retries - 1:
                backoff_time = 2 ** attempt
                print(f"请求失败，正在重试 ({attempt + 1}/{max_retries})...等待 {backoff_time} 秒")
                time.sleep(backoff_time)
            else:
                raise e

def on_submit():
    global fang4
    user_input = entry.get()
    if user_input.strip():
        if fang4 <= 0:
            messagebox.showwarning("次数不足", "您的发送次数已用完!")
            return

        try:
            response = get_completion(user_input)
            chat_history.config(state=tk.NORMAL)
            chat_history.insert(tk.END, "-----------------------""\n" + "你: " + user_input + "\n")
            entry.delete(0, tk.END)
            fang4 -= 1
            sql = f"UPDATE users SET cishu = '{fang4}' WHERE user_name = '{username}';"
            cursor.execute(sql)

            def update_response():
                try:
                    part = next(response)
                    delta = part.choices[0].delta.content
                    if delta:
                        chat_history.insert(tk.END, delta)
                        chat_history.see(tk.END)
                        chat_history.update_idletasks()
                except StopIteration:
                    chat_history.insert(tk.END, "\n")
                    chat_history.config(state=tk.DISABLED)
                    return
                except Exception as e:
                    chat_history.insert(tk.END, f"\n发生错误：{str(e)}\n")
                    chat_history.config(state=tk.DISABLED)
                    return
                chat_history.after(100, update_response)

            chat_history.insert(tk.END, "沁枫: ")
            chat_history.after(100, update_response)
        except Exception as e:
            chat_history.config(state=tk.NORMAL)
            chat_history.insert(tk.END, f"请求超时，请稍后再试。\n详细信息：{str(e)}\n\n")
            chat_history.config(state=tk.DISABLED)

def main():
    global entry, chat_history

    root = tk.Tk()
    root.title("沁枫AI")
    root.resizable(0, 0)

    frame = tk.Frame(root)
    frame.pack(padx=100, pady=10)

    w = tk.Label(root, text=f"你还有({fang})次发送机会\n")
    w.pack()

    entry = tk.Entry(frame, width=100)
    entry.pack(side=tk.LEFT, padx=(0, 10))

    submit_button = tk.Button(frame, text="发送", command=on_submit)
    submit_button.pack(side=tk.LEFT)

    chat_history = scrolledtext.ScrolledText(root, state='disabled', height=30, width=150, wrap=tk.WORD)
    chat_history.pack(expand=True, fill=tk.BOTH)

    def update_label():
        global fang4
        simulate_data_update()
        w.config(text=f"你还有({fang4})次发送机会\n")
        root.after(1000, update_label)  # 每隔1秒更新一次

    def simulate_data_update():
        global fang4
        # 获取新数据，就是1秒查询一次的那个
        # 数据更新的 emmmm（下下面的才是）
        sql = "SELECT cishu FROM users WHERE user_name=%s AND user_password=%s;"

        # SQL查询(数据更新用的)
        cursor.execute(sql, (username, password))
        data = cursor.fetchone()
        fang4 = data[0]

    update_label()
    root.mainloop()


def login():
    global fang,username,password,key1,moxing
    # 获取输入框中的数据
    username = username_entry.get()
    password = password_entry.get()
    sql = "SELECT api_key,cishu,moxing FROM users WHERE user_name=%s AND user_password=%s;"
    # SQL查询
    cursor.execute(sql, (username, password))
    data = cursor.fetchone()

    if data:
        messagebox.showinfo("登录成功", "欢迎，{}".format(username))
        window.destroy()
        fang = data[1]
        key1 = data[0]
        moxing = data[2]
        main()
    else:
        messagebox.showerror("登录失败", "用户名或密码错误")

# 创建数据库连接
conn = Connection(
    host="IP",  #主机名（IP）
    port=3306,  #端口
    user="databases_user",  #账户
    password="databases_password",  #密码
    autocommit=True  #自动提交（确认）
)
cursor = conn.cursor()
conn.select_db("databases_name") #数据库名

# 创建GUI窗口
window = Tk()
window.title("沁枫AI登录系统")
window.geometry('280x150')  # 设置窗口大小
window.resizable(0, 0)
Label(window, text="用户名").pack()
username_entry = Entry(window)
username_entry.pack()


Label(window, text="密码").pack()
password_entry = Entry(window, show="*")
password_entry.pack()

Button(window, text="登录", command=login).pack()
username = username_entry.get()
window.mainloop()