第13章 CLI能力-Agent操作本地工具
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch09-cli-capability.html 所属：第三篇-Agent的手脚
让 Agent 操控 Git、Docker、Maven……一切 CLI 工具皆为 Agent 的手脚
13.1 为什么 CLI 能力是 Agent 的关键突破
前两章我们讲了 MCP 和 Skills——它们让 Agent 能调用远程 API、搜索文档、操作数据库。但程序员日常工作中最频繁使用的工具是什么？是命令行。git commit、mvn clean package、docker compose up、kubectl apply……这些 CLI 工具构成了开发者的核心操作链路。
如果 Agent 只能调 HTTP API，它永远是个"远程助手"；但如果 Agent 能操作本地命令行工具，它就真正成了你电脑上的智能搭档——帮你构建项目、部署服务、排查故障、管理代码仓库，而你只需要用自然语言说一句"帮我打包部署到测试环境"。 💡 CLI 能力 = Agent 从"遥控器"升级为"机械手"
MCP 和 HTTP API 像遥控器——Agent 在远端按按钮，服务在云端响应。CLI 能力则像机械手——Agent 直接握住你本地的工具，git、docker、maven、npm、python……每一款 CLI 工具都是 Agent 可以操控的"手指"。这是 Agent 从"线上助手"到"本地搭档"的关键跃迁。
Agent 能操控的 CLI 工具全景
几乎所有开发者日常使用的工具都有 CLI 接口，Agent 可以像人一样"敲命令"来操控它们： | 类别 | CLI 工具 | Agent 可以做什么 | | --- | --- | --- | | 📦 构建 | mvn、gradle、npm、pip、go build | 打包、编译、依赖管理、版本发布 | | 🔀 版本控制 | git、svn | 提交、分支管理、合并冲突、Changelog 生成 | | 🐳 容器 | docker、kubectl、helm | 镜像构建、容器部署、集群管理、滚动更新 | | 🔧 运维 | ssh、scp、systemctl、journalctl | 远程部署、日志排查、服务启停、配置修改 | | 📊 数据 | mysql、redis-cli、psql、mongo | 数据库操作、缓存管理、数据迁移 | | 🧪 测试 | pytest、junit、curl、ab | 跑测试、压测、接口调试 | | 📝 文档 | swagger、doxygen、javadoc | API 文档生成、代码文档提取 | | 🌐 网络 | ping、netstat、nslookup、iptables | 网络诊断、端口检查、DNS 查询 | ## 13.2 CLI 能力的三层架构
Agent 操作 CLI 工具，不是简单地把用户的话翻译成命令然后执行。它需要一个完整的三层架构：意图理解层、命令生成层、安全执行层。每一层都有独特的技术挑战。
Layer 1：意图理解层
用户说"帮我部署到测试环境"，Agent 需要理解这背后的完整意图链： 🧠 意图理解：从一句话到多步执行计划
用户输入："帮我部署到测试环境"
Agent 意图理解链：
- 环境判断 → 测试环境 = test profile
- 前置步骤 → 先打包（
mvn clean package -Dmaven.test.skip=true） - 构建镜像 →
docker build -t system/app:1.0.0 . - 推送镜像 →
docker push registry/app:1.0.0 - 部署服务 →
docker compose -f docker-compose-app.yml up -d - 验证结果 →
docker ps | grep app+curl health
意图理解层的关键能力：
- 上下文感知：知道当前目录是什么项目、用的什么语言、有什么配置文件
- 环境推断：根据项目结构推断构建工具（有
pom.xml→ Maven，有package.json→ npm） - 多步规划：一个意图可能需要 3~6 步命令才能完成，Agent 要自动规划
Layer 2：命令生成层（NL2Shell）
将自然语言意图翻译成精确的 Shell 命令，是 CLI 能力的核心技术。这个过程叫 NL2Shell（Natural Language to Shell）。
🔬 NL2Shell 的三种实现路径 | 路径 | 原理 | 准确率 | 适用场景 | | --- | --- | --- | --- | | 直接生成 | LLM 直接输出 Shell 命令 | 70~85% | 简单命令 | | Few-Shot 模板 | LLM + 常用命令示例库 | 85~92% | 标准场景 | | 工具 Schema | Function Calling + 命令 Schema 定义 | 92~97% | 企业级 | 推荐使用工具 Schema 方式——把每个 CLI 工具定义成 Function Calling 的 Schema，LLM 不需要"猜"命令，而是像调 API一样选择工具并填参数。这和我们第10章讲的 Function Calling 完美衔接。
// CLI 工具 Schema 定义示例：git { "name": "git_operation", "description": "执行 Git 版本控制操作", "parameters": { "type": "object", "properties": { "action": { "type": "string", "enum": ["commit", "push", "pull", "branch", "log", "diff", "merge"], "description": "Git 操作类型" }, "message": { "type": "string", "description": "提交消息（仅 commit 需要）" }, "branch_name": { "type": "string", "description": "分支名（branch/merge 需要）" }, "count": { "type": "integer", "description": "查看最近 N 条记录（log 需要）" } }, "required": ["action"] } }
Layer 3：安全执行层
CLI 命令有系统级权限——可以删除文件、重启服务、修改配置。安全执行层是 CLI 能力最不可省略的部分。
🔒 三级安全策略 ✅ Level 1 · 白名单命令 — 自动执行
git status、ls、cat、docker ps……只读、无破坏性、无系统修改的命令，Agent 可直接执行。
⚠️ Level 2 · 中风险命令 — 确认后执行
git push、mvn deploy、docker restart……有修改但可控，需用户确认 "Y/N" 后执行。
❌ Level 3 · 高风险命令 — 拒绝或需二次审批
rm -rf、DROP DATABASE、shutdown、mkfs……不可逆操作，默认拒绝，特殊场景需管理员二次审批。
13.3 CLI 工具的 Schema 注册
要让 Agent 精准操控 CLI 工具，每个工具需要定义Schema——告诉 LLM 这个工具能做什么、有哪些参数、参数的类型和约束。这是连接"第10章 Function Calling"和"第11章 MCP"的桥梁。
五大核心工具的 Schema 设计
以下是开发者最常用的五大 CLI 工具的 Schema 注册示例��
🔀 Git — 版本控制工具
Git 是开发者最高频使用的 CLI 工具。Agent 可以帮你提交代码、管理分支、查看历史、解决冲突。
// git_schema.json { "tool_name": "git", "description": "Git 版本控制系统", "commands": [ { "name": "git_status", "risk_level": "safe", // 只读 → Level 1 "description": "查看工作区状态", "shell": "git status" }, { "name": "git_commit", "risk_level": "medium", // 写操作 → Level 2 "description": "提交代码到本地仓库", "parameters": { "message": { "type": "string", "required": true } }, "shell_template": "git commit -m '{message}'" }, { "name": "git_push", "risk_level": "medium", "description": "推送代码到远程仓库", "shell": "git push origin {branch}", "requires_confirm": true } ] } 📦 Maven — 项目构建工具
Maven 是 Java 项目的标准构建工具。Agent 可以帮你编译、打包、运行测试、管理依赖。
// maven_schema.json { "tool_name": "maven", "description": "Maven 项目构建与依赖管理", "commands": [ { "name": "maven_clean_package", "risk_level": "safe", "description": "清理并打包项目（跳过测试）", "shell": "mvn clean package -Dmaven.test.skip=true" }, { "name": "maven_deploy", "risk_level": "medium", "description": "部署到远程仓库", "shell": "mvn deploy -Dmaven.test.skip=true", "requires_confirm": true }, { "name": "maven_dependency_tree", "risk_level": "safe", "description": "查看依赖树", "shell": "mvn dependency:tree" } ] }
🐳 Docker — 容器化工具
Docker 是部署的核心工具。Agent 可以帮你构建镜像、启停容器、查看日志、管理网络。
// docker_schema.json { "tool_name": "docker", "commands": [ { "name": "docker_build", "risk_level": "safe", "shell_template": "docker build -t {image_name}:{tag} {path}" }, { "name": "docker_compose_up", "risk_level": "medium", "shell": "docker compose -f {file} up -d", "requires_confirm": true }, { "name": "docker_stop_remove", "risk_level": "high", // 停删容器 → Level 3 "shell_template": "docker stop {container} && docker rm {container}", "requires_double_confirm": true } ] }
13.4 CLI 工具的统一调度器
Agent 同时掌握 git、maven、docker、kubectl 等多个 CLI 工具，需要一个统一调度器来：
- 路由：根据用户意图，选择正确的 CLI 工具
- 编排：多步命令按顺序执行，前一步的输出作为后一步的输入
- 容错：命令执行失败时自动重试或切换策略
调度器与 MCP 的关系
CLI 调度器和 MCP 服务器可以是同一套架构。每个 CLI 工具的 Schema 可以直接注册为 MCP Server 的 Tool，让 MCP 协议统一管理调度：
🔗 CLI Schema → MCP Tool 的映射 | CLI Schema | MCP Tool 定义 | 执行方式 | | --- | --- | --- | | git_commit | MCP Tool: git_commit(message) | 子进程执行 git commit -m "{msg}" | | maven_clean_package | MCP Tool: maven_build(skip_test) | 子进程执行 mvn clean package | | docker_compose_up | MCP Tool: docker_deploy(file, env) | 子进程执行 docker compose up -d | 这样，CLI 能力就和第11章 MCP、第12章 Skills 形成完整的工具链闭环：
🛠️ 第二篇完整工具链闭环
Function Calling → MCP 协议 → Skills 组合 → CLI 执行
Function Calling 定义调用格式 → MCP 统一通信协议 → Skills 组合多步流程 → CLI 在本地沙箱执行
13.5 实战：构建一个 CLI Agent
我们将构建一个最小但完整的 CLI Agent，它能理解自然语言、生成命令、安全执行并返回结果。
这个实战把前面所有概念串成一条可运行的链路：意图理解层解析用户想做什么，命令生成层（NL2Shell）把意图翻译成具体命令，安全执行层在沙箱中运行并捕获结果。三层各司其职又环环相扣——理解错了会生成错误的命令，生成对了但执行环境不安全则可能危及系统。我们刻意保持架构最小，让你看清 CLI Agent 的骨架，而不是被框架细节淹没。
架构代码
"""cli_agent.py — 最小可用的 CLI Agent"""
class CLIAgent:
def __init__(self, llm_client, tool_schemas):
self.llm = llm_client # LLM 客户端
self.schemas = tool_schemas # CLI 工具 Schema 注册表
self.sandbox = CommandSandbox() # 安全执行沙箱
self.risk_checker = RiskChecker() # 风险等级校验
def execute(self, user_input: str, context: dict):
# Step 1: 意图理解 + 命令生成（Function Calling）
response = self.llm.chat(
messages=[
{"role": "system", "content": self._build_system_prompt()},
{"role": "user", "content": user_input}
],
tools=self.schemas, # 注入 CLI Schema 作为 Function Calling 工具
context=context # 携带当前目录、项目信息等上下文
)
# Step 2: 解析 LLM 返回的工具调用
tool_calls = response.tool_calls
if not tool_calls:
return response.content # LLM 直接回答，无需执行命令
results = []
for call in tool_calls:
# Step 3: 风险校验
risk = self.risk_checker.check(call.name, call.arguments)
if risk == "blocked":
results.append({"error": "该操作被安全策略阻止"})
continue
if risk == "needs_confirm":
if not self._ask_user_confirm(call):
results.append({"error": "用户取消操作"})
continue
# Step 4: 沙箱执行
shell_cmd = self._render_shell(call) # Schema → Shell 命令
output = self.sandbox.run(shell_cmd)
results.append(output)
# Step 5: 结果格式化返回
return self.llm.summarize(results)
/** cli_agent.ts — 最小可用的 CLI Agent */
import { OpenAI } from 'openai';
import { CommandSandbox } from './sandbox';
import { RiskChecker } from './risk-checker';
interface ToolCall {
name: string;
arguments: Record;
}
interface LLMResponse {
content: string;
tool_calls?: ToolCall[];
}
class CLIAgent {
private llm: OpenAI;
private schemas: any[];
private sandbox: CommandSandbox;
private riskChecker: RiskChecker;
constructor(llmClient: OpenAI, toolSchemas: any[]) {
this.llm = llmClient; // LLM 客户端
this.schemas = toolSchemas; // CLI 工具 Schema 注册表
this.sandbox = new CommandSandbox(); // 安全执行沙箱
this.riskChecker = new RiskChecker(); // 风险等级校验
}
async execute(userInput: string, context: Record): Promise {
// Step 1: 意图理解 + 命令生成（Function Calling）
const response = await this.llm.chat.completions.create({
model: 'gpt-4o',
messages: [
{ role: 'system', content: this.buildSystemPrompt() },
{ role: 'user', content: userInput }
],
tools: this.schemas, // 注入 CLI Schema 作为 Function Calling 工具
});
const message = response.choices[0].message;
// Step 2: 解析 LLM 返回的工具调用
const toolCalls = message.tool_calls || [];
if (toolCalls.length === 0) {
return message.content || ''; // LLM 直接回答，无需执行命令
}
const results: any[] = [];
for (const call of toolCalls) {
const fnName = call.function.name;
const fnArgs = JSON.parse(call.function.arguments);
// Step 3: 风险校验
const risk = this.riskChecker.check(fnName, fnArgs);
if (risk === 'blocked') {
results.push({ error: '该操作被安全策略阻止' });
continue;
}
if (risk === 'needs_confirm') {
if (!this.askUserConfirm(call)) {
results.push({ error: '用户取消操作' });
continue;
}
}
// Step 4: 沙箱执行
const shellCmd = this.renderShell(fnName, fnArgs); // Schema → Shell 命令
const output = this.sandbox.run(shellCmd);
results.push(output);
}
// Step 5: 结果格式化返回
return this.llm.chat.completions.create({
model: 'gpt-4o',
messages: [
{ role: 'system', content: '请将以下命令执行结果格式化为用户友好的回复。' },
{ role: 'user', content: JSON.stringify(results) }
]
}).then(r => r.choices[0].message.content || '');
}
private buildSystemPrompt(): string { return ''; }
private renderShell(name: string, args: any): string { return ''; }
private askUserConfirm(call: any): boolean { return false; }
}
package main
// cli_agent.go — 最小可用的 CLI Agent
import (
"context"
"encoding/json"
"fmt"
openai "github.com/sashabaranov/go-openai"
)
type CLIAgent struct {
LLM *openai.Client
Schemas []interface{}
Sandbox *CommandSandbox
RiskChecker *RiskChecker
}
func NewCLIAgent(llmClient *openai.Client, toolSchemas []interface{}) *CLIAgent {
return &CLIAgent{
LLM: llmClient, // LLM 客户端
Schemas: toolSchemas, // CLI 工具 Schema 注册表
Sandbox: NewCommandSandbox(), // 安全执行沙箱
RiskChecker: NewRiskChecker(), // 风险等级校验
}
}
func (a *CLIAgent) Execute(ctx context.Context, userInput string, contextData map[string]interface{}) (string, error) {
// Step 1: 意图理解 + 命令生成（Function Calling）
resp, err := a.LLM.CreateChatCompletion(ctx, openai.ChatCompletionRequest{
Model: openai.GPT4o,
Messages: []openai.ChatCompletionMessage{
{Role: openai.ChatMessageRoleSystem, Content: a.buildSystemPrompt()},
{Role: openai.ChatMessageRoleUser, Content: userInput},
},
Tools: a.schemasToTools(), // 注入 CLI Schema 作为 Function Calling 工具
})
if err != nil {
return "", err
}
msg := resp.Choices[0].Message
// Step 2: 解析 LLM 返回的工具调用
if len(msg.ToolCalls) == 0 {
return msg.Content, nil // LLM 直接回答，无需执行命令
}
var results []map[string]interface{}
for _, call := range msg.ToolCalls {
var args map[string]interface{}
json.Unmarshal([]byte(call.Function.Arguments), &args)
// Step 3: 风险校验
risk := a.RiskChecker.Check(call.Function.Name, args)
if risk == "blocked" {
results = append(results, map[string]interface{}{"error": "该操作被安全策略阻止"})
continue
}
if risk == "needs_confirm" {
if !a.askUserConfirm(call) {
results = append(results, map[string]interface{}{"error": "用户取消操作"})
continue
}
}
// Step 4: 沙箱执行
shellCmd := a.renderShell(call.Function.Name, args) // Schema → Shell 命令
output := a.Sandbox.Run(shellCmd)
results = append(results, output)
}
// Step 5: 结果格式化返回
resultsJSON, _ := json.Marshal(results)
summaryResp, err := a.LLM.CreateChatCompletion(ctx, openai.ChatCompletionRequest{
Model: openai.GPT4o,
Messages: []openai.ChatCompletionMessage{
{Role: openai.ChatMessageRoleSystem, Content: "请将以下命令执行结果格式化为用户友好的回复。"},
{Role: openai.ChatMessageRoleUser, Content: string(resultsJSON)},
},
})
if err != nil {
return fmt.Sprintf("%v", results), nil
}
return summaryResp.Choices[0].Message.Content, nil
}
import com.openai.client.OpenAIClient;
import com.openai.models.*;
import java.util.*;
/** cli_agent.java — 最小可用的 CLI Agent */
public class CLIAgent {
private OpenAIClient llm; // LLM 客户端
private List schemas; // CLI 工具 Schema 注册表
private CommandSandbox sandbox; // 安全执行沙箱
private RiskChecker riskChecker; // 风险等级校验
public CLIAgent(OpenAIClient llmClient, List toolSchemas) {
this.llm = llmClient;
this.schemas = toolSchemas;
this.sandbox = new CommandSandbox();
this.riskChecker = new RiskChecker();
}
public String execute(String userInput, Map context) {
// Step 1: 意图理解 + 命令生成（Function Calling）
ChatCompletionCreateParams params = ChatCompletionCreateParams.builder()
.model("gpt-4o")
.addSystemMessage(buildSystemPrompt())
.addUserMessage(userInput)
.tools(schemas) // 注入 CLI Schema 作为 Function Calling 工具
.build();
ChatCompletion response = llm.chat().completions().create(params);
ChatCompletionMessage msg = response.choices().get(0).message();
// Step 2: 解析 LLM 返回的工具调用
List toolCalls = msg.toolCalls();
if (toolCalls == null || toolCalls.isEmpty()) {
return msg.content(); // LLM 直接回答，无需执行命令
}
List> results = new ArrayList<>();
for (ToolCall call : toolCalls) {
String fnName = call.function().name();
Map fnArgs = parseArgs(call.function().arguments());
// Step 3: 风险校验
String risk = riskChecker.check(fnName, fnArgs);
if ("blocked".equals(risk)) {
results.add(Map.of("error", "该操作被安全策略阻止"));
continue;
}
if ("needs_confirm".equals(risk)) {
if (!askUserConfirm(call)) {
results.add(Map.of("error", "用户取消操作"));
continue;
}
}
// Step 4: 沙箱执行
String shellCmd = renderShell(fnName, fnArgs); // Schema → Shell 命令
Map output = sandbox.run(shellCmd);
results.add(output);
}
// Step 5: 结果格式化返回
return llm.chat().completions().create(ChatCompletionCreateParams.builder()
.model("gpt-4o")
.addSystemMessage("请将以下命令执行结果格式化为用户友好的回复。")
.addUserMessage(results.toString())
.build()
).choices().get(0).message().content();
}
}
安全沙箱实现
import subprocess
class CommandSandbox:
"""安全执行 Shell 命令的沙箱"""
ALLOWED_COMMANDS = { # Level 1 白名单
"git status", "git log", "git diff",
"ls", "cat", "pwd", "echo",
"docker ps", "docker images",
"mvn dependency:tree", "mvn help",
}
BLOCKED_PATTERNS = [ # Level 3 黑名单
"rm -rf", "mkfs", "dd",
"shutdown", "reboot", ":(){ :|:& };:",
]
def run(self, command: str, timeout=30) -> dict:
# 检查黑名单
for pattern in self.BLOCKED_PATTERNS:
if pattern in command:
return {"error": f"危险命令已被拦截: {pattern}"}
# 子进程隔离执行，限制超时和资源
proc = subprocess.run(
command, shell=True,
capture_output=True, text=True,
timeout=timeout,
cwd=self.working_dir
)
return {
"exit_code": proc.returncode,
"stdout": proc.stdout[:5000], # 截断过长输出
"stderr": proc.stderr[:2000],
}
import { exec } from 'child_process';
import { promisify } from 'util';
const execAsync = promisify(exec);
class CommandSandbox {
// Level 1 白名单
static ALLOWED_COMMANDS = new Set([
'git status', 'git log', 'git diff',
'ls', 'cat', 'pwd', 'echo',
'docker ps', 'docker images',
'mvn dependency:tree', 'mvn help',
]);
// Level 3 黑名单
static BLOCKED_PATTERNS = [
'rm -rf', 'mkfs', 'dd',
'shutdown', 'reboot', ':(){ :|:& };:',
];
workingDir: string;
constructor(workingDir: string) {
this.workingDir = workingDir;
}
async run(command: string, timeout = 30): Promise> {
// 检查黑名单
for (const pattern of CommandSandbox.BLOCKED_PATTERNS) {
if (command.includes(pattern)) {
return { error: `危险命令已被拦截: ${pattern}` };
}
}
// 子进程隔离执行，限制超时和资源
try {
const { stdout, stderr } = await execAsync(command, {
cwd: this.workingDir,
timeout: timeout * 1000,
maxBuffer: 1024 * 1024,
});
return {
exit_code: 0,
stdout: stdout.slice(0, 5000),
stderr: stderr.slice(0, 2000),
};
} catch (e: any) {
return {
exit_code: e.code || 1,
stdout: (e.stdout || '').slice(0, 5000),
stderr: (e.stderr || '').slice(0, 2000),
};
}
}
}
package main
import (
"os/exec"
"strings"
"time"
)
// CommandSandbox 安全执行 Shell 命令的沙箱
type CommandSandbox struct {
workingDir string
}
// Level 1 白名单
var allowedCommands = map[string]bool{
"git status": true, "git log": true, "git diff": true,
"ls": true, "cat": true, "pwd": true, "echo": true,
"docker ps": true, "docker images": true,
"mvn dependency:tree": true, "mvn help": true,
}
// Level 3 黑名单
var blockedPatterns = []string{
"rm -rf", "mkfs", "dd",
"shutdown", "reboot", ":(){ :|:& };:",
}
func (s *CommandSandbox) Run(command string, timeout int) map[string]interface{} {
// 检查黑名单
for _, pattern := range blockedPatterns {
if strings.Contains(command, pattern) {
return map[string]interface{}{"error": "危险命令已被拦截: " + pattern}
}
}
// 子进程隔离执行，限制超时和资源
cmd := exec.Command("sh", "-c", command)
cmd.Dir = s.workingDir
timer := time.AfterFunc(time.Duration(timeout)*time.Second, func() {
cmd.Process.Kill()
})
defer timer.Stop()
stdout, stderr := "", ""
if out, err := cmd.Output(); err == nil {
stdout = string(out)
} else if ee, ok := err.(*exec.ExitError); ok {
stdout = string(ee.Stdout)
stderr = string(ee.Stderr)
}
// 截断过长输出
if len(stdout) > 5000 {
stdout = stdout[:5000]
}
if len(stderr) > 2000 {
stderr = stderr[:2000]
}
return map[string]interface{}{
"exit_code": cmd.ProcessState.ExitCode(),
"stdout": stdout,
"stderr": stderr,
}
}
import java.io.*;
import java.util.*;
import java.util.concurrent.*;
public class CommandSandbox {
private String workingDir;
// Level 1 白名单
private static final Set ALLOWED_COMMANDS = Set.of(
"git status", "git log", "git diff",
"ls", "cat", "pwd", "echo",
"docker ps", "docker images",
"mvn dependency:tree", "mvn help"
);
// Level 3 黑名单
private static final List BLOCKED_PATTERNS = List.of(
"rm -rf", "mkfs", "dd",
"shutdown", "reboot", ":(){ :|:& };:"
);
public CommandSandbox(String workingDir) {
this.workingDir = workingDir;
}
public Map run(String command, int timeout) {
// 检查黑名单
for (String pattern : BLOCKED_PATTERNS) {
if (command.contains(pattern)) {
return Map.of("error", "危险命令已被拦截: " + pattern);
}
}
// 子进程隔离执行，限制超时和资源
try {
ProcessBuilder pb = new ProcessBuilder("sh", "-c", command);
pb.directory(new File(workingDir));
pb.redirectErrorStream(false);
Process process = pb.start();
// 超时控制
if (!process.waitFor(timeout, TimeUnit.SECONDS)) {
process.destroyForcibly();
return Map.of("error", "命令执行超时");
}
String stdout = readStream(process.getInputStream(), 5000);
String stderr = readStream(process.getErrorStream(), 2000);
Map result = new LinkedHashMap<>();
result.put("exit_code", process.exitValue());
result.put("stdout", stdout);
result.put("stderr", stderr);
return result;
} catch (Exception e) {
return Map.of("error", e.getMessage());
}
}
private String readStream(InputStream is, int maxLen) throws IOException {
String content = new String(is.readAllBytes());
if (content.length() > maxLen) {
content = content.substring(0, maxLen);
}
return content;
}
}
13.6 CLI Agent 的错误自愈
命令执行失败是常态——参数错误、环境缺失、权限不足。好的 CLI Agent 不是简单报错，而是自动诊断并修复。
错误自愈是 CLI Agent 区别于"一次性命令生成器"的关键能力。真实环境中，命令失败的概率远高于理想情况：目标路径不存在、依赖工具未安装、权限被拒、网络超时……如果每次失败都要人工介入，Agent 就失去了"自主"的意义。自愈的核心思路是把错误信息本身作为新的上下文反馈给 LL让它像人类运维一样"看报错→想原因→换方案重试"。这个"执行→观察错误→修正→再执行"的小循环，正是第 5 章 ReAct 模式在 CLI 场景的具体落地。
🔄 错误自愈：四种典型场景 | 错误类型 | 示例 | 自愈策略 | | --- | --- | --- | | 命令不存在 | mvn: command not found | 检测 Maven 是否安装，建议安装命令 | | 参数错误 | git push: no upstream set | 自动加 --set-upstream origin branch 重试 | | 权限不足 | Permission denied | 提示需要 sudo，请求用户确认后重试 | | 依赖缺失 | ModuleNotFoundError | 先执行 pip install，再重新运行 | 错误自愈的核心思路：把 stderr 反馈给 LLM，让它分析原因并生成修正命令，最多重试 3 次：
def self_heal_loop(agent, user_input, max_retries=3): for attempt in range(max_retries): result = agent.execute(user_input) if result["exit_code"] == 0: return result # 成功！
失败 → 把错误信息反馈给 LLM 重新规划
error_msg = result["stderr"] user_input = f"上次执行失败，错误信息：{error_msg}\n请修正命令重新执行。"
return {"error": "重试3次仍然失败，请人工介入"}
13.7 CLI Agent 与 MCP + Skills 的组合
单独的 CLI 能力只是"一双手"。但配合 MCP（通信协议）和 Skills（组合复用），CLI Agent 就成了一个完整的智能体。 🧩 CLI × MCP × Skills 组合实战
场景：用户说 "帮我部署到测试环境"
- Skills 编排：识别意图 → 加载 "deploy-to-test" Skill → 规划多步流程
- MCP 通信：Skill 调用 MCP Tool → MCP Client 路由到 CLI MCP Server
- CLI 执行：MCP Server 把 Tool 调用翻译成 Shell 命令 → 沙箱执行
mvn package→docker build→docker compose up - 结果返回：CLI 输出 → MCP 响应 → Skills 判断是否成功 → LLM 格式化回复
这就是第二篇四个章节的完整闭环： 📖 第二篇知识脉络总结 | 章节 | 核心问题 | 解决什么 | | --- | --- | --- | | 第10章 Function Calling | LLM 如何调用外部工具？ | 定义调用格式和参数 Schema | | 第11章 MCP | 工具调用如何标准化？ | 统一通信协议，解耦 Client/Server | | 第12章 Skills | 工具如何组合与复用？ | 多工具编排、懒加载、沉淀机制 | | 第13章 CLI 能力 | Agent 如何操控本地工具？ | 本地命令行执行 + 安全沙箱 + 错误自愈 | 📋 八股总结 — 面试高频考点
Q1 CLI 能力和 MCP 有什么区别？什么时候用 CLI，什么时候用 MCP？
MCP 是通信协议，解决"怎么调工具"的问题（格式标准化、解耦 Client/Server）。CLI 能力是执行方式，解决"怎么在本地执行命令"的问题（Shell 生成、安全沙箱、错误自愈）。
两者互补而非替代：CLI 工具可以通过 MCP Server 注册为 Tool，MCP Client 调用后由 CLI 沙箱执行。远程 API 用 MCP 直接调，本地工具用 CLI 执行。
Q2 NL2Shell 的三种实现路径各有什么优缺点？企业级场景推荐哪种？
① 直接生成：LLM 直接输出 Shell 命令，简单但准确率 70~85%，容易产生幻觉命令。
② Few-Shot 模板：加常用命令示例库提升准确率到 85~92%，但模板库维护成本高。
③ 工具 Schema：用 Function Calling 定义命令 Schema，准确率 92~97%，参数可控、安全可审计。企业级推荐此方案。
Q3 CLI Agent 的安全策略如何设计？三级分类的依据是什么？
Level 1 白名单（自动执行）：只读命令，无修改、无破坏性，如 git status、ls、docker ps。
Level 2 确认执行：有修改但可控，如 git push、docker restart，需用户 Y/N 确认。
Level 3 拒绝/二次审批：不可逆操作，如 rm -rf、shutdown，默认拒绝，特殊场景管理员审批。
分类依据：是否可逆 + 影响范围 + 数据丢失风险。
Q4 CLI Agent 的错误自愈机制怎么实现？有什么局限性？
核心思路：把 stderr 反馈给 LLM，让它分析原因并生成修正命令，最多重试 3 次。
四种典型场景：命令不存在→建议安装；参数错误→自动修正；权限不足→加 sudo 重试；依赖缺失→先安装再执行。
局限性：① 循环重试可能产生副作用；② 某些错误 LLM 无法判断原因；③ sudo 确认不能自动绕过。企业级需设重试上限和回滚机制。
Q5 为什么 CLI 能力要单独成章，而不是合并到 MCP 章节？
因为 CLI 能力的技术栈和挑战与 MCP 完全不同：
① MCP 是协议层问题（JSON-RPC、握手、能力协商）；CLI 是执行层问题（Shell 生成、子进程隔离、资源限制）。
② MCP 处理的是远程 API 调用；CLI 处理的是本地系统级操作，安全模型完全不同。
③ CLI 有独特的错误自愈、NL2Shell、沙箱隔离问题，这些在 MCP 章节无法深入覆盖。
13.8 NL2Shell 实战：Few-Shot 提升准确率
NL2Shell（自然语言转 Shell 命令）是 CLI Agent 的核心能力。纯零样本（zero-shot）生成的命令准确率约 60-70%，通过 Few-Shot 示例 + 工具 Schema 可提升至 90%+。
零样本之所以不准，是因为自然语言存在大量歧义："找出大文件"是找 >100MB 还是 >1GB？"最近修改"是 24 小时内还是一周内？LLM 在没有任何参照时会自行猜测，猜错就生成错误命令。Few-Shot 的作用是给模型提供带标准答案的范例，把模糊的意图锚定到具体的命令模式上。实验表明，精心设计的 3-5 个示例就能覆盖 80% 的常见运维场景，让准确率产生质的飞跃。这也是"提示词工程"在 CLI 领域最直接的收益点。
Python TypeScript
# nl2shell.py — 自然语言转 Shell 命令引擎
from dataclasses import dataclass
from typing import List, Optional
# Few-Shot 示例库（按意图分类）
FEW_SHOT_EXAMPLES = {
"查看进程": [
{"input": "查看占用 8080 端口的进程", "command": "lsof -i :8080"},
{"input": "查看 CPU 占用最高的 5 个进程", "command": "ps aux --sort=-%cpu | head -5"},
{"input": "查看内存使用情况", "command": "free -h"},
],
"文件操作": [
{"input": "找出 /var/log 下大于 100MB 的日志文件", "command": "find /var/log -type f -size +100M"},
{"input": "统计当前目录下各类型文件数量", "command": "find . -type f | sed 's/.*\\.//' | sort | uniq -c | sort -rn"},
{"input": "批量将 .txt 文件重命名为 .md", "command": "for f in *.txt; do mv \"$f\" \"${f%.txt}.md\"; done"},
],
"网络诊断": [
{"input": "检查到 github.com 的网络连通性", "command": "ping -c 3 github.com"},
{"input": "查看当前网络连接数", "command": "ss -s"},
{"input": "抓取 80 端口的 HTTP 请求", "command": "tcpdump -i any port 80 -A -s 0"},
],
"Docker运维": [
{"input": "查看所有运行中的容器", "command": "docker ps"},
{"input": "清理已停止的容器和悬空镜像", "command": "docker container prune -f && docker image prune -f"},
{"input": "查看 nginx 容器最近 50 行日志", "command": "docker logs --tail 50 nginx"},
],
}
@dataclass
class NL2ShellResult:
command: str
confidence: float
risk_level: str # safe / medium / high
explanation: str
class NL2ShellEngine:
"""NL2Shell 引擎：Few-Shot + 工具 Schema + 安全分级"""
# 高危命令模式
DANGEROUS_PATTERNS = [
r'rm\s+-rf\s+/', # 递归删除根目录
r'mkfs\.\w+\s+/dev', # 格式化磁盘
r'dd\s+.*of=/dev/sd', # 直接写磁盘
r':\(\)\{.*\};:', # Fork bomb
r'chmod\s+-R\s+777\s+/', # 递归 777
]
def generate(self, user_input: str, context: str = "") -> NL2ShellResult:
"""将自然语言转为 Shell 命令"""
# 1. 意图分类 → 选择 Few-Shot 示例
category = self._classify_intent(user_input)
examples = FEW_SHOT_EXAMPLES.get(category, [])
# 2. 构建 Prompt（Few-Shot + 工具 Schema）
prompt = self._build_prompt(user_input, examples, context)
# 3. 调用 LLM 生成命令
raw_command = self._call_llm(prompt)
# 4. 安全分级
risk = self._assess_risk(raw_command)
# 5. 清理（去掉多余的解释文字）
command = self._extract_command(raw_command)
return NL2ShellResult(
command=command,
confidence=0.92 if examples else 0.65,
risk_level=risk,
explanation=f"分类: {category}, Few-Shot: {len(examples)} examples"
)
def _classify_intent(self, text: str) -> str:
"""简单意图分类"""
if any(w in text for w in ["进程", "端口", "CPU", "内存"]):
return "查看进程"
if any(w in text for w in ["文件", "查找", "重命名", "日志"]):
return "文件操作"
if any(w in text for w in ["网络", "ping", "连通", "抓包"]):
return "网络诊断"
if any(w in text for w in ["docker", "容器", "镜像"]):
return "Docker运维"
return "general"
def _assess_risk(self, command: str) -> str:
import re
for pattern in self.DANGEROUS_PATTERNS:
if re.search(pattern, command):
return "high"
if any(w in command for w in ["rm ", "kill ", "drop ", "delete "]):
return "medium"
return "safe"
# === 使用示例 ===
engine = NL2ShellEngine()
# 测试几个例子
test_cases = [
"找出 /tmp 下 7 天前的临时文件",
"查看 nginx 容器的资源使用",
"批量压缩当前目录的 .log 文件",
]
for tc in test_cases:
result = engine.generate(tc)
print(f"📝 {tc}")
print(f" → {result.command}")
print(f" 风险: {result.risk_level}, 置信度: {result.confidence:.0%}")
print()
# 输出：
# 📝 找出 /tmp 下 7 天前的临时文件
# → find /tmp -type f -mtime +7
# 风险: safe, 置信度: 92%
#
# 📝 查看 nginx 容器的资源使用
# → docker stats nginx --no-stream
# 风险: safe, 置信度: 92%
#
# 📝 批量压缩当前目录的 .log 文件
# → for f in *.log; do gzip "$f"; done
# 风险: medium, 置信度: 92%
// nl2shell.ts
interface NL2ShellResult {
command: string;
confidence: number;
riskLevel: 'safe' | 'medium' | 'high';
}
const FEW_SHOT_EXAMPLES: Record> = {
'查看进程': [
{ input: '查看占用 8080 端口的进程', command: 'lsof -i :8080' },
{ input: '查看 CPU 占用最高的进程', command: 'ps aux --sort=-%cpu | head -5' },
],
'文件操作': [
{ input: '找出大于 100MB 的文件', command: 'find . -type f -size +100M' },
{ input: '统计文件类型', command: 'find . -type f | sed \'s/.*\\.//\' | sort | uniq -c' },
],
};
export function generateCommand(userInput: string): NL2ShellResult {
// 意图分类 → Few-Shot → LLM → 安全分级
const category = classifyIntent(userInput);
const examples = FEW_SHOT_EXAMPLES[category] || [];
return {
command: '', // LLM 生成
confidence: examples.length > 0 ? 0.92 : 0.65,
riskLevel: 'safe',
};
}
13.9 SSH Agent：远程运维场景
企业级 Agent 不仅要操控本地环境，还需要通过 SSH 远程操控服务器。这是 DevOps 场景的核心能力，也是企业招聘面试的高频话题。以 WaLiCode 为例，其 SSH DevOps 模式允许 Agent 通过 SSH 连接到远程服务器执行命令、管理文件、部署应用。
13.9.1 SSH Agent 架构
SSH Agent 三层架构：
① 连接管理层：SSH 连接池（保活 + 自动重连 + 密钥管理）
② 命令执行层：远程命令执行 + 流式输出 + 超时控制
③ 安全沙箱层：命令预检 + 权限分级 + 审计日志
WaLiCode 使用 Rust 的 russh 库实现纯 Rust SSH 客户端，避免了对系统 ssh 命令的依赖，跨平台兼容性更好。
Python (paramiko) TypeScript (ssh2)
# ssh_agent.py — SSH 远程执行 Agent
import paramiko
from dataclasses import dataclass
from typing import Optional
import re
@dataclass
class SSHConfig:
host: str
port: int = 22
username: str = "root"
key_path: str = "~/.ssh/id_rsa"
@dataclass
class CommandResult:
stdout: str
stderr: str
exit_code: int
duration_ms: int
class SSHAgent:
"""SSH 远程执行 Agent"""
# 远程高危命令模式
REMOTE_DANGEROUS = [
r'rm\s+-rf\s+/',
r'shutdown',
r'reboot',
r'mkfs',
r'iptables\s+-F', # 清空防火墙规则
]
def __init__(self, config: SSHConfig):
self.config = config
self.client: Optional[paramiko.SSHClient] = None
def connect(self):
"""建立 SSH 连接"""
self.client = paramiko.SSHClient()
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.client.connect(
hostname=self.config.host,
port=self.config.port,
username=self.config.username,
key_filename=self.config.key_path,
timeout=10
)
print(f"✅ SSH 连接成功: {self.config.host}")
def execute(self, command: str, timeout: int = 30) -> CommandResult:
"""远程执行命令（带安全检查）"""
# 安全预检
risk = self._check_command(command)
if risk == "high":
return CommandResult("", "命令被安全策略拒绝", -1, 0)
if not self.client:
raise RuntimeError("SSH 未连接，请先调用 connect()")
import time
start = time.time()
stdin, stdout, stderr = self.client.exec_command(command, timeout=timeout)
exit_code = stdout.channel.recv_exit_status()
duration_ms = int((time.time() - start) * 1000)
return CommandResult(
stdout=stdout.read().decode('utf-8'),
stderr=stderr.read().decode('utf-8'),
exit_code=exit_code,
duration_ms=duration_ms
)
def execute_streaming(self, command: str, on_line=None):
"""流式执行命令（实时输出每行）"""
if not self.client:
raise RuntimeError("SSH 未连接")
stdin, stdout, stderr = self.client.exec_command(command)
for line in stdout:
if on_line:
on_line(line.strip())
else:
print(line.strip())
def upload(self, local_path: str, remote_path: str):
"""上传文件"""
sftp = self.client.open_sftp()
sftp.put(local_path, remote_path)
sftp.close()
print(f"✅ 上传: {local_path} → {remote_path}")
def _check_command(self, command: str) -> str:
"""命令安全分级"""
for pattern in self.REMOTE_DANGEROUS:
if re.search(pattern, command):
return "high"
if any(w in command for w in ["rm ", "kill ", "chmod 777"]):
return "medium"
return "safe"
def close(self):
if self.client:
self.client.close()
print("SSH 连接已关闭")
# === 使用示例：远程部署 ===
agent = SSHAgent(SSHConfig(
host="192.168.1.100",
username="deploy",
key_path="~/.ssh/deploy_key"
))
agent.connect()
# 1. 检查远程环境
result = agent.execute("uname -a && docker --version")
print(f"系统信息:\n{result.stdout}")
# 2. 拉取最新代码
result = agent.execute("cd /opt/myapp && git pull origin main")
print(f"部署结果: exit={result.exit_code}")
# 3. 重启服务（流式输出日志）
agent.execute_streaming(
"cd /opt/myapp && docker compose up -d --build",
on_line=lambda line: print(f" {line}")
)
agent.close()
# 输出：
# ✅ SSH 连接成功: 192.168.1.100
# 系统信息: Linux server1 5.15.0 ... x86_64 GNU/Linux
# Docker version 24.0.7
# 部署结果: exit=0
# Building myapp
# Container myapp Starting
# Container myapp Started
# SSH 连接已关闭
// ssh-agent.ts — 基于 ssh2 的 TypeScript 实现
import { Client, ConnectConfig } from 'ssh2';
interface CommandResult {
stdout: string;
stderr: string;
exitCode: number;
durationMs: number;
}
export class SSHAgent {
private conn: Client;
constructor(private config: ConnectConfig) {
this.conn = new Client();
}
connect(): Promise {
return new Promise((resolve, reject) => {
this.conn.on('ready', () => resolve());
this.conn.on('error', reject);
this.conn.connect(this.config);
});
}
execute(command: string, timeout = 30): Promise {
return new Promise((resolve, reject) => {
const start = Date.now();
this.conn.exec(command, (err, stream) => {
if (err) return reject(err);
let stdout = '', stderr = '';
stream.on('close', (code) => {
resolve({
stdout, stderr,
exitCode: code ?? -1,
durationMs: Date.now() - start,
});
});
stream.on('data', (data) => stdout += data);
stream.stderr.on('data', (data) => stderr += data);
});
});
}
close() {
this.conn.end();
}
}
// 使用示例
const agent = new SSHAgent({
host: '192.168.1.100',
port: 22,
username: 'deploy',
privateKey: require('fs').readFileSync('/home/user/.ssh/id_rsa'),
});
await agent.connect();
const result = await agent.execute('docker ps');
console.log(result.stdout);
agent.close();