今天我們要讓工作流程變得更智能！透過條件判斷與分支邏輯，AI 助理將能夠根據不同情況做出決策，選擇最適合的處理路徑。
在實際應用中，工作流程往往不是線性的：
條件分支讓工作流程更靈活、更智能，能夠應對各種複雜場景。
conditional_workflow/
├── main.py # 主程式
├── core/
│ ├── __init__.py
│ ├── conditional_engine.py # 條件判斷引擎
│ └── decision_rules.py # 決策規則定義
├── workflows/
│ ├── __init__.py
│ ├── smart_approval.py # 智能審批流程
│ └── content_router.py # 內容路由
└── utils/
├── __init__.py
└── condition_evaluator.py # 條件評估器
from typing import Dict, Any, Callable, List, Optional
from dataclasses import dataclass
from enum import Enum
import operator
class ConditionOperator(Enum):
"""條件運算符"""
EQUAL = "=="
NOT_EQUAL = "!="
GREATER = ">"
LESS = "<"
GREATER_EQUAL = ">="
LESS_EQUAL = "<="
CONTAINS = "contains"
NOT_CONTAINS = "not_contains"
IN = "in"
NOT_IN = "not_in"
@dataclass
class Condition:
"""條件定義"""
field: str # 要檢查的欄位
operator: ConditionOperator # 運算符
value: Any # 比較值
def evaluate(self, data: Dict[str, Any]) -> bool:
"""評估條件"""
field_value = self._get_field_value(data, self.field)
if field_value is None:
return False
# 執行對應的運算
operations = {
ConditionOperator.EQUAL: lambda a, b: a == b,
ConditionOperator.NOT_EQUAL: lambda a, b: a != b,
ConditionOperator.GREATER: lambda a, b: a > b,
ConditionOperator.LESS: lambda a, b: a < b,
ConditionOperator.GREATER_EQUAL: lambda a, b: a >= b,
ConditionOperator.LESS_EQUAL: lambda a, b: a <= b,
ConditionOperator.CONTAINS: lambda a, b: b in str(a),
ConditionOperator.NOT_CONTAINS: lambda a, b: b not in str(a),
ConditionOperator.IN: lambda a, b: a in b,
ConditionOperator.NOT_IN: lambda a, b: a not in b
}
op_func = operations.get(self.operator)
if op_func:
try:
return op_func(field_value, self.value)
except Exception as e:
print(f"條件評估錯誤: {e}")
return False
return False
def _get_field_value(self, data: Dict[str, Any], field: str) -> Any:
"""獲取欄位值，支援巢狀欄位"""
keys = field.split('.')
value = data
for key in keys:
if isinstance(value, dict):
value = value.get(key)
else:
return None
return value
@dataclass
class Branch:
"""分支定義"""
name: str
conditions: List[Condition]
condition_logic: str = "AND" # AND 或 OR
action: Callable = None
next_branch: str = None
def should_execute(self, data: Dict[str, Any]) -> bool:
"""判斷是否應該執行此分支"""
if not self.conditions:
return True # 沒有條件則總是執行
results = [cond.evaluate(data) for cond in self.conditions]
if self.condition_logic == "AND":
return all(results)
elif self.condition_logic == "OR":
return any(results)
else:
return False
class ConditionalEngine:
"""條件判斷引擎"""
def __init__(self):
self.branches = {}
self.execution_history = []
def add_branch(self, branch: Branch) -> None:
"""添加分支"""
self.branches[branch.name] = branch
def execute(self, start_branch: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""執行條件工作流程"""
current_branch = start_branch
execution_path = []
results = {}
while current_branch:
if current_branch not in self.branches:
print(f"⚠️ 分支 {current_branch} 不存在")
break
branch = self.branches[current_branch]
execution_path.append(current_branch)
print(f"🔍 評估分支: {branch.name}")
if branch.should_execute(data):
print(f"✅ 條件滿足，執行分支: {branch.name}")
if branch.action:
try:
result = branch.action(data)
results[branch.name] = result
# 更新數據上下文
if isinstance(result, dict):
data.update(result)
except Exception as e:
print(f"❌ 分支執行失敗: {e}")
results[branch.name] = {'error': str(e)}
current_branch = branch.next_branch
else:
print(f"❌ 條件不滿足，跳過分支: {branch.name}")
current_branch = None
execution_record = {
'execution_path': execution_path,
'results': results,
'final_data': data
}
self.execution_history.append(execution_record)
return execution_record
def find_matching_branch(self, data: Dict[str, Any],
branch_names: List[str]) -> Optional[str]:
"""從多個分支中找到第一個滿足條件的"""
for branch_name in branch_names:
if branch_name in self.branches:
branch = self.branches[branch_name]
if branch.should_execute(data):
return branch_name
return None
from core.conditional_engine import ConditionalEngine, Condition, Branch, ConditionOperator
from typing import Dict, Any
import google.generativeai as genai
import os
genai.configure(api_key=os.getenv('GEMINI_API_KEY'))
model = genai.GenerativeModel('gemini-2.5-flash')
class SmartApprovalWorkflow:
"""智能審批工作流程"""
def __init__(self):
self.engine = ConditionalEngine()
self._setup_approval_workflow()
def _setup_approval_workflow(self):
"""設置審批工作流程"""
# 1. 初始檢查分支
initial_check = Branch(
name="initial_check",
conditions=[], # 無條件執行
action=self.validate_request,
next_branch="amount_check"
)
# 2. 金額檢查分支（小額自動通過）
auto_approve = Branch(
name="auto_approve",
conditions=[
Condition("amount", ConditionOperator.LESS_EQUAL, 1000),
Condition("is_valid", ConditionOperator.EQUAL, True)
],
condition_logic="AND",
action=self.auto_approve_request,
next_branch=None # 流程結束
)
# 3. 中額主管審批
manager_review = Branch(
name="manager_review",
conditions=[
Condition("amount", ConditionOperator.GREATER, 1000),
Condition("amount", ConditionOperator.LESS_EQUAL, 10000),
Condition("is_valid", ConditionOperator.EQUAL, True)
],
condition_logic="AND",
action=self.manager_approval,
next_branch="risk_assessment"
)
# 4. 高額多級審批
senior_review = Branch(
name="senior_review",
conditions=[
Condition("amount", ConditionOperator.GREATER, 10000),
Condition("is_valid", ConditionOperator.EQUAL, True)
],
condition_logic="AND",
action=self.senior_management_approval,
next_branch="risk_assessment"
)
# 5. 風險評估分支
risk_assessment = Branch(
name="risk_assessment",
conditions=[
Condition("manager_approved", ConditionOperator.EQUAL, True)
],
action=self.assess_risk,
next_branch="final_decision"
)
# 6. 最終決策
final_decision = Branch(
name="final_decision",
conditions=[],
action=self.make_final_decision,
next_branch=None
)
# 添加所有分支
for branch in [initial_check, auto_approve, manager_review,
senior_review, risk_assessment, final_decision]:
self.engine.add_branch(branch)
def validate_request(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""驗證申請資料"""
print("📋 驗證申請資料...")
required_fields = ['applicant', 'amount', 'purpose']
is_valid = all(field in data and data[field] for field in required_fields)
if not is_valid:
return {
'is_valid': False,
'validation_message': '申請資料不完整'
}
# 檢查金額是否合理
if data['amount'] <= 0:
return {
'is_valid': False,
'validation_message': '金額必須大於零'
}
return {
'is_valid': True,
'validation_message': '資料驗證通過',
'validated_at': '2024-01-15 10:00:00'
}
def auto_approve_request(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""自動批准（小額）"""
print("✅ 小額申請，自動批准")
return {
'approval_status': 'approved',
'approval_level': 'auto',
'approved_amount': data['amount'],
'approval_message': f"申請金額 NT${data['amount']} 已自動批准",
'approved_at': '2024-01-15 10:01:00'
}
def manager_approval(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""主管審批"""
print("👔 提交主管審批...")
# 使用 Gemini 模擬主管審批邏輯
prompt = f"""
作為主管，審核以下申請：
申請人：{data['applicant']}
金額：NT$ {data['amount']}
用途：{data['purpose']}
請評估是否批准，只回答 approved 或 rejected 和簡短理由。
"""
try:
response = model.generate_content(prompt)
decision = response.text.strip().lower()
approved = 'approved' in decision or '批准' in decision
return {
'manager_approved': approved,
'approval_level': 'manager',
'manager_comment': response.text,
'reviewed_at': '2024-01-15 10:05:00'
}
except:
return {
'manager_approved': True, # 預設批准
'approval_level': 'manager',
'manager_comment': '主管已審核',
'reviewed_at': '2024-01-15 10:05:00'
}
def senior_management_approval(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""高層審批"""
print("👨💼 提交高層審批...")
return {
'manager_approved': True, # 簡化處理
'approval_level': 'senior',
'senior_comment': f"高額申請 NT${data['amount']} 需要額外審核",
'requires_board_approval': data['amount'] > 50000,
'reviewed_at': '2024-01-15 10:10:00'
}
def assess_risk(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""風險評估"""
print("⚠️ 進行風險評估...")
# 簡單的風險評分
risk_score = 0
if data['amount'] > 5000:
risk_score += 2
if data['amount'] > 20000:
risk_score += 3
risk_level = 'low' if risk_score < 2 else 'medium' if risk_score < 4 else 'high'
return {
'risk_score': risk_score,
'risk_level': risk_level,
'risk_assessment': f"風險等級：{risk_level}，評分：{risk_score}/10"
}
def make_final_decision(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""最終決策"""
print("🎯 做出最終決策...")
# 綜合考慮各項因素
approved = (
data.get('manager_approved', False) and
data.get('risk_level', 'high') != 'high'
)
if approved:
message = f"""
✅ **申請已批准**
申請人：{data['applicant']}
批准金額：NT$ {data['amount']}
審批等級：{data.get('approval_level', 'standard')}
風險評估：{data.get('risk_level', 'low')}
備註：{data.get('purpose', '無')}
"""
else:
message = f"""
❌ **申請被拒絕**
申請人：{data['applicant']}
申請金額：NT$ {data['amount']}
拒絕原因：風險等級過高或未通過審核
請修改申請內容後重新提交。
"""
return {
'final_decision': 'approved' if approved else 'rejected',
'decision_message': message
}
def process_approval(self, request_data: Dict[str, Any]) -> Dict[str, Any]:
"""處理審批申請"""
print(f"\n🚀 開始處理審批申請...")
print(f"申請人：{request_data.get('applicant', '未知')}")
print(f"金額：NT$ {request_data.get('amount', 0)}")
print("-" * 50)
# 決定起始分支
start_branch = "initial_check"
# 執行工作流程
result = self.engine.execute(start_branch, request_data)
# 根據金額選擇審批路徑
if request_data.get('is_valid'):
amount = request_data.get('amount', 0)
if amount <= 1000:
self.engine.execute("auto_approve", request_data)
elif amount <= 10000:
self.engine.execute("manager_review", request_data)
else:
self.engine.execute("senior_review", request_data)
print("-" * 50)
print("✅ 審批流程完成")
return result
from core.conditional_engine import ConditionalEngine, Condition, Branch, ConditionOperator
from typing import Dict, Any
class ContentRouter:
"""智能內容路由器"""
def __init__(self):
self.engine = ConditionalEngine()
self._setup_routing_rules()
def _setup_routing_rules(self):
"""設置路由規則"""
# 緊急內容快速通道
urgent_branch = Branch(
name="urgent_processing",
conditions=[
Condition("priority", ConditionOperator.EQUAL, "urgent")
],
action=self.handle_urgent_content,
next_branch=None
)
# 圖片內容處理
image_branch = Branch(
name="image_processing",
conditions=[
Condition("content_type", ConditionOperator.EQUAL, "image")
],
action=self.handle_image_content,
next_branch=None
)
# 文字內容處理
text_branch = Branch(
name="text_processing",
conditions=[
Condition("content_type", ConditionOperator.EQUAL, "text")
],
action=self.handle_text_content,
next_branch=None
)
# 預設處理
default_branch = Branch(
name="default_processing",
conditions=[], # 無條件
action=self.handle_default_content,
next_branch=None
)
for branch in [urgent_branch, image_branch, text_branch, default_branch]:
self.engine.add_branch(branch)
def handle_urgent_content(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""處理緊急內容"""
print("🚨 緊急內容，快速處理通道")
return {
'processing_time': '< 1分鐘',
'handler': 'urgent_team',
'status': 'processing'
}
def handle_image_content(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""處理圖片內容"""
print("🖼️ 圖片內容，視覺處理")
return {
'processing_time': '2-5分鐘',
'handler': 'vision_ai',
'status': 'analyzing'
}
def handle_text_content(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""處理文字內容"""
print("📝 文字內容，NLP 處理")
return {
'processing_time': '1-3分鐘',
'handler': 'text_ai',
'status': 'processing'
}
def handle_default_content(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""預設處理"""
print("📦 一般內容，標準處理")
return {
'processing_time': '3-10分鐘',
'handler': 'general_queue',
'status': 'queued'
}
def route_content(self, content_data: Dict[str, Any]) -> Dict[str, Any]:
"""路由內容到適當的處理器"""
print(f"\n🔀 內容路由中...")
print(f"類型：{content_data.get('content_type', '未知')}")
print(f"優先級：{content_data.get('priority', '一般')}")
# 按優先級檢查分支
branch_order = ["urgent_processing", "image_processing",
"text_processing", "default_processing"]
matched_branch = self.engine.find_matching_branch(content_data, branch_order)
if matched_branch:
result = self.engine.execute(matched_branch, content_data)
return result
return {'error': '無法路由內容'}
from workflows.smart_approval import SmartApprovalWorkflow
from workflows.content_router import ContentRouter
def demo_approval_workflow():
"""示範審批工作流程"""
print("=" * 60)
print("📋 智能審批工作流程示範")
print("=" * 60)
approval = SmartApprovalWorkflow()
# 測試案例
test_cases = [
{
'applicant': '張三',
'amount': 500,
'purpose': '辦公用品採購'
},
{
'applicant': '李四',
'amount': 5000,
'purpose': '團隊建設活動'
},
{
'applicant': '王五',
'amount': 50000,
'purpose': '設備升級採購'
}
]
for i, case in enumerate(test_cases, 1):
print(f"\n案例 {i}:")
result = approval.process_approval(case)
if 'decision_message' in result['final_data']:
print(result['final_data']['decision_message'])
def demo_content_routing():
"""示範內容路由"""
print("\n" + "=" * 60)
print("🔀 智能內容路由示範")
print("=" * 60)
router = ContentRouter()
test_contents = [
{'content_type': 'text', 'priority': 'normal', 'content': '一般文字內容'},
{'content_type': 'image', 'priority': 'normal', 'content': 'image.jpg'},
{'content_type': 'text', 'priority': 'urgent', 'content': '緊急通知！'}
]
for content in test_contents:
result = router.route_content(content)
print(f"處理結果：{result['final_data']}\n")
def main():
"""主程式"""
print("🔀 條件判斷與分支邏輯系統")
print("智能路由、動態決策、靈活流程")
print("=" * 60)
while True:
print("\n選擇示範功能：")
print("1. 智能審批工作流程")
print("2. 智能內容路由")
print("3. 退出")
choice = input("\n請選擇 (1-3)：").strip()
try:
if choice == '1':
demo_approval_workflow()
elif choice == '2':
demo_content_routing()
elif choice == '3':
print("👋 再見！")
break
else:
print("❌ 無效的選擇")
except KeyboardInterrupt:
print("\n👋 再見！")
break
except Exception as e:
print(f"❌ 發生錯誤：{e}")
if __name__ == "__main__":
main()
✅ 靈活的條件判斷：支援多種運算符和邏輯組合
✅ 動態分支選擇：根據數據自動選擇處理路徑
✅ 多級審批流程：根據金額自動分配審批等級
✅ 智能路由：內容類型和優先級的智能分發
✅ 可擴展架構：輕鬆添加新的條件和分支
案例 1:
🚀 開始處理審批申請...
申請人：張三
金額：NT$ 500
--------------------------------------------------
📋 驗證申請資料...
🔍 評估分支: initial_check
✅ 條件滿足，執行分支: initial_check
✅ 小額申請，自動批准
--------------------------------------------------
✅ 審批流程完成
✅ **申請已批准**
申請人：張三
批准金額：NT$ 500
審批等級：auto
今天我們學會了如何在工作流程中加入條件判斷與分支邏輯，讓 AI 助理能夠根據不同情況做出智能決策。明天我們將學習並行處理與效能優化，讓系統運行更快更有效率！
iThome鐵人賽
看影片追技術