I’m building an AI agent app using Langchain, but because of the 30sec timeout limit, I need to run my agent via a background task. Find the cloned app below. However, I’m have a hard time returning the langchain agent’s work from the background task. I’ve tried using GPT4 to assist in the programming, but it’s just not working.
I’d like the agent to run in the background (as research can take up to 5mins), and then once it’s done, populate the ‘company profile textbox’, or even the backend table’s ‘company profile’ value. The agent runs and the background task is completed,
I have a check_task_status_timer in the Company’s front-end ‘Invisible components’ box, as wanted to check if the task was done.
SOLVED
FRONT END
def company_research_button_click(self, **event_args):
with anvil.server.no_loading_indicator:
# This method should handle the UI logic
print("Research button clicked")
# Start the progress bar with a small value
self.progress_bar.value = 5
company_name = self.company_name_input.text
company_url = self.company_url_input.text
# Launch the background task and store the task ID
self.task_id = anvil.server.call('launch_company_summary', company_name, company_url)
print("Task ID:", self.task_id)
# Loop to check the status of the background task
while True:
# Increment the counter
self.counter += 1
# Update the progress bar (example: increment by 10 each tick)
self.progress_bar.value = min(self.counter * 5 + 5, 100)
# Check if the background task is complete
task_status = anvil.server.call('get_task_status', self.task_id)
print("Task status:", task_status)
if task_status is not None and task_status == "completed":
# Get the result of the background task
company_context = anvil.server.call('get_task_result', self.task_id)
# Update the textbox with the result
print("Company Context:", company_context)
self.company_profile_textbox.text = company_context
break # Exit the loop
# Sleep for 1 second before checking again
time.sleep(0.1)
BACK END:
@anvil.server.callable
def launch_company_summary(company_name, company_url):
# Launch the background task
task = anvil.server.launch_background_task('company_summary', company_name, company_url)
# Return the task ID
return task.get_id()
@anvil.server.background_task
def company_summary(company_name, company_url):
print("Background task started for researching company:", company_name)
# Here, you should write the code that uses the company_name and company_url
# to research the company and generate a context. For example:
llm_agents = ChatOpenAI(temperature=0.5, model_name='gpt-4', openai_api_key=openai_api_key)
....etc
You’ll generally get better advice if you say what trouble you’re having. Are you getting an error message of some sort? Is it crashing? Just not displaying anything?
You need a background task to run the 5 minute job. As it works, you can update a datatable row with two columns: status (text) and value (string or, maybe, simple object).
The client calls a server function
The server function adds one row to a datatable, starts the background task passing the row id to the task and returns the row id to the client
The client uses a timer to call a second server function to check the status every few seconds
The background task does its job while updating the status column of the row, and updating the value when it’s done
The client keeps updating a label with the status, until the status is ‘complete’ and it updates the text box with the value
I have seen a few other threads with people using LLMs. Here’s one that would be helpful:
In short, you can take one of many approaches but the simplest is to use a while loop in your client code that is based on the status of a background task.
As the suggestion above. The simplest solution would be to add a timer to you hidden component, set it’s interval to 0 first.
Call your background task:
#start the task and assign the task retrived from background to self.task
self.task = anvil.server.call('function_server')
self.timer_1.interval = 1#set timer
example how to set the timer function to check the status every interval tick.
def timer_1_tick(self, **event_args):
"""This method is called Every [interval] seconds. Does not trigger if [interval] is 0."""
with anvil.server.no_loading_indicator:
# Show progress
state = self.task.get_state()
if self.task.is_running():
#optional to get some progress data from the task
progress = state.get('progress', 0)
self.button_1.text = f"Exporting {progress}%"
else:
#function finished retrieve data. for e.g. self.task.get_return_value()
Hey! Thanks so much. I actually posted it early by accident. The reason I didn’t share it was because I couldn’t find any confirmation that my Secrets wouldn’t be accessed by sharing the clone. I’m assuming not, but couldn’t find it.
A while loop with a sleep(0.1) may work, but I wouldn’t do it this way. The Anvil way to do that is to use a timer with the interval set to 0.1.
I also wouldn’t set the timer to 0.1 for a task that lasts 5 minutes. If the task lasts 5 minutes, I would set the timer to a few seconds. There is not point in (1) hammering the server with one call every 0.1 seconds and (2) trying to get the resolution down to 0.1 seconds, when the user will wait 5 minutes anyway.
Also, I find it cleaner to put the code for each task in its place:
When the user clicks the button, the task starts. This is the job for a button’s click handler.
Every few seconds I want to know if the job is done. This is the job for a timer’s tick handler.
I have plenty of apps like this and I set my timer to 2 to 5 seconds. Some just show the status of a long running job, some check whether other users have modified the document one is working on and ask to reload the document, etc. I’ve tried with 0.1 seconds and with 3 seconds, and I haven’t noticed any difference from the user experience point of view. Instead I have noticed that when the timer is set to 0.1 I risk transaction conflicts when a few round trips per second are triggered by multiple users using the app at the same time.