import jsonfrom src.agent.capability import MatchingCapabilityfrom src.main import AgentWorkerfrom src.agent.capability_worker import CapabilityWorker# =============================================================================# LOOP TEMPLATE# For interactive Abilities with multi-turn conversations.# Pattern: Greet → Loop (Listen → Process → Respond) → Exit on command## Replace the processing logic inside the while loop with your own.# =============================================================================# Words that will exit the Ability and return to normal flowEXIT_WORDS = {"stop", "exit", "quit", "done", "cancel", "bye", "goodbye", "leave"}class LoopTemplateCapability(MatchingCapability): worker: AgentWorker = None capability_worker: CapabilityWorker = None # Do not change following tag of register capability #{{register capability}} def call(self, worker: AgentWorker): self.worker = worker self.capability_worker = CapabilityWorker(self) self.worker.session_tasks.create(self.run()) async def run(self): # Greet the user await self.capability_worker.speak( "I'm ready to help. Ask me anything, or say stop when you're done." ) while True: # Listen for user input user_input = await self.capability_worker.user_response() # Skip empty input if not user_input: continue # Check for exit commands if any(word in user_input.lower() for word in EXIT_WORDS): await self.capability_worker.speak("Goodbye!") break # --- YOUR LOGIC HERE --- # Process the input and generate a response. # This example uses the LLM, but you can do anything: # call APIs, play audio, run calculations, etc. response = self.capability_worker.text_to_text_response( f"Respond in one short sentence: {user_input}" ) await self.capability_worker.speak(response) # ALWAYS resume normal flow when the loop ends self.capability_worker.resume_normal_flow()
The core pattern is a while True loop that runs until an exit command is detected:
while True: # Listen for user input user_input = await self.capability_worker.user_response() # Check for exit if any(word in user_input.lower() for word in EXIT_WORDS): await self.capability_worker.speak("Goodbye!") break # Process and respond response = self.capability_worker.text_to_text_response( f"Respond in one short sentence: {user_input}" ) await self.capability_worker.speak(response)
You can maintain state across loop iterations using instance variables:
class QuizCapability(MatchingCapability): score: int = 0 questions_asked: int = 0 async def run(self): while True: # Ask question user_answer = await self.capability_worker.user_response() # Check answer and update state if self.check_answer(user_answer): self.score += 1 self.questions_asked += 1 # Continue or exit if self.questions_asked >= 10: await self.capability_worker.speak( f"Quiz complete! Your score: {self.score} out of 10" ) break
The template uses text_to_text_response() as a placeholder. Replace this with your own logic:
# Use the LLM to generate responsesresponse = self.capability_worker.text_to_text_response( f"Respond in one short sentence: {user_input}")await self.capability_worker.speak(response)