Workflow Session State

1shopping_workflow = Workflow(
2    name="Shopping List Workflow",
3    steps=[manage_items_step, view_list_step],
4    session_state={"shopping_list": []},  # Initialize with structured data
5)

1from kern.agent import Agent
2from kern.db.sqlite import SqliteDb
3from kern.models.openai import OpenAIResponses
4from kern.workflow.step import Step
5from kern.workflow.workflow import Workflow
6from kern.run import RunContext
7
8db = SqliteDb(db_file="tmp/workflow.db")
9
10
11# Define tools to manage a shopping list in workflow session state
12def add_item(run_context: RunContext, item: str) -> str:
13    """Add an item to the shopping list in workflow session state.
14
15    Args:
16        item (str): The item to add to the shopping list
17    """
18    if not run_context.session_state:
19        run_context.session_state = {}
20
21    # Check if item already exists (case-insensitive)
22    existing_items = [
23        existing_item.lower() for existing_item in run_context.session_state["shopping_list"]
24    ]
25    if item.lower() not in existing_items:
26        run_context.session_state["shopping_list"].append(item)
27        return f"Added '{item}' to the shopping list."
28    else:
29        return f"'{item}' is already in the shopping list."
30
31
32def remove_item(run_context: RunContext, item: str) -> str:
33    """Remove an item from the shopping list in workflow session state.
34
35    Args:
36        item (str): The item to remove from the shopping list
37    """
38    if not run_context.session_state:
39        run_context.session_state = {}
40
41    if len(run_context.session_state["shopping_list"]) == 0:
42        return f"Shopping list is empty. Cannot remove '{item}'."
43
44    # Find and remove item (case-insensitive)
45    shopping_list = run_context.session_state["shopping_list"]
46    for i, existing_item in enumerate(shopping_list):
47        if existing_item.lower() == item.lower():
48            removed_item = shopping_list.pop(i)
49            return f"Removed '{removed_item}' from the shopping list."
50
51    return f"'{item}' not found in the shopping list."
52
53
54def remove_all_items(run_context: RunContext) -> str:
55    """Remove all items from the shopping list in workflow session state."""
56    if not run_context.session_state:
57        run_context.session_state = {}
58
59    run_context.session_state["shopping_list"] = []
60    return "Removed all items from the shopping list."
61
62
63def list_items(run_context: RunContext) -> str:
64    """List all items in the shopping list from workflow session state."""
65    if not run_context.session_state:
66        run_context.session_state = {}
67
68    if len(run_context.session_state["shopping_list"]) == 0:
69        return "Shopping list is empty."
70
71    items = run_context.session_state["shopping_list"]
72    items_str = "\n".join([f"- {item}" for item in items])
73    return f"Shopping list:\n{items_str}"
74
75
76# Create agents with tools that use workflow session state
77shopping_assistant = Agent(
78    name="Shopping Assistant",
79    model=OpenAIResponses(id="gpt-5.2"),
80    tools=[add_item, remove_item, list_items],
81    instructions=[
82        "You are a helpful shopping assistant.",
83        "You can help users manage their shopping list by adding, removing, and listing items.",
84        "Always use the provided tools to interact with the shopping list.",
85        "Be friendly and helpful in your responses.",
86    ],
87)
88
89list_manager = Agent(
90    name="List Manager",
91    model=OpenAIResponses(id="gpt-5.2"),
92    tools=[list_items, remove_all_items],
93    instructions=[
94        "You are a list management specialist.",
95        "You can view the current shopping list and clear it when needed.",
96        "Always show the current list when asked.",
97        "Confirm actions clearly to the user.",
98    ],
99)
100
101# Create steps
102manage_items_step = Step(
103    name="manage_items",
104    description="Help manage shopping list items (add/remove)",
105    agent=shopping_assistant,
106)
107
108view_list_step = Step(
109    name="view_list",
110    description="View and manage the complete shopping list",
111    agent=list_manager,
112)
113
114# Create workflow with workflow_session_state
115shopping_workflow = Workflow(
116    name="Shopping List Workflow",
117    db=db,
118    steps=[manage_items_step, view_list_step],
119    session_state={"shopping_list": []},
120)
121
122if __name__ == "__main__":
123    # Example 1: Add items to the shopping list
124    print("=== Example 1: Adding Items ===")
125    shopping_workflow.print_response(
126        input="Please add milk, bread, and eggs to my shopping list."
127    )
128    print("Workflow session state:", shopping_workflow.get_session_state())
129
130    # Example 2: Add more items and view list
131    print("\n=== Example 2: Adding More Items ===")
132    shopping_workflow.print_response(
133        input="Add apples and bananas to the list, then show me the complete list."
134    )
135    print("Workflow session state:", shopping_workflow.get_session_state())
136
137    # Example 3: Remove items
138    print("\n=== Example 3: Removing Items ===")
139    shopping_workflow.print_response(
140        input="Remove bread from the list and show me what's left."
141    )
142    print("Workflow session state:", shopping_workflow.get_session_state())
143
144    # Example 4: Clear the entire list
145    print("\n=== Example 4: Clearing List ===")
146    shopping_workflow.print_response(
147        input="Clear the entire shopping list and confirm it's empty."
148    )
149    print("Final workflow session state:", shopping_workflow.get_session_state())

1from kern.run import RunContext
2
3def evaluator_function(step_input: StepInput, run_context: RunContext):
4    return run_context.session_state["test"] == "test_1"
5
6condition_step = Condition(
7    name="condition_step",
8    evaluator=evaluator_function,
9    steps=[step_1, step_2],
10)

1from kern.run import RunContext
2
3def selector_function(step_input: StepInput, run_context: RunContext):
4    return run_context.session_state["test"] == "test_1"
5
6router_step = Router(
7    name="router_step",
8    selector=selector_function,
9    choices=[step_1, step_2],
10)