项目状态与 Tokens

project_status

project_status 是一个响应式代理对象，提供对项目共享状态的读写访问。项目中每个「Status」对应一个命名的状态对象，其字段值通过 Redis 实时共享。

# 读取字段值
current = project_status.pipeline.current_stage
count = project_status.metrics["processed_count"]

# 写入字段值（立即同步到 Redis）
project_status.pipeline.current_stage = "embedding"
project_status.metrics.processed_count += 1

遍历所有状态

for name, store in project_status.items():
    print(f"{name}: {store.state}")

使用 get()

status = project_status.get("pipeline")
if status is None:
    print("状态不存在")

注意：project_status 的字段定义在 TaskFlow UI 的「项目状态」页面配置。脚本只能读写已声明的字段，字段类型和默认值由 Schema 决定。

---

响应式状态（defineStore）

对于脚本内部的临时响应式状态，使用 defineStore：

from workflow import defineStore, action, getter, field

CartStore = defineStore("cart", {
    "state": lambda: {"items": field(default_factory=list), "total": 0.0},
    "actions": {
        "add_item": action(lambda self, item: (
            self.items.append(item),
            setattr(self, "total", self.total + item["price"]),
        )),
    },
    "getters": {
        "item_count": getter(lambda self: len(self.items)),
    },
})

cart = CartStore()
cart.add_item({"name": "书", "price": 49.0})
print(cart.item_count)  # 1

defineStore(store_id, options)

参数	类型	说明
store_id	str	Store 唯一标识符
options	dict	包含 state、actions、getters 键

action(fn)

装饰状态修改方法，确保状态变更被正确追踪。

getter(fn)

装饰计算属性，返回派生值。

field(default_factory)

用于 state 中的 list/dict 类型字段，避免可变默认值问题（类似 dataclasses.field）。

---

project_tokens

只读的项目 Token 字典，在项目设置中配置。

def run(params: dict, reporter) -> None:
    api_key = project_tokens["openai_key"]
    token = project_tokens.get("github_pat", "")