Self-Programming Workflow · 1 / 3

编程模型与组合

本页只定义 Workflow、goal()、agent()、llm()、Python package、跨 Workflow import 和运行快照。创建与发布、DAG 和管理网页分别位于另外两页。

1. Workflow 是最高层 Python 组合

Workflow → goal()任务具有明确完成条件，需要多轮工作与判定。

Workflow → agent()需要开放式工具循环，但不需要独立 Goal 条件。

Workflow → llm()只需一次分类、提取、判断或局部生成。

Goal建立在 Agent loop 上

→

Agent loop反复调用模型和工具

→

Model callllm() 是直接入口

Workflow 可以任选三层中的一种或多种进行组合；能力依赖顺序固定为 Goal → Agent loop → 单次模型调用 → Model Runtime。Workflow 不要求包含 Goal。

2. 四个现有入口的职责

入口	执行语义	适用位置
`@agentic_function`	包装普通Python函数，建立ambient Runtime、模型工具入口和DAG call node。	公开Workflow入口，以及需要独立显示的内部Goal函数。
`goal(prompt, condition)`	以 `agent()` 执行每轮工作，以内部模型调用判断condition。	有明确完成条件、需要多轮修正的阶段目标。
`agent(prompt, tools)`	模型可反复请求工具，直到文本完成或达到迭代上限。	Goal的执行基础，也可作为Workflow的直接步骤。
`llm(prompt)`	一次模型调用，`tools=[]`、`max_iterations=1`。	分类、提取、判断、改写和局部生成。

3. 具体例子：论文调研

from openprogram.agentic_programming import agentic_function, goal, llm

@agentic_function(as_tool=False)
def collect_evidence(topic: str) -> str:
    return goal(
        prompt=f"检索并核对与 {topic} 有关的论文",
        condition="候选论文已核对年份、出处和研究问题",
    )

@agentic_function(as_tool=False)
def screen_papers(evidence: str) -> str:
    return llm(f"筛选与任务直接相关且来源可核对的论文：\n{evidence}")

@agentic_function
def literature_review(task: str) -> str:
    evidence = collect_evidence(task)
    return screen_papers(evidence)

内部Goal函数使用 @agentic_function(as_tool=False) 保留Python调用与DAG记录，但不注册成模型工具。确定性helper保持普通函数。

4. 一个 Workflow 是一个 Git 管理的 Python 小项目

~/.openprogram/workflows/literature_review/
├── .git/
├── README.md
├── pyproject.toml
├── __init__.py
├── workflow.py
├── goals/              # 可选
├── helpers/
└── tests/
    └── test_workflow.py

公开入口	`workflow.py` 中与项目同名的 `@agentic_function(task)`。
项目元数据	`pyproject.toml` 与 `openprogram.workflows` entry point。
内部代码	相对import组织goals与helpers。
验证	标准库 `unittest`；替换模型和外部Workflow调用。

研究主题、输入路径、输出目录和条件必须来自每次调用参数，不能写死在首次生成的源码中。

5. Workflow 之间只使用 import 和函数调用

from openprogram.agentic_programming import agentic_function
from workflows.paper_search import paper_search
from workflows.paper_screening import paper_screening
from workflows.research_synthesis import research_synthesis

@agentic_function
def research_pipeline(task: str, output_dir: str) -> str:
    papers = paper_search(task, year_from=2025)
    selected = paper_screening(papers)
    return research_synthesis(task, selected, output_dir)

不增加字符串dispatcher、callee id、@workflow、call_workflow() 或 invoke_workflow()。跨Workflow调用只接受静态 from workflows.<package> import <function>，以便运行前计算Calls / Used by、循环和缺失依赖。

6. 加载与执行只准备 import 环境

扫描依赖

从根项目的静态imports计算全部可达Workflow。

固定版本

用 git archive HEAD 把根项目和依赖放入同一临时namespace。

准备import path

隔离worker只把本次run root加入普通Python import path，不实现自定义importer。

正常调用

导入顶层 @agentic_function 后直接调用；所有嵌套调用继承同一Runtime、权限、取消和DAG caller。

边界	合同
代码版本	一次执行固定完整commit；未提交工作树内容不进入run。
Import范围	只包含本次可达Workflow快照、OpenProgram和获准依赖。
失败	ImportError、函数异常、Goal未满足和Agent超限按普通错误进入caller，不自动修改代码。
循环	静态依赖环在worker启动前拒绝。