# 用 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 闭环：

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

它支持三个只读场景：

1. `df -h`：查看磁盘空间；
2. `uname -a`：查看系统信息；
3. `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 用户需要替换本教程中的系统查询命令。

创建项目：

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

目录结构：

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

## 2. 安装 Ollama Python SDK

### 文件：`requirements.txt`

```text
ollama==0.6.2
```

安装依赖：

```bash
python -m pip install -r requirements.txt
```

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

```bash
ollama pull qwen2.5
```

检查本地服务与模型：

```bash
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`

```python
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`

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

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

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

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

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

## 4. 先运行离线 Mock 闭环

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

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

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

```text
[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`

```python
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()
```

运行：

```bash
python -m unittest -v
```

本教程实际执行结果：

```text
test_blocks_unapproved_command ... ok
test_mock_tool_loop ... ok

Ran 2 tests in 0.005s
OK
```

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

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

预期输出：

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

## 6. 接入真实 Ollama

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

```bash
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?"
```

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

```text
[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 服务没有启动。先检查：

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

### `model not found`

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

```bash
ollama pull qwen2.5
ollama list
```

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

### 模型一直不调用工具

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

### 命令被 `BLOCKED`

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

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

### 工具调用陷入循环

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

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

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

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

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

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

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

## 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