Assembly: Jimi
Filename: app.py
Author: Terry D. Eppler
Created: 05-31-2022
Last Modified By: Terry D. Eppler
Last Modified On: 12-27-2025
app.py
Copyright © 2022 Terry Eppler
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the “Software”),
to deal in the Software without restriction,
including without limitation the rights to use,
copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software,
and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE.
You can contact me at: terryeppler@gmail.com or eppler.terry@epa.gov
app.py
local_model_available
local_model_available() -> bool
Purpose:
Determine whether the configured local GGUF model exists.
Parameters:
None
Returns:
bool
True when the configured model file exists; otherwise False.
Source code in app.py
| def local_model_available( ) -> bool:
"""
Purpose:
--------
Determine whether the configured local GGUF model exists.
Parameters:
-----------
None
Returns:
--------
bool
True when the configured model file exists; otherwise False.
"""
try:
return MODEL_PATH_OBJ.exists( )
except Exception:
return False
|
initialize_database
initialize_database() -> None
Purpose:
Ensure required SQLite tables exist and that the Prompts table contains the
columns required by the prompt utilities and Prompt Engineering mode.
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 and Prompt Engineering mode.
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
)
)
"""
)
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( )
|
decode_embedding_rows
decode_embedding_rows() -> List[Tuple[str, np.ndarray]]
Purpose:
Read and decode rows from the semantic embeddings table.
Parameters:
None
Returns:
List[Tuple[str, np.ndarray]]
Source code in app.py
| def decode_embedding_rows( ) -> List[ Tuple[ str, np.ndarray ] ]:
"""
Purpose:
--------
Read and decode rows from the semantic embeddings table.
Parameters:
-----------
None
Returns:
--------
List[Tuple[str, np.ndarray]]
"""
rows_out: List[ Tuple[ str, np.ndarray ] ] = [ ]
with sqlite3.connect( cfg.DB_PATH ) as conn:
rows = conn.execute( 'SELECT chunk, vector FROM embeddings' ).fetchall( )
for chunk_text_value, vector_blob in rows:
if not vector_blob:
continue
vec = np.frombuffer( vector_blob, dtype=np.float32 )
if vec.size == 0:
continue
rows_out.append( (str( chunk_text_value or '' ), vec) )
return rows_out
|
clear_semantic_index
clear_semantic_index() -> None
Purpose:
Clear the semantic embeddings table and reset Semantic Search diagnostics.
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.
Parameters:
-----------
None
Returns:
--------
None
"""
with sqlite3.connect( cfg.DB_PATH ) 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
|
build_semantic_index
build_semantic_index(
uploaded_files: List[Any],
) -> Dict[str, Any]
Purpose:
Build or append a semantic chunk index from uploaded files.
Parameters:
uploaded_files : List[Any]
Returns:
Dict[str, Any]
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.
Parameters:
-----------
uploaded_files : List[Any]
Returns:
--------
Dict[str, Any]
"""
embedder = load_embedder( )
if embedder is None:
return {
'success': False,
'message': 'Embedding model unavailable.',
'doc_count': 0,
'chunk_count': 0,
'vector_dim': 0,
}
chunk_size = int( st.session_state.get( 'semantic_chunk_size', 1200 ) )
chunk_overlap = int( st.session_state.get( 'semantic_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 clear_existing and not append_existing:
clear_semantic_index( )
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
text = extract_text_from_bytes( file_bytes=file_bytes, file_name=file_name )
if not text:
try:
text = file_bytes.decode( errors='ignore' )
except Exception:
text = ''
if not text:
continue
chunks = chunk_text( text=text, size=chunk_size, overlap=chunk_overlap )
if not chunks:
continue
all_chunks.extend( chunks )
doc_names.append( file_name )
if len( all_chunks ) == 0:
return {
'success': False,
'message': 'No extractable text was found in the uploaded files.',
'doc_count': 0,
'chunk_count': 0,
'vector_dim': 0,
}
vecs = embedder.encode( all_chunks, show_progress_bar=False )
vecs = np.asarray( vecs, dtype=np.float32 )
with sqlite3.connect( cfg.DB_PATH ) as conn:
for chunk_text_value, vec in zip( all_chunks, vecs ):
conn.execute(
'INSERT INTO embeddings (chunk, vector) VALUES (?, ?)',
(chunk_text_value, vec.tobytes( ))
)
conn.commit( )
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
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.
Parameters:
query_text : str
Returns:
List[Dict[str, Any]]
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.
Parameters:
-----------
query_text : str
Returns:
--------
List[Dict[str, Any]]
"""
if not query_text or not query_text.strip( ):
return [ ]
embedder = load_embedder( )
if embedder is None:
return [ ]
top_k = int( st.session_state.get( 'semantic_top_k', 8 ) )
min_similarity = float( st.session_state.get( 'semantic_min_similarity', 0.0 ) )
rows = decode_embedding_rows( )
if not rows:
return [ ]
q = embedder.encode( [ query_text.strip( ) ], show_progress_bar=False )[ 0 ]
q = np.asarray( q, dtype=np.float32 )
scored_rows: List[ Dict[ str, Any ] ] = [ ]
for idx, (chunk_text_value, vec) in enumerate( rows, start=1 ):
score = cosine_similarity( q, vec )
if score < min_similarity:
continue
scored_rows.append(
{
'Selected': False,
'Rank': idx,
'Score': float( score ),
'Chunk': chunk_text_value,
'Length': len( chunk_text_value ),
}
)
scored_rows.sort( key=lambda r: r[ 'Score' ], reverse=True )
scored_rows = scored_rows[ :top_k ]
if bool( st.session_state.get( 'semantic_group_by_document', False ) ):
# Current embeddings table stores chunk text but not document name.
# Preserve current schema and grouping behavior as a no-op until schema expansion.
pass
st.session_state[ 'semantic_last_query' ] = query_text.strip( )
st.session_state[ 'semantic_result_rows' ] = scored_rows
return scored_rows
|
build_semantic_context_from_selection
build_semantic_context_from_selection() -> str
Purpose:
Build a semantic-context text block from selected search rows.
Parameters:
None
Returns:
str
Source code in app.py
| def build_semantic_context_from_selection( ) -> str:
"""
Purpose:
--------
Build a semantic-context text block from selected search rows.
Parameters:
-----------
None
Returns:
--------
str
"""
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( )
|
send_selected_semantic_chunks_to_text_generation
send_selected_semantic_chunks_to_text_generation() -> None
Purpose:
Push selected semantic chunks into the shared basic document context buffer.
Parameters:
None
Returns:
None
Source code in app.py
| def send_selected_semantic_chunks_to_text_generation( ) -> None:
"""
Purpose:
--------
Push selected semantic chunks into the shared basic document context buffer.
Parameters:
-----------
None
Returns:
--------
None
"""
context_text = build_semantic_context_from_selection( )
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_selected_semantic_chunks_to_doc_qna
send_selected_semantic_chunks_to_doc_qna() -> None
Purpose:
Push selected semantic chunks into the shared document context buffer used by prompts.
Parameters:
None
Returns:
None
Source code in app.py
| def send_selected_semantic_chunks_to_doc_qna( ) -> None:
"""
Purpose:
--------
Push selected semantic chunks into the shared document context buffer used by prompts.
Parameters:
-----------
None
Returns:
--------
None
"""
context_text = build_semantic_context_from_selection( )
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
|
extract_selected_semantic_rows(
edited_rows: List[Dict[str, Any]],
) -> List[Dict[str, Any]]
Extract selected semantic rows from a data_editor result payload.
edited_rows : List[Dict[str, Any]]
List[Dict[str, Any]]
Source code in app.py
| def extract_selected_semantic_rows( edited_rows: List[ Dict[ str, Any ] ] ) -> List[ Dict[ str, Any ] ]:
"""
Purpose:
--------
Extract selected semantic rows from a data_editor result payload.
Parameters:
-----------
edited_rows : List[Dict[str, Any]]
Returns:
--------
List[Dict[str, Any]]
"""
selected: List[ Dict[ str, Any ] ] = [ ]
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
|
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 = None,
overlap: int | None = 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 = None, overlap: int | None = 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 = jimil( 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() -> 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.
Parameters:
None
Returns:
str
Source code in app.py
| def build_task_instruction_block( ) -> str:
"""
Purpose:
--------
Build a task-specific instruction block for Text Generation mode.
Parameters:
-----------
None
Returns:
--------
str
"""
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( )
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.' )
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( )
|
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.
Parameters:
user_input : str
Returns:
str
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.
Parameters:
-----------
user_input : str
Returns:
--------
str
"""
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', [ ] )
top_k_value = int( st.session_state.get( 'top_k', 0 ) )
if top_k_value <= 0:
top_k_value = 4
system_parts: List[ str ] = [ ]
if system_instructions:
system_parts.append( system_instructions )
if task_block:
system_parts.append( 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:
if embedder is None:
embedder = load_embedder( )
if embedder is not None:
with sqlite3.connect( cfg.DB_PATH ) as conn:
rows = conn.execute( 'SELECT chunk, vector FROM embeddings' ).fetchall( )
if rows:
q = embedder.encode( [ user_input ] )[ 0 ]
scored = [ (c, cosine_similarity( q, np.frombuffer( v ) )) for c, v in rows ]
for c, _ in sorted( scored, key=lambda x: x[ 1 ], reverse=True )[ :top_k_value ]:
prompt += f'<|system|>\nSemantic Context:\n{c}\n</s>\n'
if use_document_context and isinstance( basic_docs, list ):
for d in basic_docs[ :6 ]:
prompt += f'<|system|>\nDocument Context:\n{d}\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_input}\n</s>\n<|assistant|>\n'
return prompt
|
run_llm_turn
run_llm_turn(
user_input: str,
temperature: float,
top_p: float,
repeat_penalty: float,
max_tokens: int,
stream: bool,
output: Any | None = None,
) -> str
Purpose:
Run a single LLM turn using the current session-state runtime settings.
Parameters:
user_input : str
temperature : float
top_p : float
repeat_penalty : float
max_tokens : int
stream : bool
output : Any | None
Returns:
str
Source code in app.py
| def run_llm_turn( user_input: str, temperature: float, top_p: float, repeat_penalty: float,
max_tokens: int, stream: bool, output: Any | None = None ) -> str:
"""
Purpose:
--------
Run a single LLM turn using the current session-state runtime settings.
Parameters:
-----------
user_input : str
temperature : float
top_p : float
repeat_penalty : float
max_tokens : int
stream : bool
output : Any | None
Returns:
--------
str
"""
global llm
if user_input is None:
return ''
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 )
llm = load_llm( ctx_value, thread_value )
if llm is None:
st.error( f'Local model unavailable at {cfg.MODEL_PATH}' )
return ''
prompt = build_prompt( user_input )
max_token_value = int( max_tokens ) if int( max_tokens ) > 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
if not stream:
resp = llm(
prompt,
stream=False,
max_tokens=max_token_value,
temperature=temperature_value,
top_p=top_p_value,
repeat_penalty=repeat_penalty_value,
stop=[ '</s>' ]
)
text = (resp.get( 'choices', [ { 'text': '' } ] )[ 0 ].get( 'text', '' ) or '')
return text.strip( )
buf = ''
if output is None:
output = st.empty( )
for chunk in llm(
prompt,
stream=True,
max_tokens=max_token_value,
temperature=temperature_value,
top_p=top_p_value,
repeat_penalty=repeat_penalty_value,
stop=[ '</s>' ]
):
buf += chunk[ 'choices' ][ 0 ][ 'text' ]
output.markdown( buf + '▌' )
output.markdown( buf )
return buf.strip( )
|
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',
'Audio',
'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 'audio' in blob or 'transcrib' in blob:
return 'Audio'
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 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 == 'Audio':
lines.append(
'Work from transcript/audio-derived text and preserve meaning and speaker intent.' )
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 a 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(
category: str,
task_type: str,
response_format: str,
language: str,
) -> None
Apply prompt metadata to the shared app contract.
category : str
task_type : str
response_format : str
language : str
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 )
|
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 "jimil" 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": jimil,
"primary_key": jimil,
"auto_increment": jimil
}
]
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": jimil,
"primary_key": jimil,
"auto_increment": jimil
}
]
"""
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_doc_action_instruction
get_doc_action_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 get_doc_action_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_document_instruction_block
build_document_instruction_block() -> str
Purpose:
Build a unified instruction block for document-grounded answering.
Parameters:
None
Returns:
str
Source code in app.py
| def build_document_instruction_block( ) -> str:
"""
Purpose:
--------
Build a unified instruction block for document-grounded answering.
Parameters:
-----------
None
Returns:
--------
str
"""
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( 'doc_action', 'Answer Question' ) or 'Answer Question' )
lines: List[ str ] = [ ]
lines.append( 'Document Q&A Instructions:' )
lines.append( f'- Action: {doc_action}' )
lines.append( f'- Response Format: {response_format}' )
lines.append( f'- Action Guidance: {get_doc_action_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_bytes(
file_bytes: bytes, file_name: str = ""
) -> str
Extract text from PDF or text-based documents using the current document parsing settings.
file_bytes : bytes
file_name : str
str
Source code in app.py
| def extract_text_from_bytes( file_bytes: bytes, file_name: str = '' ) -> str:
"""
Purpose:
--------
Extract text from PDF or text-based documents using the current document parsing settings.
Parameters:
-----------
file_bytes : bytes
file_name : str
Returns:
--------
str
"""
if not file_bytes:
return ''
file_name_value = str( file_name or '' ).lower( )
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 ) )
try:
if file_name_value.endswith( '.pdf' ) or file_name_value == '':
if prefer_native_pdf_text:
import fitz
doc = fitz.open( stream=file_bytes, filetype='pdf' )
parts: List[ str ] = [ ]
page_index = 0
for page in doc:
page_index += 1
page_text = page.get_text( 'text' ) or ''
if include_page_markers:
parts.append( f'[Page {page_index}]' )
parts.append( page_text )
return '\n'.join( parts ).strip( )
except Exception:
pass
try:
return file_bytes.decode( errors='ignore' ).strip( )
except Exception:
return ''
|
route_document_query
route_document_query(prompt: str) -> str
Purpose:
Route a document question or action through the unified chat pipeline.
Parameters:
prompt : str
Returns:
str
Source code in app.py
| def route_document_query( prompt: str ) -> str:
"""
Purpose:
--------
Route a document question or action through the unified chat pipeline.
Parameters:
-----------
prompt : str
Returns:
--------
str
"""
user_input = build_document_user_input(
user_query=prompt,
k=int( st.session_state.get( 'retrieval_k', 6 ) )
)
if not user_input:
user_input = (prompt or '').strip( )
return run_llm_turn(
user_input=user_input,
temperature=float( st.session_state.get( 'temperature', 0.0 ) ),
top_p=float( st.session_state.get( 'top_percent', 0.95 ) ),
repeat_penalty=float( st.session_state.get( 'repeat_penalty', 1.1 ) ),
max_tokens=int( st.session_state.get( 'max_tokens', 1024 ) ) or 1024,
stream=False,
output=None
)
|
summarize_active_document
summarize_active_document() -> str
Purpose:
Summarize the currently active document set using the document routing layer.
Parameters:
None
Returns:
str
Source code in app.py
| def summarize_active_document( ) -> str:
"""
Purpose:
--------
Summarize the currently active document set using the document routing layer.
Parameters:
-----------
None
Returns:
--------
str
"""
system_instructions = get_effective_system_instructions( )
summary_prompt = """
Provide a clear, structured summary of the active document set.
Include:
- Purpose
- Key themes
- Major conclusions
- Important data points
- Open questions or uncertainties
"""
if system_instructions:
summary_prompt = f'{system_instructions}\n\n{summary_prompt}'
return route_document_query( summary_prompt.strip( ) )
|
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( )
|
extract_text(file_bytes: bytes, file_name: str = '') -> str
Extract document text using the configured parsing behavior.
file_bytes : bytes
file_name : str
str
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
file_name : str
Returns:
--------
str
"""
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_document_inventory_rows
build_document_inventory_rows() -> 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_document_inventory_rows( ) -> 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_active_document_names_text
get_active_document_names_text() -> str
Purpose:
Build a human-readable string of active document names.
Parameters:
None
Returns:
str
Source code in app.py
| def get_active_document_names_text( ) -> 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 ] )
|
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.
Parameters:
embedder : Any | None
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.
Parameters:
-----------
embedder : Any | None
Returns:
--------
None
"""
if embedder is None:
st.session_state[ 'docqna_vec_ready' ] = False
st.session_state[ 'docqna_chunk_count' ] = 0
st.session_state[ 'docqna_fallback_rows' ] = [ ]
st.session_state[ 'doc_inventory_rows' ] = build_document_inventory_rows( )
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[ 'doc_inventory_rows' ] = build_document_inventory_rows( )
return
st.session_state[ 'docqna_fingerprint' ] = fp
st.session_state[ 'docqna_chunk_count' ] = 0
st.session_state[ 'docqna_fallback_rows' ] = [ ]
st.session_state[ 'doc_inventory_rows' ] = build_document_inventory_rows( )
dim_value = getattr( embedder, 'get_sentence_embedding_dimension', lambda: 384 )( )
dim = int( dim_value ) if dim_value else 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
vecs = embedder.encode( chunks, show_progress_bar=False )
vecs = np.asarray( vecs, dtype=np.float32 )
if vec_ready:
for chunk_text_value, v in zip( chunks, vecs ):
cur.execute(
'INSERT INTO docqna_vec ( embedding, doc_name, chunk ) VALUES ( ?, ?, ? );',
(v.tobytes( ), name, chunk_text_value)
)
else:
for chunk_text_value, v in zip( chunks, vecs ):
fallback_rows.append( (name, chunk_text_value, v.tobytes( )) )
total_chunks += int( len( chunks ) )
conn.commit( )
st.session_state[ 'docqna_chunk_count' ] = total_chunks
if not vec_ready:
st.session_state[ 'docqna_fallback_rows' ] = fallback_rows
else:
st.session_state[ 'docqna_fallback_rows' ] = [ ]
except Exception:
st.session_state[ 'docqna_vec_ready' ] = False
st.session_state[ 'docqna_chunk_count' ] = 0
st.session_state[ 'docqna_fallback_rows' ] = [ ]
finally:
conn.close( )
|
retrieve_chunks
retrieve_chunks(
query: str, k: int | None = 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.
Parameters:
query : str
k : int | None
Returns:
List[Tuple[str, str, float]]
Source code in app.py
| def retrieve_chunks( query: str, k: int | None = 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.
Parameters:
-----------
query : str
k : int | None
Returns:
--------
List[Tuple[str, str, float]]
"""
if not query or not query.strip( ):
return [ ]
embedder = load_embedder( )
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
qv = embedder.encode( [ query ], show_progress_bar=False )
qv = np.asarray( qv, dtype=np.float32 )[ 0 ]
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( )
return [ (r[ 0 ], r[ 1 ], float( r[ 2 ] )) for r in rows ]
except Exception:
st.session_state[ 'docqna_vec_ready' ] = False
finally:
conn.close( )
if not bool( st.session_state.get( 'allow_similarity_fallback', True ) ):
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:
if not vec_blob:
continue
v = np.frombuffer( vec_blob, dtype=np.float32 )
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 )
return results[ : int( k_value ) ]
|
build_document_user_input(
user_query: str, k: int | None = None
) -> str
Build a document-grounded prompt using retrieved excerpts and the current document action.
user_query : str
k : int | None
str
Source code in app.py
| def build_document_user_input( user_query: str, k: int | None = None ) -> str:
"""
Purpose:
--------
Build a document-grounded prompt using retrieved excerpts and the current document action.
Parameters:
-----------
user_query : str
k : int | None
Returns:
--------
str
"""
system = get_effective_system_instructions( )
doc_instruction_block = build_document_instruction_block( )
hits = retrieve_chunks( user_query, k=k )
st.session_state[ 'doc_last_retrieval_hits' ] = hits
context_blocks: List[ str ] = [ ]
for doc_name, chunk, score in hits:
context_blocks.append( f'[Document: {doc_name}]\n{chunk}'.strip( ) )
context = '\n\n'.join( context_blocks ).strip( )
active_doc_names = get_active_document_names_text( )
prompt_parts: List[ str ] = [ ]
if system:
prompt_parts.append( system )
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:
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( )
|
load_llm
load_llm(ctx: int, threads: int) -> Any | None
Purpose:
Lazily load the local llama.cpp model using the supplied runtime settings.
Parameters:
ctx : int
Context window size.
threads : int
CPU thread count.
Returns:
Any | None
Source code in app.py
| @st.cache_resource
def load_llm( ctx: int, threads: int ) -> Any | None:
"""
Purpose:
--------
Lazily load the local llama.cpp model using the supplied runtime settings.
Parameters:
-----------
ctx : int
Context window size.
threads : int
CPU thread count.
Returns:
--------
Any | None
"""
try:
if not local_model_available( ):
return None
from llama_cpp import Llama
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 )
return Llama(
model_path=str( cfg.MODEL_PATH ),
n_ctx=ctx_value,
n_threads=thread_value,
n_batch=512,
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
|