一、引言

随着智能体（Agent）技术在企业场景中的落地，Agent Skill（智能体技能）作为Agent的核心功能模块，承担着“封装业务逻辑、对接外部服务、执行具体操作”的关键角色。本报告整合Agent Skill从定义、配置、匹配、调用，到沙盒基础设施支撑、Agent服务落地的全流程内容，详细阐述各环节的实现方案、核心逻辑及技术细节，为企业级Agent Skill的设计、开发、部署提供完整的参考依据。

本报告核心覆盖内容：Agent Skill核心概念与生命周期、SKILL.md配置规范、沙盒基础设施设计（bash支持、文件系统、内网连通、危险拦截）、Agent服务全模块实现、Skill与沙盒的调用链路、全流程保障机制，所有内容均基于企业内网落地场景设计，可直接复用。

二、Agent Skill 核心概念与生命周期

2.1 核心定义

Agent Skill是智能体的“功能单元”，每个Skill独立封装特定业务逻辑（如内网日志查询、第三方API调用、bash命令执行、文件处理等），拥有唯一标识（skill_id）、输入输出规则、执行逻辑及运行依赖，可按需配置、加载、调用，实现Agent功能的模块化扩展与灵活复用。

核心特点：

• 独立性：每个Skill独立封装逻辑，不依赖其他Skill，可单独加载、更新、销毁；

• 可配置性：Skill的所有规则（匹配描述、执行逻辑、依赖配置）均通过SKILL.md文件定义，与代码解耦；

• 安全性：所有Skill的执行均在隔离沙盒中进行，避免影响宿主机及内网核心资产；

• 可扩展性：新增Skill无需修改Agent核心代码，仅需更新SKILL.md配置。

2.2 完整生命周期

Agent Skill的生命周期从“创建”到“销毁”，贯穿配置、加载、匹配、调用、执行、回收全环节，与Agent服务、沙盒环境深度联动，具体流程如下：

生命周期核心状态说明：

• 就绪（ready）：Skill加载完成，已注册到Agent技能池，可被匹配和调用；

• 运行中（running）：Skill已被调用，正在沙盒内执行逻辑（如bash命令、内网调用）；

• 错误（error）：Skill执行过程中出现异常（如沙盒调用失败、第三方服务错误），需人工排查后重置为就绪状态；

• 销毁（destroyed）：Skill从Agent技能池中移除，沙盒实例回收，释放资源。

三、SKILL.md 配置规范（核心基础）

SKILL.md是Agent Skill的“配置中心”，所有Skill的匹配规则、执行逻辑、沙盒配置、内网访问规则均集中定义，实现“配置与代码解耦”。其核心作用是：为Agent提供Skill匹配依据、为沙盒提供执行规则、为大模型提供结果加工标准。

3.1 配置结构规范

每个Skill需包含“匹配描述、沙盒执行配置、第三方服务配置（可选）、大模型执行规则”四大核心模块，标准化结构如下（可直接复用）：

# SKILL.md - 企业级Agent Skill配置库
## 技能N：[技能名称]
- skill_id: [唯一标识，如skill_log_query_001]
- 描述（匹配用）：[简洁说明技能用途，用于大模型匹配，如“查询公司内网服务器的nginx日志（近1小时）”]
- 沙盒执行配置：
  1. 执行类型：[bash/script/http，如bash]
  2. 执行命令/脚本：[具体执行内容，支持参数占位符，如"grep 'error' /var/log/nginx/access.log"]
  3. 参数映射：
     - {{占位符}} → 技能参数：[用户输入的参数，如server_ip]
     - {{占位符}} → 固定值：[固定参数，如/tmp/output.log]
  4. 内网访问白名单：[JSON数组，如["172.16.0.0/16"]]
  5. 危险操作豁免：[JSON数组，无豁免则填[]，如["mkdir", "rm -rf /tmp/*"]]
  6. 超时时间：[秒数，如30s]
- 第三方服务配置（可选）：
  1. 服务类型：[REST API/SDK等]
  2. API地址：[第三方服务地址]
  3. 请求方式：[GET/POST]
  4. 请求参数映射：[API参数与Skill参数的对应关系]
  5. 响应解析规则：[提取第三方返回的关键数据]
