| from itertools import count | |
| from typing import Tuple | |
| from lagent.schema import AgentMessage | |
| class Hook: | |
| def before_agent( | |
| self, | |
| agent, | |
| message: Tuple[AgentMessage], | |
| session_id: int, | |
| ): | |
| pass | |
| def after_agent( | |
| self, | |
| agent, | |
| message: AgentMessage, | |
| session_id: int, | |
| ): | |
| pass | |
| def before_action( | |
| self, | |
| executor, | |
| message: AgentMessage, | |
| session_id: int, | |
| ): | |
| pass | |
| def after_action( | |
| self, | |
| executor, | |
| message: AgentMessage, | |
| session_id: int, | |
| ): | |
| pass | |
| class RemovableHandle: | |
| _id_iter = count(0) | |
| def __init__(self, hooks_dict): | |
| self.hooks_dict = hooks_dict | |
| self.id = next(self._id_iter) | |
| def remove(self): | |
| del self.hooks_dict[self.id] | |