NOTICE
Open WebUI Community is currently undergoing a major revamp to improve user experience
and performance. Expected completion by year-end! ✨
Function
action
v2.0.0
Visualize Data
This function makes charts out of data in the conversation and render it in the chat. Forked from @u/omar97 to fix new FileForm schema
Function Content
python
"""
title: Make charts out of your data v2
author: Iqbal Maulana
author_url: https://github.com/iqballx?tab=repositories
author_linkedin: https://www.linkedin.com/in/iqbaalm/
funding_url: https://github.com/open-webui
version: 2.0.0
"""
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any
import os
from open_webui.apps.webui.models.files import Files
import uuid
import logging
from openai import OpenAI
import time
# Set up logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
SYSTEM_PROMPT_BUILD_CHARTS = """
Objective:
Your goal is to read the query, extract the data, choose the appropriate chart to present the data, and produce the HTML to display it.
Steps:
1. Read and Examine the Query:
• Understand the user’s question and identify the data provided.
2. Analyze the Data:
• Examine the data in the query to determine the appropriate chart type (e.g., bar chart, pie chart, line chart) for effective visualization.
3. Generate HTML:
• Create the HTML code to present the data using the selected chart format.
4. Handle No Data Situations:
• If there is no data in the query or the data cannot be presented as a chart, generate a humorous or funny HTML response indicating that the data cannot be presented.
5. Calibrate the chart scale based on the data:
• based on the data try to make the scale of the chart as readable as possible.
Key Considerations:
- Your output should only include HTML code, without any additional text.
- Generate only HTML. Do not include any additional words or explanations.
- Make to remove any character other non alpha numeric from the data.
- is the generated HTML Calibrate the chart scale based on the data for eveything to be readable.
- Generate only html code , nothing else , only html.
Example1 :
'''
Interactive Chart
'''
Example2:
'''
Collaborateurs par Métier/Fonction
'''
2. No Data or Unchartable Data:
'''
We''re sorry, but your data can''t be charted.
Maybe try feeding it some coffee first?
'''
"""
USER_PROMPT_GENERATE_HTML = """
Giving this query {Query} generate the necessary html qurty.
"""
# Initialize OpenAI client
class FileData(BaseModel):
id: str
filename: str
meta: Dict[str, Any]
path: str
class Action:
class Valves(BaseModel):
show_status: bool = Field(
default=True, description="Show status of the action."
)
html_filename: str = Field(
default="json_visualizer.html",
description="Name of the HTML file to be created or retrieved.",
)
OPENIA_KEY: str = Field(
default="",
description="key to consume OpenIA interface like LLM for example a litellm key.",
)
OPENIA_URL: str = Field(
default="",
description="Host where to consume the OpenIA interface like llm",
)
MODEL_NAME: str = Field(
default="llama3.1:latest",
description="model name",
)
def __init__(self):
self.valves = self.Valves()
self.openai = None
self.html_content = """
"""
def create_or_get_file(self, user_id: str, json_data: str) -> str:
filename = str(int(time.time() * 1000)) + self.valves.html_filename
directory = "action_embed"
logger.debug(f"Attempting to create or get file: {filename}")
# Check if the file already exists
existing_files = Files.get_files()
for file in existing_files:
if (
file.filename == f"{directory}/{user_id}/{filename}"
and file.user_id == user_id
):
logger.debug(f"Existing file found. Updating content.")
# Update the existing file with new JSON data
self.update_html_content(file.meta["path"], json_data)
return file.id
# If the file doesn''t exist, create it
base_path = os.path.join("uploads", directory)
os.makedirs(base_path, exist_ok=True)
file_path = os.path.join(base_path, filename)
logger.debug(f"Creating new file at: {file_path}")
self.update_html_content(file_path, json_data)
file_id = str(uuid.uuid4())
meta = {
"source": file_path,
"title": "Modern JSON Visualizer",
"content_type": "text/html",
"size": os.path.getsize(file_path),
"path": file_path,
}
# Create a new file entry
file_data = FileData(
id=file_id,
filename=f"{directory}/{user_id}/{filename}",
meta=meta,
path=file_path,
)
new_file = Files.insert_new_file(user_id, file_data)
logger.debug(f"New file created with ID: {new_file.id}")
return new_file.id
def update_html_content(self, file_path: str, html_content: str):
with open(file_path, "w", encoding="utf-8") as f:
f.write(html_content)
logger.debug(f"HTML content updated at: {file_path}")
async def action(
self,
body: dict,
__user__=None,
__event_emitter__=None,
__event_call__=None,
) -> Optional[dict]:
logger.debug(f"action:{__name__} started")
await __event_emitter__(
{
"type": "status",
"data": {
"description": "Analysing Data",
"done": False,
},
}
)
if __event_emitter__:
try:
original_content = body["messages"][-1]["content"]
self.openai = OpenAI(
api_key=self.valves.OPENIA_KEY, base_url=self.valves.OPENIA_URL
)
response = self.openai.chat.completions.create(
model=self.valves.MODEL_NAME,
messages=[
{"role": "system", "content": SYSTEM_PROMPT_BUILD_CHARTS},
{
"role": "user",
"content": USER_PROMPT_GENERATE_HTML.format(
Query=body["messages"][-1]["content"]
),
},
],
max_tokens=1000,
n=1,
stop=None,
temperature=0.7,
)
html_content = response.choices[0].message.content
print("-----------------------------")
# print html content in pretty and readable format
# this is to help debug
print(html_content)
print("-----------------------------")
user_id = __user__["id"]
print({"user": __user__})
file_id = self.create_or_get_file(user_id, html_content)
# Create the HTML embed tag
html_embed_tag = f"{{{{HTML_FILE_ID_{file_id}}}}}"
# Append the HTML embed tag to the original content on a new line
body["messages"][-1][
"content"
] = f"{original_content}\n\n{html_embed_tag}"
await __event_emitter__(
{
"type": "status",
"data": {
"description": "Visualise the chart",
"done": True,
},
}
)
logger.debug(f" objects visualized")
except Exception as e:
error_message = f"Error visualizing JSON: {str(e)}"
logger.error(f"Error: {error_message}")
body["messages"][-1]["content"] += f"\n\nError: {error_message}"
if self.valves.show_status:
await __event_emitter__(
{
"type": "status",
"data": {
"description": "Error Visualizing JSON",
"done": True,
},
}
)
logger.debug(f"action:{__name__} completed")
return body