- 大模型执行规则（结果加工）：
  1. 输入：[用户问题、沙盒/第三方返回结果]
  2. 处理规则：[如何将原始结果转换为自然语言]
  3. 异常处理：[执行失败/结果为空时的兜底话术]

3.2 配置示例（可直接落地）

# SKILL.md - 带沙盒执行配置的技能库
## 技能1：内网日志查询
- skill_id: skill_log_query_001
- 描述（匹配用）：查询公司内网服务器的nginx日志（近1小时）
- 沙盒执行配置：
  1. 执行类型：bash
  2. 执行命令：ssh -i /home/agent/.ssh/key.pem root@{{server_ip}} "grep 'error' /var/log/nginx/access.log | grep -E '$(date -d -1hour +%Y-%m-%d\ %H):'"
  3. 参数映射：
     - {{server_ip}} → 技能参数：server_ip（用户输入的服务器IP）
  4. 内网访问白名单：["172.16.0.0/16"]
  5. 危险操作豁免：[]
  6. 超时时间：30s
- 第三方服务配置（可选）：无
- 大模型执行规则（结果加工）：
  1. 输入：用户问题、沙盒返回的日志内容；
  2. 处理规则：提取日志中的错误时间、请求URL、错误码，用自然语言汇总；
  3. 异常处理：若日志为空，返回“近1小时该服务器nginx无error日志”。

## 技能2：内网API数据导出
- skill_id: skill_api_export_001
- 描述（匹配用）：导出内网用户中心API的指定用户数据
- 沙盒执行配置：
  1. 执行类型：script
  2. 执行脚本：/scripts/export_user_data.sh
  3. 参数映射：
     - {{user_id}} → 技能参数：user_id（用户输入的用户ID）
     - {{output_path}} → 固定值：/tmp/user_data_{{user_id}}.csv
  4. 内网访问白名单：["172.16.10.5:8080"]（用户中心API地址）
  5. 危险操作豁免：["mkdir", "rm -rf /tmp/*"]
  6. 超时时间：60s
- 第三方服务配置（可选）：无
- 大模型执行规则（结果加工）：
  1. 输入：用户问题、沙盒返回的导出结果（文件路径/执行状态）；
  2. 处理规则：说明“用户ID {{user_id}} 数据已导出至沙盒路径 {{output_path}}，共XX条记录”；
  3. 异常处理：若导出失败，返回具体错误原因。

3.3 配置解析逻辑

Agent服务通过配置管理模块，定期加载并解析SKILL.md，将配置信息拆分为三个核心字典，用于后续匹配、调用环节：

• skill_descriptions：key=skill_id，value=匹配用描述（用于大模型匹配skill_id）；

• skill_sandbox_configs：key=skill_id，value=沙盒执行配置（用于封装沙盒请求）；

• skill_exec_rules：key=skill_id，value=大模型执行规则（用于加工最终结果）。

解析过程中会校验必填字段（skill_id、执行类型、命令/脚本），避免配置错误导致Skill无法加载。

四、Agent Skill 运行基础设施——沙盒环境

沙盒环境是Agent Skill的“安全执行容器”，所有需要执行bash命令、读写文件、访问内网服务的Skill，均在沙盒内隔离运行，核心满足四大需求：支持bash命令、完整文件系统、内网连通、危险操作拦截，确保Skill执行的安全性、隔离性、可控性。

4.1 沙盒整体架构

沙盒环境基于轻量级容器（Docker/containerd）或微虚拟机（Firecracker MicroVM）构建，与Agent服务通过gRPC协议通信，整体架构如下：

4.2 四大核心需求实现方案

4.2.1 支持bash相关命令

• 技术选型：Docker容器（轻量级、易部署）或Firecracker MicroVM（高安全隔离）；

• 核心能力：内置标准Linux bash环境，支持单行命令、脚本执行、管道、重定向、循环等完整bash语法；

• 执行机制：每一次Skill执行对应一个临时沙盒实例，执行完成后自动销毁或放回实例池复用，避免资源占用。

4.2.2 完整的文件系统

采用OverlayFS分层文件系统，实现“隔离性+可读写”需求，分为三层：

