用 Python 与 Ollama 构建一个更安全的本地 CLI Agent

从最小工具调用闭环开始，用离线 Mock 验证流程，再接入本地 Ollama。本文不是生产级自治代理方案，而是一份可运行、可检查、默认只读的入门教程。

原文：How to Build CLI Agents with Python & Ollama
作者：Mauro Di Pietro
来源：Towards Data Science，2026-08-03
原文链接：https://towardsdatascience.com/cli-agents-with-python-ollama/
提取说明：原页面的直接 HTTP 请求返回 403，正文通过浏览器 DOM 提取；原文部分代码片段在页面中并不完整，因此本文对缺失部分进行了明确标注的实践补全。

你将完成什么

最终程序包含一个最小 Agent 闭环：

用户请求
   ↓
本地模型判断是否调用工具
   ↓
工具 schema → 工具名称与参数
   ↓
Python 执行只读命令
   ↓
工具结果写回消息历史
   ↓
模型生成最终回答

它支持三个只读场景：

df -h：查看磁盘空间；
uname -a：查看系统信息；
ps -eo pid,comm,%cpu,%mem --sort=-%cpu：查看 CPU 占用靠前的进程。

同时，它会拒绝不在白名单中的命令。删除、清理和写文件操作没有开放。

原文方法覆盖清单

原文方法	是否纳入	落点	若遗漏原因
使用 Python 与 Ollama/Qwen 构建完全本地 Agent	是	环境准备、Ollama 实机运行	—
用 Python 函数执行 Shell 命令	是	`execute_safe_command()`	—
用工具名称映射和 JSON schema 暴露工具	是	`TOOL_MAP`、`TOOLS_SCHEMA`	—
使用 system/user/assistant/tool 消息维护上下文	是	`run_turn()`	—
循环处理模型的 `tool_calls` 并回传结果	是	`run_turn()` 的工具调用循环	—
用磁盘、系统和进程查询作为示例	是	三个只读白名单命令	—
使用 `shell=True` 直接执行模型生成的任意命令	否	改为 `shell=False` 与精确白名单	原文已提醒模糊清理请求可能造成损害；直接照搬不适合作为分享教程默认实现
扩展 Excel、Python 等更多工具	部分	文末扩展原则	原文只提出方向，没有给出完整实现；本文不把概念扩展伪装成已验证功能

内容边界

原文事实：原文使用 ollama==0.6.2、qwen2.5、工具 schema、消息历史和工具调用循环来演示本地 CLI Agent。
本文提炼：最小 Agent 的关键不是聊天界面，而是“模型决策—工具执行—结果回传—继续推理”的闭环。
实践扩展：下面增加了命令白名单、shell=False、工具轮次上限、离线 Mock、单元测试和明确的验证步骤。这些安全措施不是原文完整实现的一部分。

1. 环境准备

需要：

Python 3.10 或更高版本；
Ollama；
能进行工具调用的本地模型；
Linux 或 macOS。Windows 用户需要替换本教程中的系统查询命令。

创建项目：

mkdir local-cli-agent
cd local-cli-agent
python3 -m venv .venv
source .venv/bin/activate

目录结构：

local-cli-agent/
├── cli_agent.py
├── requirements.txt
└── test_cli_agent.py

2. 安装 Ollama Python SDK

文件：`requirements.txt`

ollama==0.6.2

安装依赖：

python -m pip install -r requirements.txt

安装并启动 Ollama 后，拉取原文使用的模型：

ollama pull qwen2.5

检查本地服务与模型：

curl -fsS http://127.0.0.1:11434/api/version
curl -fsS http://127.0.0.1:11434/api/tags

检查点：第一条命令应返回版本 JSON；第二条结果中应能找到 qwen2.5。Ollama 官方文档确认其聊天 API 支持 function/tool calling，并要求把工具结果作为 role=tool、带 tool_name 的消息写回上下文。

3. 编写 Agent、工具和双后端

文件：`cli_agent.py`

from __future__ import annotations

import argparse
import json
import shlex
import subprocess
from typing import Any, Protocol

MODEL = "qwen2.5"
MAX_TOOL_ROUNDS = 4
ALLOWED_COMMANDS = {
    ("df", "-h"),
    ("uname", "-a"),
    ("ps", "-eo", "pid,comm,%cpu,%mem", "--sort=-%cpu"),
}

TOOLS_SCHEMA = [
    {
        "type": "function",
        "function": {
            "name": "execute_safe_command",
            "description": "Execute one approved read-only system inspection command.",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "enum": [
                            "df -h",
                            "uname -a",
                            "ps -eo pid,comm,%cpu,%mem --sort=-%cpu",
                        ],
                    }
                },
                "required": ["command"],
            },
        },
    }
]


class ChatBackend(Protocol):
    def chat(
        self,
        *,
        model: str,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
    ) -> Any: ...


