18825 字
约 62 分钟
1
第18章 CLI-Agent-命令行智能助手
2026-09-18
第18章 CLI-Agent-命令行智能助手
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch14-cli-agent.html 所属：第六篇-综合实战
第五篇：综合实战 — 终端里的智能助手：用自然语言驱动命令行
18.1 CLI Agent 概述
命令行是程序员的母语。每天我们在终端里敲入 git、docker、kubectl、ssh 等命令来完成工作。但这些命令的参数繁多、语法复杂，即使是经验丰富的工程师也经常需要查文档。CLI Agent 的出现，让我们可以用自然语言与命令行交互——你告诉它"查看最近5次提交记录并找出改动最多的文件"，它自动翻译成 git log --oneline -5 加上一系列管道命令，然后执行并返回结果。
🖥️ 什么是 CLI Agent？
CLI Agent（Command-Line Interface Agent）是指通过命令行终端与 AI Agent 交互的系统。用户输入自然语言或简化的指令，Agent 理解意图后，自动生成并执行对应的 Shell 命令，将结果格式化返回给用户。它本质上是自然语言与 Shell 之间的智能桥梁。
与传统命令行工具不同，CLI Agent 具备上下文理解、多步推理、错误自愈和工具链组合能力。它不是简单的命令补全工具，而是一个能理解你的意图、规划执行步骤、调用多种工具并处理异常的智能体。比如你说"清理所有未使用的 Docker 镜像"，CLI Agent 不会简单执行 docker rmi，而是先执行 docker images -f "dangling=true" 查找悬挂镜像，确认列表后再批量删除，还会跳过正在使用的镜像。
CLI Agent vs Web Agent vs API Agent
AI Agent 按交互界面可分为三大类，它们的差异不仅是前端形态不同，更深刻影响着架构设计、安全模型和适用场景。 📊 三种 Agent 形态对比 | 维度 | CLI Agent | Web Agent | API Agent | | --- | --- | --- | --- | | 交互界面 | 终端命令行 | 浏览器网页 | HTTP/REST 接口 | | 目标用户 | 开发者、运维人员 | 普通用户 | 其他系统/服务 | | 脚本化 | ✓ 天然支持管道和重定向 | ✗ 需要额外自动化工具 | ✓ 通过代码调用 | | 安全边界 | 系统级（文件、进程、网络） | 浏览器沙箱 | API 权限和 Token | | 响应速度 | 极快（本地执行） | 中等（需渲染页面） | 快（无 UI 渲染） | | 上下文感知 | ✓ 感知文件系统、环境变量 | ✓ 感知 DOM、Cookie | ✗ 无环境感知 | | 管道组合 | ✓ stdout → stdin 天然管道 | ✗ 无管道概念 | △ 需手动编排 | 从上表可以看出，CLI Agent 最大的优势在于它生活在开发者的原生工作环境中。开发者不需要切换到浏览器、不需要打开 Postman，直接在终端里就能用自然语言完成复杂操作。更重要的是，CLI Agent 可以无缝接入 Unix 哲学中的管道（pipe）机制——一个 Agent 的输出可以成为另一个命令的输入，形成强大的组合能力。
CLI Agent 的核心优势
⚡ 快速直接
无需 GUI 渲染，命令直达系统。从输入到执行的延迟通常在毫秒级。对于开发者来说，终端永远是最快的信息入口。
📜 可脚本化
CLI Agent 的输入输出可以被 Shell 脚本捕获和组合。今天用 Agent 执行的操作，明天可以封装成自动化脚本。
🔗 管道友好
Unix 管道是 CLI 的灵魂。Agent 生成的命令天然支持 |、>、` 关键文件需确认 | echo "" > /etc/passwd | ### 沙箱执行环境
白名单拦截的是"不能做什么"，沙箱控制的是"能做什么的边界"。即使命令通过了白名单检查，也必须在受限的沙箱环境中执行，防止意外损害。主流的沙箱技术有三种：
Docker 容器沙箱
最强的隔离方案。每个命令在独立容器中执行，文件系统、网络、进程完全隔离。适合执行不信任的命令。缺点是启动慢（秒级）、资源开销大。
Firejail
Linux 轻量级沙箱，使用 Linux namespaces 实现进程隔离。启动快（毫秒级），开销小。适合在宿主机上隔离执行命令。
bubblewrap
Flatpak 使用的沙箱工具，基于 user namespaces。不需要 root 权限，安全性高。适合需要非特权沙箱的场景。
权限降级与命令注入防御
权限降级是沙箱的补充手段。CLI Agent 应始终以非 root 用户运行，对敏感目录（/etc、/var、/root）只读挂载，限制网络访问范围。命令注入防御则要处理用户输入中的特殊字符（;、|、&、$()、反引号），防止恶意构造的输入突破命令边界。
import re
import shlex
import subprocess
from typing import Optional, Tuple
class SafeCommandExecutor:
"""CLI Agent 安全命令执行器"""
# 危险命令正则模式
DANGEROUS_PATTERNS = [
r'rm\s+-rf\s+/(?:\s|$)', # rm -rf /
r'mkfs\.\w+\s+/dev/', # 格式化磁盘
r'dd\s+.*of=/dev/[sh]d', # dd 写入磁盘设备
r':\(\)\{.*\};:', # fork bomb
r'curl\s+.*\|\s*(bash|sh)', # 远程脚本执行
r'wget\s+.*\|\s*(bash|sh)', # 远程脚本执行
r'sudo\s+rm\s+-rf', # sudo 删除
r'chmod\s+-R\s+777\s+/', # 全局权限修改
r'>\s*/etc/(passwd|shadow)', # 覆盖系统文件
r'\$\(.*\).*\$\(.*\)', # 嵌套命令替换
]
# 命令白名单（命令前缀 → 最大超时秒数）
WHITELIST = {
'ls': 10, 'cat': 10, 'grep': 30, 'find': 60,
'git': 120, 'docker': 120, 'kubectl': 60,
'ps': 10, 'top': 10, 'df': 10, 'du': 30,
'head': 10, 'tail': 10, 'wc': 10, 'sort': 30,
'awk': 30, 'sed': 30, 'curl': 30, 'ping': 10,
'netstat': 10, 'lsof': 10, 'ss': 10,
}
@classmethod
def validate(cls, command: str) -> Tuple[bool, str]:
"""验证命令安全性：白名单检查 + 危险模式拦截"""
# 1. 命令注入防御：检查危险模式
for pattern in cls.DANGEROUS_PATTERNS:
if re.search(pattern, command):
return False, f"危险命令模式被拦截: {pattern}"
# 2. 解析命令
try:
parts = shlex.split(command)
except ValueError as e:
return False, f"命令解析失败: {e}"
if not parts:
return False, "空命令"
# 3. 白名单检查
cmd_name = parts[0]
if cmd_name not in cls.WHITELIST:
return False, f"命令 '{cmd_name}' 不在白名单中"
# 4. 检查是否有命令链注入（; && || 等）
if any(op in command for op in [';', '&&', '||']):
# 对链式命令，每一段都要验证
segments = re.split(r'[;|&]+', command)
for seg in segments:
seg = seg.strip()
if seg:
ok, msg = cls.validate(seg)
if not ok:
return False, f"链式命令段验证失败: {msg}"
return True, "通过"
@classmethod
def execute(cls, command: str, timeout: Optional[int] = None) -> dict:
"""在安全环境中执行命令"""
# 1. 验证命令
ok, msg = cls.validate(command)
if not ok:
return {"success": False, "error": msg, "output": ""}
# 2. 确定超时
cmd_name = shlex.split(command)[0]
max_timeout = cls.WHITELIST.get(cmd_name, 30)
timeout = min(timeout or max_timeout, max_timeout)
# 3. 执行命令（非 root、捕获输出、限制超时）
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
user='nobody', # 权限降级：非 root
env={'PATH': '/usr/bin:/bin'}, # 最小化环境变量
)
return {
"success": result.returncode == 0,
"output": result.stdout,
"error": result.stderr,
"returncode": result.returncode,
}
except subprocess.TimeoutExpired:
return {"success": False, "error": f"命令超时（{timeout}s）", "output": ""}
except Exception as e:
return {"success": False, "error": str(e), "output": ""}
# 使用示例
executor = SafeCommandExecutor()
result = executor.execute("git log --oneline -5")
print(result["output"] if result["success"] else result["error"])
# 危险命令会被拦截
result = executor.execute("rm -rf /")
# → {"success": False, "error": "危险命令模式被拦截: rm\\s+-rf\\s+/(?:\\s|$)"}
import { exec } from 'child_process';
import * as re from 're2';
interface ExecResult {
success: boolean;
output: string;
error: string;
returncode: number;
}
class SafeCommandExecutor {
/** CLI Agent 安全命令执行器 */
// 危险命令正则模式
static readonly DANGEROUS_PATTERNS: string[] = [
r'rm\s+-rf\s+/(?:\s|$)', // rm -rf /
r'mkfs\.\w+\s+/dev/', // 格式化磁盘
r'dd\s+.*of=/dev/[sh]d', // dd 写入磁盘设备
r':\(\)\{.*\};:', // fork bomb
r'curl\s+.*\|\s*(bash|sh)', // 远程脚本执行
r'wget\s+.*\|\s*(bash|sh)', // 远程脚本执行
r'sudo\s+rm\s+-rf', // sudo 删除
r'chmod\s+-R\s+777\s+/', // 全局权限修改
r'>\s*/etc/(passwd|shadow)', // 覆盖系统文件
r'\$\(.*\).*\$\(.*\)', // 嵌套命令替换
];
// 命令白名单（命令前缀 → 最大超时秒数）
static readonly WHITELIST: Record = {
'ls': 10, 'cat': 10, 'grep': 30, 'find': 60,
'git': 120, 'docker': 120, 'kubectl': 60,
'ps': 10, 'top': 10, 'df': 10, 'du': 30,
'head': 10, 'tail': 10, 'wc': 10, 'sort': 30,
'awk': 30, 'sed': 30, 'curl': 30, 'ping': 10,
'netstat': 10, 'lsof': 10, 'ss': 10,
};
static validate(command: string): [boolean, string] {
/** 验证命令安全性：白名单检查 + 危险模式拦截 */
// 1. 命令注入防御：检查危险模式
for (const pattern of this.DANGEROUS_PATTERNS) {
const regex = new RegExp(pattern);
if (regex.test(command)) {
return [false, `危险命令模式被拦截: ${pattern}`];
}
}
// 2. 解析命令（简单按空格分割）
const parts = command.trim().split(/\s+/);
if (parts.length === 0 || !parts[0]) {
return [false, '空命令'];
}
// 3. 白名单检查
const cmdName = parts[0];
if (!(cmdName in this.WHITELIST)) {
return [false, `命令 '${cmdName}' 不在白名单中`];
}
// 4. 检查是否有命令链注入（; && || 等）
if ([';', '&&', '||'].some(op => command.includes(op))) {
const segments = command.split(/[;|&]+/);
for (const seg of segments) {
const trimmed = seg.trim();
if (trimmed) {
const [ok, msg] = this.validate(trimmed);
if (!ok) {
return [false, `链式命令段验证失败: ${msg}`];
}
}
}
}
return [true, '通过'];
}
static execute(command: string, timeout?: number): Promise {
/** 在安全环境中执行命令 */
// 1. 验证命令
const [ok, msg] = this.validate(command);
if (!ok) {
return Promise.resolve({ success: false, error: msg, output: '', returncode: -1 });
}
// 2. 确定超时
const cmdName = command.trim().split(/\s+/)[0];
const maxTimeout = this.WHITELIST[cmdName] ?? 30;
const effectiveTimeout = Math.min(timeout ?? maxTimeout, maxTimeout);
// 3. 执行命令（捕获输出、限制超时）
return new Promise((resolve) => {
exec(command, {
timeout: effectiveTimeout * 1000,
env: { ...process.env, PATH: '/usr/bin:/bin' },
maxBuffer: 1024 * 1024,
}, (error, stdout, stderr) => {
if (error) {
if (error.killed) {
resolve({ success: false, error: `命令超时（${effectiveTimeout}s）`, output: '', returncode: -1 });
} else {
resolve({ success: false, error: stderr || error.message, output: stdout, returncode: error.code ?? -1 });
}
} else {
resolve({ success: true, output: stdout, error: stderr, returncode: 0 });
}
});
});
}
}
// 使用示例
(async () => {
const result = await SafeCommandExecutor.execute('git log --oneline -5');
console.log(result.success ? result.output : result.error);
// 危险命令会被拦截
const blocked = await SafeCommandExecutor.execute('rm -rf /');
// → { success: false, error: "危险命令模式被拦截: rm\s+-rf\s+/(?:\s|$)" }
})();
package main
import (
"fmt"
"os/exec"
"regexp"
"strings"
"time"
)
// ExecResult 命令执行结果
type ExecResult struct {
Success bool `json:"success"`
Output string `json:"output"`
Error string `json:"error"`
ReturnCode int `json:"returncode"`
}
// SafeCommandExecutor CLI Agent 安全命令执行器
type SafeCommandExecutor struct{}
// 危险命令正则模式
var dangerousPatterns = []string{
`rm\s+-rf\s+/(?:\s|$)`, // rm -rf /
`mkfs\.\w+\s+/dev/`, // 格式化磁盘
`dd\s+.*of=/dev/[sh]d`, // dd 写入磁盘设备
`:\(\)\{.*\};:`, // fork bomb
`curl\s+.*\|\s*(bash|sh)`, // 远程脚本执行
`wget\s+.*\|\s*(bash|sh)`, // 远程脚本执行
`sudo\s+rm\s+-rf`, // sudo 删除
`chmod\s+-R\s+777\s+/`, // 全局权限修改
`>\s*/etc/(passwd|shadow)`, // 覆盖系统文件
`\$\(.*\).*\$\(.*\)`, // 嵌套命令替换
}
// 命令白名单（命令前缀 → 最大超时秒数）
var whitelist = map[string]int{
"ls": 10, "cat": 10, "grep": 30, "find": 60,
"git": 120, "docker": 120, "kubectl": 60,
"ps": 10, "top": 10, "df": 10, "du": 30,
"head": 10, "tail": 10, "wc": 10, "sort": 30,
"awk": 30, "sed": 30, "curl": 30, "ping": 10,
"netstat": 10, "lsof": 10, "ss": 10,
}
// Validate 验证命令安全性：白名单检查 + 危险模式拦截
func Validate(command string) (bool, string) {
// 1. 命令注入防御：检查危险模式
for _, pattern := range dangerousPatterns {
matched, _ := regexp.MatchString(pattern, command)
if matched {
return false, fmt.Sprintf("危险命令模式被拦截: %s", pattern)
}
}
// 2. 解析命令
parts := strings.Fields(command)
if len(parts) == 0 {
return false, "空命令"
}
// 3. 白名单检查
cmdName := parts[0]
if _, ok := whitelist[cmdName]; !ok {
return false, fmt.Sprintf("命令 '%s' 不在白名单中", cmdName)
}
// 4. 检查是否有命令链注入（; && || 等）
for _, op := range []string{";", "&&", "||"} {
if strings.Contains(command, op) {
var segments []string
if op == ";" {
segments = strings.Split(command, ";")
} else {
segments = strings.Split(command, op)
}
for _, seg := range segments {
seg = strings.TrimSpace(seg)
if seg != "" {
ok, msg := Validate(seg)
if !ok {
return false, fmt.Sprintf("链式命令段验证失败: %s", msg)
}
}
}
}
}
return true, "通过"
}
// Execute 在安全环境中执行命令
func Execute(command string, timeout ...int) ExecResult {
// 1. 验证命令
ok, msg := Validate(command)
if !ok {
return ExecResult{Success: false, Error: msg, Output: "", ReturnCode: -1}
}
// 2. 确定超时
parts := strings.Fields(command)
cmdName := parts[0]
maxTimeout := whitelist[cmdName]
if maxTimeout == 0 {
maxTimeout = 30
}
effectiveTimeout := maxTimeout
if len(timeout) > 0 && timeout[0] > 0 {
if timeout[0] \\s*/etc/(passwd|shadow)"), // 覆盖系统文件
Pattern.compile("\\$\\(.*\\).*\\$\\(.*\\)") // 嵌套命令替换
);
// 命令白名单（命令前缀 → 最大超时秒数）
static final Map WHITELIST = Map.ofEntries(
Map.entry("ls", 10), Map.entry("cat", 10), Map.entry("grep", 30), Map.entry("find", 60),
Map.entry("git", 120), Map.entry("docker", 120), Map.entry("kubectl", 60),
Map.entry("ps", 10), Map.entry("top", 10), Map.entry("df", 10), Map.entry("du", 30),
Map.entry("head", 10), Map.entry("tail", 10), Map.entry("wc", 10), Map.entry("sort", 30),
Map.entry("awk", 30), Map.entry("sed", 30), Map.entry("curl", 30), Map.entry("ping", 10),
Map.entry("netstat", 10), Map.entry("lsof", 10), Map.entry("ss", 10)
);
// 验证命令安全性：白名单检查 + 危险模式拦截
public static Object[] validate(String command) {
// 1. 命令注入防御：检查危险模式
for (Pattern pattern : DANGEROUS_PATTERNS) {
if (pattern.matcher(command).find()) {
return new Object[]{false, "危险命令模式被拦截: " + pattern.pattern()};
}
}
// 2. 解析命令
String[] parts = command.trim().split("\\s+");
if (parts.length == 0 || parts[0].isEmpty()) {
return new Object[]{false, "空命令"};
}
// 3. 白名单检查
String cmdName = parts[0];
if (!WHITELIST.containsKey(cmdName)) {
return new Object[]{false, "命令 '" + cmdName + "' 不在白名单中"};
}
// 4. 检查是否有命令链注入（; && || 等）
if (command.contains(";") || command.contains("&&") || command.contains("||")) {
String[] segments = command.split("[;|&]+");
for (String seg : segments) {
seg = seg.trim();
if (!seg.isEmpty()) {
Object[] result = validate(seg);
if (!(Boolean) result[0]) {
return new Object[]{false, "链式命令段验证失败: " + result[1]};
}
}
}
}
return new Object[]{true, "通过"};
}
// 在安全环境中执行命令
public static Map execute(String command, Integer timeout) {
// 1. 验证命令
Object[] validation = validate(command);
if (!(Boolean) validation[0]) {
return Map.of("success", false, "error", validation[1], "output", "");
}
// 2. 确定超时
String cmdName = command.trim().split("\\s+")[0];
int maxTimeout = WHITELIST.getOrDefault(cmdName, 30);
int actualTimeout = timeout != null ? Math.min(timeout, maxTimeout) : maxTimeout;
// 3. 执行命令（捕获输出、限制超时）
try {
ProcessBuilder pb = new ProcessBuilder("sh", "-c", command);
pb.environment().clear();
pb.environment().put("PATH", "/usr/bin:/bin");
pb.redirectErrorStream(false);
Process process = pb.start();
// 超时控制
if (!process.waitFor(actualTimeout, TimeUnit.SECONDS)) {
process.destroyForcibly();
return Map.of("success", false, "error", "命令超时（" + actualTimeout + "s）", "output", "");
}
String stdout = new String(process.getInputStream().readAllBytes());
String stderr = new String(process.getErrorStream().readAllBytes());
int returnCode = process.exitValue();
Map result = new LinkedHashMap<>();
result.put("success", returnCode == 0);
result.put("output", stdout);
result.put("error", stderr);
result.put("returncode", returnCode);
return result;
} catch (Exception e) {
return Map.of("success", false, "error", e.getMessage(), "output", "");
}
}
public static void main(String[] args) {
// 使用示例
Map result = execute("git log --oneline -5", null);
if ((Boolean) result.get("success")) {
System.out.println(result.get("output"));
} else {
System.out.println(result.get("error"));
}
// 危险命令会被拦截
Map blocked = execute("rm -rf /", null);
System.out.println(blocked.get("error"));
// → 危险命令模式被拦截: rm\s+-rf\s+/(?:\s|$)
}
}
上面的安全命令执行器实现了四层防御：① 正则匹配拦截已知危险模式；② 命令白名单限制可执行范围；③ 链式命令分段验证；④ 运行时权限降级和环境隔离。在实际生产环境中，还需要加入命令审计日志（记录谁在什么时候执行了什么命令）和实时告警（检测到异常命令模式时通知管理员）。
18.4 自然语言转命令（NL2Shell）
NL2Shell（Natural Language to Shell）是 CLI Agent 的核心技术能力。它将用户的自然语言描述转化为可执行的 Shell 命令。这不仅仅是简单的翻译——同一条命令在不同操作系统上语法不同，同一意图可以用多种命令实现，用户描述可能模糊或省略关键信息。NL2Shell 需要结合操作系统上下文、用户历史命令和常识推理来生成正确的命令。
NL2Shell 核心原理
NL2Shell 的核心是一个经过特殊设计的 LLM Prompt。这个 Prompt 包含：操作系统信息（Linux/macOS/WSL）、已安装���工具列表、常用命令模板和约束规则。当用户输入自然语言时，LLM 根据这些上下文信息生成最合适的命令。关键是让 LLM "知道"当前环境——在 macOS 上不应该建议 apt-get，在没有 Docker 的机器上不应该建议 docker 命令。
Few-Shot 示例库设计
Few-Shot 示例是提升 NL2Shell 准确率的关键。好的示例库应该覆盖常见场景、包含边界情况、标注操作系统差异。示例不是越多越好——精选 20-30 个高质量示例比堆砌 200 个重复示例更有效。示例应按场景分类（文件操作、进程管理、网络诊断、版本控制、容器管理），每类 5-6 个。
import platform
import subprocess
from typing import Optional
class NL2Shell:
"""自然语言转 Shell 命令引擎"""
SYSTEM_PROMPT = """你是一个 Shell 命令专家。用户用自然语言描述需求，你生成对应的 Shell 命令。
当前环境：
- 操作系统：{os}
- Shell：{shell}
- 已安装工具：{tools}
规则：
1. 只输出命令，不要解释
2. 如果需要 sudo，在命令前加 sudo
3. 优先使用系统自带工具
4. 危险操作（删除、格式化）加 [需确认] 前缀
5. 不确定时输出最安全的命令
示例：
用户：查看当前目录下最大的5个文件
命令：du -ah . | sort -rh | head -5
用户：找出8080端口被哪个进程占用
命令：lsof -i :8080
用户：统计当前Git仓库的提交人数
命令：git shortlog -sn | wc -l
用户：查看Docker容器的资源使用情况
命令：docker stats --no-stream
用户：将最近3天的日志中ERROR提取出来
命令：find . -name "*.log" -mtime -3 -exec grep -l "ERROR" {{}} \\;
"""
FEW_SHOT_EXAMPLES = [
("查看所有运行的容器", "docker ps"),
("列出最近5次Git提交", "git log --oneline -5"),
("查看磁盘使用情况", "df -h"),
("找出当前目录下所有.py文件", "find . -name '*.py' -type f"),
("查看系统内存使用", "free -h" ]
def __init__(self, api_key: str, base_url: str, model: str):
self.api_key = api_key
self.base_url = base_url
self.model = model
def detect_os(self) -> dict:
"""检测当前操作系统和可用工具"""
import shutil
os_name = platform.system()
shell = "zsh" if os_name == "Darwin" else "bash"
tools = []
for tool in ["docker", "git", "kubectl", "python3", "node", "curl", "jq"]:
if shutil.which(tool):
tools.append(tool)
return {"os": os_name, "shell": shell, "tools": tools}
def generate_command(self, user_input: str) -> Optional[str]:
"""将自然语言转换为 Shell 命令"""
env = self.detect_os()
prompt = self.SYSTEM_PROMPT.format(
os=env["os"], shell=env["shell"], tools=", ".join(env["tools"])
)
# 调用 LLM API 生成命令
examples_text = "\n".join(
f"用户：{q}\n命令：{c}" for q, c in self.FEW_SHOT_EXAMPLES
)
full_prompt = f"{prompt}\n\n示例：\n{examples_text}\n\n用户：{user_input}\n命令："
# 实际调用 LLM...
return full_prompt # 简化示例
# 使用示例
nl2shell = NL2Shell(api_key="sk-xxx", base_url="https://api.openai.com/v1", model="gpt-4")
cmd = nl2shell.generate_command("查看当前目录下所有大于100MB的文件")
print(f"生成的命令：{cmd}")
// Node.js built-in or npm package for: platform
// Node.js built-in or npm package for: subprocess
// TypeScript has built-in types, no import needed for Optional
class NL2Shell {
/** docstring */
static readonly SYSTEM_PROMPT = """你是一个 Shell 命令专家。用户用自然语言描述需求，你生成对应的 Shell 命令。;
// 当前环境：
// - 操作系统：{os}
// - Shell：{shell}
// - 已安装工具：{tools}
// 规则：
// 1. 只输出命令，不要解释
// 2. 如果需要 sudo，在命令前加 sudo
// 3. 优先使用系统自带工具
// 4. 危险操作（删除、格式化）加 [需确认] 前缀
// 5. 不确定时输出最安全的命令
// 示例：
// 用户：查看当前目录下最大的5个文件
// 命令：du -ah . | sort -rh | head -5
// 用户：找出8080端口被哪个进程占用
// 命令：lsof -i :8080
// 用户：统计当前Git仓库的提交人数
// 命令：git shortlog -sn | wc -l
// 用户：查看Docker容器的资源使用情况
// 命令：docker stats --no-stream
// 用户：将最近3天的日志中ERROR提取出来
// 命令：find . -name "*.log" -mtime -3 -exec grep -l "ERROR" {{}} \\;
/** docstring */
static readonly FEW_SHOT_EXAMPLES = [;
// ("查看所有运行的容器", "docker ps"),
// ("列出最近5次Git提交", "git log --oneline -5"),
// ("查看磁盘使用情况", "df -h"),
// ("找出当前目录下所有.py文件", "find . -name '*.py' -type f"),
// ("查看系统内存使用", "free -h" ]
constructor(api_key: string, base_url: string, model: string) {
// self.api_key = api_key
// self.base_url = base_url
// self.model = model
// def detect_os(self) -> dict:
/** docstring */
// Node.js built-in or npm package for: shutil
// os_name = platform.system()
// shell = "zsh" if os_name == "Darwin" else "bash"
// tools = []
for (const tool of ["docker", "git", "kubectl", "python3", "node", "curl", "jq"]) {
if (shutil.which(tool)) {
// tools.append(tool)
return {"os": os_name, "shell": shell, "tools": tools};
// def generate_command(self, user_input: str) -> Optional[str]:
/** docstring */
// env = self.detect_os()
// prompt = self.SYSTEM_PROMPT.format(
// os = env["os"], shell=env["shell"], tools=", ".join(env["tools"])
// )
// 调用 LLM API 生成命令
// examples_text = "\n".join(
// f"用户：{q}\n命令：{c}" for q, c in self.FEW_SHOT_EXAMPLES
// )
// full_prompt = f"{prompt}\n\n示例：\n{examples_text}\n\n用户：{user_input}\n命令："
// 实际调用 LLM...
return full_prompt # 简化示例;
// 使用示例
// nl2shell = NL2Shell(api_key="sk-xxx", base_url="https://api.openai.com/v1", model="gpt-4")
// cmd = nl2shell.generate_command("查看当前目录下所有大于100MB的文件")
console.log(`生成的命令：${$1}`);
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// import platform
// import subprocess
// from typing import Optional
// NL2Shell - CLI Agent class
type NL2Shell struct {
// Python: SYSTEM_PROMPT = """你是一个 Shell 命令专家。用户用自然语言描述需求，你生成对应的 Shell 命令。
// Python: 当前环境：
// Python: - 操作系统：{os}
// Python: - Shell：{shell}
// Python: - 已安装工具：{tools}
// Python: 规则：
// Python: 1. 只输出命令，不要解释
// Python: 2. 如果需要 sudo，在命令前加 sudo
// Python: 3. 优先使用系统自带工具
// Python: 4. 危险操作（删除、格式化）加 [需确认] 前缀
// Python: 5. 不确定时输出最安全的命令
// Python: 示例：
// Python: 用户：查看当前目录下最大的5个文件
// Python: 命令：du -ah . | sort -rh | head -5
// Python: 用户：找出8080端口被哪个进程占用
// Python: 命令：lsof -i :8080
// Python: 用户：统计当前Git仓库的提交人数
// Python: 命令：git shortlog -sn | wc -l
// Python: 用户：查看Docker容器的资源使用情况
// Python: 命令：docker stats --no-stream
// Python: 用户：将最近3天的日志中ERROR提取出来
// Python: 命令：find . -name "*.log" -mtime -3 -exec grep -l "ERROR" {{}} \\;
// Python: FEW_SHOT_EXAMPLES = [
// Python: ("查看所有运行的容器", "docker ps"),
// Python: ("列出最近5次Git提交", "git log --oneline -5"),
// Python: ("查看磁盘使用情况", "df -h"),
// Python: ("找出当前目录下所有.py文件", "find . -name '*.py' -type f"),
// Python: ("查看系统内存使用", "free -h" ]
func New__init__() *__init__ {
return &__init__{}
}
// Python: self.api_key = api_key
// Python: self.base_url = base_url
// Python: self.model = model
// Python: def detect_os(self) -> dict:
// import shutil
// Python: os_name = platform.system()
// Python: shell = "zsh" if os_name == "Darwin" else "bash"
// Python: tools = []
for _, tool := range ["docker", "git", "kubectl", "python3", "node", "curl", "jq"] {
if shutil.which(tool) {
// Python: tools.append(tool)
return {"os": os_name, "shell": shell, "tools": tools}
// Python: def generate_command(self, user_input: str) -> Optional[str]:
// Python: env = self.detect_os()
// Python: prompt = self.SYSTEM_PROMPT.format(
// Python: os=env["os"], shell=env["shell"], tools=", ".join(env["tools"])
// Python: )
// 调用 LLM API 生成命令
// Python: examples_text = "\n".join(
// Python: f"用户：{q}\n命令：{c}" for q, c in self.FEW_SHOT_EXAMPLES
// Python: )
// Python: full_prompt = f"{prompt}\n\n示例：\n{examples_text}\n\n用户：{user_input}\n命令："
// 实际调用 LLM...
return full_prompt # 简化示例
// 使用示例
// Python: nl2shell = NL2Shell(api_key="sk-xxx", base_url="https://api.openai.com/v1", model="gpt-4")
// Python: cmd = nl2shell.generate_command("查看当前目录下所有大于100MB的文件")
fmt.Println(f"生成的命令：{cmd}")
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// import platform
// import subprocess
// from typing import Optional
public class NL2Shell {
// Python: SYSTEM_PROMPT = """你是一个 Shell 命令专家。用户用自然语言描述需求，你生成对应的 Shell 命令。
// Python: 当前环境：
// Python: - 操作系统：{os}
// Python: - Shell：{shell}
// Python: - 已安装工具：{tools}
// Python: 规则：
// Python: 1. 只输出命令，不要解释
// Python: 2. 如果需要 sudo，在命令前加 sudo
// Python: 3. 优先使用系统自带工具
// Python: 4. 危险操作（删除、格式化）加 [需确认] 前缀
// Python: 5. 不确定时输出最安全的命令
// Python: 示例：
// Python: 用户：查看当前目录下最大的5个文件
// Python: 命令：du -ah . | sort -rh | head -5
// Python: 用户：找出8080端口被哪个进程占用
// Python: 命令：lsof -i :8080
// Python: 用户：统计当前Git仓库的提交人数
// Python: 命令：git shortlog -sn | wc -l
// Python: 用户：查看Docker容器的资源使用情况
// Python: 命令：docker stats --no-stream
// Python: 用户：将最近3天的日志中ERROR提取出来
// Python: 命令：find . -name "*.log" -mtime -3 -exec grep -l "ERROR" {{}} \\;
// Python: FEW_SHOT_EXAMPLES = [
// Python: ("查看所有运行的容器", "docker ps"),
// Python: ("列出最近5次Git提交", "git log --oneline -5"),
// Python: ("查看磁盘使用情况", "df -h"),
// Python: ("找出当前目录下所有.py文件", "find . -name '*.py' -type f"),
// Python: ("查看系统内存使用", "free -h" ]
public NL2Shell(api_keyString, base_urlString, modelString) {
// Python: self.api_key = api_key
// Python: self.base_url = base_url
// Python: self.model = model
// Python: def detect_os(self) -> dict:
// import shutil
// Python: os_name = platform.system()
// Python: shell = "zsh" if os_name == "Darwin" else "bash"
// Python: tools = []
for (var tool : ["docker", "git", "kubectl", "python3", "node", "curl", "jq"]) {
if (shutil.which(tool)) {
// Python: tools.append(tool)
return {"os": os_name, "shell": shell, "tools": tools};
// Python: def generate_command(self, user_input: str) -> Optional[str]:
// Python: env = self.detect_os()
// Python: prompt = self.SYSTEM_PROMPT.format(
// Python: os=env["os"], shell=env["shell"], tools=", ".join(env["tools"])
// Python: )
// 调用 LLM API 生成命令
// Python: examples_text = "\n".join(
// Python: f"用户：{q}\n命令：{c}" for q, c in self.FEW_SHOT_EXAMPLES
// Python: )
// Python: full_prompt = f"{prompt}\n\n示例：\n{examples_text}\n\n用户：{user_input}\n命令："
// 实际调用 LLM...
return full_prompt # 简化示例;
// 使用示例
// Python: nl2shell = NL2Shell(api_key="sk-xxx", base_url="https://api.openai.com/v1", model="gpt-4")
// Python: cmd = nl2shell.generate_command("查看当前目录下所有大于100MB的文件")
System.out.println(String.format("$1"));
}
}
NL2Shell 准确率评测 | 方法 | 准确率 | 平均延迟 | 适用场景 | | --- | --- | --- | --- | | Zero-Shot | 62% | 0.8s | 简单命令 | | Few-Shot (5例) | 78% | 1.2s | 常见操作 | | Few-Shot (20例) | 89% | 1.5s | 复杂场景 | | Fine-tuned Model | 94% | 0.5s | 特定领域 | | RAG + Few-Shot | 91% | 1.8s | 混合场景 | 用户输入
"查看大文件"
环境检测 OS/Shell/Tools
Few-Shot 匹配 20+ 示例
命令生成 du -ah . | sort -rh
安全校验 白名单/拦截
18.5 场景化对接能力
CLI Agent 的真正价值不在于单条命令的执行，而在于它能将自然语言意图串联成完整的运维流程。一个"部署回滚"的指令背后，可能涉及版本查询、健康检查、镜像回退、流量切换和通知发送五个步骤。CLI Agent 需要理解每个场景的上下文和依赖关系，自动编排命令序列。
五大核心场景 | 场景 | 典型操作 | Agent 价值 | 复杂度 | | --- | --- | --- | --- | | DevOps | CI/CD、日志分析、部署回滚 | 自动编排流水线、智能回滚决策 | ★★★★☆ | | 运维 | 监控、告警、自动修复 | 7×24 值守、故障自愈 | ★★★★★ | | 开发 | 代码审查、分支管理、自动测试 | PR 自动审查、冲突解决 | ★★★☆☆ | | 数据分析 | SQL 生成、数据清洗、报表 | 自然语言查数据、自动可视化 | ★★★☆☆ | | 安全 | 漏洞扫描、日志审计、合规检查 | 批量扫描、异常行为识别 | ★★★★☆ | ### DevOps CI/CD 对接实战
import subprocess
import json
from typing import List
class DevOpsAgent:
"""CLI Agent 的 DevOps 场景对接"""
def __init__(self, project_name: str):
self.project = project_name
self.pipeline_steps = []
def deploy(self, env: str = "staging") -> dict:
"""一键部署：构建 → 测试 → 推送 → 发布"""
steps = [
("构建镜像", f"docker build -t {self.project}:{env} ."),
("运行测试", "pytest tests/ --cov --cov-report=xml"),
("推送镜像", f"docker push registry.cn-hangzhou.aliyuncs.com/xfg/{self.project}:{env}"),
("滚动更新", f"kubectl set image deployment/{self.project} app={self.project}:{env}"),
("健康检查", f"kubectl rollout status deployment/{self.project} --timeout=300s"),
]
results = {}
for step_name, cmd in steps:
print(f"▶ {step_name}...")
ret = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=300)
if ret.returncode != 0:
results[step_name] = f"FAIL: {ret.stderr[:200]}"
self.rollback(env)
break
results[step_name] = "OK"
return results
def rollback(self, env: str):
"""自动回滚到上一个稳定版本"""
print("⚠️ 部署失败，执行回滚...")
subprocess.run(f"kubectl rollout undo deployment/{self.project}", shell=True)
def analyze_logs(self, since: str = "5m") -> str:
"""智能日志分析：提取错误 + 归类 + 建议"""
cmd = f"kubectl logs deployment/{self.project} --since={since} --tail=1000"
logs = subprocess.run(cmd, shell=True, capture_output=True, text=True).stdout
errors = [l for l in logs.split("\n") if "ERROR" in l or "Exception" in l]
return f"发现 {len(errors)} 条错误\n" + "\n".join(errors[:10])
# 使用：用户说"部署到staging" → Agent 自动执行5步流水线
agent = DevOpsAgent("user-service")
result = agent.deploy("staging")
print(json.dumps(result, indent=2, ensure_ascii=False))
// Node.js built-in or npm package for: subprocess
// Node.js built-in or npm package for: json
// TypeScript has built-in types, no import needed for List
class DevOpsAgent {
/** docstring */
constructor(project_name: string) {
// self.project = project_name
// self.pipeline_steps = []
// def deploy(self, env: str = "staging") -> dict:
/** docstring */
// steps = [
// ("构建镜像", f"docker build -t {self.project}:{env} ."),
// ("运行测试", "pytest tests/ --cov --cov-report=xml"),
// ("推送镜像", f"docker push registry.cn-hangzhou.aliyuncs.com/xfg/{self.project}:{env}"),
// ("滚动更新", f"kubectl set image deployment/{self.project} app={self.project}:{env}"),
// ("健康检查", f"kubectl rollout status deployment/{self.project} --timeout=300s"),
// ]
// results = {}
for (const [step_name, cmd] of steps) {
console.log(`▶ ${$1}...`);
// ret = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=300)
if (ret.returncode != 0) {
// results[step_name] = f"FAIL: {ret.stderr[:200]}"
// self.rollback(env)
break;
// results[step_name] = "OK"
return results;
rollback(env: string) {
/** docstring */
console.log("⚠️ 部署失败，执行回滚...");
// subprocess.run(f"kubectl rollout undo deployment/{self.project}", shell=True)
// def analyze_logs(self, since: str = "5m") -> str:
/** docstring */
// cmd = f"kubectl logs deployment/{self.project} --since={since} --tail=1000"
// logs = subprocess.run(cmd, shell=True, capture_output=True, text=True).stdout
// errors = [l for l in logs.split("\n") if "ERROR" in l or "Exception" in l]
return `发现 {len(errors)} 条错误\n` + "\n".join(errors[:10]);
// 使用：用户说"部署到staging" → Agent 自动执行5步流水线
// agent = DevOpsAgent("user-service")
// result = agent.deploy("staging")
console.log(json.dumps(result, indent=2, ensure_ascii=False));
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// import subprocess
// import json
// from typing import List
// DevOpsAgent - CLI Agent class
type DevOpsAgent struct {
func New__init__() *__init__ {
return &__init__{}
}
// Python: self.project = project_name
// Python: self.pipeline_steps = []
// Python: def deploy(self, env: str = "staging") -> dict:
// Python: steps = [
// Python: ("构建镜像", f"docker build -t {self.project}:{env} ."),
// Python: ("运行测试", "pytest tests/ --cov --cov-report=xml"),
// Python: ("推送镜像", f"docker push registry.cn-hangzhou.aliyuncs.com/xfg/{self.project}:{env}"),
// Python: ("滚动更新", f"kubectl set image deployment/{self.project} app={self.project}:{env}"),
// Python: ("健康检查", f"kubectl rollout status deployment/{self.project} --timeout=300s"),
// Python: ]
// Python: results = {}
// Python: for step_name, cmd in steps:
fmt.Println(f"▶ {step_name}...")
// Python: ret = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=300)
if ret.returncode != 0 {
// Python: results[step_name] = f"FAIL: {ret.stderr[:200]}"
// Python: self.rollback(env)
break
// Python: results[step_name] = "OK"
return results
func rollback() {
fmt.Println("⚠️ 部署失败，执行回滚...")
// Python: subprocess.run(f"kubectl rollout undo deployment/{self.project}", shell=True)
// Python: def analyze_logs(self, since: str = "5m") -> str:
// Python: cmd = f"kubectl logs deployment/{self.project} --since={since} --tail=1000"
// Python: logs = subprocess.run(cmd, shell=True, capture_output=True, text=True).stdout
// Python: errors = [l for l in logs.split("\n") if "ERROR" in l or "Exception" in l]
return f"发现 {len(errors)} 条错误\n" + "\n".join(errors[:10])
// 使用：用户说"部署到staging" → Agent 自动执行5步流水线
// Python: agent = DevOpsAgent("user-service")
// Python: result = agent.deploy("staging")
fmt.Println(json.dumps(result, indent=2, ensure_ascii=False))
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// import subprocess
// import json
// from typing import List
public class DevOpsAgent {
public DevOpsAgent(project_nameString) {
// Python: self.project = project_name
// Python: self.pipeline_steps = []
// Python: def deploy(self, env: str = "staging") -> dict:
// Python: steps = [
// Python: ("构建镜像", f"docker build -t {self.project}:{env} ."),
// Python: ("运行测试", "pytest tests/ --cov --cov-report=xml"),
// Python: ("推送镜像", f"docker push registry.cn-hangzhou.aliyuncs.com/xfg/{self.project}:{env}"),
// Python: ("滚动更新", f"kubectl set image deployment/{self.project} app={self.project}:{env}"),
// Python: ("健康检查", f"kubectl rollout status deployment/{self.project} --timeout=300s"),
// Python: ]
// Python: results = {}
// Python: for step_name, cmd in steps:
System.out.println(String.format("$1"));
// Python: ret = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=300)
if (ret.returncode != 0) {
// Python: results[step_name] = f"FAIL: {ret.stderr[:200]}"
// Python: self.rollback(env)
break;
// Python: results[step_name] = "OK"
return results;
public static void rollback(envString) {
System.out.println("⚠️ 部署失败，执行回滚...");
// Python: subprocess.run(f"kubectl rollout undo deployment/{self.project}", shell=True)
// Python: def analyze_logs(self, since: str = "5m") -> str:
// Python: cmd = f"kubectl logs deployment/{self.project} --since={since} --tail=1000"
// Python: logs = subprocess.run(cmd, shell=True, capture_output=True, text=True).stdout
// Python: errors = [l for l in logs.split("\n") if "ERROR" in l or "Exception" in l]
return f"发现 {len(errors)} 条错误\n" + "\n".join(errors[:10]);
// 使用：用户说"部署到staging" → Agent 自动执行5步流水线
// Python: agent = DevOpsAgent("user-service")
// Python: result = agent.deploy("staging")
System.out.println(json.dumps(result, indent=2, ensure_ascii=False));
}
}
18.6 主流 CLI Agent 框架实战
CLI Agent 领域已经涌现出多个成熟框架，各有侧重。OpenClaw CLI 强调安全沙箱和多模态对接；Claude Code 专注于代码理解和编辑；Aider 走 Git 原生路线；Cursor CLI 则将 IDE 体验搬到了终端。理解这些框架的设计理念，有助于选择合适的工具或自建框架。
主流框架对比 | 框架 | 核心理念 | 安全机制 | 扩展性 | 适用场景 | | --- | --- | --- | --- | --- | | OpenClaw CLI | 安全优先、Skill 可插拔 | 四级沙箱、审批机制 | Skill 生态、MCP 协议 | 运维、DevOps、通用 | | Claude Code | 代码理解、上下文感知 | 只读默认、权限申请 | MCP 工具、自定义命令 | 代码开发、重构 | | Aider | Git 原生、PAIR 编程 | Git 回滚、dry-run | 自定义命令、模型可换 | 代码编辑、快速原型 | | Cursor CLI | IDE 级体验、AI 原生 | 工作区隔离 | 插件系统 | 全栈开发、调试 | | 自建框架 | 完全可控、深度定制 | 自定义沙箱 | 无限扩展 | 特殊需求、内部工具 | ### 自建 CLI Agent 框架
import subprocess
import re
import os
from typing import Optional, List
from dataclasses import dataclass
@dataclass
class CommandResult:
success: bool
output: str
error: str
command: str
class CLIAgent:
"""最小可用 CLI Agent 框架"""
# 危险命令模式
DANGEROUS_PATTERNS = [
r"rm\s+-rf\s+/",
r"dd\s+if=",
r"mkfs\.",
r"shutdown",
r"reboot",
r">\s*/dev/sda",
r"chmod\s+777\s+/",
r":\(\)\{.*\|.*&\};", # fork bomb
]
def __init__(self, allowed_commands: List[str] = None):
self.history: List[str] = []
self.allowed = allowed_commands or [
"ls", "cat", "grep", "find", "wc", "head", "tail", "sort",
"git", "docker", "kubectl", "python3", "node", "curl", "jq",
"du", "df", "ps", "top", "lsof", "netstat", "diff", "echo",
]
def is_safe(self, command: str) -> bool:
"""安全检查：白名单 + 危险模式拦截"""
# 检查危险模式
for pattern in self.DANGEROUS_PATTERNS:
if re.search(pattern, command):
return False
# 检查白名单
base_cmd = command.strip().split()[0] if command.strip() else ""
return base_cmd in self.allowed
def execute(self, command: str, timeout: int = 30) -> CommandResult:
"""安全执行命令"""
if not self.is_safe(command):
return CommandResult(False, "", "命令被安全策略拦截", command)
try:
ret = subprocess.run(
command, shell=True, capture_output=True,
text=True, timeout=timeout,
env={**os.environ, "LANG": "en_US.UTF-8"}
)
self.history.append(command)
return CommandResult(
ret.returncode == 0, ret.stdout, ret.stderr, command
)
except subprocess.TimeoutExpired:
return CommandResult(False, "", f"命令超时（{timeout}s）", command)
def chat(self, user_input: str) -> str:
"""主循环：接收自然语言 → 执行命令 → 返回结果"""
# 这里接入 NL2Shell 模块
# 简化示例：直接执行用户输入的命令
result = self.execute(user_input)
if result.success:
return f"✅ {result.output[:500]}"
else:
return f"❌ {result.error}"
# 使用
agent = CLIAgent()
print(agent.chat("ls -la"))
print(agent.chat("git log --oneline -5"))
// Node.js built-in or npm package for: subprocess
// Node.js built-in or npm package for: re
// Node.js built-in or npm package for: os
// TypeScript has built-in types, no import needed for Optional, List
import {dataclass} from 'dataclasses';
// @dataclass
class CommandResult {
// success: bool
// output: str
// error: str
// command: str
class CLIAgent {
/** docstring */
// 危险命令模式
static readonly DANGEROUS_PATTERNS = [;
// r"rm\s+-rf\s+/",
// r"dd\s+if=",
// r"mkfs\.",
// r"shutdown",
// r"reboot",
// r">\s*/dev/sda",
// r"chmod\s+777\s+/",
// r":\(\)\{.*\|.*&\};", # fork bomb
// ]
constructor(allowed_commands: str[] = null) {
// self.history: List[str] = []
// self.allowed = allowed_commands or [
// "ls", "cat", "grep", "find", "wc", "head", "tail", "sort",
// "git", "docker", "kubectl", "python3", "node", "curl", "jq",
// "du", "df", "ps", "top", "lsof", "netstat", "diff", "echo",
// ]
// def is_safe(self, command: str) -> bool:
/** docstring */
// 检查危险模式
for (const pattern of self.DANGEROUS_PATTERNS) {
if (re.search(pattern, command)) {
return false;
// 检查白名单
// base_cmd = command.strip().split()[0] if command.strip() else ""
return base_cmd in self.allowed;
// def execute(self, command: str, timeout: int = 30) -> CommandResult:
/** docstring */
if (!self.is_safe(command)) {
return CommandResult(false, "", "命令被安全策略拦截", command);
try {
// ret = subprocess.run(
// command, shell=True, capture_output=True,
// text = True, timeout=timeout,
// env = {**os.environ, "LANG": "en_US.UTF-8"}
// )
// self.history.append(command)
return CommandResult(;
// ret.returncode == 0, ret.stdout, ret.stderr, command
// )
// except subprocess.TimeoutExpired:
return CommandResult(false, "", `命令超时（${$1}s）`, command);
// def chat(self, user_input: str) -> str:
/** docstring */
// 这里接入 NL2Shell 模块
// 简化示例：直接执行���户输入的命令
// result = self.execute(user_input)
if (result.success) {
return `✅ {result.output[:500]}`;
} else {
return `❌ {result.error}`;
// 使用
// agent = CLIAgent()
console.log(agent.chat("ls -la"));
console.log(agent.chat("git log --oneline -5"));
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// import subprocess
// import re
// import os
// from typing import Optional, List
// from dataclasses import dataclass
// CommandResult - CLI Agent class
type CommandResult struct {
// Python: success: bool
// Python: output: str
// Python: error: str
// Python: command: str
// CLIAgent - CLI Agent class
type CLIAgent struct {
// 危险命令模式
// Python: DANGEROUS_PATTERNS = [
// Python: r"rm\s+-rf\s+/",
// Python: r"dd\s+if=",
// Python: r"mkfs\.",
// Python: r"shutdown",
// Python: r"reboot",
// Python: r">\s*/dev/sda",
// Python: r"chmod\s+777\s+/",
// Python: r":\(\)\{.*\|.*&\};", # fork bomb
// Python: ]
func New__init__() *__init__ {
return &__init__{}
}
// Python: self.history: List[str] = []
// Python: self.allowed = allowed_commands or [
// Python: "ls", "cat", "grep", "find", "wc", "head", "tail", "sort",
// Python: "git", "docker", "kubectl", "python3", "node", "curl", "jq",
// Python: "du", "df", "ps", "top", "lsof", "netstat", "diff", "echo",
// Python: ]
// Python: def is_safe(self, command: str) -> bool:
// 检查危险模式
for _, pattern := range self.DANGEROUS_PATTERNS {
if re.search(pattern, command) {
return false
// 检查白名单
// Python: base_cmd = command.strip().split()[0] if command.strip() else ""
return base_cmd in self.allowed
// Python: def execute(self, command: str, timeout: int = 30) -> CommandResult:
if not self.is_safe(command) {
return CommandResult(false, "", "命令被安全策略拦截", command)
// try block
// Python: ret = subprocess.run(
// Python: command, shell=True, capture_output=True,
// Python: text=True, timeout=timeout,
// Python: env={**os.environ, "LANG": "en_US.UTF-8"}
// Python: )
// Python: self.history.append(command)
return CommandResult(
// Python: ret.returncode == 0, ret.stdout, ret.stderr, command
// Python: )
// except block
return CommandResult(false, "", f"命令超时（{timeout}s）", command)
// Python: def chat(self, user_input: str) -> str:
// 这里接入 NL2Shell 模块
// 简化示例：直接执行用户输入的命令
// Python: result = self.execute(user_input)
if result.success {
return f"✅ {result.output[:500]}"
} else {
return f"❌ {result.error}"
// 使用
// Python: agent = CLIAgent()
fmt.Println(agent.chat("ls -la"))
fmt.Println(agent.chat("git log --oneline -5"))
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// import subprocess
// import re
// import os
// from typing import Optional, List
// from dataclasses import dataclass
public class CommandResult {
// Python: success: bool
// Python: output: str
// Python: error: str
// Python: command: str
public class CLIAgent {
// 危险命令模式
// Python: DANGEROUS_PATTERNS = [
// Python: r"rm\s+-rf\s+/",
// Python: r"dd\s+if=",
// Python: r"mkfs\.",
// Python: r"shutdown",
// Python: r"reboot",
// Python: r">\s*/dev/sda",
// Python: r"chmod\s+777\s+/",
// Python: r":\(\)\{.*\|.*&\};", # fork bomb
// Python: ]
public CLIAgent(allowed_commandsList ) {
// Python: self.history: List[str] = []
// Python: self.allowed = allowed_commands or [
// Python: "ls", "cat", "grep", "find", "wc", "head", "tail", "sort",
// Python: "git", "docker", "kubectl", "python3", "node", "curl", "jq",
// Python: "du", "df", "ps", "top", "lsof", "netstat", "diff", "echo",
// Python: ]
// Python: def is_safe(self, command: str) -> bool:
// 检查危险模式
for (var pattern : self.DANGEROUS_PATTERNS) {
if (re.search(pattern, command)) {
return false;
// 检查白名单
// Python: base_cmd = command.strip().split()[0] if command.strip() else ""
return base_cmd in self.allowed;
// Python: def execute(self, command: str, timeout: int = 30) -> CommandResult:
if (!self.is_safe(command)) {
return CommandResult(false, "", "命令被安全策略拦截", command);
// Python: try:
// Python: ret = subprocess.run(
// Python: command, shell=True, capture_output=True,
// Python: text=True, timeout=timeout,
// Python: env={**os.environ, "LANG": "en_US.UTF-8"}
// Python: )
// Python: self.history.append(command)
return CommandResult(;
// Python: ret.returncode == 0, ret.stdout, ret.stderr, command
// Python: )
// Python: except subprocess.TimeoutExpired:
return CommandResult(false, "", f"命令超时（{timeout}s）", command);
// Python: def chat(self, user_input: str) -> str:
// 这里接入 NL2Shell 模块
// 简化示例：直接执行用户输入的命令
// Python: result = self.execute(user_input)
if (result.success) {
return f"✅ {result.output[:500]}";
} else {
return f"❌ {result.error}";
// 使用
// Python: agent = CLIAgent()
System.out.println(agent.chat("ls -la"));
System.out.println(agent.chat("git log --oneline -5"));
}
}
OpenClaw CLI 安全沙箱 + Skill MCP 协议
Claude Code 代码理解 上下文感知
Aider Git 原生 PAIR 编程
自建框架 完全可控 深度定制
共同能力：NL2Shell · 安全沙箱 · 命令执行 · 上下文管理
选型决策 安全优先 → OpenClaw CLI 代码优先 → Claude Code / Aider 特殊需求 → 自建框架 📋 八股总结 — 面试高频考点
Q1: CLI Agent 和 Web Agent 的核心区别是什么？
CLI Agent 通过命令行终端与用户交互，输入输出都是文本，适合开发者和技术运维人员。Web Agent 运行在浏览器中，有丰富的 UI 交互能力。核心区别体现在三个方面：交互模式——CLI 是纯文本流，支持管道和重定向，可以与其他命令行工具无缝串联；Web 是 GUI 交互，依赖点击和表单。执行环境——CLI Agent 直接运行在用户的操作系统上，可以访问文件系统、进程、网络等全部系统资源；Web Agent 受浏览器沙箱限制，只能通过 API 间接访问系统资源。自动化能力——CLI Agent 可以轻松嵌入 Shell 脚本、CI/CD Pipeline、Cron 任务中实现无人值守自动化；Web Agent 通常需要人工操作或额外的 RPA 工具。CLI Agent 的优势在于快速、可脚本化、管道友好，但学习门槛较高；Web Agent 门槛低但灵活性受限。
Q2: 如何设计 CLI Agent 的安全沙箱？请描述四层防呆机制。
安全沙箱是 CLI Agent 的生命线。四层防呆机制从外到内依次为：第一层：命令白名单——只允许预定义的安全命令执行，基于命令前缀匹配，如 ls、cat、grep、git 等常见命令，拦截未知命令。第二层：危险模式正则拦截——即使命令在白名单中，也要检查参数是否包含危险模式，如 rm -rf /、dd if=、fork bomb 等，使用正则表达式匹配已知攻击模式。第三层：沙箱执行环境——在隔离的环境中执行命令，方案包括 Docker 容器（文件系统隔离 + 资源限制）、Firejail（Linux 轻量沙箱）、bubblewrap（Flatpak 使用的沙箱），以及权限降级（非 root 运行、只读挂载关键目录）。第四层：人工审批机制——对于敏感操作（如部署、删除、网络请求），弹出确认提示要求用户明确同意后才执行。四层机制层层递进，即使某一层被绕过，下一层仍能兜底。
Q3: 如何提升 NL2Shell 的准确率？
NL2Shell 准确率提升是一个系统工程，从五个方面入手：1. Few-Shot 示例优化——精选 20-30 个高质量示例，按场景分类（文件操作、进程管理、网络诊断、版本控制、容器管理），每类 5-6 个，覆盖常见和边界情况。示例质量比数量更重要。2. 上下文感知——在 Prompt 中注入操作系统类型、Shell 版本、已安装工具列表、当前工作目录等信息，让 LLM 生成与环境匹配的命令。比如在 macOS 上不生成 apt-get 命令。3. RAG 增强——将用户历史命令库和 man page 作为知识库，通过向量检索找到相似场景的参考命令，作为 Few-Shot 示例动态注入。4. 多候选 + 投票——让 LLM 生成 3-5 个候选命令，通过规则校验（语法检查、安全检查）和二次 LLM 评分选出最佳命令。5. 持续学习——记录用户对生成命令的修改和反馈，将有价值的修正加入 Few-Shot 库，形成正向飞轮。
Q4: CLI Agent 命令执行失败后如何实现自愈？
命令执行失败后的自愈是 CLI Agent 智能化的核心体现。自愈流程分四步：第一步：错误分类——将失败原因分为可重试类（网络超时、资源暂时不可用）、需修正类（命令语法错误、权限不足）、需人工介入类（文件不存在、服务未安装）。第二步：自动修正——对于语法错误，将错误信息回传给 LLM 重新生成命令；对于权限问题，尝试加 sudo 或切换用户；对于路径问题，尝试模糊匹配查找正确路径。第三步：重试策略——对于瞬时故障，采用指数退避重试（1s → 2s → 4s），最多重试 3 次；对于复杂操作，回退到上一个已知稳定状态。第四步：降级与上报——如果自动修正失败，降级为提示用户手动处理，同时提供错误分析和建议命令。关键设计：每次自愈都要记录日志，形成"错误 → 修正"映射表，逐步提升自愈能力。
Q5: 命令注入攻击的原理和防御策略是什么？
命令注入是 CLI Agent 面临的最严重的安全威胁。攻击原理：攻击者通过自然语言输入中嵌入 Shell 元字符（如 ;、|、&、$()、``），使 LLM 生成的命令包含恶意载荷。例如用户输入"查看文件 ; rm -rf /"，如果 LLM 直接拼接为 cat file ; rm -rf /，就会导致灾难性后果。另一个向量是环境变量注入，如 $HOME 可能被篡改。防御策略：1. 输入净化——在自然语言输入阶段检测并过滤 Shell 元字符，对 $、;、|、&、反引号等字符进行转义。2. 命令结构化解析——不使用 shell=True 执行，而是将命令拆分为程序名 + 参数列表（subprocess.run(["ls", "-la"])），避免 Shell 解释器介入。3. 参数白名单——对每个命令定义允许的参数集合，拒绝不在白名单中的参数。4. 沙箱隔离——即使在容器内执行，也限制网络访问、文件系统写入范围和系统调用。5. 审计日志——记录所有执行的命令、输入来源和执行结果，便于事后追溯。
Q6: OpenClaw CLI 的架构优势和设计理念是什么？
OpenClaw CLI 是一个以安全为核心、Skill 可插拔的 CLI Agent 框架。架构优势：1. Skill 生态——每个 Skill 是一个独立的能力包（SKILL.md + 脚本），可以热加载、组合使用。用户可以自建 Skill 也可以从 SkillHub 安装社区 Skill，形成能力飞轮。2. 四级安全模型——命令白名单 → 危险模式拦截 → 沙箱执行 → 人工审批，层层递进。尤其审批机制让用户对敏感操作有最终控制权。3. MCP 协议支持——通过 Model Context Protocol 标准化工具调用，支持跨模型、跨平台的工具互操作。4. 多模态对接——不只是命令行，还能对接消息平台（Telegram、Discord、企业微信）、设备节点（手机摄像头、屏幕录制）等。5. 会话管理——支持主会话和子会话隔离，复杂任务可以派发子任务并行执行，主会话不被阻塞。设计理念是"安全优先、可扩展、开发者友好"——先保证不出事，再通过 Skill 生态无限扩展能力。
Q7: CLI Agent 在 DevOps 场景中有哪些典型应用？
CLI Agent 在 DevOps 场景中大有可为，三个典型应用：应用一：智能 CI/CD 流水线——传统 CI/CD 需要手写 YAML 配置，CLI Agent 可以根据自然语言描述自动生成 Pipeline 配置，在部署失败时自动分析日志、定位问题、执行回滚。例如"部署 user-service 到 staging 环境"，Agent 自动执行构建→测试→推送→滚动更新→健康检查五步流水线，任一步骤失败自动回滚。应用二：日志智能分析——线上服务出问题时，运维人员需要在一堆日志中找到根因。CLI Agent 可以自动收集多服务日志（kubectl logs）、过滤错误（grep ERROR）、归类异常模式、关联时间线，最后给出根因假设和修复建议。比传统 ELK 查询更灵活，因为 Agent 能理解日志语义。应用三：故障自愈——监控告警触发后，CLI Agent 自动执行诊断命令（查看 CPU/内存/磁盘、检查进程状态、分析网络连接），根据预设规则执行修复操作（重启服务、清理缓存、扩容节点），并将处理过程记录到工单系统。
Q8: CLI Agent 如何管理上下文和处理长命令输出？
上下文管理是 CLI Agent 的核心技术挑战。LLM 的上下文窗口有限（4K-128K tokens），而命令输出可能动辄上万行。策略一：输出摘要——对超长输出进行分层摘要：首先截取前 N 行和后 N 行（头尾通常最重要），中间部分用 LLM 生成摘要。对于结构化输出（如 JSON、表格），提取关键字段而非全文。策略二：滑动窗口——维护一个滑动上下文窗口，保留最近 K 轮对话和命令结果。超过窗口的旧内容通过摘要压缩，保留关键信息（执行了什么命令、结果成功/失败、关键输出）。策略三：语义检索——将所有历史命令和输出存入向量数据库，当需要历史信息时通过语义检索找到相关上下文，而不是全量加载。策略四：上下文分区——将上下文分为"任务上下文"（当前任务目标）、"环境上下文"（OS、目录、工具）、"历史上下文"（最近命令和结果），按优先级分配 token 预算，确保任务关键信息不被挤占。
Q9: CLI Agent 中如何落地 RBAC 权限控制？
RBAC（Role-Based Access Control）在 CLI Agent 中的落地需要考虑命令级别的权限粒度。角色定义：通常分为 Viewer（只读权限，可执行 ls/ps/cat/grep 等查看类命令）、Operator（可执行运维操作，如 git/docker/kubectl 的非破坏性命令）、Admin（全权限，包括删除、部署、配置修改）。权限矩阵：为每个角色定义命令白名单和参数约束。例如 Operator 可以执行 kubectl get/describe/logs 但不能执行 kubectl delete。实现机制：1. 命令分类——将所有命令分为 read/write/admin 三级，写入配置文件。2. 参数检查——不仅检查命令本身，还检查参数。如 git push 属于 write 级别，需要 Operator 以上权限。3. 审批流程——越权操作需要更高级别用户审批。CLI Agent 可以生成审批请求，发送到 IM 工具（如企业微信/飞书），审批通过后自动执行。4. 审计日志——记录每条命令的执行者、时间、结果、权限级别，支持事后审计和合规检查。5. 动态权限——支持临时提权（如紧急运维时 1 小时内获得 Admin 权限），超时自动降级。
Q10: CLI Agent 的未来发展趋势是什么？
CLI Agent 正处于快速发展期，五个趋势值得关注：趋势一：多模态融合——未来的 CLI Agent 不只处理文本，还能接收截图（分析 UI 错误）、语音指令（免手输入）、甚至视频（分析操作录屏）。多模态输入让 Agent 的感知能力大幅提升。趋势二：Agent 间协作——多个 CLI Agent 可以组成协作网络，A Agent 负责前端部署，B Agent 负责后端部署，C Agent 负责数据库迁移，通过 A2A 协议协调完成复杂任务。类似微服务架构，每个 Agent 专注一个领域。趋势三：自进化能力——Agent 持续学习用户的命令习惯、偏好和工作流，自动优化 Few-Shot 示例库、调整命令推荐策略。从"通用 Agent"演进为"个人定制 Agent"。趋势四：安全可信——引入形式化验证（执行前数学证明命令安全）、可解释 AI（解释为什么选择这条命令）、零信任架构（每条命令都需要动态授权）。趋势五：与 IDE/CI 深度集成——CLI Agent 不再是独立工具，而是嵌入 VS Code、JetBrains、GitHub Actions 中，成为开发流程的有机组成部分。
CLI-Anything：让所有软件变成 Agent 的手脚
上面的框架对比聚焦于Agent 本身的架构——Agent 如何安全地执行 Shell 命令。但 CLI Agent 还有一个更深层的问题：**Agent 的"手脚"够不够多？**一个只会跑 Shell 命令的 Agent，能操作数据库、能调用 API，但如果你让它打开 GIMP 处理一张图片、用 Blender 渲染一个 3D 场景、在 LibreOffice 里生成报表——这些软件只有 GUI，没有 API，Agent 就"没手"了。
港大 HKUDS 团队开源的 CLI-Anything 解决的正是这个问题：用自动化的方式，为任何软件生成一套 Agent 可直接调用的 CLI 接口。项目口号 "Making ALL Software Agent-Native"——今天的软件服务于人类👨💻，明天的用户将是 Agent 🤖。
面试高频问题："如果让你做一个企业级 Agent，你会如何设计架构？"——这道题考的不是你会不会写 LangChain 代码，而是你能不能区分 Demo 和 Production。一个能跑 Demo 的架构，放到生产环境可能处处漏水。本节从 Demo 级 Agent 的五大缺陷出发，逐模块构建一个能扛住真实业务压力的企业级架构。
Demo 级 vs 企业级 Agent
⚠️ 面试踩坑：很多候选人一上来就画 LangChain 的链式调用图——那是在选框架，不是在设计架构。面试官问的是"企业级"，关键词是可靠性、可观测性、成本控制。先说清楚你解决了什么生产问题，再说你用什么技术实现。
八大核心模块
企业级 Agent 不是把 Demo 代码部署到服务器就完了——它需要在八个维度上都有独立的模块负责。每个模块只做一件事，但做到极致。下面是架构全景图： 🏗️ 八大核心模块职责与接口 | 模块 | 职责 | 核心接口 | 生产兜底 | | --- | --- | --- | --- | | ① Gateway层 | 请求接入、负载均衡、限流熔断、认证鉴权 | handleRequest()、rateLimit()、authenticate() | 熔断降级→返回默认响应 | | ② Agent Engine层 | 意图识别、任务规划、执行调度、工具编排 | classifyIntent()、planTasks()、executeStep() | 意图未知→主动确认 | | ③ Memory层 | 短期记忆（对话上下文）、长期记忆（向量DB）、工作记忆（任务状态） | store()、retrieve()、compress()、evict() | 记忆丢失→从摘要重建 | | ④ Tool/Skill层 | 基础工具、组合流程、业务技能、MCP集成 | register()、match()、invoke()、fallback() | 工具挂→备选工具兜底 | | ⑤ LLM Router层 | 模型选型路由、成本优化、缓存命中、降级回退 | selectModel()、cacheQuery()、fallbackModel() | 主模型挂→备模型接管 | | ⑥ Context Engine层 | 上下文组装、压缩策略、缓存管理、token预算 | assemble()、compress()、budget()、cache() | 超预算→强制压缩 | | ⑦ Safety层 | 权限控制、内容过滤、敏感操作审批、审计日志 | checkPermission()、filterContent()、auditLog() | 越权→阻断并记录 | | ⑧ Observability层 | 运行日志、性能指标、成本追踪、异常告警 | log()、trackMetrics()、alert()、reportCost() | 旁路采集，不阻塞主流程 | ### 与第2章架构的关系
第2章（ch02）讲的是概念架构——Agent 的核心组件有哪些、各组件做什么、组件间如何协作。它回答的是"What"。
本章（ch14）讲的是生产实现架构——如何把这些概念组件变成可部署、可运维、可降级的工程系统。它回答的是"How"。
两者的关系：
• 第2章的"感知→决策→执行→记忆" → 本章拆分为 Gateway→Safety→Engine→Context→LLM→Tool→Memory
• 第2章的"LLM" → 本章细化为 LLM Router（带选型、缓存、降级）
• 第2章没有提到的 → 本章新增了 Observability（生产必须有观测）和 Context Engine（生产必须有上下文管理）
企业级 Agent 架构配置示例
# ===== 企业级 Agent 架构配置 =====
# 1. Gateway 层配置
gateway_config = {
"rate_limit": {
"max_requests_per_minute": 100,
"max_tokens_per_day": 500000
},
"auth": {
"type": "oauth2",
"scopes": ["agent:read", "agent:write", "agent:admin"]
},
"circuit_breaker": {
"failure_threshold": 5, # 连续5次失败→熔断
"recovery_timeout": 30, # 30秒后尝试恢复
"fallback_response": "服务暂时不可用，请稍后重试"
}
}
# 2. LLM Router 层配置
llm_router_config = {
"models": {
"complex": "gpt-4o", # 复杂推理：强模型
"simple": "gpt-4o-mini", # 简单对话：便宜模型
"fallback": "qwen-turbo" # 降级：国产模型兜底
},
"cache": {
"enabled": True,
"ttl": 3600, # 缓存1小时
"similarity_threshold": 0.95 # 相似度≥0.95命中缓存
},
"cost_budget": {
"daily_limit": 50.0, # 日预算$50
"alert_threshold": 0.8 # 超80%告警
}
}
# 3. Memory 层配置
memory_config = {
"short_term": {
"max_turns": 10, # 短期记忆保留10轮
"compression_threshold": 4000 # 超4000 tokens触发压缩
},
"long_term": {
"vector_db": "milvus",
"embedding_model": "text-embedding-3-small"
},
"working": {
"max_tasks": 5, # 同时追踪5个任务状态
"ttl": 86400 # 任务状态24小时过期
}
}
# 4. Safety 层配置
safety_config = {
"permissions": {
"admin": ["all"],
"operator": ["tool:search", "tool:weather", "tool:translate"],
"viewer": ["tool:search", "tool:weather"]
},
"content_filter": {
"input_filter": True, # 输入过滤（防注入）
"output_filter": True # 输出过滤（防泄露）
},
"audit": {
"log_level": "full", # 记录所有操作
"retention_days": 90 # 审计日志保留90天
},
"approval_required": ["tool:delete", "tool:payment", "tool:email_send"]
}
# 5. Observability 层配置
observability_config = {
"logging": {
"level": "INFO",
"format": "structured_json",
"destination": "elasticsearch"
},
"metrics": {
"tracer": "langfuse",
"prometheus": {
"track": ["latency", "token_usage", "task_success_rate", "cost"]
}
},
"alerts": {
"channels": ["feishu_webhook", "email"],
"rules": [
{"metric": "latency_p99", "threshold": 10, "unit": "seconds"},
{"metric": "error_rate", "threshold": 0.05},
{"metric": "daily_cost", "threshold": 40, "unit": "USD"}
]
}
}
# ===== 组装完整架构 =====
from agent_framework import EnterpriseAgent
agent = EnterpriseAgent(
gateway=gateway_config,
llm_router=llm_router_config,
memory=memory_config,
safety=safety_config,
observability=observability_config
)
# 运行时，每个请求经过完整链路：
# 用户请求 → Gateway(认证限流) → Safety(权限检查)
# → Engine(意图识别) → Context(prompt组装) → LLM Router(模型路由)
# → Tool(执行) → Memory(存入) → 输出 → Observability(旁路记录)
// ===== 企业级 Agent 架构配置 =====
// 1. Gateway 层配置
const gateway_config = {;
// "rate_limit": {
// "max_requests_per_minute": 100,
// "max_tokens_per_day": 500000
// },
// "auth": {
// "type": "oauth2",
// "scopes": ["agent:read", "agent:write", "agent:admin"]
// },
// "circuit_breaker": {
// "failure_threshold": 5, # 连续5次失败→熔断
// "recovery_timeout": 30, # 30秒后尝试恢复
// "fallback_response": "服务暂时不可用，请稍后重试"
// }
// }
// 2. LLM Router 层配置
const llm_router_config = {;
// "models": {
// "complex": "gpt-4o", # 复杂推理：强模型
// "simple": "gpt-4o-mini", # 简单对话：便宜模型
// "fallback": "qwen-turbo" # 降级：国产模型兜底
// },
// "cache": {
// "enabled": True,
// "ttl": 3600, # 缓存1小时
// "similarity_threshold": 0.95 # 相似度≥0.95命中缓存
// },
// "cost_budget": {
// "daily_limit": 50.0, # 日预算$50
// "alert_threshold": 0.8 # 超80%告警
// }
// }
// 3. Memory 层配置
const memory_config = {;
// "short_term": {
// "max_turns": 10, # 短期记忆保留10轮
// "compression_threshold": 4000 # 超4000 tokens触发压缩
// },
// "long_term": {
// "vector_db": "milvus",
// "embedding_model": "text-embedding-3-small"
// },
// "working": {
// "max_tasks": 5, # 同时追踪5个任务状态
// "ttl": 86400 # 任务状态24小时过期
// }
// }
// 4. Safety 层配置
const safety_config = {;
// "permissions": {
// "admin": ["all"],
// "operator": ["tool:search", "tool:weather", "tool:translate"],
// "viewer": ["tool:search", "tool:weather"]
// },
// "content_filter": {
// "input_filter": True, # 输入过滤（防注入）
// "output_filter": True # 输出过滤（防泄露）
// },
// "audit": {
// "log_level": "full", # 记录所有操作
// "retention_days": 90 # 审计日志保留90天
// },
// "approval_required": ["tool:delete", "tool:payment", "tool:email_send"]
// }
// 5. Observability 层配置
const observability_config = {;
// "logging": {
// "level": "INFO",
// "format": "structured_json",
// "destination": "elasticsearch"
// },
// "metrics": {
// "tracer": "langfuse",
// "prometheus": {
// "track": ["latency", "token_usage", "task_success_rate", "cost"]
// }
// },
// "alerts": {
// "channels": ["feishu_webhook", "email"],
// "rules": [
// {"metric": "latency_p99", "threshold": 10, "unit": "seconds"},
// {"metric": "error_rate", "threshold": 0.05},
// {"metric": "daily_cost", "threshold": 40, "unit": "USD"}
// ]
// }
// }
// ===== 组装完整架构 =====
import {EnterpriseAgent} from 'agent_framework';
const agent = EnterpriseAgent(;
const gateway = gateway_config,;
const llm_router = llm_router_config,;
const memory = memory_config,;
const safety = safety_config,;
const observability = observability_config;
// )
// 运行时，每个请求经过完整链路：
// 用户请求 → Gateway(认证限流) → Safety(权限检查)
// → Engine(意图识别) → Context(prompt组装) → LLM Router(模型路由)
// → Tool(执行) → Memory(存入) → 输出 → Observability(旁路记录)
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// ===== 企业级 Agent 架构配置 =====
// 1. Gateway 层配置
// Python: gateway_config = {
// Python: "rate_limit": {
// Python: "max_requests_per_minute": 100,
// Python: "max_tokens_per_day": 500000
// Python: },
// Python: "auth": {
// Python: "type": "oauth2",
// Python: "scopes": ["agent:read", "agent:write", "agent:admin"]
// Python: },
// Python: "circuit_breaker": {
// Python: "failure_threshold": 5, # 连续5次失败→熔断
// Python: "recovery_timeout": 30, # 30秒后尝试恢复
// Python: "fallback_response": "服务暂时不可用，请稍后重试"
// Python: }
// Python: }
// 2. LLM Router 层配置
// Python: llm_router_config = {
// Python: "models": {
// Python: "complex": "gpt-4o", # 复杂推理：强模型
// Python: "simple": "gpt-4o-mini", # 简单对话：便宜模型
// Python: "fallback": "qwen-turbo" # 降级：国产模型兜底
// Python: },
// Python: "cache": {
// Python: "enabled": True,
// Python: "ttl": 3600, # 缓存1小时
// Python: "similarity_threshold": 0.95 # 相似度≥0.95命中缓存
// Python: },
// Python: "cost_budget": {
// Python: "daily_limit": 50.0, # 日预算$50
// Python: "alert_threshold": 0.8 # 超80%告警
// Python: }
// Python: }
// 3. Memory 层配置
// Python: memory_config = {
// Python: "short_term": {
// Python: "max_turns": 10, # 短期记忆保留10轮
// Python: "compression_threshold": 4000 # 超4000 tokens触发压缩
// Python: },
// Python: "long_term": {
// Python: "vector_db": "milvus",
// Python: "embedding_model": "text-embedding-3-small"
// Python: },
// Python: "working": {
// Python: "max_tasks": 5, # 同时追踪5个任务状态
// Python: "ttl": 86400 # 任务状态24小时过期
// Python: }
// Python: }
// 4. Safety 层配置
// Python: safety_config = {
// Python: "permissions": {
// Python: "admin": ["all"],
// Python: "operator": ["tool:search", "tool:weather", "tool:translate"],
// Python: "viewer": ["tool:search", "tool:weather"]
// Python: },
// Python: "content_filter": {
// Python: "input_filter": True, # 输入过滤（防注入）
// Python: "output_filter": True # 输出过滤（防泄露）
// Python: },
// Python: "audit": {
// Python: "log_level": "full", # 记录所有操作
// Python: "retention_days": 90 # 审计日志保留90天
// Python: },
// Python: "approval_required": ["tool:delete", "tool:payment", "tool:email_send"]
// Python: }
// 5. Observability 层配置
// Python: observability_config = {
// Python: "logging": {
// Python: "level": "INFO",
// Python: "format": "structured_json",
// Python: "destination": "elasticsearch"
// Python: },
// Python: "metrics": {
// Python: "tracer": "langfuse",
// Python: "prometheus": {
// Python: "track": ["latency", "token_usage", "task_success_rate", "cost"]
// Python: }
// Python: },
// Python: "alerts": {
// Python: "channels": ["feishu_webhook", "email"],
// Python: "rules": [
// Python: {"metric": "latency_p99", "threshold": 10, "unit": "seconds"},
// Python: {"metric": "error_rate", "threshold": 0.05},
// Python: {"metric": "daily_cost", "threshold": 40, "unit": "USD"}
// Python: ]
// Python: }
// Python: }
// ===== 组装完整架构 =====
// from agent_framework import EnterpriseAgent
// Python: agent = EnterpriseAgent(
// Python: gateway=gateway_config,
// Python: llm_router=llm_router_config,
// Python: memory=memory_config,
// Python: safety=safety_config,
// Python: observability=observability_config
// Python: )
// 运行时，每个请求经过完整链路：
// 用户请求 → Gateway(认证限流) → Safety(权限检查)
// → Engine(意图识别) → Context(prompt组装) → LLM Router(模型路由)
// → Tool(执行) → Memory(存入) → 输出 → Observability(旁路记录)
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// ===== 企业级 Agent 架构配置 =====
// 1. Gateway 层配置
// Python: gateway_config = {
// Python: "rate_limit": {
// Python: "max_requests_per_minute": 100,
// Python: "max_tokens_per_day": 500000
// Python: },
// Python: "auth": {
// Python: "type": "oauth2",
// Python: "scopes": ["agent:read", "agent:write", "agent:admin"]
// Python: },
// Python: "circuit_breaker": {
// Python: "failure_threshold": 5, # 连续5次失败→熔断
// Python: "recovery_timeout": 30, # 30秒后尝试恢复
// Python: "fallback_response": "服务暂时不可用，请稍后重试"
// Python: }
// Python: }
// 2. LLM Router 层配置
// Python: llm_router_config = {
// Python: "models": {
// Python: "complex": "gpt-4o", # 复杂推理：强模型
// Python: "simple": "gpt-4o-mini", # 简单对话：便宜模型
// Python: "fallback": "qwen-turbo" # 降级：国产模型兜底
// Python: },
// Python: "cache": {
// Python: "enabled": True,
// Python: "ttl": 3600, # 缓存1小时
// Python: "similarity_threshold": 0.95 # 相似度≥0.95命中缓存
// Python: },
// Python: "cost_budget": {
// Python: "daily_limit": 50.0, # 日预算$50
// Python: "alert_threshold": 0.8 # 超80%告警
// Python: }
// Python: }
// 3. Memory 层配置
// Python: memory_config = {
// Python: "short_term": {
// Python: "max_turns": 10, # 短期记忆保留10轮
// Python: "compression_threshold": 4000 # 超4000 tokens触发压缩
// Python: },
// Python: "long_term": {
// Python: "vector_db": "milvus",
// Python: "embedding_model": "text-embedding-3-small"
// Python: },
// Python: "working": {
// Python: "max_tasks": 5, # 同时追踪5个任务状态
// Python: "ttl": 86400 # 任务状态24小时过期
// Python: }
// Python: }
// 4. Safety 层配置
// Python: safety_config = {
// Python: "permissions": {
// Python: "admin": ["all"],
// Python: "operator": ["tool:search", "tool:weather", "tool:translate"],
// Python: "viewer": ["tool:search", "tool:weather"]
// Python: },
// Python: "content_filter": {
// Python: "input_filter": True, # 输入过滤（防注入）
// Python: "output_filter": True # 输出过滤（防泄露）
// Python: },
// Python: "audit": {
// Python: "log_level": "full", # 记录所有操作
// Python: "retention_days": 90 # 审计日志保留90天
// Python: },
// Python: "approval_required": ["tool:delete", "tool:payment", "tool:email_send"]
// Python: }
// 5. Observability 层配置
// Python: observability_config = {
// Python: "logging": {
// Python: "level": "INFO",
// Python: "format": "structured_json",
// Python: "destination": "elasticsearch"
// Python: },
// Python: "metrics": {
// Python: "tracer": "langfuse",
// Python: "prometheus": {
// Python: "track": ["latency", "token_usage", "task_success_rate", "cost"]
// Python: }
// Python: },
// Python: "alerts": {
// Python: "channels": ["feishu_webhook", "email"],
// Python: "rules": [
// Python: {"metric": "latency_p99", "threshold": 10, "unit": "seconds"},
// Python: {"metric": "error_rate", "threshold": 0.05},
// Python: {"metric": "daily_cost", "threshold": 40, "unit": "USD"}
// Python: ]
// Python: }
// Python: }
// ===== 组装完整架构 =====
// from agent_framework import EnterpriseAgent
// Python: agent = EnterpriseAgent(
// Python: gateway=gateway_config,
// Python: llm_router=llm_router_config,
// Python: memory=memory_config,
// Python: safety=safety_config,
// Python: observability=observability_config
// Python: )
// 运行时，每个请求经过完整链路：
// 用户请求 → Gateway(认证限流) → Safety(权限检查)
// → Engine(意图识别) → Context(prompt组装) → LLM Router(模型路由)
// → Tool(执行) → Memory(存入) → 输出 → Observability(旁路记录)
}
18.8 核心模块拆分与职责边界
面试高频问题："一个 Agent 系统应该拆分成哪些核心模块？每个模块分别负责什么？"——模块拆分不是"拆得越细越好"，而是拆得越清晰越好。每个模块的职责边界必须明确——只负责A，不负责B。如果边界模糊，模块间就会互相耦合，改一个模块牵连三个模块。
模块拆分三原则
📐 三大拆分原则
单一职责 每个模块只做一件事。Planner只负责规划，不负责执行；Memory只负责存取，不负责推理。就像公司里的岗位——HR不写代码，开发不做招聘。
接口隔离 模块间通过接口通信，不直接依赖内部实现。Planner不关心Executor用什么框架执行——它只关心Executor返回的result格式。换框架不影响Planner。
可替换性 任何模块都可以替换实现。LLM从GPT换到Claude不影响其他模块；Memory从Redis换到Milvus不影响Engine。关键：接口稳定，实现可变。
⚠️ 反面教材：把所有功能塞在一个"超级Agent"类里——意图识别、工具调用、记忆管理、安全校验都在一个3000行的类里。改个工具调用逻辑，可能影响意图识别的缓存。这种代码在Demo里能跑，在生产里是定时炸弹。
六大核心模块详解
职责边界卡片
🎯 Planner（规划器）
只负责：接收任务→生成执行计划→返回子任务序列
不负责：执行子任务、调用工具、存储结果
接口：plan(task: str) → List[SubTask]
降级：无法规划→返回单步执行（退化为ReAct） ⚡ Executor（执行器）
只负责：按计划执行→调用工具/模型→返回结果
不负责：规划下一步、决定用什么模型、管理记忆
接口：execute(plan: List[SubTask]) → ExecutionResult
降级：执行失败→重试1次→备选工具→报告失败 🧠 Memory（记忆器）
只负责：存储→索引→检索→压缩→淘汰
不负责：决定什么时候检索、检索结果怎么用
接口：store(key, value)、retrieve(query) → List[MemoryItem]
降级：向量DB挂→从Redis缓存召回；Redis挂→空记忆继续执行 🔧 ToolRouter（工具路由）
只负责：注册→匹配→调用→超时→降级
不负责：决定调用哪个工具（这是Executor的事）
接口：register(tool)、match(intent) → Tool、invoke(tool, params) → Result
降级：主工具超时→3秒后切备选工具→备选也挂→返回"无法执行" 🤖 LLMRouter（模型路由）
只负责：选型→路由→缓存→降级→计费
不负责：组装prompt（Context Engine的事）、解析LLM输出
接口：selectModel(complexity) → ModelName、invoke(model, prompt) → Response
降级：主模型挂→切备模型→切国产模型→切本地模型→返回默认响应
🔒 SafetyGuard（安全守卫）
只负责：权限检查→内容过滤→审计记录
不负责：执行业务逻辑、决定是否允许（只检查规则）
接口：checkPermission(user, action) → bool、filterContent(text) → SafeText
降级：安全服务挂→全阻断（宁可不可用，不可不安全）
模块间通信机制
📡 三种通信机制对比 | 机制 | 原理 | 优点 | 缺点 | 适用 | | --- | --- | --- | --- | --- | | 消息队列 | 模块间通过异步消息传递，发布者→队列→订阅者 | 解耦彻底、可削峰、可重试 | 延迟高、调试难、需消息顺序保证 | 大规模 | | 共享状态 | 模块间通过共享State对象通信，读写同一份数据 | 实时、简单、LangGraph原生方式 | 耦合度高、并发难、需状态锁 | 单Agent | | 事件驱动 | 模块发布事件，其他模块按需订阅，松耦合 | 灵活、可扩展、支持多消费者 | 事件顺序不保证、可能事件风暴 | 多Agent | 实际选择：大多数 Agent 系统用混合模式——核心链路（意图→规划→执行）用共享状态（LangGraph的StateGraph），旁路模块（Observability、Safety）用事件驱动（异步采集不影响主流程），跨Agent协作用消息队列（Kafka/RabbitMQ）。
模块接口定义示例
# ===== 六大核心模块接口定义 =====
from typing import Protocol, List, Optional
from dataclasses import dataclass
# ----- 数据结构定义 -----
@dataclass
class SubTask:
task_id: str
description: str
tool_name: Optional[str]
params: dict
depends_on: List[str] = [] # 依赖哪些子任务
@dataclass
class ExecutionResult:
task_id: str
success: bool
output: str
tool_used: Optional[str]
latency_ms: int
@dataclass
class MemoryItem:
key: str
value: str
timestamp: float
relevance_score: float
@dataclass
class SafeText:
text: str
is_safe: bool
filtered_reason: Optional[str]
# ----- 模块接口定义 -----
class Planner(Protocol):
"""规划器：只负责生成执行计划"""
def plan(self, task: str, context: dict) -> List[SubTask]:
"""接收任务，返回子任务序列"""
...
class Executor(Protocol):
"""执行器：只负责按计划执行"""
def execute(self, plan: List[SubTask]) -> List[ExecutionResult]:
"""按计划执行，返回每个子任务的结果"""
...
class Memory(Protocol):
"""记忆器：只负责存储和检索"""
def store(self, key: str, value: str, ttl: Optional[int] = None) -> None:
"""存储记忆"""
...
def retrieve(self, query: str, top_k: int = 5) -> List[MemoryItem]:
"""检索相关记忆"""
...
def compress(self) -> str:
"""压缩长期记忆"""
...
class ToolRouter(Protocol):
"""工具路由：只负责匹配和调用工具"""
def register(self, name: str, tool: callable, fallback: Optional[callable] = None) -> None:
"""注册工具（含备选）"""
...
def match(self, intent: str) -> Optional[str]:
"""根据意图匹配工具名"""
...
def invoke(self, name: str, params: dict, timeout: int = 5) -> str:
"""调用工具，超时自动切备选"""
...
class LLMRouter(Protocol):
"""模型路由：只负责选型和调用模型"""
def select_model(self, complexity: str) -> str:
"""根据复杂度选模型"""
...
def invoke(self, model: str, prompt: str) -> str:
"""调用模型，失败自动降级"""
...
class SafetyGuard(Protocol):
"""安全守卫：只负责检查和过滤"""
def check_permission(self, user: str, action: str) -> bool:
"""检查用户是否有权限执行该操作"""
...
def filter_content(self, text: str) -> SafeText:
"""过滤敏感内容"""
...
def audit_log(self, user: str, action: str, result: str) -> None:
"""记录审计日志"""
...
# ===== 关键设计点 =====
# 1. 每个Protocol只定义自己职责范围内的方法
# 2. 模块间不互相引用具体实现类，只依赖Protocol
# 3. 任何模块可以替换实现（如Memory从Redis→Milvus，只要实现Protocol即可）
# 4. 接口设计遵循"只进不出"原则：方法返回结果，不修改共享状态
// ===== 六大核心模块接口定义 =====
// TypeScript has built-in types, no import needed for Protocol, List, Optional
import {dataclass} from 'dataclasses';
// ----- 数据结构定义 -----
// @dataclass
class SubTask {
// task_id: str
// description: str
// tool_name: Optional[str]
// params: dict
// depends_on: List[str] = [] # 依赖哪些子任务
// @dataclass
class ExecutionResult {
// task_id: str
// success: bool
// output: str
// tool_used: Optional[str]
// latency_ms: int
// @dataclass
class MemoryItem {
// key: str
// value: str
// timestamp: float
// relevance_score: float
// @dataclass
class SafeText {
// text: str
// is_safe: bool
// filtered_reason: Optional[str]
// ----- 模块接口定义 -----
class Planner {
/** docstring */
// def plan(self, task: str, context: dict) -> List[SubTask]:
/** docstring */
// ...
class Executor {
/** docstring */
// def execute(self, plan: List[SubTask]) -> List[ExecutionResult]:
/** docstring */
// ...
class Memory {
/** docstring */
// def store(self, key: str, value: str, ttl: Optional[int] = None) -> None:
/** docstring */
// ...
// def retrieve(self, query: str, top_k: int = 5) -> List[MemoryItem]:
/** docstring */
// ...
// def compress(self) -> str:
/** docstring */
// ...
class ToolRouter {
/** docstring */
// def register(self, name: str, tool: callable, fallback: Optional[callable] = None) -> None:
/** docstring */
// ...
// def match(self, intent: str) -> Optional[str]:
/** docstring */
// ...
// def invoke(self, name: str, params: dict, timeout: int = 5) -> str:
/** docstring */
// ...
class LLMRouter {
/** docstring */
// def select_model(self, complexity: str) -> str:
/** docstring */
// ...
// def invoke(self, model: str, prompt: str) -> str:
/** docstring */
// ...
class SafetyGuard {
/** docstring */
// def check_permission(self, user: str, action: str) -> bool:
/** docstring */
// ...
// def filter_content(self, text: str) -> SafeText:
/** docstring */
// ...
// def audit_log(self, user: str, action: str, result: str) -> None:
/** docstring */
// ...
// ===== 关键设计点 =====
// 1. 每个Protocol只定义自己职责范围内的方法
// 2. 模块间不互相引用具体实现类，只依赖Protocol
// 3. 任何模块可以替换实现（如Memory从Redis→Milvus，只要实现Protocol即可）
// 4. 接口设计遵循"只进不出"原则：方法返回结果，不修改共享状态
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// ===== 六大核心模块接口定义 =====
// from typing import Protocol, List, Optional
// from dataclasses import dataclass
// ----- 数据结构定义 -----
// SubTask - CLI Agent class
type SubTask struct {
// Python: task_id: str
// Python: description: str
// Python: tool_name: Optional[str]
// Python: params: dict
// Python: depends_on: List[str] = [] # 依赖哪些子任务
// ExecutionResult - CLI Agent class
type ExecutionResult struct {
// Python: task_id: str
// Python: success: bool
// Python: output: str
// Python: tool_used: Optional[str]
// Python: latency_ms: int
// MemoryItem - CLI Agent class
type MemoryItem struct {
// Python: key: str
// Python: value: str
// Python: timestamp: float
// Python: relevance_score: float
// SafeText - CLI Agent class
type SafeText struct {
// Python: text: str
// Python: is_safe: bool
// Python: filtered_reason: Optional[str]
// ----- 模块接口定义 -----
// Planner - CLI Agent class
type Planner struct {
// Python: def plan(self, task: str, context: dict) -> List[SubTask]:
// Python: ...
// Executor - CLI Agent class
type Executor struct {
// Python: def execute(self, plan: List[SubTask]) -> List[ExecutionResult]:
// Python: ...
// Memory - CLI Agent class
type Memory struct {
// Python: def store(self, key: str, value: str, ttl: Optional[int] = None) -> None:
// Python: ...
// Python: def retrieve(self, query: str, top_k: int = 5) -> List[MemoryItem]:
// Python: ...
// Python: def compress(self) -> str:
// Python: ...
// ToolRouter - CLI Agent class
type ToolRouter struct {
// Python: def register(self, name: str, tool: callable, fallback: Optional[callable] = None) -> None:
// Python: ...
// Python: def match(self, intent: str) -> Optional[str]:
// Python: ...
// Python: def invoke(self, name: str, params: dict, timeout: int = 5) -> str:
// Python: ...
// LLMRouter - CLI Agent class
type LLMRouter struct {
// Python: def select_model(self, complexity: str) -> str:
// Python: ...
// Python: def invoke(self, model: str, prompt: str) -> str:
// Python: ...
// SafetyGuard - CLI Agent class
type SafetyGuard struct {
// Python: def check_permission(self, user: str, action: str) -> bool:
// Python: ...
// Python: def filter_content(self, text: str) -> SafeText:
// Python: ...
// Python: def audit_log(self, user: str, action: str, result: str) -> None:
// Python: ...
// ===== 关键设计点 =====
// 1. 每个Protocol只定义自己职责范围内的方法
// 2. 模块间不互相引用具体实现类，只依赖Protocol
// 3. 任何模块可以替换实现（如Memory从Redis→Milvus，只要实现Protocol即可）
// 4. 接口设计遵循"只进不出"原则：方法返回结果，不修改共享状态
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// ===== 六大核心模块接口定义 =====
// from typing import Protocol, List, Optional
// from dataclasses import dataclass
// ----- 数据结构定义 -----
public class SubTask {
// Python: task_id: str
// Python: description: str
// Python: tool_name: Optional[str]
// Python: params: dict
// Python: depends_on: List[str] = [] # 依赖哪些子任务
public class ExecutionResult {
// Python: task_id: str
// Python: success: bool
// Python: output: str
// Python: tool_used: Optional[str]
// Python: latency_ms: int
public class MemoryItem {
// Python: key: str
// Python: value: str
// Python: timestamp: float
// Python: relevance_score: float
public class SafeText {
// Python: text: str
// Python: is_safe: bool
// Python: filtered_reason: Optional[str]
// ----- 模块接口定义 -----
public class Planner {
// Python: def plan(self, task: str, context: dict) -> List[SubTask]:
// Python: ...
public class Executor {
// Python: def execute(self, plan: List[SubTask]) -> List[ExecutionResult]:
// Python: ...
public class Memory {
// Python: def store(self, key: str, value: str, ttl: Optional[int] = None) -> None:
// Python: ...
// Python: def retrieve(self, query: str, top_k: int = 5) -> List[MemoryItem]:
// Python: ...
// Python: def compress(self) -> str:
// Python: ...
public class ToolRouter {
// Python: def register(self, name: str, tool: callable, fallback: Optional[callable] = None) -> None:
// Python: ...
// Python: def match(self, intent: str) -> Optional[str]:
// Python: ...
// Python: def invoke(self, name: str, params: dict, timeout: int = 5) -> str:
// Python: ...
public class LLMRouter {
// Python: def select_model(self, complexity: str) -> str:
// Python: ...
// Python: def invoke(self, model: str, prompt: str) -> str:
// Python: ...
public class SafetyGuard {
// Python: def check_permission(self, user: str, action: str) -> bool:
// Python: ...
// Python: def filter_content(self, text: str) -> SafeText:
// Python: ...
// Python: def audit_log(self, user: str, action: str, result: str) -> None:
// Python: ...
// ===== 关键设计点 =====
// 1. 每个Protocol只定义自己职责范围内的方法
// 2. 模块间不互相引用具体实现类，只依赖Protocol
// 3. 任何模块可以替换实现（如Memory从Redis→Milvus，只要实现Protocol即可）
// 4. 接口设计遵循"只进不出"原则：方法返回结果，不修改共享状态
}
}
← 第17章 Dify/Coze 第19章 GUI Agent →
第18章 CLI-Agent-命令行智能助手
http://www.clxhxhhr.top/posts/721/ 评论
0 条
还没有评论，先写一条吧。