• 底层镜像层（只读）：包含基础Linux系统、bash、curl、python等常用工具，所有沙盒实例共享该层，节省资源；

• 上层可写层（临时）：每个沙盒实例独立拥有，用于存储执行过程中产生的文件（如日志、导出数据），沙盒销毁后可写层自动丢弃；

• 挂载层（可选）：有限制地挂载内网共享目录或Skill专用工作目录，支持文件的临时持久化。

支持全套文件操作命令（ls、cd、mkdir、rm、cat、echo等），仅允许操作沙盒内部路径，不影响宿主机及其他沙盒。

4.2.3 连通公司内网

采用“网络命名空间+白名单”机制，实现受控内网访问：

• 独立网络命名空间：每个沙盒拥有独立的网络栈，与宿主机、其他沙盒网络隔离；

• 内网白名单配置：通过网络策略，仅放通指定内网IP段、端口、域名（如内网API地址、数据库代理）；

• 禁止访问：禁止访问宿主机网络、互联网（除非专门配置）、内网核心资产（如数据库主节点）；

• 支持能力：可通过curl、wget访问内网API，调用内网RPC/HTTP服务，访问内网Git、对象存储等。

4.2.4 危险操作拦截

采用四层拦截体系，全面拦截危险操作，保障沙盒及内网安全：

1. 系统调用级拦截（Seccomp/AppArmor/SELinux）：禁止mount、fork炸弹、kill -9关键进程、内核相关操作等；

2. bash命令黑名单：直接拦截rm -rf /、sudo、su、nc、nmap、tcpdump、挖矿程序、反弹shell等危险命令；

3. 文件权限拦截：沙盒默认以非root用户运行，仅允许写/tmp、/workdir等指定路径，禁止写/etc、/bin、/usr等系统路径；

4. 行为审计+实时阻断：全量上报沙盒执行的所有命令，检测到异常行为（如端口扫描、暴力破解）时，立即kill沙盒实例并返回错误提示，同时同步至公司安全平台。

4.3 沙盒与Agent通信协议

Agent与沙盒通过gRPC协议通信（高性能、适合内网、支持流式传输），定义标准化的请求/响应格式，确保通信稳定、可扩展。

Protobuf协议定义（sandbox_agent.proto）

syntax = "proto3";

package sandbox;

// 沙盒执行类型
enum ExecuteType {
  BASH = 0;        // 单行bash命令
  SCRIPT = 1;      // 脚本执行
  HTTP = 2;        // 内网HTTP请求
}

// 沙盒请求体
message SandboxRequest {
  string skill_id = 1;                // 匹配到的skill_id
  ExecuteType execute_type = 2;       // 执行类型
  string command = 3;                 // bash命令/脚本路径
  map<string, string> params = 4;     // 执行参数（如server_ip、user_id）
  repeated string inner_net_whitelist = 5; // 内网访问白名单
  repeated string danger_exemption = 6;    // 危险操作豁免
  int32 timeout = 7;                  // 超时时间（秒）
}

// 沙盒响应体
message SandboxResponse {
  bool success = 1;                   // 执行是否成功
  string stdout = 2;                  // 执行输出（stdout）
  string stderr = 3;                  // 错误输出（stderr）
  string error_msg = 4;               // 错误信息（如危险操作拦截、超时）
  map<string, string> result_data = 5;// 结构化结果（如文件路径、API返回）
  int32 execute_time = 6;             // 执行耗时（毫秒）
}

// 沙盒服务接口
service SandboxService {
  // 执行技能请求
  rpc ExecuteSkill(SandboxRequest) returns (SandboxResponse);
  // 创建沙盒实例（池化复用）
  rpc CreateSandboxInstance(Empty) returns (SandboxInstance);
  // 销毁沙盒实例
  rpc DestroySandboxInstance(SandboxInstance) returns (Empty);
}

// 空消息
message Empty {}

// 沙盒实例信息
message SandboxInstance {
  string instance_id = 1;             // 沙盒实例ID
  string ip = 2;                      // 沙盒IP
  int32 port = 3;                     // 沙盒端口
  bool is_idle = 4;                   // 是否空闲
}

五、Agent 服务完整实现（核心调度）

