3 min read680 words

Workflow 与 Agent 的产品设计

选 Workflow 还是 Agent，不是技术架构问题，是产品风险判断问题。选错了，要么系统僵硬跑不动，要么 Agent 失控出事故。

Workflow 与 Agent 的选择决策树

graph TD A[新的 AI 任务需求] --> B{任务步骤是否固定？} B -- 是 --> C[Workflow 方案] B -- 否 --> D{任务失败成本？} D -- 高风险/不可逆 --> E[Workflow + 人工节点] D -- 可接受/可回滚 --> F{步骤数量？} F -- 少于 5 步 --> G[Workflow 也可以] F -- 超过 5 步且路径多变 --> H[Agent 方案] C --> I[定义固定节点与触发条件] E --> J[插入 Human-in-the-loop 审批节点] H --> K{是否有足够的测试覆盖？} K -- 否 --> L[先做 Workflow 原型验证] K -- 是 --> M[Agent 设计 + 降级机制] style A fill:#e3f2fd,stroke:#1565c0 style C fill:#e8f5e9,stroke:#2e7d32 style H fill:#fff3e0,stroke:#e65100 style L fill:#fff9c4,stroke:#f9a825

Workflow vs Agent 核心差异

维度	Workflow（确定性流程）	Agent（自主规划）
步骤路径	预定义，有限分支	动态规划，路径不固定
适用场景	审批流、数据处理管道、表单填写	复杂调研、多工具协作、开放任务
失败模式	可预测，易排查	不可预测，难复现
成本控制	精确可算	难以预估，易超支
上线门槛	低，工程师熟悉	高，需要专项测试
适合阶段	MVP、稳定期	探索期、内部工具

Human-in-the-loop 的四个插入点

不是所有 Agent 任务都应该完全自动化。PM 需要判断在哪里加人工节点：

graph LR A[任务开始] --> B[AI 规划步骤] B --> C{高风险操作？} C -- 是 --> D[人工确认 ①] C -- 否 --> E[AI 执行] E --> F{输出置信度低？} F -- 是 --> G[人工审核 ②] F -- 否 --> H[继续下一步] H --> I{任务完成？} I -- 否 --> B I -- 是 --> J{涉及对外发布？} J -- 是 --> K[人工终审 ③] J -- 否 --> L[结束] K --> M{异常情况？} M -- 是 --> N[人工接管全流程 ④] M -- 否 --> L

降级设计：Agent 产品的生命线

Agent 产品最容易犯的错误是"只设计 Happy Path"。PM 必须为每个 Agent 产品定义：

降级触发条件：置信度低于阈值 / 连续失败次数 / 执行时间超限
降级行为：自动切换到 Workflow 模式 / 暂停并通知人工 / 返回最近成功状态
恢复机制：人工干预后如何重新进入 Agent 流程

Python 示例：Workflow/Agent 设计顾问

