扩展场景与事件¶

TFRobot 内置了 Robot、Brain、Chain、LLM、Tool、Memory 等场景的遥测事件。但当你开发自定义组件——比如一个新的推理引擎、一个自定义存储后端，或者一个领域特定的处理器——你可能希望这些组件也能融入 TFRobot 的实时可观测体系，让 Hook 能够监听它们的运行状态。

本文档说明如何为自定义组件添加场景和事件。

你真的需要扩展吗¶

在动手之前，先判断你的需求属于哪种情况：

需求	推荐方案	工作量
在现有组件的 Hook 中添加自定义逻辑	编写 Hook，监听已有事件	低
在业务代码中手动发出事件	直接调用 `span.add_event("my_event", attributes)`	低
让自定义基类的子类自动获得遥测能力	完整扩展（本文档）	中

如果只是偶尔需要在某个方法中记录一个事件，直接用 span.add_event 即可，不需要走完整扩展流程。完整扩展适用于：你有一个基类，它的所有子类都应该自动被遥测系统包装。

完整扩展的四个步骤¶

以一个虚构的"Reasoner"（推理器）组件为例，演示完整扩展流程。

第一步：定义事件枚举¶

在 SpanEvent 枚举中添加你的事件。每个被监控的操作遵循四事件模型：BEFORE、AFTER、RAISE、ABORT。

# tfrobot/telemetry/context_schema.py

class SpanEvent(Enum):
    # ... 已有事件 ...

    # Reasoner 事件
    BEFORE_REASONER_INFER = "before_reasoner_infer"
    AFTER_REASONER_INFER = "after_reasoner_infer"
    REASONER_INFER_RAISE = "reasoner_infer_raise"    # 推理期间抛出异常
    REASONER_INFER_ABORT = "reasoner_infer_abort"    # 用户主动中止推理

四事件模型

BEFORE: 操作开始前触发，EventContext 包含输入信息
AFTER: 操作正常完成后触发，EventContext 包含返回值
RAISE: 操作抛出异常时触发，EventContext 包含异常信息
ABORT: 用户通过 TFUserInterruptError 主动中止时触发

RAISE 和 ABORT 的区分在 span_decorator 中自动完成：它通过 isinstance(e, TFUserInterruptError) 判断应该触发哪个事件。

第二步：定义 EventContext¶

创建一个继承 BaseEventContext 的 Pydantic 模型，声明你的场景专属字段。

# tfrobot/telemetry/context_schema.py

class ReasonerEventContext(BaseEventContext):
    scene: Literal["Reasoner"] = "Reasoner"
    input_query: str = ""               # 推理输入
    reasoning_steps: int = 0            # 推理步数
    reasoner_result: str = "null"       # 推理结果（JSON 字符串）

设计要点：

scene 必须是固定值。HookManager 根据这个字段路由事件到对应的 Hook。
复杂类型序列化为字符串。OpenTelemetry 的 attributes 只接受基本类型（str、int、float、bool），如果你的字段是字典或 Pydantic 模型，需要 json.dumps 后存为字符串。参考 LLMEventContext 的 llm_result 字段。
如果需要自定义序列化，使用 Pydantic 的 @field_serializer：

from pydantic import field_serializer

class ReasonerEventContext(BaseEventContext):
    scene: Literal["Reasoner"] = "Reasoner"
    step_details: dict = {}

    @field_serializer("step_details")
    @classmethod
    def serialize_steps(cls, v: dict) -> str:
        import json
        return json.dumps(v, ensure_ascii=False)

第三步：实现 decorate_span 和 _generate_context¶

这是核心步骤。你的基类需要：

在 __init_subclass__ 中调用 decorate_span()，让所有子类自动获得遥测能力
decorate_span 用 span_decorator 包装目标方法
_generate_context 构造 EventContext

from functools import partial
from typing import Any, ClassVar, Optional

from pydantic import ConfigDict
from typing_extensions import Unpack

from tfrobot.telemetry.context_schema import (
    SPAN_WRAP_KEY,
    SpanEvent,
    span_decorator,
)


class BaseReasoner:
    __register_key__: ClassVar[str] = "REASONER"

    def __init_subclass__(cls, **kwargs: Unpack[ConfigDict]) -> None:
        super().__init_subclass__(**kwargs)
        cls.decorate_span()

    @classmethod
    def decorate_span(cls) -> None:
        # 包装同步方法
        if hasattr(cls, "infer") and not getattr(cls.infer, "__is_tf_span_wrapped__", False):
            span_attr = {"scene": "Reasoner", "desc": "Reasoner infer method"}
            decorated = span_decorator(
                span_attr,
                SpanEvent.BEFORE_REASONER_INFER,
                SpanEvent.AFTER_REASONER_INFER,
                SpanEvent.REASONER_INFER_RAISE,
                SpanEvent.REASONER_INFER_ABORT,
                partial(cls._generate_context, desc="Before infer"),
                partial(cls._generate_context, desc="After infer"),
                # 第 8 个参数（可选）：异常上下文生成器
                # partial(cls._generate_exception_context, desc="Exception during infer"),
            )(cls.infer)
            setattr(cls, "infer", decorated)

        # 包装异步方法（如果有）
        if hasattr(cls, "async_infer") and not getattr(cls.async_infer, "__is_tf_span_wrapped__", False):
            span_attr = {"scene": "Reasoner", "desc": "Reasoner async infer method"}
            decorated = span_decorator(
                span_attr,
                SpanEvent.BEFORE_REASONER_INFER,
                SpanEvent.AFTER_REASONER_INFER,
                SpanEvent.REASONER_INFER_RAISE,
                SpanEvent.REASONER_INFER_ABORT,
                partial(cls._generate_context, desc="Before async infer"),
                partial(cls._generate_context, desc="After async infer"),
            )(cls.async_infer)
            setattr(cls, "async_infer", decorated)

    def _generate_context(self, *args: Any, desc: str, **kwargs: Any) -> "ReasonerEventContext":
        before_run: bool = not kwargs.get(SPAN_WRAP_KEY)
        query = args[0] if args else kwargs.get("query", "")

        if before_run:
            return ReasonerEventContext(
                entity_id=str(id(self)),
                desc=desc,
                info=str(query),
                input_query=str(query),
            )
        else:
            result = kwargs.get(SPAN_WRAP_KEY)
            return ReasonerEventContext(
                entity_id=str(id(self)),
                desc=desc,
                info=str(result),
                input_query=str(query),
                reasoner_result=str(result),
            )

