使用Gradio编写大模型ollama客户端 -界面版

使用Gradio编写大模型ollama客户端 -界面版

flyfish

文末包含完整源码

图例

sqlite3 是 Python 内置的一个库，用于与 SQLite 数据库进行交互。SQLite 是一个轻量级的数据库，它不需要单独的服务器进程或系统的配置，非常适合于嵌入式应用和小型项目。

从创建连接到关闭连接的全过程：

1. 导入模块：
   首先需要导入 sqlite3 模块，以便在 Python 程序中使用它。

   import sqlite3
   

2. 连接到数据库：
   使用 connect() 方法来连接到 SQLite 数据库文件。如果文件不存在，它将被创建。你可以传递数据库文件名（包括路径）作为参数。如果你想创建一个内存中的数据库，可以传递 :memory: 作为参数。

   conn = sqlite3.connect('example.db')
   

3. 创建游标对象：
   创建一个游标对象（Cursor），这是用来执行 SQL 命令的对象。

   c = conn.cursor()
   

4. 定义表格结构并创建表：
   使用 SQL 语句定义表结构，并通过游标对象的 execute() 方法执行这些语句。在你的例子中已经展示了如何创建两个表：session 和 conversations。

5. 插入数据：
   插入数据到表中可以通过 execute() 方法完成，通常会用占位符 (?) 来防止 SQL 注入攻击。

   c.execute("INSERT INTO session (name) VALUES (?)", ('SessionName',))
   

6. 查询数据：
   使用 execute() 方法执行 SELECT 语句，然后使用游标对象的 fetchall(), fetchone(), 或 fetchmany() 方法获取结果。

   c.execute("SELECT * FROM session")
   all_rows = c.fetchall()
   

7. 提交事务：
   在插入、更新或删除数据后，你需要调用 commit() 方法来保存更改。

   conn.commit()
   

8. 关闭连接：
   完成所有操作后，应该关闭数据库连接。

   conn.close()
   

9. 处理异常：
   为了保证程序的健壮性，你应该总是用 try-except 结构来处理可能出现的错误，并确保即使出现错误也能够正确关闭数据库连接。

   try:# 执行数据库操作
   except sqlite3.Error as e:print(f"An error occurred: {e}")
   finally:if conn:conn.close()
   

sqlite3 高级的功能

1. 事务管理

在 SQLite 中，事务可以确保一组操作要么全部成功执行，要么完全不执行。这对于保持数据的一致性非常重要。可以通过 BEGIN TRANSACTION 和 COMMIT 或 ROLLBACK 来手动控制事务。

例子：

import sqlite3# 连接到SQLite数据库
conn = sqlite3.connect('example.db')
c = conn.cursor()try:# 开始事务c.execute('BEGIN TRANSACTION')# 执行多个SQL语句c.execute("INSERT INTO users (name, age) VALUES (?, ?)", ('Alice', 30))c.execute("INSERT INTO users (name, age) VALUES (?, ?)", ('Bob', 25))# 提交事务conn.commit()
except sqlite3.Error as e:print(f"An error occurred: {e}")# 如果发生错误，则回滚事务conn.rollback()
finally:# 关闭连接conn.close()

2. 绑定参数

绑定参数是防止 SQL 注入攻击的一种方法，它允许在 SQL 查询中使用占位符（如 ?），然后传递一个参数列表或字典来填充这些占位符。

例子：

import sqlite3# 连接到SQLite数据库
conn = sqlite3.connect('example.db')
c = conn.cursor()# 使用问号占位符
c.execute("SELECT * FROM users WHERE age > ?", (25,))# 使用命名参数
c.execute("SELECT * FROM users WHERE name=:name AND age=:age", {"name": "Alice", "age": 30})results = c.fetchall()
print(results)

3. 自定义聚合函数

SQLite 内置了一些聚合函数，如 SUM, AVG, COUNT 等。但你也可以创建自己的聚合函数。

例子：

