用 LFM2.5-2.6B 部署只读 Linux 运维 Agent

目标：用本地模型完成主机巡检、服务状态分诊和日志容量排查；第一阶段只读、可审计、人工复核，不允许自动修复。

原文：VentureBeat：No cloud, no GPUs, no problem: Liquid AI’s new model LFM2.5-2.6B…

官方资料：模型文档 · 本地 Agent 指南 · GGUF · llama.cpp

1. 方案结论

LFM2.5-2.6B 适合高频、边界清楚、工具驱动的本地运维任务。安全边界不能只靠提示词：模型负责选工具和写报告，程序负责白名单、参数校验、超时、审计与人工审批。

运维人员 → Agent 编排器 → LFM2.5-2.6B（llama.cpp /v1）
                         ↓
              只读白名单工具 + 参数校验
                         ↓
              reports/*.md + traces/*.json
                         ↓
                       人工复核

原文事实：模型为 26 亿参数稠密文本模型，支持 128K 上下文和原生工具调用；官方提供 GGUF/MLX/ONNX，并支持 llama.cpp、vLLM、SGLang。文章中的吞吐和内存主要是厂商或特定第三方数据，不等于你的实测。复杂编码与高难度推理仍适合更大模型。年收入超过 1000 万美元的企业须单独确认 LFM Open License v1.0 商业授权。

本文提炼：固定运维任务优先改进 Harness 和工具，而不是先微调模型。

实践扩展：下文的只读 Python Agent、systemd 单元和安全检查是本文新增实现，不是原文命令。

2. 原文方法覆盖清单

原文方法	纳入	落点	说明
CPU/端侧运行	是	llama.cpp CPU 部署	—
原生工具调用	是	OpenAI-compatible tools 循环	—
Harness 补足模型弱点	是	白名单、校验、trace	—
换工具复用同一模型	是	三个运维工具	—
128K 上下文	是	容量规划	默认 8K，避免内存浪费
生产微调	部分	分阶段上线	只读试点无需先微调
手机/树莓派 Harness	否	—	本教程聚焦 Linux 节点
商业许可	是	上线检查	—

3. 边界与资源选择

适合：每日主机健康报告、指定 systemd 服务状态读取、指定日志目录的容量排查、数据不能离开内网的环境。

禁止：通用 Shell；自动 restart、rm、deploy、rollback；数据库/Kubernetes 写入；把模型输出当监控或 CMDB 的事实源。

官方 GGUF 参考：Q4_K_M 约 1.67 GB，Q6_K 约 2.22 GB，Q8_0 约 2.87 GB，BF16 约 5.4 GB。低内存 CPU 节点从 Q4_K_M + 8K context 开始；发生真实截断再升到 32K。支持 128K 不代表应默认开满，KV cache 也占内存。

4. 先跑无模型安全闭环

mkdir -p ~/lfm-ops-agent
cd ~/lfm-ops-agent

保存为 ops_agent.py：

#!/usr/bin/env python3
"""Read-only Linux operations agent: mock-first, OpenAI-compatible backend optional."""

from __future__ import annotations

import argparse
import ast
import json
import operator
import os
import re
import subprocess
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

SERVICE_RE = re.compile(r"^[A-Za-z0-9_.@-]{1,128}$")
ALLOWED_LOG_ROOT = Path(os.environ.get("OPS_LOG_ROOT", "/var/log")).resolve()
TIMEOUT = float(os.environ.get("OPS_COMMAND_TIMEOUT", "5"))
OPS: dict[type[ast.operator], Any] = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.FloorDiv: operator.floordiv,
    ast.Mod: operator.mod,
    ast.Pow: operator.pow,
}


def run(argv: list[str]) -> dict[str, Any]:
    try:
        p = subprocess.run(argv, text=True, capture_output=True, timeout=TIMEOUT, check=False)
        return {"argv": argv, "exit_code": p.returncode, "stdout": p.stdout.strip(), "stderr": p.stderr.strip()}
    except (OSError, subprocess.TimeoutExpired) as exc:
        return {"argv": argv, "error": type(exc).__name__, "detail": str(exc)}


def host_health(_: dict[str, Any] | None = None) -> dict[str, Any]:
    return {"uptime": run(["uptime"]), "root_disk": run(["df", "-h", "--output=source,size,used,avail,pcent,target", "/"])}


def service_status(args: dict[str, Any]) -> dict[str, Any]:
    name = str(args.get("name", ""))
    if not SERVICE_RE.fullmatch(name):
        raise ValueError("invalid service name")
    return run(["systemctl", "show", name, "--no-pager", "--property=Id,LoadState,ActiveState,SubState,Result"])


def log_sizes(args: dict[str, Any]) -> dict[str, Any]:
    requested = Path(str(args.get("path", ALLOWED_LOG_ROOT))).resolve()
    if requested != ALLOWED_LOG_ROOT and ALLOWED_LOG_ROOT not in requested.parents:
        raise ValueError("path outside OPS_LOG_ROOT")
    if not requested.is_dir():
        raise ValueError("log path is not a directory")
    limit = max(1, min(int(args.get("limit", 10)), 50))
    rows = []
    for item in requested.iterdir():
        try:
            if item.is_file():
                rows.append({"path": str(item), "bytes": item.stat().st_size})
        except OSError:
            continue
    rows.sort(key=lambda row: row["bytes"], reverse=True)
    return {"root": str(requested), "files": rows[:limit]}


def safe_calc(args: dict[str, Any]) -> dict[str, Any]:
    expression = str(args.get("expression", "")).strip()
    if not expression:
        raise ValueError("empty expression")
    tree = ast.parse(expression, mode="eval")

    def evaluate(node: ast.AST) -> float | int:
        if isinstance(node, ast.Expression):
            return evaluate(node.body)
        if isinstance(node, ast.Constant):
            value = node.value
            if isinstance(value, (int, float)) and not isinstance(value, bool):
                return value
        if isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.UAdd, ast.USub)):
            value = evaluate(node.operand)
            return value if isinstance(node.op, ast.UAdd) else -value
        if isinstance(node, ast.BinOp) and type(node.op) in OPS:
            left, right = evaluate(node.left), evaluate(node.right)
            if isinstance(node.op, ast.Pow) and abs(right) > 10:
                raise ValueError("exponent too large")
            return OPS[type(node.op)](left, right)
        raise ValueError("unsupported expression")

    value = evaluate(tree)
    if abs(float(value)) > 1e15:
        raise ValueError("result too large")
    return {"expression": expression, "result": value}


TOOLS = {
    "host_health": host_health,
    "service_status": service_status,
    "log_sizes": log_sizes,
    "safe_calc": safe_calc,
}
TOOL_SCHEMA = [
    {"type": "function", "function": {"name": "host_health", "description": "Read uptime and root filesystem usage.", "parameters": {"type": "object", "properties": {}}}},
    {"type": "function", "function": {"name": "service_status", "description": "Read systemd service state; never restart it.", "parameters": {"type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"]}}},
    {"type": "function", "function": {"name": "log_sizes", "description": "List largest files under the allowed log root; never delete files.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}}}},
    {"type": "function", "function": {"name": "safe_calc", "description": "Evaluate bounded arithmetic only.", "parameters": {"type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"]}}},
]


def invoke(name: str, args: dict[str, Any]) -> dict[str, Any]:
    if name not in TOOLS:
        return {"ok": False, "error": "tool_not_allowed"}
    try:
        return {"ok": True, "data": TOOLS[name](args)}
    except Exception as exc:
        return {"ok": False, "error": type(exc).__name__, "detail": str(exc)}


def post_chat(messages: list[dict[str, Any]]) -> dict[str, Any]:
    base = os.environ.get("MODEL_BASE_URL", "http://127.0.0.1:8080/v1").rstrip("/")
    payload = json.dumps({"model": os.environ.get("MODEL_NAME", "LFM2.5-2.6B"), "messages": messages, "tools": TOOL_SCHEMA, "temperature": 0.1}, ensure_ascii=False).encode()
    req = urllib.request.Request(base + "/chat/completions", data=payload, headers={"Content-Type": "application/json", "Authorization": "Bearer " + os.environ.get("MODEL_API_KEY", "local")})
    with urllib.request.urlopen(req, timeout=60) as response:
        return json.load(response)


def real_agent(prompt: str) -> tuple[str, list[dict[str, Any]]]:
    messages: list[dict[str, Any]] = [
        {"role": "system", "content": "You are a read-only Linux operations assistant. Use tools for host facts. Never invent missing observations or propose that a write already happened."},
        {"role": "user", "content": prompt},
    ]
    trace: list[dict[str, Any]] = []
    first = post_chat(messages)
    assistant = first["choices"][0]["message"]
    messages.append(assistant)
    for call in assistant.get("tool_calls") or []:
        name = call["function"]["name"]
        try:
            args = json.loads(call["function"].get("arguments") or "{}")
        except json.JSONDecodeError:
            args = {}
        result = invoke(name, args)
        trace.append({"tool": name, "args": args, "result": result})
        messages.append({"role": "tool", "tool_call_id": call["id"], "content": json.dumps(result, ensure_ascii=False)})
    if trace:
        final = post_chat(messages)["choices"][0]["message"].get("content") or ""
    else:
        final = assistant.get("content") or "模型未调用工具，不能视为已完成主机观测。"
    return final, trace


def mock_agent(scenario: str, target: str | None) -> tuple[str, list[dict[str, Any]]]:
    if scenario == "health":
        name, args = "host_health", {}
    elif scenario == "service":
        name, args = "service_status", {"name": target or "ssh.service"}
    else:
        name, args = "log_sizes", {"path": target or str(ALLOWED_LOG_ROOT), "limit": 5}
    result = invoke(name, args)
    return f"Mock executed read-only tool {name}; ok={result['ok']}", [{"tool": name, "args": args, "result": result}]


def save_report(summary: str, trace: list[dict[str, Any]]) -> Path:
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    report_dir, trace_dir = Path("reports"), Path("traces")
    report_dir.mkdir(exist_ok=True); trace_dir.mkdir(exist_ok=True)
    (trace_dir / f"{stamp}.json").write_text(json.dumps(trace, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    report = report_dir / f"{stamp}.md"
    report.write_text(f"# Read-only operations report\n\n{summary}\n\n## Tool evidence\n\n```json\n{json.dumps(trace, ensure_ascii=False, indent=2)}\n```\n", encoding="utf-8")
    return report


def self_check() -> None:
    assert invoke("service_status", {"name": "ssh;reboot"})["ok"] is False
    assert invoke("log_sizes", {"path": "/etc"})["ok"] is False
    assert invoke("safe_calc", {"expression": "__import__('os').system('id')"})["ok"] is False
    assert invoke("safe_calc", {"expression": "open('/etc/passwd').read()"})["ok"] is False
    assert invoke("safe_calc", {"expression": ""})["ok"] is False
    assert invoke("safe_calc", {"expression": "(80-17)/63*100"}) == {"ok": True, "data": {"expression": "(80-17)/63*100", "result": 100.0}}
    print("SELF_CHECK_OK: 6 checks passed")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--self-check", action="store_true")
    parser.add_argument("--mock", choices=("health", "service", "logs"))
    parser.add_argument("--target")
    parser.add_argument("--prompt", default="检查当前主机健康状态并生成只读报告")
    args = parser.parse_args()
    if args.self_check:
        self_check(); return
    summary, trace = mock_agent(args.mock, args.target) if args.mock else real_agent(args.prompt)
    report = save_report(summary, trace)
    print(json.dumps({"summary": summary, "tool_calls": len(trace), "report": str(report)}, ensure_ascii=False))


if __name__ == "__main__":
    main()

4.1 负向自检

python3 ops_agent.py --self-check

实际输出：

SELF_CHECK_OK: 6 checks passed

覆盖：服务名 Shell 元字符、越界路径、模块导入、主机文件读取、空计算全部拒绝；普通算术成功。

4.2 三个运维场景

每日健康报告

python3 ops_agent.py --mock health

预期形状：

{"summary":"Mock executed read-only tool host_health; ok=True","tool_calls":1,"report":"reports/<UTC时间>.md"}

验收：trace 包含 uptime 和根文件系统 df 的原始返回；命令失败必须原样保留。

服务分诊，不重启

python3 ops_agent.py --mock service --target ssh.service

程序只调用参数数组形式的 systemctl show；没有 restart/stop。服务不存在时保留退出码或 LoadState=not-found。

日志容量排查，不清理

python3 ops_agent.py --mock logs --target /var/log
OPS_LOG_ROOT=/var/log python3 -c 'import ops_agent; print(ops_agent.invoke("log_sizes", {"path":"/etc"}))'

第二条必须返回：

{'ok': False, 'error': 'ValueError', 'detail': 'path outside OPS_LOG_ROOT'}

工具只列白名单目录中的普通文件，不递归、不读内容、不删除。

5. Linux 部署 llama.cpp 与模型

5.1 编译 CPU 版

Debian/Ubuntu：

sudo apt-get update
sudo apt-get install -y git cmake build-essential libcurl4-openssl-dev
git clone --depth 1 https://github.com/ggml-org/llama.cpp.git
cd llama.cpp
cmake -B build -DCMAKE_BUILD_TYPE=Release
cmake --build build --config Release -j"$(nproc)"
sudo install -m 0755 build/bin/llama-server /usr/local/bin/llama-server
llama-server --version

正式环境应固定并登记验证过的 llama.cpp commit，不要长期跟踪最新 main。

5.2 前台试跑

llama-server   -hf LiquidAI/LFM2.5-2.6B-GGUF:Q4_K_M   --jinja --host 127.0.0.1 --port 8080 -c 8192   --temp 0.1 --top-k 50 --repeat-penalty 1.1

--jinja：启用模型模板中的工具调用。
CPU-only 不照抄 -ngl 99 与 -fa on；二者面向可用的 Metal/CUDA 构建。
只监听 loopback；不要暴露无鉴权模型端口。
首次运行会下载模型。隔离网应提前完成合规下载、恶意文件扫描与哈希登记。

另开终端：

curl --fail --silent http://127.0.0.1:8080/v1/models

HTTP 成功且含模型条目才算服务可用，只有端口监听不算。

5.3 systemd 持久化

sudo useradd --system --home /var/lib/lfm --create-home --shell /usr/sbin/nologin lfm

/etc/systemd/system/lfm-server.service：

[Unit]
Description=LFM2.5-2.6B local inference server
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=lfm
Group=lfm
Environment=HOME=/var/lib/lfm
Environment=HF_HOME=/var/lib/lfm/huggingface
ExecStart=/usr/local/bin/llama-server -hf LiquidAI/LFM2.5-2.6B-GGUF:Q4_K_M --jinja --host 127.0.0.1 --port 8080 -c 8192 --temp 0.1 --top-k 50 --repeat-penalty 1.1
Restart=on-failure
RestartSec=5
TimeoutStopSec=30
NoNewPrivileges=true
PrivateTmp=true
ProtectHome=true
ProtectSystem=strict
ReadWritePaths=/var/lib/lfm

[Install]
WantedBy=multi-user.target

先静态检查，再由维护窗口内人工启用：

sudo systemd-analyze verify /etc/systemd/system/lfm-server.service
sudo systemctl daemon-reload
sudo systemctl enable --now lfm-server.service
systemctl status lfm-server.service --no-pager
curl --fail --silent http://127.0.0.1:8080/v1/models

回滚：

sudo systemctl disable --now lfm-server.service
sudo rm /etc/systemd/system/lfm-server.service
sudo systemctl daemon-reload

/var/lib/lfm 含模型缓存，教程不自动删除。

6. 连接真实模型

cd ~/lfm-ops-agent
export MODEL_BASE_URL=http://127.0.0.1:8080/v1
export MODEL_NAME=LFM2.5-2.6B
export MODEL_API_KEY=local
export OPS_LOG_ROOT=/var/log
export OPS_COMMAND_TIMEOUT=5
python3 ops_agent.py --prompt '读取当前主机 uptime 和根文件系统使用率，基于工具证据生成简短巡检报告；不得执行任何修复。'

成功标准：tool_calls > 0；traces/*.json 中存在 host_health；报告数值与原始工具结果一致。模型只描述命令却未调用工具时，本次任务失败，不能通过扩大权限来补救。

7. 鉴权、网络与审计

单机：保持 127.0.0.1:8080，Agent 与模型同机。

企业内网：不要把 llama.cpp 直接绑定 0.0.0.0。采用：

运维客户端 → VPN/零信任入口 → HTTPS 反向代理（mTLS/企业 SSO）
                              → 127.0.0.1:8080 llama.cpp

最低要求：TLS、身份校验、来源 allowlist、请求大小/并发限制、脱敏访问日志；模型端口不直接暴露互联网。

审计：本例记录工具名、参数和原始结果。生产试点再补请求 ID、操作者、模型/量化版本、主机 ID、策略版本和人工处置结果。禁止记录密钥、Cookie、私钥、.env 或数据库连接串。

8. 失败处理

故障	判定	处理
`/v1/models` 不通	推理不可用	查 systemd 日志、内存、下载；不生成伪报告
模型不调用工具	未完成观测	检查 `--jinja`、tools 请求和提示词；权限不变
工具超时	数据缺失	trace 保留错误，报告明确“未取得数据”
服务名/路径越界	输入越界	拒绝，不临时扩大 allowlist
输出合理但无 trace	不可信	任务失败，不进入运维记录
内存不足	容量配置错误	降上下文或用 Q4KM，先不加并发

9. 安全与验收

没有通用 Shell 和任何写操作工具；
子进程均使用参数数组、超时与原始退出码；
服务名和路径经过校验；
工具失败不会被改写成成功；
模型 API 只监听 loopback；跨主机经过 TLS 和身份控制；
不读取秘密文件；报告与 trace 有权限和保留周期；
候选修复必须附风险、回滚、人工确认；
上线前完成 LFM Open License v1.0 法务检查。

最小验收：三个 Mock 场景各产生报告和 trace；六项负向检查通过；同秒任务不会覆盖；真实模型至少产生一次可核对的工具调用；无 trace 的回答不进入记录。

10. 分阶段上线

开发机 Mock：自检 + 三场景。
单台非生产机：真实 LFM 只读运行一周，与人工结果对比。
小批量内网：集中模型、受控反代、限制并发。
影子运行：不接告警/工单主链路，统计漏报、误报、工具失败率、人工采纳率。
候选建议：仅生成待审批命令，不执行。
写操作评审：只有真实数据证明收益后，才为单一动作设计独立权限、dry-run、审批和回滚；仍不添加通用 Shell。

微调也放在真实失败样本之后：先判断问题来自模型、工具 schema、Harness 还是数据缺失。

11. 本教程验证证据

static_publish_ok：true，Markdown/HTML 生成、目录、复制代码、远端非空与 SHA-256 一致性检查均通过；
mock_code_run_ok：true，已实际运行自检及 health/service/logs 三场景；
real_backend_contract_ok：true，脚本通过 Python 编译检查，并使用标准 OpenAI-compatible tools/tool_calls 消息形状；
real_backend_smoke_ok：false，未在目标读者的 LFM2.5-2.6B 实例上运行真实推理，厂商基准不能替代你的环境实测。

12. 刻意跳过

自动修复、RAG、向量库、多 Agent、微调、通用 Shell 和 Kubernetes 控制器都不是第一阶段必需。只读工具 + 本地模型 + 可核验 trace 足以验证价值；影子运行数据证明需要时再加。