手写一个 AI Agent：从零理解它的原理和本质

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

你: "我的机器有几个 CPU 核心？"
                  │
                  ▼
LLM（收到问题，开始推理）:
  "用户想知道 CPU 核心数 -> 用 nproc 命令"
                  │
                  ▼
Agent（解析 LLM 输出 → 发现 Action）:
  执行: nproc
                  │
                  ▼
命令行返回: "8"
                  │
                  ▼
LLM（看到结果，继续推理）:
  "8 个核心 -> 可以回答了"
                  │
                  ▼
你收到: "你的机器有 8 个 CPU 核心 ✅"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

class LLMClient:
    def __init__(self):
        load_dotenv()
        self.client = OpenAI(
            api_key=os.getenv("DEEPSEEK_API_KEY"),
            base_url=os.getenv("DEEPSEEK_BASE_URL"),
        )

    def chat(self, messages) -> str:
        response = self.client.chat.completions.create(
            model="deepseek-v4-flash",
            messages=messages,
            temperature=0.0,
        )
        return response.choices[0].message.content or ""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

class CLIExecutor:
    DANGEROUS_PATTERNS = [
        r"\bsudo\b", r"\bmkfs\b", r"\brm\s+-rf\s+/$", ...
    ]

    def run(self, command: str) -> str:
        # 1. 安全检查：拒绝危险命令
        for pattern in self.DANGEROUS_PATTERNS:
            if re.search(pattern, command, re.IGNORECASE):
                return f"[Safety Error] 危险命令被拦截"

        # 2. 执行命令（带超时）
        result = subprocess.run(
            command, shell=True,
            capture_output=True, text=True,
            timeout=30,
        )

        # 3. 处理输出（截断、报错）
        output = result.stdout + result.stderr
        if result.returncode != 0:
            output = f"[Exit code: {result.returncode}]\n{output}"
        return output[:3000]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

可用工具:
- run_command: 执行 shell 命令

输出格式（二选一）:

=== 要执行命令 ===
Thought: <你的推理>
Action: run_command
Action Input: <命令>

=== 要给出答案 ===
Final Answer: <你的回答>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def parse_response(response: str) -> dict:
    # 是 Final Answer 吗？
    if re.search(r"Final Answer:\s*(.*)", response, re.DOTALL):
        return {"type": "final", "content": "..."}

    # 是 Action 吗？
    if re.search(r"Action:\s*run_command\s*Action Input:\s*(.*)", response):
        return {"type": "action", "command": "..."}

    # 都不是 → 还在思考
    return {"type": "thought", "content": "..."}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

def run_agent(user_input: str):
    llm = LLMClient()
    executor = CLIExecutor()

    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_input},
    ]

    for turn in range(1, 11):  # 最多 10 轮
        # 1. 调 LLM
        response = llm.chat(messages)
        parsed = parse_response(response)

        # 2. 根据解析结果做不同的事
        if parsed["type"] == "final":
            print(parsed["content"])
            return

        if parsed["type"] == "action":
            observation = executor.run(parsed["command"])
            # 把"思考"和"观察结果"都追加到消息历史
            messages.append({"role": "assistant", "content": response})
            messages.append({"role": "user", "content": observation})
            # 继续下一轮循环

        if parsed["type"] == "thought":
            # 还没给出明确指令，推它一把
            messages.append({"role": "user", "content": "请给出 Action 或 Final Answer。"})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

用户输入
    │
    ▼
┌─────────────────────────────┐
│  给 LLM 发送消息历史          │
│  (system + 之前的对话)        │
└──────────┬──────────────────┘
           │
           ▼
┌─────────────────────────────┐
│  LLM 返回文本                 │
│  "Thought... Action: xxx"    │
└──────────┬──────────────────┘
           │
    ┌──────┴──────┐
    ▼              ▼
 Final?         Action?
    │              │
    ▼              ▼
 输出答案     执行命令 → 拿到结果
                             │
                  ┌──────────┘
                  ▼
            追加到消息历史
                  │
                  ▼
            回到 LLM（继续循环）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

---
name: code-review
description: 审查代码质量，检查 bug、安全漏洞和代码风格
---

你是一个资深代码审查专家。请从以下维度审查用户提供的代码：
1. 逻辑正确性
2. 安全漏洞
3. 代码风格
4. 性能问题

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

User: "帮我审查这段代码"
                      │
                      ▼
LLM（看到 system prompt 里的 skill 列表）:
  "这需要 code-review skill"
  Action: load_skill
  Action Input: code-review
                      │
                      ▼
Agent（加载 skill.md 文件）:
  把 skill 的 prompt 追加为 system message
                      │
                      ▼
LLM（现在拥有了审查专家的指令）:
  开始按 skill 要求工作
  Action: run_command
  Action Input: cat file.py
  ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// 不用 Action/Final Answer，用 XML
<tool_call>
  <name>search_web</name>
  <args>{"q": "weather in Beijing"}</args>
</tool_call>

// 或者用 JSON
{"tool": "search_web", "params": {"q": "weather"}}

// 或者用任意你喜欢的格式
→ SEARCH: weather in Beijing ←

1
2
3
4
5
6
7
8
9

# 1. 配置 API Key
cp .env.example .env
# 编辑 .env，填入 DEEPSEEK_API_KEY

# 2. 安装依赖
uv sync

# 3. 运行
uv run python agent.py