Agent服务是Agent Skill全流程的“调度中心”，负责SKILL.md加载解析、skill_id匹配、沙盒请求封装、沙盒调用、大模型交互、结果返回等全环节，采用“模块化设计”，实现高内聚、低耦合，便于维护和扩展。

5.1 整体架构设计

Agent服务拆分为6个核心模块，各模块职责单一、可独立扩展，整体架构如下：

5.2 各模块完整实现（Python代码，可直接复用）

5.2.1 配置管理模块（config_manager.py）

负责加载、解析、缓存SKILL.md，支持热更新，避免重复读取文件，为其他模块提供配置数据。

import re
import time
import json
from typing import Dict, List, Tuple

class SkillConfigManager:
    def __init__(self, skill_md_path: str, cache_ttl: int = 300):
        self.skill_md_path = skill_md_path
        self.cache_ttl = cache_ttl  # 缓存有效期（5分钟）
        self.last_load_time = 0
        self.skill_descriptions: Dict[str, str] = {}  # skill_id → 匹配描述
        self.skill_sandbox_configs: Dict[str, dict] = {}  # skill_id → 沙盒执行配置
        self.skill_exec_rules: Dict[str, str] = {}  # skill_id → 大模型执行规则
        self._load_skill_md()  # 初始化加载

    def _load_skill_md(self):
        """加载并解析SKILL.md，更新缓存"""
        try:
            with open(self.skill_md_path, 'r', encoding='utf-8') as f:
                content = f.read()
            
            # 匹配每个技能块
            skill_blocks = re.findall(r'## 技能\d+：.*?\n(.*?)(?=## 技能|$)', content, re.DOTALL)
            
            # 清空旧配置
            self.skill_descriptions.clear()
            self.skill_sandbox_configs.clear()
            self.skill_exec_rules.clear()

            for block in skill_blocks:
                # 提取skill_id
                skill_id_match = re.search(r'skill_id:\s*(\w+)', block)
                if not skill_id_match:
                    continue
                skill_id = skill_id_match.group(1)

                # 提取匹配描述
                desc_match = re.search(r'描述（匹配用）：\s*(.*?)\n', block)
                if desc_match:
                    self.skill_descriptions[skill_id] = desc_match.group(1)

                # 提取沙盒执行配置
                sandbox_match = re.search(r'沙盒执行配置：\s*(.*?)(?=第三方服务配置|大模型执行规则)', block, re.DOTALL)
                if sandbox_match:
                    sandbox_content = sandbox_match.group(1).strip()
                    sandbox_config = self._parse_sandbox_config(sandbox_content)
                    self.skill_sandbox_configs[skill_id] = sandbox_config

                # 提取大模型执行规则
                exec_rule_match = re.search(r'大模型执行规则（结果加工）：\s*(.*?)$', block, re.DOTALL)
                if exec_rule_match:
                    self.skill_exec_rules[skill_id] = exec_rule_match.group(1).strip()

            self.last_load_time = time.time()
            print(f"SKILL.md加载成功，共加载{len(self.skill_descriptions)}个技能")
        except Exception as e:
            raise Exception(f"加载SKILL.md失败：{str(e)}")

    def _parse_sandbox_config(self, content: str) -> dict:
        """解析沙盒执行配置为字典"""
        config = {}
        lines = [line.strip() for line in content.split('\n') if line.strip()]
        for line in lines:
            if line.startswith('1. 执行类型：'):
                config['execute_type'] = line.replace('1. 执行类型：', '')
            elif line.startswith('2. 执行命令：'):
                config['command'] = line.replace('2. 执行命令：', '')
            elif line.startswith('2. 执行脚本：'):
                config['command'] = line.replace('2. 执行脚本：', '')
            elif line.startswith('3. 参数映射：'):
                param_lines = re.findall(r'- \{\{(.*?)\}\} → 技能参数：(.*?)(?:\（|$)', line)
                config['param_map'] = {k: v for k, v in param_lines}
                # 提取固定值参数
                fixed_param_lines = re.findall(r'- \{\{(.*?)\}\} → 固定值：(.*?)$', line)
                config['fixed_param_map'] = {k: v for k, v in fixed_param_lines}
            elif line.startswith('4. 内网访问白名单：'):
                config['inner_net_whitelist'] = json.loads(line.replace('4. 内网访问白名单：', ''))
            elif line.startswith('5. 危险操作豁免：'):
                config['danger_exemption'] = json.loads(line.replace('5. 危险操作豁免：', ''))
            elif line.startswith('6. 超时时间：'):
                config['timeout'] = int(line.replace('6. 超时时间：', '').replace('s', ''))
        return config

    def get_skill_descriptions(self) -> Dict[str, str]:
        """获取技能描述（用于匹配），自动刷新缓存"""
        if time.time() - self.last_load_time > self.cache_ttl:
            self._load_skill_md()
        return self.skill_descriptions

    def get_sandbox_config(self, skill_id: str) -> dict:
        """获取指定skill_id的沙盒配置"""
        if time.time() - self.last_load_time > self.cache_ttl:
            self._load_skill_md()
        return self.skill_sandbox_configs.get(skill_id, {})

    def get_exec_rule(self, skill_id: str) -> str:
        """获取指定skill_id的大模型执行规则"""
        if time.time() - self.last_load_time > self.cache_ttl:
            self._load_skill_md()
        return self.skill_exec_rules.get(skill_id, "")

    def is_skill_valid(self, skill_id: str) -> bool:
        """校验skill_id是否有效"""
        return skill_id in self.skill_descriptions

