Self-Programmed Agentic Workflow

agent 自己把任务写成一个真正的 Python 程序,再由框架执行。程序用框架的积木 (注册的 agentic function、自由 agent 调用)组合而成,控制流就是 Python 本身的控制流。 运行模型和人写代码一样:整个程序一次跑完;报错就看报错、改代码、整个重跑, 已完成的调用直接用上次结果,效果上从出错处继续。入口是内置 agentic function run_task_list

1. 全流程

run_task_list(task) 每次调用 = 新建一个独立实例 planner:把任务写成程序 只读工具 read / grep / glob / list 输入:任务 + 函数注册表(积木清单) 产出:code.py,def workflow() if / for / 异常处理随便写 校验 ast 能解析 · 必须有 def workflow() 禁止 import(积木由框架注入) 通过 无效:打回重写,直到能跑 exec 整个程序 workflow() 从头到尾跑完 没有调度器,没有"取下一项" 已完成的调用直接回放上次结果 正常返回 completed 返回结果与全部运行记录 程序抛异常:planner 看报错和运行记录,改代码 带着异常栈、 state.json 现场 改完整个程序重跑 = 从出错处继续 实例目录 workflows/<run_id>/ code.py:当前程序(修订即重写) state.json:每次调用的全量记录 状态变化先写盘再动手 实例之间完全独立: 想同时跑几个流程就跑几个, 没有共享的板,没有互相匹配
全流程。planner 写程序,框架 exec;出错就改代码整个重跑,调用回放让重跑等价于断点续跑。 每个实例一个目录,并发互不相关。

2. 生成的程序长什么样

planner 产出的是一个普通 Python 模块,入口 workflow(),组合就是函数互调, 控制流就是 Python 的控制流。程序里的 LLM 原语是单次请求的 llm() 与自主工具循环的 agent()。 "让模型想/写一段"和"让一个 agent 动手干活"是同一个调用,区别只在 agent_id 选的 profile 带不带工具:

def find_issues() -> str:
    return agent(
        "审查 openprogram/auth/ 的代码,逐文件检查错误处理与并发安全,"
        "输出问题清单,按 HIGH / MEDIUM / LOW 分级,每条带文件路径与行号。",
        description="find issues")


def workflow() -> str:
    findings = find_issues()
    if "HIGH" in findings:
        agent("按清单修复 HIGH 级问题,改完跑通相关测试:" + findings,
              description="fix auth")          # 干活的 agent,profile 带工具
        checks = run_tests()                    # 注册表里的 agentic function 直接调
        if "failed" in checks:
            raise RuntimeError("修复后测试仍失败:" + checks)
    return agent("汇总以上结果写报告", description="report")

可用的积木由框架在 exec 时注入,程序里禁止 import

注入的名字说明
llm一次模型请求;不创建 session 分支,不进入工具循环。 签名为 llm(prompt, model=, effort=, response_format=, choices=, web_search=, timeout_s=),并与其他注入调用一样由 checkpoint 层包装。
agent现有的 spawn 工具原样注入 (functions/tools/agent/,签名 agent(prompt, description=, agent_id=, run_in_background=, …))。模型与工具集通过 agent_id 选 agent profile; 需要自主工具循环或独立分支时使用。
注册的 agentic functionAGENTIC_MODULES 注册表里的全部函数,按名直接调用。 planner 的 prompt 里带着这份清单(函数名、签名、docstring 首行),它据此选材。
程序里没有任何 checkpoint 语法。存档是框架的事:注入环境里的每个可调用都被包了一层, 真实执行前后写 state.json,键 =(函数名,该名第几次被调,参数摘要)。验证也不是框架强加的: 要验证就像上面那样把检查写成程序的一步,不满足就 raise,交给修订回环。

3. 断点续跑:整个程序重跑,调用回放

