AIGC领域多智能体系统的群体智能特性

          +-------------------+
|共享环境|
|---|

          +-------------------+
                ^       ^       ^
|||

    +-----------+   |   +-----------+   +-----------+
|智能体A|---+---|智能体B|---+---|智能体C|
|---|---|---|---|---|

    +-----------+   |   +-----------+   +-----------+
|||

                +-------+-------+
               交互协议（消息传递）
    
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/tbyDosaS9g4dCPri8YZR2cJNBMTp.png)

    import numpy as np
    
    class AntColony:
    def __init__(self, n_ants, n_iterations, alpha=1, beta=1, rho=0.5, q=100):
        self.n_ants = n_ants          # 蚂蚁数量
        self.n_iterations = n_iterations  # 迭代次数
        self.alpha = alpha            # 信息素权重
        self.beta = beta              # 启发式权重
        self.rho = rho                # 信息素挥发率
        self.q = q                    # 信息素强度
        
    def _init_pheromone(self, distance_matrix):
        n_cities = distance_matrix.shape[0]
        return np.ones((n_cities, n_cities)) / (n_cities * n_cities)
    
    def _select_next_city(self, current_city, visited, pheromone, distance_matrix):
        n_cities = distance_matrix.shape[0]
        allowed_cities = [city for city in range(n_cities) if city not in visited]
        probabilities = np.zeros(n_cities)
        for city in allowed_cities:
            tau = pheromone[current_city][city]
            eta = 1.0 / distance_matrix[current_city][city]
            probabilities[city] = (tau ** self.alpha) * (eta ** self.beta)
        prob_sum = np.sum(probabilities)
        if prob_sum == 0:
            return np.random.choice(allowed_cities)
        probabilities /= prob_sum
        return np.random.choice(n_cities, p=probabilities)
    
    def solve(self, distance_matrix):
        n_cities = distance_matrix.shape[0]
        best_path = None
        best_distance = float('inf')
        pheromone = self._init_pheromone(distance_matrix)
        
        for _ in range(self.n_iterations):
            ant_paths = []
            for _ in range(self.n_ants):
                path = [np.random.randint(n_cities)]
                visited = set(path)
                while len(visited) < n_cities:
                    next_city = self._select_next_city(path[-1], visited, pheromone, distance_matrix)
                    path.append(next_city)
                    visited.add(next_city)
                ant_paths.append(path)
            
            new_pheromone = np.zeros_like(pheromone)
            for path in ant_paths:
                distance = sum(distance_matrix[path[i]][path[i+1]] for i in range(n_cities-1))
                for i in range(n_cities-1):
                    u = path[i]
                    v = path[i+1]
                    new_pheromone[u][v] += self.q / distance
                    new_pheromone[v][u] = new_pheromone[u][v]  # 无向图
            
            pheromone = (1 - self.rho) * pheromone + new_pheromone
            
            current_best_distance = min(sum(distance_matrix[path[i]][path[i+1]] for i in range(n_cities-1)) for path in ant_paths)
            if current_best_distance < best_distance:
                best_distance = current_best_distance
                best_path = [path for path in ant_paths if sum(distance_matrix[path[i]][path[i+1]] for i in range(n_cities-1)) == best_distance][0]
        
        return best_path, best_distance
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/GsSYJTc8lOXuN0jVBk6QLW2wexEz.png)

    import numpy as np
    
    class ParticleSwarmOptimizer:
    def __init__(self, n_particles, dimensions, max_iterations, w=0.8, c1=2, c2=2):
        self.n_particles = n_particles
        self.dimensions = dimensions
        self.max_iterations = max_iterations
        self.w = w          # 惯性权重
        self.c1 = c1        # 个体学习因子
        self.c2 = c2        # 全局学习因子
        
        # 初始化粒子位置和速度
        self.positions = np.random.uniform(-10, 10, (n_particles, dimensions))
        self.velocities = np.zeros_like(self.positions)
        
        # 个体最优和全局最优
        self.pbest_positions = np.copy(self.positions)
        self.pbest_values = np.inf * np.ones(n_particles)
        self.gbest_position = np.zeros(dimensions)
        self.gbest_value = np.inf
    
    def _evaluate(self, position):
        # 示例目标函数：Sphere函数
        return np.sum(position ** 2)
    
    def optimize(self):
        for _ in range(self.max_iterations):
            for i in range(self.n_particles):
                current_value = self._evaluate(self.positions[i])
                # 更新个体最优
                if current_value < self.pbest_values[i]:
                    self.pbest_values[i] = current_value
                    self.pbest_positions[i] = self.positions[i]
                # 更新全局最优
                if current_value < self.gbest_value:
                    self.gbest_value = current_value
                    self.gbest_position = self.positions[i]
            
            # 更新速度和位置
            r1 = np.random.rand(self.n_particles, self.dimensions)
            r2 = np.random.rand(self.n_particles, self.dimensions)
            self.velocities = self.w * self.velocities + \
                              self.c1 * r1 * (self.pbest_positions - self.positions) + \
                              self.c2 * r2 * (self.gbest_position - self.positions)
            self.positions += self.velocities
        
        return self.gbest_position, self.gbest_value
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/gX6HieVrOxJGnEpzUhPbaSlcByM8.png)

    {
    "sender": "agent_id_001",
    "receiver": "agent_group_content_generators",
    "message_type": "TASK_ASSIGNMENT",
    "content": {
        "task_id": "20231001_001",
        "task_description": "生成科幻小说开头段落",
        "constraints": {
            "word_count": 500-800,
            "genre": "硬科幻",
            "key_elements": ["量子计算机", "星际航行"]
        }
    },
    "timestamp": "2023-10-01T12:00:00Z"
    }
    
    
    json
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/PSRgqyViw1YBWKJuXt4O7AQbUG6l.png)

    # 安装依赖
    pip install mesa transformers flask redis torch
    
    # 下载预训练模型
    python -m transformers-cli download --model gpt2
    
    
    bash

    from mesa import Agent
    from transformers import pipeline
    
    class StoryGeneratorAgent(Agent):
    def __init__(self, unique_id, model_name, max_length=200):
        super().__init__(unique_id)
        self.model_name = model_name
        self.generator = pipeline("text-generation", model=model_name)
        self.max_length = max_length
        self.current_task = None
        self.collaborative_context = []  # 存储协作历史
    
    def receive_task(self, task):
        """接收任务：包含主题、风格、上下文等信息"""
        self.current_task = task
    
    def generate_content(self, prompt):
        """基于提示生成内容"""
        output = self.generator(
            prompt,
            max_length=self.max_length,
            num_return_sequences=1,
            temperature=0.8
        )
        return output[0]['generated_text']
    
    def collaborate(self, other_agent):
        """与其他智能体交换中间结果"""
        if self.current_task and other_agent.current_task and self.current_task == other_agent.current_task:
            # 交换最新生成的段落
            self.collaborative_context.append(other_agent.last_generated)
            other_agent.collaborative_context.append(self.last_generated)
    
    def step(self):
        """智能体执行步骤"""
        if self.current_task:
            # 构建生成提示（包含任务要求和协作历史）
            prompt = f"主题：{self.current_task['theme']}，风格：{self.current_task['style']}\n"
            prompt += "上下文：" + " ".join(self.collaborative_context) + "\n"
            prompt += "当前段落："
            
            # 生成内容
            generated_text = self.generate_content(prompt)
            self.last_generated = generated_text
            # 发布到共享黑板
            self.model.blackboard.append(generated_text)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/8OdWvarbcyiALeN5TfMPkhoEqpgw.png)

    from mesa import Model
    from mesa.time import RandomActivation
    
    class StoryWorld(Model):
    def __init__(self, n_agents, model_name):
        self.num_agents = n_agents
        self.schedule = RandomActivation(self)
        self.blackboard = []  # 共享协作空间
        
        # 创建智能体
        for i in range(n_agents):
            agent = StoryGeneratorAgent(i, model_name)
            self.schedule.add(agent)
            self.agents.append(agent)
    
    def assign_task(self, task):
        """向所有智能体分配任务"""
        for agent in self.schedule.agents:
            agent.receive_task(task)
    
    def step(self):
        """执行一轮协作"""
        self.schedule.step()
        # 随机选择一对智能体进行协作
        agents = list(self.schedule.agents)
        if len(agents) >= 2:
            a1, a2 = np.random.choice(agents, 2, replace=False)
            a1.collaborate(a2)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/jC4PGJprH0eWBDwYzViscZfk3KQl.png)

    class TaskScheduler:
    def __init__(self, story_world):
        self.story_world = story_world
    
    def split_task(self, full_task):
        """将完整故事生成任务拆分为章节任务"""
        chapters = full_task['chapters']
        sub_tasks = []
        for chapter_idx, chapter_desc in enumerate(chapters):
            sub_tasks.append({
                "theme": full_task['theme'],
                "style": full_task['style'],
                "chapter": chapter_idx,
                "description": chapter_desc
            })
        return sub_tasks
    
    def execute_task(self, full_task):
        """执行任务流程：拆分->分配->生成->整合"""
        sub_tasks = self.split_task(full_task)
        for sub_task in sub_tasks:
            self.story_world.assign_task(sub_task)
            for _ in range(10):  # 迭代协作次数
                self.story_world.step()
        
        # 整合黑板内容
        return "\n\n".join(self.story_world.blackboard)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/PcuWnxSAhk461fOmMw0EaV8tTI3U.png)

    需求解析层（智能体A） -> 创意生成层（智能体B1-Bn） -> 细节完善层（智能体C1-Cn）

    ↓                          ↑                             ↑
    ├────────────── 质量评估层（智能体D） ────────────────┤