#3904 Prompt Sequence Manager
#3904 Prompt Sequence Manager
#3904 Python, SQLite3, Google Gemini API application
Read a "Prompt Set" text file including: Name, Description, Number of Prompts, <i..N> System Instructions, <i..N> Prompt
SQLite3 database with Tables: Prompt_Sets, Prompt_Steps, and Prompt_Responses
Python GUI to select Prompt Set and starting Prompt Step
Prompt manager to run the Prompt_Steps and save responses to the Prompt_Responses table, managing state from prompt to prompt in the sequence
I will generate a complete, single-file Python application. This application includes:
SQLite3 Database Management: Automatically creates and manages the
Prompt_Sets,Prompt_Steps, andPrompt_Responsestables.File Parser: A system to read a structured text file (sample included) to populate the database.
Gemini API Integration: A logic handler that runs the prompts sequentially, managing the conversation history (state) to ensure subsequent prompts know about previous answers.
Tkinter GUI: A graphical interface to select files, choose prompt sets, enter your API key, and monitor the execution.
You will need to install the Google Generative AI library:
pip install google-generativeai
import sqlite3
import tkinter as tk
from tkinter import ttk, filedialog, messagebox, scrolledtext
import google.generativeai as genai
import os
import threading
from datetime import datetime
import textwrap
# -----------------------------------------------------------------------------
# DATABASE MANAGER
# -----------------------------------------------------------------------------
class DatabaseManager:
def __init__(self, db_name="prompt_manager.db"):
self.db_name = db_name
self.conn = None
self.init_db()
def get_connection(self):
# We use check_same_thread=False for this simple GUI app to allow
# the worker thread to write to DB.
return sqlite3.connect(self.db_name, check_same_thread=False)
def init_db(self):
self.conn = self.get_connection()
cursor = self.conn.cursor()
# Table: Prompt_Sets
cursor.execute('''
CREATE TABLE IF NOT EXISTS Prompt_Sets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
description TEXT,
total_prompts INTEGER
)
''')
# Table: Prompt_Steps
cursor.execute('''
CREATE TABLE IF NOT EXISTS Prompt_Steps (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prompt_set_id INTEGER,
step_number INTEGER,
system_instruction TEXT,
prompt_text TEXT,
FOREIGN KEY(prompt_set_id) REFERENCES Prompt_Sets(id)
)
''')
# Table: Prompt_Responses
cursor.execute('''
CREATE TABLE IF NOT EXISTS Prompt_Responses (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prompt_step_id INTEGER,
response_text TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY(prompt_step_id) REFERENCES Prompt_Steps(id)
)
''')
self.conn.commit()
def insert_prompt_set(self, name, description, steps):
"""
steps: list of dicts [{'step': i, 'system': '...', 'prompt': '...'}]
"""
cursor = self.conn.cursor()
try:
# Check if exists, if so, delete old to overwrite (simplification for this app)
cursor.execute("SELECT id FROM Prompt_Sets WHERE name=?", (name,))
existing = cursor.fetchone()
if existing:
set_id = existing[0]
cursor.execute("DELETE FROM Prompt_Responses WHERE prompt_step_id IN (SELECT id FROM Prompt_Steps WHERE prompt_set_id=?)", (set_id,))
cursor.execute("DELETE FROM Prompt_Steps WHERE prompt_set_id=?", (set_id,))
cursor.execute("DELETE FROM Prompt_Sets WHERE id=?", (set_id,))
cursor.execute("INSERT INTO Prompt_Sets (name, description, total_prompts) VALUES (?, ?, ?)",
(name, description, len(steps)))
set_id = cursor.lastrowid
for step in steps:
cursor.execute('''
INSERT INTO Prompt_Steps (prompt_set_id, step_number, system_instruction, prompt_text)
VALUES (?, ?, ?, ?)
''', (set_id, step['step'], step['system'], step['prompt']))
self.conn.commit()
return True, f"Imported set '{name}' with {len(steps)} steps."
except Exception as e:
return False, str(e)
def get_all_sets(self):
cursor = self.conn.cursor()
cursor.execute("SELECT id, name, total_prompts FROM Prompt_Sets")
return cursor.fetchall()
def get_steps_for_set(self, set_id):
cursor = self.conn.cursor()
cursor.execute("SELECT id, step_number, system_instruction, prompt_text FROM Prompt_Steps WHERE prompt_set_id=? ORDER BY step_number ASC", (set_id,))
return cursor.fetchall()
def save_response(self, step_id, response):
cursor = self.conn.cursor()
cursor.execute("INSERT INTO Prompt_Responses (prompt_step_id, response_text) VALUES (?, ?)", (step_id, response))
self.conn.commit()
# -----------------------------------------------------------------------------
# FILE PARSER
# -----------------------------------------------------------------------------
class PromptFileParser:
@staticmethod
def parse_file(filepath):
"""
Expected format per block:
NAME: My Prompt Set
DESC: Description here
### STEP 1
SYS: System instruction here
PROMPT: User prompt here
### STEP 2
...
"""
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
sets_found = []
# Simple splitter by "NAME:" assuming file contains one or more sets
raw_sets = content.split('NAME:')
for raw in raw_sets:
if not raw.strip(): continue
lines = raw.strip().split('\n')
name = lines[0].strip()
desc = ""
steps = []
current_step = {}
# Simple state machine parser
mode = None # 'DESC', 'SYS', 'PROMPT'
for line in lines[1:]:
line = line.strip()
if line.startswith('DESC:'):
desc = line.replace('DESC:', '').strip()
elif line.startswith('### STEP'):
# Save previous step if exists
if current_step:
steps.append(current_step)
# Start new step
step_num = int(line.replace('### STEP', '').strip())
current_step = {'step': step_num, 'system': '', 'prompt': ''}
elif line.startswith('SYS:'):
current_step['system'] = line.replace('SYS:', '').strip()
mode = 'SYS'
elif line.startswith('PROMPT:'):
current_step['prompt'] = line.replace('PROMPT:', '').strip()
mode = 'PROMPT'
else:
# Append continuation lines
if mode == 'SYS' and current_step:
current_step['system'] += " " + line
elif mode == 'PROMPT' and current_step:
current_step['prompt'] += " " + line
# Append final step
if current_step:
steps.append(current_step)
if name and steps:
sets_found.append({'name': name, 'desc': desc, 'steps': steps})
return sets_found
except Exception as e:
raise ValueError(f"Error parsing file: {e}")
@staticmethod
def create_sample_file(filename="sample_prompts.txt"):
if os.path.exists(filename): return
content = """NAME: Creative Writer Helper
DESC: A 3-step process to generate a story, critique it, and rewrite it.
### STEP 1
SYS: You are a creative fiction writer specializing in Sci-Fi.
PROMPT: Write a short paragraph about a robot discovering a flower on Mars.
### STEP 2
SYS: You are a harsh literary critic.
PROMPT: Critique the previous story for emotional depth and scientific accuracy.
### STEP 3
SYS: You are an editor.
PROMPT: Rewrite the story based on the critique provided.
"""
with open(filename, 'w', encoding='utf-8') as f:
f.write(content)
# -----------------------------------------------------------------------------
# GUI APPLICATION
# -----------------------------------------------------------------------------
class PromptApp:
def __init__(self, root):
self.root = root
self.root.title("Gemini Pipeline Manager")
self.root.geometry("900x700")
self.db = DatabaseManager()
self.parser = PromptFileParser()
self.parser.create_sample_file() # Ensure a sample exists
self.api_key_var = tk.StringVar()
self.selected_set_id = tk.IntVar()
self.start_step_var = tk.IntVar(value=1)
self.setup_ui()
self.refresh_set_list()
def setup_ui(self):
# --- Top Frame: Config ---
config_frame = ttk.LabelFrame(self.root, text="Configuration", padding=10)
config_frame.pack(fill="x", padx=10, pady=5)
ttk.Label(config_frame, text="Gemini API Key:").pack(side="left")
ttk.Entry(config_frame, textvariable=self.api_key_var, width=40, show="*").pack(side="left", padx=5)
ttk.Button(config_frame, text="Load Prompts File", command=self.load_file).pack(side="right")
# --- Middle Frame: Selection ---
selection_frame = ttk.LabelFrame(self.root, text="Execution Control", padding=10)
selection_frame.pack(fill="x", padx=10, pady=5)
# Set List
ttk.Label(selection_frame, text="Select Prompt Set:").grid(row=0, column=0, sticky="w")
self.set_combo = ttk.Combobox(selection_frame, state="readonly", width=40)
self.set_combo.grid(row=0, column=1, padx=5, sticky="w")
self.set_combo.bind("<<ComboboxSelected>>", self.on_set_selected)
# Step Selection
ttk.Label(selection_frame, text="Start at Step:").grid(row=0, column=2, padx=5)
self.step_spin = ttk.Spinbox(selection_frame, from_=1, to=100, textvariable=self.start_step_var, width=5)
self.step_spin.grid(row=0, column=3, padx=5)
# Run Button
self.run_btn = ttk.Button(selection_frame, text="RUN PIPELINE", command=self.start_thread)
self.run_btn.grid(row=0, column=4, padx=20)
# --- Bottom Frame: Output ---
log_frame = ttk.LabelFrame(self.root, text="Execution Log & Responses", padding=10)
log_frame.pack(fill="both", expand=True, padx=10, pady=5)
self.log_area = scrolledtext.ScrolledText(log_frame, state='disabled', font=("Consolas", 10))
self.log_area.pack(fill="both", expand=True)
# Tags for coloring logs
self.log_area.tag_config("INFO", foreground="blue")
self.log_area.tag_config("USER", foreground="green")
self.log_area.tag_config("AI", foreground="purple")
self.log_area.tag_config("ERROR", foreground="red")
self.log_area.tag_config("SYS", foreground="orange")
def log(self, message, tag="INFO"):
self.log_area.config(state='normal')
timestamp = datetime.now().strftime("%H:%M:%S")
self.log_area.insert(tk.END, f"[{timestamp}] {message}\n", tag)
self.log_area.see(tk.END)
self.log_area.config(state='disabled')
def load_file(self):
filename = filedialog.askopenfilename(filetypes=[("Text Files", "*.txt")])
if filename:
try:
sets = self.parser.parse_file(filename)
count = 0
for s in sets:
success, msg = self.db.insert_prompt_set(s['name'], s['desc'], s['steps'])
if success: count += 1
self.log(f"Loaded {count} sets from file.", "INFO")
self.refresh_set_list()
except Exception as e:
messagebox.showerror("Error", str(e))
def refresh_set_list(self):
sets = self.db.get_all_sets() # [(id, name, total), ...]
self.set_map = {f"{s[1]} ({s[2]} steps)": s[0] for s in sets}
self.set_combo['values'] = list(self.set_map.keys())
if self.set_map:
self.set_combo.current(0)
self.on_set_selected(None)
def on_set_selected(self, event):
selection = self.set_combo.get()
if selection in self.set_map:
self.selected_set_id.set(self.set_map[selection])
def start_thread(self):
api_key = self.api_key_var.get().strip()
if not api_key:
messagebox.showwarning("Missing Key", "Please enter a valid Gemini API Key.")
return
set_id = self.selected_set_id.get()
start_step = self.start_step_var.get()
self.run_btn.config(state="disabled")
thread = threading.Thread(target=self.run_pipeline, args=(api_key, set_id, start_step))
thread.daemon = True
thread.start()
def run_pipeline(self, api_key, set_id, start_step):
try:
genai.configure(api_key=api_key)
# Using 1.5 Flash as it is efficient and generally available
model = genai.GenerativeModel('gemini-1.5-flash')
steps = self.db.get_steps_for_set(set_id) # [(id, step_num, sys, prompt), ...]
# Filter steps based on start_step
steps_to_run = [s for s in steps if s[1] >= start_step]
if not steps_to_run:
self.log("No steps found starting from " + str(start_step), "ERROR")
return
# Chat History (State Management)
# We will use a list of content dicts to manually manage history
# so we can inject specific system instructions into the flow context
chat_history = []
for step in steps_to_run:
step_id, step_num, sys_instr, user_prompt = step
self.log(f"--- Running Step {step_num} ---", "INFO")
# Constructing the message for this turn.
# To ensure the 'System Instruction' for this specific step is respected
# alongside the history, we combine it with the user prompt.
# (Gemini 1.5 Flash supports system_instruction at init, but changing it
# mid-chat requires a new session or context injection. Context injection is safer here).
combined_prompt = f"System Instruction: {sys_instr}\n\nTask: {user_prompt}"
self.log(f"System: {sys_instr}", "SYS")
self.log(f"Prompt: {user_prompt}", "USER")
# Send to Gemini
try:
chat = model.start_chat(history=chat_history)
response = chat.send_message(combined_prompt)
response_text = response.text
self.log(f"Response: {response_text}", "AI")
# Save to DB
self.db.save_response(step_id, response_text)
# Update manual history for next iteration's context
# Note: We append the raw prompt and the response to history
chat_history.append({'role': 'user', 'parts': [combined_prompt]})
chat_history.append({'role': 'model', 'parts': [response_text]})
except Exception as e:
self.log(f"API Error on step {step_num}: {str(e)}", "ERROR")
break # Stop sequence on error
self.log("--- Pipeline Complete ---", "INFO")
except Exception as e:
self.log(f"Critical Error: {str(e)}", "ERROR")
finally:
self.root.after(0, lambda: self.run_btn.config(state="normal"))
# -----------------------------------------------------------------------------
# MAIN
# -----------------------------------------------------------------------------
if __name__ == "__main__":
root = tk.Tk()
app = PromptApp(root)
root.mainloop()
From <https://gemini.google.com/app/3721cb314cf7f13f> Google Gemini 3.0 Pro (Thinking/Canvas)
Comments
Post a Comment