智能体角色分工：Factorio Learning Environment矿工-运输工协作模型

在《异星工厂（Factorio）》这款游戏中，资源采集与运输的效率直接决定了工厂的扩张速度。传统单人模式下，玩家往往需要在矿工与运输工角色间频繁切换，导致操作繁琐且效率低下。Factorio Learning Environment（FLE）通过多智能体协作模型，实现了矿工与运输工的专业化分工，显著提升了资源流转效率。本文将详细解析这一协作模型的实现原理与应用方法。## 协作模型架构FLE...

gitblog_00071

469人浏览 · 2025-10-06 04:11:58

gitblog_00071 · 2025-10-06 04:11:58 发布

智能体角色分工：Factorio Learning Environment矿工-运输工协作模型

【免费下载链接】factorio-learning-environment A non-saturating, open-ended environment for evaluating LLMs in Factorio 项目地址: https://gitcode.com/GitHub_Trending/fa/factorio-learning-environment

协作模型架构

FLE的多智能体系统基于抽象基类AgentABC构建，该类定义了智能体的核心接口，包括初始化、决策和清理等方法。矿工与运输工智能体均继承自此类，并通过重写step方法实现专业化行为逻辑。

class AgentABC:
    @abstractmethod
    async def step(
        self,
        conversation: Conversation,
        response: Response,
        namespace: FactorioNamespace,
    ) -> Policy:
        """执行智能体的一个决策步骤"""
        pass

角色分工设计

角色	核心职责	关键API	决策依据
矿工	资源采集与初级加工	`mine()`, `place_entity()`	资源分布、矿脉剩余量
运输工	资源转运与库存管理	`move_to()`, `insert_item()`	库存水平、需求优先级

矿工智能体专注于资源开采，通过调用mine()方法实现自动化挖掘，并能根据资源分布动态调整开采策略。运输工则利用路径规划算法，通过move_to()方法实现高效物资转运，确保资源在生产线上的连续流动。

矿工智能体实现

矿工智能体基于BasicAgent类实现，专注于资源采集的自动化。其核心逻辑包括资源定位、开采决策和异常处理三个模块。

资源定位机制

通过namespace._render()方法生成地图视图，结合视觉识别算法实现矿脉自动定位：

render = namespace._render(
    position=player_pos,
    layers=Layer.ALL,  # 渲染所有图层以获取完整信息
)

视觉智能体VisualAgent会周期性渲染游戏地图，生成包含资源分布的可视化数据。矿工智能体利用这些信息，通过get_resource_patch()方法识别高价值矿脉，优先开采资源密度高的区域。

开采决策逻辑

矿工智能体的决策流程遵循"探测-评估-执行"三步模型：

资源探测：通过get_entities()获取当前视野内的资源实体
开采评估：分析矿脉大小、资源类型和可开采性
执行开采：调用mine()方法进行资源采集

关键实现代码位于examples/agents/basic_agent.py中，核心逻辑如下：

async def step(self, conversation, response, namespace):
    # 格式化对话历史获取上下文
    formatted_conversation = await self.formatter.format_conversation(conversation, namespace)
    
    # 资源探测与评估
    resources = namespace.get_entities(filter=lambda e: e.type == "resource")
    prioritized_resources = self._prioritize_resources(resources)
    
    # 生成开采策略
    return await self._generate_mining_policy(prioritized_resources)

运输工智能体实现

运输工智能体基于BacktrackingAgent类实现，专注于资源的高效转运。其核心特点是具备路径规划能力和异常处理机制，能够应对复杂环境下的物资运输挑战。

路径规划算法

运输工使用改进的A*算法进行路径规划，通过request_path()方法获取最优运输路线：

path = namespace.request_path(
    start_pos=current_position,
    end_pos=target_position,
    avoid_entities=["transport-belt", "assembling-machine"]
)

该算法会自动避开已有建筑和障碍物，确保运输路线的安全性和高效性。路径规划结果会通过walk.lua脚本转化为具体移动指令：

game.players[arg1].walking_state = {walking = arg3, direction = defines.direction.arg2}

库存管理策略

运输工智能体通过inspect_inventory()方法实时监控库存状态，并根据需求优先级动态调整运输策略。当检测到库存积压或短缺时，会自动触发紧急转运流程：

inventory = namespace.inspect_inventory(player_id)
if inventory.is_full():
    # 触发紧急卸货流程
    target_chest = self._find_nearest_chest(inventory.contents)
    return await self._generate_transport_policy(current_position, target_chest)

协作通信机制

矿工与运输工通过共享命名空间FactorioNamespace实现信息交互，确保协作的协调性和同步性。该命名空间作为信息枢纽，存储了游戏状态、实体位置和资源数量等关键数据。

信息共享接口

set_entity_recipe()：更新实体状态，通知运输工资源产出
get_inventory_status()：查询库存信息，协调资源转运
broadcast_event()：发送事件通知，如"矿脉枯竭"或"库存已满"

冲突解决策略

当多个智能体竞争同一资源或路径时，系统会基于优先级机制进行仲裁。矿工智能体在资源采集方面具有优先权，而运输工则在路径使用上拥有优先通行权。这种设计既保证了资源采集的连续性，又确保了物资运输的流畅性。

协作效率优化

为进一步提升协作效率，FLE引入了动态任务分配机制和性能监控系统。通过持续分析智能体的工作负载和效率指标，系统会自动调整任务分配，实现资源的最优配置。

动态负载均衡

基于AgentCapabilities类定义的能力矩阵，系统会根据智能体的实时表现调整任务分配：

capabilities=AgentCapabilities(
    pushNotifications=False, 
    stateTransitionHistory=True, 
    streaming=False
)

当检测到矿工负载过高时，系统会自动将部分矿脉分配给空闲矿工；同理，当运输线路拥堵时，会调度备用运输工分担压力。

性能监控指标

系统通过metrics.py模块持续监控关键性能指标，包括：

资源采集速率：单位时间内的矿石采集量
运输延迟：从资源产出到送达工厂的平均时间
智能体利用率：有效工作时间占比

这些指标通过可视化界面实时展示，帮助用户评估协作系统的整体性能。

应用案例

早期游戏阶段

在游戏初期，单个矿工-运输工组合即可满足基本资源需求。矿工专注于铁矿和铜矿的采集，运输工则将资源运往初始熔炉。此阶段的协作重点是建立稳定的资源供应链，为后续工厂扩张奠定基础。

大规模工厂阶段

随着工厂规模扩大，系统会自动扩展智能体团队。典型配置为：

3-5名矿工：负责不同区域的资源采集
2-3名运输工：专注于不同类型资源的转运
1名协调者：优化整体资源分配策略

这种配置可实现资源的并行采集与运输，显著提升工厂的扩张速度。

扩展与定制

FLE的模块化设计使得智能体行为的定制与扩展变得简单。用户可通过修改prompts/目录下的提示文件，调整智能体的决策逻辑。例如，修改prompts_for_rag/system_message_planning.md可调整任务规划策略，而steps_to_script/system_message.md则控制脚本生成逻辑。

自定义角色示例

通过继承AgentABC，用户可创建全新的智能体角色，如"建筑师"或"防御专家"。以下是一个简单的自定义智能体框架：

class ArchitectAgent(AgentABC):
    async def step(self, conversation, response, namespace):
        # 分析当前工厂布局
        # 生成优化方案
        # 执行建筑操作
        return self._generate_build_policy()