250 行 Python 写一个 CLI AI Agent

$ agent "把今天的 nginx 499 日志按接口聚合一下"终端卡了两秒，然后它不是回答一堆废话，而是先扫日志、再跑一段本地统计、最后把可疑接口列出来。 这东西才像 CLI AI Agent。不是聊天框搬到命令行。

我一直不太喜欢那种“先设计一个宏大的 Agent 框架”的写法。真落地的时候，第一版不用复杂。250 行 Python 足够干三件事：

1. 接收命令行问题
2. 让模型判断要不要调用工具
3. 工具跑完以后，再把结果丢回模型整理

目录我一般就这么放：

mini-agent/
  agent.py
  tools.py
  config.json

config.json 也别整太花：

{
"api_base": "https://api.example.com/v1/chat/completions",
"api_key": "换成自己的 key",
"model": "your-model-name"
}

这个地方我会先写死一层薄封装。不要上来就 LangChain、插件市场、记忆系统。排障工具最怕还没干活，依赖先炸一屏。

# agent.py
import argparse
import json
import subprocess
import urllib.request
from pathlib import Path
ROOT = Path(__file__).parent
CONF = json.loads((ROOT / "config.json").read_text("utf-8"))
SYSTEM_PROMPT = """
你是一个命令行 AI Agent。
遇到需要本地信息的问题，可以选择调用工具。
只能返回 JSON，不要解释。
格式：
{"thought":"你的判断","tool":"工具名或 final","args":{}}
可用工具：
- shell: 执行只读命令，例如 ls、grep、cat、du、ps
- read_file: 读取文件
- final: 给用户最终答案
"""

这里有个细节，shell 工具必须收口。否则用户一句“帮我清理一下”，模型可能真给你跑 rm -rf。我现场写这种东西，第一版只允许读命令。

# tools.py
import subprocess
from pathlib import Path
SAFE_PREFIX = (
"ls", "cat", "grep", "head", "tail", "wc",
"du", "df", "ps", "find", "sed", "awk"
)
defrun_shell(cmd: str) -> str:
    first = cmd.strip().split()[0] if cmd.strip() else""
if first notin SAFE_PREFIX:
returnf"blocked command: {cmd}"
    p = subprocess.run(
        cmd,
        shell=True,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        timeout=8
    )
return p.stdout[-6000:]
defread_file(path: str) -> str:
    p = Path(path).expanduser()
ifnot p.exists():
returnf"file not found: {path}"
if p.is_dir():
returnf"{path} is a directory"
return p.read_text("utf-8", errors="ignore")[-8000:]

我第一眼会防两个东西：危险命令，超长输出。 Agent 很容易死在这里。模型一激动让你 find /，终端跑半天，最后上下文还被塞爆。

模型请求也没必要封装成 SDK。用标准库就够：

defcall_llm(messages):
    body = json.dumps({
"model": CONF["model"],
"messages": messages,
"temperature": 0.2
    }).encode("utf-8")
    req = urllib.request.Request(
        CONF["api_base"],
        data=body,
        headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {CONF['api_key']}"
        }
    )
with urllib.request.urlopen(req, timeout=30) as resp:
        data = json.loads(resp.read().decode("utf-8"))
return data["choices"][0]["message"]["content"]

主循环是整个 Agent 的心脏。别写复杂，就是模型决策、工具执行、结果回填。

from tools import run_shell, read_file
defparse_json(text):
try:
return json.loads(text)
except Exception:
        left = text.find("{")
        right = text.rfind("}")
if left >= 0and right > left:
return json.loads(text[left:right + 1])
raise
defrun_agent(user_input: str):
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_input}
    ]
for step in range(6):
        raw = call_llm(messages)
        action = parse_json(raw)
        tool = action.get("tool")
        args = action.get("args", {})
if tool == "final":
return args.get("answer", raw)
if tool == "shell":
            result = run_shell(args.get("cmd", ""))
elif tool == "read_file":
            result = read_file(args.get("path", ""))
else:
            result = f"unknown tool: {tool}"
        messages.append({"role": "assistant", "content": raw})
        messages.append({
"role": "tool",
"content": result
        })
return"执行步数太多，已停止。可以把问题缩小一点再跑。"

这个循环我一般限制 6 步。不是舍不得，是 Agent 很容易绕圈。 比如它先 ls logs，再 tail app.log，再 grep ERROR，这还正常。要是第 5 步还在翻目录，多半是问题描述太散，继续跑也不会聪明多少。

命令行入口就几行：

defmain():
    parser = argparse.ArgumentParser()
    parser.add_argument("question", nargs="+")
    args = parser.parse_args()
    question = " ".join(args.question)
    answer = run_agent(question)
    print(answer)
if __name__ == "__main__":
    main()

跑一下：

python agent.py "看一下 logs/app.log 里最近 20 条 ERROR，判断是不是数据库连接池问题"

模型理想情况下会先决策：

{
"thought": "需要先读取错误日志",
"tool": "shell",
"args": {
"cmd": "tail -n 500 logs/app.log | grep ERROR | tail -n 20"
  }
}

工具结果回来后，它再总结：

最近错误集中在 HikariPool timeout，像是连接池耗尽。
优先看两个点：
1. 是否有慢 SQL 长时间占用连接
2. 是否有连接泄漏
可以继续 grep connection is not available 前后的 traceId。

这就够用了。

后面真要增强，我不会先加“长期记忆”，那玩意十个 Agent 九个用不上。 我会先加三个小东西。

第一个，命令执行确认。只读命令直接跑，疑似危险命令必须人工确认。

第二个，工具白名单拆细。不要一个 shell 管天下，日志分析就给 grep_log，文件读取就给 read_file，进程查看就给 list_process。工具越窄，Agent 越稳。

第三个，把每次执行过程落盘。

defaudit(step, action, result):
    line = json.dumps({
"step": step,
"action": action,
"result_preview": result[:300]
    }, ensure_ascii=False)
with open("agent.audit.log", "a", encoding="utf-8") as f:
        f.write(line + "\n")

出了问题能复盘。 不然你只看到最后一句回答，根本不知道它中间跑了什么命令、读了哪个文件、是不是被一段脏日志带偏了。

250 行 Python 写 CLI AI Agent，重点不在“250 行”。重点是别把它写成玩具聊天框。 能拿本地信息，能调用小工具，能控制风险，能留下痕迹，这东西才开始有点用。