5.2.2 技能匹配模块（skill_matcher.py）

调用大模型，根据用户提问和SKILL.md中的技能描述，匹配最合适的skill_id，确保匹配精准，处理匹配失败场景。

from openai import OpenAI
from typing import Dict, Optional

class SkillMatcher:
    def __init__(self, openai_api_key: str, openai_base_url: str, model: str = "gpt-3.5-turbo"):
        self.client = OpenAI(api_key=openai_api_key, base_url=openai_base_url)
        self.model = model

    def match_skill_id(self, user_query: str, skill_descriptions: Dict[str, str]) -> Optional[str]:
        """
        调用大模型匹配skill_id
        :param user_query: 用户提问
        :param skill_descriptions: 技能描述字典（skill_id → 描述）
        :return: 匹配的skill_id，无匹配返回None
        """
        # 构建匹配Prompt（核心：明确指令，只返回skill_id）
        skill_list = "\n".join([f"- {sid}: {desc}" for sid, desc in skill_descriptions.items()])
        system_prompt = f"""
        你的唯一任务是根据用户问题，匹配最合适的技能ID（skill_id）。
        可用技能列表：
        {skill_list}
        
        严格遵守以下规则：
        1. 仅返回匹配的skill_id（如skill_log_query_001），不要返回任何多余内容；
        2. 若没有任何匹配的技能，返回"no_match"；
        3. 必须从可用技能列表中选择，禁止编造skill_id。
        """

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_query}
                ],
                temperature=0,  # 0温度保证匹配结果稳定
                timeout=10
            )
            matched_skill_id = response.choices[0].message.content.strip()
            # 校验匹配结果
            if matched_skill_id == "no_match":
                return None
            if matched_skill_id in skill_descriptions:
                return matched_skill_id
            else:
                # 模型返回无效skill_id，返回None
                return None
        except Exception as e:
            raise Exception(f"匹配skill_id失败：{str(e)}")

5.2.3 技能请求封装模块（request_packer.py）

将skill配置和用户参数，封装为沙盒能识别的gRPC请求格式，同时做参数校验，拦截注入风险。

import re
from typing import Dict, Optional
from sandbox_agent_pb2 import SandboxRequest, ExecuteType

