AgentExecution 与自动编排
有些任务看起来只是一次提问,但模型可能需要调用 Action、使用 Skills,或者把输入拆成 TaskDAG。Agently 的主线不是让 Agent “随便自治”,而是把一次运行收束到 AgentExecution:Agent 保存可复用定义,AgentExecution 保存本轮输入、目标、输出契约、状态和诊断。
自动编排只是 AgentExecution 里的路线选择:先把允许使用的能力注册清楚,再让 Agent 在这些候选里完成一次有边界的执行。
最小形态
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。
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。
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:
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:
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 或 Skills | Agent 自动编排 |
| 模型生成或应用提交一个 DAG | Dynamic Task |
| 流程阶段固定、需要分支 / 并行 / pause / stream | TriggerFlow |
| 单 Agent 做 2 到 5 轮有边界任务 | agent.create_task(...) 返回的 AgentExecution draft |
| 宿主系统自己写循环,只要一个 bounded Agent step | create_execution(mode="task_step") |
常见误用
| 写法 | 问题 |
|---|---|
| 不注册候选能力,却期待 Agent 自动调用工具 | 没有能力边界,仍是普通模型请求 |
| 把 AgentTask 当新项目默认 handle | 新代码应优先消费 AgentExecutionResult |
| 把 task strategy 当长期后台自治 | 当前公开 slice 只覆盖有边界单任务 |
| 用 Agent 自动编排替代明确流程 | 固定阶段流程更适合 TriggerFlow |
| 在单例 Agent 上累积每个用户的临时 input | 请求之间会互相污染,应使用 execution draft |
另见
- Action Runtime - 给 Agent 开放外部能力
- Skills Executor - coding-agent 风格能力执行
- Dynamic Task - DAG 计划与执行
- TriggerFlow 概览 - 固定流程编排
- Workspace - AgentTask Loop 的证据和 checkpoint