def _as_dict(value: Any) -> dict[str, Any]:
    if isinstance(value, dict):
        return value
    if hasattr(value, "model_dump"):
        return value.model_dump(exclude_none=True)
    if hasattr(value, "dict"):
        return value.dict()
    raise TypeError(f"Unsupported response object: {type(value)!r}")


def execute_safe_command(command: str) -> str:
    argv = tuple(shlex.split(command))
    if argv not in ALLOWED_COMMANDS:
        return f"BLOCKED: command is not allowlisted: {command}"
    try:
        result = subprocess.run(
            list(argv),
            shell=False,
            capture_output=True,
            text=True,
            timeout=5,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        return f"ERROR: {exc}"
    output = result.stdout if result.returncode == 0 else result.stderr
    return output[:4000] or (
        f"Command exited with status {result.returncode} and no output."
    )


TOOL_MAP = {"execute_safe_command": execute_safe_command}


class MockBackend:
    """Offline backend used to verify the agent/tool loop."""

    def __init__(self) -> None:
        self.calls = 0

    def chat(
        self,
        *,
        model: str,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
    ) -> dict[str, Any]:
        self.calls += 1
        if self.calls == 1:
            return {
                "message": {
                    "role": "assistant",
                    "content": "",
                    "tool_calls": [
                        {
                            "function": {
                                "name": "execute_safe_command",
                                "arguments": {"command": "df -h"},
                            }
                        }
                    ],
                }
            }
        tool_output = next(
            m["content"] for m in reversed(messages) if m["role"] == "tool"
        )
        first_line = tool_output.splitlines()[0] if tool_output else "no output"
        return {
            "message": {
                "role": "assistant",
                "content": (
                    "Mock verification complete. "
                    f"Tool output starts with: {first_line}"
                ),
            }
        }


class OllamaBackend:
    def chat(
        self,
        *,
        model: str,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
    ) -> Any:
        import ollama

        return ollama.chat(model=model, messages=messages, tools=tools)


def run_turn(
    backend: ChatBackend,
    messages: list[dict[str, Any]],
    user_input: str,
) -> str:
    messages.append({"role": "user", "content": user_input})
    for _ in range(MAX_TOOL_ROUNDS + 1):
        response = _as_dict(
            backend.chat(model=MODEL, messages=messages, tools=TOOLS_SCHEMA)
        )
        message = _as_dict(response["message"])
        tool_calls = message.get("tool_calls") or []
        messages.append(message)
        if not tool_calls:
            return str(message.get("content", ""))

        for raw_call in tool_calls:
            call = _as_dict(raw_call)
            function = _as_dict(call["function"])
            name = str(function.get("name", ""))
            arguments = function.get("arguments", {})
            if isinstance(arguments, str):
                arguments = json.loads(arguments)
            tool = TOOL_MAP.get(name)
            result = tool(**arguments) if tool else f"BLOCKED: unknown tool: {name}"
            print(f"[tool] {name}({arguments})")
            messages.append(
                {"role": "tool", "tool_name": name, "content": result}
            )

    raise RuntimeError("Tool-call round limit exceeded")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "prompt",
        nargs="?",
        default="How much disk space is left?",
    )
    parser.add_argument(
        "--mock",
        action="store_true",
        help="Run the offline verification backend",
    )
    args = parser.parse_args()

    backend: ChatBackend = MockBackend() if args.mock else OllamaBackend()
    messages = [
        {
            "role": "system",
            "content": (
                "You are a local CLI assistant. Use only the provided "
                "read-only inspection tool. Never invent command output."
            ),
        }
    ]
    print(run_turn(backend, messages, args.prompt))


if __name__ == "__main__":
    main()

为什么不照搬 `shell=True`

原文的概念代码把模型生成的字符串交给：

subprocess.run(command, shell=True, ...)

这意味着 Shell 会解释管道、重定向、命令替换和连接符。仅靠模型提示词中的“safe”无法限制实际执行能力。本文改用三层边界：

shlex.split() 把字符串解析为参数列表；
精确匹配 ALLOWED_COMMANDS；
shell=False，不让 Shell 解释额外语法。

这仍不是强隔离。真正处理不可信输入时，还需要容器、低权限用户、只读文件系统和人工审批。

4. 先运行离线 Mock 闭环

Mock 后端会模拟一次模型工具调用，但工具函数会真实执行只读的 df -h。它不需要 Ollama，也不需要下载模型。

python cli_agent.py --mock "How much disk space is left?"

本教程实际验证时的输出形状：

[tool] execute_safe_command({'command': 'df -h'})
Mock verification complete. Tool output starts with: Filesystem      Size  Used Avail Use% Mounted on

不同系统的表头和磁盘数据可能不同。验收标准是：

出现 [tool] execute_safe_command(...)；
最终回答包含 Mock verification complete；
工具结果确实进入消息历史，而不是模型凭空编造。

5. 添加单元测试

文件：`test_cli_agent.py`

import unittest

from cli_agent import MockBackend, execute_safe_command, run_turn


