Class CacheWorkflow

examples/agno/agno_workflow_setup.py:32–84

A workflow that demonstrates intelligent caching capabilities. This workflow: - Caches agent responses to avoid redundant API calls - Maintains session state across multiple invocations - Provides instant responses for repeated queries - Reduces costs and improves performan

class CacheWorkflow(Workflow):
    """
    A workflow that demonstrates intelligent caching capabilities.

    This workflow:
    - Caches agent responses to avoid redundant API calls
    - Maintains session state across multiple invocations
    - Provides instant responses for repeated queries
    - Reduces costs and improves performance

    Use cases:
    - FAQ systems where questions repeat frequently
    - Development/testing to avoid repeated API calls
    - Systems with predictable query patterns
    """

    # Workflow metadata (descriptive, not functional)
    description: str = "A workflow that caches previous outputs for efficiency"

    # Initialize agents as workflow attributes
    # This agent will be used to generate responses when cache misses occur
    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():