🤖 第11课：任务规划

📌 任务规划概述

任务规划是服务机器人的大脑，将高层指令分解为可执行的原子动作序列：

📌 任务依赖与调度

import heapq, json

class Task:
    def __init__(self, tid, name, priority=0, deps=None, duration=30, location=None):
        self.id = tid; self.name = name; self.priority = priority
        self.deps = deps or []; self.duration = duration; self.location = location
        self.status = "pending"  # pending/running/done/failed

class TaskPlanner:
    def __init__(self):
        self.tasks = {}; self.completed = set()

    def add_task(self, task):
        self.tasks[task.id] = task

    def plan(self):
        """基于优先级和依赖的任务规划"""
        order = []; ready = []; remaining = set(self.tasks.keys())
        
        while remaining:
            for tid in list(remaining):
                task = self.tasks[tid]
                if all(d in self.completed for d in task.deps):
                    heapq.heappush(ready, (-task.priority, tid))
            
            if not ready:
                print("⚠️ 依赖循环，无法继续")
                break
            
            _, tid = heapq.heappop(ready)
            order.append(tid)
            self.completed.add(tid)
            remaining.remove(tid)
        
        return order

    def estimate_total_time(self, order):
        """估算总执行时间"""
        total = 0
        for tid in order:
            total += self.tasks[tid].duration
        return total

planner = TaskPlanner()
planner.add_task(Task("T1", "导航到咖啡机", 3, [], 45, "1楼茶水间"))
planner.add_task(Task("T2", "取咖啡", 2, ["T1"], 20, "1楼茶水间"))
planner.add_task(Task("T3", "导航到电梯", 3, ["T2"], 30, "1楼电梯口"))
planner.add_task(Task("T4", "乘电梯到3楼", 4, ["T3"], 60, "电梯"))
planner.add_task(Task("T5", "导航到会议室A", 3, ["T4"], 40, "3楼会议室A"))
planner.add_task(Task("T6", "递送咖啡", 5, ["T5"], 15, "3楼会议室A"))
planner.add_task(Task("T7", "语音确认送达", 2, ["T6"], 10, "3楼会议室A"))
planner.add_task(Task("T8", "返回待命点", 1, ["T7"], 50, "1楼前台"))

order = planner.plan()
total = planner.estimate_total_time(order)

print("任务规划 - 基于优先级和依赖")
print("=" * 55)
for i, tid in enumerate(order):
    t = planner.tasks[tid]
    deps = ",".join(t.deps) if t.deps else "无"
    print(f"  {i+1}. [{tid}] {t.name} (优先级P{t.priority}, 依赖:{deps}, {t.duration}s, {t.location})")
print(f"\n⏱️ 总预计时间: {total}秒 ({total/60:.1f}分钟)")
print("✅ 任务规划验证通过")

📌 行为树

行为树是游戏AI和服务机器人的主流任务表示方法，支持顺序/选择/装饰器等组合：

class BehaviorTreeNode:
    """行为树节点"""
    def __init__(self, name, node_type="sequence", children=None, action=None):
        self.name = name; self.type = node_type
        self.children = children or []; self.action = action
        self.status = "idle"

    def tick(self, context=None):
        if self.type == "sequence":
            for child in self.children:
                result = child.tick(context)
                if result != "success":
                    self.status = result
                    return result
            self.status = "success"
            return "success"
        
        elif self.type == "selector":
            for child in self.children:
                result = child.tick(context)
                if result != "failure":
                    self.status = result
                    return result
            self.status = "failure"
            return "failure"
        
        elif self.type == "action":
            result = self.action() if self.action else "success"
            self.status = result
            return result
        
        elif self.type == "decorator":
            if self.children:
                result = self.children[0].tick(context)
                self.status = "success" if result == "failure" else "failure"
                return self.status
            return "failure"

# 构建配送任务行为树
log = []
def act(name, result="success"):
    def f():
        log.append(f"执行: {name} → {result}")
        return result
    return f