import sqlite3class MySum:def __init__(self):self.total = 0def step(self, value):if value is not None:self.total += valuedef finalize(self):return self.total# 连接到SQLite数据库
conn = sqlite3.connect('example.db')
conn.create_aggregate("mysum", 1, MySum)c = conn.cursor()
c.execute("CREATE TABLE IF NOT EXISTS sales (id INTEGER PRIMARY KEY, amount REAL)")
c.execute("INSERT INTO sales (amount) VALUES (100), (200), (300)")# 使用自定义聚合函数
c.execute("SELECT mysum(amount) FROM sales")
result = c.fetchone()[0]
print(f"The sum of all sales is: {result}")conn.close()

4. 自定义排序规则

默认情况下，SQLite 使用字母顺序进行排序。但是可以通过创建一个 Python 函数并将其注册为排序函数来改变这一行为。

例子：

import sqlite3def reverse_sort_order(a, b):# 反转排序顺序return -1 if a < b else 1 if a > b else 0# 连接到SQLite数据库
conn = sqlite3.connect('example.db')
conn.create_collation("reverse", reverse_sort_order)c = conn.cursor()
c.execute("CREATE TABLE IF NOT EXISTS names (name TEXT)")
c.execute("INSERT INTO names (name) VALUES ('Alice'), ('Bob'), ('Charlie')")# 使用自定义排序规则
c.execute("SELECT name FROM names ORDER BY name COLLATE reverse")
results = c.fetchall()
for row in results:print(row[0])conn.close()

一、Gradio 介绍

1. Gradio是什么？
   • Gradio是一个Python库，主要用于快速构建和分享机器学习模型与数据科学项目的用户界面。它的核心优势在于让开发者能够利用简单的API创建出用户友好的Web界面，而且不需要开发者有很深入的前端开发知识。
   • 从功能角度看，它的界面组件非常丰富。例如，它支持多种数据类型的输入输出，像文本（可以用于文本生成模型、文本分类模型等的输入输出）、图像（适用于图像分类、图像生成等任务）、音频（如语音识别、语音合成相关应用）、视频等。这些组件可以很方便地和自定义的Python函数关联起来，实现交互功能。

关键概念

• 接口（Interface）：这是 Gradio 中最基本的概念，指的是用户与你的模型之间进行交互的方式。你可以定义输入组件（如文本框、图像上传器等）和输出组件（如文本显示、图表等），并指定这些组件如何与底层函数交互。

• 组件（Components）：Gradio 提供了一套丰富的组件，包括但不限于 Textbox、Image、Video、Audio、Slider 等等。每个组件都有特定的功能，例如 Textbox 用来接收用户的文本输入，而 Image 则可以展示图片结果。

• 事件（Events）：在 Gradio 中，事件是指用户在界面上的动作，比如点击按钮、选择下拉菜单项等。你可以为这些事件绑定处理函数，当事件发生时，Gradio 将自动调用相应的函数来更新界面。

• 状态（State）：为了保持应用的状态信息（如对话历史记录、当前选择的选项等），Gradio 提供了 State 组件。它可以用来存储临时的数据，并在不同的回调函数间传递。

• 样式（Styles）：虽然 Gradio 默认提供了一个美观的主题，但你也可以自定义 CSS 样式以适应特定的设计需求。

2. 为什么Gradio能做Web界面？

   • Gradio是基于Web技术构建的。它在内部封装了许多Web开发中的常见操作，例如HTML、CSS和JavaScript的相关操作。当开发者使用Gradio的API来定义各种组件（如文本框、按钮等）时，Gradio会在后台将这些组件对应的HTML元素生成出来，并且通过JavaScript来处理用户交互事件（如按钮点击、文本输入等）。
   • 它能够将Python函数和这些Web组件进行绑定。例如，当用户在界面上输入文本并点击按钮后，Gradio会将输入的数据传递给关联的Python函数进行处理，然后再将函数的输出结果以合适的方式（如更新文本显示区域）展示在Web界面上。这种方式使得开发者可以专注于数据处理和业务逻辑（通过Python函数实现），而把界面展示和交互的复杂细节交给Gradio来处理。