class SkillRequestPacker:
    def __init__(self, config_manager):
        self.config_manager = config_manager

    def _extract_user_params(self, user_query: str, skill_id: str) -> Dict[str, str]:
        """
        从用户提问中提取技能参数（进阶版可调用大模型提取，此处简化为正则）
        """
        params = {}
        sandbox_config = self.config_manager.get_sandbox_config(skill_id)
        param_map = sandbox_config.get('param_map', {})
        
        # 提取服务器IP（示例）
        if 'server_ip' in param_map.values():
            ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', user_query)
            if ip_match:
                params['server_ip'] = ip_match.group(1)
        # 提取用户ID（示例）
        if 'user_id' in param_map.values():
            user_id_match = re.search(r'用户ID[:：]?(\w+)', user_query)
            if user_id_match:
                params['user_id'] = user_id_match.group(1)
        return params

    def pack_sandbox_request(self, skill_id: str, user_query: str) -> Optional[SandboxRequest]:
        """
        封装沙盒请求
        :param skill_id: 匹配到的skill_id
        :param user_query: 用户提问
        :return: SandboxRequest对象，参数缺失返回None
        """
        # 1. 获取沙盒配置
        sandbox_config = self.config_manager.get_sandbox_config(skill_id)
        if not sandbox_config:
            raise Exception(f"skill_id={skill_id} 无沙盒执行配置")
        
        # 2. 提取用户参数
        user_params = self._extract_user_params(user_query, skill_id)
        
        # 3. 校验必填参数
        param_map = sandbox_config.get('param_map', {})
        required_params = list(param_map.values())
        missing_params = [p for p in required_params if p not in user_params]
        if missing_params:
            raise Exception(f"缺少必填参数：{','.join(missing_params)}")
        
        # 4. 替换命令中的参数占位符
        command = sandbox_config['command']
        # 替换用户参数
        for placeholder, param_name in param_map.items():
            command = command.replace(f"{{{{{placeholder}}}}}", user_params[param_name])
        # 替换固定值参数
        fixed_param_map = sandbox_config.get('fixed_param_map', {})
        for placeholder, fixed_value in fixed_param_map.items():
            command = command.replace(f"{{{{{placeholder}}}}}", fixed_value)
        
        # 5. 转换执行类型
        execute_type_str = sandbox_config['execute_type'].upper()
        execute_type = ExecuteType.BASH if execute_type_str == "BASH" else ExecuteType.SCRIPT

        # 6. 封装SandboxRequest
        sandbox_request = SandboxRequest(
            skill_id=skill_id,
            execute_type=execute_type,
            command=command,
            params=user_params,
            inner_net_whitelist=sandbox_config.get('inner_net_whitelist', []),
            danger_exemption=sandbox_config.get('danger_exemption', []),
            timeout=sandbox_config.get('timeout', 30)
        )
        return sandbox_request

5.2.4 沙盒通信模块（sandbox_communicator.py）

实现Agent与沙盒的gRPC通信，发送封装好的Skill请求，接收沙盒执行结果，处理通信异常。

import grpc
from typing import Optional
from sandbox_agent_pb2 import SandboxRequest, SandboxResponse
from sandbox_agent_pb2_grpc import SandboxServiceStub

class SandboxCommunicator:
    def __init__(self, sandbox_server_addr: str):
        self.sandbox_server_addr = sandbox_server_addr
        self.channel = grpc.insecure_channel(sandbox_server_addr)
        self.stub = SandboxServiceStub(self.channel)

    def execute_skill_in_sandbox(self, sandbox_request: SandboxRequest) -> SandboxResponse:
        """
        发送请求到沙盒执行技能
        :param sandbox_request: 封装好的沙盒请求
        :return: 沙盒响应
        """
        try:
            # 调用沙盒的ExecuteSkill接口
            response = self.stub.ExecuteSkill(sandbox_request, timeout=sandbox_request.timeout + 5)
            return response
        except grpc.RpcError as e:
            raise Exception(f"沙盒通信失败：{e.code()} - {e.details()}")
        finally:
            # 可选：关闭通道（池化场景可复用通道）
            self.channel.close()

5.2.5 大模型交互模块（llm_interactor.py）

拼装用户问题、沙盒执行结果、Skill执行规则，调用大模型生成自然语言最终结果，满足用户易懂的需求。

from openai import OpenAI
from typing import Dict

