Bases: ABC
The base processor for the session. A session consists of multiple rounds of conversation with the user, completing a task.
At each round, the HostAgent and AppAgent interact with the user and the application with the processor.
Each processor is responsible for processing the user request and updating the HostAgent and AppAgent at a single step in a round.
Initialize the processor.
Parameters: |
-
context
(Context )
–
The context of the session.
-
agent
(BasicAgent )
–
The agent who executes the processor.
|
Source code in agents/processors/basic.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68 | def __init__(self, agent: BasicAgent, context: Context) -> None:
"""
Initialize the processor.
:param context: The context of the session.
:param agent: The agent who executes the processor.
"""
self._context = context
self._agent = agent
self.photographer = PhotographerFacade()
self.control_inspector = ControlInspectorFacade(BACKEND)
self._prompt_message = None
self._status = None
self._response = None
self._cost = 0
self._control_label = None
self._control_text = None
self._response_json = {}
self._memory_data = MemoryItem()
self._results = None
self._question_list = []
self._agent_status_manager = self.agent.status_manager
self._is_resumed = False
self._action = None
self._plan = None
self._control_log = {
"control_class": None,
"control_type": None,
"control_automation_id": None,
}
self._total_time_cost = 0
self._time_cost = {}
|
action: str
property
writable
agent: BasicAgent
property
app_root: str
property
writable
Get the application root.
application_process_name: str
property
writable
Get the application process name.
Returns: |
-
str
–
The application process name.
|
application_window: UIAWrapper
property
writable
context: Context
property
control_reannotate: List[str]
property
writable
Get the control reannotation.
Returns: |
-
List[str]
–
The control reannotation.
|
control_text: str
property
writable
Get the active application.
cost: float
property
writable
Get the cost of the processor.
Returns: |
-
float
–
The cost of the processor.
|
host_message: List[str]
property
writable
name: str
property
Get the name of the processor.
Returns: |
-
str
–
The name of the processor.
|
plan: str
property
writable
Get the plan of the agent.
prev_plan: List[str]
property
Get the previous plan.
Returns: |
-
List[str]
–
The previous plan of the agent.
|
previous_subtasks: List[str]
property
writable
Get the previous subtasks.
question_list: List[str]
property
writable
request_logger: str
property
round_cost: float
property
writable
round_step: int
property
writable
round_subtask_amount: int
property
Get the round subtask amount.
Returns: |
-
int
–
The round subtask amount.
|
session_cost: float
property
writable
session_step: int
property
writable
status: str
property
writable
Get the status of the processor.
Returns: |
-
str
–
The status of the processor.
|
subtask: str
property
writable
capture_screenshot()
abstractmethod
Capture the screenshot.
Source code in agents/processors/basic.py
| @abstractmethod
def capture_screenshot(self) -> None:
"""
Capture the screenshot.
"""
pass
|
error_log(response_str, error)
Error handler for the session.
Source code in agents/processors/basic.py
671
672
673
674
675
676
677
678
679
680
681
682
683 | def error_log(self, response_str: str, error: str) -> None:
"""
Error handler for the session.
"""
log = json.dumps(
{
"step": self.session_step,
"status": self._agent_status_manager.ERROR.value,
"response": response_str,
"error": error,
}
)
self.logger.info(log)
|
execute_action()
abstractmethod
Execute the action.
Source code in agents/processors/basic.py
| @abstractmethod
def execute_action(self) -> None:
"""
Execute the action.
"""
pass
|
general_error_handler()
Error handler for the general error.
Source code in agents/processors/basic.py
693
694
695
696
697
698
699
700
701
702 | def general_error_handler(self) -> None:
"""
Error handler for the general error.
"""
error_trace = traceback.format_exc()
utils.print_with_color(f"Error Occurs at {self.name}", "red")
utils.print_with_color(str(error_trace), "red")
utils.print_with_color(self._response, "red")
self.error_log(self._response, str(error_trace))
self._status = self._agent_status_manager.ERROR.value
|
get_control_info()
abstractmethod
Get the control information.
Source code in agents/processors/basic.py
| @abstractmethod
def get_control_info(self) -> None:
"""
Get the control information.
"""
pass
|
get_prompt_message()
abstractmethod
Get the prompt message.
Source code in agents/processors/basic.py
| @abstractmethod
def get_prompt_message(self) -> None:
"""
Get the prompt message.
"""
pass
|
get_response()
abstractmethod
Get the response from the LLM.
Source code in agents/processors/basic.py
| @abstractmethod
def get_response(self) -> None:
"""
Get the response from the LLM.
"""
pass
|
is_confirm()
Check if the process is confirm.
Returns: |
-
bool
–
The boolean value indicating if the process is confirm.
|
Source code in agents/processors/basic.py
651
652
653
654
655
656
657
658
659 | def is_confirm(self) -> bool:
"""
Check if the process is confirm.
:return: The boolean value indicating if the process is confirm.
"""
self.agent.status = self.status
return self.status == self._agent_status_manager.CONFIRM.value
|
is_error()
Check if the process is in error.
Returns: |
-
bool
–
The boolean value indicating if the process is in error.
|
Source code in agents/processors/basic.py
619
620
621
622
623
624
625
626 | def is_error(self) -> bool:
"""
Check if the process is in error.
:return: The boolean value indicating if the process is in error.
"""
self.agent.status = self.status
return self.status == self._agent_status_manager.ERROR.value
|
is_paused()
Check if the process is paused.
Returns: |
-
bool
–
The boolean value indicating if the process is paused.
|
Source code in agents/processors/basic.py
628
629
630
631
632
633
634
635
636
637
638
639 | def is_paused(self) -> bool:
"""
Check if the process is paused.
:return: The boolean value indicating if the process is paused.
"""
self.agent.status = self.status
return (
self.status == self._agent_status_manager.PENDING.value
or self.status == self._agent_status_manager.CONFIRM.value
)
|
is_pending()
Check if the process is pending.
Returns: |
-
bool
–
The boolean value indicating if the process is pending.
|
Source code in agents/processors/basic.py
641
642
643
644
645
646
647
648
649 | def is_pending(self) -> bool:
"""
Check if the process is pending.
:return: The boolean value indicating if the process is pending.
"""
self.agent.status = self.status
return self.status == self._agent_status_manager.PENDING.value
|
llm_error_handler()
Error handler for the LLM error.
Source code in agents/processors/basic.py
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721 | def llm_error_handler(self) -> None:
"""
Error handler for the LLM error.
"""
error_trace = traceback.format_exc()
log = json.dumps(
{
"step": self.session_step,
"prompt": self._prompt_message,
"status": str(error_trace),
}
)
utils.print_with_color(
"Error occurs when calling LLM: {e}".format(e=str(error_trace)), "red"
)
self.request_logger.info(log)
self._status = self._agent_status_manager.ERROR.value
return
|
log(response_json)
Set the result of the session, and log the result.
result: The result of the session.
response_json: The response json.
return: The response json.
Source code in agents/processors/basic.py
661
662
663
664
665
666
667
668
669 | def log(self, response_json: dict) -> None:
"""
Set the result of the session, and log the result.
result: The result of the session.
response_json: The response json.
return: The response json.
"""
self.logger.info(json.dumps(response_json))
|
log_save()
Save the log.
Source code in agents/processors/basic.py
243
244
245
246
247
248
249
250
251 | def log_save(self) -> None:
"""
Save the log.
"""
self._memory_data.set_values_from_dict(
{"total_time_cost": self._total_time_cost}
)
self.log(self._memory_data.to_dict())
|
method_timer(func)
classmethod
Decorator to calculate the time cost of the method.
Parameters: |
-
func
–
The method to be decorated.
|
Source code in agents/processors/basic.py
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172 | @classmethod
def method_timer(cls, func):
"""
Decorator to calculate the time cost of the method.
:param func: The method to be decorated.
:return: The decorated method.
"""
@wraps(func)
def wrapper(self, *args, **kwargs):
start_time = time.time()
result = func(self, *args, **kwargs)
end_time = time.time()
self._time_cost[func.__name__] = end_time - start_time
return result
return wrapper
|
parse_response()
abstractmethod
Parse the response.
Source code in agents/processors/basic.py
| @abstractmethod
def parse_response(self) -> None:
"""
Parse the response.
"""
pass
|
print_step_info()
abstractmethod
Print the step information.
Source code in agents/processors/basic.py
| @abstractmethod
def print_step_info(self) -> None:
"""
Print the step information.
"""
pass
|
process()
Process a single step in a round.
The process includes the following steps:
1. Print the step information.
2. Capture the screenshot.
3. Get the control information.
4. Get the prompt message.
5. Get the response.
6. Update the cost.
7. Parse the response.
8. Execute the action.
9. Update the memory.
10. Update the step and status.
11. Save the log.
Source code in agents/processors/basic.py
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133 | def process(self) -> None:
"""
Process a single step in a round.
The process includes the following steps:
1. Print the step information.
2. Capture the screenshot.
3. Get the control information.
4. Get the prompt message.
5. Get the response.
6. Update the cost.
7. Parse the response.
8. Execute the action.
9. Update the memory.
10. Update the step and status.
11. Save the log.
"""
start_time = time.time()
# Step 1: Print the step information.
self.print_step_info()
# Step 2: Capture the screenshot.
self.capture_screenshot()
# Step 3: Get the control information.
self.get_control_info()
# Step 4: Get the prompt message.
self.get_prompt_message()
# Step 5: Get the response.
self.get_response()
if self.is_error():
return
# Step 6: Update the context.
self.update_cost()
# Step 7: Parse the response, if there is no error.
self.parse_response()
if self.is_error() or self.is_paused():
# If the session is pending, update the step and memory, and return.
if self.is_pending():
self.update_step()
self.update_memory()
return
# Step 8: Execute the action.
self.execute_action()
# Step 9: Update the memory.
self.update_memory()
# Step 10: Update the status.
self.update_status()
self._total_time_cost = time.time() - start_time
# Step 11: Save the log.
self.log_save()
|
resume()
Resume the process of action execution after the session is paused.
Source code in agents/processors/basic.py
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154 | def resume(self) -> None:
"""
Resume the process of action execution after the session is paused.
"""
self._is_resumed = True
# Step 1: Execute the action.
self.execute_action()
# Step 2: Update the memory.
self.update_memory()
# Step 3: Update the status.
self.update_status()
# Step 4: Update the step.
self.update_step()
self._is_resumed = False
|
string2list(string)
staticmethod
Convert a string to a list of string if the input is a string.
Source code in agents/processors/basic.py
723
724
725
726
727
728
729
730
731
732
733 | @staticmethod
def string2list(string: Any) -> List[str]:
"""
Convert a string to a list of string if the input is a string.
:param string: The string.
:return: The list.
"""
if isinstance(string, str):
return [string]
else:
return string
|
update_cost()
Update the cost.
Source code in agents/processors/basic.py
261
262
263
264
265
266
267 | def update_cost(self) -> None:
"""
Update the cost.
"""
self.round_cost += self.cost
self.session_cost += self.cost
|
update_memory()
abstractmethod
Update the memory of the Agent.
Source code in agents/processors/basic.py
| @abstractmethod
def update_memory(self) -> None:
"""
Update the memory of the Agent.
"""
pass
|
update_status()
Update the status of the session.
Source code in agents/processors/basic.py
230
231
232
233
234
235
236
237
238
239
240
241 | def update_status(self) -> None:
"""
Update the status of the session.
"""
self.agent.step += 1
self.agent.status = self.status
if self.status != self._agent_status_manager.FINISH.value:
time.sleep(configs["SLEEP_TIME"])
self.round_step += 1
self.session_step += 1
|