Event Hook¶

Event Hook主要用于在事件发生时，触发一些操作。例如，当一个任务完成时，可以触发一个事件，然后在事件发生时执行一些操作。

TFRobot会在scenes_events对应的场景发生时，触发相应的事件。用户可以通过实现EventHook类，来实现自己的事件处理逻辑。

目前定义EventHook主要有两种方法：

EventHook类：用户可以继承EventHook类，然后实现on_event方法，来处理事件。
tf_event_hook装饰器：用户可以使用tf_event_hook装饰器，来装饰一个函数，然后在函数中处理事件。

基于`EventHook`类的定义方式¶

用户可以导入from tfrobot.telemetry.hook import BaseHook，然后继承BaseHook类，实现execute方法，来处理事件。

from opentelemetry.util import types
from tfrobot.telemetry.hook import BaseHook
from tfrobot.telemetry.hook import HookManager

class MyHook(BaseHook):
    def execute(self, name: str, context: types.Attributes) -> None:
        print(f"事件名称 {name} 被触发。相关上下文如下: {context}")

HookManager().register_hook(MyHook(), scene="all")

Notes: 如果使用类方式进行定义，需要显式地将自定义Hook类注册到HookManager中，以便在事件发生时调用。如果需要注册多个场景，需要多次注册。

基于`tf_event_hook`装饰器的定义方式¶

用户可以导入from tfrobot.telemetry.hook import tf_event_hook，然后使用tf_event_hook装饰器，来装饰一个函数，然后在函数中处理事件。

from opentelemetry.util import types
from tfrobot.telemetry.hook import tf_event_hook

@tf_event_hook(scenes="all")
def my_hook(name: str, context: types.Attributes) -> None:
    print(f"事件名称 {name} 被触发。相关上下文如下: {context}")

注意装饰器可以接收一个scenes参数。scenes参数是一个列表或者单个字符串，用于指定装饰器装饰的函数，只有在指定的场景发生时，才会触发事件。场景没有特殊限制，只需要是字符串即可，目前已经支持的场景见：scenes_events。

Context 上下文¶

在Hook开发过程中，所有的上下文信息存储于context参数中。针对不同的场景（Scene）其属性字段不尽相同。具体可以参考如下：

基础上下文¶

所有上下文的公共字段如下：

class BaseEventContext(BaseModel):
    scene: Literal["Robot", "Brain", "Drive", "Memory", "MemoryStore", "Chain", "Tool", "LLM"]
    entity_id: str  # 发送事件的实体ID
    desc: Optional[str] = None  # 发送事件的描述信息。Desc往往与scene+action有关，并非具体内容，旨在帮助程序员了解信息而非用户
    info: str  # 事件的详细信息，一般可以用于打印日志，旨在帮助用户了解具体信息而非程序员，每个事件的info都不同

Robot 机器人场景上下文¶

class RobotEventContext(BaseEventContext):
    scene: Literal["Robot"] = "Robot"
    tool_names: Optional[str] = None  # 机器人中所有工具的名称列表
    tool_info_str: Optional[str] = None  # 机器人中所有工具的信息字符串
    msg_content: Optional[str] = None  # 机器人工作上下文中收到的消息
    msg_from_user: Optional[str] = None  # 机器人工作上下文中消息的发送者
    msg_from_user_id: Optional[str] = None  # 机器人工作上下文中消息的发送者ID
    msg_id: Optional[str] = None  # 机器人工作上下文中消息的ID

Brain 大脑场景上下文¶

class BrainEventContext(BaseEventContext):
    scene: Literal["Brain"] = "Brain"
    msg_content: Optional[str] = None  # 机器人工作上下文中收到的消息
    msg_from_user: Optional[str] = None  # 机器人工作上下文中消息的发送者
    msg_from_user_id: Optional[str] = None  # 机器人工作上下文中消息的发送者ID
    msg_id: Optional[str] = None  # 机器人工作上下文中消息的ID

Chain 链场景上下文¶

class ChainEventContext(BaseEventContext):
    scene: Literal["Chain"] = "Chain"
    current_input: Optional[str] = None

LLM 大语言模型场景上下文¶

class LLMEventContext(BaseEventContext):
    scene: Literal["LLM"]
    streaming: bool  # 是否是流式的输出
    # 当前上下文里调用大模型的token_usage. 如果是整形，代表本次请求总计的Token数量。如果是字典，则会分开请求Token，
    # 响应Token与总计的Token。如果是流式输出，中间响应时有可能不会填充此字段。
    token_usage: Annotated[Optional[int | dict], BeforeValidator(format_token_usage)] = None
    user_input: Optional[str] = None
    llm_model_name: str  # Pydantic 不建议使用 model_ 开头的属性进行命名。

    @field_serializer("prompt", "token_usage")
    def serialize_dict_attr(self, value: Optional[str | int | dict]) -> str | int | None:
        if isinstance(value, dict):
            return json.dumps(value, indent=2)
        return value

Memory 记忆场景上下文¶

class MemoryEventContext(BaseEventContext):
    scene: Literal["Memory"]
    query: Optional[str] = None
    max_token: Optional[str] = None
    top_k: Optional[int] = None

MemoryStore 记忆存储场景上下文¶

class MemoryStoreEventContext(BaseEventContext):
    scene: Literal["MemoryStore"]
    key: Optional[str] = None

Drive 驱动场景上下文¶

class DriveEventContext(BaseEventContext):
    scene: Literal["Drive"]

Tool 工具场景上下文¶

class ToolEventContext(BaseEventContext):
    scene: Literal["Tool"]
    tool_name: str
    tool_return: str = (
        "null"  # 工具的返回值，经过json.dumps进行序列化。如果是BeforeToolRun，则此值为"None" 因为SpanContext不允许None值。故而使用字符串None代替。需要重点注意与处理。
    )
    tool_version: Optional[str] = None
    tool_description: Optional[str] = None

如果需要使用工具的全量返回数据，比如需要使用ToolReturn元数据信息，则需要将tool_return字段进行反序列化。

如果仅仅是使用工具返回的文本信息，比如打印日志，或者给用户渲染返回。则可以直接使用info字段（此字段在基类中定义）。

Note:

由于tool_return字段是经过json.dumps序列化的，所以在使用时需要进行反序列化。同时需要注意，如果工具返回的不是标准ToolReturn类型，而是一个迭代器类型，比如Iterator[ToolReturn]则不会进行序列化。则这里记录"null"