I am trying to register a callback, after I have an instance of my class.
I am doing this because, I can overwrite the same function on different pages to provide different data for the callback with the same function. This works well as long as I don’t use a cancle button and switch states while runnning.
I am not exactly sure why it does not work. My file looks like this
import uuid
import copy
import time
import os
import plotly.graph_objects as go
from pypdf import PdfReader
from dash_drag_grid import DashboardItemResponsive
import dash_mantine_components as dmc
from dash import (
dcc,
html,
callback,
Input,
ALL,
MATCH,
Output,
State,
no_update,
clientside_callback,
)
from dash_app.config.ids import GeneralIDs
from dash.exceptions import PreventUpdate
from flask import current_app
from dash_app.components.chat import single_message_card
from dash_app.services.NotificationProvider import NotificationProvider
from dash_app.services.agents.lightweight.agent_light import create_agent_light
from dash_app.utils.functions import get_icon, stringify_id
from dash_app.services.agents.inferenceAgent.executer.members.pdfAnalyst import (
make_retriever,
)
from langchain_core.messages import HumanMessage, AIMessage
notify = NotificationProvider()
cache = current_app.cache
class QueryTool:
def __init__(
self,
members: dict = {},
page_id: str = MATCH,
with_reflection=False,
):
"""Creating a QueryTool instance for a page
1. Init the QueryTool and add members. The format is a dict:
members={
"ResearchAnalyst": "provides information about content of disclosures. Cannot provid insigts about data or graph! Needs specific information such as titles to provide useful information. It does not have information about graphs!",
"SummerizeAnalyst": "When asked provides an informative summery of the collected insights",
}
The above dictionary are also the default members.
Possible members are:
1. ResearchAnalyst
2. SummerizeAnalyst
3. GraphAnalyst
4. DataAnalyst
.and the page_id
2. If you use a DataAnalyst, overwrite the get_dataframes method(self, current_selection).
The inputs are the same, you can only change whats happening inside the function. The function needs to return a list of dataframes.
InstaceXYZ.get_dataframes = types.MethodType(get_dataframes, InstaceXYZ)
3. Add external data sources with `bot.register_data_source()`. Currently only figures from dcc.Graph supported
A dictionary needs to be passed
{"figure": TradeActivityGraph.ids.chart}
Where the key is the value of the instance and the value is the id function which has one argument, page_id.
If you have none, just execute function without passing parameters
3. Bot.get_layout() at the point where you want to add the QueryTool
"""
self.members = members
self.page_id = page_id
self.datasources = False
self.with_reflection = with_reflection
class ids:
user_chat_input = lambda page_id, component: {
"component": component,
"name": "user_chat_input",
"page_id": page_id,
}
submit_chat = lambda page_id, component: {
"component": component,
"name": "submit_chat",
"page_id": page_id,
}
cancle_request = lambda page_id, component: {
"component": component,
"name": "cancle_request",
"page_id": page_id,
}
answer = lambda page_id, component: {
"component": component,
"name": "answer",
"page_id": page_id,
}
data_inputs = lambda page_id, component: {
"component": component,
"name": "data_inputs",
"page_id": page_id,
}
ids = ids
def get_dataframes(self, current_selection):
"""Overwrite this method with the quering of the data you want to have
Args:
current_selection (dict): The only available input to the queries
Returns:
list: list of dataframes you want to make available
"""
return []
def get_pdf(self, current_selection):
"""Overwrite this method with the query to get the pdf to analyse
Args:
current_selection (dict): The only available input to the queries
Returns:
list: with the document
"""
return []
@staticmethod
def extract_text_from_pdf(reader):
text = ""
for page in reader.pages:
text += page.extract_text()
return text
@staticmethod
def add_graph(figure):
image_path = f"{current_app.config['temp_path']}/fig_{str(uuid.uuid4())}.png"
fig_object = go.Figure(figure)
fig_object.write_image(image_path)
return image_path
@staticmethod
def remove_image(image_path):
try:
os.remove(image_path)
except:
pass
@staticmethod
def get_signatures(signature_type, signiture_dict, page_id=MATCH):
signatures = {}
signatures = {
key: signature_type(source(page_id), key)
for key, source in signiture_dict.items()
}
return signatures
def register_data_source(self):
@callback(
Output(
QueryTool.ids.submit_chat(self.page_id, "QueryTool"),
"loading",
allow_duplicate=True,
),
Input(QueryTool.ids.submit_chat(self.page_id, "QueryTool"), "n_clicks"),
State(QueryTool.ids.user_chat_input(self.page_id, "QueryTool"), "value"),
State(QueryTool.ids.data_inputs(self.page_id, "QueryTool"), "value"),
State(GeneralIDs.socket("base"), "socketId"),
State(GeneralIDs.page_selection_store(self.page_id), "data"),
prevent_initial_call=True,
running=[
(
Output(
QueryTool.ids.submit_chat(self.page_id, "QueryTool"), "disabled"
),
True,
False,
),
(
Output(
QueryTool.ids.cancle_request(self.page_id, "QueryTool"),
"disabled",
),
False,
True,
),
],
cancel=[
Input(
QueryTool.ids.cancle_request(self.page_id, "QueryTool"), "n_clicks"
)
],
)
def ask_question(n_clicks, question, data_inputs, socket_id, current_selection):
if n_clicks:
data_inputs = [] if data_inputs is None else data_inputs
pdf_text = []
card_message = single_message_card(message=question, role="user")
notify.send_chat_message(socket_id, card_message, event="queryTool")
time.sleep(0.5)
current_app.logger.info(f"QueryTool: User input is - {question}")
try:
if "pdf" in data_inputs:
pdf = self.get_pdf(current_selection)
pdf_text = self.extract_text_from_pdf(pdf[0])
chattbot = create_agent_light(
data_inputs=data_inputs, pdf_text=pdf_text
)
chat_history = cache.get(f"query_{str(socket_id)}")
if chat_history is None:
chat_history = []
inputs = {"task": question, "chat_history": chat_history}
card_answer = single_message_card(role="agent")
notify.send_chat_message(socket_id, card_answer, event="queryTool")
agent_answer = ""
for token in chattbot.stream(inputs):
agent_answer += token.content
notify.send_chat_message(
socket_id, token.content, event="queryToolToken"
)
time.sleep(0.2)
chat_history.append(HumanMessage(content=question))
chat_history.append(AIMessage(content=agent_answer))
cache.set(f"query_{str(socket_id)}", chat_history)
except Exception as e:
current_app.logger.error(f"QueryTool: {e}")
if e == "Could not parse function call: 'function_call":
message = (
"There was an internal error of the agent. Please ask again"
)
elif (
e
== "Recursion limit of 30 reachedwithout hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key."
):
message = "You hit the recusion limit and the agent is likely stuck in a loop. Make your question more specific"
else:
message = "There was an error with the API"
notify.send_socket(
to=socket_id,
id=str(uuid.uuid4()),
title="We are sorry!",
message=message,
type="error",
)
return no_update
raise PreventUpdate
def get_layout(self, **props):
derived_kwargs = props.copy()
derived_kwargs["x"] = derived_kwargs.get("x", 0)
derived_kwargs["y"] = derived_kwargs.get("y", 0)
derived_kwargs["w"] = derived_kwargs.get("w", 4)
derived_kwargs["h"] = derived_kwargs.get("h", 5)
derived_kwargs["id"] = derived_kwargs.get("id", "QueryTool")
derived_kwargs["inToolbox"] = derived_kwargs.get("inToolbox", True)
derived_kwargs["defaultName"] = derived_kwargs.get("defaultName", "Query Tool")
component = "QueryTool"
page_id = self.page_id
if page_id is None:
page_id = str(uuid.uuid4())
return DashboardItemResponsive(
children=[
html.Div(
[
dmc.Stack(
children=[],
id=getattr(self.ids, "answer")(page_id, component),
style={
"overflow": "auto",
"gap": "10px",
"overflow-y": "auto",
"flex-grow": "1",
},
),
dmc.Space(h=10),
dmc.Center(
[
dmc.Stack(
[
dmc.CheckboxGroup(
id=getattr(self.ids, "data_inputs")(
page_id, component
),
label="Select your sources",
orientation="horizontal",
offset="md",
mb=10,
children=[
dmc.Checkbox(
label="Disclosure", value="pdf"
),
],
),
dmc.Textarea(
placeholder="Send a message",
id=getattr(self.ids, "user_chat_input")(
page_id, component
),
autosize=True,
style={
"width": "100%",
},
spellCheck=False,
minRows=2,
radius="xl",
rightSectionWidth=80,
rightSection=dmc.Group(
[
dmc.Tooltip(
label="Send message",
children=dmc.ActionIcon(
get_icon(
"ph:arrow-up-bold",
height=20,
),
size="lg",
variant="transparent",
id=getattr(
self.ids, "submit_chat"
)(page_id, component),
n_clicks=None,
mb=10,
radius=10,
style={
"margin": 0,
"color": "black",
},
),
),
dmc.Tooltip(
label="Stop chat",
children=dmc.ActionIcon(
get_icon(
"material-symbols:stop-outline",
height=20,
),
size="lg",
variant="transparent",
id=getattr(
self.ids,
"cancle_request",
)(page_id, component),
color="red",
n_clicks=None,
mb=10,
radius=10,
style={"margin": 0},
),
),
]
),
),
],
style={"width": "100%"},
)
],
style={
"padding": "10px 10px 10px 10px",
"min-height": "unset",
},
),
],
style={
"padding": "10px",
"display": "flex",
"flex-direction": "column",
"height": "100%",
},
)
],
x=derived_kwargs["x"],
y=derived_kwargs["y"],
w=derived_kwargs["w"],
h=derived_kwargs["h"],
id=derived_kwargs["id"],
inToolbox=derived_kwargs["inToolbox"],
defaultName=derived_kwargs["defaultName"],
)
clientside_callback(
"""(n_clicks) => {
if( n_clicks){return ""}
return dash_clientside.no_update
}""",
Output(ids.user_chat_input(MATCH, "QueryTool"), "value", allow_duplicate=True),
Input(ids.submit_chat(MATCH, "QueryTool"), "n_clicks"),
prevent_initial_call=True,
)
# JavaScript callback to scroll to the bottom of the messages div
clientside_callback(
"""(textAreaId, submitButtonId, chatLogId) => {
window.setupMutationObserver(stringifyId(textAreaId), stringifyId(submitButtonId), stringifyId(chatLogId));
return dash_clientside.no_update
}""",
Output(ids.submit_chat(MATCH, "QueryTool"), "children"),
Input(ids.user_chat_input(MATCH, "QueryTool"), "id"),
Input(ids.submit_chat(MATCH, "QueryTool"), "id"),
Input(ids.answer(MATCH, "QueryTool"), "id"),
)
Maybe you have a better way to do this? If you need I can also create an MRE.
Thank you
