AI人工智能领域多智能体系统：实现智能工业的自动化

    class Agent:
    def __init__(self, id, environment):
        self.id = id
        self.environment = environment
        self.knowledge = {}
        self.neighbors = []
    
    def perceive(self):
        """感知环境状态"""
        return self.environment.get_state(self.id)
    
    def decide(self, perception):
        """基于感知做出决策"""
        # 简单的基于规则的决策
        if perception['task_available']:
            return 'accept_task'
        elif perception['resource_needed']:
            return 'request_resource'
        else:
            return 'idle'
    
    def act(self, decision):
        """执行决策"""
        if decision == 'accept_task':
            self.environment.assign_task(self.id)
        elif decision == 'request_resource':
            self.communicate({'type': 'resource_request', 'sender': self.id})
    
    def communicate(self, message):
        """与其他智能体通信"""
        for neighbor in self.neighbors:
            neighbor.receive(message)
    
    def receive(self, message):
        """处理接收到的消息"""
        if message['type'] == 'resource_request':
            if self.has_available_resource():
                self.communicate({'type': 'resource_response', 'sender': self.id, 'resource': True})
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/rn4Zp6EAyUzqg1LxfNVSdcCG3Mew.png)

    import numpy as np
    import random
    
    class RLAgent(Agent):
    def __init__(self, id, environment, learning_rate=0.1, discount_factor=0.9, exploration_rate=0.2):
        super().__init__(id, environment)
        self.q_table = {}
        self.learning_rate = learning_rate
        self.discount_factor = discount_factor
        self.exploration_rate = exploration_rate
    
    def get_state_key(self, state):
        """将状态转换为Q表的键"""
        return str(sorted(state.items()))
    
    def decide(self, perception):
        state_key = self.get_state_key(perception)
    
        # 初始化Q值
        if state_key not in self.q_table:
            self.q_table[state_key] = {'accept_task': 0, 'request_resource': 0, 'idle': 0}
    
        # ε-贪心策略
        if random.random() < self.exploration_rate:
            return random.choice(list(self.q_table[state_key].keys()))
        else:
            return max(self.q_table[state_key].items(), key=lambda x: x[1])[0]
    
    def learn(self, state, action, reward, next_state):
        state_key = self.get_state_key(state)
        next_state_key = self.get_state_key(next_state)
    
        # 初始化下一状态的Q值
        if next_state_key not in self.q_table:
            self.q_table[next_state_key] = {'accept_task': 0, 'request_resource': 0, 'idle': 0}
    
        # Q-learning更新规则
        current_q = self.q_table[state_key][action]
        max_next_q = max(self.q_table[next_state_key].values())
        new_q = current_q + self.learning_rate * (reward + self.discount_factor * max_next_q - current_q)
        self.q_table[state_key][action] = new_q
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/ea4dEzj2GAhsXHkWibpw65PBoVnU.png)

    # 创建Python虚拟环境
    python -m venv mas_env
    source mas_env/bin/activate  # Linux/Mac
    mas_env\Scripts\activate     # Windows
    
    # 安装必要库
    pip install numpy torch matplotlib pyyaml tensorboardx
    pip install git+https://github.com/oxwhirl/pymarl.git
    
    
    bash

    import numpy as np
    from collections import defaultdict
    
    class IndustrialEnvironment:
    def __init__(self, num_agents, num_tasks, num_resources):
        self.num_agents = num_agents
        self.agents = [Agent(i, self) for i in range(num_agents)]
        self.tasks = [{'id': i, 'complexity': np.random.randint(1,5)} for i in range(num_tasks)]
        self.resources = [{'id': i, 'type': np.random.choice(['A','B','C'])} for i in range(num_resources)]
        self.agent_positions = np.random.rand(num_agents, 2)  # 2D坐标
        self.task_allocations = defaultdict(list)
        self.resource_allocations = defaultdict(list)
        self.time = 0
    
    def get_state(self, agent_id):
        """返回指定智能体的局部状态"""
        agent_pos = self.agent_positions[agent_id]
        nearby_tasks = [t for t in self.tasks if
                       np.linalg.norm(agent_pos - np.random.rand(2)) < 0.3]  # 简化距离计算
        nearby_resources = [r for r in self.resources if
                           np.linalg.norm(agent_pos - np.random.rand(2)) < 0.3]
    
        return {
            'position': agent_pos,
            'nearby_tasks': nearby_tasks,
            'nearby_resources': nearby_resources,
            'current_task': next((t for t in self.task_allocations if agent_id in self.task_allocations[t]), None),
            'current_resources': self.resource_allocations.get(agent_id, []),
            'time': self.time
        }
    
    def assign_task(self, agent_id, task_id):
        """分配任务给智能体"""
        if task_id in self.task_allocations:
            self.task_allocations[task_id].append(agent_id)
        else:
            self.task_allocations[task_id] = [agent_id]
    
    def step(self):
        """环境步进"""
        self.time += 1
        # 更新任务状态
        for task_id, agents in list(self.task_allocations.items()):
            if len(agents) >= self.tasks[task_id]['complexity']:
                print(f"Task {task_id} completed by agents {agents}")
                del self.task_allocations[task_id]
                # 给参与任务的智能体奖励
                for agent in agents:
                    if isinstance(self.agents[agent], RLAgent):
                        self.agents[agent].learn(
                            self.get_state(agent),
                            'accept_task',
                            10,  # 奖励
                            self.get_state(agent)
                        )
    
        # 随机生成新任务
        if np.random.rand() < 0.1:
            new_task = {'id': len(self.tasks), 'complexity': np.random.randint(1,5)}
            self.tasks.append(new_task)
            print(f"New task generated: {new_task}")
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/SBxEfUgIkTCG8w4s0QntOR6iDoe7.png)

 * 模拟了任务、资源和智能体的空间分布
 * 每个智能体只能感知局部环境信息
 * 任务有复杂度属性，需要多个智能体协作完成

 * 基础智能体使用简单的基于规则的决策
 * RL智能体使用Q-learning进行决策学习
 * 智能体通过通信协调任务分配和资源共享

 * 任务需要足够数量的智能体才能完成
 * 智能体通过感知附近的任务和资源做出决策
 * 成功完成任务后，参与智能体会获得奖励