3. 使用gr.Row和gr.Column进行布局

   • 基本原理：在Gradio中，gr.Row用于创建水平行，gr.Column用于在行内创建垂直列。这类似于HTML中的<div>标签与CSS的flex - box布局模式。它们可以嵌套使用，以构建复杂的网格布局。
   • 示例代码及解释：

     import gradio as gr
     with gr.Blocks() as demo:with gr.Row():with gr.Column():input_text = gr.Textbox()with gr.Column():output_text = gr.Textbox()
     demo.launch()
     

     在这个例子中，首先使用gr.Blocks()创建了一个Gradio应用。然后，with gr.Row():创建了一个水平行，在这个行里面，有两个gr.Column()，分别用于放置一个Textbox组件。这样就形成了一个简单的两列布局，左边是输入文本框input_text，右边是输出文本框output_text。

4. 空间分配

   • 默认均匀分配：当使用gr.Row和gr.Column时，默认情况下，各个列或行中的组件会均匀地分配空间。例如，在一个包含三个gr.Column的gr.Row中，每个列大约会占据水平方向1/3的空间；在一个包含两个gr.Row的gr.Column中，每个行大约会占据垂直方向1/2的空间。
   • 使用参数调整空间比例：
     • scale参数：可以通过scale参数来调整空间比例。例如，在gr.Row中，如果想让其中一个gr.Column占据2/3的空间，另一个占据1/3的空间，可以这样设置：

     import gradio as gr
     with gr.Blocks() as demo:with gr.Row():with gr.Column(scale = 2):input_text = gr.Textbox()with gr.Column(scale = 1):output_text = gr.Textbox()
     demo.launch()
     

     在这个例子中，左边的input_text所在列的scale参数为2，右边的output_text所在列的scale参数为1。这意味着左边的列会占据水平方向2/3的空间，右边的列会占据1/3的空间。
     • min_width和min_height参数：这些参数可以用于设置组件或布局元素的最小宽度和最小高度。例如，gr.Column(min_width = 200)可以确保这个列的最小宽度为200像素。这在需要保证某些组件有足够的显示空间时很有用，比如一个图像显示组件，可能需要设置一个最小宽度来保证图像能够正常显示。
   • 相对布局和绝对布局的结合：除了使用gr.Row和gr.Column的相对布局方式，Gradio也允许在一定程度上使用绝对布局。例如，可以通过设置组件的width和height属性来指定其绝对大小。但是，这种方式可能会影响到布局的灵活性和响应性。通常建议优先使用gr.Row和gr.Column的相对布局方式，在必要时再结合绝对布局来满足特定的设计需求。例如：

     import gradio as gr
     with gr.Blocks() as demo:with gr.Row():input_text = gr.Textbox(width = 300)  # 设置绝对宽度output_text = gr.Textbox()
     demo.launch()
     

     在这个例子中，input_text的宽度被设置为300像素，而output_text会根据剩余空间和布局规则来分配宽度。这样可以实现一种混合的布局方式，在保证某些组件有固定大小的同时，让其他组件自适应剩余空间。

二、运作机制

Gradio 应用程序的工作流程大致如下：

1. 定义接口：首先，你需要确定模型的输入和输出类型，并选择合适的 Gradio 组件来表示它们。

2. 编写逻辑：接着，实现一个 Python 函数，该函数接受来自输入组件的数据作为参数，并返回要展示给用户的输出数据。

3. 创建实例：使用 gr.Interface() 或更高级别的布局构造函数（如 gr.Blocks()）来组合输入组件、输出组件以及处理函数，从而形成完整的 Gradio 应用程序。

4. 启动服务：最后，调用 .launch() 方法启动 Web 服务器，使你的应用程序可以在浏览器中访问。

三、使用方法

步骤概述

1. 安装 Gradio：

   pip install gradio
   

2. 导入库并创建基本结构：

   import gradio as grdef greet(name):return f"Hello {name}!"demo = gr.Interface(fn=greet, inputs="text", outputs="text")
   demo.launch()
   

3. 定义更复杂的界面：

   • 使用 gr.Blocks() 构建包含多个组件的复杂布局。
   • 添加更多类型的输入和输出组件，如文件上传、滑块、表格等。
   • 设置事件监听器来响应用户的操作。

4. 管理状态：

   • 使用 gr.State() 来保存需要跨请求保留的信息。
   • 在事件处理器中更新状态对象的值。

5. 部署与共享：

   • 可以通过 .launch(share=True) 参数生成一个可共享的链接。
   • 对于生产环境，考虑使用 Docker 容器或其他云服务托管。

