dsl/_start.py 全文很短（flow/dsl/_start.py:18）：

# flow/dsl/_start.py:18
def start(condition: FlowTrigger | None = None) -> FlowMethodDecorator:
    """Marks a method as a flow's starting point."""

    def decorator(func: Callable[P, R]) -> StartMethod[P, R]:
        wrapper = StartMethod(func)                     # ① 包成 StartMethod 对象

        _merge_flow_method_definition(                  # ② 挂上元数据
            wrapper,
            FlowMethodDefinition(
                do=_method_action(func),                # 记住"这方法怎么调"
                start=(                                 # 记住"起点 + 触发条件"
                    _to_definition_condition(condition)
                    if condition is not None
                    else True                           # 无条件 → start=True
                ),
            ),
        )
        return wrapper
    return cast(FlowMethodDecorator, decorator)

dsl/_listen.py（flow/dsl/_listen.py:18），对比着看：

# flow/dsl/_listen.py:18
def listen(condition: FlowTrigger) -> FlowMethodDecorator:   # 注意：condition 必填
    def decorator(func: Callable[P, R]) -> ListenMethod[P, R]:
        wrapper = ListenMethod(func)                     # 包成 ListenMethod

        _merge_flow_method_definition(
            wrapper,
            FlowMethodDefinition(
                do=_method_action(func),
                listen=_to_definition_condition(condition),  # ← 填的是 listen 字段
            ),
        )
        return wrapper
    return cast(FlowMethodDecorator, decorator)

三个装饰器都用 FlowMethod 作基类（flow/flow_wrappers.py:48）：

# flow/flow_wrappers.py:56
def __init__(self, meth: Callable[P, R], instance: Any = None) -> None:
    self._meth = meth
    self._instance = instance
    functools.update_wrapper(self, meth, updated=[])   # ① 复制 __name__/__doc__ 等
    self.__name__ = FlowMethodName(self.__name__)
    self.__signature__ = inspect.signature(meth)       # ② 保留原签名
    if instance is not None:
        self.__self__ = instance
    if inspect.iscoroutinefunction(meth):              # ③ 异步方法也要能被认出
        inspect.markcoroutinefunction(self)
    # 把 @human_feedback / @persist 等其它装饰器留下的标记也一并搬过来
    for attr in ["__human_feedback_config__", "__conversational_only__",
                 "__flow_persistence_config__", "__flow_method_definition__"]:
        if hasattr(meth, attr):
            setattr(self, attr, getattr(meth, attr))

# flow_wrappers.py:90
def __call__(self, *args, **kwargs) -> R:
    if self._instance is not None:
        return self._meth(self._instance, *args, **kwargs)  # 绑定了实例：补上 self
    return self._meth(*args, **kwargs)

靠实现描述符协议 __get__（flow/flow_wrappers.py:112）：

# flow/flow_wrappers.py:112
def __get__(self, instance: Any, owner: type | None = None) -> Self:
    if instance is None:
        return self                          # 从类上访问 → 返回未绑定的自己
    bound = type(self)(self._meth, instance) # ★从实例访问 → 造一个"绑定了 instance"的新包装
    skip = {"_meth", "_instance", "__name__", "__doc__", "__signature__", ...}
    for attr, value in self.__dict__.items():
        if attr not in skip:
            setattr(bound, attr, value)      # 把元数据（含 __flow_method_definition__）复制给绑定版
    return bound

dsl/_conditions.py（flow/dsl/_conditions.py:22）：

# flow/dsl/_conditions.py:22
def or_(*triggers: FlowTrigger) -> FlowCondition:
    """任一 trigger 触发就满足。"""
    return _condition_tree(OR_CONDITION, triggers)

def and_(*triggers: FlowTrigger) -> FlowCondition:
    """所有 trigger 都触发才满足。"""
    return _condition_tree(AND_CONDITION, triggers)

# :65
def _condition_tree(condition_type, triggers) -> FlowCondition:
    return {
        "type": condition_type,                        # "OR" 或 "AND"
        "conditions": [_coerce_trigger(t) for t in triggers],  # 每个 trigger 归一成"名字"或子树
    }

方法引用会被 _coerce_trigger 归一成方法名字符串（flow/dsl/_conditions.py:32）：

# flow/dsl/_conditions.py:32
def _trigger_name(value: Any) -> str | None:
    if isinstance(value, str):
        return value                        # 直接给字符串（方法名/路由标签）
    name = getattr(value, "__name__", None)
    if callable(value) and isinstance(name, str):
        return name                         # 给的是方法引用 → 取它的 __name__
    return None

_merge_flow_method_definition 让多个装饰器往同一个方法上"累加"元数据（flow/dsl/_utils.py:109）：

# flow/dsl/_utils.py:109
def _merge_flow_method_definition(wrapper, definition) -> None:
    existing = _get_flow_method_definition(wrapper)
    if existing is None:
        _set_flow_method_definition(wrapper, definition)   # 第一个装饰器：直接设
        return
    updates = {                                            # 后续装饰器：只覆盖"显式设过的字段"
        field_name: getattr(definition, field_name)
        for field_name in definition.model_fields_set
    }
    _set_flow_method_definition(
        wrapper, existing.model_copy(deep=True, update=updates))

# :89  do 动作只是记一个"import 路径"
def _method_action(method: Any) -> FlowActionDefinition:
    return FlowCodeActionDefinition(ref=f"{method.__module__}:{method.__qualname__}")