class CliAgentTests(unittest.TestCase):
    def test_blocks_unapproved_command(self) -> None:
        self.assertTrue(
            execute_safe_command("rm -rf /tmp/example").startswith("BLOCKED:")
        )

    def test_mock_tool_loop(self) -> None:
        messages = [{"role": "system", "content": "test"}]
        answer = run_turn(
            MockBackend(),
            messages,
            "How much disk space is left?",
        )
        self.assertIn("Mock verification complete", answer)
        self.assertTrue(
            any(message["role"] == "tool" for message in messages)
        )


if __name__ == "__main__":
    unittest.main()

运行：

python -m unittest -v

本教程实际执行结果：

test_blocks_unapproved_command ... ok
test_mock_tool_loop ... ok

Ran 2 tests in 0.005s
OK

失败边界也可以手动检查：

python -c 'from cli_agent import execute_safe_command; print(execute_safe_command("rm -rf /tmp/example"))'

预期输出：

BLOCKED: command is not allowlisted: rm -rf /tmp/example

6. 接入真实 Ollama

确认 Ollama 正在运行且模型已下载后，执行：

python cli_agent.py "How much disk space is left?"
python cli_agent.py "Show my operating system information"
python cli_agent.py "Which processes use the most CPU?"

真实模型的措辞不固定，但验证时不要只看最终答案是否“像真的”。还要确认终端出现类似日志：

[tool] execute_safe_command({'command': 'uname -a'})

如果没有 [tool] 日志，而回答中却出现具体系统数据，就不能证明模型真的调用了工具。

三个实践案例

案例一：磁盘空间

用户意图：查看剩余磁盘容量；
允许命令：df -h；
验证：回答中的容量应能在直接执行 df -h 的结果中找到。

案例二：系统信息

用户意图：识别内核和机器架构；
允许命令：uname -a；
验证：回答应与直接执行 uname -a 一致。

案例三：CPU 进程

用户意图：查看高 CPU 进程；
允许命令：ps -eo pid,comm,%cpu,%mem --sort=-%cpu；
验证：检查工具日志和输出前几行，不要把瞬时排序当成长时间性能结论。

7. 常见故障

`Connection refused`

Ollama 服务没有启动。先检查：

curl -fsS http://127.0.0.1:11434/api/version

`model not found`

模型尚未下载，或 MODEL 与本地名称不同：

ollama pull qwen2.5
ollama list

如改用其他支持工具调用的模型，请同步修改 cli_agent.py 中的 MODEL，并重新做真实工具调用验证。

模型一直不调用工具

先确认模型支持 tool calling；
让用户请求明确对应一个白名单能力；
检查 TOOLS_SCHEMA 的命令枚举和工具描述；
不要通过放开任意 Shell 权限来“修复”调用率。

命令被 `BLOCKED`

这是预期的失败关闭行为。新增命令时，应同时：

确认它只读且参数固定；
加入 ALLOWED_COMMANDS；
加入 schema 的 enum；
新增允许和拒绝两类测试。

工具调用陷入循环

MAX_TOOL_ROUNDS 会在超过上限时抛出异常。不要无限重试；记录最后一次工具名、参数和结果，再修正 schema 或提示词。

8. 从演示走向可维护工具

原文提出可以继续增加 Excel、Python 等工具。更稳妥的扩展方式不是扩大通用 Shell 权限，而是为每项能力创建窄工具，例如：

read_csv_summary(path)：只读取指定目录中的 CSV，并限制文件大小；
list_files(directory)：仅列出工作区，不允许任意路径；
run_python_report(report_id)：从预注册脚本中选择，而不是执行任意 Python 字符串。

每个写操作都应增加独立的预览与确认阶段：

模型提出操作 → 程序生成变更预览 → 用户确认 → 执行 → 读取结果验证

不要把“模型再次检查参数”当作权限控制。模型复核仍然是概率性判断，不能替代程序化白名单、隔离和人工批准。

9. 验证等级

static_publish_ok：需在 Markdown、HTML 生成并完成远端非空检查后确认。
mock_code_run_ok：是。离线工具循环已实际运行，2 个单元测试通过。
real_backend_contract_ok：是（静态）。代码与 Ollama 官方工具调用消息结构一致，并通过 Python 编译检查。
real_backend_smoke_ok：否。本教程制作过程中没有调用本地 Ollama 模型，因此不声称真实模型已完成工具调用。

总结

一个最小 CLI Agent 只需要模型、消息历史、工具 schema、函数映射和工具调用循环；真正困难的是执行边界。先用 Mock 证明闭环，再用只读白名单接入 Ollama，最后以工具日志而不是“像真的答案”验证调用发生。这条路径比直接把模型输出交给 shell=True 更适合学习、演示和后续扩展。

参考资料

原文：https://towardsdatascience.com/cli-agents-with-python-ollama/
Ollama Quickstart：https://docs.ollama.com/quickstart
Ollama Tool Calling：https://docs.ollama.com/capabilities/tool-calling
Ollama Python 工具调用示例：https://github.com/ollama/ollama-python/blob/main/examples/tools.py