关键机制说明：

SPAN_WRAP_KEY 如何区分 before 和 after

span_decorator 在方法执行后，会将返回值注入 kwargs[SPAN_WRAP_KEY]（即 kwargs["__func_res"]），然后调用 after 上下文生成器。因此：

kwargs.get(SPAN_WRAP_KEY) 为 falsy → 这是 BEFORE 事件
kwargs.get(SPAN_WRAP_KEY) 有值 → 这是 AFTER 事件，值就是方法的返回结果

这允许同一个 _generate_context 方法同时处理 before 和 after 两种情况。

__is_tf_span_wrapped__ 防重复包装

在多层继承中，__init_subclass__ 会在每个子类定义时被调用。如果父类已经包装了 infer 方法，子类如果没有覆写它，infer 上已经有 __is_tf_span_wrapped__ = True 标记，decorate_span 会跳过。只有当子类覆写了 infer 时（新方法没有标记），才会重新包装。

异常上下文生成器（可选）

span_decorator 的第 8 个参数 generate_exception_context 是可选的。如果不提供：

框架会自动构造一个 BaseEventContext，其 info 字段为异常信息
适用于大多数场景

如果你需要在异常时携带更丰富的上下文（如 LLM 场景需要记录模型名、输入等），可以提供自定义的异常上下文生成器。注意它的签名多一个 exception 参数：

def _generate_exception_context(
    self, *args: Any, desc: str, exception: Exception, **kwargs: Any
) -> ReasonerEventContext:
    return ReasonerEventContext(
        entity_id=str(id(self)),
        desc=desc,
        info=f"Exception: {type(exception).__name__}: {exception}",
        input_query=str(args[0]) if args else "",
    )

第四步：更新 EventScene 类型（如需精确路由）¶

如果你希望 Hook 能够通过 scene="Reasoner" 精确订阅你的事件，需要将新场景添加到 EventScene 类型定义中：

# tfrobot/telemetry/context_schema.py

EventScene: TypeAlias = Literal[
    "Robot", "Brain", "Drive", "Memory", "MemoryStore",
    "Chain", "Tool", "LLM",
    "Reasoner",  # 新增
]

同时更新 HookManager 中 SCENE 类型的定义以包含新场景。

不更新 EventScene 也能工作

HookManager.execute_hooks 从 context.get("scene") 提取场景名称，然后查找该场景下注册的 Hook。即使你不修改 EventScene 类型定义，事件仍然能正常发出和分发。但此时 scene 字段的类型检查（Pydantic Literal 校验）会失败。如果你的组件不需要严格的类型安全，可以跳过这一步。

运行时行为总结¶

扩展完成后，当 MyReasoner().infer(query) 被调用时，以下流程自动发生：

1. __init_subclass__ 触发 → decorate_span() 包装 infer 方法
2. infer() 被调用
3. span_decorator 创建 OpenTelemetry Span
4. _generate_context(desc="Before infer") → 构造 ReasonerEventContext
5. TFSpan.add_event("before_reasoner_infer", attributes) → HookManager 分发
6. 原始 infer() 执行
7. 正常完成 → _generate_context(desc="After infer") → add_event("after_reasoner_infer")
   异常抛出 → generate_exception_context() → add_event("reasoner_infer_raise" 或 "reasoner_infer_abort")

Hook 端只需：

from tfrobot.telemetry.hook import tf_event_hook


@tf_event_hook(scenes=["Reasoner"])
def on_reasoner_event(name, context):
    print(f"[Reasoner] {name}: {context.get('info')}")

已有实现参考¶

组件	文件	特点
Robot	`tfrobot/base.py:84-144`	最简模式：一个 `_generate_context` 同时处理 before/after
Brain	`tfrobot/brain/base.py:44-101`	与 Robot 结构相同
Chain	`tfrobot/brain/chain/base.py:174-225`	包含状态机信息（`self.state`）在 info 中
LLM	`tfrobot/brain/chain/llms/base.py:166-276`	最复杂：独立的 `_generate_exception_context`、`field_serializer`、媒体缓存
Tool	`tfrobot/drive/tool/base.py:135-599`	同一个 `_generate_context` 处理 before/after/exception 三种路径
Memory	`tfrobot/brain/memory/base.py:58-114`	包装 `recall`/`async_recall` 而非 `run`

推荐起步：从 Robot（tfrobot/base.py）的实现开始参考，它是最简单的完整示例。如果需要异常上下文定制，再参考 LLM 的实现。