"""
Workflow vs Agent 设计决策顾问
PM 视角：根据任务特征推荐架构并生成设计要点
"""
from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum
class ArchType(Enum):
WORKFLOW = "Workflow（确定性流程）"
AGENT = "Agent（自主规划）"
WORKFLOW_WITH_HITL = "Workflow + Human-in-the-loop"
HYBRID = "混合架构（Workflow 主干 + 局部 Agent）"
@dataclass
class TaskProfile:
"""任务特征描述"""
name: str
steps_fixed: bool           # 步骤是否固定
step_count_max: int         # 最大步骤数
failure_reversible: bool    # 失败是否可回滚
failure_cost: str           # 失败代价: low / medium / high
has_external_actions: bool  # 是否涉及外部操作（发邮件、转账等）
knowledge_required: bool    # 是否需要复杂推理
team_ai_maturity: str       # 团队 AI 成熟度: low / medium / high
@dataclass
class DesignRecommendation:
"""设计建议"""
task_name: str
recommended_arch: ArchType
confidence: str             # high / medium / low
key_design_points: List[str] = field(default_factory=list)
hitl_points: List[str] = field(default_factory=list)
fallback_strategy: List[str] = field(default_factory=list)
risks: List[str] = field(default_factory=list)
def decide_architecture(profile: TaskProfile) -> ArchType:
"""根据任务特征决定架构类型"""
# 强制 Workflow 的条件
if profile.failure_cost == "high" and not profile.failure_reversible:
if profile.steps_fixed:
return ArchType.WORKFLOW_WITH_HITL
else:
return ArchType.WORKFLOW_WITH_HITL
# 强制 Workflow 的条件：团队 AI 成熟度低
if profile.team_ai_maturity == "low":
return ArchType.WORKFLOW
# 适合 Agent 的条件
if (not profile.steps_fixed
and profile.step_count_max > 7
and profile.knowledge_required
and profile.team_ai_maturity == "high"):
if profile.has_external_actions:
return ArchType.WORKFLOW_WITH_HITL
return ArchType.AGENT
# 混合架构
if not profile.steps_fixed and profile.step_count_max > 4:
return ArchType.HYBRID
return ArchType.WORKFLOW
def generate_hitl_points(profile: TaskProfile, arch: ArchType) -> List[str]:
"""生成 Human-in-the-loop 插入建议"""
points = []
if arch in (ArchType.WORKFLOW_WITH_HITL, ArchType.HYBRID):
if profile.has_external_actions:
points.append("外部操作执行前需人工确认（如发送通知、修改数据）")
if profile.failure_cost in ("medium", "high"):
points.append("AI 输出置信度 < 0.75 时，触发人工审核队列")
if not profile.failure_reversible:
points.append("所有不可逆操作节点必须设置审批步骤，审批记录存档")
points.append("设置人工接管入口，任意节点支持人工覆盖 AI 决策")
return points if points else ["当前架构风险可控，暂不需要强制人工节点"]
def generate_fallback_strategy(profile: TaskProfile, arch: ArchType) -> List[str]:
"""生成降级策略"""
strategies = []
if arch in (ArchType.AGENT, ArchType.HYBRID):
strategies.append("Agent 连续失败 3 次后，自动切换到 Workflow 备用路径")
strategies.append("单次执行超过 30 秒，触发超时降级并通知用户")
strategies.append("Agent 调用工具失败率 > 20%，暂停 Agent 模式并告警")
if profile.failure_cost == "high":
strategies.append("任何异常状态下，保留最近一次成功的状态快照供回滚")
strategies.append("所有降级事件记录到监控系统，每周 Review 频率和原因")
return strategies
def generate_design_points(profile: TaskProfile, arch: ArchType) -> List[str]:
"""生成关键设计要点"""
points = []
if arch == ArchType.WORKFLOW:
points.extend([
"绘制完整流程图，标注每个节点的输入/输出/失败处理",
"定义每个节点的超时时间和重试次数",
"为每个分支条件写明判断逻辑，避免歧义",
])
elif arch == ArchType.AGENT:
points.extend([
"明确 Agent 可调用的工具清单和每个工具的权限边界",
"定义 Agent 的最大步骤数上限，防止无限循环",
"设计 Agent 执行日志格式，支持事后审查每一步决策",
"先用小规模（10 个案例）验证 Agent 逻辑再扩大流量",
])
elif arch == ArchType.WORKFLOW_WITH_HITL:
points.extend([
"标注所有人工节点的 SLA（如 1 小时内必须处理）",
"设计人工节点的待办界面，展示 AI 决策依据方便审核",
"人工拒绝时，定义流程如何回退或终止",
])
elif arch == ArchType.HYBRID:
points.extend([
"明确 Workflow 主干节点和 Agent 局部节点的边界",
"Agent 局部节点必须有确定性的输入/输出接口定义",
"Workflow 主干对 Agent 节点的结果做格式校验再传递",
])
return points
def recommend_design(profile: TaskProfile) -> DesignRecommendation:
"""生成完整设计建议"""
arch = decide_architecture(profile)
risks = []
if not profile.steps_fixed and arch == ArchType.AGENT:
risks.append("Agent 路径不固定，测试覆盖度难以保证，需专项测试计划")
if profile.has_external_actions and arch == ArchType.AGENT:
risks.append("Agent 执行外部操作存在不可逆风险，必须加人工确认节点")
if profile.team_ai_maturity == "low" and arch != ArchType.WORKFLOW:
risks.append("团队 AI 经验不足，建议先从 Workflow 积累经验再升级架构")
confidence = "high" if profile.steps_fixed else \
"medium" if profile.team_ai_maturity != "low" else "low"
return DesignRecommendation(
task_name=profile.name,
recommended_arch=arch,
confidence=confidence,
key_design_points=generate_design_points(profile, arch),
hitl_points=generate_hitl_points(profile, arch),
fallback_strategy=generate_fallback_strategy(profile, arch),
risks=risks,
)
def print_recommendation(rec: DesignRecommendation):
print(f"\n{'='*55}")
print(f"  任务：{rec.task_name}")
print(f"{'='*55}")
print(f"  推荐架构：{rec.recommended_arch.value}")
print(f"  判断置信度：{rec.confidence}")
print("\n  关键设计要点：")
for p in rec.key_design_points:
print(f"    • {p}")
print("\n  Human-in-the-loop 设计：")
for p in rec.hitl_points:
print(f"    • {p}")
print("\n  降级策略：")
for s in rec.fallback_strategy:
print(f"    • {s}")
if rec.risks:
print("\n  注意风险：")
for r in rec.risks:
print(f"    ⚠ {r}")
# ── 演示 ──────────────────────────────────────────────
if __name__ == "__main__":
tasks = [
TaskProfile(
name="客服工单自动分类与路由",
steps_fixed=True, step_count_max=3,
failure_reversible=True, failure_cost="low",
has_external_actions=False, knowledge_required=False,
team_ai_maturity="medium",
),
TaskProfile(
name="合同审查与风险标注 Agent",
steps_fixed=False, step_count_max=12,
failure_reversible=False, failure_cost="high",
has_external_actions=True, knowledge_required=True,
team_ai_maturity="medium",
),
TaskProfile(
name="市场调研报告自动生成",
steps_fixed=False, step_count_max=8,
failure_reversible=True, failure_cost="low",
has_external_actions=False, knowledge_required=True,
team_ai_maturity="high",
),
]
for task in tasks:
rec = recommend_design(task)
print_recommendation(rec)

本章 checklist

[ ] 在方案评审时，明确要求工程师说明"为什么选 Agent 而不是 Workflow"
[ ] 为每个 Agent 任务定义最大步骤数上限和超时时间
[ ] 识别所有不可逆操作节点，并确认是否有人工确认步骤
[ ] 确认降级方案已设计：Agent 失败后系统如何响应、用户看到什么
[ ] 新 Agent 功能上线前，先在内部环境跑至少 50 个真实案例验证

本章小结

Workflow 是确定性的，可测试、可排查、成本可控，大多数 AI 产品功能优先选 Workflow
Agent 适合路径多变、需要复杂推理的场景，但必须配套降级策略和人工接管机制
PM 对这两类架构的最大贡献是定义"什么情况下 AI 不应该自己决定"，而不是让工程师自己判断

本章完：下一章进入 07-场景拆解与流程设计，学习如何从用户旅程拆解出可执行的 AI 产品流程。