CEL Expressions

pip install cel-python

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.workflow import Condition, Step, Workflow

urgent_handler = Agent(
    name="Urgent Handler",
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions="Handle urgent requests with high priority.",
)

normal_handler = Agent(
    name="Normal Handler",
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions="Handle normal requests thoroughly.",
)

workflow = Workflow(
    name="CEL Input Routing",
    steps=[
        Condition(
            name="Urgent Check",
            evaluator='input.contains("urgent")',
            steps=[Step(name="Handle Urgent", agent=urgent_handler)],
            else_steps=[Step(name="Handle Normal", agent=normal_handler)],
        ),
    ],
)

workflow.print_response("This is an urgent request - please help immediately!")

classifier = Agent(
    name="Classifier",
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions="Classify the request as TECHNICAL or GENERAL. Respond with one word.",
)

workflow = Workflow(
    name="Classify and Route",
    steps=[
        Step(name="Classify", agent=classifier),
        Condition(
            name="Route by Classification",
            evaluator='previous_step_content.contains("TECHNICAL")',
            steps=[Step(name="Technical Help", agent=technical_agent)],
            else_steps=[Step(name="General Help", agent=general_agent)],
        ),
    ],
)

workflow = Workflow(
    name="Priority Routing",
    steps=[
        Condition(
            name="Priority Gate",
            evaluator="additional_data.priority > 5",
            steps=[Step(name="High Priority", agent=high_priority_agent)],
            else_steps=[Step(name="Low Priority", agent=low_priority_agent)],
        ),
    ],
)

workflow.print_response("Review this report.", additional_data={"priority": 8})

workflow = Workflow(
    name="Retry Logic",
    steps=[
        Step(name="Increment", executor=increment_retry),
        Condition(
            name="Retry Check",
            evaluator="session_state.retry_count <= 3",
            steps=[Step(name="Attempt", agent=retry_agent)],
            else_steps=[Step(name="Give Up", agent=fallback_agent)],
        ),
    ],
    session_state={"retry_count": 0},
)

from agno.workflow import Loop, Step, Workflow

workflow = Workflow(
    name="Iteration Limit",
    steps=[
        Loop(
            name="Writing Loop",
            max_iterations=10,
            end_condition="current_iteration >= 2",
            steps=[Step(name="Write", agent=writer)],
        ),
    ],
)

editor = Agent(
    name="Editor",
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions="Edit the text. When polished, include the word DONE at the end.",
)

workflow = Workflow(
    name="Content Keyword Loop",
    steps=[
        Loop(
            name="Editing Loop",
            max_iterations=5,
            end_condition='last_step_content.contains("DONE")',
            steps=[Step(name="Edit", agent=editor)],
        ),
    ],
)

workflow = Workflow(
    name="Compound Exit",
    steps=[
        Loop(
            name="Research Loop",
            max_iterations=5,
            end_condition="all_success && current_iteration >= 2",
            steps=[
                Step(name="Research", agent=researcher),
                Step(name="Review", agent=reviewer),
            ],
        ),
    ],
)

from agno.workflow import Step, Workflow
from agno.workflow.router import Router

workflow = Workflow(
    name="Style Router",
    steps=[
        Router(
            name="Analysis Style",
            selector="session_state.preferred_handler",
            choices=[
                Step(name="Detailed Analyst", agent=detailed_agent),
                Step(name="Brief Analyst", agent=brief_agent),
            ],
        ),
    ],
    session_state={"preferred_handler": "Brief Analyst"},
)

workflow = Workflow(
    name="Media Router",
    steps=[
        Router(
            name="Media Router",
            selector='input.contains("video") ? "Video Handler" : "Image Handler"',
            choices=[
                Step(name="Video Handler", agent=video_agent),
                Step(name="Image Handler", agent=image_agent),
            ],
        ),
    ],
)

workflow = Workflow(
    name="Index Router",
    steps=[
        Router(
            name="Analysis Router",
            selector='input.contains("quick") ? step_choices[0] : step_choices[1]',
            choices=[
                Step(name="Quick Analysis", agent=quick_analyzer),
                Step(name="Detailed Analysis", agent=detailed_analyzer),
            ],
        ),
    ],
)