# 配送任务行为树
root = BehaviorTreeNode("配送任务", "selector", [
    BehaviorTreeNode("正常流程", "sequence", [
        BehaviorTreeNode("导航到取物点", "action", action=act("导航到茶水间")),
        BehaviorTreeNode("取物品", "action", action=act("取咖啡")),
        BehaviorTreeNode("导航到电梯", "action", action=act("导航到电梯口")),
        BehaviorTreeNode("乘电梯", "selector", [
            BehaviorTreeNode("电梯正常", "sequence", [
                BehaviorTreeNode("呼叫电梯", "action", action=act("呼叫电梯")),
                BehaviorTreeNode("进入电梯", "action", action=act("进入电梯")),
                BehaviorTreeNode("到达目标层", "action", action=act("到达3楼")),
            ]),
            BehaviorTreeNode("走楼梯", "sequence", [
                BehaviorTreeNode("导航到楼梯", "action", action=act("导航到楼梯")),
                BehaviorTreeNode("爬楼梯", "action", action=act("爬楼梯到3楼")),
            ]),
        ]),
        BehaviorTreeNode("导航到目的地", "action", action=act("导航到会议室A")),
        BehaviorTreeNode("递送物品", "action", action=act("递送咖啡")),
    ]),
    BehaviorTreeNode("失败处理", "sequence", [
        BehaviorTreeNode("通知管理员", "action", action=act("通知管理员配送失败")),
        BehaviorTreeNode("返回待命点", "action", action=act("返回1楼前台")),
    ]),
])

print("行为树执行模拟")
print("=" * 55)
result = root.tick()
for entry in log:
    print(f"  {entry}")
print(f"\n最终状态: {result}")
print("✅ 行为树验证通过")

📌 状态机任务执行

有限状态机(FSM)是经典且可靠的任务执行模型：

class StateMachineTask:
    """状态机驱动的任务执行"""
    def __init__(self):
        self.states = {
            "IDLE": {"start": "NAVIGATE_PICKUP"},
            "NAVIGATE_PICKUP": {"arrived": "PICK_ITEM", "failed": "ERROR"},
            "PICK_ITEM": {"success": "NAVIGATE_ELEVATOR", "failed": "RETRY_PICK"},
            "RETRY_PICK": {"retry": "PICK_ITEM", "give_up": "ERROR"},
            "NAVIGATE_ELEVATOR": {"arrived": "CALL_ELEVATOR", "failed": "TRY_STAIRS"},
            "CALL_ELEVATOR": {"arrived": "ENTER_ELEVATOR", "timeout": "CALL_ELEVATOR", "failed": "TRY_STAIRS"},
            "ENTER_ELEVATOR": {"entered": "RIDE_ELEVATOR", "door_closed": "CALL_ELEVATOR"},
            "RIDE_ELEVATOR": {"arrived": "EXIT_ELEVATOR"},
            "EXIT_ELEVATOR": {"exited": "NAVIGATE_DEST"},
            "TRY_STAIRS": {"arrived": "NAVIGATE_DEST", "failed": "ERROR"},
            "NAVIGATE_DEST": {"arrived": "DELIVER", "failed": "ERROR"},
            "DELIVER": {"success": "CONFIRM", "failed": "RETRY_DELIVER"},
            "RETRY_DELIVER": {"retry": "DELIVER", "give_up": "NOTIFY_FAIL"},
            "CONFIRM": {"confirmed": "RETURN_HOME"},
            "RETURN_HOME": {"arrived": "IDLE"},
            "NOTIFY_FAIL": {"done": "RETURN_HOME"},
            "ERROR": {"reset": "IDLE"},
        }
        self.current = "IDLE"
        self.history = []

    def transition(self, event):
        state_trans = self.states.get(self.current, {})
        new_state = state_trans.get(event, self.current)
        self.history.append((self.current, event, new_state))
        self.current = new_state
        return new_state

    def get_available_events(self):
        return list(self.states.get(self.current, {}).keys())

sm = StateMachineTask()
print("状态机任务执行模拟")
print("=" * 55)

# 模拟正常流程
events = ["start","arrived","success","arrived","arrived","entered","arrived","exited","arrived","success","confirmed","arrived"]
for e in events:
    old = sm.current
    new = sm.transition(e)
    print(f"  {old} --{e}--> {new}")

print(f"\n状态历史: {len(sm.history)}次转换")
print("✅ 状态机验证通过")

📌 行为树 vs 状态机

📌 练习

📌 成就