App

app

---

Assembly:                Leeroy
Filename:                app.py
Author:                  Terry D. Eppler
Created:                 05-31-2024

Last Modified By:        Terry D. Eppler
Last Modified On:        05-01-2025

---

       Leeroy is a data analysis tool integrating various Generative GPT, Text-Processing, and
       Machine-Learning algorithms for federal analysts.
       Copyright ©  2022  Terry Eppler

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

You can contact me at: terryeppler@gmail.com or eppler.terry@epa.gov

app.py

---

write_error

write_error(
    error: Exception, cause: str, method: str
) -> None

Write an application error record without interrupting fallback handling.

Purpose

Wraps a caught exception in the project Error object and writes it through the SQLite-backed Logger. The helper is used by existing fallback handlers that must preserve their original return behavior while still recording diagnostic metadata.

Parameters:

Name	Type	Description	Default
error	Exception	Caught exception instance to wrap and persist.	required
cause	str	Logical workflow component associated with the failure.	required
method	str	Stable function or method signature associated with the failure.	required

Source code in app.py

def write_error( error: Exception, cause: str, method: str ) -> None:
	"""Write an application error record without interrupting fallback handling.

	Purpose:
		Wraps a caught exception in the project ``Error`` object and writes it through the
		SQLite-backed ``Logger``. The helper is used by existing fallback handlers that must
		preserve their original return behavior while still recording diagnostic metadata.

	Args:
		error: Caught exception instance to wrap and persist.
		cause: Logical workflow component associated with the failure.
		method: Stable function or method signature associated with the failure.
	"""
	try:
		exception = Error( error )
		exception.module = 'app'
		exception.cause = cause
		exception.method = method
		Logger( ).write( exception )
	except Exception:
		return None

local_llm_available

local_llm_available() -> bool

Determine whether the configured local model file is available.

Purpose

Checks the resolved model-path state created during module import and returns whether the optional local GGUF file exists in the current execution environment. This helper allows the Streamlit UI and local model loader to degrade safely when the model file is not present.

Returns:

Name	Type	Description
bool	bool	Return value produced by the operation.

Source code in app.py

def local_llm_available( ) -> bool:
	"""Determine whether the configured local model file is available.

	Purpose:
		Checks the resolved model-path state created during module import and returns whether
		the optional local GGUF file exists in the current execution environment. This helper
		allows the Streamlit UI and local model loader to degrade safely when the model file
		is not present.

	Returns:
		bool: Return value produced by the operation.
	"""
	try:
		return bool( LOCAL_MODEL_AVAILABLE )
	except Exception as e:
		write_error( e, 'local_llm', 'local_llm_available( ) -> bool' )
		return False

throw_if

throw_if(name: str, value: object) -> None

Throw if.

Purpose

Validates that a required argument contains a usable value before the surrounding workflow continues. This guard centralizes early validation so provider wrappers and UI routines fail with consistent, readable error messages.

Parameters:

Name	Type	Description	Default
name	str	Name value used by the operation.	required
value	object	Value value used by the operation.	required

Returns:

Name	Type	Description
None	None	This function performs its work through side effects and does not return a value.

Source code in app.py

def throw_if( name: str, value: object ) -> None:
	"""Throw if.

	Purpose:
	    Validates that a required argument contains a usable value before the surrounding workflow
	    continues. This guard centralizes early validation so provider wrappers and UI routines
	    fail
	    with consistent, readable error messages.

	Args:
	    name (str): Name value used by the operation.
	    value (object): Value value used by the operation.

	Returns:
	    None: This function performs its work through side effects and does not return a value."""
	if value is None:
		raise ValueError( f'Argument "{name}" cannot be None.' )

	if isinstance( value, str ) and not value.strip( ):
		raise ValueError( f'Argument "{name}" cannot be empty.' )

image_to_base64

image_to_base64(path: str) -> str

Convert an image file to a Base64 text string.

Purpose

Reads a local image file and converts the raw bytes into a Base64-encoded string that can be embedded in Streamlit or Markdown output. The helper supports UI image rendering workflows that need inline image data.

Parameters:

Name	Type	Description	Default
path	str	Filesystem path to the image file.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def image_to_base64( path: str ) -> str:
	"""Convert an image file to a Base64 text string.

	Purpose:
		Reads a local image file and converts the raw bytes into a Base64-encoded string that
		can be embedded in Streamlit or Markdown output. The helper supports UI image
		rendering workflows that need inline image data.

	Args:
		path: Filesystem path to the image file.

	Returns:
		str: Return value produced by the operation.
	"""
	with open( path, "rb" ) as f:
		return base64.b64encode( f.read( ) ).decode( )

cosine_sim

cosine_sim(a: ndarray, b: ndarray) -> float

Calculate cosine similarity between two vectors.

Purpose

Computes normalized dot-product similarity for semantic retrieval, document chunk ranking, and fallback vector matching when sqlite-vec is unavailable. The helper returns a safe zero score when either vector has zero magnitude.

Parameters:

Name	Type	Description	Default
a	ndarray	First vector.	required
b	ndarray	Second vector.	required

Returns:

Name	Type	Description
float	float	Return value produced by the operation.

Source code in app.py

def cosine_sim( a: np.ndarray, b: np.ndarray ) -> float:
	"""Calculate cosine similarity between two vectors.

	Purpose:
		Computes normalized dot-product similarity for semantic retrieval, document chunk
		ranking, and fallback vector matching when sqlite-vec is unavailable. The helper
		returns a safe zero score when either vector has zero magnitude.

	Args:
		a: First vector.
		b: Second vector.

	Returns:
		float: Return value produced by the operation.
	"""
	denom = np.linalg.norm( a ) * np.linalg.norm( b )
	return float( np.dot( a, b ) / denom ) if denom else 0.0

initialize_database

initialize_database() -> None

Create required application database tables.

Purpose

Ensures the local SQLite storage directory exists and creates the tables required for chat history, semantic embeddings, and categorized prompt templates.

Raises:

Type	Description
Error	Raised when SQLite initialization fails after writing diagnostic metadata.

Source code in app.py

def initialize_database( ) -> None:
	"""Create required application database tables.

	Purpose:
		Ensures the local SQLite storage directory exists and creates the tables required for
		chat history, semantic embeddings, and categorized prompt templates.

	Raises:
		Error: Raised when SQLite initialization fails after writing diagnostic metadata.
	"""
	Path( 'stores/sqlite' ).mkdir( parents=True, exist_ok=True )
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		conn.execute(
			"""
            CREATE TABLE IF NOT EXISTS chat_history
            (
                id
                INTEGER
                PRIMARY
                KEY
                AUTOINCREMENT,
                role
                TEXT,
                content
                TEXT
            )
			"""
		)

		conn.execute(
			"""
            CREATE TABLE IF NOT EXISTS embeddings
            (
                id
                INTEGER
                PRIMARY
                KEY
                AUTOINCREMENT,
                chunk
                TEXT,
                vector
                BLOB
            )
			"""
		)

		conn.execute(
			"""
            CREATE TABLE IF NOT EXISTS Prompts
            (
				ID INTEGER NOT NULL UNIQUE,
				Caption TEXT(80),
				Name TEXT(80),
				Category TEXT(80),
				Text TEXT(2048),
				PRIMARY KEY(ID AUTOINCREMENT)
			)
			"""
		)

		conn.commit( )

normalize_text

normalize_text(text: str) -> str

Normalize text for matching and comparison workflows.

Purpose

Standardizes free text by lowercasing content, removing punctuation except sentence delimiters, normalizing sentence-boundary spacing, and collapsing repeated whitespace. The helper supports prompt, document, and search workflows that need consistent text comparison behavior.

Parameters:

Name	Type	Description	Default
text	str	Source text to process.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def normalize_text( text: str ) -> str:
	"""Normalize text for matching and comparison workflows.

	Purpose:
		Standardizes free text by lowercasing content, removing punctuation except sentence
		delimiters, normalizing sentence-boundary spacing, and collapsing repeated whitespace.
		The helper supports prompt, document, and search workflows that need consistent text
		comparison behavior.

	Args:
		text: Source text to process.

	Returns:
		str: Return value produced by the operation.
	"""
	if not text:
		return ""

	# Lowercase
	text = text.lower( )

	# Remove punctuation except . ! ?
	text = re.sub( r"[^\w\s\.\!\?]", "", text )

	# Ensure single space after sentence delimiters
	text = re.sub( r"([.!?])\s*", r"\1 ", text )

	# Normalize whitespace
	text = re.sub( r"\s+", " ", text ).strip( )

	return text

chunk_text

chunk_text(
    text: str, size: int = 1200, overlap: int = 200
) -> List[str]

Split text into overlapping chunks.

Purpose

Creates overlapping text windows used by semantic indexing, retrieval-augmented generation, and document Q&A workflows. The overlap preserves local context across chunk boundaries so retrieved excerpts remain coherent.

Parameters:

Name	Type	Description	Default
text	str	Source text to process.	required
size	int	Maximum character length for each chunk.	1200
overlap	int	Number of characters shared between adjacent chunks.	200

Returns:

Type	Description
List[str]	list[str]: Return value produced by the operation.

Source code in app.py

def chunk_text( text: str, size: int = 1200, overlap: int = 200 ) -> List[ str ]:
	"""Split text into overlapping chunks.

	Purpose:
		Creates overlapping text windows used by semantic indexing, retrieval-augmented
		generation, and document Q&A workflows. The overlap preserves local context across
		chunk boundaries so retrieved excerpts remain coherent.

	Args:
		text: Source text to process.
		size: Maximum character length for each chunk.
		overlap: Number of characters shared between adjacent chunks.

	Returns:
		list[str]: Return value produced by the operation.
	"""
	chunks, i = [ ], 0
	while i < len( text ):
		chunks.append( text[ i:i + size ] )
		i += size - overlap
	return chunks

convert_xml

convert_xml(text: str) -> str

Convert XML-like prompt sections into Markdown.

Purpose

Transforms prompt text containing XML-like opening and closing tags into Markdown section blocks. The function treats tags as lightweight section delimiters instead of strict XML, allowing prompt templates to be rendered more clearly in the UI.

Parameters:

Name	Type	Description	Default
text	str	Source text to process.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def convert_xml( text: str ) -> str:
	"""Convert XML-like prompt sections into Markdown.

	Purpose:
		Transforms prompt text containing XML-like opening and closing tags into Markdown
		section blocks. The function treats tags as lightweight section delimiters instead of
		strict XML, allowing prompt templates to be rendered more clearly in the UI.

	Args:
		text: Source text to process.

	Returns:
		str: Return value produced by the operation.
	"""
	markdown_blocks: List[ str ] = [ ]
	for match in cfg.XML_BLOCK_PATTERN.finditer( text ):
		raw_tag: str = match.group( "tag" )
		body: str = match.group( "body" ).strip( )

		# Humanize tag name for Markdown heading
		heading: str = raw_tag.replace( "_", " " ).replace( "-", " " ).title( )
		markdown_blocks.append( f"## {heading}" )
		if body:
			markdown_blocks.append( body )
	return "\n\n".join( markdown_blocks )

markdown_converter

markdown_converter(text: Any) -> str

Convert between Markdown headings and XML-like heading tags.

Purpose

Auto-detects whether the supplied text contains simple hN heading tags or Markdown heading syntax. HTML-like headings are converted to Markdown headings; otherwise Markdown headings are converted to matching hN tags. Non-string or empty inputs return an empty string.

Parameters:

Name	Type	Description	Default
text	Any	Source text to process.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def markdown_converter( text: Any ) -> str:
	"""Convert between Markdown headings and XML-like heading tags.

	Purpose:
		Auto-detects whether the supplied text contains simple hN heading tags or Markdown
		heading syntax. HTML-like headings are converted to Markdown headings; otherwise
		Markdown headings are converted to matching hN tags. Non-string or empty inputs return
		an empty string.

	Args:
		text: Source text to process.

	Returns:
		str: Return value produced by the operation.
	"""
	if not isinstance( text, str ) or not text.strip( ):
		return ""

	# Normalize newlines
	src = text.replace( "\r\n", "\n" ).replace( "\r", "\n" )

	htag_pattern = re.compile( r"<h([1-6])>(.*?)</h\1>", flags=re.IGNORECASE | re.DOTALL )
	md_heading_pattern = re.compile( r"^(#{1,6})[ \t]+(.+?)[ \t]*$", flags=re.MULTILINE )

	# ------------------------------------------------------------------
	# Direction detection
	# ------------------------------------------------------------------
	contains_htags = bool( htag_pattern.search( src ) )

	# ------------------------------------------------------------------
	# XML-like heading tags -> Markdown headings
	# ------------------------------------------------------------------
	if contains_htags:
		def _htag_to_md( match: re.Match ) -> str:
			level = int( match.group( 1 ) )
			content = match.group( 2 ).strip( )

			# Preserve inner newlines safely by collapsing interior whitespace
			# while keeping content readable.
			content = re.sub( r"[ \t]+\n", "\n", content )
			content = re.sub( r"\n[ \t]+", "\n", content )

			return f"{'#' * level} {content}"

		out = htag_pattern.sub( _htag_to_md, src )
		return out.strip( )

	# ------------------------------------------------------------------
	# Markdown headings -> XML-like heading tags
	# ------------------------------------------------------------------
	def _md_to_htag( match: re.Match ) -> str:
		hashes = match.group( 1 )
		content = match.group( 2 ).strip( )
		level = len( hashes )
		return f"<h{level}>{content}</h{level}>"

	out = md_heading_pattern.sub( _md_to_htag, src )
	return out.strip( )

inject_response_css

inject_response_css() -> None

Inject chat-response CSS into the Streamlit page.

Purpose

Adds inline CSS that styles chat-message paragraphs, headings, and links inside Streamlit chat responses. The style layer keeps generated responses visually consistent with the Leeroy dark-mode interface and shared blue accent color.

Source code in app.py

def inject_response_css( ) -> None:
	"""Inject chat-response CSS into the Streamlit page.

	Purpose:
		Adds inline CSS that styles chat-message paragraphs, headings, and links inside
		Streamlit chat responses. The style layer keeps generated responses visually
		consistent with the Leeroy dark-mode interface and shared blue accent color.
	"""
	st.markdown(
		"""
		<style>
		/* Chat message text */
		.stChatMessage p {
			color: rgb(220, 220, 220);
			font-size: 1rem;
			line-height: 1.6;
		}

		/* Headings inside chat responses */
		.stChatMessage h1 {
			color: rgb(0, 120, 252); /* DoD Blue */
			font-size: 1.6rem;
		}

		.stChatMessage h2 {
			color: rgb(0, 120, 252);
			font-size: 1.35rem;
		}

		.stChatMessage h3 {
			color: rgb(0, 120, 252);
			font-size: 1.15rem;
		}

		.stChatMessage a {
			color: rgb(0, 120, 252); /* DoD Blue */
			text-decoration: underline;
		}

		.stChatMessage a:hover {
			color: rgb(80, 160, 255);
		}

		</style>
		""", unsafe_allow_html=True )

style_subheaders

style_subheaders() -> None

Inject shared subheader CSS into the Streamlit page.

Purpose

Applies the Leeroy blue accent color to selected Markdown and chat subheaders in the main Streamlit UI. The helper centralizes visual styling used by sidebar and mode- rendering sections.

Source code in app.py

def style_subheaders( ) -> None:
	"""Inject shared subheader CSS into the Streamlit page.

	Purpose:
		Applies the Leeroy blue accent color to selected Markdown and chat subheaders in the
		main Streamlit UI. The helper centralizes visual styling used by sidebar and mode-
		rendering sections.
	"""
	st.markdown(
		"""
		<style>
		div[data-testid="stMarkdownContainer"] h2,
		div[data-testid="stMarkdownContainer"] h3,
		div[data-testid="stChatMessage"] div[data-testid="stMarkdownContainer"] h2,
		div[data-testid="stChatMessage"] div[data-testid="stMarkdownContainer"] h3 {
			color: rgb(0, 120, 252) !important;
		}
		</style>
		""",
		unsafe_allow_html=True, )

save_message

save_message(role: str, content: str) -> None

Persist a chat message to SQLite.

Purpose

Writes a single chat-history record to the configured application database. The function supports durable conversation history for the Text Generation and Document Q&A modes.

Parameters:

Name	Type	Description	Default
role	str	Chat role associated with the message.	required
content	str	Message content to store.	required

Source code in app.py

def save_message( role: str, content: str ) -> None:
	"""Persist a chat message to SQLite.

	Purpose:
		Writes a single chat-history record to the configured application database. The
		function supports durable conversation history for the Text Generation and Document
		Q&A modes.

	Args:
		role: Chat role associated with the message.
		content: Message content to store.
	"""
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		conn.execute( 'INSERT INTO chat_history (role, content) VALUES (?, ?)', (role, content) )

load_history

load_history() -> List[Tuple[str, str]]

Load persisted chat history from SQLite.

Purpose

Reads stored chat messages in insertion order so Streamlit session state can be reconstructed when the application starts or reruns.

Returns:

Type	Description
List[Tuple[str, str]]	list[tuple[str, str]]: Return value produced by the operation.

Source code in app.py

def load_history( ) -> List[ Tuple[ str, str ] ]:
	"""Load persisted chat history from SQLite.

	Purpose:
		Reads stored chat messages in insertion order so Streamlit session state can be
		reconstructed when the application starts or reruns.

	Returns:
		list[tuple[str, str]]: Return value produced by the operation.
	"""
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		return conn.execute( 'SELECT role, content FROM chat_history ORDER BY id' ).fetchall( )

clear_history

clear_history() -> None

Delete persisted chat history.

Purpose

Removes all rows from the SQLite chat_history table so the Streamlit chat UI can be reset without affecting prompt templates, embeddings, or imported data.

Source code in app.py

def clear_history( ) -> None:
	"""Delete persisted chat history.

	Purpose:
		Removes all rows from the SQLite chat_history table so the Streamlit chat UI can be
		reset without affecting prompt templates, embeddings, or imported data.
	"""
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		conn.execute( "DELETE FROM chat_history" )

supports_prompt_category

supports_prompt_category(category: str) -> bool

Determine whether a prompt category fits the local text model.

Purpose

Excludes categories that clearly require unsupported image, audio, video, speech, or hosted-tool capabilities while retaining text, code, retrieval, and document workflows.

Parameters:

Name	Type	Description	Default
category	str	Stored prompt category name.	required

Returns:

Name	Type	Description
bool	bool	True when the category is suitable for Leeroy's local Llama 3.2 model.

Source code in app.py

def supports_prompt_category( category: str ) -> bool:
	"""Determine whether a prompt category fits the local text model.

	Purpose:
		Excludes categories that clearly require unsupported image, audio, video, speech, or
		hosted-tool capabilities while retaining text, code, retrieval, and document workflows.

	Args:
		category: Stored prompt category name.

	Returns:
		bool: True when the category is suitable for Leeroy's local Llama 3.2 model.
	"""
	unsupported_terms = ( 'audio', 'computer use', 'hosted tool', 'image', 'speech',
		'text-to-speech', 'video', 'vision' )
	category_name = category.casefold( )
	return not any( term in category_name for term in unsupported_terms )

fetch_prompt_categories

fetch_prompt_categories(db_path: str) -> List[str]

Retrieve prompt categories supported by the local text model.

Purpose

Reads distinct category names from the configured Prompts table and returns the alphabetically ordered categories that fit Leeroy's text-only local model.

Parameters:

Name	Type	Description	Default
db_path	str	SQLite database path.	required

Returns:

Type	Description
List[str]	list[str]: Supported prompt categories.

Source code in app.py

def fetch_prompt_categories( db_path: str ) -> List[ str ]:
	"""Retrieve prompt categories supported by the local text model.

	Purpose:
		Reads distinct category names from the configured Prompts table and returns the
		alphabetically ordered categories that fit Leeroy's text-only local model.

	Args:
		db_path: SQLite database path.

	Returns:
		list[str]: Supported prompt categories.
	"""
	try:
		with sqlite3.connect( db_path ) as conn:
			rows = conn.execute(
				"SELECT DISTINCT Category FROM Prompts "
				"WHERE Category IS NOT NULL AND TRIM(Category) <> '' ORDER BY Category;"
			).fetchall( )
		return [ str( row[ 0 ] ) for row in rows if supports_prompt_category( str( row[ 0 ] ) ) ]
	except Exception as e:
		write_error( e, 'prompts', 'fetch_prompt_categories( db_path: str ) -> List[str]' )
		return [ ]

fetch_prompt_choices

fetch_prompt_choices(
    db_path: str, category: str
) -> List[Tuple[int, str]]

Retrieve prompt identifiers and display labels for a category.

Purpose

Loads category-filtered prompt choices using the primary key as the selector value and Caption plus Name as the human-readable label.

Parameters:

Name	Type	Description	Default
db_path	str	SQLite database path.	required
category	str	Selected prompt category.	required

Returns:

Type	Description
List[Tuple[int, str]]	list[tuple[int, str]]: Prompt identifiers paired with display labels.

Source code in app.py

def fetch_prompt_choices( db_path: str, category: str ) -> List[ Tuple[ int, str ] ]:
	"""Retrieve prompt identifiers and display labels for a category.

	Purpose:
		Loads category-filtered prompt choices using the primary key as the selector value and
		Caption plus Name as the human-readable label.

	Args:
		db_path: SQLite database path.
		category: Selected prompt category.

	Returns:
		list[tuple[int, str]]: Prompt identifiers paired with display labels.
	"""
	if not category:
		return [ ]

	try:
		with sqlite3.connect( db_path ) as conn:
			rows = conn.execute(
				"SELECT ID, Caption, Name FROM Prompts WHERE Category = ? "
				"ORDER BY Caption, Name, ID;", (category,) ).fetchall( )
		choices: List[ Tuple[ int, str ] ] = [ ]
		for prompt_id, caption, name in rows:
			caption_text = str( caption or '' ).strip( )
			name_text = str( name or '' ).strip( )
			label = caption_text
			if name_text and name_text != caption_text:
				label = f'{caption_text} — {name_text}' if caption_text else name_text
			if not label:
				label = f'Prompt {prompt_id}'
			choices.append( (int( prompt_id ), label) )
		return choices
	except Exception as e:
		write_error( e, 'prompts',
			'fetch_prompt_choices( db_path: str, category: str ) -> List[Tuple[int, str]]' )
		return [ ]

fetch_prompts

fetch_prompts() -> pd.DataFrame

Load prompt metadata for prompt administration.

Purpose

Reads prompt metadata from the configured SQLite Prompts table, orders rows by most recent identifier first, and inserts a selection flag used by the Prompt Engineering data editor.

Returns:

Type	Description
DataFrame	pd.DataFrame: Return value produced by the operation.

Source code in app.py

def fetch_prompts( ) -> pd.DataFrame:
	"""Load prompt metadata for prompt administration.

	Purpose:
		Reads prompt metadata from the configured SQLite Prompts table, orders rows by most
		recent identifier first, and inserts a selection flag used by the Prompt Engineering
		data editor.

	Returns:
		pd.DataFrame: Return value produced by the operation.
	"""
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		df = pd.read_sql_query(
			"SELECT ID, Caption, Name, Category, Text FROM Prompts ORDER BY ID DESC",
			conn )
	df.insert( 0, "Selected", False )
	return df

fetch_prompt_by_id

fetch_prompt_by_id(pid: int) -> Dict[str, Any] | None

Load a prompt record by primary key.

Purpose

Retrieves the full prompt record for a selected ID value and returns a dictionary keyed by SQLite column names for prompt-editing workflows.

Parameters:

Name	Type	Description	Default
pid	int	Prompt primary key value.	required

Returns:

Type	Description
Dict[str, Any] | None	dict[str, Any] | None: Return value produced by the operation.

Source code in app.py

def fetch_prompt_by_id( pid: int ) -> Dict[ str, Any ] | None:
	"""Load a prompt record by primary key.

	Purpose:
		Retrieves the full prompt record for a selected ID value and returns a
		dictionary keyed by SQLite column names for prompt-editing workflows.

	Args:
		pid: Prompt primary key value.

	Returns:
		dict[str, Any] | None: Return value produced by the operation.
	"""
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		cur = conn.execute(
			"SELECT ID, Caption, Name, Category, Text FROM Prompts WHERE ID=?",
			(pid,)
		)
		row = cur.fetchone( )
		return dict( zip( [ c[ 0 ] for c in cur.description ], row ) ) if row else None

fetch_prompt_by_name

fetch_prompt_by_name(name: str) -> Dict[str, Any] | None

Load a prompt record by caption.

Purpose

Retrieves the full prompt record for a selected prompt caption and returns a dictionary keyed by SQLite column names for template-loading workflows.

Parameters:

Name	Type	Description	Default
name	str	Prompt caption or object name used by the operation.	required

Returns:

Type	Description
Dict[str, Any] | None	dict[str, Any] | None: Return value produced by the operation.

Source code in app.py

def fetch_prompt_by_name( name: str ) -> Dict[ str, Any ] | None:
	"""Load a prompt record by caption.

	Purpose:
		Retrieves the full prompt record for a selected prompt caption and returns a
		dictionary keyed by SQLite column names for template-loading workflows.

	Args:
		name: Prompt caption or object name used by the operation.

	Returns:
		dict[str, Any] | None: Return value produced by the operation.
	"""
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		cur = conn.execute(
			"SELECT ID, Caption, Name, Category, Text FROM Prompts WHERE Caption=? ORDER BY ID",
			(name,)
		)
		row = cur.fetchone( )
		return dict( zip( [ c[ 0 ] for c in cur.description ], row ) ) if row else None

insert_prompt

insert_prompt(data: Dict[str, Any]) -> None

Insert a prompt record.

Purpose

Writes a new prompt-template record to the configured SQLite Prompts table using the fields collected by the Prompt Engineering UI.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Prompt field dictionary.	required

Source code in app.py

def insert_prompt( data: Dict[ str, Any ] ) -> None:
	"""Insert a prompt record.

	Purpose:
		Writes a new prompt-template record to the configured SQLite Prompts table using the
		fields collected by the Prompt Engineering UI.

	Args:
		data: Prompt field dictionary.
	"""
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		conn.execute(
			'INSERT INTO Prompts (Caption, Name, Category, Text) VALUES (?, ?, ?, ?)',
			(data[ 'Caption' ], data[ 'Name' ], data[ 'Category' ], data[ 'Text' ]) )

update_prompt

update_prompt(pid: int, data: Dict[str, Any]) -> None

Update an existing prompt record.

Purpose

Updates an existing prompt-template record in the configured SQLite Prompts table using the selected ID value and edited field values.

Parameters:

Name	Type	Description	Default
pid	int	Prompt primary key value.	required
data	Dict[str, Any]	Prompt field dictionary.	required

Source code in app.py

def update_prompt( pid: int, data: Dict[ str, Any ] ) -> None:
	"""Update an existing prompt record.

	Purpose:
		Updates an existing prompt-template record in the configured SQLite Prompts table
		using the selected ID value and edited field values.

	Args:
		pid: Prompt primary key value.
		data: Prompt field dictionary.
	"""
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		conn.execute(
			"UPDATE Prompts SET Caption=?, Name=?, Category=?, Text=? WHERE ID=?",
			(data[ 'Caption' ], data[ 'Name' ], data[ 'Category' ], data[ 'Text' ], pid)
		)

delete_prompt

delete_prompt(pid: int) -> None

Delete a prompt record.

Purpose

Removes a prompt-template record from the configured SQLite Prompts table using the selected ID value.

Parameters:

Name	Type	Description	Default
pid	int	Prompt primary key value.	required

Source code in app.py

def delete_prompt( pid: int ) -> None:
	"""Delete a prompt record.

	Purpose:
		Removes a prompt-template record from the configured SQLite Prompts table using the
		selected ID value.

	Args:
		pid: Prompt primary key value.
	"""
	with sqlite3.connect( cfg.DB_PATH ) as conn:
		conn.execute( "DELETE FROM Prompts WHERE ID=?", (pid,) )

clear_active_prompt_metadata

clear_active_prompt_metadata() -> None

Clear metadata for the active system-instruction template.

Purpose

Resets the shared prompt identity fields without changing conversation, document, retrieval, or inference state.

Returns:

Name	Type	Description
None	None	This function performs its work through side effects and does not return a value.

Source code in app.py

def clear_active_prompt_metadata( ) -> None:
	"""Clear metadata for the active system-instruction template.

	Purpose:
		Resets the shared prompt identity fields without changing conversation, document,
		retrieval, or inference state.

	Returns:
		None: This function performs its work through side effects and does not return a value.
	"""
	st.session_state[ 'active_prompt_id' ] = None
	st.session_state[ 'active_prompt_caption' ] = ''
	st.session_state[ 'active_prompt_name' ] = ''
	st.session_state[ 'active_prompt_category' ] = ''

render_system_instructions

render_system_instructions(
    category_key: str, prompt_key: str, clear_key: str
) -> None

Render a categorized System Instructions expander.

Purpose

Displays the shared editable system text with mode-specific category and ID-backed template selectors. The selector exposes only categories supported by Leeroy's local text model while keeping selection state independent between application modes.

Parameters:

Name	Type	Description	Default
category_key	str	Session-state key for the mode-specific category selector.	required
prompt_key	str	Session-state key for the mode-specific prompt selector.	required
clear_key	str	Widget key for the mode-specific clear button.	required

Returns:

Name	Type	Description
None	None	This function performs its work through side effects and does not return a value.

Source code in app.py

def render_system_instructions( category_key: str, prompt_key: str, clear_key: str ) -> None:
	"""Render a categorized System Instructions expander.

	Purpose:
		Displays the shared editable system text with mode-specific category and ID-backed
		template selectors. The selector exposes only categories supported by Leeroy's local
		text model while keeping selection state independent between application modes.

	Args:
		category_key: Session-state key for the mode-specific category selector.
		prompt_key: Session-state key for the mode-specific prompt selector.
		clear_key: Widget key for the mode-specific clear button.

	Returns:
		None: This function performs its work through side effects and does not return a value.
	"""
	categories = fetch_prompt_categories( cfg.DB_PATH )
	selected_category = st.session_state.get( category_key )
	if selected_category not in categories:
		st.session_state[ category_key ] = None
		selected_category = None

	prompt_choices = fetch_prompt_choices( cfg.DB_PATH, str( selected_category or '' ) )
	prompt_labels = { prompt_id: label for prompt_id, label in prompt_choices }
	prompt_ids = list( prompt_labels.keys( ) )
	if st.session_state.get( prompt_key ) not in prompt_ids:
		st.session_state[ prompt_key ] = None

	def on_category_change( ) -> None:
		st.session_state[ prompt_key ] = None
		clear_active_prompt_metadata( )

	def on_template_change( ) -> None:
		prompt_id = st.session_state.get( prompt_key )
		if prompt_id is None:
			clear_active_prompt_metadata( )
			return None

		prompt_record = fetch_prompt_by_id( int( prompt_id ) )
		if prompt_record is None:
			clear_active_prompt_metadata( )
			return None

		st.session_state[ 'system_instructions' ] = str( prompt_record.get( 'Text' ) or '' )
		st.session_state[ 'active_prompt_id' ] = int( prompt_record[ 'ID' ] )
		st.session_state[ 'active_prompt_caption' ] = str(
			prompt_record.get( 'Caption' ) or '' )
		st.session_state[ 'active_prompt_name' ] = str( prompt_record.get( 'Name' ) or '' )
		st.session_state[ 'active_prompt_category' ] = str(
			prompt_record.get( 'Category' ) or '' )

	def on_clear( ) -> None:
		st.session_state[ 'system_instructions' ] = ''
		st.session_state[ category_key ] = None
		st.session_state[ prompt_key ] = None
		clear_active_prompt_metadata( )

	with st.expander( label='System Instructions', icon='🖥️', expanded=False, width='stretch' ):
		ins_left, ins_right = st.columns( [ 0.8, 0.2 ] )

		with ins_left:
			st.text_area( label='Enter Text', height=50, width='stretch',
				help=cfg.SYSTEM_INSTRUCTIONS, key='system_instructions' )

		with ins_right:
			st.selectbox( label='Category', options=categories, index=None, key=category_key,
				on_change=on_category_change, placeholder='Select category' )
			st.selectbox( label='Use Template', options=prompt_ids, index=None, key=prompt_key,
				format_func=lambda prompt_id: prompt_labels.get( prompt_id, str( prompt_id ) ),
				on_change=on_template_change, disabled=selected_category is None,
				placeholder='Select template' )

		st.button( label='Clear Instructions', width='stretch', key=clear_key,
			on_click=on_clear )

build_prompt

build_prompt(user_input: str) -> str

Build a llama.cpp-compatible chat prompt.

Purpose

Combines system instructions, optional semantic retrieval context, basic document context, and in-memory chat history into the chat-template prompt consumed by the local llama.cpp model.

Parameters:

Name	Type	Description	Default
user_input	str	User message or constructed prompt text for the current generation turn.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def build_prompt( user_input: str ) -> str:
	"""Build a llama.cpp-compatible chat prompt.

	Purpose:
		Combines system instructions, optional semantic retrieval context, basic document
		context, and in-memory chat history into the chat-template prompt consumed by the
		local llama.cpp model.

	Args:
		user_input: User message or constructed prompt text for the current generation turn.

	Returns:
		str: Return value produced by the operation.
	"""
	system_instructions = st.session_state.get( 'system_instructions', '' )
	use_semantic = bool( st.session_state.get( 'use_semantic', False ) )
	basic_docs = st.session_state.get( 'basic_docs', [ ] )
	messages = st.session_state.get( 'messages', [ ] )

	top_k_value = int( st.session_state.get( 'top_k', 0 ) )
	if top_k_value <= 0:
		top_k_value = 4

	prompt = f"<|system|>\n{system_instructions}\n</s>\n"

	if use_semantic:
		local_embedder = get_embedder( )
		if local_embedder is not None:
			with sqlite3.connect( cfg.DB_PATH ) as conn:
				rows = conn.execute( "SELECT chunk, vector FROM embeddings" ).fetchall( )

			if rows:
				q = local_embedder.encode( [ user_input ] )[ 0 ]
				scored = [ (c, cosine_sim( q, np.frombuffer( v ) )) for c, v in rows ]
				for c, _ in sorted( scored, key=lambda x: x[ 1 ], reverse=True )[ :top_k_value ]:
					prompt += f"<|system|>\n{c}\n</s>\n"

	for d in basic_docs[ :6 ]:
		prompt += f"<|system|>\n{d}\n</s>\n"

	if isinstance( messages, list ):
		for msg in messages:
			role = ''
			content = ''

			if isinstance( msg, tuple ) or isinstance( msg, list ):
				if len( msg ) == 2:
					role = str( msg[ 0 ] or '' ).strip( )
					content = str( msg[ 1 ] or '' )
			elif isinstance( msg, dict ):
				role = str( msg.get( 'role', '' ) or '' ).strip( )
				content = str( msg.get( 'content', '' ) or '' )

			if role:
				prompt += f"<|{role}|>\n{content}\n</s>\n"

	prompt += f"<|user|>\n{user_input}\n</s>\n<|assistant|>\n"
	return prompt

run_llm_turn

