GPT_workflow：GPT API 工作流引擎
Python GPT API 多步骤 AI 任务编排引擎，含用户认证。
项目背景
GPT_workflow 是一个基于 GPT API 的多步骤 AI 任务编排引擎。在使用 GPT API 的过程中，我发现很多场景不是简单的"一问一答"，而是需要多步骤的复杂流程：先生成大纲，再展开每个部分，然后做翻译或润色，最后输出到文件。如果每一步都手动调用 API，效率很低且难以复用。
这个项目就是为了解决这个问题——让用户能够编排多步骤的 AI 工作流，每个步骤可以配置不同的 Prompt 模板、参数和模型，整个工作流可以一键执行并持久化结果。
技术选型
| 组件 | 选择 | 理由 |
|---|---|---|
| 后端 | Python web 框架 | 轻量级，与 OpenAI SDK 无缝集成 |
| AI 引擎 | OpenAI GPT API | 最成熟的 LLM API |
| 数据库 | SQLite（开发）/ PostgreSQL（生产） | 灵活切换 |
| 前端 | React + Vite | 现代化 UI 框架 |
| 部署 | Docker | 容器化部署 |
| 认证 | JWT | 用户系统 |
架构设计
系统分为三层：
- 前端层（React）：工作流设计器、执行监控、历史查看
- API 层：工作流 CRUD、执行触发、结果查询
- 引擎层：工作流解析、步骤编排、LLM 调用、结果聚合
工作流的结构类似于 DAG（有向无环图），每个步骤可以引用上一步的输出：
用户输入
→ Step 1: 生成大纲（系统提示：你是专业写手）
→ Step 2: 展开第一部分（引用 Step 1 的输出）
→ Step 3: 展开第二部分（引用 Step 1 的输出）
→ Step 4: 翻译成英文（引用 Step 2 + Step 3 的输出）
→ 输出最终结果
核心实现
工作流定义模型
class WorkflowStep(BaseModel):
id: str
name: str
prompt_template: str # 支持 {{input}} 和 {{step_id.output}} 模板语法
model: str = "gpt-4"
temperature: float = 0.7
max_tokens: int = 2048
depends_on: list[str] = [] # 依赖的步骤 ID 列表
class Workflow(BaseModel):
id: str
name: str
description: str = ""
steps: list[WorkflowStep]
created_at: datetime
updated_at: datetime
工作流执行引擎
class WorkflowEngine:
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
async def execute(self, workflow: Workflow,
user_input: str) -> dict[str, str]:
outputs: dict[str, str] = {}
# Topological sort steps by dependencies
sorted_steps = self._topological_sort(workflow.steps)
for step in sorted_steps:
# 构建 Prompt：替换模板变量
prompt = step.prompt_template
prompt = prompt.replace("{{input}}", user_input)
for dep_id in step.depends_on:
placeholder = f"{{{{{dep_id}.output}}}}"
prompt = prompt.replace(placeholder, outputs.get(dep_id, ""))
# 调用 GPT API
response = await self.client.chat.completions.create(
model=step.model,
messages=[{"role": "user", "content": prompt}],
temperature=step.temperature,
max_tokens=step.max_tokens,
)
outputs[step.id] = response.choices[0].message.content
return outputs
模板变量解析
import re
class TemplateParser:
VARIABLE_PATTERN = re.compile(r"{{(w+).(input|output)}}")
@classmethod
def resolve(cls, template: str,
user_input: str,
step_outputs: dict[str, str]) -> str:
def replace_var(match: re.Match) -> str:
step_id = match.group(1)
var_type = match.group(2)
if step_id == "input":
return user_input if var_type == "input" else ""
return step_outputs.get(step_id, "")
return cls.VARIABLE_PATTERN.sub(replace_var, template)
前端工作流设计器（React）
function WorkflowDesigner({ workflow, onUpdate }) {
const [steps, setSteps] = useState(workflow.steps);
const addStep = () => {
const newStep = {
id: uuid(),
name: `步骤 ${steps.length + 1}`,
prompt_template: "",
model: "gpt-4",
temperature: 0.7,
max_tokens: 2048,
depends_on: [],
};
setSteps([...steps, newStep]);
};
return (
<div className="workflow-designer">
{steps.map((step, index) => (
<StepCard
key={step.id}
step={step}
stepIndex={index}
onUpdate={(updated) => {
const newSteps = [...steps];
newSteps[index] = updated;
setSteps(newSteps);
onUpdate({ ...workflow, steps: newSteps });
}}
/>
))}
<Button onClick={addStep}>添加步骤</Button>
</div>
);
}
踩坑记录
1. Token 限制
多步骤工作流中，每步的输出作为下步的输入，可能导致 Prompt 长度呈指数增长，超过 GPT 的上下文窗口。解决方案：在每个步骤中加入"摘要"指令，要求模型在生成完整输出的同时，也输出一个摘要版本供后续步骤使用。
2. 错误处理的粒度
工作流中间某个步骤失败，应该重试还是终止？如果终止，已经成功的步骤结果是否保留？解决方案：支持可配置的错误处理策略——"立即终止"（默认）、"重试 N 次"、"跳过继续"。同时保留每个步骤的输出，方便定位问题。
3. 任务编排的循环检测
用户设计工作流时可能创建循环依赖（A→B→A），导致引擎死循环。解决方案：在拓扑排序前进行循环依赖检测，发现循环时给出明确的错误提示。
4. Prompt 模板的转义问题
用户输入中如果包含 {{}} 字符串，会被模板引擎误解析为变量。解决方案：支持转义语法（如 \\{{literal}}），并在模板解析前做转义处理。
总结
GPT_workflow 是一个"让 AI 自动化更自动化的工具"。它解决的核心问题是：当 AI 任务从单步走向多步时，如何优雅地编排和管理。通过工作流引擎，用户可以将复杂的 AI 任务拆解为多个简单的步骤，每个步骤专注于一件事，最终组合成强大的自动化流水线。
这个项目让我深入理解了 Prompt Engineering 中的"链式推理"（Chain-of-Thought）概念——把大问题拆成小问题，一步步解决。同时也积累了任务编排、模板引擎、拓扑排序等方面的工程实践。未来可以考虑集成更多 LLM 提供商（Claude、Gemini），以及支持非线性的工作流结构（条件分支、并行执行等）。