-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathresult.py
More file actions
192 lines (155 loc) · 5.1 KB
/
Copy pathresult.py
File metadata and controls
192 lines (155 loc) · 5.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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
69
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
"""
Result classes for the Gopher Security MCP SDK.
Provides structured result objects with status, response, and metadata.
"""
from enum import Enum
from typing import Optional
class AgentResultStatus(Enum):
"""Status of an agent operation."""
SUCCESS = "success"
ERROR = "error"
TIMEOUT = "timeout"
class AgentResult:
"""
Result of an agent operation.
Contains the response, status, and optional metadata.
"""
def __init__(
self,
response: str,
status: AgentResultStatus,
iteration_count: int = 0,
tokens_used: int = 0,
error_message: Optional[str] = None,
) -> None:
"""
Initialize an AgentResult.
Args:
response: The agent's response text
status: The result status
iteration_count: Number of iterations performed
tokens_used: Number of tokens consumed
error_message: Error message if status is ERROR or TIMEOUT
"""
self._response = response
self._status = status
self._iteration_count = iteration_count
self._tokens_used = tokens_used
self._error_message = error_message
@property
def response(self) -> str:
"""Get the response text."""
return self._response
@property
def status(self) -> AgentResultStatus:
"""Get the result status."""
return self._status
@property
def iteration_count(self) -> int:
"""Get the number of iterations."""
return self._iteration_count
@property
def tokens_used(self) -> int:
"""Get the number of tokens used."""
return self._tokens_used
@property
def error_message(self) -> Optional[str]:
"""Get the error message if any."""
return self._error_message
def is_success(self) -> bool:
"""Check if the result is successful."""
return self._status == AgentResultStatus.SUCCESS
def is_error(self) -> bool:
"""Check if the result is an error."""
return self._status == AgentResultStatus.ERROR
def is_timeout(self) -> bool:
"""Check if the result is a timeout."""
return self._status == AgentResultStatus.TIMEOUT
@staticmethod
def success(
response: str, iteration_count: int = 1, tokens_used: int = 0
) -> "AgentResult":
"""
Create a successful result.
Args:
response: The response text
iteration_count: Number of iterations
tokens_used: Number of tokens used
Returns:
AgentResult with SUCCESS status
"""
return AgentResult(
response=response,
status=AgentResultStatus.SUCCESS,
iteration_count=iteration_count,
tokens_used=tokens_used,
)
@staticmethod
def error(message: str) -> "AgentResult":
"""
Create an error result.
Args:
message: The error message
Returns:
AgentResult with ERROR status
"""
return AgentResult(
response="",
status=AgentResultStatus.ERROR,
error_message=message,
)
@staticmethod
def timeout(message: str = "Operation timed out") -> "AgentResult":
"""
Create a timeout result.
Args:
message: The timeout message
Returns:
AgentResult with TIMEOUT status
"""
return AgentResult(
response="",
status=AgentResultStatus.TIMEOUT,
error_message=message,
)
@staticmethod
def builder() -> "AgentResultBuilder":
"""Create a new result builder."""
return AgentResultBuilder()
class AgentResultBuilder:
"""Builder for AgentResult."""
def __init__(self) -> None:
self._response: str = ""
self._status: AgentResultStatus = AgentResultStatus.SUCCESS
self._iteration_count: int = 0
self._tokens_used: int = 0
self._error_message: Optional[str] = None
def response(self, response: str) -> "AgentResultBuilder":
"""Set the response text."""
self._response = response
return self
def status(self, status: AgentResultStatus) -> "AgentResultBuilder":
"""Set the status."""
self._status = status
return self
def iteration_count(self, count: int) -> "AgentResultBuilder":
"""Set the iteration count."""
self._iteration_count = count
return self
def tokens_used(self, tokens: int) -> "AgentResultBuilder":
"""Set the tokens used."""
self._tokens_used = tokens
return self
def error_message(self, message: str) -> "AgentResultBuilder":
"""Set the error message."""
self._error_message = message
return self
def build(self) -> AgentResult:
"""Build the AgentResult."""
return AgentResult(
response=self._response,
status=self._status,
iteration_count=self._iteration_count,
tokens_used=self._tokens_used,
error_message=self._error_message,
)