开发者文档 指南

状态与记忆

三种不同生命周期的持久层:任务记忆(一次性写入的经验)、项目状态(共享可变状态)、响应式 store(带变更追踪的类型化状态)。

选对层级

层级生命周期写入模型适用场景
任务记忆永久,按元任务平台在每次运行后写入未来运行应检索的经验与模式
项目状态项目存续期通过响应式代理直接 dict 写入跨运行的开关、计数与协调状态
响应式 store运行内,可选持久化类型化字段、action、patch运行内需要变更监听的结构化状态

响应式 store

用类注解声明字段,@getter 派生,@action 变更。每次变更都会发出可订阅的类型化事件:

from allybuild_sdk.reactive import defineStore, getter, action, field

@defineStore("review-queue")
class ReviewQueue:
    items: list = field(default_factory=list)

    @getter
    def pending(self) -> int:
        return sum(1 for i in self.items if not i["done"])

    @action
    def add(self, path: str) -> None:
        self.items = self.items + [{"path": path, "done": False}]

queue = ReviewQueue()
unsub = queue["$subscribe"](lambda m, s: print(m.type, len(s["items"])))
queue.add("src/auth.py")     # 打印: direct 1
print(queue.pending)          # 1

美元符号 API

方法用途
$patch(dict | fn)批量更新,产生一次 "patch" 变更
$reset()恢复声明默认值,发出 "reset"
$subscribe(cb)监听变更;返回取消订阅函数
$on_action(cb)action 生命周期钩子:after / on_error
$atomic(fn)显式的单事件多字段更新
$state深拷贝的只读快照
$dispose()解除订阅并从注册表移除

HTTP 持久化

from allybuild_sdk.reactive import HttpBackend, create_reactive

backend = HttpBackend(
    api_url=ALLYBUILD_API_URL,
    jwt=TASK_JWT,
    project_id=PROJECT_ID,
    timeout=5.0,
)
create_reactive(redis=backend)

记忆实战

记忆由平台代写:开启记忆提取的元任务会把每次完成的运行蒸馏成经验——agent 运行取会话轨迹,脚本运行回退到 stdout。对子任务树,整树落定后只提取一次,由根元任务的开关决定。高价值模式是在下一次运行开始时检索记忆来支撑决策,并用 order="least" 反向检索以暴露新颖的失败模式:

from allybuild_sdk import MemoriesApi

memories = MemoriesApi(client)
grounding = memories.list_task_type_memories(
    "incident-triage", query="database failover", k=5
)
novel = memories.list_task_type_memories(
    "incident-triage", query="database failover", k=5, order="least"
)

跨运行的项目状态

# project_status 在 workflow 模块里已预接线
project_status["last_run"] = TASK_ID
project_status["deploy_count"] = project_status.get("deploy_count", 0) + 1

for key, value in project_status.items():
    print(key, "=", value)