ea4all/utils/utils.py

import ea4all.src.ea4all_apm.graph as e4a
import ea4all.src.ea4all_vqa.graph as e4v
import ea4all.src.ea4all_gra.graph as e4t
import ea4all.src.shared.utils as e4u
from ea4all.src.ea4all_gra.configuration import AgentConfiguration as gra
from ea4all.src.shared import vectorstore
from ea4all.src.shared.configuration import BaseConfiguration
from ea4all.src.ea4all_indexer.configuration import IndexConfiguration
from ea4all.src.ea4all_indexer.graph import indexer_graph

from langchain_community.document_loaders import ConfluenceLoader
from langchain_core.messages import ChatMessage
from langsmith import Client

import uuid
import os
import time
import pandas as pd
import gradio as gr

class UIUtils:
    #ea4all-about
    @staticmethod
    def ea4all_about(show_api=False):
        readme = e4u.load_mock_content(e4u.CFG.EA4ALL_ABOUT)
        return readme

    #vqa_chatbot (ChatInterface -> Chatbot)
    @staticmethod
    def add_message(message, history):
        if message["text"] is not None:
            history.append({"role": "user", "content": message["text"]})
        
        if len(message['files']) > 0:
            history.append({"role": "user", "content": {"path": message['files'][-1]}})
        
        return (
            gr.MultimodalTextbox(value=message, interactive=True, placeholder="Upload a diagram and type your message..."),
            history
        )

    #Upload & clear business requirement
    @staticmethod
    def load_dbr(file):
        return file.decode()

#Load demo business requirements
def init_dbr():
    # Open the file in read mode ('r')
    with open(e4u._join_paths(BaseConfiguration.ea4all_store, gra.dbr_mock), 'r') as file:
        # Read the contents of the file
        contents = file.read()
    return contents

def init_df(show_api=False):
    return vectorstore.apm_dataframe_loader(e4u._join_paths(BaseConfiguration.ea4all_store, IndexConfiguration.apm_catalogue))

#load core-architecture image
#fix the issue with gr.Image(path) inside a docker containder
def get_image(_image):
    #from PIL import Image
    # Load an image
    image = e4u._join_paths(BaseConfiguration.ea4all_images,_image)
    print(f"Full path: {image}")

    return image

def ea4all_confluence(show_api=False):

    #Confluence API Key
    confluence_api_key = os.environ['CONFLUENCE_API_KEY']

    loader = ConfluenceLoader(
        url="https://learnitall.atlassian.net/wiki", username="learn-it-all@outlook.com", api_key=confluence_api_key,
        space_key="~71202000cd55f36336455f8c07afa1860ba810", 
        include_attachments=False, limit=10,
        keep_markdown_format=True
    )

    documents = loader.load()

    data = {
            "title": [doc.metadata["title"] for doc in documents],
            "source": [doc.metadata["source"] for doc in documents],
            "page_content": [doc.page_content for doc in documents],
        }

    df = pd.DataFrame(data)

    return df        

def filter_page(page_list, title):
    x =  page_list[page_list["title"] == title]
    return x.iloc[0]['page_content']

#get LLM response user's feedback
def get_user_feedback(evt: gr.SelectData, request:gr.Request):
    ##{evt.index} {evt.value} {evt._data['liked']}
    try:     
        uuid_str = os.environ["EA4ALL_" + e4u.get_user_identification(request).replace(".","_")]
        gr.Info("Thanks for your feedback - run_id: " + uuid_str)
        run_id = uuid.UUID(uuid_str)
        client = Client()
        client.create_feedback(
            run_id,
            key="feedback-key",
            score= 1.0 if evt._data['liked'] == True else 0,
            comment=str(evt.value)
        )
    except Exception as e:
        gr.Warning(f"Couldn't capture a feedback: {e}")

#Set initial state of apm, llm and capture user-ip
async def ea4all_agent_init(request:gr.Request):

    agentic_qna_desc="""Hi,
        improve effieciency, knowledge sharing, and get valuable insights from your IT landscape using natural language.
        As an Enterprise Architect Agentic Companion I can answer questions related to Enterprise Architecture, Technology, plus the following IT Landscape sample dataset: """

    #capture user IP address
    #ea4all_user = e4u.get_user_identification(request)
    gr.Info("Thank you for using the EA4ALL Agentic MCP Server!")

    # Set initial landscape vectorstore

    #await indexer_graph.ainvoke(input={"docs":[]}, config=config)

    #set chatbot description w/ user apm columns
    df = vectorstore.apm_dataframe_loader(e4u._join_paths(BaseConfiguration.ea4all_store, IndexConfiguration.apm_catalogue))
    columns_string = ', '.join(df.columns)
    apm_columns = agentic_qna_desc + columns_string

    prompt=ChatMessage(role='assistant', content='Hi, I am your Architect Copilot! How can I help you today?')

    page_list = ea4all_confluence()

    #Load gradio.dataframe with Portfolio sample dataset
    pmo_df = pd.read_csv("ea4all/ea4all_store/ea4all-portfolio-management.csv")

    dbr_text = init_dbr()

    return (
        apm_columns, 
        [{"role": "system", "content": "You are a helpful assistant."}],
        [{"role": "system", "content": "You are a helpful assistant."}],
        [{"role": "system", "content": "You are a helpful assistant."}],
        gr.DataFrame(value=df),
        gr.DataFrame(value=pmo_df),
        dbr_text
    )

#authentication
def ea4all_login(username, password):
    return (username==password)

#TABS & Reference Architecture look-and-feel control
def off_dbrtext():
    return gr.TextArea(visible=False), gr.Tab(visible=False), gr.Tab(visible=False)

def on_dbrtext(file):
    if file:
        return gr.TextArea(visible=True)
    return gr.TextArea(visible=False)

def unload_dbr():
    return gr.TextArea(visible=False)

def get_question_diagram_from_example(value) -> list:
    """
    Extracts the question and diagram from the selected example.
    """
    if value:
        return [value['text'], value['files'][-1]] if 'files' in value else [value['text'], None]
    return ["", None]

def on_image_update(image: gr.Image) -> gr.Image:
    """
    Callback to handle image updates.
    """
    visible = True if image is not None else False

    return gr.Image(visible=visible)