无人机群体智能与分布式协同

无人机核心课程 · 大学前沿篇

第一章：群体智能基础

1.1 涌现行为

群体智能的核心思想：

个体简单 → 群体复杂行为

自然界的例子：
├── 鸟群：无领导者，却能形成整齐队形
├── 蚁群：找到最短路径
├── 蜂群：集体决策（选择新巢穴）
└── 鱼群：快速躲避捕食者

关键原理：
├── 局部交互：个体只与邻居通信
├── 正反馈：放大好的行为
├── 负反馈：抑制过度行为
└── 随机性：探索新可能性

1.2 Reynolds Boids 模型

import numpy as np

class Boid:
    """
    Reynolds Boids 模型
    
    三条简单规则产生复杂群体行为：
    1. 分离（Separation）：避免碰撞
    2. 对齐（Alignment）：与邻居方向一致
    3. 凝聚（Cohesion）：向邻居中心移动
    """
    
    def __init__(self, position, velocity):
        self.position = np.array(position, dtype=float)
        self.velocity = np.array(velocity, dtype=float)
        self.acceleration = np.zeros(2)
    
    def separation(self, neighbors, radius=5.0):
        """分离：避免与邻居太近"""
        force = np.zeros(2)
        for other in neighbors:
            distance = np.linalg.norm(self.position - other.position)
            if 0 < distance < radius:
                # 远离邻居
                force += (self.position - other.position) / distance
        return force
    
    def alignment(self, neighbors, radius=10.0):
        """对齐：与邻居速度一致"""
        if not neighbors:
            return np.zeros(2)
        
        avg_velocity = np.mean([n.velocity for n in neighbors], axis=0)
        return avg_velocity - self.velocity
    
    def cohesion(self, neighbors, radius=10.0):
        """凝聚：向邻居中心移动"""
        if not neighbors:
            return np.zeros(2)
        
        center = np.mean([n.position for n in neighbors], axis=0)
        return center - self.position
    
    def update(self, neighbors, max_speed=2.0, max_force=0.1):
        """更新位置和速度"""
        # 计算三个力
        sep = self.separation(neighbors) * 1.5
        ali = self.alignment(neighbors) * 1.0
        coh = self.cohesion(neighbors) * 1.0
        
        # 合成加速度
        self.acceleration = sep + ali + coh
        
        # 限制加速度
        if np.linalg.norm(self.acceleration) > max_force:
            self.acceleration = max_force * self.acceleration / np.linalg.norm(self.acceleration)
        
        # 更新速度
        self.velocity += self.acceleration
        
        # 限制速度
        if np.linalg.norm(self.velocity) > max_speed:
            self.velocity = max_speed * self.velocity / np.linalg.norm(self.velocity)
        
        # 更新位置
        self.position += self.velocity


class BoidSwarm:
    """群体仿真"""
    
    def __init__(self, n_boids=100):
        self.boids = [
            Boid(
                position=np.random.uniform(0, 100, 2),
                velocity=np.random.uniform(-1, 1, 2)
            )
            for _ in range(n_boids)
        ]
    
    def get_neighbors(self, boid, radius=10.0):
        """获取邻居"""
        neighbors = []
        for other in self.boids:
            if other is boid:
                continue
            distance = np.linalg.norm(boid.position - other.position)
            if distance < radius:
                neighbors.append(other)
        return neighbors
    
    def step(self):
        """仿真一步"""
        for boid in self.boids:
            neighbors = self.get_neighbors(boid)
            boid.update(neighbors)

第二章：分布式任务分配

2.1 拍卖算法

class TaskAuction:
    """
    拍卖算法用于多无人机任务分配
    
    思想：
    ├── 每个任务拍卖给最高出价者
    ├── 无人机根据收益-成本出价
    └── 分布式执行，无需中央协调
    """
    
    def __init__(self, n_drones, n_tasks):
        self.n_drones = n_drones
        self.n_tasks = n_tasks
        
        # 任务价格（初始为 0）
        self.prices = np.zeros(n_tasks)
        
        # 分配结果
        self.assignment = {}
    
    def compute_bid(self, drone_id, task_id, drone_position, task_position):
        """计算出价"""
        # 收益（任务价值）
        value = 100.0
        
        # 成本（距离）
        cost = np.linalg.norm(drone_position - task_position)
        
        # 出价 = 收益 - 成本 - 当前价格
        bid = value - cost - self.prices[task_id]
        
        return bid
    
    def auction_round(self, drone_positions, task_positions):
        """一轮拍卖"""
        bids = {}
        
        # 每个无人机对每个任务出价
        for i in range(self.n_drones):
            for j in range(self.n_tasks):
                bid = self.compute_bid(i, j, drone_positions[i], task_positions[j])
                if j not in bids or bid > bids[j][1]:
                    bids[j] = (i, bid)
        
        # 更新价格和分配
        for task_id, (drone_id, bid) in bids.items():
            self.assignment[task_id] = drone_id
            self.prices[task_id] += bid * 0.1  # 价格上升
        
        return self.assignment

