app

---

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

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

---

       Loca is python application for running local LLMs.
       Copyright ©  2023 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

---

is_docx_available

is_docx_available() -> bool

Purpose:

Determine whether python-docx is available for DOCX extraction.

Parameters:

None

Returns:

bool True when python-docx is available; otherwise False.

Source code in app.py

def is_docx_available( ) -> bool:
	"""
		Purpose:
		--------
		Determine whether python-docx is available for DOCX extraction.

		Parameters:
		-----------
		None

		Returns:
		--------
		bool
			True when python-docx is available; otherwise False.
	"""
	return Document is not None

is_llama_cpp_available

is_llama_cpp_available() -> bool

Purpose:

Determine whether llama-cpp-python is available for local GGUF inference.

Parameters:

None

Returns:

bool True when llama-cpp-python is available; otherwise False.

Source code in app.py

def is_llama_cpp_available( ) -> bool:
	"""
		Purpose:
		--------
		Determine whether llama-cpp-python is available for local GGUF inference.

		Parameters:
		-----------
		None

		Returns:
		--------
		bool
			True when llama-cpp-python is available; otherwise False.
	"""
	return Llama is not None

is_pymupdf_available

is_pymupdf_available() -> bool

Purpose:

Determine whether PyMuPDF is available for native PDF text extraction.

Parameters:

None

Returns:

bool True when PyMuPDF is available; otherwise False.

Source code in app.py

def is_pymupdf_available( ) -> bool:
	"""
		Purpose:
		--------
		Determine whether PyMuPDF is available for native PDF text extraction.

		Parameters:
		-----------
		None

		Returns:
		--------
		bool
			True when PyMuPDF is available; otherwise False.
	"""
	return fitz is not None

get_selected_model_name

get_selected_model_name() -> str

Purpose:

Return the currently selected local model name from Streamlit session state.

Parameters:

None

Returns:

str Selected model name.

Source code in app.py

def get_selected_model_name( ) -> str:
	"""
		Purpose:
		--------
		Return the currently selected local model name from Streamlit session state.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Selected model name.
	"""
	model_name = str(
		st.session_state.get( 'selected_model_name', get_default_model_name( ) ) or
		get_default_model_name( ) )

	return model_name

get_selected_model_path

get_selected_model_path() -> str

Purpose:

Return the currently selected local GGUF model path from Streamlit session state.

Parameters:

None

Returns:

str Resolved local GGUF path for the selected model.

Source code in app.py

def get_selected_model_path( ) -> str:
	"""
		Purpose:
		--------
		Return the currently selected local GGUF model path from Streamlit session state.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Resolved local GGUF path for the selected model.
	"""
	model_name = get_selected_model_name( )
	model_path = str(
		st.session_state.get( 'selected_model_path', get_model_path_for_state( model_name ) ) or
		get_model_path_for_state( model_name ) )

	return model_path

get_selected_model_spec

get_selected_model_spec() -> Dict[str, Any]

Purpose:

Return the selected model specification from Streamlit session state.

Parameters:

None

Returns:

Dict[str, Any] Selected model metadata.

Source code in app.py

