Refactor the Chatbot class to have a compisition relationship with APICall instead of inheritance. Implement and test chatbot component by writing the functions to interact with the API in the generated app script

Author: sayalaruano

report_config_micw2graph.yaml

@@ -82,10 +82,11 @@ sections:
             file_format: "csv"
             delimiter: ","
             caption: "This is the edge list of the network"
-  - title: "APICall test"
+  - title: "ChatBot test"
     subsections:
       - title: "Simple test"
         components:
-          - title: "APICall test"
-            component_type: "apicall"
-            api_url: "https://jsonplaceholder.typicode.com/todos/1"
+          - title: "ChatBot test"
+            component_type: "chatbot"
+            api_url: "http://localhost:11434/api/chat"
+            model: "llama3.2"

vuegen/main.py

@@ -11,8 +11,8 @@
     report_config = load_yaml_config(config_path)
 
     # Define logger suffix based on report type and name
-    report_name = report_config['report'].get('name')
-    logger_suffix = f"{report_type}_report_{report_name}"
+    report_title = report_config['report'].get('title')
+    logger_suffix = f"{report_type}_report_{report_title}"
 
     # Initialize logger
     logger = get_logger(f"{logger_suffix}")

vuegen/report.py

@@ -404,68 +404,84 @@ def make_api_request(self, method: str, request_body: Optional[dict] = None) ->
             self.logger.error(f"API request failed: {e}")
             return None
 
-class ChatBot(APICall):
+class ChatBot(Component):
     """
-    A specialized component for creating a ChatBot.
+    A component for creating a ChatBot that interacts with an API.
+    This component uses an APICall instance to send requests to the chatbot API and receive responses.
 
     Attributes
     ----------
     model : str
-        The language model to use.
+        The language model to use for the chatbot.
+    api_call : APICall
+        An instance of the APICall class used to interact with the API for fetching chatbot responses.
+    headers : Optional[dict]
+        Headers to include in the API request (default is None).
+    params : Optional[dict]
+        Query parameters to include in the API request (default is None).
     """
     def __init__(self, title: str, logger: logging.Logger, api_url: str, model: str,
                  caption: str = None, headers: Optional[dict] = None, params: Optional[dict] = None):
-        super().__init__(title = title, logger = logger, api_url = api_url, 
-                         caption=caption, headers=headers, params=params)
+        super().__init__(title=title, logger=logger, component_type=ComponentType.CHATBOT, caption=caption)
         self.model = model
+        self.api_call = APICall(
+            title=title, 
+            logger=logger, 
+            api_url=api_url, 
+            caption=caption,
+            headers=headers, 
+            params=params 
+        )
 
     def get_chatbot_answer(self, prompt: str) -> dict:
         """
-        Sends a RAG query and retrieves the resulting documents.
+        Sends a query to the chatbot API and retrieves the resulting response. This method constructs 
+        the request body for the chatbot, sends the request to the API, and parses the response.
 
         Parameters
         ----------
         prompt : str
-            The prompt for asking the chatbot.
+            The prompt or message to send to the chatbot for a response.
 
         Returns
         -------
-        parsed_response : dict
-            The chabtbot answer.
+        dict
+            The parsed response from the chatbot API, containing the chatbot's answer.
         """
         request_body = self._generate_query(prompt)
-        response = self.make_api_request(method="POST", request_body=request_body)
+        response = self.api_call.make_api_request(method="POST", request_body=request_body)  # Delegate API call
         if response:
-            self.logger.info(f"Request successful")
+            self.logger.info("Request successful")
         else:
-            self.logger.warning("Nothing retreived.")
+            self.logger.warning("No response retrieved.")
         parsed_response = self._parse_api_response(response)
         return parsed_response
 
+
     def _generate_query(self, messages: str) -> dict:
         """
-        Constructs the request body for a question to the chatbot.
+        Constructs the request body for the chatbot query.
 
         Parameters
         ----------
         messages : str
-            The messages for retrieval.
+            The messages to send to the chatbot.
 
         Returns
         -------
-        request_body : dict
-            The request body for the question to the chatbot.
+        dict
+            The constructed request body for the chatbot query.
         """
         self.logger.info(f"Generating request body for message: {messages}")
         return {
             "model": self.model,
             "messages": messages,
             "stream": True
         }