class LLMInteractor:
    def __init__(self, openai_api_key: str, openai_base_url: str, model: str = "gpt-3.5-turbo"):
        self.client = OpenAI(api_key=openai_api_key, base_url=openai_base_url)
        self.model = model

    def generate_final_result(self, user_query: str, skill_id: str, sandbox_response: dict, exec_rule: str) -> str:
        """
        生成最终结果
        :param user_query: 用户提问
        :param skill_id: 匹配的skill_id
        :param sandbox_response: 沙盒返回结果（字典格式）
        :param exec_rule: 大模型执行规则
        :return: 自然语言最终结果
        """
        # 构建Prompt（核心：用户问题+沙盒结果+执行规则）
        system_prompt = f"""
        你需要严格按照以下规则，将沙盒执行结果转换为用户易懂的自然语言：
        1. 用户原始问题：{user_query}
        2. 沙盒执行结果：{sandbox_response}
        3. 执行规则：
        {exec_rule}
        
        要求：
        - 结果简洁、口语化，无技术术语；
        - 若沙盒执行失败，直接返回错误提示，无需额外解释；
        - 若结果为空，按执行规则返回兜底话术。
        """

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": "请按规则生成最终回答"}
                ],
                temperature=0.1,  # 低温度保证结果稳定
                timeout=10
            )
            return response.choices[0].message.content.strip()
        except Exception as e:
            raise Exception(f"大模型生成结果失败：{str(e)}")

5.2.6 核心调度模块（agent_core.py）

串联所有模块，处理全流程异常，是Agent服务的“大脑”，负责用户请求的端到端处理。

import traceback
from config_manager import SkillConfigManager
from skill_matcher import SkillMatcher
from request_packer import SkillRequestPacker
from sandbox_communicator import SandboxCommunicator
from llm_interactor import LLMInteractor

class AgentCore:
    def __init__(self, config: dict):
        # 初始化配置
        self.skill_md_path = config['skill_md_path']
        self.openai_api_key = config['openai_api_key']
        self.openai_base_url = config['openai_base_url']
        self.sandbox_server_addr = config['sandbox_server_addr']
        
        # 初始化各模块
        self.config_manager = SkillConfigManager(self.skill_md_path)
        self.skill_matcher = SkillMatcher(self.openai_api_key, self.openai_base_url)
        self.request_packer = SkillRequestPacker(self.config_manager)
        self.sandbox_communicator = SandboxCommunicator(self.sandbox_server_addr)
        self.llm_interactor = LLMInteractor(self.openai_api_key, self.openai_base_url)

    def handle_user_query(self, user_query: str) -> str:
        """
        处理用户提问的主流程
        :param user_query: 用户输入
        :return: 最终自然语言结果
        """
        try:
            # 步骤1：加载技能描述（自动刷新缓存）
            skill_descriptions = self.config_manager.get_skill_descriptions()
            if not skill_descriptions:
                return "暂无可用技能，请联系管理员配置SKILL.md"

            # 步骤2：匹配skill_id
            skill_id = self.skill_matcher.match_skill_id(user_query, skill_descriptions)
            if not skill_id:
                return "未匹配到可用技能，请更换问题重试（支持：内网日志查询、内网API数据导出）"

            # 步骤3：封装沙盒请求
            sandbox_request = self.request_packer.pack_sandbox_request(skill_id, user_query)
            if not sandbox_request:
                return "参数解析失败，请补充关键信息（如服务器IP、用户ID）"

            # 步骤4：调用沙盒执行技能
            sandbox_response = self.sandbox_communicator.execute_skill_in_sandbox(sandbox_request)
            # 转换沙盒响应为字典（便于大模型处理）
            sandbox_response_dict = {
                "success": sandbox_response.success,
                "stdout": sandbox_response.stdout,
                "stderr": sandbox_response.stderr,
                "error_msg": sandbox_response.error_msg,
                "result_data": dict(sandbox_response.result_data),
                "execute_time": sandbox_response.execute_time
            }
            if not sandbox_response.success:
                return f"技能执行失败：{sandbox_response.error_msg}"

            # 步骤5：调用大模型生成最终结果
            exec_rule = self.config_manager.get_exec_rule(skill_id)
            final_result = self.llm_interactor.generate_final_result(
                user_query, skill_id, sandbox_response_dict, exec_rule
            )
            return final_result

        except Exception as e:
            # 全流程异常兜底
            error_detail = traceback.format_exc()
            print(f"Agent处理请求失败：{error_detail}")  # 日志记录
            return f"服务暂不可用，请稍后重试（错误：{str(e)}）"