run_llm_turn(
    user_input: str,
    temperature: float,
    top_p: float,
    repeat_penalty: float,
    max_tokens: int,
    stream: bool,
    output: Any | None = None,
) -> str

Run one local LLM generation turn.

Purpose

Builds the shared prompt, loads the optional local llama.cpp model, and either streams generated tokens into a Streamlit placeholder or returns the completed response text for downstream workflows.

Parameters:

Name	Type	Description	Default
user_input	str	User message or constructed prompt text for the current generation turn.	required
temperature	float	Sampling temperature passed to the local model.	required
top_p	float	Nucleus sampling probability passed to the local model.	required
repeat_penalty	float	Repeat penalty passed to the local model.	required
max_tokens	int	Maximum number of generated tokens.	required
stream	bool	Whether to stream generated text into the UI.	required
output	Any | None	Optional Streamlit placeholder used for streaming output.	None

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def run_llm_turn( user_input: str, temperature: float, top_p: float, repeat_penalty: float,
		max_tokens: int, stream: bool, output: Any | None = None ) -> str:
	"""Run one local LLM generation turn.

	Purpose:
		Builds the shared prompt, loads the optional local llama.cpp model, and either streams
		generated tokens into a Streamlit placeholder or returns the completed response text
		for downstream workflows.

	Args:
		user_input: User message or constructed prompt text for the current generation turn.
		temperature: Sampling temperature passed to the local model.
		top_p: Nucleus sampling probability passed to the local model.
		repeat_penalty: Repeat penalty passed to the local model.
		max_tokens: Maximum number of generated tokens.
		stream: Whether to stream generated text into the UI.
		output: Optional Streamlit placeholder used for streaming output.

	Returns:
		str: Return value produced by the operation.
	"""
	if user_input is None:
		return ''

	prompt = build_prompt( user_input )
	local_model = get_llm( int( st.session_state.get( 'context_window', cfg.DEFAULT_CTX ) ),
		int( st.session_state.get( 'cpu_threads', cfg.CORES ) ) )

	if local_model is None:
		st.error( 'Local llama.cpp model is unavailable in this environment.' )
		return ''

	if not stream:
		resp = local_model(
			prompt,
			stream=False,
			max_tokens=max_tokens,
			temperature=temperature,
			top_p=top_p,
			repeat_penalty=repeat_penalty,
			stop=[ '</s>' ]
		)
		text = (resp.get( 'choices', [ { 'text': '' } ] )[ 0 ].get( 'text', '' ) or '')
		return text.strip( )

	buf = ''
	if output is None:
		output = st.empty( )

	for chunk in local_model(
			prompt,
			stream=True,
			max_tokens=max_tokens,
			temperature=temperature,
			top_p=top_p,
			repeat_penalty=repeat_penalty,
			stop=[ '</s>' ]
	):
		buf += chunk[ 'choices' ][ 0 ][ 'text' ]
		output.markdown( buf + '▌' )

	output.markdown( buf )
	return buf.strip( )

create_connection

create_connection() -> sqlite3.Connection

Create a SQLite connection.

Purpose

Opens a connection to the configured application database used by chat history, prompt records, semantic embeddings, and data-management tables.

Returns:

Type	Description
Connection	sqlite3.Connection: Return value produced by the operation.

Source code in app.py

def create_connection( ) -> sqlite3.Connection:
	"""Create a SQLite connection.

	Purpose:
		Opens a connection to the configured application database used by chat history, prompt
		records, semantic embeddings, and data-management tables.

	Returns:
		sqlite3.Connection: Return value produced by the operation.
	"""
	return sqlite3.connect( cfg.DB_PATH )

list_tables

list_tables() -> List[str]

List user-visible SQLite tables.

Purpose

Reads table names from sqlite_master and returns them in sorted order for data- management browsing, administration, and query workflows.

Returns:

Type	Description
List[str]	list[str]: Return value produced by the operation.

Source code in app.py

def list_tables( ) -> List[ str ]:
	"""List user-visible SQLite tables.

	Purpose:
		Reads table names from sqlite_master and returns them in sorted order for data-
		management browsing, administration, and query workflows.

	Returns:
		list[str]: Return value produced by the operation.
	"""
	with create_connection( ) as conn:
		_query = "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name;"
		rows = conn.execute( _query ).fetchall( )
		return [ r[ 0 ] for r in rows ]

create_schema

create_schema(table: str) -> List[Tuple]

Read a SQLite table schema.

Purpose

Retrieves PRAGMA table_info metadata for the selected table so data-management views can display columns, types, nullability, defaults, and primary-key indicators.

Parameters:

Name	Type	Description	Default
table	str	SQLite table name.	required

Returns:

Type	Description
List[Tuple]	list[tuple]: Return value produced by the operation.

Source code in app.py

def create_schema( table: str ) -> List[ Tuple ]:
	"""Read a SQLite table schema.

	Purpose:
		Retrieves PRAGMA table_info metadata for the selected table so data-management views
		can display columns, types, nullability, defaults, and primary-key indicators.

	Args:
		table: SQLite table name.

	Returns:
		list[tuple]: Return value produced by the operation.
	"""
	with create_connection( ) as conn:
		return conn.execute( f'PRAGMA table_info("{table}");' ).fetchall( )

read_table

read_table(
    table: str, limit: int = None, offset: int = 0
) -> pd.DataFrame

Read rows from a SQLite table.

Purpose

Builds a SELECT query for the requested table and optional pagination arguments, then returns the result as a pandas DataFrame for browsing and analysis.

Parameters:

Name	Type	Description	Default
table	str	SQLite table name.	required
limit	int	Optional maximum number of rows to return.	None
offset	int	Number of rows to skip before returning results.	0

Returns:

Type	Description
DataFrame	pd.DataFrame: Return value produced by the operation.

Source code in app.py

def read_table( table: str, limit: int = None, offset: int = 0 ) -> pd.DataFrame:
	"""Read rows from a SQLite table.

	Purpose:
		Builds a SELECT query for the requested table and optional pagination arguments, then
		returns the result as a pandas DataFrame for browsing and analysis.

	Args:
		table: SQLite table name.
		limit: Optional maximum number of rows to return.
		offset: Number of rows to skip before returning results.

	Returns:
		pd.DataFrame: Return value produced by the operation.
	"""
	query = f'SELECT rowid, * FROM "{table}"'
	if limit:
		query += f" LIMIT {limit} OFFSET {offset}"
	with create_connection( ) as conn:
		return pd.read_sql_query( query, conn )

drop_table

drop_table(table: str) -> None

Drop a SQLite table when requested.

Purpose

Removes a selected table from the application database when the data-management administrator explicitly requests deletion.

Parameters:

Name	Type	Description	Default
table	str	SQLite table name.	required

Raises:

Type	Description
ValueError	Raised when the requested table name is empty.

Source code in app.py

def drop_table( table: str ) -> None:
	"""Drop a SQLite table when requested.

	Purpose:
		Removes a selected table from the application database when the data-management
		administrator explicitly requests deletion.

	Args:
		table: SQLite table name.

	Raises:
		ValueError: Raised when the requested table name is empty.
	"""
	if not table:
		return

	with create_connection( ) as conn:
		conn.execute( f'DROP TABLE IF EXISTS "{table}";' )
		conn.commit( )

rename_table

rename_table(old_name: str, new_name: str) -> None

Rename an existing SQLite table.

Purpose

Attempts a native SQLite ALTER TABLE rename and falls back to a schema-preserving rebuild when needed. The fallback preserves column data and index SQL where possible.

Parameters:

Name	Type	Description	Default
old_name	str	Existing table or column name.	required
new_name	str	Replacement table or column name.	required

Raises:

Type	Description
ValueError	Raised when the source table definition is missing or malformed.

Source code in app.py

def rename_table( old_name: str, new_name: str ) -> None:
	"""Rename an existing SQLite table.

	Purpose:
		Attempts a native SQLite ALTER TABLE rename and falls back to a schema-preserving
		rebuild when needed. The fallback preserves column data and index SQL where possible.

	Args:
		old_name: Existing table or column name.
		new_name: Replacement table or column name.

	Raises:
		ValueError: Raised when the source table definition is missing or malformed.
	"""
	if not old_name or not new_name:
		return

	with create_connection( ) as conn:
		try:
			conn.execute( f'ALTER TABLE "{old_name}" RENAME TO "{new_name}";' )
			conn.commit( )
			return
		except Exception:
			pass

		row = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='table' AND name =?
			""",
			(old_name,)
		).fetchone( )

		if not row or not row[ 0 ]:
			raise ValueError( "Table definition not found." )

		create_sql = row[ 0 ]

		indexes = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='index' AND tbl_name=? AND sql IS NOT NULL
			""",
			(old_name,)
		).fetchall( )

		open_paren = create_sql.find( "(" )
		if open_paren == -1:
			raise ValueError( "Malformed CREATE TABLE statement." )

		temp_name = f"{new_name}__rebuild_temp"

		conn.execute( "BEGIN" )
		conn.execute( f'CREATE TABLE "{temp_name}" {create_sql[ open_paren: ]}' )

		cols = [ r[ 1 ] for r in conn.execute( f'PRAGMA table_info("{old_name}");' ).fetchall( ) ]
		col_list = ", ".join( [ f'"{c}"' for c in cols ] )

		conn.execute(
			f'INSERT INTO "{temp_name}" ({col_list}) SELECT {col_list} FROM "{old_name}";'
		)

		conn.execute( f'DROP TABLE "{old_name}";' )
		conn.execute( f'ALTER TABLE "{temp_name}" RENAME TO "{new_name}";' )

		for idx in indexes:
			idx_sql = idx[ 0 ]
			if idx_sql:
				idx_sql = idx_sql.replace( f'ON "{old_name}"', f'ON "{new_name}"' )
				conn.execute( idx_sql )

		conn.commit( )

rename_column

rename_column(
    table_name: str, old_name: str, new_name: str
) -> None

Rename a column within a SQLite table.

Purpose

Attempts a native SQLite column rename and falls back to a schema-preserving table rebuild that keeps column order, data, defaults, nullability, and index SQL where possible.

Parameters:

Name	Type	Description	Default
table_name	str	SQLite table name.	required
old_name	str	Existing table or column name.	required
new_name	str	Replacement table or column name.	required

Raises:

Type	Description
ValueError	Raised when the table definition is missing or the selected column does not exist.

Source code in app.py

def rename_column( table_name: str, old_name: str, new_name: str ) -> None:
	"""Rename a column within a SQLite table.

	Purpose:
		Attempts a native SQLite column rename and falls back to a schema-preserving table
		rebuild that keeps column order, data, defaults, nullability, and index SQL where
		possible.

	Args:
		table_name: SQLite table name.
		old_name: Existing table or column name.
		new_name: Replacement table or column name.

	Raises:
		ValueError: Raised when the table definition is missing or the selected column does not exist.
	"""
	if not table_name or not old_name or not new_name:
		return

	with create_connection( ) as conn:
		try:
			conn.execute(
				f'ALTER TABLE "{table_name}" RENAME COLUMN "{old_name}" TO "{new_name}";'
			)
			conn.commit( )
			return
		except Exception:
			pass

		row = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='table' AND name =?
			""",
			(table_name,)
		).fetchone( )

		if not row or not row[ 0 ]:
			raise ValueError( "Table definition not found." )

		create_sql = row[ 0 ]

		indexes = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='index' AND tbl_name=? AND sql IS NOT NULL
			""",
			(table_name,)
		).fetchall( )

		schema = conn.execute( f'PRAGMA table_info("{table_name}");' ).fetchall( )
		cols = [ r[ 1 ] for r in schema ]
		if old_name not in cols:
			raise ValueError( "Column not found." )

		mapped_cols = [ (new_name if c == old_name else c) for c in cols ]

		temp_table = f"{table_name}__rebuild_temp"

		col_defs: List[ str ] = [ ]
		pk_cols = [ r for r in schema if int( r[ 5 ] or 0 ) > 0 ]
		single_pk = len( pk_cols ) == 1

		for row in schema:
			col_name = row[ 1 ]
			col_type = row[ 2 ] or ''
			not_null = int( row[ 3 ] or 0 )
			default_value = row[ 4 ]
			pk = int( row[ 5 ] or 0 )

			out_name = new_name if col_name == old_name else col_name
			col_def = f'"{out_name}" {col_type}'.strip( )

			if not_null:
				col_def += ' NOT NULL'

			if default_value is not None:
				col_def += f' DEFAULT {default_value}'

			if single_pk and pk == 1:
				col_def += ' PRIMARY KEY'

			col_defs.append( col_def )

		new_create_sql = f'CREATE TABLE "{temp_table}" ({", ".join( col_defs )});'

		old_select = ", ".join( [ f'"{c}"' for c in cols ] )
		new_insert = ", ".join( [ f'"{c}"' for c in mapped_cols ] )

		conn.execute( "BEGIN" )
		conn.execute( new_create_sql )
		conn.execute(
			f'INSERT INTO "{temp_table}" ({new_insert}) SELECT {old_select} FROM "{table_name}";'
		)

		conn.execute( f'DROP TABLE "{table_name}";' )
		conn.execute( f'ALTER TABLE "{temp_table}" RENAME TO "{table_name}";' )

		for idx in indexes:
			idx_sql = idx[ 0 ]
			if idx_sql:
				idx_sql = idx_sql.replace( f'"{old_name}"', f'"{new_name}"' )
				conn.execute( idx_sql )

		conn.commit( )

