Method run

examples/agno/agno_workflow_setup.py:55–84

Execute the workflow with caching logic. This method: 1. Checks if the response is already cached 2. Returns cached response immediately if found 3. Generates new response if not cached 4. Caches the new response for future use Args:

(self, message: str)

    agent = Agent(model=OpenAIChat(id="gpt-4o-mini"), description="General purpose agent for generating responses")

    def run(self, message: str) -> Iterator[RunResponse]:
        """
        Execute the workflow with caching logic.

        This method:
        1. Checks if the response is already cached
        2. Returns cached response immediately if found
        3. Generates new response if not cached
        4. Caches the new response for future use

        Args:
            message: The input query to process

        Yields:
            RunResponse: Streamed response chunks
        """
        logger.info(f"Checking cache for '{message}'")

        if self.session_state.get(message):
            logger.info(f"Cache hit for '{message}'")
            # Return cached response immediately (no API call needed)
            yield RunResponse(run_id=self.run_id, content=self.session_state.get(message))
            return

        logger.info(f"Cache miss for '{message}'")

        yield from self.agent.run(message, stream=True)

        self.session_state[message] = self.agent.run_response.content
        logger.info("Cached response for future use")


def demonstrate_workflows():