开发者文档 指南
状态与记忆
三种不同生命周期的持久层:任务记忆(一次性写入的经验)、项目状态(共享可变状态)、响应式 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)