|
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] |
|
|