create_index

create_index(table: str, column: str) -> None

Create a safe SQLite index.

Purpose

Validates the requested table and column against the active database schema, creates a sanitized index name, and creates the index using quoted identifiers.

Parameters:

Name	Type	Description	Default
table	str	SQLite table name.	required
column	str	SQLite column name.	required

Raises:

Type	Description
ValueError	Raised when the table or column is not present in the active schema.

Source code in app.py

def create_index( table: str, column: str ) -> None:
	"""Create a safe SQLite index.

	Purpose:
		Validates the requested table and column against the active database schema, creates a
		sanitized index name, and creates the index using quoted identifiers.

	Args:
		table: SQLite table name.
		column: SQLite column name.

	Raises:
		ValueError: Raised when the table or column is not present in the active schema.
	"""
	if not table or not column:
		return

	# ----------  Validate table exists
	tables = list_tables( )
	if table not in tables:
		raise ValueError( "Invalid table name." )

	# ----------  Validate column exists
	schema = create_schema( table )
	valid_columns = [ col[ 1 ] for col in schema ]

	if column not in valid_columns:
		raise ValueError( "Invalid column name." )

	# ----------  Sanitize index name (identifier only)
	safe_index_name = re.sub( r"[^0-9a-zA-Z_]+", "_", f"idx_{table}_{column}" )

	# ----------  Create index safely (quote identifiers)
	sql = f'CREATE INDEX IF NOT EXISTS "{safe_index_name}" ON "{table}"("{column}");'

	with create_connection( ) as conn:
		conn.execute( sql )
		conn.commit( )

apply_filters

apply_filters(df: DataFrame) -> pd.DataFrame

Apply an interactive DataFrame filter.

Purpose

Renders Streamlit filter controls and applies the selected comparison or containment operation to the provided DataFrame for data-management exploration.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame used by the operation.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: Return value produced by the operation.

Source code in app.py

def apply_filters( df: pd.DataFrame ) -> pd.DataFrame:
	"""Apply an interactive DataFrame filter.

	Purpose:
		Renders Streamlit filter controls and applies the selected comparison or containment
		operation to the provided DataFrame for data-management exploration.

	Args:
		df: DataFrame used by the operation.

	Returns:
		pd.DataFrame: Return value produced by the operation.
	"""
	st.subheader( 'Advanced Filters' )
	col1, col2, col3 = st.columns( 3 )
	column = col1.selectbox( 'Column', df.columns )
	operator = col2.selectbox( 'Operator', [ '=', '!=', '>', '<', '>=', '<=', 'contains' ] )
	value = col3.text_input( 'Value' )
	if value:
		if operator == '=':
			df = df[ df[ column ] == value ]
		elif operator == '!=':
			df = df[ df[ column ] != value ]
		elif operator == '>':
			df = df[ df[ column ].astype( float ) > float( value ) ]
		elif operator == '<':
			df = df[ df[ column ].astype( float ) < float( value ) ]
		elif operator == '>=':
			df = df[ df[ column ].astype( float ) >= float( value ) ]
		elif operator == '<=':
			df = df[ df[ column ].astype( float ) <= float( value ) ]
		elif operator == 'contains':
			df = df[ df[ column ].astype( str ).str.contains( value ) ]

	return df

create_aggregation

create_aggregation(df: DataFrame) -> None

Render an interactive aggregation result.

Purpose

Renders Streamlit controls for selecting a numeric column and aggregation function, computes the selected aggregate, and displays the result as a metric.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame used by the operation.	required

Source code in app.py

def create_aggregation( df: pd.DataFrame ) -> None:
	"""Render an interactive aggregation result.

	Purpose:
		Renders Streamlit controls for selecting a numeric column and aggregation function,
		computes the selected aggregate, and displays the result as a metric.

	Args:
		df: DataFrame used by the operation.
	"""
	st.subheader( 'Aggregation Engine' )

	numeric_cols = df.select_dtypes( include=[ 'number' ] ).columns.tolist( )

	if not numeric_cols:
		st.info( 'No numeric columns available.' )
		return

	col = st.selectbox( 'Column', numeric_cols )
	agg = st.selectbox( 'Aggregation', [ 'COUNT', 'SUM', 'AVG', 'MIN', 'MAX', 'MEDIAN' ] )

	if agg == 'COUNT':
		result = df[ col ].count( )
	elif agg == 'SUM':
		result = df[ col ].sum( )
	elif agg == 'AVG':
		result = df[ col ].mean( )
	elif agg == 'MIN':
		result = df[ col ].min( )
	elif agg == 'MAX':
		result = df[ col ].max( )
	elif agg == 'MEDIAN':
		result = df[ col ].median( )

	st.metric( 'Result', result )

create_visualization

create_visualization(df: DataFrame) -> None

Render an interactive DataFrame visualization.

Purpose

Renders Streamlit chart controls and uses Plotly Express to display histograms, bars, lines, scatter plots, boxes, pies, or correlation heatmaps from the provided DataFrame.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame used by the operation.	required

Source code in app.py

def create_visualization( df: pd.DataFrame ) -> None:
	"""Render an interactive DataFrame visualization.

	Purpose:
		Renders Streamlit chart controls and uses Plotly Express to display histograms, bars,
		lines, scatter plots, boxes, pies, or correlation heatmaps from the provided
		DataFrame.

	Args:
		df: DataFrame used by the operation.
	"""
	st.subheader( 'Visualization Engine' )
	numeric_cols = df.select_dtypes( include=[ 'number' ] ).columns.tolist( )
	categorical_cols = df.select_dtypes( include=[ 'object' ] ).columns.tolist( )
	chart = st.selectbox( 'Chart Type',
		[ 'Histogram', 'Bar', 'Line', 'Scatter', 'Box', 'Pie', 'Correlation' ] )

	if chart == 'Histogram' and numeric_cols:
		col = st.selectbox( 'Column', numeric_cols )
		fig = px.histogram( df, x=col )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Bar':
		x = st.selectbox( 'X', df.columns )
		y = st.selectbox( 'Y', numeric_cols )
		fig = px.bar( df, x=x, y=y )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Line':
		x = st.selectbox( 'X', df.columns )
		y = st.selectbox( 'Y', numeric_cols )
		fig = px.line( df, x=x, y=y )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Scatter':
		x = st.selectbox( 'X', numeric_cols )
		y = st.selectbox( 'Y', numeric_cols )
		fig = px.scatter( df, x=x, y=y )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Box':
		col = st.selectbox( 'Column', numeric_cols )
		fig = px.box( df, y=col )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Pie':
		col = st.selectbox( 'Category Column', categorical_cols )
		fig = px.pie( df, names=col )
		st.plotly_chart( fig, use_container_width=True )

	elif chart == 'Correlation' and len( numeric_cols ) > 1:
		corr = df[ numeric_cols ].corr( )
		fig = px.imshow( corr, text_auto=True )
		st.plotly_chart( fig, use_container_width=True )

convert_dataframe

convert_dataframe(table_name: str, df: DataFrame) -> None

Create a SQLite table from DataFrame columns.

Purpose

Maps pandas column dtypes to SQLite storage types and creates a table definition using normalized column names for imported tabular data.

Parameters:

Name	Type	Description	Default
table_name	str	SQLite table name.	required
df	DataFrame	DataFrame used by the operation.	required

Source code in app.py

def convert_dataframe( table_name: str, df: pd.DataFrame ) -> None:
	"""Create a SQLite table from DataFrame columns.

	Purpose:
		Maps pandas column dtypes to SQLite storage types and creates a table definition using
		normalized column names for imported tabular data.

	Args:
		table_name: SQLite table name.
		df: DataFrame used by the operation.
	"""
	columns = [ ]
	for col in df.columns:
		sql_type = get_sqlite_type( df[ col ].dtype )
		safe_col = col.replace( ' ', '_' )
		columns.append( f'{safe_col} {sql_type}' )

	create_stmt = f'CREATE TABLE IF NOT EXISTS {table_name} ({", ".join( columns )});'

	with create_connection( ) as conn:
		conn.execute( create_stmt )
		conn.commit( )

insert_data

insert_data(table_name: str, df: DataFrame) -> None

Insert DataFrame rows into SQLite.

Purpose

Normalizes DataFrame column names to SQLite-friendly identifiers and bulk-inserts row values into the specified table.

Parameters:

Name	Type	Description	Default
table_name	str	SQLite table name.	required
df	DataFrame	DataFrame used by the operation.	required

Source code in app.py

def insert_data( table_name: str, df: pd.DataFrame ) -> None:
	"""Insert DataFrame rows into SQLite.

	Purpose:
		Normalizes DataFrame column names to SQLite-friendly identifiers and bulk-inserts row
		values into the specified table.

	Args:
		table_name: SQLite table name.
		df: DataFrame used by the operation.
	"""
	df = df.copy( )
	df.columns = [ c.replace( ' ', '_' ) for c in df.columns ]

	placeholders = ', '.join( [ '?' ] * len( df.columns ) )
	stmt = f'INSERT INTO {table_name} VALUES ({placeholders});'

	with create_connection( ) as conn:
		conn.executemany( stmt, df.values.tolist( ) )
		conn.commit( )

get_sqlite_type

get_sqlite_type(dtype: object) -> str

Map a pandas dtype to a SQLite type.

Purpose

Converts pandas dtype information into the SQLite storage class used by import and table-creation workflows.

Parameters:

Name	Type	Description	Default
dtype	object	Pandas dtype or dtype-like object.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def get_sqlite_type( dtype: object ) -> str:
	"""Map a pandas dtype to a SQLite type.

	Purpose:
		Converts pandas dtype information into the SQLite storage class used by import and
		table-creation workflows.

	Args:
		dtype: Pandas dtype or dtype-like object.

	Returns:
		str: Return value produced by the operation.
	"""
	dtype_str = str( dtype ).lower( )

	# ----------  Integer Types
	if "int" in dtype_str:
		return "INTEGER"

	# ----------  Float Types
	if "float" in dtype_str:
		return "REAL"

	# ----------  Boolean
	if "bool" in dtype_str:
		return "INTEGER"

	# ----------  Datetime
	if "datetime" in dtype_str:
		return "TEXT"

	# ----------  Categorical
	if "category" in dtype_str:
		return "TEXT"

	# ----------  Default fallback
	return "TEXT"

create_custom_table

create_custom_table(table_name: str, columns: list) -> None

Create a custom SQLite table.

Purpose

