Spaces:
Sleeping
Sleeping
File size: 5,583 Bytes
d38e9c6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
import streamlit as st
from ibm_watsonx_ai import APIClient, Credentials
from ibm_watsonx_ai.foundation_models import ModelInference
from io import BytesIO
from reportlab.lib.pagesizes import letter
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib import colors
from reportlab.lib.enums import TA_LEFT, TA_RIGHT
from datetime import datetime
import regex
import os
def check_password():
def password_entered():
if st.session_state["password"] == os.getenv("APP_PASSWORD"):
st.session_state["password_correct"] = True
del st.session_state["password"]
else:
st.session_state["password_correct"] = False
if "password_correct" not in st.session_state:
st.markdown("\n\n")
st.text_input(
"Enter the password",
type="password",
on_change=password_entered,
key="password",
)
st.divider()
st.info("Designed and developed by Milan Mrdenovic © IBM Norway 2025")
return False
elif not st.session_state["password_correct"]:
st.markdown("\n\n")
st.text_input(
"Enter the password",
type="password",
on_change=password_entered,
key="password",
)
st.divider()
st.info("Designed and developed by Milan Mrdenovic © IBM Norway 2025")
st.error("😕 Password incorrect")
return False
else:
return True
def initialize_session_state():
if "chat_history" not in st.session_state:
st.session_state.chat_history = []
def setup_watsonxai_client(
api_key: str, project_id: str, url: str = "https://eu-de.ml.cloud.ibm.com"
):
"""Set up a watsonx.ai python SDK client using an apikey and project_id."""
from ibm_watsonx_ai import APIClient, Credentials
wx_credentials = Credentials(url=url, api_key=api_key)
wxai_client = APIClient(wx_credentials, project_id=project_id)
return wxai_client
emoji_pattern = regex.compile(r"\p{Emoji}", flags=regex.UNICODE)
def remove_emojis(text):
return emoji_pattern.sub(r"", text)
def create_pdf_from_chat(chat_history):
buffer = BytesIO()
doc = SimpleDocTemplate(buffer, pagesize=letter, topMargin=30, bottomMargin=30)
styles = getSampleStyleSheet()
flowables = []
title_style = ParagraphStyle(
"Title", parent=styles["Heading1"], fontSize=18, spaceAfter=20
)
flowables.append(
Paragraph(
f"Chat History - Generated on {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
title_style,
)
)
user_style = ParagraphStyle(
"UserStyle",
parent=styles["Normal"],
backColor=colors.lightblue,
borderPadding=10,
alignment=TA_RIGHT,
)
jimmy_style = ParagraphStyle(
"JimmyStyle",
parent=styles["Normal"],
backColor=colors.lavender,
borderPadding=10,
)
for message in chat_history:
role = message["role"]
content = remove_emojis(message["content"])
style = user_style if role == "user" else jimmy_style
flowables.append(Paragraph(f"<b>{role.capitalize()}:</b> {content}", style))
flowables.append(Spacer(1, 12))
doc.build(flowables)
buffer.seek(0)
return buffer
def watsonx_chat_prompt(
messages,
stream=False,
client=None,
wx_url=None,
wx_apikey=None,
project_id=None,
model_id=None,
params=None,
):
"""
Dynamic chat function for Watson AI
Args:
messages (list): List of message objects following watsonx schema.
Each message should have 'role' and 'content' keys.
Supports system, user, assistant, and tool messages.
stream (bool): If True, return streaming generator; if False, return complete response
client (APIClient): Pre-configured Watson client (optional)
wx_url (str): Watson URL (required if no client)
wx_apikey (str): Watson API key (required if no client)
project_id (str): Watson project ID (required if no client)
model_id (str): Model identifier
params (dict): Model parameters (optional)
Returns:
str or generator: Complete response text or streaming generator based on stream parameter
"""
# from ibm_watsonx_ai.foundation_models import ModelInference
# from ibm_watsonx_ai import APIClient, Credentials
if params is None:
params = {
"temperature": 0.7,
"max_tokens": 4096,
"top_p": 1.0,
"stop": ["</s>", "<|end_of_text|>", "<|endoftext|>"],
# "frequency_penalty": 0.5,
# "presence_penalty": 0.3,
}
# Use provided client or create new one
if client is None:
wx_credentials = Credentials(url=wx_url, api_key=wx_apikey)
client = APIClient(credentials=wx_credentials, project_id=project_id)
chat_model = ModelInference(api_client=client, model_id=model_id, params=params)
if stream:
return chat_model.chat_stream(messages=messages)
else:
return chat_model.chat(messages=messages)
def generate_response(watsonx_chat_prompt, stream):
if stream:
for chunk in watsonx_chat_prompt:
if chunk["choices"]:
yield chunk["choices"][0]["delta"].get("content", "")
else:
return watsonx_chat_prompt["choices"][0]["message"]["content"]
|