引言：AI的“最后一公里”困境

大模型已经能写出媲美人类的代码、生成逻辑严密的方案、回答五花八门的问题。但当你说“帮我把服务器上的日志查一下”或者“在本机跑一下这个脚本看看结果”时，再强大的AI也只能停留在“建议”层面——它无法真正触碰你的文件系统，无法执行一条命令，更无法修改一行代码。

这不是模型能力的问题，而是物理世界与数字智能之间的一道天堑。

iBizAIFactory 的 SkillRunner，正是为了填平这道天堑而生。

---

一、什么是SkillRunner？

SkillRunner 是 iBizAIFactory 架构中“云边协同”的执行端组件。在 AI Factory 的三层架构中——服务层（大脑）、能力层（工具箱）、基础层（物理触手）——SkillRunner 扮演的是最底层、最接近真实世界的角色。

云端 AI 负责“想”，SkillRunner 负责“做”。

它是一个部署在用户本地环境（企业内网、开发机、服务器）的轻量级客户端，通过 MQTT 协议接收云端下发的 ToolCall 指令，在本地真实执行文件读写、命令运行、补丁应用等操作，并将结果回传云端。

如果说 Hub-Agent 是“多专家调度官”、Sub-Agent 是“专项执行官”，那么 SkillRunner 就是这些“大脑”伸向物理世界的唯一一只手。

📊 整体架构流程图（云边协同）

下图清晰地展示了从“用户意图”到“物理执行”的全链路数据流向：

---

二、核心能力：它能做什么？

基于开源的 SkillRunner 代码实现，它具备以下原子化操作能力：

📂 文件操作三件套

命令	功能	特性
read_file	读取工作区或技能目录下的文件	支持文本编码自动识别
write_file	写入或覆盖文件	自动创建父目录
delete_file	删除指定文件	路径安全校验

🖥️ 系统执行

• execute_bash：在本地 Shell 中执行命令
  • 支持环境变量注入（env 参数）
  • 超时控制（默认 50 秒）
  • 工作目录切换（可指定 cwd）
  • 自动适配 Windows/Linux（python3 → python）

🔧 高级能力

命令	功能	技术亮点
apply_patch	基于 Unified Diff 精确修改文件	依赖 unidiff 库，上下文行精确匹配
download_file	分块接收并还原文件	Base64 编码，64KB/块，断点续传
upload_file	分块读取并回传文件	自动分块，索引标记
output_step	输出执行步骤信息	供机器人插件消费

🔌 插件扩展

通过 BOT_PLUGINS 环境变量动态加载外部机器人插件（如 QQ 机器人、企业微信机器人），将执行结果推送到 IM 等第三方渠道。

🤖 AI 修改代码的核心原理：apply_patch 执行流程

补丁应用是 AI 修改代码最优雅的方式——不需要重写整个文件，只传递变更的“差异”。

---

三、技术实现：安全、可靠、可扩展

⏱️ 核心机制时序图

详细拆解了 SkillRunner 启动、注册、执行任务的全生命周期：

1. 自动化注册与心跳保活

SkillRunner 启动时向 AIFactory 服务端注册，上报工作区路径、技能库路径、操作系统类型及本地技能清单。注册成功后，每 60 秒一次心跳持续上报状态，确保云端始终感知其在线情况与能力变化。

2. MQTT 消息驱动

SkillRunner 与云端之间通过 MQTT over WebSocket 通信。云端下发任务到动态分配的 listen_topic，SkillRunner 订阅并实时响应，执行结果通过 result_topic 回传。

📨 云端下发的 MQTT 消息格式（JSON）

{
  "tool_call_id": "call_abc123",
  "result_topic": "runner/response/abc123",
  "command": "execute_bash",
  "args": {
    "skill_id": "my_data_process",
    "from_template": false,
    "command": "python3 {baseDir}/main.py --input {SKILLS_WORKSPACE}/data.csv",
    "env": {
      "PYTHONPATH": "/usr/local/lib",
      "LOG_LEVEL": "DEBUG"
    }
  },
  "extended": {
    "main.py": "print('Hello from AI generated script!')"
  }
}

