一个变量的地址（selector）永远是"至少两段"：第一段是命名空间/节点 id，第二段是字段名。比如系统变量 query 的地址就是 ("sys", "query")。看最基础的一个辅助函数 system_variable_selector（core/workflow/system_variables.py:59）：

# core/workflow/system_variables.py:55
def system_variable_name(key: str | SystemVariableKey) -> str:
    return key.value if isinstance(key, SystemVariableKey) else key

def system_variable_selector(key: str | SystemVariableKey) -> tuple[str, str]:
    return SYSTEM_VARIABLE_NODE_ID, system_variable_name(key)   # ★ ("sys", "query")

那些"第一段"从哪来？Dify 把它们收拢在一个极短的文件里 variable_prefixes.py（core/workflow/variable_prefixes.py:1）：

# core/workflow/variable_prefixes.py:1
SYSTEM_VARIABLE_NODE_ID = "sys"
ENVIRONMENT_VARIABLE_NODE_ID = "env"
CONVERSATION_VARIABLE_NODE_ID = "conversation"
RAG_PIPELINE_VARIABLE_NODE_ID = "rag"

sys 这个抽屉里具体能放什么？由一个枚举钉死 SystemVariableKey（core/workflow/system_variables.py:22）：

# core/workflow/system_variables.py:22
class SystemVariableKey(StrEnum):
    QUERY = "query"                       # 用户这轮的问题
    FILES = "files"                       # 用户上传的文件
    CONVERSATION_ID = "conversation_id"   # 会话 id
    USER_ID = "user_id"                   # 用户 id
    DIALOGUE_COUNT = "dialogue_count"     # 第几轮对话
    APP_ID = "app_id"
    WORKFLOW_ID = "workflow_id"
    WORKFLOW_EXECUTION_ID = "workflow_run_id"   # 注意：枚举名和值故意不同
    ...
    INVOKE_FROM = "invoke_from"           # 从哪触发（调试/API/前端）

那把这些系统变量真正"造出来"的是 build_system_variables（core/workflow/system_variables.py:81）：

# core/workflow/system_variables.py:81
def build_system_variables(values=None, /, **kwargs) -> list[Variable]:
    normalized = _normalize_system_variable_values(values, **kwargs)   # ① 清洗
    return [
        segment_to_variable(                                           # ② 逐个包装成 Variable
            segment=build_segment(value),
            selector=system_variable_selector(key),                    #    地址 = ("sys", key)
            name=key,
        )
        for key, value in normalized.items()
    ]

工作流一启动，得先把"初始变量"一次性装进池。这活由 build_bootstrap_variables（core/workflow/system_variables.py:109）干——它把四类变量各自盖上正确的命名空间前缀：

# core/workflow/system_variables.py:109
def build_bootstrap_variables(*, system_variables=(), environment_variables=(),
                              conversation_variables=(), rag_pipeline_variables=()):
    variables = [
        *(_with_selector(v, SYSTEM_VARIABLE_NODE_ID)       for v in system_variables),        # sys.*
        *(_with_selector(v, ENVIRONMENT_VARIABLE_NODE_ID)  for v in environment_variables),   # env.*
        *(_with_selector(v, CONVERSATION_VARIABLE_NODE_ID) for v in conversation_variables),  # conversation.*
    ]
    # rag 变量按"归属节点"分组打包
    rag_pipeline_variables_map = defaultdict(dict)
    for rag_var in rag_pipeline_variables:
        node_id = rag_var.variable.belong_to_node_id
        rag_pipeline_variables_map[node_id][rag_var.variable.variable] = rag_var.value
    for node_id, value in rag_pipeline_variables_map.items():
        variables.append(segment_to_variable(
            segment=build_segment(value),
            selector=(RAG_PIPELINE_VARIABLE_NODE_ID, node_id), name=node_id))
    return variables

关键是那个 _with_selector（core/workflow/system_variables.py:102）——它保证每个变量的地址第一段是对的前缀：

# core/workflow/system_variables.py:102
def _with_selector(variable: Variable, node_id: str) -> Variable:
    selector = [node_id, variable.name]
    if list(variable.selector) == selector:
        return variable                              # 已经对了：原样返回，不复制
    return variable.model_copy(update={"selector": selector})   # 否则：改地址再复制一份

开机装的是初始料。节点跑起来后产出的新数据，靠 variable_pool_initializer.py 里的两个小函数写回池（core/workflow/variable_pool_initializer.py:8）：

# core/workflow/variable_pool_initializer.py:8
def add_variables_to_pool(variable_pool, variables) -> None:
    for variable in variables:
        variable_pool.add(variable.selector, variable)      # 逐个按自带地址塞进池

def add_node_inputs_to_pool(variable_pool, *, node_id, inputs, aliases=()):
    """Store node inputs under the primary node id and any compatible aliases."""
    node_ids = [node_id]
    for alias in aliases:                                   # ① 主 id + 若干别名
        if alias not in node_ids:
            node_ids.append(alias)
    for current_node_id in node_ids:
        for key, value in inputs.items():
            variable_pool.add((current_node_id, key), value)   # ② 每个字段都存一份

读取端也有一组便捷函数（core/workflow/system_variables.py:140），把"查池 + 拆 Segment"包成一步：

# core/workflow/system_variables.py:140
def get_system_segment(variable_pool, key) -> Segment | None:
    return variable_pool.get(system_variable_selector(key))   # 按 ("sys",key) 查

def get_system_value(variable_pool, key) -> Any:
    segment = get_system_segment(variable_pool, key)
    return None if segment is None else segment.value         # 拆出裸值

def get_all_system_variables(variable_pool):
    return variable_pool.get_by_prefix(SYSTEM_VARIABLE_NODE_ID)   # 一次拿回整个 sys 抽屉

cd /Users/bitmart/work/codes/github/AI_WORK/dify/api

# 1. 四个前缀（就 4 行）
cat core/workflow/variable_prefixes.py

# 2. 系统变量清单 + 构造
sed -n '22,39p'   core/workflow/system_variables.py    # SystemVariableKey 枚举
sed -n '81,96p'   core/workflow/system_variables.py    # build_system_variables
sed -n '109,138p' core/workflow/system_variables.py    # build_bootstrap_variables

# 3. 写回池
cat core/workflow/variable_pool_initializer.py         # 全文就 28 行

# 4. 数一数系统变量有几个
grep -c '= "' core/workflow/system_variables.py