Validates table and column identifiers, translates user-provided column definitions into SQLite DDL, and creates the requested table when it does not exist.

Parameters:

Name	Type	Description	Default
table_name	str	SQLite table name.	required
columns	list	Column-definition dictionaries used to create the table.	required

Raises:

Type	Description
ValueError	Raised when table or column definitions are invalid.

Source code in app.py

def create_custom_table( table_name: str, columns: list ) -> None:
	"""Create a custom SQLite table.

	Purpose:
		Validates table and column identifiers, translates user-provided column definitions
		into SQLite DDL, and creates the requested table when it does not exist.

	Args:
		table_name: SQLite table name.
		columns: Column-definition dictionaries used to create the table.

	Raises:
		ValueError: Raised when table or column definitions are invalid.
	"""
	if not table_name:
		raise ValueError( "Table name required." )

	# ----------  Validate identifier
	if not re.match( r"^[A-Za-z_][A-Za-z0-9_]*$", table_name ):
		raise ValueError( "Invalid table name." )

	col_defs = [ ]
	for col in columns:
		col_name = col[ "name" ]
		col_type = col[ "type" ].upper( )
		if not re.match( r"^[A-Za-z_][A-Za-z0-9_]*$", col_name ):
			raise ValueError( f"Invalid column name: {col_name}" )

		definition = f'"{col_name}" {col_type}'
		if col[ "primary_key" ]:
			definition += " PRIMARY KEY"
			if col[ "auto_increment" ] and col_type == "INTEGER":
				definition += " AUTOINCREMENT"

		if col[ "not_null" ]:
			definition += " NOT NULL"

		col_defs.append( definition )

	sql = f'CREATE TABLE IF NOT EXISTS "{table_name}" ({", ".join( col_defs )});'
	with create_connection( ) as conn:
		conn.execute( sql )
		conn.commit( )

is_safe_query

is_safe_query(query: str) -> bool

Validate whether a SQL query is read-only.

Purpose

Checks query text before execution in the SQL console, allowing read-oriented statements and blocking destructive or mutating operations and multiple statements.

Parameters:

Name	Type	Description	Default
query	str	SQL or search query text.	required

Returns:

Name	Type	Description
bool	bool	Return value produced by the operation.

Source code in app.py

def is_safe_query( query: str ) -> bool:
	"""Validate whether a SQL query is read-only.

	Purpose:
		Checks query text before execution in the SQL console, allowing read-oriented
		statements and blocking destructive or mutating operations and multiple statements.

	Args:
		query: SQL or search query text.

	Returns:
		bool: Return value produced by the operation.
	"""
	if not query or not isinstance( query, str ):
		return False

	q = query.strip( ).lower( )

	# ----------  Block multiple statements
	if ';' in q[ :-1 ]:
		return False

	# ----------  Remove SQL comments
	q = re.sub( r"--.*?$", "", q, flags=re.MULTILINE )
	q = re.sub( r"/\*.*?\*/", "", q, flags=re.DOTALL )
	q = q.strip( )

	# ----------  Allowed starting keywords
	allowed_starts = ('select', 'with', 'explain', 'pragma')
	if not q.startswith( allowed_starts ):
		return False

	# ----------  Block dangerous keywords anywhere
	blocked_keywords = ('insert ', 'update ', 'delete ', 'drop ', 'alter ',
	                    'create ', 'attach ', 'detach ', 'vacuum ', 'replace ', 'trigger ')

	for keyword in blocked_keywords:
		if keyword in q:
			return False

	return True

create_identifier

create_identifier(name: str) -> str

Create a safe SQLite identifier.

Purpose

Normalizes arbitrary text into a SQLite-safe identifier by replacing invalid characters, ensuring a valid leading character, and rejecting empty results.

Parameters:

Name	Type	Description	Default
name	str	Prompt caption or object name used by the operation.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Raises:

Type	Description
ValueError	Raised when the input cannot be converted into a valid identifier.

Source code in app.py

def create_identifier( name: str ) -> str:
	"""Create a safe SQLite identifier.

	Purpose:
		Normalizes arbitrary text into a SQLite-safe identifier by replacing invalid
		characters, ensuring a valid leading character, and rejecting empty results.

	Args:
		name: Prompt caption or object name used by the operation.

	Returns:
		str: Return value produced by the operation.

	Raises:
		ValueError: Raised when the input cannot be converted into a valid identifier.
	"""
	if not name or not isinstance( name, str ):
		raise ValueError( 'Invalid Identifier.' )

	safe = re.sub( r'[^0-9a-zA-Z_]', '_', name.strip( ) )
	if not re.match( r'^[A-Za-z_]', safe ):
		safe = f'_{safe}'

	if not safe:
		raise ValueError( 'Invalid identifier after sanitization.' )

	return safe

get_indexes

get_indexes(table: str) -> List[Tuple]

List indexes for a SQLite table.

Purpose

Reads PRAGMA index_list metadata for the selected table so the data-management administration UI can display available indexes.

Parameters:

Name	Type	Description	Default
table	str	SQLite table name.	required

Returns:

Type	Description
List[Tuple]	list[tuple]: Return value produced by the operation.

Source code in app.py

def get_indexes( table: str ) -> List[ Tuple ]:
	"""List indexes for a SQLite table.

	Purpose:
		Reads PRAGMA index_list metadata for the selected table so the data-management
		administration UI can display available indexes.

	Args:
		table: SQLite table name.

	Returns:
		list[tuple]: Return value produced by the operation.
	"""
	with create_connection( ) as conn:
		rows = conn.execute( f'PRAGMA index_list("{table}");' ).fetchall( )
		return rows

add_column

add_column(table: str, column: str, col_type: str) -> None

Add a column to a SQLite table.

Purpose

Sanitizes the requested column name and executes an ALTER TABLE statement that appends the column with the requested SQLite type.

Parameters:

Name	Type	Description	Default
table	str	SQLite table name.	required
column	str	SQLite column name.	required
col_type	str	SQLite column type.	required

Source code in app.py

def add_column( table: str, column: str, col_type: str ) -> None:
	"""Add a column to a SQLite table.

	Purpose:
		Sanitizes the requested column name and executes an ALTER TABLE statement that appends
		the column with the requested SQLite type.

	Args:
		table: SQLite table name.
		column: SQLite column name.
		col_type: SQLite column type.
	"""
	column = create_identifier( column )
	col_type = col_type.upper( )

	with create_connection( ) as conn:
		conn.execute(
			f'ALTER TABLE "{table}" ADD COLUMN "{column}" {col_type};' )
		conn.commit( )

create_profile_table

create_profile_table(table: str) -> pd.DataFrame

Create a profile summary for a SQLite table.

Purpose

Reads the selected table into a DataFrame and computes per-column type, null percentage, distinct percentage, and numeric summary statistics for exploration.

Parameters:

Name	Type	Description	Default
table	str	SQLite table name.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: Return value produced by the operation.

Source code in app.py

def create_profile_table( table: str ) -> pd.DataFrame:
	"""Create a profile summary for a SQLite table.

	Purpose:
		Reads the selected table into a DataFrame and computes per-column type, null
		percentage, distinct percentage, and numeric summary statistics for exploration.

	Args:
		table: SQLite table name.

	Returns:
		pd.DataFrame: Return value produced by the operation.
	"""
	df = read_table( table )
	profile_rows = [ ]
	total_rows = len( df )
	for col in df.columns:
		series = df[ col ]
		null_count = series.isna( ).sum( )
		distinct_count = series.nunique( dropna=True )
		row = \
			{
					'column': col, 'dtype': str( series.dtype ),
					'null_%': round( (null_count / total_rows) * 100, 2 ) if total_rows else 0,
					'distinct_%': round( (
							                     distinct_count / total_rows) * 100,
						2 ) if total_rows else 0,
			}

		if pd.api.types.is_numeric_dtype( series ):
			row[ "min" ] = series.min( )
			row[ "max" ] = series.max( )
			row[ "mean" ] = series.mean( )
		else:
			row[ "min" ] = None
			row[ "max" ] = None
			row[ "mean" ] = None

		profile_rows.append( row )

	return pd.DataFrame( profile_rows )

drop_column

drop_column(table: str, column: str) -> None

Drop a column from a SQLite table.

Purpose

Rebuilds the selected table without the specified column while preserving remaining column definitions, data, and indexes that do not depend on the dropped column.

Parameters:

Name	Type	Description	Default
table	str	SQLite table name.	required
column	str	SQLite column name.	required

Raises:

Type	Description
ValueError	Raised when the table, column, or resulting schema would be invalid.

Source code in app.py

def drop_column( table: str, column: str ) -> None:
	"""Drop a column from a SQLite table.

	Purpose:
		Rebuilds the selected table without the specified column while preserving remaining
		column definitions, data, and indexes that do not depend on the dropped column.

	Args:
		table: SQLite table name.
		column: SQLite column name.

	Raises:
		ValueError: Raised when the table, column, or resulting schema would be invalid.
	"""
	if not table or not column:
		raise ValueError( "Table and column required." )

	with create_connection( ) as conn:
		schema = conn.execute( f'PRAGMA table_info("{table}");' ).fetchall( )
		if not schema:
			raise ValueError( "Table definition not found." )

		col_names = [ r[ 1 ] for r in schema ]
		if column not in col_names:
			raise ValueError( "Column not found." )

		remaining = [ r for r in schema if r[ 1 ] != column ]
		if not remaining:
			raise ValueError( "Cannot drop the only remaining column." )

		temp_table = f"{table}_rebuild_temp"

		pk_cols = [ r for r in remaining if int( r[ 5 ] or 0 ) > 0 ]
		single_pk = len( pk_cols ) == 1

		new_defs: List[ str ] = [ ]
		for row in remaining:
			col_name = row[ 1 ]
			col_type = row[ 2 ] or ''
			not_null = int( row[ 3 ] or 0 )
			default_value = row[ 4 ]
			pk = int( row[ 5 ] or 0 )

			col_def = f'"{col_name}" {col_type}'.strip( )

			if not_null:
				col_def += ' NOT NULL'

			if default_value is not None:
				col_def += f' DEFAULT {default_value}'

			if single_pk and pk == 1:
				col_def += ' PRIMARY KEY'

			new_defs.append( col_def )

		new_create_sql = f'CREATE TABLE "{temp_table}" ({", ".join( new_defs )});'

		conn.execute( "BEGIN" )
		conn.execute( new_create_sql )

		remaining_cols = [ r[ 1 ] for r in remaining ]
		col_list = ", ".join( [ f'"{c}"' for c in remaining_cols ] )

		conn.execute(
			f'INSERT INTO "{temp_table}" ({col_list}) '
			f'SELECT {col_list} FROM "{table}";'
		)

		indexes = conn.execute(
			"""
            SELECT sql
            FROM sqlite_master
            WHERE type ='index' AND tbl_name=? AND sql IS NOT NULL
			""",
			(table,)
		).fetchall( )

		conn.execute( f'DROP TABLE "{table}";' )
		conn.execute( f'ALTER TABLE "{temp_table}" RENAME TO "{table}";' )

		for idx in indexes:
			idx_sql = idx[ 0 ]
			if idx_sql and column not in idx_sql:
				conn.execute( idx_sql )

		conn.commit( )

extract_text_from_bytes

extract_text_from_bytes(file_bytes: bytes) -> str

Extract text from uploaded document bytes.

Purpose

Attempts PDF extraction with PyMuPDF and falls back to defensive text decoding when PDF parsing fails. The helper supports document preview, summarization, and retrieval workflows.

Parameters:

Name	Type	Description	Default
file_bytes	bytes	Uploaded file bytes or PDF byte stream.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def extract_text_from_bytes( file_bytes: bytes ) -> str:
	"""Extract text from uploaded document bytes.

	Purpose:
		Attempts PDF extraction with PyMuPDF and falls back to defensive text decoding when
		PDF parsing fails. The helper supports document preview, summarization, and retrieval
		workflows.

	Args:
		file_bytes: Uploaded file bytes or PDF byte stream.

	Returns:
		str: Return value produced by the operation.
	"""
	try:
		import fitz  # PyMuPDF

		doc = fitz.open( stream=file_bytes, filetype="pdf" )
		text = ""
		for page in doc:
			text += page.get_text( )
		return text.strip( )

	except Exception:
		try:
			return file_bytes.decode( errors="ignore" )
		except Exception:
			return ""

route_document_query

route_document_query(prompt: str) -> str

Route a document question through the chat pipeline.

Purpose

Builds a retrieval-grounded Document Q&A input from the user prompt and submits it to the shared local LLM generation workflow using current Streamlit runtime controls.

Parameters:

Name	Type	Description	Default
prompt	str	Document question or prompt text.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def route_document_query( prompt: str ) -> str:
	"""Route a document question through the chat pipeline.

	Purpose:
		Builds a retrieval-grounded Document Q&A input from the user prompt and submits it to
		the shared local LLM generation workflow using current Streamlit runtime controls.

	Args:
		prompt: Document question or prompt text.

	Returns:
		str: Return value produced by the operation.
	"""
	user_input = build_docqna_input( prompt )
	if not user_input:
		user_input = (prompt or '').strip( )

	return run_llm_turn( user_input=user_input,
		temperature=float( st.session_state.get( 'temperature', 0.0 ) ),
		top_p=float( st.session_state.get( 'top_percent', 0.95 ) ),
		repeat_penalty=float( st.session_state.get( 'repeat_penalty', 1.1 ) ),
		max_tokens=int( st.session_state.get( 'max_tokens', 1024 ) ) or 1024,
		stream=False, output=None )

summarize_active_document

summarize_active_document() -> str

Summarize the active document context.

Purpose

Constructs a structured summary prompt and routes the request through the Document Q&A workflow. The shared generation pipeline applies current system instructions once.

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def summarize_active_document( ) -> str:
	"""Summarize the active document context.

	Purpose:
		Constructs a structured summary prompt and routes the request through the Document Q&A
		workflow. The shared generation pipeline applies current system instructions once.

	Returns:
		str: Return value produced by the operation.
	"""
	summary_prompt = """
		Provide a clear, structured summary of this document.
		Include:
		- Purpose
		- Key themes
		- Major conclusions
		- Important data points (if any)
		- Policy implications (if applicable)

		Be precise and concise.
		"""
	return route_document_query( summary_prompt.strip( ) )

compute_fingerprint

compute_fingerprint(
    active_docs: List[str], doc_bytes: Dict[str, bytes]
) -> str

Compute a stable active-document fingerprint.

Purpose

Hashes active document names, byte lengths, and byte content hashes to support cache invalidation when uploaded Document Q&A inputs change.

Parameters:

Name	Type	Description	Default
active_docs	List[str]	Active document names selected in session state.	required
doc_bytes	Dict[str, bytes]	Mapping of document names to uploaded bytes.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def compute_fingerprint( active_docs: List[ str ], doc_bytes: Dict[ str, bytes ] ) -> str:
	"""Compute a stable active-document fingerprint.

	Purpose:
		Hashes active document names, byte lengths, and byte content hashes to support cache
		invalidation when uploaded Document Q&A inputs change.

	Args:
		active_docs: Active document names selected in session state.
		doc_bytes: Mapping of document names to uploaded bytes.

	Returns:
		str: Return value produced by the operation.
	"""
	h = hashlib.sha256( )
	for name in sorted( active_docs ):
		b = doc_bytes.get( name, b'' )
		h.update( name.encode( 'utf-8', errors='ignore' ) )
		h.update( len( b ).to_bytes( 8, 'little', signed=False ) )
		h.update( hashlib.sha256( b ).digest( ) )
	return h.hexdigest( )

extract_text

extract_text(file_bytes: bytes) -> str

Extract text from PDF bytes.

Purpose

Uses PyMuPDF to read text from each page of a PDF byte stream and returns the combined text for chunking and retrieval workflows.

Parameters:

Name	Type	Description	Default
file_bytes	bytes	Uploaded file bytes or PDF byte stream.	required

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def extract_text( file_bytes: bytes ) -> str:
	"""Extract text from PDF bytes.

	Purpose:
		Uses PyMuPDF to read text from each page of a PDF byte stream and returns the combined
		text for chunking and retrieval workflows.

	Args:
		file_bytes: Uploaded file bytes or PDF byte stream.

	Returns:
		str: Return value produced by the operation.
	"""
	if not file_bytes:
		return ''

	try:
		doc = fitz.open( stream=file_bytes, filetype='pdf' )
		parts: List[ str ] = [ ]
		for page in doc:
			parts.append( page.get_text( 'text' ) or '' )
		return '\n'.join( parts ).strip( )
	except Exception as e:
		write_error( e, 'document_qna', 'extract_text( file_bytes: bytes ) -> str' )
		return ''

load_sqlite_vec

load_sqlite_vec(conn: Connection) -> bool

Load the sqlite-vec extension into a connection.

Purpose

Attempts to register sqlite-vec support on the provided SQLite connection so Document Q&A can use vector-table retrieval when available.

Parameters:

Name	Type	Description	Default
conn	Connection	SQLite connection.	required

Returns:

Name	Type	Description
bool	bool	Return value produced by the operation.

Source code in app.py

def load_sqlite_vec( conn: sqlite3.Connection ) -> bool:
	"""Load the sqlite-vec extension into a connection.

	Purpose:
		Attempts to register sqlite-vec support on the provided SQLite connection so Document
		Q&A can use vector-table retrieval when available.

	Args:
		conn: SQLite connection.

	Returns:
		bool: Return value produced by the operation.
	"""
	try:
		import sqlite_vec

		sqlite_vec.load( conn )
		return True
	except Exception as e:
		write_error( e, 'document_qna', 'load_sqlite_vec( conn: sqlite3.Connection ) -> bool' )
		return False

ensure_vec_schema

ensure_vec_schema(dim: int) -> bool

Ensure the Document Q&A vector schema exists.

Purpose

Loads sqlite-vec and creates the docqna_vec virtual table for document embeddings when the extension is available in the runtime environment.

Parameters:

Name	Type	Description	Default
dim	int	Embedding dimension used by the vector table.	required

Returns:

Name	Type	Description
bool	bool	Return value produced by the operation.

Source code in app.py

def ensure_vec_schema( dim: int ) -> bool:
	"""Ensure the Document Q&A vector schema exists.

	Purpose:
		Loads sqlite-vec and creates the docqna_vec virtual table for document embeddings when
		the extension is available in the runtime environment.

	Args:
		dim: Embedding dimension used by the vector table.

	Returns:
		bool: Return value produced by the operation.
	"""
	conn = create_connection( )
	try:
		ok = load_sqlite_vec( conn )
		if not ok:
			return False

		cur = conn.cursor( )
		cur.execute(
			f'''
			CREATE VIRTUAL TABLE IF NOT EXISTS docqna_vec
			USING vec0(
				embedding float[{int( dim )}],
				doc_name TEXT,
				chunk TEXT
			);
			'''
		)
		conn.commit( )
		return True
	except Exception as e:
		write_error( e, 'document_qna', 'ensure_vec_schema( dim: int ) -> bool' )
		return False
	finally:
		conn.close( )

rebuild_index

rebuild_index(embedder: SentenceTransformer) -> None

Build or refresh the Document Q&A vector index.

Purpose

Compares the active document fingerprint with cached state, extracts active document text, chunks content, creates embeddings, and stores vectors either in sqlite-vec or in the in-memory fallback rows.

Parameters:

Name	Type	Description	Default
embedder	SentenceTransformer	SentenceTransformer used to create embeddings.	required

Source code in app.py

def rebuild_index( embedder: SentenceTransformer ) -> None:
	"""Build or refresh the Document Q&A vector index.

	Purpose:
		Compares the active document fingerprint with cached state, extracts active document
		text, chunks content, creates embeddings, and stores vectors either in sqlite-vec or
		in the in-memory fallback rows.

	Args:
		embedder: SentenceTransformer used to create embeddings.
	"""
	active_docs: List[ str ] = st.session_state.get( 'active_docs', [ ] )
	doc_bytes: Dict[ str, bytes ] = st.session_state.get( 'doc_bytes', { } )

	fp = compute_fingerprint( active_docs, doc_bytes )
	if fp and fp == st.session_state.get( 'docqna_fingerprint', '' ):
		return

	st.session_state[ 'docqna_fingerprint' ] = fp
	st.session_state[ 'docqna_chunk_count' ] = 0
	st.session_state[ 'docqna_fallback_rows' ] = [ ]

	dim_value = getattr( embedder, 'get_sentence_embedding_dimension', lambda: 384 )( )
	dim = int( dim_value ) if dim_value else 384

	vec_ready = ensure_vec_schema( dim )
	st.session_state[ 'docqna_vec_ready' ] = bool( vec_ready )

	conn = create_connection( )
	try:
		cur = conn.cursor( )

		if vec_ready:
			try:
				cur.execute( 'DELETE FROM docqna_vec;' )
				conn.commit( )
			except Exception:
				st.session_state[ 'docqna_vec_ready' ] = False
				vec_ready = False

		total_chunks = 0
		fallback_rows: List[ Tuple[ str, str, bytes ] ] = [ ]

		for name in active_docs:
			b = doc_bytes.get( name )
			if not b:
				continue

			text = extract_text( b )
			if not text:
				continue

			chunks = chunk_text( text )
			if not chunks:
				continue

			vecs = embedder.encode( chunks, show_progress_bar=False )
			vecs = np.asarray( vecs, dtype=np.float32 )

			if vec_ready:
				for chunk_text_value, v in zip( chunks, vecs ):
					cur.execute(
						'INSERT INTO docqna_vec ( embedding, doc_name, chunk ) VALUES ( ?, ?, ? );',
						(v.tobytes( ), name, chunk_text_value)
					)
			else:
				for chunk_text_value, v in zip( chunks, vecs ):
					fallback_rows.append( (name, chunk_text_value, v.tobytes( )) )

			total_chunks += int( len( chunks ) )

		conn.commit( )
		st.session_state[ 'docqna_chunk_count' ] = total_chunks

		if not vec_ready:
			st.session_state[ 'docqna_fallback_rows' ] = fallback_rows

	except Exception:
		st.session_state[ 'docqna_vec_ready' ] = False
		st.session_state[ 'docqna_fallback_rows' ] = [ ]
		st.session_state[ 'docqna_chunk_count' ] = 0
	finally:
		conn.close( )

retrieve_chunks

retrieve_chunks(
    query: str, k: int = 6
) -> List[Tuple[str, str, float]]

Retrieve document chunks relevant to a query.

Purpose

Embeds the user query, refreshes the document index when needed, retrieves nearest chunks from sqlite-vec when available, and falls back to cosine similarity over cached vectors.

Parameters:

Name	Type	Description	Default
query	str	SQL or search query text.	required
k	int	Number of retrieved chunks to include.	6

Returns:

Type	Description
List[Tuple[str, str, float]]	list[tuple[str, str, float]]: Return value produced by the operation.

Source code in app.py

