Spaces:
Sleeping
Sleeping
File size: 4,359 Bytes
d38e9c6 60a5571 d38e9c6 e648ed7 d38e9c6 e648ed7 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 |
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 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"Jimmy Chat History - {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,
wx_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=wx_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"]
|