字段说明：

• extended：在执行前预置额外的辅助脚本或配置文件
• {baseDir}：自动替换为技能目录的绝对路径
• {SKILLS_WORKSPACE}：自动替换为工作区目录的绝对路径

3. 路径安全隔离

每个技能（Skill）拥有独立目录，SkillRunner 在读写文件时严格校验路径是否落在对应技能目录或工作区内，防止越权访问。核心校验逻辑：

def _validate_path_in_workspace(self, target_path, workspace_path):
    try:
        Path(target_path).resolve().relative_to(Path(workspace_path).resolve())
        return True
    except ValueError:
        return False

4. 技能热加载与清单同步

SkillRunner 扫描 AIFACTORY_SKILLS 目录下所有子文件夹，读取每个技能的 SKILL.md 中的 YAML 元数据，构建技能清单并随心跳上报。

📄 技能定义文件（SKILL.md）规范

开发者只需在技能根目录放置 SKILL.md，SkillRunner 便能自动识别并上报能力清单：

---
name: "数据分析助手"
description: "用于清洗和聚合 CSV 数据"
version: "1.0.0"
author: "iBizLab"
commands:
  - execute_bash
  - read_file
  - write_file
---

# 使用说明
该技能依赖 Python3 环境，执行前请确保 pandas 已安装。

当技能文件变更时，自动触发异步激活信号通知云端更新——能力即改即用，无需重启。

5. 机器人插件机制

通过动态导入机制加载外部插件，将执行步骤信息实时推送到 IM 等渠道。

🔌 插件开发示例

# 文件名: qq_plugin.py
class QqPlugin(BotPlugin):
    
    def __init__(self, config, command_callback):
        super().__init__(config, command_callback)
        self.client = None

    def start(self):
        def _run():
            try:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
                intents = Intents(
                    public_guild_messages=True,
                    direct_message=True,
                    public_messages=True
                )
                client = QQBotClient(
                    plugin=self,
                    loop=loop, 
                    intents=intents
                )
                self.client = client  # 保存引用
                client.run(appid=self.config['bot_id'], secret=self.config['secret'])
            finally:
                loop.close()
        
        self.thread = threading.Thread(target=_run, daemon=True, name=f"QQ-{self.name}")
        self.thread.start()
        print(f"[QQ插件] {self.name} 已启动")
        
    def stop(self):
        # botpy 没有直接停止的方法，守护线程会在主进程退出时自动结束
        try:
            self.thread.join()
            print(f"[QQ插件] {self.name} 停止")
        except Exception as e:
            print(f"[QQ插件] 停止 {self.name} 时发生错误: {e}")

    def output_step(self, step: dict):
        if self.client is None:
            return
        self.client.output_step(step)

配置环境变量启用：

BOT_PLUGINS='[{"type": "qq","name": "my_qq_bot","bot_id": "xx",    "secret": "xx",  "agent_tag":"demo_skill@DynamicAgent" }]'

---

四、应用场景：谁在用？怎么用？

🏢 场景一：企业内网自动化运维

企业往往不允许云端 AI 直接访问内网服务器。通过在内网部署 SkillRunner，云端 AI 可以下发运维指令（执行脚本、读取日志、修改配置），由 Runner 在本地完成操作——既保障内网安全，又实现 AI 驱动的自动化运维。

💻 场景二：本地代码仓库的 AI 辅助开发

开发者在本地运行 SkillRunner，配合云端 AI Factory 可实现：

• AI 自动读取项目代码并分析
• 根据 AI 生成的 Unified Diff 补丁自动修改代码（apply_patch）
• 执行构建、测试命令并回传结果

将 AI 编程助手的能力延伸到本地开发环境。

📝 补丁内容（Unified Diff）示例

