Agentic User Input

1"""
2Agentic User Input
3=============================
4
5Human-in-the-Loop: Allowing users to provide input externally.
6"""
7
8from typing import Any, Dict, List
9
10from kern.agent import Agent
11from kern.db.sqlite import SqliteDb
12from kern.models.openai import OpenAIResponses
13from kern.tools import Toolkit
14from kern.tools.function import UserInputField
15from kern.tools.user_control_flow import UserControlFlowTools
16from kern.utils import pprint
17
18
19class EmailTools(Toolkit):
20    def __init__(self, *args, **kwargs):
21        super().__init__(
22            name="EmailTools", tools=[self.send_email, self.get_emails], *args, **kwargs
23        )
24
25    def send_email(self, subject: str, body: str, to_address: str) -> str:
26        """Send an email to the given address with the given subject and body.
27
28        Args:
29            subject (str): The subject of the email.
30            body (str): The body of the email.
31            to_address (str): The address to send the email to.
32        """
33        return f"Sent email to {to_address} with subject {subject} and body {body}"
34
35    def get_emails(self, date_from: str, date_to: str) -> list[dict[str, str]]:
36        """Get all emails between the given dates.
37
38        Args:
39            date_from (str): The start date (in YYYY-MM-DD format).
40            date_to (str): The end date (in YYYY-MM-DD format).
41        """
42        return [
43            {
44                "subject": "Hello",
45                "body": "Hello, world!",
46                "to_address": "test@test.com",
47                "date": date_from,
48            },
49            {
50                "subject": "Random other email",
51                "body": "This is a random other email",
52                "to_address": "john@doe.com",
53                "date": date_to,
54            },
55        ]
56
57
58# ---------------------------------------------------------------------------
59# Create Agent
60# ---------------------------------------------------------------------------
61agent = Agent(
62    model=OpenAIResponses(id="gpt-5-mini"),
63    tools=[EmailTools(), UserControlFlowTools()],
64    markdown=True,
65    db=SqliteDb(db_file="tmp/agentic_user_input.db"),
66)
67
68# ---------------------------------------------------------------------------
69# Run Agent
70# ---------------------------------------------------------------------------
71if __name__ == "__main__":
72    run_response = agent.run(
73        "Send an email with the body 'What is the weather in Tokyo?'"
74    )
75
76    # We use a while loop to continue the running until the agent is satisfied with the user input
77    while run_response.is_paused:
78        for requirement in run_response.active_requirements:
79            if requirement.needs_user_input:
80                input_schema: List[UserInputField] = requirement.user_input_schema  # type: ignore
81
82                for field in input_schema:
83                    # Get user input for each field in the schema
84                    field_type = field.field_type  # type: ignore
85                    field_description = field.description  # type: ignore
86
87                    # Display field information to the user
88                    print(f"\nField: {field.name}")  # type: ignore
89                    print(f"Description: {field_description}")
90                    print(f"Type: {field_type}")
91
92                    # Get user input
93                    if field.value is None:  # type: ignore
94                        user_value = input(f"Please enter a value for {field.name}: ")  # type: ignore
95                    else:
96                        print(f"Value: {field.value}")  # type: ignore
97                        user_value = field.value  # type: ignore
98
99                    # Update the field value
100                    field.value = user_value  # type: ignore
101
102        run_response = agent.continue_run(
103            run_id=run_response.run_id,
104            requirements=run_response.requirements,
105        )
106        if not run_response.is_paused:
107            pprint.pprint_run_response(run_response)
108            break
109
110    run_response = agent.run("Get me all my emails")
111
112    while run_response.is_paused:
113        for requirement in run_response.active_requirements:
114            if requirement.needs_user_input:
115                input_schema: Dict[str, Any] = requirement.user_input_schema  # type: ignore
116
117                for field in input_schema:
118                    # Get user input for each field in the schema
119                    field_type = field.field_type  # type: ignore
120                    field_description = field.description  # type: ignore
121
122                    # Display field information to the user
123                    print(f"\nField: {field.name}")  # type: ignore
124                    print(f"Description: {field_description}")
125                    print(f"Type: {field_type}")
126
127                    # Get user input
128                    if field.value is None:  # type: ignore
129                        user_value = input(f"Please enter a value for {field.name}: ")  # type: ignore
130                    else:
131                        print(f"Value: {field.value}")  # type: ignore
132                        user_value = field.value  # type: ignore
133
134                    # Update the field value
135                    field.value = user_value  # type: ignore
136
137        run_response = agent.continue_run(
138            run_id=run_response.run_id,
139            requirements=run_response.requirements,
140        )
141
142        if not run_response.is_paused:
143            pprint.pprint_run_response(run_response)
144            break

1# Clone and setup repo
2git clone https://github.com/kern-ai/kern.git
3cd kern/cookbook/02_agents/10_human_in_the_loop
4
5# Create and activate virtual environment
6./scripts/demo_setup.sh
7source .venvs/demo/bin/activate
8
9python agentic_user_input.py