6. 说明：

   • 初始化界面：使用gr.Blocks()创建一个新的Gradio应用实例。例如with gr.Blocks() as demo:这个语句块内的内容都会被当作这个应用的组成部分。就好像是在搭建一个房子，这个语句块就是房子的框架，里面的各种组件就是房子的各个房间和设施。
   • 布局设计：通过gr.Row()和gr.Column()来组织界面元素，形成合理的布局。比如，gr.Row()可以创建一行来放置组件，gr.Column()可以在这一行或者其他布局元素中创建列来放置组件。这就好比在设计网页布局时，确定不同部分是放在同一行还是分栏显示，让界面更加直观和易于使用。
   • 添加组件：
     • gr.Markdown()：用于在界面中显示Markdown格式的文本。可以用来添加标题、说明文字、项目介绍等内容。例如，如果是一个文本生成模型的界面，可以用它来写“欢迎使用文本生成应用，下面请输入主题”这样的说明。
     • gr.Dropdown()：创建下拉选择框。比如在一个图像分类应用中，可以用它来让用户选择不同的分类类别。
     • gr.Textbox()：提供文本输入框。用户可以在这个框中输入文本，如在聊天机器人应用中输入聊天内容。
     • gr.Button()：定义按钮。当用户点击按钮时，可以触发相应的函数来执行操作，比如在文件上传应用中，点击“上传”按钮来触发文件上传操作。
     • gr.Chatbot()：特别适合用来展示对话形式的内容。设置type = 'messages'参数后，消息能够以对话气泡的形式呈现，很像常见的聊天软件界面，用于聊天机器人等对话式应用。
   • 状态管理：使用gr.State()组件来保存应用程序的状态信息。比如在一个多轮对话的聊天应用中，需要保存对话历史记录或者当前会话ID等信息，就可以通过这个组件在不同的回调函数之间传递数据，保证应用的连贯性。
   • 事件绑定：将用户交互（如点击按钮、选择选项）与特定的功能函数关联起来。例如create_session_button.click(...)这样的代码，就是定义了当“创建新会话”按钮被点击时要执行的操作。这就像给电器安装开关，用户操作按钮（开关）时，对应的电器（功能函数）就会工作。
   • 启动应用：最后调用demo.launch()方法启动Gradio应用程序。这会在本地打开一个Web浏览器窗口，展示构建好的应用，就像打开商店的大门，让用户能够看到和使用里面的内容。

四、具体示例

1. 简单文本处理

import gradio as grdef reverse_text(text):return text[::-1]demo = gr.Interface(fn=reverse_text, inputs="text", outputs="text")
demo.launch()

2. 图像分类器

假设你有一个预训练好的图像分类模型 classify_image，它可以接收一张图片并返回类别标签。

import gradio as grdef classify_image(image):# 这里应该调用实际的分类函数prediction = "cat"  # 示例返回值return predictiondemo = gr.Interface(fn=classify_image, inputs="image", outputs="label")
demo.launch()

3. 聊天机器人

这里展示了如何结合 SQLite 数据库和 API 请求来创建一个简单的聊天机器人。

import gradio as gr
import sqlite3
import requests
import json# 初始化数据库并创建表（如果它们不存在）
def initialize_database():conn = sqlite3.connect('chat_history.db')c = conn.cursor()c.execute('''CREATE TABLE IF NOT EXISTS conversations (id INTEGER PRIMARY KEY AUTOINCREMENT,user_input TEXT,bot_response TEXT,timestamp DATETIME DEFAULT CURRENT_TIMESTAMP)''')conn.commit()conn.close()initialize_database()# 生成来自选定模型的响应
def generate_response(user_input):url = 'http://localhost:11434/api/generate'headers = {'Content-Type': 'application/json'}data = {'prompt': user_input,'stream': False}response = requests.post(url, headers=headers, data=json.dumps(data))if response.status_code == 200:return response.json().get('response', '')else:return "Error generating response."# 保存对话到数据库
def save_conversation(user_input, bot_response):conn = sqlite3.connect('chat_history.db')c = conn.cursor()c.execute('INSERT INTO conversations (user_input, bot_response) VALUES (?, ?)', (user_input, bot_response))conn.commit()conn.close()def chat_with_bot(message):bot_reply = generate_response(message)save_conversation(message, bot_reply)return bot_replywith gr.Blocks() as demo:chat_display = gr.Chatbot(type='messages')user_input = gr.Textbox(label="您:")send_button = gr.Button("发送")def send_message(user_message):reply = chat_with_bot(user_message)return [{"role": "user", "content": user_message}, {"role": "assistant", "content": reply}], ""send_button.click(send_message, [user_input], [chat_display, user_input])user_input.submit(send_message, [user_input], [chat_display, user_input])demo.launch()