-    
+
     def _parse_api_response(self, response: dict) -> dict:
         """
-        Extracts and processes data from the API response.
+        Processes and extracts relevant data from the API response.
 
         Parameters
         ----------
@@ -474,8 +490,8 @@ def _parse_api_response(self, response: dict) -> dict:
 
         Returns
         -------
-        output : dict
-            The extracted data from the response.
+        dict
+            The processed response, including the role and content of the assistant's reply.
         """
         output = ""
         for line in response.iter_lines():
@@ -590,8 +606,8 @@ class ReportView(ABC):
         The name of the view.
     report : Report
         The report that this ABC is associated with.
-    columns : List[str], optional
-        Column names used in the report view ABC (default is None).
+    report_type : ReportType
+        The report type. It should be one of the values of the ReportType Enum.
     
     """
     def __init__(self, report: 'Report', report_type: 'ReportType'):

vuegen/report_generator.py

@@ -1,8 +1,8 @@
 from streamlit_reportview import StreamlitReportView
 from quarto_reportview import QuartoReportView
 from metadata_manager import MetadataManager
-from report import ReportType
 from utils import assert_enum_value
+from report import ReportType
 import logging
 
 def get_report(config: dict, report_type: str, logger: logging.Logger) -> None:

vuegen/streamlit_reportview.py

@@ -15,7 +15,7 @@ class StreamlitReportView(r.WebAppReportView):
     REPORT_MANAG_SCRIPT = 'report_manager.py'
 
     def __init__(self, report: r.Report, report_type: r.ReportType):
-        super().__init__(report=report, report_type = report_type)
+        super().__init__(report = report, report_type = report_type)
 
     def generate_report(self, output_dir: str = SECTIONS_DIR, static_dir: str = STATIC_FILES_DIR) -> None:
         """
@@ -479,34 +479,86 @@ def _generate_apicall_content(self, apicall) -> List[str]:
 
     def _generate_chatbot_content(self, chatbot) -> List[str]:
         """
-        Generate content for a Markdown component.
+        Generate content for a ChatBot component.
 
         Parameters
         ----------
         chatbot : ChatBot
-            The chatbot component to generate content for.
+            The ChatBot component to generate content for.
 
         Returns
         -------
         list : List[str]
-            The list of content lines for the chatbot.
+            The list of content lines for the ChatBot.
         """
         chatbot_content = []
 
         # Add title
         chatbot_content.append(self._format_text(text=chatbot.title, type='header', level=4, color='#2b8cbe'))
-        try:
-            apicall_response = chatbot.get_chatbot_answer()
-            chatbot_content.append(f"""st.write({apicall_response})\n""")
-        except Exception as e:
-            self.report.logger.error(f"Error generating content for APICall: {chatbot.title}. Error: {str(e)}")
-            raise
-        
+
+        # Chatbot logic for embedding in the web application
+        chatbot_content.append(f"""
+def generate_query(messages):
+    response = requests.post(
+        "{chatbot.api_call.api_url}",
+        json={{"model": "{chatbot.model}", "messages": messages, "stream": True}},
+    )
+    response.raise_for_status()
+    return response               
+
+def parse_api_response(response):
+    try:
+        output = ""
+        for line in response.iter_lines():
+            body = json.loads(line)
+            if "error" in body:
+                raise Exception(f"API error: {{body['error']}}")
+            if body.get("done", False):
+                return {{"role": "assistant", "content": output}}
+            output += body.get("message", {{}}).get("content", "")
+    except Exception as e:
+        return {{"role": "assistant", "content": f"Error while processing API response: {{str(e)}}"}}
+
+def response_generator(msg_content):
+    for word in msg_content.split():
+        yield word + " "
+        time.sleep(0.1)
+    yield "\\n"
+
+# Chatbot interaction in the app
+if 'messages' not in st.session_state:
+    st.session_state['messages'] = []
+
+# Display chat history
+for message in st.session_state['messages']:
+    with st.chat_message(message['role']):
+        st.write(message['content'])
+
+# Handle new input from the user
+if prompt := st.chat_input("Enter your prompt here:"):
+    # Add user's question to the session state                           
+    st.session_state.messages.append({{"role": "user", "content": prompt}})
+    with st.chat_message("user"):
+        st.write(prompt)
+    
+    # Retrieve question and generate answer
+    combined = "\\n".join(msg["content"] for msg in st.session_state.messages if msg["role"] == "user")
+    messages = [{{"role": "user", "content": combined}}]
+    with st.spinner('Generating answer...'):                       
+        response = generate_query(messages)
+        parsed_response = parse_api_response(response)
+                               
+    # Add the assistant's response to the session state and display it
+    st.session_state.messages.append(parsed_response)
+    with st.chat_message("assistant"):
+        st.write_stream(response_generator(parsed_response["content"]))
+    """)
+
         # Add caption if available
         if chatbot.caption:
-            chatbot_content.append(self._format_text(text=chatbot.caption, type='caption', text_align="left")) 
+            chatbot_content.append(self._format_text(text=chatbot.caption, type='caption', text_align="left"))
 
-        self.report.logger.info(f"Successfully generated content for APICall: '{chatbot.title}'")
+        self.report.logger.info(f"Successfully generated content for ChatBot: '{chatbot.title}'")
         return chatbot_content
 
     def _generate_component_imports(self, component: r.Component) -> List[str]:
@@ -531,7 +583,8 @@ def _generate_component_imports(self, component: r.Component) -> List[str]:
                 r.PlotType.ALTAIR: ['import json', 'import altair as alt'],
                 r.PlotType.PLOTLY: ['import json']
             },
-            'dataframe': ['import pandas as pd']
+            'dataframe': ['import pandas as pd'],
+            'chatbot': ['import time', 'import json', 'import requests']
         }
 
         component_type = component.component_type
@@ -544,6 +597,8 @@ def _generate_component_imports(self, component: r.Component) -> List[str]:
                 component_imports.extend(components_imports['plot'][plot_type])
         elif component_type == r.ComponentType.DATAFRAME:
             component_imports.extend(components_imports['dataframe'])
+        elif component_type == r.ComponentType.CHATBOT:
+            component_imports.extend(components_imports['chatbot'])
 
         # Return the list of import statements
         return component_imports     

vuegen/utils.py

@@ -12,7 +12,7 @@ def check_path(filepath: str) -> bool:
     """
     Checks if the given file or folder path exists.
 
-    PARAMETERS
+    Parameters
     ---------
     filepath : str
         The file or folder path to check.
@@ -155,12 +155,12 @@ def load_yaml_config(file_path: str) -> dict:
     """
     Load a YAML configuration file and return its contents as a dictionary.
 
-    PARAMETERS
+    Parameters
     ----------
     file_path : str
         The path to the YAML configuration file.
 
-    RETURNS
+    Returns
     -------
     config : dict
         The contents of the YAML file as a dictionary.
@@ -199,7 +199,7 @@ def get_basename(fname: None | str = None) -> str:
     -----
     - basename of given filepath or the current file the function is executed
 
-    EXAMPLES
+    Examples
     -----
     1)
     >>> get_basename()
@@ -223,17 +223,17 @@ def get_time(incl_time: bool = True, incl_timezone: bool = True) -> str:
     """
     Gets current date, time (optional) and timezone (optional) for file naming
 
-    PARAMETERS
+    Parameters
     -----
     - incl_time (bool): whether to include timestamp in the string
     - incl_timezone (bool): whether to include the timezone in the string
 
-    RETURNS
+    Returns
     -----
     - fname (str): includes date, timestamp and/or timezone
         connected by '_' in one string e.g. yyyyMMdd_hhmm_timezone
 
-    EXAMPLES
+    Examples
     -----
     1)
     >>> get_time()
@@ -286,12 +286,12 @@ def generate_log_filename(folder: str = "logs", suffix: str = "") -> str:
     """
     Creates log file name and path
 
-    PARAMETERS
+    Parameters
     -----
     folder (str): name of the folder to put the log file in
     suffix (str): anything else you want to add to the log file name
 
-    RETURNS
+    Returns
     -----
     log_filepath (str): the file path to the log file
     """
@@ -313,18 +313,18 @@ def init_log(filename: str, display: bool = False, logger_id: str | None = None)
     - Keeps a log record file of the python application, with option to
         display in stdout
 
-    PARAMETERS
+    Parameters
     -----
     - filename (str): filepath to log record file
     - display (bool): whether to print the logs to whatever standard output
     - logger_id (str): an optional identifier for yourself,
         if None then defaults to 'root'
 
-    RETURNS
+    Returns
     -----
     - logger object
 
-    EXAMPLE
+    Examples
     -----
     >>> logger = init_log('logs/tmp.log', display=True)
     >>> logger.info('Loading things')