流程图设计与异常分支处理

只画 Happy Path 的流程图是废图。AI 产品的核心挑战恰恰在异常分支——模型不确定、用户输入不规范、外部服务超时，这些情况不设计好，上线后就是客诉。

AI 产品流程图的完整结构

流程节点的标准描述格式

每个节点都应该用以下四要素描述，而不只是一个标签名：

五类常见异常分支

PM 在画流程图时，必须逐一确认这五类异常是否有处理方案：

Python 示例：流程分析报告生成器

"""
AI 产品流程图分析工具
PM 视角：对一个流程设计进行完整性检查，输出结构化报告
"""
from dataclasses import dataclass, field
from typing import List, Optional, Dict
from enum import Enum
class NodeType(Enum):
INPUT = "输入节点"
DECISION = "判断节点"
AI_ACTION = "AI 执行节点"
HUMAN = "人工节点"
OUTPUT = "输出节点"
ERROR = "异常处理节点"
@dataclass
class FlowNode:
"""流程节点"""
node_id: str
name: str
node_type: NodeType
input_desc: str
action_desc: str
output_desc: str
failure_modes: List[str] = field(default_factory=list)
fallback_action: Optional[str] = None
timeout_ms: Optional[int] = None
has_monitoring: bool = False
@dataclass
class FlowDefinition:
"""流程定义"""
flow_name: str
nodes: List[FlowNode]
has_input_validation: bool = False
has_confidence_check: bool = False
has_output_validation: bool = False
has_human_escalation: bool = False
has_retry_mechanism: bool = False
@dataclass
class FlowAnalysisReport:
"""流程分析报告"""
flow_name: str
completeness_score: int    # 完整性得分 0-100
missing_elements: List[str] = field(default_factory=list)
node_issues: Dict[str, List[str]] = field(default_factory=dict)
suggestions: List[str] = field(default_factory=list)
def check_node_completeness(node: FlowNode) -> List[str]:
"""检查单个节点的完整性问题"""
issues = []
if not node.input_desc or node.input_desc.strip() == "":
issues.append("缺少输入描述")
if not node.action_desc or node.action_desc.strip() == "":
issues.append("缺少动作描述")
if not node.output_desc or node.output_desc.strip() == "":
issues.append("缺少输出描述")
if node.node_type == NodeType.AI_ACTION:
if not node.failure_modes:
issues.append("AI 节点未定义失败模式")
if node.fallback_action is None:
issues.append("AI 节点未设计兜底行为")
if node.timeout_ms is None:
issues.append("AI 节点未定义超时时间")
if not node.has_monitoring:
issues.append("AI 节点未接入监控")
return issues
def analyze_flow(flow: FlowDefinition) -> FlowAnalysisReport:
"""分析流程设计完整性"""
report = FlowAnalysisReport(flow_name=flow.flow_name)
deductions = 0
missing = []
# 检查流程级别的机制
if not flow.has_input_validation:
missing.append("缺少输入合法性校验机制")
deductions += 15
if not flow.has_confidence_check:
missing.append("缺少 AI 置信度检查与低置信度处理")
deductions += 15
if not flow.has_output_validation:
missing.append("缺少输出结果质量校验")
deductions += 10
if not flow.has_human_escalation:
missing.append("缺少转人工的升级路径")
deductions += 15
if not flow.has_retry_mechanism:
missing.append("缺少失败重试机制")
deductions += 10
report.missing_elements = missing
# 检查各节点
for node in flow.nodes:
issues = check_node_completeness(node)
if issues:
report.node_issues[f"{node.node_id} ({node.name})"] = issues
deductions += len(issues) * 3
# 生成建议
ai_nodes = [n for n in flow.nodes if n.node_type == NodeType.AI_ACTION]
if len(ai_nodes) > 5:
report.suggestions.append(
f"流程包含 {len(ai_nodes)} 个 AI 节点，建议分拆为子流程，降低单流程复杂度"
)
nodes_without_timeout = [
n.name for n in ai_nodes if n.timeout_ms is None
]
if nodes_without_timeout:
report.suggestions.append(
f"以下 AI 节点缺少超时配置：{', '.join(nodes_without_timeout)}"
)
nodes_without_monitor = [
n.name for n in ai_nodes if not n.has_monitoring
]
if nodes_without_monitor:
report.suggestions.append(
f"以下节点未接入监控，建议上线前配置：{', '.join(nodes_without_monitor)}"
)
if not report.suggestions:
report.suggestions.append("流程结构完整，可进入详细评审阶段")
report.completeness_score = max(0, 100 - deductions)
return report
def print_report(report: FlowAnalysisReport):
grade = (
"优秀" if report.completeness_score >= 85 else
"良好" if report.completeness_score >= 70 else
"需修订" if report.completeness_score >= 50 else
"重新设计"
)
print(f"\n{'='*55}")
print(f"  流程分析报告：{report.flow_name}")
print(f"{'='*55}")
print(f"  完整性得分：{report.completeness_score}/100  [{grade}]")
if report.missing_elements:
print(f"\n  流程级别缺失项（{len(report.missing_elements)} 条）：")
for m in report.missing_elements:
print(f"    ✗ {m}")
if report.node_issues:
print(f"\n  节点问题（共 {len(report.node_issues)} 个节点有问题）：")
for node_name, issues in report.node_issues.items():
print(f"    [{node_name}]")
for issue in issues:
print(f"      - {issue}")
print(f"\n  改进建议：")
for s in report.suggestions:
print(f"    → {s}")
# ── 演示 ──────────────────────────────────────────────
if __name__ == "__main__":
# 示例：一个不完整的客服 AI 流程
incomplete_flow = FlowDefinition(
flow_name="客服智能回复流程（初版）",
has_input_validation=False,
has_confidence_check=False,
has_output_validation=True,
has_human_escalation=True,
has_retry_mechanism=False,
nodes=[
FlowNode(
node_id="N1", name="接收用户消息",
node_type=NodeType.INPUT,
input_desc="用户文本消息",
action_desc="接收并存储消息",
output_desc="原始消息字符串",
),
FlowNode(
node_id="N2", name="意图识别",
node_type=NodeType.AI_ACTION,
input_desc="用户消息",
action_desc="调用意图分类模型",
output_desc="意图标签 + 置信度",
failure_modes=["模型超时", "返回格式错误"],
fallback_action="转人工处理",
timeout_ms=2000,
has_monitoring=True,
),
FlowNode(
node_id="N3", name="生成回复",
node_type=NodeType.AI_ACTION,
input_desc="意图标签 + 历史对话",
action_desc="调用 LLM 生成回复",
output_desc="",          # 故意留空演示
failure_modes=[],        # 故意留空演示
fallback_action=None,    # 故意留空演示
timeout_ms=None,         # 故意留空演示
has_monitoring=False,
),
],
)
report = analyze_flow(incomplete_flow)
print_report(report)

本章 checklist

• [ ] 流程图已包含 Happy Path 和至少 3 种异常分支
• [ ] 每个 AI 节点都定义了超时时间、失败模式和兜底行为
• [ ] 流程中有明确的"转人工"路径，且定义了触发条件
• [ ] 输入节点有合法性校验，输出节点有质量检查
• [ ] 流程图已通过工程师评审，确认技术可实现性

本章小结

• 流程图的价值不在于"好看"，而在于把所有人的假设显式化，让异常分支在纸上就能发现
• 每个 AI 节点必须有失败模式和兜底行为，这是 PM 在评审中必须逐一确认的，不能跳过
• 五类常见异常（输入异常、意图不明、模型超时、输出不达标、外部依赖失败）是流程设计的基本检查清单

下一节：03-多角色协同场景设计实战 — 当用户、AI、人工客服、系统同时出现在一个流程里，如何设计职责边界和交接机制。