源码

import gradio as gr
import sqlite3
import subprocess
import requests
import json
from datetime import datetime# 初始化数据库并创建表（如果它们不存在）
def initialize_database():"""创建或初始化SQLite数据库，用于存储会话和对话历史。"""conn = sqlite3.connect('chat_history.db')  # 连接到SQLite数据库c = conn.cursor()# 创建会话表，确保每个会话名称是唯一的，并记录创建时间c.execute('''CREATE TABLE IF NOT EXISTS session (id INTEGER PRIMARY KEY AUTOINCREMENT,name TEXT UNIQUE,timestamp DATETIME DEFAULT CURRENT_TIMESTAMP)''')# 创建对话表，记录与特定会话相关的对话信息c.execute('''CREATE TABLE IF NOT EXISTS conversations (id INTEGER PRIMARY KEY AUTOINCREMENT,session_id INTEGER,model_name TEXT,user_input TEXT,bot_response TEXT,timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,FOREIGN KEY (session_id) REFERENCES session(id))''')conn.commit()  # 提交更改conn.close()   # 关闭连接# 函数：从命令行获取可用模型列表
def get_available_models():"""使用ollama命令行工具列出所有可用的聊天模型。"""try:result = subprocess.run(['ollama', 'list'], capture_output=True, text=True, check=True)lines = result.stdout.strip().split('\n')models = [line.split()[0] for line in lines[1:]]  # 跳过标题行并获取模型名称return modelsexcept subprocess.CalledProcessError as e:print(f"获取模型列表时出错: {e}")return []# 函数：生成来自选定模型的响应
def generate_response(model_name, user_input):"""根据用户输入和选择的模型名，调用API生成机器人的响应。"""url = f'http://localhost:11434/api/generate'headers = {'Content-Type': 'application/json'}data = {'model': model_name,'prompt': user_input,'stream': False}response = requests.post(url, headers=headers, data=json.dumps(data))if response.status_code == 200:return response.json().get('response', '')else:print('与模型通信时出错。')return ''# 函数：保存对话到数据库
def save_conversation(session_id, model_name, user_input, bot_response):"""将一次完整的对话（用户输入和机器人响应）保存到数据库中。"""conn = sqlite3.connect('chat_history.db')c = conn.cursor()c.execute('''INSERT INTO conversations (session_id, model_name, user_input, bot_response)VALUES (?, ?, ?, ?)''', (session_id, model_name, user_input, bot_response))conn.commit()conn.close()# 函数：创建新会话并返回其ID
def create_new_session(name):"""在数据库中创建一个新的会话条目，并返回该会话的ID。"""conn = sqlite3.connect('chat_history.db')c = conn.cursor()c.execute('INSERT INTO session (name) VALUES (?)', (name,))conn.commit()session_id = c.lastrowidconn.close()return session_id# 函数：加载所有会话
def load_sessions():"""从数据库加载所有会话记录，并按创建时间降序排列。"""conn = sqlite3.connect('chat_history.db')c = conn.cursor()c.execute('SELECT id, name FROM session ORDER BY timestamp DESC')sessions = c.fetchall()conn.close()return sessions# 函数：按会话ID加载对话历史
def load_conversation_history(session_id):"""根据提供的会话ID加载特定会话的所有对话记录。"""conn = sqlite3.connect('chat_history.db')c = conn.cursor()c.execute('''SELECT user_input, bot_response, model_nameFROM conversationsWHERE session_id = ?ORDER BY timestamp DESC''', (session_id,))conversation = c.fetchall()conn.close()return conversation# 初始化数据库
initialize_database()# 加载初始数据
models = get_available_models()
if not models:models = ["未找到模型。请检查您的Ollama安装。"]
session_list = load_sessions()
session_names = [name for id, name in session_list]# 定义Gradio应用
with gr.Blocks() as demo:gr.Markdown('# Gradio客户端')with gr.Row():with gr.Column(scale=1):  # 左侧栏gr.Markdown('## 选择模型')model_dropdown = gr.Dropdown(label='模型', choices=models, value=models[0] if models else None)gr.Markdown('## 创建新会话')session_name_input = gr.Textbox(label='输入新会话名称')create_session_button = gr.Button('创建新会话')gr.Markdown('## 会话历史')session_radio = gr.Radio(label='会话历史', choices=session_names)with gr.Column(scale=3):  # 右侧栏current_session_name_display = gr.Markdown('## 对话 - 未选择会话')chat_display = gr.Chatbot(type='messages')  # 设置type参数为'messages'user_input = gr.Textbox(label='您:', lines=2)send_button = gr.Button('发送')# 初始化状态变量session_list_state = gr.State(value=session_list)conversation_history_state = gr.State(value=[])current_session_id = gr.State(value=None)current_session_name = gr.State(value='')# 添加新会话的函数def add_new_session(session_name, session_list_state):"""创建新的会话，更新界面和状态变量。"""if session_name:session_id = create_new_session(session_name)session_list_state.append((session_id, session_name))session_names = [name for id, name in session_list_state]current_session_name_display_value = gr.update(value=f'## 对话 - {session_name}')return (gr.update(choices=session_names, value=session_name),session_list_state,session_id,session_name,[],current_session_name_display_value,gr.update(value=''),gr.update(value=[]))else:return (gr.update(),session_list_state,None,'',[],gr.update(value='## 对话 - 未选择会话'),gr.update(value=''),gr.update(value=[]))# 选择会话的函数def select_session(session_name, session_list_state):"""当用户选择一个已有会话时，加载对应的对话历史。"""session_id = Nonefor id, name in session_list_state:if name == session_name:session_id = idbreakif session_id:conversation = load_conversation_history(session_id)messages = []for user_msg, bot_msg, model_name in reversed(conversation):messages.insert(0, {"role": "user", "content": user_msg})messages.insert(0, {"role": "assistant", "content": f"{model_name}: {bot_msg}"})current_session_name_display_value = gr.update(value=f'## 对话 - {session_name}')return (messages,session_id,session_name,messages,gr.update(value=''),current_session_name_display_value)else:return ([],None,'',[],gr.update(value=''),gr.update(value='## 对话 - 未选择会话'))# 发送消息的函数def send_message(user_message, model_name, current_session_id, conversation_history_state):"""用户点击发送按钮后，根据当前选中的模型生成回复，并更新对话历史。"""if user_message and current_session_id:bot_response = generate_response(model_name, user_message)if bot_response:save_conversation(current_session_id, model_name, user_message, bot_response)user_msg_dict = {"role": "user", "content": user_message}bot_msg_dict = {"role": "assistant", "content": f"{model_name}: {bot_response}"}conversation_history_state.append(user_msg_dict)conversation_history_state.append(bot_msg_dict)return conversation_history_state, '', conversation_history_statereturn conversation_history_state, '', conversation_history_state# 绑定函数到事件create_session_button.click(fn=add_new_session,inputs=[session_name_input, session_list_state],outputs=[session_radio,session_list_state,current_session_id,current_session_name,conversation_history_state,current_session_name_display,user_input,chat_display])session_radio.change(fn=select_session,inputs=[session_radio, session_list_state],outputs=[chat_display,current_session_id,current_session_name,conversation_history_state,user_input,current_session_name_display])send_button.click(fn=send_message,inputs=[user_input, model_dropdown, current_session_id, conversation_history_state],outputs=[chat_display, user_input, conversation_history_state])# 当按下回车键时清空用户输入框user_input.submit(fn=send_message,inputs=[user_input, model_dropdown, current_session_id, conversation_history_state],outputs=[chat_display, user_input, conversation_history_state])if __name__ == "__main__":demo.launch()  # 启动Gradio应用