没有"记录几个点、完成的打勾"的调度逻辑。续跑就是把 workflow() 从第一行重新执行一遍, 唯一的机制是调用边界的短路:state.json 里已 completed 的调用(同名、同次序、同参数摘要), 不真正执行,直接返回上次的结果。于是重跑时程序飞速掠过做完的部分, 到第一个没做完的调用才真正干活——控制流每次都是完整走的(if 重新判断、for 重新循环), 恢复的只是昂贵调用的结果。

首跑 find_issues() 实跑 ✓ agent("修复…") 实跑 ✓ run_tests() 崩溃 ✗ 进程死 / 抛异常 前两个调用的结果已在 state.json;planner 若需要改代码,改完仍走下面这条路 重跑 find_issues 回放 agent 回放 run_tests 从头实跑 → 继续往后 completed 虚线 = 不执行,从 state.json 直接取上次结果,耗时≈0 整个 workflow() 仍从第一行完整执行:if 重新判断、for 重新循环,恢复的只是昂贵调用的结果
重跑即续跑。程序每次都完整执行,做完的调用瞬间回放,第一个未完成的调用起真实运行。
进程被杀同理:state.json 每次状态变化先写盘,重启后拿 run_id 再跑一遍即续。 续跑是显式的——每次 run_task_list(task) 都新建实例(返回值带 run_id), 续跑要传入既有实例的 run_id。不存在按任务文本匹配旧运行这种事。

4. 出错后的修订:改代码,重跑

程序抛异常——语法错、积木调用失败、planner 自己写的检查 raise——处理方式和开发者一样: planner 拿到异常栈、当前代码和 state.json 里的运行记录,重写 code.py(改错的那段、拆步骤、 补前置,随它),然后整个程序重跑。没改动的已完成调用照旧回放,所以修订不推翻已完成的工作。 旧版代码留档在实例目录里,修订历史进返回值的 revisions。校验不过、修完还错, 都是打回 planner 再改——改到能跑为止,运行的强制停止只有 capped 一种。

5. 与现有机制的关系

机制关系
Agentic Programming 核心(@agentic_function / Runtime 生成的程序就是一个标准 agentics 模块,执行走同一套 DAG 追踪;本设计不引入第二种函数形态。
todo 计划板不参与。workflow 的状态在自己的实例目录里,todo 板继续做它自己的事。
goal 判定 / continuation loop不参与。程序的验证由 planner 写进代码,失败即异常。
函数注册表(AGENTIC_MODULESplanner 的积木清单来源,也是 exec 时注入执行环境的白名单。

6. 边界与取舍

决定理由
planner 直接写 Python 代码表达力优先:分支、循环、异常处理是任务编排的常态, 不为它们发明第二种规格语言。写错的风险由校验(ast、禁 import、必须 def workflow())和修订回环兜住。
续跑 = 重跑 + 调用回放Python 无法从函数中间恢复执行,重跑整个程序、回放已完成调用 是代码级断点续跑的唯一正确形态;控制流重走保证修订后的逻辑立即生效。
checkpoint 自动,代码里无痕存档语法一旦要生成方手写就会被写错;包在注入环境的 调用边界上,生成的代码和手写的 agentics 模块长得一模一样。
实例独立,无共享板一个 workflow 是一个程序的一次运行,多个流程并发是常态; 共享清单带来的匹配、认领、串扰问题一概不要。
小任务不写程序planner 第一步仍做规模判断,单 agent 一口气能完成的直接执行, 写程序的开销只花在配得上它的任务上。
失败不放弃没有 abandoned:无效代码、运行报错都带着具体错误打回 planner 重写, 改到能跑为止。防失控的唯一闸门是 40 次真实调用执行触发 capped(回放不计数)。

实现状态

代码在 openprogram/functions/agentics/task_list/,注册在 AGENTIC_MODULES, 入口 run_task_list。正在按本设计从 JSON item 清单形态重实现为生成代码形态; 产品文档 docs/capabilities/task-list.md 随实现落地后同步改写。