Skip to content

基本流程与推荐写法

如果你已经确定需要 TriggerFlow,这一页是最适合下手的地方:先做一条最简单、最清晰、最容易调试的流程。

适合什么时候读

  • 你刚从请求层升级到 TriggerFlow
  • 你要写第一条线性流程
  • 你想知道什么是“推荐写法”,什么是后期容易失控的写法

你会学到什么

  • 最小线性流程的写法
  • 为什么推荐具名 chunk 和显式 .end()
  • runtime_resources 应该如何注入

TIP

这页保留同步入口示例,是为了先把流程结构讲清楚。只要进入真实服务、runtime stream、或模型流式结果要继续驱动流程的场景,默认推荐切到 Async First:async chunk、execution.async_start(...)get_async_runtime_stream(...),以及 chunk 内的 response.get_async_generator(type="instant")

最小流程长什么样

这张图嵌在正文里,是为了强调:刚开始用 TriggerFlow 时,先把流程写成一条清楚的线,再去引入分支和并发。

最小示例

python
from agently import TriggerFlow, TriggerFlowRuntimeData

flow = TriggerFlow(name="hello-flow")

@flow.chunk("normalize")
async def normalize(data: TriggerFlowRuntimeData):
    return str(data.value).strip()

@flow.chunk("greet")
async def greet(data: TriggerFlowRuntimeData):
    return f"Hello, {data.value}"

flow.to(normalize).to(greet).end()
print(flow.start(" Agently "))

推荐写法

  • 优先使用具名 chunk
  • 线性链路显式 .end()
  • 新代码统一用 TriggerFlowRuntimeData
  • 需要运行依赖时优先用 runtime_resources

注入运行依赖

python
flow.update_runtime_resources(logger=my_logger)

result = flow.start(
    "topic",
    runtime_resources={"search_tool": custom_search_tool},
)

覆盖顺序:

  1. flow 级 update_runtime_resources(...)
  2. execution / start 级 runtime_resources={...}
  3. chunk 内 data.set_resource(...)

不推荐的起步方式

  • 长期逻辑都写成匿名 lambda
  • 不写 .end() 却默认期待流程自动收敛
  • 用闭包偷偷塞 logger、工具和配置

下一步去哪

  • agently-triggerflow