def retrieve_chunks( query: str, k: int = 6 ) -> List[ Tuple[ str, str, float ] ]:
	"""Retrieve document chunks relevant to a query.

	Purpose:
		Embeds the user query, refreshes the document index when needed, retrieves nearest
		chunks from sqlite-vec when available, and falls back to cosine similarity over cached
		vectors.

	Args:
		query: SQL or search query text.
		k: Number of retrieved chunks to include.

	Returns:
		list[tuple[str, str, float]]: Return value produced by the operation.
	"""
	if not query or not query.strip( ):
		return [ ]

	embedder: SentenceTransformer = load_embedder( )
	rebuild_index( embedder )

	qv = embedder.encode( [ query ], show_progress_bar=False )
	qv = np.asarray( qv, dtype=np.float32 )[ 0 ]

	if st.session_state.get( 'docqna_vec_ready', False ):
		conn = create_connection( )
		try:
			load_sqlite_vec( conn )
			cur = conn.cursor( )
			cur.execute(
				'''
                SELECT doc_name, chunk, distance
                FROM docqna_vec
                WHERE embedding MATCH ?
                ORDER BY distance ASC LIMIT ?;
				''',
				(qv.tobytes( ), int( k ))
			)
			rows = cur.fetchall( )
			return [ (r[ 0 ], r[ 1 ], float( r[ 2 ] )) for r in rows ]
		except Exception:
			st.session_state[ 'docqna_vec_ready' ] = False
		finally:
			conn.close( )

	fallback_rows: List[
		Tuple[ str, str, bytes ] ] = st.session_state.get( 'docqna_fallback_rows', [ ] )
	results: List[ Tuple[ str, str, float ] ] = [ ]

	for doc_name, chunk_text_value, vec_blob in fallback_rows:
		if not vec_blob:
			continue

		v = np.frombuffer( vec_blob, dtype=np.float32 )
		if v.size == 0:
			continue

		score = cosine_sim( qv, v )
		results.append( (doc_name, chunk_text_value, float( score )) )

	results.sort( key=lambda r: r[ 2 ], reverse=True )
	return results[ : int( k ) ]

build_docqna_input

build_docqna_input(user_query: str, k: int = 6) -> str

Build a retrieval-grounded Document Q&A prompt.

Purpose

Retrieves relevant document excerpts and combines them with the user question to form the grounded input submitted to the shared LLM pipeline. System instructions remain in the dedicated system-message path.

Parameters:

Name	Type	Description	Default
user_query	str	User question for the Document Q&A workflow.	required
k	int	Number of retrieved chunks to include.	6

Returns:

Name	Type	Description
str	str	Return value produced by the operation.

Source code in app.py

def build_docqna_input( user_query: str, k: int = 6 ) -> str:
	"""Build a retrieval-grounded Document Q&A prompt.

	Purpose:
		Retrieves relevant document excerpts and combines them with the user question to form
		the grounded input submitted to the shared LLM pipeline. System instructions remain in
		the dedicated system-message path.

	Args:
		user_query: User question for the Document Q&A workflow.
		k: Number of retrieved chunks to include.

	Returns:
		str: Return value produced by the operation.
	"""
	hits = retrieve_chunks( user_query, k=int( k ) )

	context_blocks: List[ str ] = [ ]
	for doc_name, chunk, score in hits:
		context_blocks.append( f'[Document: {doc_name}]\n{chunk}'.strip( ) )

	context = '\n\n'.join( context_blocks ).strip( )

	prompt_parts: List[ str ] = [ ]

	if context:
		prompt_parts.append(
			'Use the following document excerpts to answer the question. If the excerpts do not contain '
			'the answer, say you do not have enough information.\n\n'
			f'{context}'
		)

	prompt_parts.append( f'Question:\n{user_query}\n\nAnswer:' )

	return '\n\n'.join( prompt_parts ).strip( )

load_embedder

load_embedder() -> SentenceTransformer

Load the sentence-transformer embedder.

Purpose

Creates the cached sentence-transformers model used for semantic indexing, Document Q&A retrieval, and fallback vector scoring.

Returns:

Name	Type	Description
SentenceTransformer	SentenceTransformer	Return value produced by the operation.

Source code in app.py

@st.cache_resource
def load_embedder( ) -> SentenceTransformer:
	"""Load the sentence-transformer embedder.

	Purpose:
		Creates the cached sentence-transformers model used for semantic indexing, Document
		Q&A retrieval, and fallback vector scoring.

	Returns:
		SentenceTransformer: Return value produced by the operation.
	"""
	return SentenceTransformer( 'all-MiniLM-L6-v2' )

local_llm_enabled

local_llm_enabled() -> bool

Determine whether local LLM support is enabled.

Purpose

Checks configuration and model-file availability to decide whether the local llama.cpp model should be loaded in the current runtime environment.

Returns:

Name	Type	Description
bool	bool	Return value produced by the operation.

Source code in app.py

def local_llm_enabled( ) -> bool:
	"""Determine whether local LLM support is enabled.

	Purpose:
		Checks configuration and model-file availability to decide whether the local llama.cpp
		model should be loaded in the current runtime environment.

	Returns:
		bool: Return value produced by the operation.
	"""
	try:
		_is_enabled = bool( getattr( cfg, 'ENABLE_LOCAL_LLM', False ) )
		return bool( _is_enabled and local_llm_available( ) )
	except Exception as e:
		write_error( e, 'local_llm', 'local_llm_enabled( ) -> bool' )
		return False

load_llm

load_llm(ctx: int, threads: int) -> Any | None

Load the optional local llama.cpp model.

Purpose

Lazily imports llama_cpp and creates a cached Llama instance using the configured model path, context window, CPU thread count, and batch settings when local support is enabled.

Parameters:

Name	Type	Description	Default
ctx	int	Optional context window override.	required
threads	int	Optional CPU thread-count override.	required

Returns:

Type	Description
Any | None	Any | None: Return value produced by the operation.

Source code in app.py

@st.cache_resource
def load_llm( ctx: int, threads: int ) -> Any | None:
	"""Load the optional local llama.cpp model.

	Purpose:
		Lazily imports llama_cpp and creates a cached Llama instance using the configured
		model path, context window, CPU thread count, and batch settings when local support is
		enabled.

	Args:
		ctx: Optional context window override.
		threads: Optional CPU thread-count override.

	Returns:
		Any | None: Return value produced by the operation.
	"""
	try:
		if not local_llm_enabled( ):
			return None

		try:
			from llama_cpp import Llama
		except Exception:
			return None

		return Llama(
			model_path=str( cfg.MODEL_PATH ),
			n_ctx=ctx,
			n_threads=threads,
			n_batch=512,
			verbose=False
		)
	except Exception as e:
		write_error( e, 'local_llm', 'load_llm( ctx: int, threads: int ) -> Any | None' )
		return None

get_llm

get_llm(
    ctx: int | None = None, threads: int | None = None
) -> Any | None

Return the available local llama.cpp model.

Purpose

Applies default context and thread settings when overrides are not supplied and returns the cached local model instance when enabled and loadable.

Parameters:

Name	Type	Description	Default
ctx	int | None	Optional context window override.	None
threads	int | None	Optional CPU thread-count override.	None

Returns:

Type	Description
Any | None	Any | None: Return value produced by the operation.

Source code in app.py

def get_llm( ctx: int | None = None, threads: int | None = None ) -> Any | None:
	"""Return the available local llama.cpp model.

	Purpose:
		Applies default context and thread settings when overrides are not supplied and
		returns the cached local model instance when enabled and loadable.

	Args:
		ctx: Optional context window override.
		threads: Optional CPU thread-count override.

	Returns:
		Any | None: Return value produced by the operation.
	"""
	try:
		if not local_llm_enabled( ):
			return None

		_ctx = int( ctx ) if ctx else int( getattr( cfg, 'DEFAULT_CTX', 0 ) )
		_threads = int( threads ) if threads else int( getattr( cfg, 'CORES', 1 ) )
		return load_llm( _ctx, _threads )
	except Exception as e:
		write_error( e, 'local_llm',
			'get_llm( ctx: int | None, threads: int | None ) -> Any | None' )
		return None

get_embedder

get_embedder() -> SentenceTransformer | None

Return the cached embedding model.

Purpose

Returns the cached sentence-transformer model when it can be loaded and preserves a None fallback when embedding support is unavailable.

Returns:

Type	Description
SentenceTransformer | None	SentenceTransformer | None: Return value produced by the operation.

Source code in app.py

def get_embedder( ) -> SentenceTransformer | None:
	"""Return the cached embedding model.

	Purpose:
		Returns the cached sentence-transformer model when it can be loaded and preserves a
		None fallback when embedding support is unavailable.

	Returns:
		SentenceTransformer | None: Return value produced by the operation.
	"""
	try:
		return load_embedder( )
	except Exception as e:
		write_error( e, 'embeddings', 'get_embedder( ) -> SentenceTransformer | None' )
		return None

get_conn

get_conn() -> sqlite3.Connection

Create a Prompt Engineering database connection.

Returns:

Type	Description
Connection	sqlite3.Connection: Open connection to the configured application database.

Source code in app.py

def get_conn( ) -> sqlite3.Connection:
	"""Create a Prompt Engineering database connection.

	Returns:
		sqlite3.Connection: Open connection to the configured application database.
	"""
	return sqlite3.connect( cfg.DB_PATH )

reset_selection

reset_selection() -> None

Reset the Prompt Engineering record selection and editor fields.

Returns:

Name	Type	Description
None	None	This function performs its work through side effects and does not return a value.

Source code in app.py

def reset_selection( ) -> None:
	"""Reset the Prompt Engineering record selection and editor fields.

	Returns:
		None: This function performs its work through side effects and does not return a value.
	"""
	st.session_state.pe_selected_id = None
	st.session_state.pe_caption = ''
	st.session_state.pe_name = ''
	st.session_state.pe_category = ''
	st.session_state.pe_text = ''

load_prompt

load_prompt(pid: int) -> None

Load one prompt into the Prompt Engineering editor.

Parameters:

Name	Type	Description	Default
pid	int	Prompt primary key value.	required

Returns:

Name	Type	Description
None	None	This function performs its work through side effects and does not return a value.

Source code in app.py

def load_prompt( pid: int ) -> None:
	"""Load one prompt into the Prompt Engineering editor.

	Args:
		pid: Prompt primary key value.

	Returns:
		None: This function performs its work through side effects and does not return a value.
	"""
	with get_conn( ) as conn:
		select_query = f"SELECT Caption, Name, Category, Text FROM {TABLE} WHERE ID=?"
		cur = conn.execute( select_query, (pid,), )
		row = cur.fetchone( )
		if not row:
			return
		st.session_state.pe_caption = str( row[ 0 ] or '' )
		st.session_state.pe_name = str( row[ 1 ] or '' )
		st.session_state.pe_category = str( row[ 2 ] or '' )
		st.session_state.pe_text = str( row[ 3 ] or '' )
		if st.session_state.pe_cascade_enabled:
			st.session_state.system_instructions = str( row[ 3 ] or '' )
			st.session_state.active_prompt_id = int( pid )
			st.session_state.active_prompt_caption = str( row[ 0 ] or '' )
			st.session_state.active_prompt_name = str( row[ 1 ] or '' )
			st.session_state.active_prompt_category = str( row[ 2 ] or '' )

save_prompt_record

save_prompt_record() -> None

Create or update the prompt currently shown in the editor.

Returns:

Name	Type	Description
None	None	This function performs its work through side effects and does not return a value.

Source code in app.py

def save_prompt_record( ) -> None:
	"""Create or update the prompt currently shown in the editor.

	Returns:
		None: This function performs its work through side effects and does not return a value.
	"""
	data = {
		'Caption': st.session_state.pe_caption,
		'Name': st.session_state.pe_name,
		'Category': st.session_state.pe_category,
		'Text': st.session_state.pe_text,
	}
	if st.session_state.pe_selected_id is None:
		insert_prompt( data )
	else:
		update_prompt( int( st.session_state.pe_selected_id ), data )
	st.session_state.pe_notice = 'Saved.'
	reset_selection( )

delete_prompt_record

delete_prompt_record() -> None

Delete the selected prompt record.

Returns:

Name	Type	Description
None	None	This function performs its work through side effects and does not return a value.

Source code in app.py

def delete_prompt_record( ) -> None:
	"""Delete the selected prompt record.

	Returns:
		None: This function performs its work through side effects and does not return a value.
	"""
	if st.session_state.pe_selected_id is not None:
		delete_prompt( int( st.session_state.pe_selected_id ) )
		st.session_state.pe_notice = 'Deleted.'
		reset_selection( )