# 启动Agent服务示例
if __name__ == "__main__":
    # 配置项（可放在配置文件中）
    agent_config = {
        "skill_md_path": "./SKILL.md",
        "openai_api_key": "你的OpenAI API密钥",
        "openai_base_url": "https://api.openai.com/v1",
        "sandbox_server_addr": "172.16.0.100:50051"  # 沙盒服务器地址
    }

    # 初始化Agent核心
    agent_core = AgentCore(agent_config)

    # 测试用户请求
    user_query = "查询172.16.0.20服务器近1小时的nginx error日志"
    final_result = agent_core.handle_user_query(user_query)
    print("用户问题：", user_query)
    print("最终结果：", final_result)

六、Agent Skill 全流程调用解析

结合上述模块和基础设施，Agent Skill从用户提问到最终结果返回的完整调用流程，共7个核心步骤，流程清晰、可追溯，具体如下：

6.1 各步骤详细说明

1. 用户发起请求：用户输入自然语言提问（如“查询172.16.0.20服务器近1小时的nginx error日志”），发送至Agent服务；

2. SKILL.md加载：配置管理模块加载并解析SKILL.md，缓存技能描述、沙盒配置、执行规则，确保数据最新；

3. skill_id匹配：技能匹配模块将用户提问和技能描述传给大模型，大模型返回唯一匹配的skill_id（如skill_log_query_001），并校验有效性；

4. 沙盒请求封装：技能请求封装模块提取用户提问中的参数（如服务器IP：172.16.0.20），替换Skill配置中的占位符，封装为标准化的SandboxRequest对象，做参数校验；

5. 沙盒调用：沙盒通信模块通过gRPC协议，将SandboxRequest发送至沙盒服务器，沙盒实例接收请求后，按配置执行逻辑（如执行bash命令）；

6. 沙盒执行与结果回传：沙盒执行过程中，危险操作拦截引擎全程监控，执行完成后，将stdout、stderr、结构化结果通过gRPC回传给Agent服务；

7. 最终结果生成与返回：大模型交互模块拼装用户提问、沙盒结果、执行规则，调用大模型生成自然语言最终结果，返回给用户。

七、关键保障机制（企业级落地必备）

为确保Agent Skill服务稳定、安全、高效运行，建立四大保障机制，覆盖异常处理、性能优化、安全防护、可扩展性。

7.1 异常全链路兜底

每个模块均设置异常处理逻辑，任何环节失败均有兜底方案，避免服务崩溃，提升用户体验：

执行环节	异常场景	兜底方案
SKILL.md加载	文件不存在、解析失败	返回“技能配置加载失败，请联系管理员”
skill_id匹配	大模型调用失败、无匹配技能	返回“未识别你的需求，请换种说法”
沙盒请求封装	参数缺失、格式错误	返回“请补充关键信息（如服务器IP、用户ID）”
沙盒调用	沙盒不可达、执行超时、危险操作拦截	返回“沙盒服务暂不可用/执行失败：XXX”
大模型生成结果	大模型调用失败	返回沙盒原始结果（格式化后，便于用户理解）

7.2 性能优化

• SKILL.md缓存：设置5分钟缓存有效期，避免重复读取文件，提升加载效率；

• 沙盒实例池化：沙盒管理器提前创建多个空闲实例，Agent调用时直接复用，减少实例创建/销毁的耗时；

• 异步调用：高并发场景下，将沙盒调用、大模型调用改为异步（使用aiohttp/grpc.aio），提升并发处理能力；

• 参数提取优化：进阶版可调用大模型提取用户参数，替代正则表达式，提升参数提取的精准度和通用性。

7.3 安全防护

• 参数校验：沙盒请求封装时，过滤特殊字符（如;、|、&），避免命令注入攻击；

• 权限控制：Agent服务运行在低权限账户下，沙盒以非root用户执行，仅开放必要权限；

• 网络隔离：沙盒使用独立网络命名空间，仅放通内网白名单，禁止访问核心资产；

• 日志审计：全量记录用户请求、skill_id匹配结果、沙盒执行命令、大模型生成结果，便于安全溯源和问题排查。