Function demonstrate_workflows

examples/agno/agno_workflow_setup.py:87–123

Demonstrate workflow capabilities with caching. This function shows: - How to create and use custom workflows - The performance benefits of caching - Session state persistence - Response streaming

()

def demonstrate_workflows():
    """
    Demonstrate workflow capabilities with caching.

    This function shows:
    - How to create and use custom workflows
    - The performance benefits of caching
    - Session state persistence
    - Response streaming
    """

    tracer = agentops.start_trace(trace_name="Agno Workflow Setup Demonstration")
    try:
        workflow = CacheWorkflow()

        response: Iterator[RunResponse] = workflow.run(message="Tell me a joke.")

        pprint_run_response(response, markdown=True, show_time=True)

        response: Iterator[RunResponse] = workflow.run(message="Tell me a joke.")

        pprint_run_response(response, markdown=True, show_time=True)

        agentops.end_trace(tracer, end_state="Success")

    except Exception:
        agentops.end_trace(tracer, end_state="Error")

    # Let's check programmatically that spans were recorded in AgentOps
    print("\n" + "=" * 50)
    print("Now let's verify that our LLM calls were tracked properly...")
    try:
        agentops.validate_trace_spans(trace_context=tracer)
        print("\n✅ Success! All LLM spans were properly recorded in AgentOps.")
    except agentops.ValidationError as e:
        print(f"\n❌ Error validating spans: {e}")
        raise


demonstrate_workflows()