Agently 文档

中文

English

中文

English

Appearance

State 与 Resources ​

TriggerFlow 里最容易写错的不是连线，而是“这个值应该放哪”。一个客服工单流程里可能同时有三类东西：

  • 工单分类、审批结果、模型产出的报告。
  • 数据库 client、HTTP client、日志对象、回调函数。
  • 所有 execution 都共用的一份配置或缓存。

它们不能放在同一个地方。TriggerFlow 提供三层存储：state、runtime_resources、flow_data。

先用三个问题判断 ​

问题是否
这个值要出现在 close snapshot 或 save/load 里吗？用 state继续判断
这个值是 live 对象、函数、client、socket、文件句柄吗？用 runtime_resources继续判断
这个值真的要被同一个 flow 的所有 execution 共享吗？用 flow_data 或外部存储多数情况下仍应回到 state

日常业务里，默认答案通常是 state。只有不能序列化的活对象才进 runtime_resources。flow_data 要带着明确理由使用。

state：一次 execution 自己的业务状态 ​

state 是 execution-local、可序列化、可快照的。它会进入 close snapshot，也会被 execution.save() / execution.load(saved) 往返。

python
async def classify_ticket(data):
    result = {
        "ticket_id": data.input["ticket_id"],
        "team": "billing",
        "priority": "high",
    }
    await data.async_set_state("classification", result)
    return result

读取时不需要 await：

python
async def draft_reply(data):
    classification = data.get_state("classification")
    await data.async_set_state(
        "reply_context",
        {"team": classification["team"], "tone": "concise"},
    )

常用 API：

操作API
写入await data.async_set_state(key, value)
读取data.get_state(key, default=None)
追加 listawait data.async_append_state(key, value)
删除await data.async_del_state(key)

写入、追加、删除都有同步版本。服务和 chunk 里优先用 async 版本，和周围的 async flow 保持一致。

runtime_resources：给 chunk 用的活对象 ​

DB client、模型 agent、搜索函数、logger 都不适合进 state。它们不能可靠序列化，也不应该被 close snapshot 暴露给调用方。

创建 execution 时注入：

python
execution = flow.create_execution(
    auto_close=False,
    runtime_resources={
        "db": db_client,
        "logger": logger,
        "search": search_function,
    },
)

chunk 内读取：

python
async def enrich_ticket(data):
    db = data.require_resource("db")
    logger = data.require_resource("logger")

    ticket = await db.fetch_ticket(data.input["ticket_id"])
    logger.info("ticket loaded")
    await data.async_set_state("ticket", ticket)
    return ticket

require_resource(name) 适合必需依赖，没注入时直接报错。可选依赖用 data.get_resource(name, default=None)。

flow 级别也可以注入长期复用的资源：

python
flow.update_runtime_resources(logger=logger, search=search_function)

如果使用 execution_environments=[...]，Execution Environment Manager 可以把托管资源注入到同一套 runtime_resources 里。chunk 的读取方式不变，差别在于资源的启动和释放由环境管理器负责。

为什么 runtime_resources 不进 snapshot ​

close snapshot 要能被 JSON 化、写日志、返回 API、保存到数据库。live 对象没有稳定的序列化意义。TriggerFlow 会在 save/snapshot 里记录 resource_keys，告诉应用“恢复时需要重新注入哪些名字”，但不会保存对象本体。

python
saved = execution.save()

restored = flow.create_execution(
    auto_close=False,
    runtime_resources={
        "db": new_db_client,
        "logger": new_logger,
        "search": search_function,
    },
)
restored.load(saved)

恢复代码要重新注入兼容资源。这个边界清楚后，跨进程恢复才不会把过期连接、旧进程内存对象带到新 execution 里。

flow_data：确实要共享时才用 ​

flow_data 属于 flow，而不是某次 execution。同一个 flow 同时跑两个 execution，它们看到的是同一份 flow_data。

python
flow.set_flow_data("shared_counter", 0, no_warning=True)

默认调用会发 RuntimeWarning，原因很直接：

  • 并发 execution 可能互相覆盖。
  • execution.save() / load() 不保存 flow_data。
  • 多进程服务里，不同 worker 的 flow 实例可能不是同一份内存。

如果只是一次 execution 的中间值，用 state。如果是不可序列化的共享 client，用 flow 级 runtime_resources。如果是跨进程共享业务数据，通常应该放外部存储。

一个服务里的完整分工 ​

python
async def load_ticket(data):
    db = data.require_resource("db")
    ticket = await db.fetch_ticket(data.input["ticket_id"])
    await data.async_set_state("ticket", ticket)
    return ticket


async def classify(data):
    agent = data.require_resource("agent")
    ticket = data.get_state("ticket")
    result = (
        agent
        .input(ticket["content"])
        .output({"team": (str, "处理团队", True)})
        .get_result()
    )
    classification = await result.async_get_data(ensure_keys=["team"])
    await data.async_set_state("classification", classification)
    return classification


flow.to(load_ticket).to(classify)

execution = flow.create_execution(
    runtime_resources={"db": db_client, "agent": agent},
)
snapshot = await execution.async_start({"ticket_id": "T-1024"})

这里的边界是：

  • ticket 和 classification 是业务结果，用 state。
  • db_client 和 agent 是 live 依赖，用 runtime_resources。
  • close snapshot 只暴露业务状态，不暴露连接对象。

常见错误 ​

写法问题改法
把 DB client 放进 state保存和恢复都不可靠放进 runtime_resources
把单次工单的中间值放进 flow_data并发 execution 会互相影响放进 state
load(saved) 后直接跑缺少 live 资源按 resource_keys 重新注入
把大块流式文本不断追加进 statesnapshot 变大，前端也拿不到增量进度走 runtime stream，最终摘要进 state

另见 ​