2.2 一致性算法

class ConsensusAlgorithm:
    """
    分布式一致性算法
    
    应用：
    ├── 编队保持
    ├── 目标定位
    └── 决策融合
    """
    
    def __init__(self, n_agents, initial_values):
        self.n_agents = n_agents
        self.values = np.array(initial_values, dtype=float)
        
        # 邻接矩阵（通信图）
        self.adjacency = np.ones((n_agents, n_agents))
        np.fill_diagonal(self.adjacency, 0)
    
    def step(self, learning_rate=0.1):
        """
        一致性更新
        
        x_i(t+1) = x_i(t) + α * Σ (x_j(t) - x_i(t))
        """
        new_values = self.values.copy()
        
        for i in range(self.n_agents):
            # 获取邻居
            neighbors = np.where(self.adjacency[i] > 0)[0]
            
            # 计算差异
            diff = np.sum([self.values[j] - self.values[i] for j in neighbors])
            
            # 更新
            new_values[i] += learning_rate * diff
        
        self.values = new_values
        
        return self.values
    
    def run_until_convergence(self, threshold=1e-3, max_iter=1000):
        """运行直到收敛"""
        for _ in range(max_iter):
            old_values = self.values.copy()
            self.step()
            
            # 检查收敛
            if np.max(np.abs(self.values - old_values)) < threshold:
                break
        
        return self.values

第三章：多机协同感知

3.1 分布式 SLAM

class DistributedSLAM:
    """
    多无人机分布式 SLAM
    
    挑战：
    ├── 各无人机独立建图
    ├── 地图需要对齐和融合
    └── 通信带宽有限
    """
    
    def __init__(self, n_drones):
        self.n_drones = n_drones
        
        # 每架无人机的局部地图
        self.local_maps = [{} for _ in range(n_drones)]
        
        # 全局地图（融合后）
        self.global_map = {}
    
    def local_update(self, drone_id, observations):
        """
        局部更新
        
        参数：
            drone_id: 无人机 ID
            observations: 观测到的路标
        """
        for landmark in observations:
            self.local_maps[drone_id][landmark.id] = landmark.position
    
    def share_maps(self, communication_graph):
        """
        地图共享与融合
        
        参数：
            communication_graph: 通信拓扑
        """
        for i in range(self.n_drones):
            # 获取邻居
            neighbors = np.where(communication_graph[i] > 0)[0]
            
            for j in neighbors:
                # 融合地图
                self.local_maps[i] = self.fuse_maps(
                    self.local_maps[i], 
                    self.local_maps[j]
                )
    
    def fuse_maps(self, map1, map2):
        """融合两个地图"""
        fused = map1.copy()
        
        for landmark_id, position in map2.items():
            if landmark_id in fused:
                # 取平均
                fused[landmark_id] = (fused[landmark_id] + position) / 2
            else:
                fused[landmark_id] = position
        
        return fused

第四章：实战项目

项目 1：Boids 群体仿真

# 使用 Pygame 可视化
import pygame

def visualize_swarm(swarm):
    pygame.init()
    screen = pygame.display.set_mode((800, 600))
    
    running = True
    while running:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
        
        screen.fill((255, 255, 255))
        
        # 更新
        swarm.step()
        
        # 绘制
        for boid in swarm.boids:
            x, y = int(boid.position[0] * 8), int(boid.position[1] * 6)
            pygame.draw.circle(screen, (0, 0, 255), (x, y), 3)
        
        pygame.display.flip()
        pygame.time.delay(30)
    
    pygame.quit()

项目 2：分布式任务分配仿真

# 模拟 10 架无人机分配 20 个任务
n_drones = 10
n_tasks = 20

drone_positions = np.random.uniform(0, 100, (n_drones, 2))
task_positions = np.random.uniform(0, 100, (n_tasks, 2))

auction = TaskAuction(n_drones, n_tasks)

# 多轮拍卖
for round in range(10):
    assignment = auction.auction_round(drone_positions, task_positions)
    print(f"Round {round}: {len(assignment)} tasks assigned")

前沿论文推荐

• Dorigo et al. (2023) - "Swarm Robotics: A Review"
• Bullo et al. (2009) - "Distributed Consensus Algorithms"
• Chung et al. (2018) - "A Survey of Multi-Robot SLAM"

参考资源：
- PX4 多机仿真
- ROS 2 Multi-Robot
- Gazebo Swarm