当 AI 需要修改代码时，它会发送类似下面的文本。SkillRunner 的 apply_patch 方法会精确查找 print("old") 并将其替换。

--- a/script.py
+++ b/script.py
@@ -1,3 +1,3 @@
 def main():
-    print("old")
+    print("new")
     return True

☁️ 场景三：多云/混合云统一调度

在多个云环境或混合云中分别部署 SkillRunner，云端 Hub-Agent 根据任务需求将指令路由到对应 Runner——实现跨环境、跨地域的统一调度。

🔒 场景四：数据不出域的隐私合规

金融、医疗等行业对数据出境有严格限制。SkillRunner 的本地执行模式确保敏感数据始终停留在企业内网，云端只接收处理结果而非原始数据。

⏰ 场景五：无人值守定时/批量任务

Sub-Agent 支持独立模式（Standalone Mode），可由定时任务（Cron Job）或批量流水线直接驱动。配合 SkillRunner 的本地执行能力，实现夜间批量数据处理、定时报表生成、周期性系统巡检等无人值守场景。

---

五、为什么是 SkillRunner？——与同类工具的差异

对比维度	SkillRunner	OpenClaw 等独立智能体
定位	企业级AI平台的可插拔执行组件	独立的全能型AI管家
智能核心	无——完全被动执行指令	有——内置Agent自主规划
架构	云边协同，职责单一	单体式，集决策与执行于一体
交互方式	MQTT JSON 指令	自然语言直接交互
核心能力	原子化操作（文件/命令/补丁）	复杂工作流（浏览器/GUI/跨应用）
部署方式	后台守护进程	Gateway 进程
目标用户	企业IT团队	个人开发者、极客
可靠性	平台保障 + 心跳 + 自动重连	Gateway 单点问题

SkillRunner 不追求“全能”，而是追求 “极致专注” ——只做一件事：安全、可靠地执行云端下发的每一个原子操作。

---

六、快速上手

📦 1. 克隆仓库

git clone https://gitee.com/ibizlab/aifactory
cd aifactory

🔧 2. 安装依赖

pip install paho-mqtt python-dotenv unidiff requests

⚙️ 3. 配置环境变量

创建 .env 文件：

# 服务端地址（必填）
AIFACTORY_URL="https://your-company-aifactory.com/api"
AIFACTORY_TOKEN="sk-xxxxx"

# MQTT WebSocket 地址（必填）
AIFACTORY_MQTT="wss://mqtt.your-company.com:443/mqtt"

# 本地技能存放目录（必填）
AIFACTORY_SKILLS="/home/developer/workspace/skills"

# 运行时工作区（必填，用于存放临时数据或输出产物）
AIFACTORY_SKILL_WORKSPACE="/home/developer/workspace/runtime"

# 机器人插件配置（可选）
BOT_PLUGINS='[{"type":"dingtalk", "name":"报警机器人", "webhook_url":"https://..."}]'

🏃 4. 启动 SkillRunner

python skill_runner.py

✅ 5. 验证运行状态

启动成功后，你将看到类似以下日志：

已连接到MQTT WebSocket服务器
注册成功！获取监听主题: runner/abc123/commands
已订阅动态主题: runner/abc123/commands
开始监听MQTT消息...
成功发送激活信号 (Heartbeat)
已加载技能清单，共 3 个技能

---

结语

AI 的能力边界，从来不取决于模型的参数规模，而取决于它能触达多少真实世界。

SkillRunner 所做的，就是给 AI 一双“物理的手”——让大模型不再只是“纸上谈兵”的顾问，而是能真正动手改变世界的执行者。

在 iBizAIFactory 的体系中：

• Hub-Agent 负责思考
• Sub-Agent 负责规划
• Skill Bus 提供工具
• SkillRunner 负责把一切都落到实处

它是 AI 从“云端”走向“地面”的最后一步，也是最关键的一步。

---

SkillRunner 已在 Gitee 开源：https://gitee.com/ibizlab/aifactory