def get_selected_model_spec( ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Return the selected model specification from Streamlit session state.

		Parameters:
		-----------
		None

		Returns:
		--------
		Dict[str, Any]
			Selected model metadata.
	"""
	model_name = get_selected_model_name( )
	model_spec = st.session_state.get( 'selected_model_spec', None )

	if isinstance( model_spec, dict ) and len( model_spec ) > 0:
		return model_spec

	return get_model_spec_for_state( model_name )

local_model_available

local_model_available(
    model_path: str | None = None,
) -> bool

Purpose:

Determine whether the selected or supplied local GGUF model file exists.

Parameters:

model_path : str | None Optional GGUF model path. When omitted, the selected model path is used.

Returns:

bool True when the configured model file exists; otherwise False.

Source code in app.py

def local_model_available( model_path: str | None = None ) -> bool:
	"""
		Purpose:
		--------
		Determine whether the selected or supplied local GGUF model file exists.

		Parameters:
		-----------
		model_path : str | None
			Optional GGUF model path. When omitted, the selected model path is used.

		Returns:
		--------
		bool
			True when the configured model file exists; otherwise False.
	"""
	try:
		path_value = str( model_path or get_selected_model_path( ) or '' ).strip( )

		if not path_value:
			return False

		return Path( path_value ).exists( )
	except Exception:
		return False

get_default_model_name

get_default_model_name() -> str

Purpose:

Return the configured default local model name from config.

Parameters:

None

Returns:

str Default model name.

Source code in app.py

def get_default_model_name( ) -> str:
	"""
		Purpose:
		--------
		Return the configured default local model name from config.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Default model name.
	"""
	default_model = str( getattr( cfg, 'DEFAULT_MODEL', '' ) or '' ).strip( )

	if default_model:
		return default_model

	if hasattr( cfg, 'get_model_names' ):
		model_names = cfg.get_model_names( )
	else:
		model_names = list( getattr( cfg, 'MODEL_MAP', { } ).keys( ) )

	return str( model_names[ 0 ] ) if model_names else ''

get_model_names_for_state

get_model_names_for_state() -> List[str]

Purpose:

Return configured model names from the config registry while preserving fallback compatibility with cfg.MODEL_MAP.

Parameters:

None

Returns:

List[str] Configured model names.

Source code in app.py

def get_model_names_for_state( ) -> List[ str ]:
	"""
		Purpose:
		--------
		Return configured model names from the config registry while preserving fallback
		compatibility with cfg.MODEL_MAP.

		Parameters:
		-----------
		None

		Returns:
		--------
		List[str]
			Configured model names.
	"""
	if hasattr( cfg, 'get_model_names' ):
		model_names = cfg.get_model_names( )
	else:
		model_names = list( getattr( cfg, 'MODEL_MAP', { } ).keys( ) )

	return [ str( name ) for name in model_names ]

get_default_mode_name

get_default_mode_name(model_name: str = '') -> str

Purpose:

Return the default UI mode for the selected model.

Parameters:

model_name : str Selected local model name.

Returns:

str Default UI mode name.

Source code in app.py

def get_default_mode_name( model_name: str = '' ) -> str:
	"""
		Purpose:
		--------
		Return the default UI mode for the selected model.

		Parameters:
		-----------
		model_name : str
			Selected local model name.

		Returns:
		--------
		str
			Default UI mode name.
	"""
	model_value = str( model_name or get_default_model_name( ) ).strip( )

	if hasattr( cfg, 'get_model_modes' ):
		modes = cfg.get_model_modes( model_value )
	else:
		modes = getattr( cfg, 'MODES', [ ] )

	if isinstance( modes, list ) and len( modes ) > 0:
		return str( modes[ 0 ] )

	return str( getattr( cfg, 'DEFAULT_MODE', 'Text Generation' ) or 'Text Generation' )

get_model_modes_for_state

get_model_modes_for_state(model_name: str) -> List[str]

Purpose:

Return the supported modes for the selected model using the config model registry when available, while preserving fallback compatibility with cfg.MODES.

Parameters:

model_name : str Selected local model name.

Returns:

List[str] Supported mode names.

Source code in app.py

def get_model_modes_for_state( model_name: str ) -> List[ str ]:
	"""
		Purpose:
		--------
		Return the supported modes for the selected model using the config model registry
		when available, while preserving fallback compatibility with cfg.MODES.

		Parameters:
		-----------
		model_name : str
			Selected local model name.

		Returns:
		--------
		List[str]
			Supported mode names.
	"""
	model_value = str( model_name or get_default_model_name( ) ).strip( )

	if hasattr( cfg, 'get_model_modes' ):
		modes = cfg.get_model_modes( model_value )
	else:
		modes = getattr( cfg, 'MODES', [ ] )

	if isinstance( modes, list ) and len( modes ) > 0:
		return [ str( mode_name ) for mode_name in modes ]

	return [ 'Text Generation' ]

get_model_path_for_state

get_model_path_for_state(model_name: str) -> str

Purpose:

Return the selected model path using the config model registry when available, while preserving fallback compatibility with cfg.MODEL_MAP and cfg.MODEL_PATH.

Parameters:

model_name : str Selected local model name.

Returns:

str Resolved GGUF model path.

Source code in app.py

def get_model_path_for_state( model_name: str ) -> str:
	"""
		Purpose:
		--------
		Return the selected model path using the config model registry when available,
		while preserving fallback compatibility with cfg.MODEL_MAP and cfg.MODEL_PATH.

		Parameters:
		-----------
		model_name : str
			Selected local model name.

		Returns:
		--------
		str
			Resolved GGUF model path.
	"""
	model_value = str( model_name or get_default_model_name( ) ).strip( )

	if hasattr( cfg, 'get_model_path' ):
		return str( cfg.get_model_path( model_value ) or '' )

	if hasattr( cfg, 'MODEL_MAP' ) and model_value in cfg.MODEL_MAP:
		return str( cfg.MODEL_MAP.get( model_value, '' ) or '' )

	return str( getattr( cfg, 'MODEL_PATH', '' ) or '' )

get_model_spec_for_state

get_model_spec_for_state(model_name: str) -> Dict[str, Any]

Purpose:

Return the selected model registry specification when available.

Parameters:

model_name : str Selected local model name.

Returns:

Dict[str, Any] Model specification dictionary.

Source code in app.py

def get_model_spec_for_state( model_name: str ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Return the selected model registry specification when available.

		Parameters:
		-----------
		model_name : str
			Selected local model name.

		Returns:
		--------
		Dict[str, Any]
			Model specification dictionary.
	"""
	model_value = str( model_name or get_default_model_name( ) ).strip( )

	if hasattr( cfg, 'get_model_spec' ):
		spec = cfg.get_model_spec( model_value )
		if isinstance( spec, dict ):
			return spec

	return {
			'path': get_model_path_for_state( model_value ),
			'modes': get_model_modes_for_state( model_value ),
			'family': '',
			'size': '',
			'chat_template': 'chatml',
			'description': ''
	}

initialize_model_mode_state

initialize_model_mode_state() -> None

Purpose:

Initialize widget-owned and derived model/mode session-state keys before the sidebar widgets are created.

Parameters:

None

Returns:

None

Source code in app.py

def initialize_model_mode_state( ) -> None:
	"""
		Purpose:
		--------
		Initialize widget-owned and derived model/mode session-state keys before the
		sidebar widgets are created.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	model_names = get_model_names_for_state( )
	default_model = get_default_model_name( )

	if 'selected_model_name' not in st.session_state:
		st.session_state[ 'selected_model_name' ] = default_model

	model_name = str(
		st.session_state.get( 'selected_model_name', default_model ) or default_model )

	if model_names and model_name not in model_names:
		model_name = default_model
		st.session_state[ 'selected_model_name' ] = model_name

	model_modes = get_model_modes_for_state( model_name )

	if 'selected_mode' not in st.session_state:
		st.session_state[ 'selected_mode' ] = ( model_modes[ 0 ]
		                                        if model_modes
		                                        else get_default_mode_name( model_name ) )

	selected_mode = str( st.session_state.get( 'selected_mode', get_default_mode_name( model_name ) ) or
		get_default_mode_name( model_name ) )

	if selected_mode not in model_modes:
		selected_mode = model_modes[ 0 ] if model_modes else get_default_mode_name( model_name )
		st.session_state[ 'selected_mode' ] = selected_mode

	st.session_state[ 'selected_model_path' ] = get_model_path_for_state( model_name )
	st.session_state[ 'selected_model_modes' ] = model_modes
	st.session_state[ 'selected_model_spec' ] = get_model_spec_for_state( model_name )
	st.session_state[ 'active_model_name' ] = model_name
	st.session_state[ 'mode' ] = selected_mode

	if 'model_switch_counter' not in st.session_state:
		st.session_state[ 'model_switch_counter' ] = 0

synchronize_model_derived_state

synchronize_model_derived_state() -> None

Purpose:

Synchronize derived model state without modifying widget-owned keys after their widgets have been instantiated.

Parameters:

None

Returns:

None

Source code in app.py

def synchronize_model_derived_state( ) -> None:
	"""
		Purpose:
		--------
		Synchronize derived model state without modifying widget-owned keys after their
		widgets have been instantiated.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	model_name = str( st.session_state.get( 'selected_model_name', get_default_model_name( ) ) or
		get_default_model_name( ) )

	model_modes = get_model_modes_for_state( model_name )

	st.session_state[ 'selected_model_path' ] = get_model_path_for_state( model_name )
	st.session_state[ 'selected_model_modes' ] = model_modes
	st.session_state[ 'selected_model_spec' ] = get_model_spec_for_state( model_name )
	st.session_state[ 'active_model_name' ] = model_name

	selected_mode = str(
		st.session_state.get( 'selected_mode', get_default_mode_name( model_name ) ) or
		get_default_mode_name( model_name )
	)

	if selected_mode in model_modes:
		st.session_state[ 'mode' ] = selected_mode
	else:
		st.session_state[ 'pending_selected_mode' ] = (
				model_modes[ 0 ] if model_modes else get_default_mode_name( model_name )
		)

on_selected_model_change

on_selected_model_change() -> None

Purpose:

Streamlit callback used by the LLM selector to resynchronize derived model values after the selected model changes without directly modifying selected_mode.

Parameters:

None

Returns:

None

Source code in app.py

def on_selected_model_change( ) -> None:
	"""
		Purpose:
		--------
		Streamlit callback used by the LLM selector to resynchronize derived model values
		after the selected model changes without directly modifying selected_mode.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	previous_model = str( st.session_state.get( 'active_model_name', '' ) or '' )
	model_name = str(
		st.session_state.get( 'selected_model_name', get_default_model_name( ) ) or
		get_default_model_name( )
	)

	model_modes = get_model_modes_for_state( model_name )

	st.session_state[ 'selected_model_path' ] = get_model_path_for_state( model_name )
	st.session_state[ 'selected_model_modes' ] = model_modes
	st.session_state[ 'selected_model_spec' ] = get_model_spec_for_state( model_name )
	st.session_state[ 'active_model_name' ] = model_name

	if previous_model and previous_model != model_name:
		st.session_state[ 'model_switch_counter' ] = ( int( st.session_state.get(
			'model_switch_counter', 0 ) or 0 ) + 1 )

	current_mode = str( st.session_state.get( 'selected_mode', '' ) or '' )
	if current_mode not in model_modes:
		st.session_state[ 'pending_selected_mode' ] = ( model_modes[ 0 ]
		                                                if model_modes

		                                                else get_default_mode_name( model_name ) )
	try:
		refresh_capability_session_state( )
		apply_model_safe_retrieval_defaults( model_name )
	except Exception:
		pass

on_selected_mode_change

on_selected_mode_change() -> None

Purpose:

Streamlit callback used by the AI Mode selector to keep the legacy mode key aligned with selected_mode.

Parameters:

None

Returns:

None

Source code in app.py

def on_selected_mode_change( ) -> None:
	"""
		Purpose:
		--------
		Streamlit callback used by the AI Mode selector to keep the legacy mode key
		aligned with selected_mode.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	selected_mode = str( st.session_state.get( 'selected_mode', '' ) or '' )
	st.session_state[ 'mode' ] = selected_mode

get_mode_constant

get_mode_constant(constant_name: str, fallback: str) -> str

Purpose:

Return a mode constant from config with a stable fallback. This allows app.py to accept expanded config.py mode definitions without crashing while config updates are being staged.

Parameters:

constant_name : str Name of the config.py constant.

str

Fallback mode name.

Returns:

str Resolved mode name.

Source code in app.py

def get_mode_constant( constant_name: str, fallback: str ) -> str:
	"""
		Purpose:
		--------
		Return a mode constant from config with a stable fallback. This allows app.py to
		accept expanded config.py mode definitions without crashing while config updates
		are being staged.

		Parameters:
		-----------
		constant_name : str
			Name of the config.py constant.

		fallback : str
			Fallback mode name.

		Returns:
		--------
		str
			Resolved mode name.
	"""
	try:
		value = cfg.__dict__.get( constant_name, fallback )
		value = str( value or fallback ).strip( )
		return value if value else fallback
	except Exception:
		return fallback

get_mode_definition_text

get_mode_definition_text(mode_name: str) -> str

Purpose:

Return descriptive config.py text for expanded API modes when available.

Parameters:

mode_name : str UI mode name.

Returns:

str Mode description text.

Source code in app.py

def get_mode_definition_text( mode_name: str ) -> str:
	"""
		Purpose:
		--------
		Return descriptive config.py text for expanded API modes when available.

		Parameters:
		-----------
		mode_name : str
			UI mode name.

		Returns:
		--------
		str
			Mode description text.
	"""
	try:
		image_mode = get_mode_constant( 'IMAGE_MODE', 'Images API' )
		audio_mode = get_mode_constant( 'AUDIO_MODE', 'Audio API' )

		if mode_name == image_mode:
			return str( cfg.__dict__.get( 'IMAGES_API', '' ) or '' ).strip( )

		if mode_name == audio_mode:
			return str( cfg.__dict__.get( 'AUDIO_API', '' ) or '' ).strip( )

		return ''
	except Exception:
		return ''

get_selected_base_model

get_selected_base_model() -> str

Purpose:

Return the selected model's configured base model name.

Parameters:

None

Returns:

str Base model name.

Source code in app.py

def get_selected_base_model( ) -> str:
	"""
		Purpose:
		--------
		Return the selected model's configured base model name.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Base model name.
	"""
	try:
		spec = get_selected_model_spec( )
		return str( spec.get( 'base_model', '' ) or '' ).strip( )
	except Exception:
		return ''

get_selected_model_family

get_selected_model_family() -> str

Purpose:

Return the selected model's configured model family.

Parameters:

None

Returns:

str Model family name.

Source code in app.py

def get_selected_model_family( ) -> str:
	"""
		Purpose:
		--------
		Return the selected model's configured model family.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Model family name.
	"""
	try:
		spec = get_selected_model_spec( )
		return str( spec.get( 'family', '' ) or '' ).strip( )
	except Exception:
		return ''

get_selected_chat_template

get_selected_chat_template() -> str

Purpose:

Return the selected model's configured chat template.

Parameters:

None

Returns:

str Chat template name.

Source code in app.py

def get_selected_chat_template( ) -> str:
	"""
		Purpose:
		--------
		Return the selected model's configured chat template.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Chat template name.
	"""
	try:
		spec = get_selected_model_spec( )
		return str( spec.get( 'chat_template', '' ) or '' ).strip( )
	except Exception:
		return ''

is_buddy_model

is_buddy_model() -> bool

Purpose:

Determine whether the selected model is Buddy or a Buddy base model.

Parameters:

None

Returns:

bool True when Buddy is selected; otherwise False.

Source code in app.py

def is_buddy_model( ) -> bool:
	"""
		Purpose:
		--------
		Determine whether the selected model is Buddy or a Buddy base model.

		Parameters:
		-----------
		None

		Returns:
		--------
		bool
			True when Buddy is selected; otherwise False.
	"""
	model_name = get_selected_model_name( ).lower( )
	base_model = get_selected_base_model( ).lower( )
	return model_name == 'buddy' or base_model == 'gemma-3-270m-it'

is_gipity_model

is_gipity_model() -> bool

Purpose:

Determine whether the selected model is Gipity or a GPT-OSS base model.

Parameters:

None

Returns:

bool True when Gipity is selected; otherwise False.

Source code in app.py

def is_gipity_model( ) -> bool:
	"""
		Purpose:
		--------
		Determine whether the selected model is Gipity or a GPT-OSS base model.

		Parameters:
		-----------
		None

		Returns:
		--------
		bool
			True when Gipity is selected; otherwise False.
	"""
	model_name = get_selected_model_name( ).lower( )
	base_model = get_selected_base_model( ).lower( )
	return model_name == 'gipity' or base_model == 'gpt-oss-20b'

is_gemma4_model

is_gemma4_model() -> bool

Purpose:

Determine whether the selected model uses the Gemma 4 E4B base model.

Parameters:

None

Returns:

bool True when a Gemma 4 E4B model is selected; otherwise False.

Source code in app.py

def is_gemma4_model( ) -> bool:
	"""
		Purpose:
		--------
		Determine whether the selected model uses the Gemma 4 E4B base model.

		Parameters:
		-----------
		None

		Returns:
		--------
		bool
			True when a Gemma 4 E4B model is selected; otherwise False.
	"""
	model_name = get_selected_model_name( ).lower( )
	base_model = get_selected_base_model( ).lower( )
	return model_name in ('jimi', 'nisty') or base_model == 'gemma-4-e4b-it'

is_jimi_or_nisty_model

is_jimi_or_nisty_model() -> bool

Purpose:

Determine whether the selected model is Jimi or Nisty.

Parameters:

None

Returns:

bool True when Jimi or Nisty is selected; otherwise False.

Source code in app.py

def is_jimi_or_nisty_model( ) -> bool:
	"""
		Purpose:
		--------
		Determine whether the selected model is Jimi or Nisty.

		Parameters:
		-----------
		None

		Returns:
		--------
		bool
			True when Jimi or Nisty is selected; otherwise False.
	"""
	model_name = get_selected_model_name( ).lower( )
	return model_name in ('jimi', 'nisty')

model_supports_mode

model_supports_mode(mode_name: str) -> bool

Purpose:

Determine whether the selected model registry advertises a specific UI mode.

Parameters:

mode_name : str UI mode name.

Returns:

bool True when the mode is listed for the selected model; otherwise False.

Source code in app.py

def model_supports_mode( mode_name: str ) -> bool:
	"""
		Purpose:
		--------
		Determine whether the selected model registry advertises a specific UI mode.

		Parameters:
		-----------
		mode_name : str
			UI mode name.

		Returns:
		--------
		bool
			True when the mode is listed for the selected model; otherwise False.
	"""
	try:
		model_modes = st.session_state.get(
			'selected_model_modes',
			get_model_modes_for_state( get_selected_model_name( ) )
		)

		if not isinstance( model_modes, list ):
			return False

		return str( mode_name or '' ) in [ str( m ) for m in model_modes ]
	except Exception:
		return False

get_runtime_multimodal_status

get_runtime_multimodal_status() -> Dict[str, Any]

Purpose:

Return the current runtime's multimodal adapter status. This detects whether app.py has an image/audio-capable local adapter configured separately from the model registry. The function fails closed so newly exposed modes cannot crash.

Parameters:

None

Returns:

Dict[str, Any] Runtime multimodal status flags and message.

Source code in app.py

def get_runtime_multimodal_status( ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Return the current runtime's multimodal adapter status. This detects whether app.py
		has an image/audio-capable local adapter configured separately from the model
		registry. The function fails closed so newly exposed modes cannot crash.

		Parameters:
		-----------
		None

		Returns:
		--------
		Dict[str, Any]
			Runtime multimodal status flags and message.
	"""
	status = {
			'image_runtime_available': False,
			'audio_runtime_available': False,
			'function_runtime_available': True,
			'web_runtime_available': False,
			'runtime_name': 'llama-cpp-python',
			'message': 'The current local runtime is treated as text-only until a '
			           'multimodal adapter is explicitly wired into app.py.'
	}

	try:
		if bool( cfg.__dict__.get( 'IMAGE_RUNTIME_AVAILABLE', False ) ):
			status[ 'image_runtime_available' ] = True

		if bool( cfg.__dict__.get( 'AUDIO_RUNTIME_AVAILABLE', False ) ):
			status[ 'audio_runtime_available' ] = True

		if bool( cfg.__dict__.get( 'WEB_RUNTIME_AVAILABLE', False ) ):
			status[ 'web_runtime_available' ] = True

		runtime_name = str( cfg.__dict__.get( 'MULTIMODAL_RUNTIME_NAME', '' ) or '' ).strip( )
		if runtime_name:
			status[ 'runtime_name' ] = runtime_name

		if status[ 'image_runtime_available' ] or status[ 'audio_runtime_available' ]:
			status[ 'message' ] = 'A multimodal runtime adapter is configured.'

		return status
	except Exception:
		return status

get_active_model_capabilities

get_active_model_capabilities() -> Dict[str, Any]

Purpose:

Return selected model capability flags used by expanded Text, Image, Audio, Function Calling, Coding, Thinking, and Web Browsing workflows.

Parameters:

None

Returns:

Dict[str, Any] Capability contract for the selected model.

Source code in app.py

def get_active_model_capabilities( ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Return selected model capability flags used by expanded Text, Image, Audio,
		Function Calling, Coding, Thinking, and Web Browsing workflows.

		Parameters:
		-----------
		None

		Returns:
		--------
		Dict[str, Any]
			Capability contract for the selected model.
	"""
	model_name = get_selected_model_name( )
	base_model = get_selected_base_model( )
	family = get_selected_model_family( )
	template = get_selected_chat_template( )
	image_mode = get_mode_constant( 'IMAGE_MODE', 'Images API' )
	audio_mode = get_mode_constant( 'AUDIO_MODE', 'Audio API' )
	docqna_mode = get_mode_constant( 'DOCQNA_MODE', 'Document Q&A' )
	semantic_mode = get_mode_constant( 'SEMANTIC_MODE', 'Semantic Search' )
	prompt_mode = get_mode_constant( 'PROMPT_MODE', 'Prompt Engineering' )
	data_mode = get_mode_constant( 'DATA_MODE', 'Data Management' )
	text_mode = get_mode_constant( 'TEXT_MODE', 'Text Generation' )
	runtime_status = get_runtime_multimodal_status( )

	capabilities: Dict[ str, Any ] = {
			'model_name': model_name,
			'base_model': base_model,
			'family': family,
			'chat_template': template,
			'text_generation': model_supports_mode( text_mode ),
			'document_qna': model_supports_mode( docqna_mode ),
			'semantic_search': model_supports_mode( semantic_mode ),
			'prompt_engineering': model_supports_mode( prompt_mode ),
			'data_management': model_supports_mode( data_mode ),
			'image_mode': model_supports_mode( image_mode ) and is_jimi_or_nisty_model( ),
			'audio_mode': model_supports_mode( audio_mode ) and is_jimi_or_nisty_model( ),
			'image_runtime_available': bool(
				runtime_status.get( 'image_runtime_available', False ) ),
			'audio_runtime_available': bool(
				runtime_status.get( 'audio_runtime_available', False ) ),
			'function_calling': is_gemma4_model( ) or is_gipity_model( ),
			'coding': is_gemma4_model( ),
			'thinking': is_gemma4_model( ),
			'web_browsing': is_gipity_model( ),
			'gipity': is_gipity_model( ),
			'gemma4': is_gemma4_model( ),
			'buddy': is_buddy_model( ),
			'runtime_status': runtime_status
	}

	return capabilities

model_supports_capability

model_supports_capability(capability: str) -> bool

Purpose:

Determine whether the selected model supports a named expanded capability.

Parameters:

capability : str Capability name.

Returns:

bool True when the selected model supports the capability; otherwise False.

Source code in app.py

def model_supports_capability( capability: str ) -> bool:
	"""
		Purpose:
		--------
		Determine whether the selected model supports a named expanded capability.

		Parameters:
		-----------
		capability : str
			Capability name.

		Returns:
		--------
		bool
			True when the selected model supports the capability; otherwise False.
	"""
	try:
		capabilities = get_active_model_capabilities( )
		return bool( capabilities.get( str( capability or '' ), False ) )
	except Exception:
		return False

get_capability_status_message

get_capability_status_message(capability: str) -> str

Purpose:

Return a user-facing status message for unsupported or unavailable capabilities.

Parameters:

capability : str Capability name.

Returns:

str Status message.

Source code in app.py

def get_capability_status_message( capability: str ) -> str:
	"""
		Purpose:
		--------
		Return a user-facing status message for unsupported or unavailable capabilities.

		Parameters:
		-----------
		capability : str
			Capability name.

		Returns:
		--------
		str
			Status message.
	"""
	capabilities = get_active_model_capabilities( )
	model_name = str( capabilities.get( 'model_name', get_selected_model_name( ) ) or '' )
	runtime_status = capabilities.get( 'runtime_status', { } )

	if capability == 'image_mode':
		if not capabilities.get( 'image_mode', False ):
			return f'{model_name} is not configured for Image Mode.'

		if not capabilities.get( 'image_runtime_available', False ):
			return str( runtime_status.get( 'message', '' ) or
			            'Image Mode is configured, but no image-capable runtime is wired yet.' )

	if capability == 'audio_mode':
		if not capabilities.get( 'audio_mode', False ):
			return f'{model_name} is not configured for Audio Mode.'

		if not capabilities.get( 'audio_runtime_available', False ):
			return str( runtime_status.get( 'message', '' ) or
			            'Audio Mode is configured, but no audio-capable runtime is wired yet.' )

	if not bool( capabilities.get( capability, False ) ):
		return f'{model_name} does not advertise the "{capability}" capability.'

	return f'{model_name} supports the "{capability}" capability.'

get_default_function_schema_text

get_default_function_schema_text() -> str

Purpose:

Return a safe starter JSON schema for function-calling workflows.

Parameters:

None

Returns:

str Starter JSON function schema text.

Source code in app.py

def get_default_function_schema_text( ) -> str:
	"""
		Purpose:
		--------
		Return a safe starter JSON schema for function-calling workflows.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Starter JSON function schema text.
	"""
	return '''{
	"name": "summarize_text",
	"description": "Summarize supplied text into concise bullet points.",
	"parameters": {
		"type": "object",
		"properties": {
			"text": {
				"type": "string",
				"description": "The text to summarize."
			},
			"max_bullets": {
				"type": "integer",
				"description": "Maximum number of bullets to return."
			}
		},
		"required": [
			"text"
		]
	}
}'''

initialize_capability_session_state

initialize_capability_session_state() -> None

Purpose:

Initialize expanded capability session-state keys before Image, Audio, Function Calling, Coding, Thinking, and Web Browsing controls are introduced.

Parameters:

None

Returns:

None

Source code in app.py

def initialize_capability_session_state( ) -> None:
	"""
		Purpose:
		--------
		Initialize expanded capability session-state keys before Image, Audio, Function
		Calling, Coding, Thinking, and Web Browsing controls are introduced.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	defaults: Dict[ str, Any ] = {
			'image_prompt': '',
			'image_uploaded_name': '',
			'image_response': '',
			'image_status': '',
			'image_context_buffer': '',
			'image_send_to_text': False,
			'audio_prompt': '',
			'audio_uploaded_name': '',
			'audio_response': '',
			'audio_status': '',
			'audio_transcript': '',
			'audio_context_buffer': '',
			'audio_send_to_text': False,
			'function_schema_text': get_default_function_schema_text( ),
			'function_call_prompt': '',
			'function_call_response': '',
			'function_call_result': '',
			'function_call_status': '',
			'function_call_enabled': False,
			'function_call_model_json': '',
			'coding_mode_enabled': False,
			'coding_test_request': False,
			'coding_explain_request': False,
			'thinking_mode_enabled': False,
			'thinking_effort': 'Medium',
			'thinking_summary_enabled': True,
			'web_browse_url': '',
			'web_browse_allow_domain': '',
			'web_browse_prompt': '',
			'web_browse_result': '',
			'web_browse_status': '',
			'web_browse_context_buffer': '',
			'web_browse_send_to_text': False,
			'active_model_capabilities': { }
	}

	for key, value in defaults.items( ):
		if key not in st.session_state:
			st.session_state[ key ] = value

	st.session_state[ 'active_model_capabilities' ] = get_active_model_capabilities( )

refresh_capability_session_state

refresh_capability_session_state() -> None

Purpose:

Refresh derived capability state after model or mode changes without clearing user-owned text, uploaded-file names, generated output, or existing chat state.

Parameters:

None

Returns:

None

Source code in app.py

def refresh_capability_session_state( ) -> None:
	"""
		Purpose:
		--------
		Refresh derived capability state after model or mode changes without clearing
		user-owned text, uploaded-file names, generated output, or existing chat state.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	try:
		st.session_state[ 'active_model_capabilities' ] = get_active_model_capabilities( )
	except Exception:
		st.session_state[ 'active_model_capabilities' ] = { }

get_model_retrieval_profile

get_model_retrieval_profile(
    model_name: str,
) -> Dict[str, Any]

Purpose:

Return model-safe retrieval defaults for Document Q&A and Semantic Search. Smaller models receive narrower retrieval windows so grounded prompts stay concise and less likely to exceed practical local runtime limits.

Parameters:

model_name : str Selected local model name.

Returns:

Dict[str, Any] Retrieval profile values.

Source code in app.py

def get_model_retrieval_profile( model_name: str ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Return model-safe retrieval defaults for Document Q&A and Semantic Search. Smaller
		models receive narrower retrieval windows so grounded prompts stay concise and
		less likely to exceed practical local runtime limits.

		Parameters:
		-----------
		model_name : str
			Selected local model name.

		Returns:
		--------
		Dict[str, Any]
			Retrieval profile values.
	"""
	model_value = str( model_name or get_selected_model_name( ) or '' ).strip( ).lower( )
	base_model = ''

	try:
		spec = get_model_spec_for_state( model_name )
		if isinstance( spec, dict ):
			base_model = str( spec.get( 'base_model', '' ) or '' ).strip( ).lower( )
	except Exception:
		base_model = ''

	if model_value == 'buddy' or base_model == 'gemma-3-270m-it':
		return {
				'profile_name': 'Buddy Compact Retrieval',
				'retrieval_k': 3,
				'retrieval_chunk_size': 800,
				'retrieval_chunk_overlap': 120,
				'semantic_top_k': 4,
				'semantic_chunk_size': 800,
				'semantic_chunk_overlap': 120,
				'semantic_min_similarity': 0.05,
				'require_grounding': True,
				'answer_from_excerpts_only': True,
				'show_retrieved_chunks': True,
				'prefer_sqlite_vec': True,
				'allow_similarity_fallback': True,
				'semantic_show_diagnostics': True,
				'semantic_group_by_document': False
		}

	return {
			'profile_name': 'Standard Retrieval',
			'retrieval_k': 6,
			'retrieval_chunk_size': 1200,
			'retrieval_chunk_overlap': 200,
			'semantic_top_k': 8,
			'semantic_chunk_size': 1200,
			'semantic_chunk_overlap': 200,
			'semantic_min_similarity': 0.0,
			'require_grounding': True,
			'answer_from_excerpts_only': True,
			'show_retrieved_chunks': True,
			'prefer_sqlite_vec': True,
			'allow_similarity_fallback': True,
			'semantic_show_diagnostics': True,
			'semantic_group_by_document': False
	}

has_user_tuned_retrieval_controls

has_user_tuned_retrieval_controls() -> bool

Purpose:

Determine whether the current retrieval controls have already been changed by the user or by a previously applied model profile. This prevents model-safe defaults from overwriting user-tuned values on every Streamlit rerun.

Parameters:

None

Returns:

bool True when retrieval controls should be preserved; otherwise False.

Source code in app.py

def has_user_tuned_retrieval_controls( ) -> bool:
	"""
		Purpose:
		--------
		Determine whether the current retrieval controls have already been changed by the
		user or by a previously applied model profile. This prevents model-safe defaults
		from overwriting user-tuned values on every Streamlit rerun.

		Parameters:
		-----------
		None

		Returns:
		--------
		bool
			True when retrieval controls should be preserved; otherwise False.
	"""
	return bool( st.session_state.get( 'retrieval_controls_user_tuned', False ) )

mark_retrieval_controls_user_tuned

mark_retrieval_controls_user_tuned() -> None

Purpose:

Mark retrieval controls as user-tuned. Later controls can call this callback if needed to permanently preserve manual user settings across model changes.

Parameters:

None

Returns:

None

Source code in app.py

def mark_retrieval_controls_user_tuned( ) -> None:
	"""
		Purpose:
		--------
		Mark retrieval controls as user-tuned. Later controls can call this callback if
		needed to permanently preserve manual user settings across model changes.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	st.session_state[ 'retrieval_controls_user_tuned' ] = True

apply_retrieval_profile

apply_retrieval_profile(
    profile: Dict[str, Any], force: bool = False
) -> None

Purpose:

Apply a retrieval profile to Document Q&A and Semantic Search session-state keys. The profile is applied only when forced or when no user-tuned override exists.

Parameters:

profile : Dict[str, Any] Retrieval profile to apply.

bool

When True, apply the profile even if retrieval_controls_user_tuned is True.

Returns:

None

Source code in app.py

def apply_retrieval_profile( profile: Dict[ str, Any ], force: bool = False ) -> None:
	"""
		Purpose:
		--------
		Apply a retrieval profile to Document Q&A and Semantic Search session-state keys.
		The profile is applied only when forced or when no user-tuned override exists.

		Parameters:
		-----------
		profile : Dict[str, Any]
			Retrieval profile to apply.

		force : bool
			When True, apply the profile even if retrieval_controls_user_tuned is True.

		Returns:
		--------
		None
	"""
	if not isinstance( profile, dict ) or len( profile ) == 0:
		return

	if has_user_tuned_retrieval_controls( ) and not force:
		return

	assignments = {
			'retrieval_k': int( profile.get( 'retrieval_k', 6 ) ),
			'retrieval_chunk_size': int( profile.get( 'retrieval_chunk_size', 1200 ) ),
			'retrieval_chunk_overlap': int( profile.get( 'retrieval_chunk_overlap', 200 ) ),
			'semantic_top_k': int( profile.get( 'semantic_top_k', 8 ) ),
			'semantic_chunk_size': int( profile.get( 'semantic_chunk_size', 1200 ) ),
			'semantic_chunk_overlap': int( profile.get( 'semantic_chunk_overlap', 200 ) ),
			'semantic_min_similarity': float( profile.get( 'semantic_min_similarity', 0.0 ) ),
			'require_grounding': bool( profile.get( 'require_grounding', True ) ),
			'answer_from_excerpts_only': bool( profile.get( 'answer_from_excerpts_only', True ) ),
			'show_retrieved_chunks': bool( profile.get( 'show_retrieved_chunks', True ) ),
			'prefer_sqlite_vec': bool( profile.get( 'prefer_sqlite_vec', True ) ),
			'allow_similarity_fallback': bool( profile.get( 'allow_similarity_fallback', True ) ),
			'semantic_show_diagnostics': bool( profile.get( 'semantic_show_diagnostics', True ) ),
			'semantic_group_by_document': bool( profile.get( 'semantic_group_by_document', False ) )
	}

	for key, value in assignments.items( ):
		st.session_state[ key ] = value

	st.session_state[ 'active_retrieval_profile' ] = str(
		profile.get( 'profile_name', 'Standard Retrieval' ) or 'Standard Retrieval' )
	st.session_state[ 'active_retrieval_profile_model' ] = get_selected_model_name( )

apply_model_safe_retrieval_defaults

apply_model_safe_retrieval_defaults(
    model_name: str = "",
) -> None

Purpose:

Apply model-safe retrieval defaults when the selected model changes. Buddy receives compact retrieval settings suitable for a 270M model; other models receive standard settings unless the user has already tuned retrieval controls.

Parameters:

model_name : str Optional selected model name. When omitted, the current selected model is used.

Returns:

None

Source code in app.py

def apply_model_safe_retrieval_defaults( model_name: str = '' ) -> None:
	"""
		Purpose:
		--------
		Apply model-safe retrieval defaults when the selected model changes. Buddy receives
		compact retrieval settings suitable for a 270M model; other models receive standard
		settings unless the user has already tuned retrieval controls.

		Parameters:
		-----------
		model_name : str
			Optional selected model name. When omitted, the current selected model is used.

		Returns:
		--------
		None
	"""
	selected_model = str( model_name or get_selected_model_name( ) or '' ).strip( )
	if not selected_model:
		return

	last_profile_model = str(
		st.session_state.get( 'last_retrieval_profile_model', '' ) or '' ).strip( )

	if last_profile_model == selected_model:
		return

	profile = get_model_retrieval_profile( selected_model )
	force_apply = not has_user_tuned_retrieval_controls( )

	apply_retrieval_profile( profile=profile, force=force_apply )

	st.session_state[ 'last_retrieval_profile_model' ] = selected_model
	st.session_state[ 'retrieval_profile_status' ] = (
			f'Active retrieval profile: {st.session_state.get( "active_retrieval_profile", "" )}')

reset_model_safe_retrieval_defaults

reset_model_safe_retrieval_defaults() -> None

Purpose:

Clear manual retrieval override state and reapply the selected model's recommended Document Q&A and Semantic Search retrieval profile.

Parameters:

None

Returns:

None

Source code in app.py

def reset_model_safe_retrieval_defaults( ) -> None:
	"""
		Purpose:
		--------
		Clear manual retrieval override state and reapply the selected model's recommended
		Document Q&A and Semantic Search retrieval profile.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	st.session_state[ 'retrieval_controls_user_tuned' ] = False
	st.session_state[ 'last_retrieval_profile_model' ] = ''
	apply_model_safe_retrieval_defaults( get_selected_model_name( ) )

initialize_model_safe_retrieval_state

initialize_model_safe_retrieval_state() -> None

Purpose:

Initialize retrieval-profile tracking keys and apply model-safe defaults once after capability session state is initialized.

Parameters:

None

Returns:

None

Source code in app.py

def initialize_model_safe_retrieval_state( ) -> None:
	"""
		Purpose:
		--------
		Initialize retrieval-profile tracking keys and apply model-safe defaults once after
		capability session state is initialized.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	if 'retrieval_controls_user_tuned' not in st.session_state:
		st.session_state[ 'retrieval_controls_user_tuned' ] = False

	if 'last_retrieval_profile_model' not in st.session_state:
		st.session_state[ 'last_retrieval_profile_model' ] = ''

	if 'active_retrieval_profile' not in st.session_state:
		st.session_state[ 'active_retrieval_profile' ] = ''

	if 'active_retrieval_profile_model' not in st.session_state:
		st.session_state[ 'active_retrieval_profile_model' ] = ''

	if 'retrieval_profile_status' not in st.session_state:
		st.session_state[ 'retrieval_profile_status' ] = ''

	apply_model_safe_retrieval_defaults( get_selected_model_name( ) )

extract_json_object_from_text

extract_json_object_from_text(text: str) -> Dict[str, Any]

Purpose:

Extract the first valid JSON object from model-generated text. This supports function-calling outputs where the model may accidentally wrap the object in markdown fences or explanatory prose.

Parameters:

text : str Model-generated text.

Returns:

Dict[str, Any] Parsed JSON object.

Source code in app.py

def extract_json_object_from_text( text: str ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Extract the first valid JSON object from model-generated text. This supports
		function-calling outputs where the model may accidentally wrap the object in
		markdown fences or explanatory prose.

		Parameters:
		-----------
		text : str
			Model-generated text.

		Returns:
		--------
		Dict[str, Any]
			Parsed JSON object.
	"""
	import json

	text_value = str( text or '' ).strip( )
	if not text_value:
		raise ValueError( 'No function-call text was provided.' )

	if text_value.startswith( '```' ):
		text_value = re.sub( r'^```(?:json)?\s*', '', text_value, flags=re.IGNORECASE )
		text_value = re.sub( r'\s*```$', '', text_value )
		text_value = text_value.strip( )

	try:
		parsed = json.loads( text_value )
		if isinstance( parsed, dict ):
			return parsed
	except Exception:
		pass

	start = text_value.find( '{' )
	if start < 0:
		raise ValueError( 'No JSON object start marker was found.' )

	depth = 0
	in_string = False
	escape = False

	for idx in range( start, len( text_value ) ):
		char = text_value[ idx ]

		if escape:
			escape = False
			continue

		if char == '\\':
			escape = True
			continue

		if char == '"':
			in_string = not in_string
			continue

		if in_string:
			continue

		if char == '{':
			depth += 1
		elif char == '}':
			depth -= 1

			if depth == 0:
				candidate = text_value[ start:idx + 1 ]
				parsed = json.loads( candidate )
				if not isinstance( parsed, dict ):
					raise ValueError( 'The parsed function call was not a JSON object.' )

				return parsed

	raise ValueError( 'No complete JSON object was found.' )

normalize_tool_call

normalize_tool_call(
    tool_call: Dict[str, Any],
) -> Dict[str, Any]

Purpose:

Normalize a model-generated tool call into the app contract: {"name": "...", "arguments": {...}}.

Parameters:

tool_call : Dict[str, Any] Parsed tool-call JSON object.

Returns:

Dict[str, Any] Normalized tool-call object.

Source code in app.py

def normalize_tool_call( tool_call: Dict[ str, Any ] ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Normalize a model-generated tool call into the app contract:
		{"name": "...", "arguments": {...}}.

		Parameters:
		-----------
		tool_call : Dict[str, Any]
			Parsed tool-call JSON object.

		Returns:
		--------
		Dict[str, Any]
			Normalized tool-call object.
	"""
	if not isinstance( tool_call, dict ):
		raise ValueError( 'Tool call must be a dictionary.' )

	name = str( tool_call.get( 'name', '' ) or tool_call.get( 'function', '' ) or '' ).strip( )
	args = tool_call.get( 'arguments', { } )

	if not name:
		raise ValueError( 'Tool call is missing a function name.' )

	if isinstance( args, str ):
		try:
			args = extract_json_object_from_text( args )
		except Exception:
			args = { 'text': args }

	if not isinstance( args, dict ):
		raise ValueError( 'Tool-call arguments must be a dictionary.' )

	return {
			'name': name,
			'arguments': args
	}

get_allowed_function_names

get_allowed_function_names() -> List[str]

Purpose:

Return the function names that app.py is allowed to execute. This prevents model output from invoking arbitrary functions.

Parameters:

None

Returns:

List[str] Allowlisted function names.

Source code in app.py

def get_allowed_function_names( ) -> List[ str ]:
	"""
		Purpose:
		--------
		Return the function names that app.py is allowed to execute. This prevents model
		output from invoking arbitrary functions.

		Parameters:
		-----------
		None

		Returns:
		--------
		List[str]
			Allowlisted function names.
	"""
	return [
			'summarize_text',
			'extract_keywords',
			'web_browse_url'
	]

summarize_text_tool

summarize_text_tool(text: str, max_bullets: int = 5) -> str

Purpose:

Summarize supplied text using a deterministic local sentence extraction fallback. This is intentionally non-agentic and does not execute arbitrary model code.

Parameters:

text : str Text to summarize.

int

Maximum number of bullets to return.

Returns:

str Bullet summary.

Source code in app.py

def summarize_text_tool( text: str, max_bullets: int = 5 ) -> str:
	"""
		Purpose:
		--------
		Summarize supplied text using a deterministic local sentence extraction fallback.
		This is intentionally non-agentic and does not execute arbitrary model code.

		Parameters:
		-----------
		text : str
			Text to summarize.

		max_bullets : int
			Maximum number of bullets to return.

		Returns:
		--------
		str
			Bullet summary.
	"""
	text_value = re.sub( r'\s+', ' ', str( text or '' ) ).strip( )
	if not text_value:
		return 'No text was provided.'

	try:
		bullet_count = int( max_bullets )
	except Exception:
		bullet_count = 5

	if bullet_count <= 0:
		bullet_count = 5

	sentences = re.split( r'(?<=[.!?])\s+', text_value )
	sentences = [ s.strip( ) for s in sentences if s and s.strip( ) ]
	selected = sentences[ :bullet_count ]

	if not selected:
		selected = [ text_value[ :800 ] ]

	return '\n'.join( [ f'- {sentence}' for sentence in selected ] )

extract_keywords_tool

extract_keywords_tool(
    text: str, max_keywords: int = 15
) -> str

Purpose:

Extract simple frequency-ranked keywords from supplied text without external dependencies.

Parameters:

text : str Text to analyze.

int

Maximum number of keywords to return.

Returns:

str Comma-separated keyword list.

Source code in app.py

def extract_keywords_tool( text: str, max_keywords: int = 15 ) -> str:
	"""
		Purpose:
		--------
		Extract simple frequency-ranked keywords from supplied text without external
		dependencies.

		Parameters:
		-----------
		text : str
			Text to analyze.

		max_keywords : int
			Maximum number of keywords to return.

		Returns:
		--------
		str
			Comma-separated keyword list.
	"""
	text_value = str( text or '' ).lower( )
	if not text_value:
		return ''

	try:
		keyword_count = int( max_keywords )
	except Exception:
		keyword_count = 15

	if keyword_count <= 0:
		keyword_count = 15

	stop_words = {
			'a', 'an', 'and', 'are', 'as', 'at', 'be', 'by', 'for', 'from',
			'has', 'have', 'in', 'is', 'it', 'its', 'of', 'on', 'or', 'that',
			'the', 'their', 'this', 'to', 'was', 'were', 'with', 'you', 'your'
	}

	words = re.findall( r'[a-zA-Z][a-zA-Z0-9_\-]{2,}', text_value )
	counts: Dict[ str, int ] = { }

	for word in words:
		if word in stop_words:
			continue

		counts[ word ] = counts.get( word, 0 ) + 1

	ranked = sorted( counts.items( ), key=lambda item: item[ 1 ], reverse=True )
	keywords = [ word for word, _ in ranked[ :keyword_count ] ]

	return ', '.join( keywords )

is_private_or_local_hostname

is_private_or_local_hostname(hostname: str) -> bool

Purpose:

Determine whether a hostname resolves to a local, loopback, private, reserved, or link-local address. This blocks server-side requests to private network resources.

Parameters:

hostname : str URL hostname.

Returns:

bool True when the hostname is private or local; otherwise False.

Source code in app.py

def is_private_or_local_hostname( hostname: str ) -> bool:
	"""
		Purpose:
		--------
		Determine whether a hostname resolves to a local, loopback, private, reserved, or
		link-local address. This blocks server-side requests to private network resources.

		Parameters:
		-----------
		hostname : str
			URL hostname.

		Returns:
		--------
		bool
			True when the hostname is private or local; otherwise False.
	"""
	import ipaddress
	import socket

	host_value = str( hostname or '' ).strip( ).lower( )
	if not host_value:
		return True

	if host_value in ('localhost', '0.0.0.0') or host_value.endswith( '.local' ):
		return True

	try:
		ip_value = ipaddress.ip_address( host_value )
		return bool(
			ip_value.is_private
			or ip_value.is_loopback
			or ip_value.is_link_local
			or ip_value.is_reserved
			or ip_value.is_multicast
			or ip_value.is_unspecified
		)
	except Exception:
		pass

	try:
		addresses = socket.getaddrinfo( host_value, None )
	except Exception:
		return True

	for address in addresses:
		try:
			ip_text = address[ 4 ][ 0 ]
			ip_value = ipaddress.ip_address( ip_text )
			if (
					ip_value.is_private
					or ip_value.is_loopback
					or ip_value.is_link_local
					or ip_value.is_reserved
					or ip_value.is_multicast
					or ip_value.is_unspecified
			):
				return True
		except Exception:
			return True

	return False

validate_web_url

validate_web_url(url: str, allowed_domain: str = '') -> str

Purpose:

Validate an outbound web-browsing URL. Only HTTP and HTTPS URLs are allowed, and private/local network targets are blocked.

Parameters:

url : str User-supplied URL.

str

Optional allowed domain suffix.

Returns:

str Validated URL.

Source code in app.py

def validate_web_url( url: str, allowed_domain: str = '' ) -> str:
	"""
		Purpose:
		--------
		Validate an outbound web-browsing URL. Only HTTP and HTTPS URLs are allowed, and
		private/local network targets are blocked.

		Parameters:
		-----------
		url : str
			User-supplied URL.

		allowed_domain : str
			Optional allowed domain suffix.

		Returns:
		--------
		str
			Validated URL.
	"""
	from urllib.parse import urlparse

	url_value = str( url or '' ).strip( )
	if not url_value:
		raise ValueError( 'A URL is required.' )

	parsed = urlparse( url_value )
	if parsed.scheme.lower( ) not in ('http', 'https'):
		raise ValueError( 'Only http and https URLs are allowed.' )

	if not parsed.netloc or not parsed.hostname:
		raise ValueError( 'The URL must include a valid host.' )

	hostname = str( parsed.hostname or '' ).strip( ).lower( )

	if is_private_or_local_hostname( hostname ):
		raise ValueError( 'Private, local, loopback, reserved, and link-local hosts are blocked.' )

	domain_value = str( allowed_domain or '' ).strip( ).lower( )
	if domain_value:
		if domain_value.startswith( 'http://' ) or domain_value.startswith( 'https://' ):
			domain_value = str( urlparse( domain_value ).hostname or '' ).strip( ).lower( )

		domain_value = domain_value.lstrip( '.' )
		if hostname != domain_value and not hostname.endswith( f'.{domain_value}' ):
			raise ValueError( f'The URL host is not within the allowed domain: {domain_value}' )

	return url_value

html_to_readable_text

html_to_readable_text(html_text: str) -> str

Purpose:

Convert HTML to readable text using a dependency-free parser fallback.

Parameters:

html_text : str Raw HTML text.

Returns:

str Readable extracted text.

Source code in app.py

def html_to_readable_text( html_text: str ) -> str:
	"""
		Purpose:
		--------
		Convert HTML to readable text using a dependency-free parser fallback.

		Parameters:
		-----------
		html_text : str
			Raw HTML text.

		Returns:
		--------
		str
			Readable extracted text.
	"""
	import html

	text_value = str( html_text or '' )
	text_value = re.sub( r'(?is)<(script|style|noscript).*?>.*?</\1>', ' ', text_value )
	text_value = re.sub( r'(?is)<br\s*/?>', '\n', text_value )
	text_value = re.sub( r'(?is)</p\s*>', '\n\n', text_value )
	text_value = re.sub( r'(?is)<[^>]+>', ' ', text_value )
	text_value = html.unescape( text_value )
	text_value = re.sub( r'[ \t\r\f\v]+', ' ', text_value )
	text_value = re.sub( r'\n\s+', '\n', text_value )
	text_value = re.sub( r'\n{3,}', '\n\n', text_value )

	return text_value.strip( )

fetch_web_text

fetch_web_text(
    url: str,
    allowed_domain: str = "",
    timeout_seconds: int = 15,
    max_chars: int = 12000,
) -> Dict[str, Any]

Purpose:

Fetch readable text from a public HTTP/HTTPS URL with timeout, size, and private network safeguards.

Parameters:

url : str User-supplied URL.

str

Optional allowed domain suffix.

int

Network timeout in seconds.

int

Maximum readable text characters returned.

Returns:

Dict[str, Any] Web fetch result.

Source code in app.py

def fetch_web_text( url: str, allowed_domain: str = '', timeout_seconds: int = 15,
		max_chars: int = 12000 ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Fetch readable text from a public HTTP/HTTPS URL with timeout, size, and private
		network safeguards.

		Parameters:
		-----------
		url : str
			User-supplied URL.

		allowed_domain : str
			Optional allowed domain suffix.

		timeout_seconds : int
			Network timeout in seconds.

		max_chars : int
			Maximum readable text characters returned.

		Returns:
		--------
		Dict[str, Any]
			Web fetch result.
	"""
	from urllib.request import Request, urlopen

	try:
		timeout_value = int( timeout_seconds )
	except Exception:
		timeout_value = 15

	if timeout_value <= 0:
		timeout_value = 15

	try:
		max_char_value = int( max_chars )
	except Exception:
		max_char_value = 12000

	if max_char_value <= 0:
		max_char_value = 12000

	validated_url = validate_web_url( url=url, allowed_domain=allowed_domain )

	request = Request(
		validated_url,
		headers={
				'User-Agent': 'Loca-Llama/1.0 TextFetcher'
		}
	)

	with urlopen( request, timeout=timeout_value ) as response:
		content_type = str( response.headers.get( 'Content-Type', '' ) or '' )
		raw = response.read( max_char_value * 4 )

	text = raw.decode( 'utf-8', errors='ignore' )
	if 'html' in content_type.lower( ) or '<html' in text.lower( ):
		readable_text = html_to_readable_text( text )
	else:
		readable_text = re.sub( r'\s+', ' ', text ).strip( )

	if len( readable_text ) > max_char_value:
		readable_text = readable_text[ :max_char_value ].strip( )

	return {
			'url': validated_url,
			'content_type': content_type,
			'text': readable_text,
			'length': len( readable_text )
	}

web_browse_url_tool

web_browse_url_tool(
    url: str,
    prompt: str = "",
    allowed_domain: str = "",
    max_chars: int = 12000,
) -> str

Purpose:

Fetch a public web page and return bounded text suitable for model grounding.

Parameters:

url : str User-supplied URL.

str

Optional user task for the fetched content.

str

Optional allowed domain suffix.

int

Maximum readable text characters returned.

Returns:

str Readable web context.

Source code in app.py

def web_browse_url_tool( url: str, prompt: str = '', allowed_domain: str = '',
		max_chars: int = 12000 ) -> str:
	"""
		Purpose:
		--------
		Fetch a public web page and return bounded text suitable for model grounding.

		Parameters:
		-----------
		url : str
			User-supplied URL.

		prompt : str
			Optional user task for the fetched content.

		allowed_domain : str
			Optional allowed domain suffix.

		max_chars : int
			Maximum readable text characters returned.

		Returns:
		--------
		str
			Readable web context.
	"""
	if not model_supports_capability( 'web_browsing' ):
		return get_capability_status_message( 'web_browsing' )

	result = fetch_web_text(
		url=url,
		allowed_domain=allowed_domain,
		timeout_seconds=15,
		max_chars=max_chars
	)

	task_text = str( prompt or '' ).strip( )
	parts = [
			f'Web Source: {result.get( "url", "" )}',
			f'Content Type: {result.get( "content_type", "" )}',
			f'Characters: {result.get( "length", 0 )}'
	]

	if task_text:
		parts.append( f'User Web Task: {task_text}' )

	parts.append( 'Fetched Web Text:' )
	parts.append( str( result.get( 'text', '' ) or '' ) )

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

execute_allowlisted_function

execute_allowlisted_function(
    tool_call: Dict[str, Any],
) -> Dict[str, Any]

Purpose:

Execute a normalized, allowlisted app function. Arbitrary model-generated function names are rejected.

Parameters:

tool_call : Dict[str, Any] Normalized tool-call object.

Returns:

Dict[str, Any] Tool execution result.

Source code in app.py

def execute_allowlisted_function( tool_call: Dict[ str, Any ] ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Execute a normalized, allowlisted app function. Arbitrary model-generated function
		names are rejected.

		Parameters:
		-----------
		tool_call : Dict[str, Any]
			Normalized tool-call object.

		Returns:
		--------
		Dict[str, Any]
			Tool execution result.
	"""
	normalized = normalize_tool_call( tool_call )
	name = normalized[ 'name' ]
	args = normalized[ 'arguments' ]

	if name not in get_allowed_function_names( ):
		raise ValueError( f'Function "{name}" is not allowlisted.' )

	if name == 'summarize_text':
		result = summarize_text_tool(
			text=str( args.get( 'text', '' ) or '' ),
			max_bullets=int( args.get( 'max_bullets', 5 ) or 5 )
		)

	elif name == 'extract_keywords':
		result = extract_keywords_tool(
			text=str( args.get( 'text', '' ) or '' ),
			max_keywords=int( args.get( 'max_keywords', 15 ) or 15 )
		)

	elif name == 'web_browse_url':
		result = web_browse_url_tool(
			url=str( args.get( 'url', '' ) or '' ),
			prompt=str( args.get( 'prompt', '' ) or '' ),
			allowed_domain=str( args.get( 'allowed_domain', '' ) or '' ),
			max_chars=int( args.get( 'max_chars', 12000 ) or 12000 )
		)

	else:
		raise ValueError( f'Function "{name}" is not implemented.' )

	return {
			'name': name,
			'arguments': args,
			'result': result
	}

build_tool_call_generation_prompt

build_tool_call_generation_prompt(user_task: str) -> str

Purpose:

Build a focused prompt that asks the selected model to emit one strict JSON function-call object.

Parameters:

user_task : str User task to translate into a tool call.

Returns:

str Tool-call generation prompt.

Source code in app.py

def build_tool_call_generation_prompt( user_task: str ) -> str:
	"""
		Purpose:
		--------
		Build a focused prompt that asks the selected model to emit one strict JSON
		function-call object.

		Parameters:
		-----------
		user_task : str
			User task to translate into a tool call.

		Returns:
		--------
		str
			Tool-call generation prompt.
	"""
	schema_text = str( st.session_state.get( 'function_schema_text', '' ) or '' ).strip( )
	available_functions = ', '.join( get_allowed_function_names( ) )

	return f'''Create one app-mediated function call for the user task below.

		Rules:
		- Return one strict JSON object only.
		- Do not wrap the object in markdown.
		- Use this shape: {{"name":"function_name","arguments":{{...}}}}
		- Use only one of these allowlisted functions: {available_functions}
		- Do not invent function names.
		- Do not include prose outside the JSON object.

		Current function schema:
		{schema_text}

		User task:
		{user_task}'''

generate_function_call_json

generate_function_call_json(user_task: str) -> str

Purpose:

Ask the selected local model to generate a strict JSON function-call object.

Parameters:

user_task : str User task to convert into a function call.

Returns:

str Generated model text.

Source code in app.py

def generate_function_call_json( user_task: str ) -> str:
	"""
		Purpose:
		--------
		Ask the selected local model to generate a strict JSON function-call object.

		Parameters:
		-----------
		user_task : str
			User task to convert into a function call.

		Returns:
		--------
		str
			Generated model text.
	"""
	if not model_supports_capability( 'function_calling' ):
		return get_capability_status_message( 'function_calling' )

	task_text = str( user_task or '' ).strip( )
	if not task_text:
		return 'No function-call task was provided.'

	prior_function_enabled = bool( st.session_state.get( 'function_call_enabled', False ) )
	st.session_state[ 'function_call_enabled' ] = True

	try:
		response = run_llm_turn(
			user_input=build_tool_call_generation_prompt( task_text ),
			temperature=0.0,
			top_p=1.0,
			repeat_penalty=float( st.session_state.get( 'repeat_penalty', 1.1 ) ),
			max_tokens=512,
			stream=False,
			output=None
		)
	finally:
		st.session_state[ 'function_call_enabled' ] = prior_function_enabled

	return str( response or '' ).strip( )

execute_tool_call_text

execute_tool_call_text(
    tool_call_text: str,
) -> Dict[str, Any]

Purpose:

Parse and execute model-generated tool-call text through the app's allowlisted function layer.

Parameters:

tool_call_text : str Model-generated function-call JSON text.

Returns:

Dict[str, Any] Tool execution result.

Source code in app.py

def execute_tool_call_text( tool_call_text: str ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Parse and execute model-generated tool-call text through the app's allowlisted
		function layer.

		Parameters:
		-----------
		tool_call_text : str
			Model-generated function-call JSON text.

		Returns:
		--------
		Dict[str, Any]
			Tool execution result.
	"""
	parsed = extract_json_object_from_text( tool_call_text )
	normalized = normalize_tool_call( parsed )
	return execute_allowlisted_function( normalized )

build_tool_result_final_prompt

build_tool_result_final_prompt(
    user_task: str, tool_result: Dict[str, Any]
) -> str

Purpose:

Build a final-answer prompt from a validated tool execution result.

Parameters:

user_task : str Original user task.

Dict[str, Any]

Executed tool result.

Returns:

str Final answer prompt.

Source code in app.py

def build_tool_result_final_prompt( user_task: str, tool_result: Dict[ str, Any ] ) -> str:
	"""
		Purpose:
		--------
		Build a final-answer prompt from a validated tool execution result.

		Parameters:
		-----------
		user_task : str
			Original user task.

		tool_result : Dict[str, Any]
			Executed tool result.

		Returns:
		--------
		str
			Final answer prompt.
	"""
	return f'''Use the validated app tool result below to answer the user.

		User task:
		{str( user_task or '' ).strip( )}

		Tool used:
		{tool_result.get( 'name', '' )}

		Tool arguments:
		{tool_result.get( 'arguments', { } )}

		Tool result:
		{tool_result.get( 'result', '' )}

		Return a useful final answer grounded in the tool result.'''

generate_tool_grounded_final_answer

generate_tool_grounded_final_answer(
    user_task: str, tool_result: Dict[str, Any]
) -> str

Purpose:

Generate a final answer grounded in an executed tool result.

Parameters:

user_task : str Original user task.

Dict[str, Any]

Executed tool result.

Returns:

str Model-generated final answer.

Source code in app.py

def generate_tool_grounded_final_answer( user_task: str, tool_result: Dict[ str, Any ] ) -> str:
	"""
		Purpose:
		--------
		Generate a final answer grounded in an executed tool result.

		Parameters:
		-----------
		user_task : str
			Original user task.

		tool_result : Dict[str, Any]
			Executed tool result.

		Returns:
		--------
		str
			Model-generated final answer.
	"""
	try:
		return run_llm_turn(
			user_input=build_tool_result_final_prompt(
				user_task=user_task,
				tool_result=tool_result
			),
			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
		)
	except Exception as e:
		return f'Final answer generation failed: {e}'

send_web_context_to_text_generation

send_web_context_to_text_generation(
    context_text: str,
) -> None

Purpose:

Send fetched web context into the shared Text Generation document context buffer.

Parameters:

context_text : str Web context text.

Returns:

None

Source code in app.py

def send_web_context_to_text_generation( context_text: str ) -> None:
	"""
		Purpose:
		--------
		Send fetched web context into the shared Text Generation document context buffer.

		Parameters:
		-----------
		context_text : str
			Web context text.

		Returns:
		--------
		None
	"""
	context_value = str( context_text or '' ).strip( )
	if not context_value:
		return

	existing_docs = st.session_state.get( 'basic_docs', [ ] )
	if not isinstance( existing_docs, list ):
		existing_docs = [ ]

	existing_docs.append( context_value )
	st.session_state[ 'basic_docs' ] = existing_docs
	st.session_state[ 'use_document_context' ] = True

get_model_logo_for_state

get_model_logo_for_state(model_name: str) -> str

Purpose:

Return the logo path associated with the selected model.

Parameters:

model_name : str Selected local model name.

Returns:

str Configured logo path.

Source code in app.py

def get_model_logo_for_state( model_name: str ) -> str:
	"""
		Purpose:
		--------
		Return the logo path associated with the selected model.

		Parameters:
		-----------
		model_name : str
			Selected local model name.

		Returns:
		--------
		str
			Configured logo path.
	"""
	model_logo_map: Dict[ str, str ] = {
			'Bro': getattr( cfg, 'BRO_LOGO', '' ),
			'Gipity': getattr( cfg, 'GIPITY_LOGO', '' ),
			'Buddy': getattr( cfg, 'BUDDY_LOGO', '' ),
			'Boo': getattr( cfg, 'BOO_LOGO', '' ),
			'Jimi': getattr( cfg, 'JIMI_LOGO', '' ),
			'Leeroy': getattr( cfg, 'LEEROY_LOGO', '' ),
			'Nisty': getattr( cfg, 'NISTY_LOGO', '' )
	}

	return str( model_logo_map.get( str( model_name or '' ), '' ) or '' )

resolve_resource_path

resolve_resource_path(path: str) -> Path

Purpose:

Resolve a configured resource path relative to the application base directory when the supplied path is not already absolute.

Parameters:

path : str Configured resource path.

Returns:

Path Resolved resource path.

Source code in app.py

def resolve_resource_path( path: str ) -> Path:
	"""
		Purpose:
		--------
		Resolve a configured resource path relative to the application base directory
		when the supplied path is not already absolute.

		Parameters:
		-----------
		path : str
			Configured resource path.

		Returns:
		--------
		Path
			Resolved resource path.
	"""
	path_value = str( path or '' ).strip( )
	if not path_value:
		return Path( '' )

	resource_path = Path( path_value )
	if resource_path.is_absolute( ):
		return resource_path

	return Path( cfg.BASE_DIR ) / resource_path

render_selected_model_logo

render_selected_model_logo(
    model_name: str, size: str = "large"
) -> None

Purpose:

Render the selected model logo using Streamlit's native logo API so the logo remains visible when the sidebar is collapsed.

Parameters:

model_name : str Selected local model name.

str

Streamlit logo size. Expected values are 'small', 'medium', or 'large'.

Returns:

None

Source code in app.py

def render_selected_model_logo( model_name: str, size: str='large' ) -> None:
	"""
		Purpose:
		--------
		Render the selected model logo using Streamlit's native logo API so the logo
		remains visible when the sidebar is collapsed.

		Parameters:
		-----------
		model_name : str
			Selected local model name.

		size : str
			Streamlit logo size. Expected values are 'small', 'medium', or 'large'.

		Returns:
		--------
		None
	"""
	logo_path = get_model_logo_for_state( model_name )
	resolved_logo_path = resolve_resource_path( logo_path )
	if logo_path and resolved_logo_path.exists( ):
		st.logo( image=str( resolved_logo_path ), icon_image=str( resolved_logo_path ), size=size )
		return

	default_logo_path = resolve_resource_path( getattr( cfg, 'LOGO', '' ) )
	if default_logo_path.exists( ):
		st.logo( image=str( default_logo_path ), icon_image=str( default_logo_path ), size=size )

normalize_text

normalize_text(text: str) -> str

Purpose

Normalize text by: • Converting to lowercase • Removing punctuation except sentence delimiters (. ! ?) • Ensuring clean sentence boundary spacing • Collapsing whitespace

Parameters

text: str

Returns

str

Source code in app.py

def normalize_text( text: str ) -> str:
	"""

		Purpose
		-------
		Normalize text by:
			• Converting to lowercase
			• Removing punctuation except sentence delimiters (. ! ?)
			• Ensuring clean sentence boundary spacing
			• Collapsing whitespace

		Parameters
		----------
		text: str

		Returns
		-------
		str

	"""
	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 = None, overlap: int = None
) -> List[str]

Purpose:

Split text into overlapping chunks using session-state defaults when explicit values are not provided.

Parameters:

text : str size : int | None overlap : int | None

Returns:

List[str]

Source code in app.py

def chunk_text( text: str, size: int=None, overlap: int=None ) -> List[ str ]:
	"""
		Purpose:
		--------
		Split text into overlapping chunks using session-state defaults when explicit values
		are not provided.

		Parameters:
		-----------
		text : str
		size : int | None
		overlap : int | None

		Returns:
		--------
		List[str]
	"""
	if not text:
		return [ ]

	chunk_size = int(
		size if size is not None else st.session_state.get( 'retrieval_chunk_size', 1200 ) )
	chunk_overlap = int(
		overlap if overlap is not None else st.session_state.get( 'retrieval_chunk_overlap', 200 ) )

	if chunk_size <= 0:
		chunk_size = 1200

	if chunk_overlap < 0:
		chunk_overlap = 0

	if chunk_overlap >= chunk_size:
		chunk_overlap = max( 0, chunk_size // 4 )

	chunks: List[ str ] = [ ]
	i = 0
	step = max( 1, chunk_size - chunk_overlap )

	while i < len( text ):
		chunk = text[ i:i + chunk_size ]
		if chunk and chunk.strip( ):
			chunks.append( chunk )
		i += step

	return chunks

convert_xml

convert_xml(text: str) -> str

Purpose:

---

Convert XML-delimited prompt text into Markdown by treating XML-like tags as section delimiters, not as strict XML.

Parameters:

text (str) - Prompt text containing XML-like opening and closing tags.

Returns:

Markdown-formatted text using level-2 headings (##).

Source code in app.py

def convert_xml( text: str ) -> str:
	"""

			Purpose:
			_________
			Convert XML-delimited prompt text into Markdown by treating XML-like
			tags as section delimiters, not as strict XML.

			Parameters:
			-----------
			text (str) - Prompt text containing XML-like opening and closing tags.

			Returns:
			---------
			Markdown-formatted text using level-2 headings (##).
	"""
	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 )

convert_markdown

convert_markdown(text: Any) -> str

Purpose:

Convert between Markdown headings and simple XML-like heading tags.

Behavior:

Auto-detects direction: - If

...

/

...

... exist, converts to Markdown (# / ## / ###). - Otherwise converts Markdown headings (# / ## / ###) to ... tags.

Parameters:

text : Any Source text. Non-string values return "".

Returns:

str Converted text.

Source code in app.py

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

		Behavior:
		---------
		Auto-detects direction:
		  - If <h1>...</h1> / <h2>...</h2> ... exist, converts to Markdown (# / ## / ###).
		  - Otherwise converts Markdown headings (# / ## / ###) to <hN>...</hN> tags.

		Parameters:
		-----------
		text : Any
			Source text. Non-string values return "".

		Returns:
		--------
		str
			Converted text.
	"""
	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

Purpose:

---

Set the the format via css.

Source code in app.py

def inject_response_css( ) -> None:
	"""

		Purpose:
		_________
		Set the the format via css.

	"""
	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

Purpose:

---

Sets the style of subheaders in the main UI

Source code in app.py

def style_subheaders( ) -> None:
	"""

		Purpose:
		_________
		Sets the style of subheaders in the main UI

	"""
	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, )

fetch_prompt_names

fetch_prompt_names(db_path: str) -> list[str]

Purpose:

Retrieve template names from Prompts table.

Parameters:

db_path : str SQLite database path.

Returns:

list[str] Sorted prompt names.

Source code in app.py

def fetch_prompt_names( db_path: str ) -> list[ str ]:
	"""
		Purpose:
		--------
		Retrieve template names from Prompts table.

		Parameters:
		-----------
		db_path : str
			SQLite database path.

		Returns:
		--------
		list[str]
			Sorted prompt names.
	"""
	try:
		conn = sqlite3.connect( db_path )
		cur = conn.cursor( )
		cur.execute( "SELECT Caption FROM Prompts ORDER BY PromptsId;" )
		rows = cur.fetchall( )
		conn.close( )
		return [ r[ 0 ] for r in rows if r and r[ 0 ] is not None ]
	except Exception:
		return [ ]

fetch_prompt_text

fetch_prompt_text(db_path: str, name: str) -> str | None

Purpose:

Retrieve template text by name.

Parameters:

db_path : str SQLite database path. name : str Template name.

Returns:

str | None Prompt text if found.

Source code in app.py

def fetch_prompt_text( db_path: str, name: str ) -> str | None:
	"""
		Purpose:
		--------
		Retrieve template text by name.

		Parameters:
		-----------
		db_path : str
			SQLite database path.
		name : str
			Template name.

		Returns:
		--------
		str | None
			Prompt text if found.
	"""
	try:
		conn = sqlite3.connect( db_path )
		cur = conn.cursor( )
		cur.execute( "SELECT Text FROM Prompts WHERE Caption = ?;", (name,) )
		row = cur.fetchone( )
		conn.close( )
		return str( row[ 0 ] ) if row and row[ 0 ] is not None else None
	except Exception:
		return None

get_effective_system_instructions

get_effective_system_instructions() -> str

Purpose:

Return the authoritative system instructions text from session state.

Parameters:

None

Returns:

str

Source code in app.py

def get_effective_system_instructions( ) -> str:
	"""
		Purpose:
		--------
		Return the authoritative system instructions text from session state.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
	"""
	text = st.session_state.get( 'system_instructions', '' )
	return str( text ).strip( ) if text is not None else ''

build_task_instruction_block

build_task_instruction_block() -> str

Purpose:

Build a task-specific instruction block for Text Generation mode, including model-gated Thinking, Coding, and Function Calling directives for Gemma 4 and GPT-OSS-aligned local models.

Parameters:

None

Returns:

str Task instruction block.

Source code in app.py

def build_task_instruction_block( ) -> str:
	"""
		Purpose:
		--------
		Build a task-specific instruction block for Text Generation mode, including
		model-gated Thinking, Coding, and Function Calling directives for Gemma 4 and
		GPT-OSS-aligned local models.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Task instruction block.
	"""
	task_preset = str( st.session_state.get( 'task_preset', 'Chat' ) or 'Chat' ).strip( )
	response_format = str(
		st.session_state.get( 'response_format', 'Markdown' ) or 'Markdown' ).strip( )
	reasoning_depth = str(
		st.session_state.get( 'reasoning_depth', 'Medium' ) or 'Medium' ).strip( )
	answer_only = bool( st.session_state.get( 'answer_only', False ) )
	use_self_check = bool( st.session_state.get( 'use_self_check', False ) )
	deterministic_reasoning = bool( st.session_state.get( 'deterministic_reasoning', False ) )
	coding_language = str(
		st.session_state.get( 'coding_language', 'Python' ) or 'Python' ).strip( )
	coding_task = str( st.session_state.get( 'coding_task', 'Generate' ) or 'Generate' ).strip( )
	coding_include_comments = bool( st.session_state.get( 'coding_include_comments', True ) )
	coding_editor_format = bool( st.session_state.get( 'coding_editor_format', True ) )
	coding_fenced_output = bool( st.session_state.get( 'coding_fenced_output', True ) )
	translation_target_language = (
			str( st.session_state.get(
				'translation_target_language', 'English' ) or 'English' ).strip( ))

	thinking_mode_enabled = bool( st.session_state.get( 'thinking_mode_enabled', False ) )
	thinking_effort = str(
		st.session_state.get( 'thinking_effort', 'Medium' ) or 'Medium' ).strip( )
	thinking_summary_enabled = bool(
		st.session_state.get( 'thinking_summary_enabled', True ) )
	coding_mode_enabled = bool( st.session_state.get( 'coding_mode_enabled', False ) )
	coding_test_request = bool( st.session_state.get( 'coding_test_request', False ) )
	coding_explain_request = bool( st.session_state.get( 'coding_explain_request', False ) )
	function_call_enabled = bool( st.session_state.get( 'function_call_enabled', False ) )
	function_schema_text = str(
		st.session_state.get( 'function_schema_text', '' ) or '' ).strip( )

	lines: List[ str ] = [ ]
	lines.append( 'Task Preset:' )
	lines.append( f'- Active Task: {task_preset}' )
	lines.append( f'- Response Format: {response_format}' )

	if task_preset == 'Reasoning':
		lines.append( f'- Reasoning Depth: {reasoning_depth}' )
		lines.append(
			'- Use a careful analytical process internally and return a clear final answer.' )

		if answer_only:
			lines.append( '- Return the final answer without extra prefatory narration.' )
		if use_self_check:
			lines.append( '- Verify the conclusion against the prompt before answering.' )
		if deterministic_reasoning:
			lines.append( '- Prefer stable, conservative reasoning over creative variation.' )

	elif task_preset == 'Coding':
		lines.append( f'- Code Language: {coding_language}' )
		lines.append( f'- Coding Task: {coding_task}' )
		if coding_include_comments:
			lines.append(
				'- Include documentation comments and useful inline comments when appropriate.' )
		else:
			lines.append( '- Minimize comments unless required for clarity.' )
		if coding_editor_format:
			lines.append(
				'- Format the output as editor-ready source code, not as explanatory pseudo-code.' )
		if coding_fenced_output:
			lines.append(
				'- Return code inside fenced markdown code blocks when code is produced.' )
		else:
			lines.append(
				'- Return raw code without fenced markdown blocks when code is produced.' )

	elif task_preset == 'Translation':
		lines.append( f'- Translate the user content into {translation_target_language}.' )
		lines.append( '- Preserve original meaning, tone, and structure where practical.' )

	elif task_preset == 'Summarization':
		lines.append( '- Summarize the user content clearly and faithfully.' )
		lines.append( '- Preserve key facts, names, dates, and conclusions.' )

	elif task_preset == 'Extraction':
		lines.append( '- Extract the requested facts faithfully and do not invent missing values.' )
		if response_format == 'JSON':
			lines.append( '- Return valid JSON only.' )

	else:
		lines.append( '- Respond as a general-purpose assistant.' )

	if thinking_mode_enabled and model_supports_capability( 'thinking' ):
		lines.append( '' )
		lines.append( 'Thinking Capability:' )
		lines.append( f'- Thinking Effort: {thinking_effort}' )
		lines.append(
			'- Use private internal reasoning to solve the task, but do not expose hidden '
			'chain-of-thought.' )

		if thinking_summary_enabled:
			lines.append(
				'- Provide a concise reasoning summary only when it improves the usefulness '
				'of the final answer.' )
		else:
			lines.append(
				'- Return the final answer without a separate reasoning summary.' )

	elif thinking_mode_enabled:
		lines.append( '' )
		lines.append( 'Thinking Capability:' )
		lines.append(
			f'- {get_selected_model_name( )} does not advertise the Thinking capability. '
			'Use the standard reasoning controls only.' )

	if coding_mode_enabled and model_supports_capability( 'coding' ):
		lines.append( '' )
		lines.append( 'Advanced Coding Capability:' )
		lines.append( f'- Primary Language: {coding_language}' )
		lines.append( f'- Requested Coding Operation: {coding_task}' )
		lines.append(
			'- Preserve existing application structure and generate paste-ready source code.' )

		if coding_test_request:
			lines.append(
				'- Include a minimal verification or test strategy when appropriate.' )

		if coding_explain_request:
			lines.append(
				'- Include a concise explanation of the implementation after the code.' )

	elif coding_mode_enabled:
		lines.append( '' )
		lines.append( 'Advanced Coding Capability:' )
		lines.append(
			f'- {get_selected_model_name( )} does not advertise the advanced Coding '
			'capability. Use the standard coding task controls only.' )

	if function_call_enabled and model_supports_capability( 'function_calling' ):
		lines.append( '' )
		lines.append( 'Function Calling Capability:' )
		lines.append(
			'- When a function call is required, return a single strict JSON object and no '
			'extra prose.' )
		lines.append(
			'- The JSON object must use this shape: '
			'{"name":"function_name","arguments":{...}}' )
		lines.append(
			'- Do not invent functions. Use only the function schema supplied below.' )

		if is_gipity_model( ):
			lines.append(
				'- For GPT-OSS/Gipity, treat function calling as an app-mediated tool call. '
				'The app will validate and execute only allowlisted functions.' )

		if is_gemma4_model( ):
			lines.append(
				'- For Gemma 4 models, generate a valid function-call object only when the '
				'user request clearly requires structured tool invocation.' )

		if function_schema_text:
			lines.append( '' )
			lines.append( 'Available Function Schema:' )
			lines.append( function_schema_text )

	elif function_call_enabled:
		lines.append( '' )
		lines.append( 'Function Calling Capability:' )
		lines.append(
			f'- {get_selected_model_name( )} does not advertise function calling. '
			'Answer normally and do not emit tool-call JSON.' )

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

build_effective_prompt_preview

build_effective_prompt_preview(user_input: str) -> str

Purpose:

Build a readable preview of the effective prompt content used for generation.

Parameters:

user_input : str

Returns:

str

Source code in app.py

def build_effective_prompt_preview( user_input: str ) -> str:
	"""
		Purpose:
		--------
		Build a readable preview of the effective prompt content used for generation.

		Parameters:
		-----------
		user_input : str

		Returns:
		--------
		str
	"""
	system_instructions = get_effective_system_instructions( )
	task_block = build_task_instruction_block( )
	preview_parts: List[ str ] = [ ]

	if system_instructions:
		preview_parts.append( '[System Instructions]' )
		preview_parts.append( system_instructions )

	if task_block:
		preview_parts.append( '[Task Instructions]' )
		preview_parts.append( task_block )

	preview_parts.append( '[User Input]' )
	preview_parts.append( user_input or '' )

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

get_preset_system_instruction

get_preset_system_instruction(task_preset: str) -> str

Purpose:

Return a starter system-instruction preset for the selected task type.

Parameters:

task_preset : str Selected task preset.

Returns:

str System-instruction preset text.

Source code in app.py

def get_preset_system_instruction( task_preset: str ) -> str:
	"""
		Purpose:
		--------
		Return a starter system-instruction preset for the selected task type.

		Parameters:
		-----------
		task_preset : str
			Selected task preset.

		Returns:
		--------
		str
			System-instruction preset text.
	"""
	preset_name = str( task_preset or 'Chat' ).strip( )

	preset_map: Dict[ str, str ] = {
			'Chat':
				'You are Loca, a helpful local assistant. Be accurate, practical, and concise.',
			'Reasoning':
				'Solve the task carefully. Use a careful internal process, then return a clear '
				'final answer.',
			'Coding':
				'Produce correct, editor-ready code. Preserve the requested language, structure, '
				'and implementation intent.',
			'Translation':
				'Translate faithfully while preserving meaning, tone, and structure.',
			'Summarization':
				'Summarize faithfully and preserve key facts, names, dates, and conclusions.',
			'Extraction':
				'Extract only supported facts. Do not invent missing values.'
	}

	return preset_map.get( preset_name, preset_map[ 'Chat' ] )

get_system_instruction_action_key

get_system_instruction_action_key(prefix: str) -> str

Purpose:

Return the pending action key used by a specific System Instructions renderer.

Parameters:

prefix : str Renderer prefix.

Returns:

str Pending action session-state key.

Source code in app.py

def get_system_instruction_action_key( prefix: str ) -> str:
	"""
		Purpose:
		--------
		Return the pending action key used by a specific System Instructions renderer.

		Parameters:
		-----------
		prefix : str
			Renderer prefix.

		Returns:
		--------
		str
			Pending action session-state key.
	"""
	return f'{prefix}_pending_system_instruction_action'

get_system_instruction_template_key

get_system_instruction_template_key(prefix: str) -> str

Purpose:

Return the template-select widget key used by a specific System Instructions renderer.

Parameters:

prefix : str Renderer prefix.

Returns:

str Template widget session-state key.

Source code in app.py

def get_system_instruction_template_key( prefix: str ) -> str:
	"""
		Purpose:
		--------
		Return the template-select widget key used by a specific System Instructions
		renderer.

		Parameters:
		-----------
		prefix : str
			Renderer prefix.

		Returns:
		--------
		str
			Template widget session-state key.
	"""
	return f'{prefix}_instructions_template'

get_system_instruction_pending_template_key

get_system_instruction_pending_template_key(
    prefix: str,
) -> str

Purpose:

Return the pending template key used by a specific System Instructions renderer.

Parameters:

prefix : str Renderer prefix.

Returns:

str Pending template session-state key.

Source code in app.py

def get_system_instruction_pending_template_key( prefix: str ) -> str:
	"""
		Purpose:
		--------
		Return the pending template key used by a specific System Instructions renderer.

		Parameters:
		-----------
		prefix : str
			Renderer prefix.

		Returns:
		--------
		str
			Pending template session-state key.
	"""
	return f'{prefix}_pending_system_template_name'

request_system_template_change

request_system_template_change(prefix: str) -> None

Purpose:

Request a system-instruction template change without directly modifying the widget-owned system_instructions key.

Parameters:

prefix : str Renderer prefix.

Returns:

None

Source code in app.py

def request_system_template_change( prefix: str ) -> None:
	"""
		Purpose:
		--------
		Request a system-instruction template change without directly modifying the
		widget-owned system_instructions key.

		Parameters:
		-----------
		prefix : str
			Renderer prefix.

		Returns:
		--------
		None
	"""
	template_key = get_system_instruction_template_key( prefix )
	pending_key = get_system_instruction_pending_template_key( prefix )
	name = st.session_state.get( template_key, None )

	if name:
		st.session_state[ pending_key ] = str( name )

request_system_instruction_action

request_system_instruction_action(
    prefix: str, action: str
) -> None

Purpose:

Request a pending system-instruction action without directly modifying the widget-owned system_instructions key.

Parameters:

prefix : str Renderer prefix.

str

Requested action name.

Returns:

None

Source code in app.py

def request_system_instruction_action( prefix: str, action: str ) -> None:
	"""
		Purpose:
		--------
		Request a pending system-instruction action without directly modifying the
		widget-owned system_instructions key.

		Parameters:
		-----------
		prefix : str
			Renderer prefix.

		action : str
			Requested action name.

		Returns:
		--------
		None
	"""
	action_key = get_system_instruction_action_key( prefix )
	st.session_state[ action_key ] = str( action or '' )

process_pending_system_instruction_requests

process_pending_system_instruction_requests(
    prefix: str,
) -> None

Purpose:

Process pending System Instructions requests before the system_instructions text area is instantiated. This is the only safe place to modify the shared system_instructions widget-owned key.

Parameters:

prefix : str Renderer prefix.

Returns:

None

Source code in app.py

def process_pending_system_instruction_requests( prefix: str ) -> None:
	"""
		Purpose:
		--------
		Process pending System Instructions requests before the system_instructions text
		area is instantiated. This is the only safe place to modify the shared
		system_instructions widget-owned key.

		Parameters:
		-----------
		prefix : str
			Renderer prefix.

		Returns:
		--------
		None
	"""
	template_key = get_system_instruction_template_key( prefix )
	pending_template_key = get_system_instruction_pending_template_key( prefix )
	action_key = get_system_instruction_action_key( prefix )

	pending_template_name = st.session_state.pop( pending_template_key, None )
	if pending_template_name:
		template_text = fetch_prompt_text( cfg.DB_PATH, str( pending_template_name ) )

		if template_text is not None:
			st.session_state[ 'system_instructions' ] = str( template_text )
			st.session_state[ 'active_prompt_caption' ] = str( pending_template_name )

	pending_action = st.session_state.pop( action_key, None )
	if pending_action == 'clear':
		st.session_state[ 'system_instructions' ] = ''
		st.session_state[ 'active_prompt_caption' ] = ''

		if template_key in st.session_state:
			del st.session_state[ template_key ]

	elif pending_action == 'convert':
		text = st.session_state.get( 'system_instructions', '' )

		if isinstance( text, str ) and text.strip( ):
			src = text.strip( )

			if cfg.XML_BLOCK_PATTERN.search( src ):
				converted = convert_xml( src )
			else:
				converted = convert_markdown( src )

			st.session_state[ 'system_instructions' ] = converted

	elif pending_action == 'apply_preset':
		task_preset = str( st.session_state.get( 'task_preset', 'Chat' ) or 'Chat' ).strip( )
		st.session_state[ 'system_instructions' ] = get_preset_system_instruction( task_preset )
		st.session_state[ 'active_prompt_caption' ] = f'{task_preset} Preset'

render_system_instructions

render_system_instructions(
    prefix: str,
    include_apply_preset: bool = False,
    include_preview: bool = False,
) -> None

Purpose:

Render a Streamlit-safe shared System Instructions control surface. All writes to the widget-owned system_instructions key are processed before the text-area widget is instantiated.

Parameters:

prefix : str Unique renderer prefix, such as 'text' or 'docqna'.

bool

When True, show the Apply Preset button.

bool

When True, show the Preview Prompt button and preview text area.

Returns:

None

Source code in app.py

def render_system_instructions( prefix: str, include_apply_preset: bool = False,
		include_preview: bool = False ) -> None:
	"""
		Purpose:
		--------
		Render a Streamlit-safe shared System Instructions control surface. All writes to
		the widget-owned system_instructions key are processed before the text-area widget
		is instantiated.

		Parameters:
		-----------
		prefix : str
			Unique renderer prefix, such as 'text' or 'docqna'.

		include_apply_preset : bool
			When True, show the Apply Preset button.

		include_preview : bool
			When True, show the Preview Prompt button and preview text area.

		Returns:
		--------
		None
	"""
	process_pending_system_instruction_requests( prefix )

	template_key = get_system_instruction_template_key( prefix )
	prompt_names = fetch_prompt_names( cfg.DB_PATH )

	if not prompt_names:
		prompt_names = [ '' ]

	in_left, in_right = st.columns( [ 0.8, 0.2 ] )

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

	with in_right:
		st.selectbox( label='Use Template', options=prompt_names,
			index=None, key=template_key,
			on_change=request_system_template_change, args=(prefix,) )

	if include_apply_preset and include_preview:
		btn_c1, btn_c2, btn_c3, btn_c4 = st.columns( [ 0.35, 0.2, 0.2, 0.25 ] )

		with btn_c1:
			st.button( label='Clear Instructions', width='stretch',
				on_click=request_system_instruction_action, args=(prefix, 'clear') )

		with btn_c2:
			st.button( label='XML <-> Markdown', width='stretch',
				on_click=request_system_instruction_action, args=(prefix, 'convert') )

		with btn_c3:
			st.button( label='Apply Preset', width='stretch',
				on_click=request_system_instruction_action, args=(prefix, 'apply_preset') )

		with btn_c4:
			if st.button( label='Preview Prompt', width='stretch', key=f'{prefix}_preview_prompt' ):
				st.session_state[ 'preview_effective_prompt' ] = not bool(
					st.session_state.get( 'preview_effective_prompt', False ) )

	else:
		btn_c1, btn_c2 = st.columns( [ 0.8, 0.2 ] )

		with btn_c1:
			st.button( label='Clear Instructions', width='stretch',
				on_click=request_system_instruction_action, args=(prefix, 'clear') )

		with btn_c2:
			st.button( label='XML <-> Markdown', width='stretch', on_click=request_system_instruction_action,
				args=(prefix, 'convert') )

	if include_preview and bool( st.session_state.get( 'preview_effective_prompt', False ) ):
		user_preview_input = str( st.session_state.get( 'last_preview_input', '' ) or '' )

		st.text_area( label='Effective Prompt Preview',
			value=build_effective_prompt_preview( user_preview_input ), height=220,
			disabled=True, key=f'{prefix}_effective_prompt_preview' )

get_runtime_llm

get_runtime_llm() -> Any | None

Purpose:

Load the selected llama.cpp model using the currently selected model path and runtime settings.

Parameters:

None

Returns:

Any | None Loaded llama.cpp model instance when available; otherwise None.

Source code in app.py

def get_runtime_llm( ) -> Any | None:
	"""
		Purpose:
		--------
		Load the selected llama.cpp model using the currently selected model path and
		runtime settings.

		Parameters:
		-----------
		None

		Returns:
		--------
		Any | None
			Loaded llama.cpp model instance when available; otherwise None.
	"""
	synchronize_model_derived_state( )

	model_path = str(
		st.session_state.get(
			'selected_model_path',
			get_model_path_for_state( get_selected_model_name( ) )
		) or get_model_path_for_state( get_selected_model_name( ) )
	)

	ctx_value = int( st.session_state.get( 'context_window', cfg.DEFAULT_CTX ) or cfg.DEFAULT_CTX )
	thread_value = int( st.session_state.get( 'cpu_threads', cfg.CORES ) or cfg.CORES )
	seed_value = int( st.session_state.get( 'random_seed', -1 ) or -1 )

	if ctx_value <= 0:
		ctx_value = int( cfg.DEFAULT_CTX )

	if thread_value <= 0:
		thread_value = int( cfg.CORES )

	return load_llm( model_path=model_path, ctx=ctx_value,
		threads=thread_value, seed=seed_value )

build_prompt

build_prompt(user_input: str) -> str

Purpose:

Build a llama.cpp-compatible prompt using unified system instructions, task-specific Text Generation settings, optional semantic/basic context, and chat history. Semantic context retrieval is guarded so Text Generation cannot crash when the embedder, database, embeddings table, or stored vectors are unavailable or inconsistent.

Parameters:

user_input : str User prompt text supplied by the Text Generation mode.

Returns:

str Prompt text formatted for the local llama.cpp chat template.

Source code in app.py

def build_prompt( user_input: str ) -> str:
	"""
		Purpose:
		--------
		Build a llama.cpp-compatible prompt using unified system instructions, task-specific
		Text Generation settings, optional semantic/basic context, and chat history. Semantic
		context retrieval is guarded so Text Generation cannot crash when the embedder,
		database, embeddings table, or stored vectors are unavailable or inconsistent.

		Parameters:
		-----------
		user_input : str
			User prompt text supplied by the Text Generation mode.

		Returns:
		--------
		str
			Prompt text formatted for the local llama.cpp chat template.
	"""
	global embedder

	system_instructions = get_effective_system_instructions( )
	task_block = build_task_instruction_block( )
	use_semantic = bool( st.session_state.get( 'use_semantic', False ) )
	use_chat_history = bool( st.session_state.get( 'use_chat_history', True ) )
	use_document_context = bool( st.session_state.get( 'use_document_context', False ) )
	basic_docs = st.session_state.get( 'basic_docs', [ ] )
	messages = st.session_state.get( 'messages', [ ] )
	user_text = str( user_input or '' )

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

	system_parts: List[ str ] = [ ]
	if system_instructions:
		system_parts.append( str( system_instructions ) )
	if task_block:
		system_parts.append( str( task_block ) )

	system_text = '\n\n'.join( [ p for p in system_parts if p ] ).strip( )

	prompt = ''
	if system_text:
		prompt += f'<|system|>\n{system_text}\n</s>\n'

	if use_semantic:
		try:
			if not is_embedder_available( globals( ).get( 'embedder', None ) ):
				st.session_state[ 'semantic_status' ] = get_embedder_unavailable_message( )
			else:
				with sqlite3.connect( cfg.DB_PATH ) as conn:
					rows = conn.execute(
						'SELECT chunk, vector FROM embeddings' ).fetchall( )

				if rows:
					q_raw = embedder.encode( [ user_text ], show_progress_bar=False )[ 0 ]
					q = np.asarray( q_raw, dtype=np.float32 ).reshape( -1 )
					scored: List[ Tuple[ str, float ] ] = [ ]

					for chunk, vector_blob in rows:
						if not chunk or vector_blob is None:
							continue

						vector = np.frombuffer( vector_blob, dtype=np.float32 )
						if vector.size == 0 or vector.size != q.size:
							continue

						score = cosine_similarity( q, vector )
						scored.append( (str( chunk ), score) )

					if scored:
						scored = sorted( scored, key=lambda x: x[ 1 ], reverse=True )
						for chunk, score in scored[ :top_k_value ]:
							prompt += f'<|system|>\nSemantic Context:\n{chunk}\n</s>\n'

						st.session_state[ 'semantic_status' ] = (
								f'Loaded {min( len( scored ), top_k_value )} semantic context '
								f'chunk(s) for Text Generation.')
					else:
						st.session_state[ 'semantic_status' ] = (
								'Semantic context was enabled, but no compatible embedding '
								'vectors were available for Text Generation.')
		except Exception as e:
			st.session_state[ 'semantic_status' ] = (
					f'Semantic context was skipped because retrieval failed: {e}')

	if use_document_context and isinstance( basic_docs, list ):
		for document_text in basic_docs[ :6 ]:
			if document_text:
				prompt += f'<|system|>\nDocument Context:\n{document_text}\n</s>\n'

	if use_chat_history and 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 in ('user', 'assistant', 'system'):
				prompt += f'<|{role}|>\n{content}\n</s>\n'

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

build_llama_call_args

build_llama_call_args(
    max_tokens: int,
    temperature: float,
    top_p: float,
    repeat_penalty: float,
    stream: bool,
) -> Dict[str, Any]

Purpose:

Build llama.cpp generation arguments from the current Streamlit runtime settings.

Parameters:

max_tokens : int Maximum number of generated tokens.

float

Sampling temperature.

float

Nucleus sampling value.

float

Repeat penalty value.

bool

Whether streaming output is requested.

Returns:

Dict[str, Any] Generation argument dictionary for llama.cpp.

Source code in app.py

def build_llama_call_args( max_tokens: int, temperature: float, top_p: float,
		repeat_penalty: float, stream: bool ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Build llama.cpp generation arguments from the current Streamlit runtime settings.

		Parameters:
		-----------
		max_tokens : int
			Maximum number of generated tokens.

		temperature : float
			Sampling temperature.

		top_p : float
			Nucleus sampling value.

		repeat_penalty : float
			Repeat penalty value.

		stream : bool
			Whether streaming output is requested.

		Returns:
		--------
		Dict[str, Any]
			Generation argument dictionary for llama.cpp.
	"""
	max_token_value = int( max_tokens ) if int( max_tokens or 0 ) > 0 else 1024
	temperature_value = float( temperature ) if temperature is not None else 0.0
	top_p_value = float( top_p ) if top_p is not None else 0.95
	repeat_penalty_value = float( repeat_penalty ) if repeat_penalty is not None else 1.1
	top_k_value = int( st.session_state.get( 'top_k', 0 ) or 0 )
	repeat_window_value = int( st.session_state.get( 'repeat_window', 0 ) or 0 )

	call_args: Dict[ str, Any ] = {
			'stream': bool( stream ),
			'max_tokens': max_token_value,
			'temperature': temperature_value,
			'top_p': top_p_value,
			'repeat_penalty': repeat_penalty_value,
			'stop': [ '</s>' ]
	}

	if top_k_value > 0:
		call_args[ 'top_k' ] = top_k_value

	if repeat_window_value > 0:
		call_args[ 'repeat_last_n' ] = repeat_window_value

	return call_args

get_missing_model_message

get_missing_model_message() -> str

Purpose:

Build a clear user-facing message when the selected local GGUF model or required llama.cpp dependency is not available.

Parameters:

None

Returns:

str Model availability message.

Source code in app.py

def get_missing_model_message( ) -> str:
	"""
		Purpose:
		--------
		Build a clear user-facing message when the selected local GGUF model or required
		llama.cpp dependency is not available.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Model availability message.
	"""
	model_name = get_selected_model_name( )
	model_path = get_selected_model_path( )

	if Llama is None:
		return (
				'Local model inference is unavailable because `llama-cpp-python` is not '
				'installed or could not be imported.\n\n'
				f'**Selected Model:** {model_name}\n\n'
				'Install or repair `llama-cpp-python`, then restart the Streamlit app.'
		)

	if model_path:
		return (
				'The selected local GGUF model is unavailable.\n\n'
				f'**Model:** {model_name}\n\n'
				f'**Configured Path:** `{model_path}`\n\n'
				'Verify that the file exists or update the model path in `config.py` '
				'or the corresponding environment variable.'
		)

	return (
			'The selected local GGUF model is not configured.\n\n'
			f'**Model:** {model_name}\n\n'
			'Update the model registry path in `config.py` or provide the corresponding '
			'environment variable.'
	)

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,
) -> str

Purpose:

Run a single local llama.cpp LLM turn using the selected model, current prompt contract, and runtime settings. Missing or unavailable GGUF models are handled safely with a user-facing diagnostic response instead of a callable None failure.

Parameters:

user_input : str User prompt or prepared application prompt.

float

Sampling temperature.

float

Nucleus sampling value.

float

Repeat penalty value.

int

Maximum number of generated tokens.

bool

Whether to stream output tokens into the supplied Streamlit placeholder.

Any | None

Optional Streamlit output placeholder.

Returns:

str Generated response text or diagnostic message.

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 ) -> str:
	"""
		Purpose:
		--------
		Run a single local llama.cpp LLM turn using the selected model, current prompt
		contract, and runtime settings. Missing or unavailable GGUF models are handled
		safely with a user-facing diagnostic response instead of a callable None failure.

		Parameters:
		-----------
		user_input : str
			User prompt or prepared application prompt.

		temperature : float
			Sampling temperature.

		top_p : float
			Nucleus sampling value.

		repeat_penalty : float
			Repeat penalty value.

		max_tokens : int
			Maximum number of generated tokens.

		stream : bool
			Whether to stream output tokens into the supplied Streamlit placeholder.

		output : Any | None
			Optional Streamlit output placeholder.

		Returns:
		--------
		str
			Generated response text or diagnostic message.
	"""
	if user_input is None:
		return ''

	user_text = str( user_input or '' ).strip( )
	if not user_text:
		return ''

	synchronize_model_derived_state( )

	runtime_llm = get_runtime_llm( )
	if runtime_llm is None:
		message = get_missing_model_message( )
		if output is not None:
			output.markdown( message )
		return message

	prompt = build_prompt( user_text )
	call_args = build_llama_call_args(
		max_tokens=max_tokens,
		temperature=temperature,
		top_p=top_p,
		repeat_penalty=repeat_penalty,
		stream=stream
	)

	try:
		if not stream:
			resp = runtime_llm( prompt, **call_args )
			text = (
					resp.get( 'choices', [ { 'text': '' } ] )[ 0 ]
					.get( 'text', '' ) or ''
			)

			return str( text ).strip( )

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

		for chunk in runtime_llm( prompt, **call_args ):
			try:
				token = chunk[ 'choices' ][ 0 ][ 'text' ]
			except Exception:
				token = ''

			if token:
				buf += token
				output.markdown( buf + '▌' )

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

	except TypeError:
		fallback_args: Dict[ str, Any ] = {
				'stream': bool( stream ),
				'max_tokens': int( max_tokens ) if int( max_tokens or 0 ) > 0 else 1024,
				'temperature': float( temperature ) if temperature is not None else 0.0,
				'top_p': float( top_p ) if top_p is not None else 0.95,
				'repeat_penalty': float( repeat_penalty ) if repeat_penalty is not None else 1.1,
				'stop': [ '</s>' ]
		}

		if not stream:
			resp = runtime_llm( prompt, **fallback_args )
			text = ( resp.get( 'choices', [ { 'text': '' } ] )[ 0 ].get( 'text', '' ) or '' )

			return str( text ).strip( )

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

		for chunk in runtime_llm( prompt, **fallback_args ):
			try:
				token = chunk[ 'choices' ][ 0 ][ 'text' ]
			except Exception:
				token = ''

			if token:
				buf += token
				output.markdown( buf + '▌' )

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

	except Exception as e:
		message = (
				'Local model generation failed.\n\n'
				f'**Model:** {get_selected_model_name( )}\n\n'
				f'**Path:** `{get_selected_model_path( )}`\n\n'
				f'**Error:** {e}'
		)

		if output is not None:
			output.markdown( message )

		return message

get_prompt_categories

get_prompt_categories() -> List[str]

Purpose:

Return supported prompt categories.

Parameters:

None

Returns:

List[str]

Source code in app.py

def get_prompt_categories( ) -> List[ str ]:
	"""
		Purpose:
		--------
		Return supported prompt categories.

		Parameters:
		-----------
		None

		Returns:
		--------
		List[str]
	"""
	return [
			'General Chat',
			'Reasoning',
			'Coding',
			'Translation',
			'Summarization',
			'Extraction',
			'Document Extraction',
			'OCR',
			'JSON Output'
	]

get_prompt_task_types

get_prompt_task_types() -> List[str]

Purpose:

Return supported task types.

Parameters:

None

Returns:

List[str]

Source code in app.py

def get_prompt_task_types( ) -> List[ str ]:
	"""
		Purpose:
		--------
		Return supported task types.

		Parameters:
		-----------
		None

		Returns:
		--------
		List[str]
	"""
	return [
			'Chat',
			'Reasoning',
			'Coding',
			'Translation',
			'Summarization',
			'Extraction'
	]

infer_prompt_category

infer_prompt_category(
    prompt_row: Dict[str, Any] | None,
) -> str

Purpose:

Infer a prompt category from the prompt row content.

Parameters:

prompt_row : Dict[str, Any] | None

Returns:

str

Source code in app.py

def infer_prompt_category( prompt_row: Dict[ str, Any ] | None ) -> str:
	"""
		Purpose:
		--------
		Infer a prompt category from the prompt row content.

		Parameters:
		-----------
		prompt_row : Dict[str, Any] | None

		Returns:
		--------
		str
	"""
	if not isinstance( prompt_row, dict ):
		return 'General Chat'

	caption = str( prompt_row.get( 'Caption', '' ) or '' ).lower( )
	name = str( prompt_row.get( 'Name', '' ) or '' ).lower( )
	text = str( prompt_row.get( 'Text', '' ) or '' ).lower( )

	blob = f'{caption} {name} {text}'

	if 'json' in blob:
		return 'JSON Output'
	if 'ocr' in blob:
		return 'OCR'
	if 'document' in blob and 'extract' in blob:
		return 'Document Extraction'
	if 'extract' in blob:
		return 'Extraction'
	if 'summar' in blob:
		return 'Summarization'
	if 'translat' in blob:
		return 'Translation'
	if 'coding' in blob or 'code' in blob or 'debug' in blob or 'refactor' in blob:
		return 'Coding'
	if 'reason' in blob or 'analysis' in blob:
		return 'Reasoning'

	return 'General Chat'

build_starter_prompt_template

build_starter_prompt_template(
    category: str,
    task_type: str,
    response_format: str,
    language: str,
) -> str

Purpose:

Build a starter prompt template from high-level prompt metadata.

Parameters:

category : str task_type : str response_format : str language : str

Returns:

str

Source code in app.py

def build_starter_prompt_template( category: str, task_type: str, response_format: str,
		language: str ) -> str:
	"""
		Purpose:
		--------
		Build a starter prompt template from high-level prompt metadata.

		Parameters:
		-----------
		category : str
		task_type : str
		response_format : str
		language : str

		Returns:
		--------
		str
	"""
	category_value = str( category or 'General Chat' ).strip( )
	task_value = str( task_type or 'Chat' ).strip( )
	format_value = str( response_format or 'Markdown' ).strip( )
	language_value = str( language or 'English' ).strip( )
	lines: List[ str ] = [ ]
	lines.append( f'You are Loca, a local AI assistant operating in the category "{category_value}".' )
	lines.append( f'Primary task type: {task_value}.' )
	lines.append( f'Response format: {format_value}.' )
	lines.append( f'Preferred language: {language_value}.' )

	if category_value == 'Reasoning':
		lines.append(
			'Provide careful, structured analytical answers grounded in the supplied information.' )
	elif category_value == 'Coding':
		lines.append(
			'Produce editor-ready code and explain only what is necessary for correct implementation.' )
	elif category_value == 'Translation':
		lines.append( 'Translate faithfully while preserving meaning, tone, and structure.' )
	elif category_value == 'Summarization':
		lines.append( 'Summarize faithfully and preserve key facts, names, and dates.' )
	elif category_value == 'Extraction':
		lines.append( 'Extract only supported facts. Do not invent missing values.' )
	elif category_value == 'Document Extraction':
		lines.append(
			'Use the document content as the evidence base and extract structured facts faithfully.' )
	elif category_value == 'OCR':
		lines.append( 'Extract visible text accurately and preserve structural cues where possible.' )
	elif category_value == 'JSON Output':
		lines.append( 'Return valid JSON only, matching the requested structure exactly.' )
	else:
		lines.append( 'Respond helpfully, accurately, and concisely.' )

	lines.append( 'If information is missing, state that clearly.' )
	return '\n'.join( lines ).strip( )

generate_prompt_template_draft

generate_prompt_template_draft(
    goal: str,
    constraints: str,
    style: str,
    category: str,
    task_type: str,
    response_format: str,
    language: str,
) -> str

Purpose:

Generate a draft system prompt using the local model.

Parameters:

goal : str constraints : str style : str category : str task_type : str response_format : str language : str

Returns:

str

Source code in app.py

def generate_prompt_template_draft( goal: str, constraints: str, style: str,
		category: str, task_type: str, response_format: str, language: str ) -> str:
	"""
		Purpose:
		--------
		Generate a draft system prompt using the local model.

		Parameters:
		-----------
		goal : str
		constraints : str
		style : str
		category : str
		task_type : str
		response_format : str
		language : str

		Returns:
		--------
		str
	"""
	prompt = f"""
	Create a strong system prompt for the Loca local AI application.

	Category: {category}
	Task Type: {task_type}
	Response Format: {response_format}
	Language: {language}
	Goal: {goal}
	Constraints: {constraints}
	Style: {style}

	Write only the system prompt text. Do not add explanation.
	""".strip( )

	return run_llm_turn( user_input=prompt,
		temperature=float( st.session_state.get( 'temperature', 0.2 ) ),
		top_p=float( st.session_state.get( 'top_percent', 0.95 ) ),
		repeat_penalty=float( st.session_state.get( 'repeat_penalty', 1.05 ) ),
		max_tokens=512, stream=False, output=None )

apply_prompt_to_text_generation

apply_prompt_to_text_generation(prompt_text: str) -> None

Purpose:

Apply a prompt to shared Text Generation settings.

Parameters:

prompt_text : str

Returns:

None

Source code in app.py

def apply_prompt_to_text_generation( prompt_text: str ) -> None:
	"""
		Purpose:
		--------
		Apply a prompt to shared Text Generation settings.

		Parameters:
		-----------
		prompt_text : str

		Returns:
		--------
		None
	"""
	st.session_state[ 'system_instructions' ] = str( prompt_text or '' )

apply_prompt_to_document_qna

apply_prompt_to_document_qna(prompt_text: str) -> None

Purpose:

Apply a prompt to shared Document Q&A settings.

Parameters:

prompt_text : str

Returns:

None

Source code in app.py

def apply_prompt_to_document_qna( prompt_text: str ) -> None:
	"""
		Purpose:
		--------
		Apply a prompt to shared Document Q&A settings.

		Parameters:
		-----------
		prompt_text : str

		Returns:
		--------
		None
	"""
	st.session_state[ 'system_instructions' ] = str( prompt_text or '' )
	st.session_state[ 'require_grounding' ] = True
	st.session_state[ 'answer_from_excerpts_only' ] = True

apply_prompt_metadata_to_shared_state

apply_prompt_metadata_to_shared_state(
    category: str,
    task_type: str,
    response_format: str,
    language: str,
) -> None

Purpose:

Apply prompt metadata to the shared app contract.

Parameters:

category : str task_type : str response_format : str language : str

Returns:

None

Source code in app.py

def apply_prompt_metadata_to_shared_state( category: str, task_type: str,
		response_format: str, language: str ) -> None:
	"""
		Purpose:
		--------
		Apply prompt metadata to the shared app contract.

		Parameters:
		-----------
		category : str
		task_type : str
		response_format : str
		language : str

		Returns:
		--------
		None
	"""
	st.session_state[ 'task_preset' ] = str( task_type or 'Chat' )
	st.session_state[ 'response_format' ] = str( response_format or 'Markdown' )
	st.session_state[ 'translation_target_language' ] = str( language or 'English' )

clone_prompt_record

clone_prompt_record(
    source_prompt: Dict[str, Any] | None,
) -> None

Purpose:

Clone a selected prompt into the edit surface as a new prompt draft.

Parameters:

source_prompt : Dict[str, Any] | None

Returns:

None

Source code in app.py

def clone_prompt_record( source_prompt: Dict[ str, Any ] | None ) -> None:
	"""
		Purpose:
		--------
		Clone a selected prompt into the edit surface as a new prompt draft.

		Parameters:
		-----------
		source_prompt : Dict[str, Any] | None

		Returns:
		--------
		None
	"""
	if not isinstance( source_prompt, dict ):
		return

	st.session_state.pe_selected_id = None
	st.session_state.pe_caption = f'{str( source_prompt.get( "Caption", "" ) )} Copy'.strip( )
	st.session_state.pe_name = str( source_prompt.get( 'Name', '' ) or '' )
	st.session_state.pe_text = str( source_prompt.get( 'Text', '' ) or '' )
	st.session_state.pe_version = str( source_prompt.get( 'Version', '' ) or '' )
	st.session_state.pe_id = source_prompt.get( 'ID', 0 )

initialize_database

initialize_database() -> None

Purpose:

Ensure required SQLite tables exist and that the Prompts table contains the columns required by the prompt utilities, Prompt Engineering mode, and AI-asset governance features.

Parameters:

None

Returns:

None

Source code in app.py

def initialize_database( ) -> None:
	"""
		Purpose:
		--------
		Ensure required SQLite tables exist and that the Prompts table contains the
		columns required by the prompt utilities, Prompt Engineering mode, and
		AI-asset governance features.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	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
            (
                PromptsId
                INTEGER
                NOT
                NULL
                PRIMARY
                KEY
                AUTOINCREMENT,
                Caption
                TEXT,
                Name
                TEXT
            (
                80
            ),
                Text TEXT,
                Version TEXT
            (
                80
            ),
                ID TEXT
            (
                80
            )
                )
			""" )

		conn.execute( """
            CREATE TABLE IF NOT EXISTS documents
            (
                DocumentId
                INTEGER
                PRIMARY
                KEY
                AUTOINCREMENT,
                Name
                TEXT
                NOT
                NULL,
                Type
                TEXT,
                SizeBytes
                INTEGER,
                Source
                TEXT,
                Fingerprint
                TEXT,
                TextLength
                INTEGER,
                ChunkCount
                INTEGER,
                CreatedOn
                TEXT
            )
			""" )

		conn.execute( """
            CREATE TABLE IF NOT EXISTS document_chunks
            (
                ChunkId
                INTEGER
                PRIMARY
                KEY
                AUTOINCREMENT,
                DocumentName
                TEXT
                NOT
                NULL,
                ChunkIndex
                INTEGER,
                ChunkText
                TEXT,
                ChunkLength
                INTEGER,
                Fingerprint
                TEXT,
                CreatedOn
                TEXT
            )
			""" )

		conn.execute( """
            CREATE TABLE IF NOT EXISTS document_embeddings
            (
                EmbeddingId
                INTEGER
                PRIMARY
                KEY
                AUTOINCREMENT,
                DocumentName
                TEXT
                NOT
                NULL,
                ChunkIndex
                INTEGER,
                VectorDim
                INTEGER,
                Fingerprint
                TEXT,
                CreatedOn
                TEXT
            )
			""" )

		conn.execute( """
            CREATE TABLE IF NOT EXISTS images
            (
                ImageId
                INTEGER
                PRIMARY
                KEY
                AUTOINCREMENT,
                Name
                TEXT
                NOT
                NULL,
                MimeType
                TEXT,
                SizeBytes
                INTEGER,
                Fingerprint
                TEXT,
                Source
                TEXT,
                CreatedOn
                TEXT
            )
			""" )

		prompt_columns = [ row[ 1 ] for row in conn.execute( 'PRAGMA table_info("Prompts");' ).fetchall( ) ]

		if 'Caption' not in prompt_columns:
			conn.execute( 'ALTER TABLE "Prompts" ADD COLUMN "Caption" TEXT;' )

		conn.commit( )

drop_table

drop_table(table: str) -> None

Purpose:

Safely drop a table if it exists.

Parameters:

table : str Table name.

Source code in app.py

def drop_table( table: str ) -> None:
	"""
		Purpose:
		--------
		Safely drop a table if it exists.

		Parameters:
		-----------
		table : str
			Table name.
	"""
	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

Purpose:

Rename an existing SQLite table. Attempts native ALTER TABLE rename first; if it fails, falls back to a schema-safe rebuild using the original CREATE TABLE statement and preserves indexes.

Parameters:

old_name : str Existing table name.

str

New table name.

Returns:

None

Source code in app.py

def rename_table( old_name: str, new_name: str ) -> None:
	"""
		Purpose:
		--------
		Rename an existing SQLite table. Attempts native ALTER TABLE rename first; if it fails,
		falls back to a schema-safe rebuild using the original CREATE TABLE statement and
		preserves indexes.

		Parameters:
		-----------
		old_name : str
			Existing table name.

		new_name : str
			New table name.

		Returns:
		--------
		None
	"""
	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

Purpose:

Rename a column within an existing SQLite table. Attempts native ALTER TABLE rename first; if it fails, falls back to a schema-safe rebuild preserving column order, data, and indexes.

Parameters:

table_name : str Table containing the column.

str

Existing column name.

str

New column name.

Returns:

None

Source code in app.py

def rename_column( table_name: str, old_name: str, new_name: str ) -> None:
	"""
		Purpose:
		--------
		Rename a column within an existing SQLite table. Attempts native ALTER TABLE rename
		first; if it fails, falls back to a schema-safe rebuild preserving column order, data,
		and indexes.

		Parameters:
		-----------
		table_name : str
			Table containing the column.

		old_name : str
			Existing column name.

		new_name : str
			New column name.

		Returns:
		--------
		None
	"""
	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

Purpose:

Create a safe SQLite index on a specified table column.

Handles

• Spaces in column names
• Special characters
• Reserved words
• Duplicate index names
• Validation against actual table schema

Parameters:

table : str Table name. column : str Column name to index.

Source code in app.py

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

		Handles:
			- Spaces in column names
			- Special characters
			- Reserved words
			- Duplicate index names
			- Validation against actual table schema

		Parameters:
		-----------
		table : str
			Table name.
		column : str
			Column name to index.
	"""
	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( )

get_sqlite_type

get_sqlite_type(dtype) -> str

Purpose:

Map a pandas dtype to an appropriate SQLite column type.

Parameters:

dtype : pandas dtype The dtype of a pandas Series.

Returns:

str SQLite column type.

Source code in app.py

def get_sqlite_type( dtype ) -> str:
	"""
		Purpose:
		--------
		Map a pandas dtype to an appropriate SQLite column type.

		Parameters:
		-----------
		dtype : pandas dtype
			The dtype of a pandas Series.

		Returns:
		--------
		str
			SQLite column type.
	"""
	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

Purpose:

Create a custom SQLite table from column definitions.

Parameters:

table_name : str Name of table.

list of dict

[ { "name": str, "type": str, "not_null": bool, "primary_key": bool, "auto_increment": bool } ]

Source code in app.py

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

		Parameters:
		-----------
		table_name : str
			Name of table.

		columns : list of dict
			[
				{
					"name": str,
					"type": str,
					"not_null": bool,
					"primary_key": bool,
					"auto_increment": bool
				}
			]
	"""
	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

Purpose:

Determine whether a SQL query is read-only and safe to execute.

Allows

SELECT WITH (CTE returning SELECT) EXPLAIN SELECT PRAGMA (read-only)

Blocks

INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, ATTACH, DETACH, VACUUM, REPLACE, TRIGGER, and multiple statements.

Source code in app.py

def is_safe_query( query: str ) -> bool:
	"""

		Purpose:
		--------
		Determine whether a SQL query is read-only and safe to execute.

		Allows:
			SELECT
			WITH (CTE returning SELECT)
			EXPLAIN SELECT
			PRAGMA (read-only)

		Blocks:
			INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, ATTACH,
			DETACH, VACUUM, REPLACE, TRIGGER, and multiple statements.

	"""
	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

Purpose:

Sanitize a string into a safe SQLite identifier.

• Replaces invalid characters with underscores
• Ensures it starts with a letter or underscore
• Prevents empty names

Source code in app.py

def create_identifier( name: str ) -> str:
	"""

		Purpose:
		--------
		Sanitize a string into a safe SQLite identifier.

		- Replaces invalid characters with underscores
		- Ensures it starts with a letter or underscore
		- Prevents empty names

	"""
	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_ai_asset_tables

get_ai_asset_tables() -> List[str]

Purpose:

Return the AI-asset governance table names.

Parameters:

None

Returns:

List[str] AI-asset table names.

Source code in app.py

def get_ai_asset_tables( ) -> List[ str ]:
	"""
		Purpose:
		--------
		Return the AI-asset governance table names.

		Parameters:
		-----------
		None

		Returns:
		--------
		List[str]
			AI-asset table names.
	"""
	return [
			'documents',
			'document_chunks',
			'document_embeddings',
			'images'
	]

get_table_row_count

get_table_row_count(
    conn: Connection, table_name: str
) -> int

Purpose:

Return the row count for a SQLite table.

Parameters:

conn : sqlite3.Connection Open SQLite connection.

str

Table name.

Returns:

int Number of rows in the table.

Source code in app.py

def get_table_row_count( conn: sqlite3.Connection, table_name: str ) -> int:
	"""
		Purpose:
		--------
		Return the row count for a SQLite table.

		Parameters:
		-----------
		conn : sqlite3.Connection
			Open SQLite connection.

		table_name : str
			Table name.

		Returns:
		--------
		int
			Number of rows in the table.
	"""
	if not table_name:
		return 0

	try:
		row = conn.execute(
			f'SELECT COUNT(*) FROM "{table_name}";'
		).fetchone( )

		return int( row[ 0 ] ) if row else 0
	except Exception:
		return 0

get_ai_asset_counts

get_ai_asset_counts() -> Dict[str, int]

Purpose:

Count rows in the AI asset governance tables used by Document Q&A, Semantic Search, and Data Management.

Parameters:

None

Returns:

Dict[str, int] Dictionary keyed by AI asset table name with row counts as values.

Source code in app.py

def get_ai_asset_counts( ) -> Dict[ str, int ]:
	"""
		Purpose:
		--------
		Count rows in the AI asset governance tables used by Document Q&A,
		Semantic Search, and Data Management.

		Parameters:
		-----------
		None

		Returns:
		--------
		Dict[str, int]
			Dictionary keyed by AI asset table name with row counts as values.
	"""
	counts: Dict[ str, int ] = { }
	tables = list_tables( )

	with create_connection( ) as conn:
		for table_name in get_ai_asset_tables( ):
			if table_name not in tables:
				counts[ table_name ] = 0
				continue

			counts[ table_name ] = get_table_row_count( conn, table_name )

	return counts

purge_orphaned_document_chunks

purge_orphaned_document_chunks(conn: Connection) -> int

Purpose:

Delete document chunk rows whose DocumentName no longer exists in the documents table.

Parameters:

conn : sqlite3.Connection Open SQLite connection.

Returns:

int Deleted row count.

Source code in app.py

def purge_orphaned_document_chunks( conn: sqlite3.Connection ) -> int:
	"""
		Purpose:
		--------
		Delete document chunk rows whose DocumentName no longer exists in the documents
		table.

		Parameters:
		-----------
		conn : sqlite3.Connection
			Open SQLite connection.

		Returns:
		--------
		int
			Deleted row count.
	"""
	try:
		cur = conn.execute(
			"""
            DELETE
            FROM document_chunks
            WHERE DocumentName NOT IN
                  (SELECT Name
                   FROM documents);
			"""
		)

		return int( cur.rowcount if cur.rowcount is not None else 0 )
	except Exception:
		return 0

purge_orphaned_document_embeddings

purge_orphaned_document_embeddings(conn: Connection) -> int

Purpose:

Delete document embedding metadata rows whose DocumentName no longer exists in the documents table.

Parameters:

conn : sqlite3.Connection Open SQLite connection.

Returns:

int Deleted row count.

Source code in app.py

def purge_orphaned_document_embeddings( conn: sqlite3.Connection ) -> int:
	"""
		Purpose:
		--------
		Delete document embedding metadata rows whose DocumentName no longer exists in
		the documents table.

		Parameters:
		-----------
		conn : sqlite3.Connection
			Open SQLite connection.

		Returns:
		--------
		int
			Deleted row count.
	"""
	try:
		cur = conn.execute(
			"""
            DELETE
            FROM document_embeddings
            WHERE DocumentName NOT IN
                  (SELECT Name
                   FROM documents);
			"""
		)

		return int( cur.rowcount if cur.rowcount is not None else 0 )
	except Exception:
		return 0

purge_orphaned_ai_assets

purge_orphaned_ai_assets() -> Dict[str, int]

Purpose:

Delete orphaned AI asset rows that depend on the governed documents table.

Parameters:

None

Returns:

Dict[str, int] Dictionary containing deleted chunk and embedding row counts.

Source code in app.py

def purge_orphaned_ai_assets( ) -> Dict[ str, int ]:
	"""
		Purpose:
		--------
		Delete orphaned AI asset rows that depend on the governed documents table.

		Parameters:
		-----------
		None

		Returns:
		--------
		Dict[str, int]
			Dictionary containing deleted chunk and embedding row counts.
	"""
	result: Dict[ str, int ] = {
			'deleted_chunks': 0,
			'deleted_embeddings': 0
	}

	tables = list_tables( )

	if 'documents' not in tables:
		return result

	with create_connection( ) as conn:
		if 'document_chunks' in tables:
			result[ 'deleted_chunks' ] = purge_orphaned_document_chunks( conn )

		if 'document_embeddings' in tables:
			result[ 'deleted_embeddings' ] = purge_orphaned_document_embeddings( conn )

		conn.commit( )

	return result

get_timestamp_text

get_timestamp_text() -> str

Purpose:

Return a UTC-like timestamp string for metadata rows.

Parameters:

None

Returns:

str

Source code in app.py

def get_timestamp_text( ) -> str:
	"""
		Purpose:
		--------
		Return a UTC-like timestamp string for metadata rows.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
	"""
	return time.strftime( '%Y-%m-%d %H:%M:%S' )

register_session_documents

register_session_documents() -> Dict[str, int]

Purpose:

Register active uploaded documents into the governed documents table.

Parameters:

None

Returns:

Dict[str, int]

Source code in app.py

def register_session_documents( ) -> Dict[ str, int ]:
	"""
		Purpose:
		--------
		Register active uploaded documents into the governed documents table.

		Parameters:
		-----------
		None

		Returns:
		--------
		Dict[str, int]
	"""
	active_docs = st.session_state.get( 'active_docs', [ ] )
	doc_bytes = st.session_state.get( 'doc_bytes', { } )

	inserted = 0
	updated = 0

	with create_connection( ) as conn:
		for name in active_docs:
			file_bytes = doc_bytes.get( name, b'' )
			if not file_bytes:
				continue

			text = extract_text( file_bytes, name )
			chunks = chunk_text( text ) if text else [ ]
			fingerprint = hashlib.sha256( file_bytes ).hexdigest( )
			file_type = Path( name ).suffix.lower( ).replace( '.', '' )
			created_on = get_timestamp_text( )

			existing = conn.execute(
				'''
                SELECT DocumentId
                FROM documents
                WHERE Name = ?
                  AND Fingerprint = ?
				''',
				(name, fingerprint)
			).fetchone( )

			if existing:
				conn.execute(
					'''
                    UPDATE documents
                    SET Type       = ?,
                        SizeBytes  = ?,
                        Source     = ?,
                        TextLength = ?,
                        ChunkCount = ?,
                        CreatedOn  = ?
                    WHERE DocumentId = ?
					''',
					(
							file_type,
							len( file_bytes ),
							'uploadlocal',
							len( text ),
							len( chunks ),
							created_on,
							existing[ 0 ]
					)
				)
				updated += 1
			else:
				conn.execute(
					'''
                    INSERT INTO documents
                    (Name,
                     Type,
                     SizeBytes,
                     Source,
                     Fingerprint,
                     TextLength,
                     ChunkCount,
                     CreatedOn)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
					''',
					(
							name,
							file_type,
							len( file_bytes ),
							'uploadlocal',
							fingerprint,
							len( text ),
							len( chunks ),
							created_on
					)
				)
				inserted += 1

		conn.commit( )

	return { 'inserted': inserted, 'updated': updated }

register_session_chunks

register_session_chunks() -> Dict[str, int]

Purpose:

Register active document chunks into the governed document_chunks table.

Parameters:

None

Returns:

Dict[str, int]

Source code in app.py

def register_session_chunks( ) -> Dict[ str, int ]:
	"""
		Purpose:
		--------
		Register active document chunks into the governed document_chunks table.

		Parameters:
		-----------
		None

		Returns:
		--------
		Dict[str, int]
	"""
	active_docs = st.session_state.get( 'active_docs', [ ] )
	doc_bytes = st.session_state.get( 'doc_bytes', { } )
	inserted = 0

	with create_connection( ) as conn:
		for name in active_docs:
			file_bytes = doc_bytes.get( name, b'' )
			if not file_bytes:
				continue

			text = extract_text( file_bytes, name )
			chunks = chunk_text( text ) if text else [ ]
			file_fingerprint = hashlib.sha256( file_bytes ).hexdigest( )
			created_on = get_timestamp_text( )

			conn.execute(
				'DELETE FROM document_chunks WHERE DocumentName = ? AND Fingerprint = ?',
				(name, file_fingerprint)
			)

			for idx, chunk_value in enumerate( chunks ):
				conn.execute(
					'''
                    INSERT INTO document_chunks
                    (DocumentName,
                     ChunkIndex,
                     ChunkText,
                     ChunkLength,
                     Fingerprint,
                     CreatedOn)
                    VALUES (?, ?, ?, ?, ?, ?)
					''',
					(
							name,
							idx,
							chunk_value,
							len( chunk_value ),
							file_fingerprint,
							created_on
					)
				)
				inserted += 1

		conn.commit( )

	return { 'inserted': inserted }

register_session_embeddings

register_session_embeddings() -> Dict[str, int]

Purpose:

Register active document embedding metadata into the governed document_embeddings table.

Parameters:

None

Returns:

Dict[str, int]

Source code in app.py

def register_session_embeddings( ) -> Dict[ str, int ]:
	"""
		Purpose:
		--------
		Register active document embedding metadata into the governed
		document_embeddings table.

		Parameters:
		-----------
		None

		Returns:
		--------
		Dict[str, int]
	"""
	active_docs = st.session_state.get( 'active_docs', [ ] )
	doc_bytes = st.session_state.get( 'doc_bytes', { } )
	inserted = 0

	if embedder is None:
		return { 'inserted': 0 }

	vector_dim = getattr( embedder, 'get_sentence_embedding_dimension', lambda: 384 )( )
	vector_dim = int( vector_dim ) if vector_dim else 384
	with create_connection( ) as conn:
		for name in active_docs:
			file_bytes = doc_bytes.get( name, b'' )
			if not file_bytes:
				continue

			text = extract_text( file_bytes, name )
			chunks = chunk_text( text ) if text else [ ]
			file_fingerprint = hashlib.sha256( file_bytes ).hexdigest( )
			created_on = get_timestamp_text( )

			conn.execute(
				'DELETE FROM document_embeddings WHERE DocumentName = ? AND Fingerprint = ?',
				(name, file_fingerprint) )

			for idx, _chunk_value in enumerate( chunks ):
				conn.execute( '''
                    INSERT INTO document_embeddings
                    (DocumentName,
                     ChunkIndex,
                     VectorDim,
                     Fingerprint,
                     CreatedOn)
                    VALUES (?, ?, ?, ?, ?)
					''', (name, idx, vector_dim, file_fingerprint, created_on) )
				inserted += 1

		conn.commit( )

	return { 'inserted': inserted }

register_upload_images

register_upload_images(
    uploaded_files: List[Any],
) -> Dict[str, int]

Purpose:

Register uploaded image metadata into the governed images table.

Parameters:

uploaded_files : List[Any]

Returns:

Dict[str, int]

Source code in app.py

def register_upload_images( uploaded_files: List[ Any ] ) -> Dict[ str, int ]:
	"""
		Purpose:
		--------
		Register uploaded image metadata into the governed images table.

		Parameters:
		-----------
		uploaded_files : List[Any]

		Returns:
		--------
		Dict[str, int]
	"""
	inserted = 0
	updated = 0

	with create_connection( ) as conn:
		for f in uploaded_files:
			try:
				name = str( getattr( f, 'name', '' ) or '' ).strip( )
				file_bytes = f.getvalue( )
				mime_type = str( getattr( f, 'type', '' ) or '' ).strip( )
			except Exception:
				continue

			if not name or not file_bytes:
				continue

			fingerprint = hashlib.sha256( file_bytes ).hexdigest( )
			created_on = get_timestamp_text( )

			existing = conn.execute(
				'''
                SELECT ImageId
                FROM images
                WHERE Name = ?
                  AND Fingerprint = ?
				''',
				(name, fingerprint)
			).fetchone( )

			if existing:
				conn.execute(
					'''
                    UPDATE images
                    SET MimeType  = ?,
                        SizeBytes = ?,
                        Source    = ?,
                        CreatedOn = ?
                    WHERE ImageId = ?
					''',
					(
							mime_type,
							len( file_bytes ),
							'uploadlocal',
							created_on,
							existing[ 0 ]
					)
				)
				updated += 1
			else:
				conn.execute(
					'''
                    INSERT INTO images
                    (Name,
                     MimeType,
                     SizeBytes,
                     Fingerprint,
                     Source,
                     CreatedOn)
                    VALUES (?, ?, ?, ?, ?, ?)
					''',
					(
							name,
							mime_type,
							len( file_bytes ),
							fingerprint,
							'uploadlocal',
							created_on
					)
				)
				inserted += 1

		conn.commit( )

	return { 'inserted': inserted, 'updated': updated }

load_llm

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

Purpose:

Lazily load the selected local llama.cpp GGUF model using the supplied runtime settings. The model path is part of the cache key so switching models creates a distinct cached model resource.

Parameters:

model_path : str Local GGUF model path.

int

Context window size.

int

CPU thread count.

int

Random seed used by llama.cpp.

Returns:

Any | None Loaded llama.cpp model instance when available; otherwise None.

Source code in app.py

@st.cache_resource
def load_llm( model_path: str, ctx: int, threads: int, seed: int ) -> Any | None:
	"""
		Purpose:
		--------
		Lazily load the selected local llama.cpp GGUF model using the supplied runtime
		settings. The model path is part of the cache key so switching models creates a
		distinct cached model resource.

		Parameters:
		-----------
		model_path : str
			Local GGUF model path.

		ctx : int
			Context window size.

		threads : int
			CPU thread count.

		seed : int
			Random seed used by llama.cpp.

		Returns:
		--------
		Any | None
			Loaded llama.cpp model instance when available; otherwise None.
	"""
	try:
		if Llama is None:
			return None

		model_path_value = str( model_path or '' ).strip( )

		if not model_path_value:
			return None

		if not Path( model_path_value ).exists( ):
			return None

		ctx_value = int( ctx ) if int( ctx ) > 0 else int( cfg.DEFAULT_CTX )
		thread_value = int( threads ) if int( threads ) > 0 else int( cfg.CORES )
		seed_value = int( seed ) if seed is not None else -1

		return Llama(
			model_path=model_path_value,
			n_ctx=ctx_value,
			n_threads=thread_value,
			n_batch=512,
			seed=seed_value,
			verbose=False
		)

	except Exception:
		return None

load_embedder

load_embedder() -> Any | None

Purpose:

Lazily load the sentence embedding model when the dependency is available.

Parameters:

None

Returns:

Any | None A sentence-transformer model instance when available; otherwise None.

Source code in app.py

@st.cache_resource
def load_embedder( ) -> Any | None:
	"""
		Purpose:
		--------
		Lazily load the sentence embedding model when the dependency is available.

		Parameters:
		-----------
		None

		Returns:
		--------
		Any | None
			A sentence-transformer model instance when available; otherwise None.
	"""
	try:
		from sentence_transformers import SentenceTransformer

		return SentenceTransformer( 'all-MiniLM-L6-v2' )
	except Exception:
		return None

create_docqna_instruction

create_docqna_instruction(action_name: str) -> str

Purpose:

Return an instruction block for a selected document action.

Parameters:

action_name : str

Returns:

str

Source code in app.py

def create_docqna_instruction( action_name: str ) -> str:
	"""
		Purpose:
		--------
		Return an instruction block for a selected document action.

		Parameters:
		-----------
		action_name : str

		Returns:
		--------
		str
	"""
	action = str( action_name or 'Answer Question' ).strip( )
	action_map = {
			'Answer Question':
				'Answer the user question directly using the retrieved excerpts.',
			'Summarize Active Document':
				'Provide a clear, structured summary of the active document.',
			'Extract Key Points':
				'Extract the most important points as a concise bullet list.',
			'Generate Outline':
				'Generate a structured outline of the document.',
			'Extract Entities':
				'Extract named entities, important organizations, dates, and references.',
			'Extract Tables':
				'Describe tabular information or structured fields present in the excerpts.',
			'Compare Active Documents':
				'Compare the active documents, noting agreements, differences, and gaps.'
	}

	return action_map.get( action, action_map[ 'Answer Question' ] )

build_instruction_block

build_instruction_block() -> str

Purpose:

Build a unified instruction block for document-grounded answering.

Parameters:

None

Returns:

str

Source code in app.py

def build_instruction_block( ) -> str:
	"""
		Purpose:
		--------
		Build a unified instruction block for document-grounded answering.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
	"""
	system_instructions = get_effective_system_instructions( )
	require_grounding = bool( st.session_state.get( 'require_grounding', True ) )
	answer_from_excerpts_only = bool( st.session_state.get( 'answer_from_excerpts_only', True ) )
	response_format = str( st.session_state.get( 'response_format', 'Markdown' ) or 'Markdown' ).strip( )
	doc_action = str( st.session_state.get( 'docqna_action', 'Answer Question' ) or 'Answer Question' )
	lines: List[ str ] = [ ]
	if system_instructions:
		lines.append( system_instructions )

	lines.append( 'Document Q&A Instructions:' )
	lines.append( f'- Action: {doc_action}' )
	lines.append( f'- Response Format: {response_format}' )
	lines.append( f'- Action Guidance: {create_docqna_instruction( doc_action )}' )
	if require_grounding:
		lines.append( '- Ground every answer in the retrieved document excerpts.' )

	if answer_from_excerpts_only:
		lines.append(
			'- If the retrieved excerpts do not contain the answer, '
			'state clearly that there is not enough information.' )

	if response_format == 'JSON':
		lines.append( '- Return valid JSON only.' )

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

extract_text_from_pdf_bytes

extract_text_from_pdf_bytes(
    file_bytes: bytes, include_page_markers: bool = False
) -> str

Purpose:

Extract native text from PDF bytes using PyMuPDF when it is available.

Parameters:

file_bytes : bytes PDF file bytes.

bool

When True, include page markers before each extracted page.

Returns:

str Extracted PDF text.

Source code in app.py

def extract_text_from_pdf_bytes( file_bytes: bytes, include_page_markers: bool = False ) -> str:
	"""
		Purpose:
		--------
		Extract native text from PDF bytes using PyMuPDF when it is available.

		Parameters:
		-----------
		file_bytes : bytes
			PDF file bytes.

		include_page_markers : bool
			When True, include page markers before each extracted page.

		Returns:
		--------
		str
			Extracted PDF text.
	"""
	if not file_bytes:
		return ''

	if fitz is None:
		return ''

	try:
		doc = fitz.open( stream=file_bytes, filetype='pdf' )
		parts: List[ str ] = [ ]

		for page_index, page in enumerate( doc, start=1 ):
			page_text = page.get_text( 'text' ) or ''

			if include_page_markers:
				parts.append( f'[Page {page_index}]' )

			if page_text:
				parts.append( page_text )

		doc.close( )
		return '\n'.join( parts ).strip( )
	except Exception:
		return ''

extract_text_from_docx_bytes

extract_text_from_docx_bytes(file_bytes: bytes) -> str

Purpose:

Extract text from DOCX bytes using python-docx when it is available.

Parameters:

file_bytes : bytes DOCX file bytes.

Returns:

str Extracted DOCX text.

Source code in app.py

def extract_text_from_docx_bytes( file_bytes: bytes ) -> str:
	"""
		Purpose:
		--------
		Extract text from DOCX bytes using python-docx when it is available.

		Parameters:
		-----------
		file_bytes : bytes
			DOCX file bytes.

		Returns:
		--------
		str
			Extracted DOCX text.
	"""
	if not file_bytes:
		return ''

	if Document is None:
		return ''

	try:
		buffer = BytesIO( file_bytes )
		document = Document( buffer )
		parts: List[ str ] = [ ]

		for paragraph in document.paragraphs:
			text = str( paragraph.text or '' ).strip( )
			if text:
				parts.append( text )

		for table in document.tables:
			for row in table.rows:
				values: List[ str ] = [ ]
				for cell in row.cells:
					cell_text = str( cell.text or '' ).strip( )
					values.append( cell_text )

				row_text = ' | '.join( values ).strip( )
				if row_text:
					parts.append( row_text )

		return '\n'.join( parts ).strip( )
	except Exception:
		return ''

compute_fingerprint

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

Purpose:

Computes a stable fingerprint for the currently selected active documents and their byte contents.

Parameters:

active_docs: A List[ str ] of active document names. doc_bytes: A Dict[ str, bytes ] mapping document name to file bytes.

Returns:

A str fingerprint suitable for cache invalidation.

Source code in app.py

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

		Purpose:
		--------
		Computes a stable fingerprint for the currently selected active documents and their byte contents.

		Parameters:
		-----------
		active_docs:
			A List[ str ] of active document names.
		doc_bytes:
			A Dict[ str, bytes ] mapping document name to file bytes.

		Returns:
		--------
		A str fingerprint suitable for cache invalidation.

	'''
	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( )

decode_text_bytes

decode_text_bytes(file_bytes: bytes) -> str

Purpose:

Decode text-like document bytes using common encodings and a permissive fallback.

Parameters:

file_bytes : bytes File bytes to decode.

Returns:

str Decoded text.

Source code in app.py

def decode_text_bytes( file_bytes: bytes ) -> str:
	"""
		Purpose:
		--------
		Decode text-like document bytes using common encodings and a permissive fallback.

		Parameters:
		-----------
		file_bytes : bytes
			File bytes to decode.

		Returns:
		--------
		str
			Decoded text.
	"""
	if not file_bytes:
		return ''

	encodings = [ 'utf-8', 'utf-8-sig', 'cp1252', 'latin-1' ]

	for encoding in encodings:
		try:
			return file_bytes.decode( encoding ).strip( )
		except Exception:
			continue

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

extract_text_from_bytes

extract_text_from_bytes(
    file_bytes: bytes, file_name: str = ""
) -> str

Purpose:

Extract text from supported document bytes using the file name extension and current parsing preferences.

Parameters:

file_bytes : bytes Source document bytes.

str

Source document name.

Returns:

str Extracted text.

Source code in app.py

def extract_text_from_bytes( file_bytes: bytes, file_name: str = '' ) -> str:
	"""
		Purpose:
		--------
		Extract text from supported document bytes using the file name extension and
		current parsing preferences.

		Parameters:
		-----------
		file_bytes : bytes
			Source document bytes.

		file_name : str
			Source document name.

		Returns:
		--------
		str
			Extracted text.
	"""
	if not file_bytes:
		return ''

	file_name_value = str( file_name or '' ).lower( ).strip( )
	include_page_markers = bool( st.session_state.get( 'include_page_markers', False ) )
	prefer_native_pdf_text = bool( st.session_state.get( 'prefer_native_pdf_text', True ) )

	if file_name_value.endswith( '.pdf' ):
		if prefer_native_pdf_text:
			text = extract_text_from_pdf_bytes(
				file_bytes=file_bytes,
				include_page_markers=include_page_markers
			)

			if text:
				return text

		return decode_text_bytes( file_bytes )

	if file_name_value.endswith( '.docx' ):
		text = extract_text_from_docx_bytes( file_bytes )

		if text:
			return text

		return decode_text_bytes( file_bytes )

	if (
			file_name_value.endswith( '.txt' )
			or file_name_value.endswith( '.md' )
			or file_name_value.endswith( '.csv' )
			or file_name_value.endswith( '.json' )
			or file_name_value.endswith( '.xml' )
			or file_name_value.endswith( '.html' )
			or file_name_value.endswith( '.htm' )
	):
		return decode_text_bytes( file_bytes )

	if not file_name_value:
		pdf_text = extract_text_from_pdf_bytes(
			file_bytes=file_bytes,
			include_page_markers=include_page_markers
		)

		if pdf_text:
			return pdf_text

	return decode_text_bytes( file_bytes )

extract_text_bytes

extract_text_bytes(
    file_bytes: bytes, file_name: str = ""
) -> str

Purpose:

Backward-compatible wrapper for extracting text from document bytes.

Parameters:

file_bytes : bytes Source document bytes.

str

Source document name.

Returns:

str Extracted text.

Source code in app.py

def extract_text_bytes( file_bytes: bytes, file_name: str = '' ) -> str:
	"""
		Purpose:
		--------
		Backward-compatible wrapper for extracting text from document bytes.

		Parameters:
		-----------
		file_bytes : bytes
			Source document bytes.

		file_name : str
			Source document name.

		Returns:
		--------
		str
			Extracted text.
	"""
	return extract_text_from_bytes( file_bytes=file_bytes, file_name=file_name )

extract_text

extract_text(file_bytes: bytes, file_name: str = '') -> str

Purpose:

Extract document text using the configured parsing behavior.

Parameters:

file_bytes : bytes Source document bytes.

str

Source document name.

Returns:

str Extracted text.

Source code in app.py

def extract_text( file_bytes: bytes, file_name: str = '' ) -> str:
	"""
		Purpose:
		--------
		Extract document text using the configured parsing behavior.

		Parameters:
		-----------
		file_bytes : bytes
			Source document bytes.

		file_name : str
			Source document name.

		Returns:
		--------
		str
			Extracted text.
	"""
	return extract_text_from_bytes( file_bytes=file_bytes, file_name=file_name )

load_sqlite_vec

load_sqlite_vec(conn: Connection) -> bool

Purpose:

Attempts to load sqlite-vec into the provided SQLite connection.

Parameters:

conn: The sqlite3.Connection.

Returns:

True if sqlite-vec loaded successfully; otherwise False.

Source code in app.py

def load_sqlite_vec( conn: sqlite3.Connection ) -> bool:
	'''

		Purpose:
		--------
		Attempts to load sqlite-vec into the provided SQLite connection.

		Parameters:
		-----------
		conn:
			The sqlite3.Connection.

		Returns:
		--------
		True if sqlite-vec loaded successfully; otherwise False.

	'''
	try:
		import sqlite_vec

		sqlite_vec.load( conn )
		return True
	except Exception:
		return False

ensure_schema

ensure_schema(dim: int) -> bool

Purpose:

Creates the sqlite-vec virtual table used for Document Q&A embeddings if possible.

Parameters:

dim: The embedding dimension (e.g., 384 for all-MiniLM-L6-v2).

Returns:

True if the schema exists and is usable; otherwise False.

Source code in app.py

def ensure_schema( dim: int ) -> bool:
	'''

		Purpose:
		--------
		Creates the sqlite-vec virtual table used for Document Q&A embeddings if possible.

		Parameters:
		-----------
		dim:
			The embedding dimension (e.g., 384 for all-MiniLM-L6-v2).

		Returns:
		--------
		True if the schema exists and is usable; otherwise False.

	'''
	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:
		return False
	finally:
		conn.close( )

build_docqna_inventory

build_docqna_inventory() -> List[Dict[str, Any]]

Purpose:

Build inventory rows for the currently active uploaded documents.

Parameters:

None

Returns:

List[Dict[str, Any]]

Source code in app.py

def build_docqna_inventory( ) -> List[ Dict[ str, Any ] ]:
	"""
		Purpose:
		--------
		Build inventory rows for the currently active uploaded documents.

		Parameters:
		-----------
		None

		Returns:
		--------
		List[Dict[str, Any]]
	"""
	rows: List[ Dict[ str, Any ] ] = [ ]
	active_docs = st.session_state.get( 'active_docs', [ ] )
	doc_bytes = st.session_state.get( 'doc_bytes', { } )
	for name in active_docs:
		b = doc_bytes.get( name, b'' )
		text = extract_text( b, name ) if b else ''
		chunks = chunk_text( text ) if text else [ ]
		rows.append( {
					'Name': name,
					'SizeBytes': len( b ) if b else 0,
					'TextLength': len( text ) if text else 0,
					'ChunkCount': len( chunks ),
					'Loaded': bool( b )
			} )

	return rows

get_docqna_names

get_docqna_names() -> str

Purpose:

Build a human-readable string of active document names.

Parameters:

None

Returns:

str

Source code in app.py

def get_docqna_names( ) -> str:
	"""
		Purpose:
		--------
		Build a human-readable string of active document names.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
	"""
	active_docs = st.session_state.get( 'active_docs', [ ] )
	if not isinstance( active_docs, list ) or len( active_docs ) == 0:
		return 'No active documents'
	return ', '.join( [ str( name ) for name in active_docs ] )

is_embedder_available

is_embedder_available(candidate: Any | None = None) -> bool

Purpose:

Determine whether a sentence embedding model is available and usable.

Parameters:

candidate : Any | None Optional embedding model instance. When omitted, the global embedder is used.

Returns:

bool True when an embedder with an encode method is available; otherwise False.

Source code in app.py

def is_embedder_available( candidate: Any | None = None ) -> bool:
	"""
		Purpose:
		--------
		Determine whether a sentence embedding model is available and usable.

		Parameters:
		-----------
		candidate : Any | None
			Optional embedding model instance. When omitted, the global embedder is used.

		Returns:
		--------
		bool
			True when an embedder with an encode method is available; otherwise False.
	"""
	model = candidate if candidate is not None else globals( ).get( 'embedder', None )
	return model is not None and hasattr( model, 'encode' )

get_embedder_unavailable_message

get_embedder_unavailable_message() -> str

Purpose:

Return a standard diagnostic message when sentence-transformer embeddings are unavailable.

Parameters:

None

Returns:

str Diagnostic message.

Source code in app.py

def get_embedder_unavailable_message( ) -> str:
	"""
		Purpose:
		--------
		Return a standard diagnostic message when sentence-transformer embeddings are
		unavailable.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Diagnostic message.
	"""
	return (
			'Embedding retrieval is unavailable because the sentence-transformer model '
			'could not be loaded. Install or repair `sentence-transformers`, then restart '
			'the Streamlit app.'
	)

decode_embedding_vector

decode_embedding_vector(
    vector_blob: bytes | memoryview | None,
) -> np.ndarray

Purpose:

Decode a stored embedding vector BLOB into a NumPy float32 array.

Parameters:

vector_blob : bytes | memoryview | None Stored vector BLOB.

Returns:

np.ndarray Decoded vector. Empty array when decoding fails.

Source code in app.py

def decode_embedding_vector( vector_blob: bytes | memoryview | None ) -> np.ndarray:
	"""
		Purpose:
		--------
		Decode a stored embedding vector BLOB into a NumPy float32 array.

		Parameters:
		-----------
		vector_blob : bytes | memoryview | None
			Stored vector BLOB.

		Returns:
		--------
		np.ndarray
			Decoded vector. Empty array when decoding fails.
	"""
	if not vector_blob:
		return np.asarray( [ ], dtype=np.float32 )

	try:
		return np.frombuffer( vector_blob, dtype=np.float32 )
	except Exception:
		return np.asarray( [ ], dtype=np.float32 )

rebuild_index

rebuild_index(embedder: Any | None) -> None

Purpose:

Build or refresh the Document Q&A vector index when active documents or chunk settings change. The function fails closed when embeddings are unavailable instead of raising an AttributeError from embedder.encode(...).

Parameters:

embedder : Any | None Sentence embedding model instance.

Returns:

None

Source code in app.py

def rebuild_index( embedder: Any | None ) -> None:
	"""
		Purpose:
		--------
		Build or refresh the Document Q&A vector index when active documents or chunk
		settings change. The function fails closed when embeddings are unavailable instead
		of raising an AttributeError from embedder.encode(...).

		Parameters:
		-----------
		embedder : Any | None
			Sentence embedding model instance.

		Returns:
		--------
		None
	"""
	if not is_embedder_available( embedder ):
		st.session_state[ 'docqna_vec_ready' ] = False
		st.session_state[ 'docqna_fallback_rows' ] = [ ]
		st.session_state[ 'docqna_chunk_count' ] = 0
		st.session_state[ 'docqna_last_retrieval' ] = [ ]
		st.session_state[ 'docqna_inventory_rows' ] = build_docqna_inventory( )
		st.session_state[ 'docqna_retrieval_status' ] = get_embedder_unavailable_message( )
		return

	active_docs: List[ str ] = st.session_state.get( 'active_docs', [ ] )
	doc_bytes: Dict[ str, bytes ] = st.session_state.get( 'doc_bytes', { } )
	retrieval_chunk_size = int( st.session_state.get( 'retrieval_chunk_size', 1200 ) )
	retrieval_chunk_overlap = int( st.session_state.get( 'retrieval_chunk_overlap', 200 ) )

	fp_seed = f'{retrieval_chunk_size}|{retrieval_chunk_overlap}|'
	fp_seed += compute_fingerprint( active_docs, doc_bytes )
	fp = hashlib.sha256( fp_seed.encode( 'utf-8', errors='ignore' ) ).hexdigest( )

	if fp and fp == st.session_state.get( 'docqna_fingerprint', '' ):
		st.session_state[ 'docqna_inventory_rows' ] = build_docqna_inventory( )
		return

	st.session_state[ 'docqna_fingerprint' ] = fp
	st.session_state[ 'docqna_chunk_count' ] = 0
	st.session_state[ 'docqna_fallback_rows' ] = [ ]
	st.session_state[ 'docqna_inventory_rows' ] = build_docqna_inventory( )
	st.session_state[ 'docqna_retrieval_status' ] = ''

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

	prefer_sqlite_vec = bool( st.session_state.get( 'prefer_sqlite_vec', True ) )
	vec_ready = False

	if prefer_sqlite_vec:
		vec_ready = ensure_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, name )
			if not text:
				continue

			chunks = chunk_text(
				text,
				size=retrieval_chunk_size,
				overlap=retrieval_chunk_overlap
			)

			if not chunks:
				continue

			try:
				vecs = embedder.encode( chunks, show_progress_bar=False )
				vecs = np.asarray( vecs, dtype=np.float32 )
			except Exception as e:
				st.session_state[ 'docqna_retrieval_status' ] = (
						f'Embedding generation failed for {name}: {e}'
				)
				continue

			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
		else:
			st.session_state[ 'docqna_fallback_rows' ] = [ ]

	except Exception as e:
		st.session_state[ 'docqna_vec_ready' ] = False
		st.session_state[ 'docqna_fallback_rows' ] = [ ]
		st.session_state[ 'docqna_chunk_count' ] = 0
		st.session_state[ 'docqna_retrieval_status' ] = f'Document index rebuild failed: {e}'

	finally:
		conn.close( )

retrieve_chunks

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

Purpose:

Retrieve top-k document chunks relevant to the query using sqlite-vec when available, with optional cosine-similarity fallback. Missing embeddings fail safely.

Parameters:

query : str User query.

int | None

Number of chunks to retrieve.

Returns:

List[Tuple[str, str, float]] Ranked retrieval results as document name, chunk text, and score or distance.

Source code in app.py

def retrieve_chunks( query: str, k: int = None ) -> List[ Tuple[ str, str, float ] ]:
	"""
		Purpose:
		--------
		Retrieve top-k document chunks relevant to the query using sqlite-vec when available,
		with optional cosine-similarity fallback. Missing embeddings fail safely.

		Parameters:
		-----------
		query : str
			User query.

		k : int | None
			Number of chunks to retrieve.

		Returns:
		--------
		List[Tuple[str, str, float]]
			Ranked retrieval results as document name, chunk text, and score or distance.
	"""
	if not query or not query.strip( ):
		return [ ]

	if not is_embedder_available( globals( ).get( 'embedder', None ) ):
		st.session_state[ 'docqna_last_retrieval' ] = [ ]
		st.session_state[ 'docqna_retrieval_status' ] = get_embedder_unavailable_message( )
		return [ ]

	rebuild_index( embedder )

	k_value = int( k ) if k is not None else int( st.session_state.get( 'retrieval_k', 6 ) )
	if k_value <= 0:
		k_value = 6

	try:
		qv = embedder.encode( [ query ], show_progress_bar=False )
		qv = np.asarray( qv, dtype=np.float32 )[ 0 ]
	except Exception as e:
		st.session_state[ 'docqna_last_retrieval' ] = [ ]
		st.session_state[ 'docqna_retrieval_status' ] = f'Query embedding failed: {e}'
		return [ ]

	if bool( 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_value ))
			)
			rows = cur.fetchall( )
			results = [ (r[ 0 ], r[ 1 ], float( r[ 2 ] )) for r in rows ]
			st.session_state[ 'docqna_last_retrieval' ] = results
			return results

		except Exception as e:
			st.session_state[ 'docqna_vec_ready' ] = False
			st.session_state[ 'docqna_retrieval_status' ] = (
					f'sqlite-vec retrieval failed; using fallback when enabled. Error: {e}'
			)

		finally:
			conn.close( )

	if not bool( st.session_state.get( 'allow_similarity_fallback', True ) ):
		st.session_state[ 'docqna_last_retrieval' ] = [ ]
		return [ ]

	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:
		v = decode_embedding_vector( vec_blob )

		if v.size == 0:
			continue

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

	results.sort( key=lambda r: r[ 2 ], reverse=True )
	results = results[ : int( k_value ) ]
	st.session_state[ 'docqna_last_retrieval' ] = results
	return results

build_docqna_input

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

Purpose:

Build a document-grounded prompt using retrieved excerpts and the current document action. Missing retrieval returns a safe prompt rather than failing.

Parameters:

user_query : str User request.

int | None

Number of chunks to retrieve.

Returns:

str Document-grounded LLM prompt.

Source code in app.py

def build_docqna_input( user_query: str, k: int = None ) -> str:
	"""
		Purpose:
		--------
		Build a document-grounded prompt using retrieved excerpts and the current document
		action. Missing retrieval returns a safe prompt rather than failing.

		Parameters:
		-----------
		user_query : str
			User request.

		k : int | None
			Number of chunks to retrieve.

		Returns:
		--------
		str
			Document-grounded LLM prompt.
	"""
	doc_instruction_block = build_instruction_block( )
	hits = retrieve_chunks( user_query, k=k )
	st.session_state[ 'docqna_last_retrieval' ] = hits

	context_blocks: List[ str ] = [ ]

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

	semantic_context_buffer = st.session_state.get( 'semantic_context_buffer', [ ] )
	if isinstance( semantic_context_buffer, list ):
		for value in semantic_context_buffer:
			if isinstance( value, str ) and value.strip( ):
				context_blocks.append( f'[Semantic Context]\n{value.strip( )}' )

	context = '\n\n'.join( context_blocks ).strip( )
	active_doc_names = get_docqna_names( )
	retrieval_status = str( st.session_state.get( 'docqna_retrieval_status', '' ) or '' ).strip( )

	prompt_parts: List[ str ] = [ ]

	if doc_instruction_block:
		prompt_parts.append( doc_instruction_block )

	prompt_parts.append( f'Active Documents:\n{active_doc_names}' )

	if context:
		prompt_parts.append(
			'Use the following retrieved document excerpts as the evidence base for your answer.\n\n'
			f'{context}'
		)
	else:
		if retrieval_status:
			prompt_parts.append(
				'No retrieved document excerpts were available.\n\n'
				f'Retrieval Status: {retrieval_status}'
			)
		else:
			prompt_parts.append(
				'No retrieved document excerpts were available for this question.'
			)

	prompt_parts.append( f'User Request:\n{user_query}\n\nAnswer:' )
	return '\n\n'.join( prompt_parts ).strip( )

decode_embedding_rows

decode_embedding_rows() -> List[Tuple[str, np.ndarray]]

Purpose:

Read and decode rows from the semantic embeddings table. Database failures, missing tables, corrupt blobs, and empty vectors fail closed so Semantic Search and Text Generation context reuse cannot crash.

Parameters:

None

Returns:

List[Tuple[str, np.ndarray]] Decoded chunk/vector rows.

Source code in app.py

def decode_embedding_rows( ) -> List[ Tuple[ str, np.ndarray ] ]:
	"""
		Purpose:
		--------
		Read and decode rows from the semantic embeddings table. Database failures,
		missing tables, corrupt blobs, and empty vectors fail closed so Semantic Search
		and Text Generation context reuse cannot crash.

		Parameters:
		-----------
		None

		Returns:
		--------
		List[Tuple[str, np.ndarray]]
			Decoded chunk/vector rows.
	"""
	rows_out: List[ Tuple[ str, np.ndarray ] ] = [ ]

	try:
		initialize_database( )

		with sqlite3.connect( cfg.DB_PATH, timeout=30 ) as conn:
			rows = conn.execute( 'SELECT chunk, vector FROM embeddings' ).fetchall( )
	except Exception as e:
		st.session_state[ 'semantic_status' ] = (
				f'Semantic index could not be read: {e}')
		return rows_out

	for chunk_text_value, vector_blob in rows:
		try:
			chunk = str( chunk_text_value or '' )
			vec = decode_embedding_vector( vector_blob )

			if not chunk or vec.size == 0:
				continue

			rows_out.append( (chunk, vec) )
		except Exception:
			continue

	return rows_out

clear_semantic_index

clear_semantic_index() -> None

Purpose:

Clear the semantic embeddings table and reset Semantic Search diagnostics without raising hard database errors into the Streamlit UI execution path.

Parameters:

None

Returns:

None

Source code in app.py

def clear_semantic_index( ) -> None:
	"""
		Purpose:
		--------
		Clear the semantic embeddings table and reset Semantic Search diagnostics without
		raising hard database errors into the Streamlit UI execution path.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	try:
		initialize_database( )

		with sqlite3.connect( cfg.DB_PATH, timeout=30 ) as conn:
			conn.execute( 'DELETE FROM embeddings' )
			conn.commit( )

		st.session_state[ 'semantic_result_rows' ] = [ ]
		st.session_state[ 'semantic_selected_rows' ] = [ ]
		st.session_state[ 'semantic_index_chunk_count' ] = 0
		st.session_state[ 'semantic_index_dim' ] = 0
		st.session_state[ 'semantic_index_doc_count' ] = 0
		st.session_state[ 'semantic_uploaded_names' ] = [ ]
		st.session_state[ 'semantic_last_query' ] = ''
		st.session_state[ 'semantic_status' ] = 'Semantic index deleted.'
	except Exception as e:
		st.session_state[ 'semantic_status' ] = (
				f'Semantic index could not be deleted: {e}')

build_semantic_index

build_semantic_index(
    uploaded_files: List[Any],
) -> Dict[str, Any]

Purpose:

Build or append a semantic chunk index from uploaded files. The function preserves the existing embeddings table contract while guarding extraction, embedding, vector shape, and SQLite write failures.

Parameters:

uploaded_files : List[Any] Uploaded files from Streamlit.

Returns:

Dict[str, Any] Index build result.

Source code in app.py

def build_semantic_index( uploaded_files: List[ Any ] ) -> Dict[ str, Any ]:
	"""
		Purpose:
		--------
		Build or append a semantic chunk index from uploaded files. The function preserves
		the existing embeddings table contract while guarding extraction, embedding, vector
		shape, and SQLite write failures.

		Parameters:
		-----------
		uploaded_files : List[Any]
			Uploaded files from Streamlit.

		Returns:
		--------
		Dict[str, Any]
			Index build result.
	"""
	global embedder

	if not is_embedder_available( globals( ).get( 'embedder', None ) ):
		message = get_embedder_unavailable_message( )
		st.session_state[ 'semantic_status' ] = message

		return {
				'success': False,
				'message': message,
				'doc_count': 0,
				'chunk_count': 0,
				'vector_dim': 0
		}

	if not isinstance( uploaded_files, list ) or len( uploaded_files ) == 0:
		message = 'No files were provided for semantic indexing.'
		st.session_state[ 'semantic_status' ] = message

		return {
				'success': False,
				'message': message,
				'doc_count': 0,
				'chunk_count': 0,
				'vector_dim': 0
		}

	try:
		chunk_size = int( st.session_state.get( 'semantic_chunk_size', 1200 ) )
		chunk_overlap = int( st.session_state.get( 'semantic_chunk_overlap', 200 ) )
	except Exception:
		chunk_size = 1200
		chunk_overlap = 200

	clear_existing = bool( st.session_state.get( 'semantic_clear_existing', True ) )
	append_existing = bool( st.session_state.get( 'semantic_append_existing', False ) )

	if append_existing:
		clear_existing = False

	all_chunks: List[ str ] = [ ]
	doc_names: List[ str ] = [ ]

	for f in uploaded_files:
		try:
			file_name = str( getattr( f, 'name', '' ) or '' ).strip( )
			file_bytes = f.getvalue( )
		except Exception:
			continue

		if not file_name or not file_bytes:
			continue

		try:
			text = extract_text( file_bytes=file_bytes, file_name=file_name )
		except Exception:
			text = ''

		if not text:
			try:
				text = file_bytes.decode( errors='ignore' )
			except Exception:
				text = ''

		if not text:
			continue

		try:
			chunks = chunk_text( text=text, size=chunk_size, overlap=chunk_overlap )
		except Exception:
			chunks = [ ]

		if not chunks:
			continue

		all_chunks.extend( [ str( chunk ) for chunk in chunks if chunk ] )
		doc_names.append( file_name )

	if len( all_chunks ) == 0:
		message = 'No extractable text was found in the uploaded files.'
		st.session_state[ 'semantic_status' ] = message

		return {
				'success': False,
				'message': message,
				'doc_count': 0,
				'chunk_count': 0,
				'vector_dim': 0
		}

	try:
		vecs = embedder.encode( all_chunks, show_progress_bar=False )
		vecs = np.asarray( vecs, dtype=np.float32 )

		if len( vecs.shape ) != 2 or vecs.shape[ 0 ] != len( all_chunks ):
			message = (
					'Semantic embedding generation returned an unexpected vector shape.')
			st.session_state[ 'semantic_status' ] = message

			return {
					'success': False,
					'message': message,
					'doc_count': len( doc_names ),
					'chunk_count': len( all_chunks ),
					'vector_dim': 0
			}
	except Exception as e:
		message = f'Semantic embedding generation failed: {e}'
		st.session_state[ 'semantic_status' ] = message

		return {
				'success': False,
				'message': message,
				'doc_count': len( doc_names ),
				'chunk_count': len( all_chunks ),
				'vector_dim': 0
		}

	try:
		initialize_database( )

		with sqlite3.connect( cfg.DB_PATH, timeout=30 ) as conn:
			if clear_existing:
				conn.execute( 'DELETE FROM embeddings' )

			for chunk_text_value, vec in zip( all_chunks, vecs ):
				vec = np.asarray( vec, dtype=np.float32 ).reshape( -1 )

				if vec.size == 0:
					continue

				conn.execute(
					'INSERT INTO embeddings (chunk, vector) VALUES (?, ?)',
					(chunk_text_value, vec.tobytes( ))
				)

			conn.commit( )
	except Exception as e:
		message = f'Semantic index database write failed: {e}'
		st.session_state[ 'semantic_status' ] = message

		return {
				'success': False,
				'message': message,
				'doc_count': len( doc_names ),
				'chunk_count': len( all_chunks ),
				'vector_dim': int( vecs.shape[ 1 ] ) if len( vecs.shape ) == 2 else 0
		}

	vector_dim = int( vecs.shape[ 1 ] ) if len( vecs.shape ) == 2 else 0
	st.session_state[ 'semantic_uploaded_names' ] = doc_names
	st.session_state[ 'semantic_index_doc_count' ] = len( doc_names )
	st.session_state[ 'semantic_index_chunk_count' ] = len( all_chunks )
	st.session_state[ 'semantic_index_dim' ] = vector_dim
	st.session_state[ 'semantic_result_rows' ] = [ ]
	st.session_state[ 'semantic_selected_rows' ] = [ ]
	st.session_state[ 'semantic_status' ] = 'Semantic index built successfully.'

	return {
			'success': True,
			'message': 'Semantic index built successfully.',
			'doc_count': len( doc_names ),
			'chunk_count': len( all_chunks ),
			'vector_dim': vector_dim
	}

query_semantic_index

query_semantic_index(
    query_text: str,
) -> List[Dict[str, Any]]

Purpose:

Query the semantic index and return ranked chunk results. Missing embeddings, database failures, empty indexes, malformed vectors, and vector dimension mismatches fail closed instead of raising hard runtime errors.

Parameters:

query_text : str Query text.

Returns:

List[Dict[str, Any]] Ranked semantic result rows.

Source code in app.py

def query_semantic_index( query_text: str ) -> List[ Dict[ str, Any ] ]:
	"""
		Purpose:
		--------
		Query the semantic index and return ranked chunk results. Missing embeddings,
		database failures, empty indexes, malformed vectors, and vector dimension mismatches
		fail closed instead of raising hard runtime errors.

		Parameters:
		-----------
		query_text : str
			Query text.

		Returns:
		--------
		List[Dict[str, Any]]
			Ranked semantic result rows.
	"""
	global embedder

	if not query_text or not str( query_text ).strip( ):
		st.session_state[ 'semantic_result_rows' ] = [ ]
		st.session_state[ 'semantic_status' ] = 'Enter a semantic query before searching.'
		return [ ]

	if not is_embedder_available( globals( ).get( 'embedder', None ) ):
		st.session_state[ 'semantic_result_rows' ] = [ ]
		st.session_state[ 'semantic_status' ] = get_embedder_unavailable_message( )
		return [ ]

	try:
		top_k = int( st.session_state.get( 'semantic_top_k', 8 ) )
	except Exception:
		top_k = 8

	try:
		min_similarity = float( st.session_state.get( 'semantic_min_similarity', 0.0 ) )
	except Exception:
		min_similarity = 0.0

	rows = decode_embedding_rows( )
	if not rows:
		st.session_state[ 'semantic_result_rows' ] = [ ]
		if not st.session_state.get( 'semantic_status', '' ):
			st.session_state[ 'semantic_status' ] = 'Semantic index is empty.'
		return [ ]

	try:
		q = embedder.encode( [ str( query_text ).strip( ) ], show_progress_bar=False )[ 0 ]
		q = np.asarray( q, dtype=np.float32 ).reshape( -1 )
	except Exception as e:
		st.session_state[ 'semantic_result_rows' ] = [ ]
		st.session_state[ 'semantic_status' ] = f'Semantic query embedding failed: {e}'
		return [ ]

	if q.size == 0:
		st.session_state[ 'semantic_result_rows' ] = [ ]
		st.session_state[ 'semantic_status' ] = (
				'Semantic query embedding returned an empty vector.')
		return [ ]

	scored_rows: List[ Dict[ str, Any ] ] = [ ]
	skipped_rows = 0

	for idx, (chunk_text_value, vec) in enumerate( rows, start=1 ):
		try:
			vec = np.asarray( vec, dtype=np.float32 ).reshape( -1 )

			if vec.size == 0 or vec.size != q.size:
				skipped_rows += 1
				continue

			score = cosine_similarity( q, vec )
		except Exception:
			skipped_rows += 1
			continue

		if score < min_similarity:
			continue

		scored_rows.append(
			{
					'Selected': False,
					'Rank': idx,
					'Score': float( score ),
					'Chunk': str( chunk_text_value or '' ),
					'Length': len( str( chunk_text_value or '' ) )
			}
		)

	scored_rows.sort( key=lambda r: r[ 'Score' ], reverse=True )
	scored_rows = scored_rows[ :top_k ]

	st.session_state[ 'semantic_last_query' ] = str( query_text ).strip( )
	st.session_state[ 'semantic_result_rows' ] = scored_rows

	if scored_rows:
		st.session_state[ 'semantic_status' ] = 'Semantic search completed.'
	elif skipped_rows:
		st.session_state[ 'semantic_status' ] = (
				'No semantic matches found. Some stored vectors were skipped because '
				'their dimensions did not match the active embedding model.')
	else:
		st.session_state[ 'semantic_status' ] = 'No semantic matches found.'

	return scored_rows

create_semantic_context

create_semantic_context() -> str

Purpose:

Build a semantic-context text block from selected search rows.

Parameters:

None

Returns:

str Semantic context text.

Source code in app.py

def create_semantic_context( ) -> str:
	"""
		Purpose:
		--------
		Build a semantic-context text block from selected search rows.

		Parameters:
		-----------
		None

		Returns:
		--------
		str
			Semantic context text.
	"""
	selected_rows = st.session_state.get( 'semantic_selected_rows', [ ] )

	if not isinstance( selected_rows, list ) or len( selected_rows ) == 0:
		return ''

	context_parts: List[ str ] = [ ]

	for idx, row in enumerate( selected_rows, start=1 ):
		chunk_text_value = str( row.get( 'Chunk', '' ) or '' ).strip( )
		score_value = row.get( 'Score', '' )

		if not chunk_text_value:
			continue

		context_parts.append( f'[Semantic Chunk {idx} | Score: {score_value}]\n{chunk_text_value}' )

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

extract_selected_rows

extract_selected_rows(
    edited_rows: Any,
) -> List[Dict[str, Any]]

Purpose:

Extract selected semantic rows from a data_editor result payload.

Parameters:

edited_rows : Any Data editor result payload.

Returns:

List[Dict[str, Any]] Selected rows.

Source code in app.py

def extract_selected_rows( edited_rows: Any ) -> List[ Dict[ str, Any ] ]:
	"""
		Purpose:
		--------
		Extract selected semantic rows from a data_editor result payload.

		Parameters:
		-----------
		edited_rows : Any
			Data editor result payload.

		Returns:
		--------
		List[Dict[str, Any]]
			Selected rows.
	"""
	selected: List[ Dict[ str, Any ] ] = [ ]

	if isinstance( edited_rows, pd.DataFrame ):
		for _, row in edited_rows.iterrows( ):
			row_dict = row.to_dict( )
			if bool( row_dict.get( 'Selected', False ) ):
				selected.append( row_dict )

		return selected

	if not isinstance( edited_rows, list ):
		return selected

	for row in edited_rows:
		if isinstance( row, dict ) and bool( row.get( 'Selected', False ) ):
			selected.append( row )

	return selected

send_text_chunks

send_text_chunks() -> None

Purpose:

Push selected semantic chunks into the shared basic document context buffer.

Parameters:

None

Returns:

None

Source code in app.py

def send_text_chunks( ) -> None:
	"""
		Purpose:
		--------
		Push selected semantic chunks into the shared basic document context buffer.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	context_text = create_semantic_context( )

	if not context_text:
		return

	existing_docs = st.session_state.get( 'basic_docs', [ ] )

	if not isinstance( existing_docs, list ):
		existing_docs = [ ]

	existing_docs.append( context_text )
	st.session_state[ 'basic_docs' ] = existing_docs
	st.session_state[ 'use_semantic' ] = True

send_docqna_chunks

send_docqna_chunks() -> None

Purpose:

Push selected semantic chunks into the shared document context buffer used by Document Q&A prompts.

Parameters:

None

Returns:

None

Source code in app.py

def send_docqna_chunks( ) -> None:
	"""
		Purpose:
		--------
		Push selected semantic chunks into the shared document context buffer used by
		Document Q&A prompts.

		Parameters:
		-----------
		None

		Returns:
		--------
		None
	"""
	context_text = create_semantic_context( )

	if not context_text:
		return

	buffer_rows = st.session_state.get( 'semantic_context_buffer', [ ] )

	if not isinstance( buffer_rows, list ):
		buffer_rows = [ ]

	buffer_rows.append( context_text )
	st.session_state[ 'semantic_context_buffer' ] = buffer_rows

request_text_generation_reset

request_text_generation_reset(reset_name: str) -> None

Purpose:

Request a Text Generation control reset without directly modifying any widget-owned keys after their widgets have been instantiated.

Parameters:

reset_name : str Name of the reset group to process on the next safe script pass.

Returns:

None

Source code in app.py

def request_text_generation_reset( reset_name: str ) -> None:
	"""
	Purpose:
	--------
	Request a Text Generation control reset without directly modifying any
	widget-owned keys after their widgets have been instantiated.

	Parameters:
	-----------
	reset_name : str
		Name of the reset group to process on the next safe script pass.

	Returns:
	--------
	None
	"""
	st.session_state[ 'pending_text_generation_reset' ] = str( reset_name or '' )

request_docqna_reset

request_docqna_reset(reset_name: str) -> None

Purpose:

Request a Document Q&A control reset without directly modifying any widget-owned keys after their widgets have been instantiated.

Parameters:

reset_name : str Name of the reset group to process on the next safe script pass.

Returns:

None

Source code in app.py

def request_docqna_reset( reset_name: str ) -> None:
	"""
	Purpose:
	--------
	Request a Document Q&A control reset without directly modifying any
	widget-owned keys after their widgets have been instantiated.

	Parameters:
	-----------
	reset_name : str
		Name of the reset group to process on the next safe script pass.

	Returns:
	--------
	None
	"""
	st.session_state[ 'pending_docqna_reset' ] = str( reset_name or '' )

request_document_unload

request_document_unload() -> None

Purpose:

Request that active uploaded documents be unloaded on the next safe script pass.

Parameters:

None

Returns:

None

Source code in app.py

def request_document_unload( ) -> None:
	"""
	Purpose:
	--------
	Request that active uploaded documents be unloaded on the next safe script pass.

	Parameters:
	-----------
	None

	Returns:
	--------
	None
	"""
	st.session_state[ 'pending_doc_loader_action' ] = 'unload'

render_pdf_preview

render_pdf_preview(
    file_bytes: bytes, preview_name: str
) -> None

Purpose:

Render a PDF preview using st.pdf when available, otherwise fall back to a base64 iframe and, if needed, extracted text.

Parameters:

file_bytes : bytes PDF file bytes.

str

Display name for the active PDF.

Returns:

None

Source code in app.py

def render_pdf_preview( file_bytes: bytes, preview_name: str ) -> None:
	"""
	Purpose:
	--------
	Render a PDF preview using st.pdf when available, otherwise fall back to a
	base64 iframe and, if needed, extracted text.

	Parameters:
	-----------
	file_bytes : bytes
		PDF file bytes.

	preview_name : str
		Display name for the active PDF.

	Returns:
	--------
	None
	"""
	if not file_bytes:
		st.info( 'PDF preview unavailable.' )
		return

	try:
		if hasattr( st, 'pdf' ):
			st.pdf( file_bytes, height=420 )
			return
	except Exception:
		pass

	try:
		encoded_pdf = base64.b64encode( file_bytes ).decode( 'utf-8' )
		pdf_html = (
				f'<iframe src="data:application/pdf;base64,{encoded_pdf}" '
				f'width="100%" height="420" type="application/pdf"></iframe>'
		)
		st.markdown( pdf_html, unsafe_allow_html=True )
		return
	except Exception:
		pass

	preview_text = extract_text( file_bytes, preview_name )
	if preview_text:
		st.text_area(
			label=f'Preview: {preview_name}',
			value=preview_text[ :4000 ],
			height=420,
			disabled=True,
			key='doc_loader_pdf_text_fallback'
		)
	else:
		st.info( 'PDF preview unavailable.' )

run_image_mode_adapter

run_image_mode_adapter(
    image_bytes: bytes, image_name: str, prompt: str
) -> str

Purpose:

Run an optional image-analysis adapter when one has been wired into app.py. The function fails closed when the selected model or runtime does not support image analysis.

Parameters:

image_bytes : bytes Uploaded image bytes.

str

Uploaded image filename.

str

User prompt for image analysis.

Returns:

str Image analysis response text.

Source code in app.py

def run_image_mode_adapter( image_bytes: bytes, image_name: str, prompt: str ) -> str:
	"""
		Purpose:
		--------
		Run an optional image-analysis adapter when one has been wired into app.py.
		The function fails closed when the selected model or runtime does not support
		image analysis.

		Parameters:
		-----------
		image_bytes : bytes
			Uploaded image bytes.

		image_name : str
			Uploaded image filename.

		prompt : str
			User prompt for image analysis.

		Returns:
		--------
		str
			Image analysis response text.
	"""
	try:
		if not model_supports_capability( 'image_mode' ):
			return get_capability_status_message( 'image_mode' )

		runtime_status = get_runtime_multimodal_status( )
		if not bool( runtime_status.get( 'image_runtime_available', False ) ):
			return get_capability_status_message( 'image_mode' )

		adapter = globals( ).get( 'analyze_image_with_model', None )
		if not callable( adapter ):
			return (
					'Image Mode is configured for this model, but no image adapter named '
					'analyze_image_with_model is wired into app.py yet.')

		try:
			result = adapter(
				model_path=get_selected_model_path( ),
				model_name=get_selected_model_name( ),
				image_bytes=image_bytes,
				image_name=image_name,
				prompt=prompt
			)
		except TypeError:
			result = adapter( image_bytes, prompt )

		if result is None:
			return ''

		return str( result )
	except Exception as e:
		return f'Image analysis failed: {e}'

build_image_context_text

build_image_context_text(
    image_name: str, prompt: str, response: str
) -> str

Purpose:

Build reusable image context text for Text Generation, Document Q&A, or Prompt Engineering workflows.

Parameters:

image_name : str Uploaded image filename.

str

User image-analysis prompt.

str

Image-analysis response or runtime status.

Returns:

str Reusable image context text.

Source code in app.py

def build_image_context_text( image_name: str, prompt: str, response: str ) -> str:
	"""
		Purpose:
		--------
		Build reusable image context text for Text Generation, Document Q&A, or Prompt
		Engineering workflows.

		Parameters:
		-----------
		image_name : str
			Uploaded image filename.

		prompt : str
			User image-analysis prompt.

		response : str
			Image-analysis response or runtime status.

		Returns:
		--------
		str
			Reusable image context text.
	"""
	name_value = str( image_name or '' ).strip( )
	prompt_value = str( prompt or '' ).strip( )
	response_value = str( response or '' ).strip( )

	parts: List[ str ] = [ ]
	if name_value:
		parts.append( f'Image Source: {name_value}' )
	if prompt_value:
		parts.append( f'Image Prompt: {prompt_value}' )
	if response_value:
		parts.append( f'Image Analysis:\n{response_value}' )

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

run_audio_mode_adapter

run_audio_mode_adapter(
    audio_bytes: bytes, audio_name: str, prompt: str
) -> str

Purpose:

Run an optional audio-analysis adapter when one has been wired into app.py. The function fails closed when the selected model or runtime does not support audio transcription, translation, or audio analysis.

Parameters:

audio_bytes : bytes Uploaded audio bytes.

str

Uploaded audio filename.

str

User prompt for audio transcription or analysis.

Returns:

str Audio transcription, translation, analysis, or runtime status text.

Source code in app.py

def run_audio_mode_adapter( audio_bytes: bytes, audio_name: str, prompt: str ) -> str:
	"""
		Purpose:
		--------
		Run an optional audio-analysis adapter when one has been wired into app.py.
		The function fails closed when the selected model or runtime does not support
		audio transcription, translation, or audio analysis.

		Parameters:
		-----------
		audio_bytes : bytes
			Uploaded audio bytes.

		audio_name : str
			Uploaded audio filename.

		prompt : str
			User prompt for audio transcription or analysis.

		Returns:
		--------
		str
			Audio transcription, translation, analysis, or runtime status text.
	"""
	try:
		if not model_supports_capability( 'audio_mode' ):
			return get_capability_status_message( 'audio_mode' )

		runtime_status = get_runtime_multimodal_status( )
		if not bool( runtime_status.get( 'audio_runtime_available', False ) ):
			return get_capability_status_message( 'audio_mode' )

		adapter = globals( ).get( 'analyze_audio_with_model', None )
		if not callable( adapter ):
			return (
					'Audio Mode is configured for this model, but no audio adapter named '
					'analyze_audio_with_model is wired into app.py yet.')

		try:
			result = adapter(
				model_path=get_selected_model_path( ),
				model_name=get_selected_model_name( ),
				audio_bytes=audio_bytes,
				audio_name=audio_name,
				prompt=prompt
			)
		except TypeError:
			result = adapter( audio_bytes, prompt )

		if result is None:
			return ''

		return str( result )
	except Exception as e:
		return f'Audio analysis failed: {e}'

build_audio_context_text

build_audio_context_text(
    audio_name: str, prompt: str, response: str
) -> str

Purpose:

Build reusable audio context text for Text Generation, Document Q&A, Semantic Search, or Prompt Engineering workflows.

Parameters:

audio_name : str Uploaded audio filename.

str

User audio-analysis prompt.

str

Audio transcription, translation, analysis, or runtime status.

Returns:

str Reusable audio context text.

Source code in app.py

def build_audio_context_text( audio_name: str, prompt: str, response: str ) -> str:
	"""
		Purpose:
		--------
		Build reusable audio context text for Text Generation, Document Q&A, Semantic
		Search, or Prompt Engineering workflows.

		Parameters:
		-----------
		audio_name : str
			Uploaded audio filename.

		prompt : str
			User audio-analysis prompt.

		response : str
			Audio transcription, translation, analysis, or runtime status.

		Returns:
		--------
		str
			Reusable audio context text.
	"""
	name_value = str( audio_name or '' ).strip( )
	prompt_value = str( prompt or '' ).strip( )
	response_value = str( response or '' ).strip( )

	parts: List[ str ] = [ ]
	if name_value:
		parts.append( f'Audio Source: {name_value}' )
	if prompt_value:
		parts.append( f'Audio Prompt: {prompt_value}' )
	if response_value:
		parts.append( f'Audio Transcript / Analysis:\n{response_value}' )

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

get_audio_mime_type

get_audio_mime_type(audio_name: str) -> str

Purpose:

Return a browser-friendly MIME type for Streamlit audio preview based on the uploaded audio filename.

Parameters:

audio_name : str Uploaded audio filename.

Returns:

str Audio MIME type.

Source code in app.py

def get_audio_mime_type( audio_name: str ) -> str:
	"""
		Purpose:
		--------
		Return a browser-friendly MIME type for Streamlit audio preview based on the
		uploaded audio filename.

		Parameters:
		-----------
		audio_name : str
			Uploaded audio filename.

		Returns:
		--------
		str
			Audio MIME type.
	"""
	name_value = str( audio_name or '' ).strip( ).lower( )

	if name_value.endswith( '.mp3' ):
		return 'audio/mpeg'
	if name_value.endswith( '.m4a' ):
		return 'audio/mp4'
	if name_value.endswith( '.flac' ):
		return 'audio/flac'
	if name_value.endswith( '.ogg' ):
		return 'audio/ogg'
	if name_value.endswith( '.wav' ):
		return 'audio/wav'

	return 'audio/wav'