Skip to content

AgentExecution 与自动编排

有些任务看起来只是一次提问,但模型可能需要调用 Action、使用 Skills,或者把输入拆成 TaskDAG。Agently 的主线不是让 Agent “随便自治”,而是把一次运行收束到 AgentExecution:Agent 保存可复用定义,AgentExecution 保存本轮输入、目标、输出契约、状态和诊断。

自动编排只是 AgentExecution 里的路线选择:先把允许使用的能力注册清楚,再让 Agent 在这些候选里完成一次有边界的执行。

最小形态

python
from agently import Agently

agent = Agently.create_agent()

execution = (
    agent
    .use_actions([market_data_action])
    .use_skills_packs(["equity-research"])
    .use_dynamic_task(mode="auto", max_tasks=8)
    .input("Review this renewal risk.")
    .output({"answer": (str, "final answer", True)})
)

result = execution.get_result()
data = await result.async_get_data(ensure_keys=["answer"])

如果没有注册 Actions、Skills、Skills Packs 或 Dynamic Task 候选,这仍然是一条普通模型请求。候选注入是能力边界,模型只能在你提供的能力范围内工作。

Agent 保存长期设置,AgentExecution 保存本轮输入

Agent 可以作为服务单例,保存模型配置、固定角色、Actions、Skills、Workspace 和 policy。每次 .input(...).output(...)、附件和本轮 options,会进入隔离的 AgentExecution draft。

python
agent = (
    Agently.create_agent()
    .define(prompt={"role": "你是客服工单助手。"})
    .use_actions([lookup_ticket, create_refund])
)

execution_a = agent.input("查 T-001 的状态").output({"answer": (str, "回答", True)})
execution_b = agent.input("给 T-002 发退款").output({"answer": (str, "回答", True)})

result_a = execution_a.get_result()
result_b = execution_b.get_result()

不要把多个用户请求堆在同一个 Agent 的临时 prompt 状态上。共享内容放 Agent 定义;单次内容放 AgentExecution draft。always=True 这类历史写法仍可兼容,但新文档优先使用 agent.define(...)

路由怎么发生

自动编排是候选驱动:

候选情况结果
没有候选能力普通模型请求
只有一个明确候选直接走该路线
有 submitted Dynamic Task优先按提交的 DAG 执行
有 required Skills进入 Skills route
多个可选候选并存模型按输入和候选描述选择 route

这样做能避免“模型说要做某事,但代码层根本没有开放该能力”。如果业务需要确定性路线,直接选择对应 API,例如手动创建 Dynamic Task 或直接调用 TriggerFlow。

什么时候选择 task strategy

一次回复不够,但又不想自己手写完整循环时,用 agent.create_task(...) 或 task strategy。它返回的仍是 AgentExecution draft,内部执行一个有边界的任务循环:计划、执行一个 bounded step、写 Workspace 证据、验证、必要时 replan。

python
execution = agent.create_task(
    goal="将旧版 Agently 脚本迁移到当前 API,并确认它可以运行。",
    success_criteria=[
        "旧 API 已被替换。",
        "脚本可以运行。",
        "输出保留原来的结构化字段。",
    ],
    workspace="./.agently/tasks/legacy-script-upgrade",
    max_iterations=4,
    verify="before_done",
    options={
        "agent_task": {
            "stream_progress": True,
            "stream_snapshots": True,
        },
    },
)

result = execution.get_result()

async for item in result.get_async_generator():
    meta = item.meta or {}
    if meta.get("stream_kind") == "progress":
        print("[progress]", item.value["message"])
    elif meta.get("stream_kind") == "snapshot":
        print("[snapshot]", item.path)

data = await result.async_get_data()
meta = await result.async_get_meta()
task_refs = result.task_refs

这一版 task strategy 适合单任务、单 Agent owner、2 到 5 次迭代的场景。AgentTaskLoop 是 AgentExecution 背后的执行策略,不是第二个推荐公开生命周期,也不是多 agent 协作平台或后台长期自治系统。

什么时候直接用 Execution 对象

调用方需要过程流、路线诊断、meta 或多种结果视图时,显式创建 execution:

python
execution = (
    agent
    .use_dynamic_task(mode="submitted", plan=graph, handlers=handlers)
    .input("Run the reviewed graph.")
    .create_execution()
)

async for item in execution.get_async_generator(type="instant"):
    if item.is_complete:
        print(item.path, item.value)

data = await execution.async_get_data()
meta = await execution.async_get_meta()

create_execution() 默认是普通单轮 Agent execution。开发者自己写循环时,可以用 mode="task_step" 给每一步加 lineage 和 limits:

python
execution = agent.input("Try one bounded fix step.").create_execution(
    mode="task_step",
    lineage={
        "task_id": "issue-123",
        "iteration_id": "iter-2",
        "step_id": "execute-fix",
    },
    limits={
        "max_model_requests": 3,
        "max_seconds": 180,
        "max_no_progress_seconds": 60,
    },
)

task_step 仍然只是一次 Agent execution,不是多轮循环本身。它适合让宿主系统自己掌控循环,并把每一步的预算、诊断和 lineage 记录清楚。

选择表

任务形态
可复用 Agent 定义 + 本轮隔离执行agent.define(...) + AgentExecution
一次请求,可能需要 Action 或 SkillsAgent 自动编排
模型生成或应用提交一个 DAGDynamic Task
流程阶段固定、需要分支 / 并行 / pause / streamTriggerFlow
单 Agent 做 2 到 5 轮有边界任务agent.create_task(...) 返回的 AgentExecution draft
宿主系统自己写循环,只要一个 bounded Agent stepcreate_execution(mode="task_step")

常见误用

写法问题
不注册候选能力,却期待 Agent 自动调用工具没有能力边界,仍是普通模型请求
把 AgentTask 当新项目默认 handle新代码应优先消费 AgentExecutionResult
把 task strategy 当长期后台自治当前公开 slice 只覆盖有边界单任务
用 Agent 自动编排替代明确流程固定阶段流程更适合 TriggerFlow
在单例 Agent 上累积每个用户的临时 input请求之间会互相污染,应使用 execution draft

另见