5 分钟跑通第一个 Agent

你的电脑上已经装了 Python 3.10+ 和 pip。如果没有，先去 python.org 下载安装，装完在终端跑 python --version 确认版本号大于等于 3.10。还要准备一个 LLM API key——用 OpenAI、Anthropic 或任何兼容 OpenAI 格式的都可以。如果你手头没有，用 DeepSeek 或智谱的免费额度也行，后面我们会看到怎么切换。

先搞清楚我们要做什么

很多人第一次接触 Agent 时，会被各种框架吓住——LangChain、AutoGPT、CrewAI，每个都有一堆概念要学。但 Agent 最核心的东西其实很简单：一个循环，让模型能看、能想、能做。

这一章我们不用任何框架，只用 30 行 Bash 脚本，搭出一个能真正干活的 Agent。它接收一个任务描述，自己决定要做什么，然后执行。整个过程不超过 5 分钟。

第一步：准备你的 API 密钥

把 API key 设成环境变量，这样脚本里可以直接读取，不用硬编码在文件里。

export OPENAI_API_KEY="sk-your-key-here"

如果你用的是兼容 OpenAI 格式的其他服务商（比如 DeepSeek、智谱、Groq），还需要设一个自定义的 API 地址：

export OPENAI_BASE_URL="https://api.deepseek.com/v1"

预期结果：运行 echo $OPENAI_API_KEY 能看到你的 key，而不是空行。

第二步：写一个最简单的 LLM 调用

创建一个文件叫 agent.sh，用你喜欢的编辑器打开。先写一个最基础的函数，能跟 LLM 对话：

#!/bin/bash

# 调用 LLM 并返回回复内容
call_llm() {
    local system_prompt="$1"
    local user_message="$2"
    
    curl -s "$OPENAI_BASE_URL/chat/completions" \
        -H "Content-Type: application/json" \
        -H "Authorization: Bearer $OPENAI_API_KEY" \
        -d "$(cat <<EOF
{
    "model": "gpt-4o-mini",
    "messages": [
        {"role": "system", "content": "$system_prompt"},
        {"role": "user", "content": "$user_message"}
    ],
    "temperature": 0
}
EOF
    )" | jq -r '.choices[0].message.content'
}

这里有几个要点：

• temperature: 0 让模型输出尽可能确定，适合工具调用场景
• 用 jq 解析 JSON 响应——如果你的系统没有 jq，用 brew install jq 或 apt install jq 装一下
• $OPENAI_BASE_URL 默认是 OpenAI 的地址，如果你没设这个变量，脚本会报错。我们后面会处理这个默认值

预期结果：先别急着跑，这个函数还没被调用。我们继续往下搭。

第三步：让 Agent 能「思考」和「行动」

Agent 的核心循环是：收到任务 → 让模型决定下一步 → 执行 → 把结果反馈给模型 → 重复直到完成。

我们给 Agent 一个简单的工具：执行 Shell 命令。模型会输出一个特殊格式的指令，我们解析它并执行。

# Agent 主循环
agent_loop() {
    local task="$1"
    local max_steps=10
    local step=0
    local messages='[]'
    
    # 系统提示词：告诉模型它能做什么
    local system_prompt='你是一个能执行 Shell 命令的 AI 助手。
当你需要执行命令时，请用以下格式输出：

```bash
你的命令

如果你认为任务已经完成，请输出：

完成：任务已完成的说明

请直接输出，不要有多余的思考过程。'

# 把用户任务加入消息列表
messages=$(echo "$messages" | jq --arg task "$task" '. + [{"role": "user", "content": $task}]')

while [ $step -lt $max_steps ]; do
    step=$((step + 1))
    echo "=== 步骤 $step ==="
    
    # 调用 LLM
    local response=$(call_llm "$system_prompt" "$(echo "$messages" | jq -r '.[-1].content')")
    echo "模型回复：$response"
    
    # 检查是否包含 bash 代码块
    if echo "$response" | grep -q '```bash'; then
        # 提取命令
        local cmd=$(echo "$response" | sed -n '/```bash/,/```/p' | sed '1d;$d')
        echo "执行命令：$cmd"
        
        # 执行命令并捕获输出
        local output=$(eval "$cmd" 2>&1)
        echo "命令输出：$output"
        
        # 把结果反馈给模型
        messages=$(echo "$messages" | jq \
            --arg response "$response" \
            --arg output "$output" \
            '. + [{"role": "assistant", "content": $response}, {"role": "user", "content": "命令执行结果：\n" + $output}]')
    elif echo "$response" | grep -q '^完成：'; then
        echo "任务完成！"
        echo "$response" | sed 's/^完成：//'
        return 0
    else
        echo "模型没有输出有效指令，尝试继续..."
        messages=$(echo "$messages" | jq \
            --arg response "$response" \
            '. + [{"role": "assistant", "content": $response}]')
    fi
done

echo "达到最大步骤数 $max_steps，任务可能未完成。"
return 1

}

这段代码做了几件事：
1. 把用户的任务放进消息列表
2. 循环调用 LLM，每次最多 10 步防止死循环
3. 检查模型回复里有没有 ` ```bash ` 代码块——有就提取并执行
4. 把执行结果追加到消息列表，让模型知道刚才的命令发生了什么
5. 如果模型输出 `完成：` 开头的内容，就结束循环

**常见问题**：如果模型输出的代码块格式不对（比如用了 ` ```shell ` 而不是 ` ```bash `），我们的解析会失败。你可以扩展 grep 匹配更多格式，或者让模型输出更规范的格式。这里为了简单，我们只匹配一种。

## 第四步：加上入口和默认值

在脚本末尾加上主入口，同时处理 API 地址的默认值：

```bash
# 设置默认 API 地址
if [ -z "$OPENAI_BASE_URL" ]; then
    export OPENAI_BASE_URL="https://api.openai.com/v1"
fi

# 检查 API key
if [ -z "$OPENAI_API_KEY" ]; then
    echo "错误：请设置 OPENAI_API_KEY 环境变量"
    exit 1
fi

# 检查 jq
if ! command -v jq &> /dev/null; then
    echo "错误：需要 jq 来解析 JSON，请先安装"
    exit 1
fi

# 运行 Agent
if [ $# -eq 0 ]; then
    echo "用法：./agent.sh <任务描述>"
    echo "示例：./agent.sh '列出当前目录下最大的三个文件'"
    exit 1
fi

agent_loop "$*"

预期结果：现在整个脚本应该能跑了。先给它一个简单的任务试试。

第五步：跑起来

给脚本执行权限，然后运行：

chmod +x agent.sh
./agent.sh "列出当前目录下所有的 Python 文件，并统计总行数"

你会看到类似这样的输出：

=== 步骤 1 ===
模型回复：```bash
find . -name "*.py" -type f

执行命令：find . -name "*.py" -type f 命令输出：./agent.py ./utils.py ./main.py

=== 步骤 2 === 模型回复：```bash wc -l *.py

执行命令：wc -l *.py
命令输出：      23 agent.py
      56 utils.py
     102 main.py
     181 total

=== 步骤 3 ===
模型回复：完成：当前目录下有 3 个 Python 文件，总行数为 181 行。
任务完成！
当前目录下有 3 个 Python 文件，总行数为 181 行。

如果报错：

• curl: (6) Could not resolve host：检查网络连接和 OPENAI_BASE_URL 是否正确
• jq: error：检查 API 返回的 JSON 格式，可能是 key 无效或额度用完了
• command not found: jq：按前面说的安装 jq

让它做点更有意思的事

试试这些任务，感受 Agent 的自主性：

./agent.sh "创建一个叫 test_project 的目录，在里面初始化一个 Git 仓库，创建一个 README.md 文件，内容写'这是我的第一个 Agent 项目'，然后提交"

./agent.sh "下载 https://example.com 的首页内容，保存到 example.html，然后统计它有多少行"

注意第二个任务需要你的机器能访问外网。如果被墙了，换个国内能访问的网址。

完成后你已经得到了...

一个能自主执行 Shell 命令的 Agent。虽然简陋，但它具备了 Agent 最核心的三个要素：

1. 感知：通过命令输出了解环境状态
2. 推理：LLM 根据任务和反馈决定下一步
3. 行动：执行 Shell 命令改变环境

这个脚本只有 60 行左右，但你已经跑通了 Agent 的基本循环。下一章我们会给它装上更多「手」——文件读写、更精细的工具控制。但现在你已经有了一个能用的东西，可以拿它做自动化测试、批量文件处理、甚至简单的部署脚本。

记住这个模式：模型决定做什么，脚本负责执行。后面所有复杂的功能都是在这个基础上加东西。

第 2 章：给 Agent 装上「手」——实现文件读写与 Shell 执行工具

上一章我们用 Bash 脚本调通了 LLM，但那个 Agent 只能「想」，不能「做」。它回答了你问题，却没法帮你改代码、读文件、跑命令。这一章我们就给它装上两只手：文件读写和 Shell 执行。

装好之后，你的 Agent 就能做这样的事：

你：帮我看看 src/main.py 里有没有语法错误，然后修好它
Agent：读取 src/main.py → 发现第 15 行少了个括号 → 执行 sed 修复 → 运行 python src/main.py 验证

前置条件

• 已完成第 1 章，有一个能调通 LLM 的 Bash 脚本（我们叫它 agent.sh）
• 你的脚本里已经配置好了 API key 和模型名称
• 当前工作目录下有一个测试用的 Python 文件，比如 test.py，内容随意

第一步：设计工具接口

工具的本质是一个函数：输入参数，输出结果。为了让 LLM 能调用工具，我们需要把工具描述成它看得懂的格式。

打开你的 agent.sh，在调用 API 之前，先定义两个工具。我们用 JSON 来描述它们：

# 工具定义
TOOLS='[
  {
    "name": "read_file",
    "description": "读取文件内容，返回文件文本",
    "parameters": {
      "type": "object",
      "properties": {
        "path": {
          "type": "string",
          "description": "文件路径"
        }
      },
      "required": ["path"]
    }
  },
  {
    "name": "write_file",
    "description": "写入内容到文件，如果文件不存在则创建",
    "parameters": {
      "type": "object",
      "properties": {
        "path": {
          "type": "string",
          "description": "文件路径"
        },
        "content": {
          "type": "string",
          "description": "要写入的内容"
        }
      },
      "required": ["path", "content"]
    }
  },
  {
    "name": "run_shell",
    "description": "执行一条 Shell 命令并返回输出",
    "parameters": {
      "type": "object",
      "properties": {
        "command": {
          "type": "string",
          "description": "要执行的命令"
        }
      },
      "required": ["command"]
    }
  }
]'

预期结果：你定义了一个 JSON 数组，里面有三个工具。每个工具都有 name、description 和 parameters。description 要写得足够清楚，让 LLM 知道什么时候该用这个工具。

第二步：把工具传给 LLM

现在修改你的 API 调用，把 TOOLS 传进去。以 OpenAI 兼容 API 为例，在 curl 命令里加上 tools 参数：

# 构建消息
MESSAGES='[
  {"role": "user", "content": "读取 test.py 的内容"}
]'

# 调用 API，传入工具
RESPONSE=$(curl -s https://api.openai.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -d "$(jq -n \
    --arg model "$MODEL" \
    --argjson messages "$MESSAGES" \
    --argjson tools "$TOOLS" \
    '{model: $model, messages: $messages, tools: $tools}')")

预期结果：API 返回的响应里会多出一个 tool_calls 字段。LLM 看到用户说「读取 test.py 的内容」，就会决定调用 read_file 工具，并填好 path 参数。

第三步：解析工具调用

API 返回的响应结构大概是这样的：

{
  "choices": [{
    "message": {
      "role": "assistant",
      "content": null,
      "tool_calls": [{
        "id": "call_xxx",
        "type": "function",
        "function": {
          "name": "read_file",
          "arguments": "{\"path\": \"test.py\"}"
        }
      }]
    }
  }]
}

我们需要从响应里提取出工具名称和参数。用 jq 来解析：

# 提取工具调用
TOOL_CALLS=$(echo "$RESPONSE" | jq -c '.choices[0].message.tool_calls[]?')

# 如果没有工具调用，直接输出文本回复
if [ -z "$TOOL_CALLS" ]; then
  echo "$RESPONSE" | jq -r '.choices[0].message.content'
  exit 0
fi

# 遍历每个工具调用
echo "$TOOL_CALLS" | while read -r call; do
  TOOL_NAME=$(echo "$call" | jq -r '.function.name')
  TOOL_ARGS=$(echo "$call" | jq -r '.function.arguments')
  
  echo "调用工具: $TOOL_NAME"
  echo "参数: $TOOL_ARGS"
done

预期结果：运行脚本后，你会看到控制台打印出「调用工具: read_file」和「参数: {"path": "test.py"}」。这说明 LLM 正确识别了意图并选择了工具。

第四步：实现工具函数

现在写真正的工具逻辑。在脚本里添加三个函数：

# 读取文件
read_file() {
  local path="$1"
  if [ ! -f "$path" ]; then
    echo "错误: 文件 $path 不存在"
    return 1
  fi
  cat "$path"
}

# 写入文件
write_file() {
  local path="$1"
  local content="$2"
  echo "$content" > "$path"
  echo "成功写入 $path"
}

# 执行 Shell 命令
run_shell() {
  local command="$1"
  eval "$command" 2>&1
}

注意：eval 有安全风险，生产环境应该用更严格的执行方式。这里为了教学简单，先用 eval。

第五步：执行工具并返回结果

把工具调用和实现连接起来。修改循环部分：

echo "$TOOL_CALLS" | while read -r call; do
  TOOL_NAME=$(echo "$call" | jq -r '.function.name')
  TOOL_ARGS=$(echo "$call" | jq -r '.function.arguments')
  
  # 提取参数
  PATH_ARG=$(echo "$TOOL_ARGS" | jq -r '.path // empty')
  CONTENT_ARG=$(echo "$TOOL_ARGS" | jq -r '.content // empty')
  COMMAND_ARG=$(echo "$TOOL_ARGS" | jq -r '.command // empty')
  
  # 执行对应工具
  case "$TOOL_NAME" in
    "read_file")
      RESULT=$(read_file "$PATH_ARG")
      ;;
    "write_file")
      RESULT=$(write_file "$PATH_ARG" "$CONTENT_ARG")
      ;;
    "run_shell")
      RESULT=$(run_shell "$COMMAND_ARG")
      ;;
    *)
      RESULT="未知工具: $TOOL_NAME"
      ;;
  esac
  
  echo "工具执行结果: $RESULT"
done

预期结果：现在运行脚本，它会读取 test.py 并打印内容。如果文件不存在，会打印错误信息。

第六步：把结果送回 LLM 继续对话

工具执行完了，但 LLM 还不知道结果。我们需要把结果作为新的消息发回去，让 LLM 继续推理。

完整的流程应该是：

1. 用户提问
2. LLM 决定调用工具
3. 我们执行工具
4. 把结果发给 LLM
5. LLM 根据结果给出最终回答

修改脚本，加入循环：

# 构建初始消息
MESSAGES='[{"role": "user", "content": "读取 test.py 的内容，然后在文件末尾加一行 # 注释"}]'

# 循环直到 LLM 不再调用工具
while true; do
  RESPONSE=$(curl -s https://api.openai.com/v1/chat/completions \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer $OPENAI_API_KEY" \
    -d "$(jq -n \
      --arg model "$MODEL" \
      --argjson messages "$MESSAGES" \
      --argjson tools "$TOOLS" \
      '{model: $model, messages: $messages, tools: $tools}')")
  
  # 检查是否有工具调用
  TOOL_CALLS=$(echo "$RESPONSE" | jq -c '.choices[0].message.tool_calls[]?')
  
  if [ -z "$TOOL_CALLS" ]; then
    # 没有工具调用，输出最终回答
    echo "$RESPONSE" | jq -r '.choices[0].message.content'
    break
  fi
  
  # 把 assistant 的消息加入对话历史
  ASSISTANT_MSG=$(echo "$RESPONSE" | jq '.choices[0].message')
  MESSAGES=$(echo "$MESSAGES" | jq --argjson msg "$ASSISTANT_MSG" '. + [$msg]')
  
  # 执行每个工具调用
  echo "$TOOL_CALLS" | while read -r call; do
    TOOL_NAME=$(echo "$call" | jq -r '.function.name')
    TOOL_ARGS=$(echo "$call" | jq -r '.function.arguments')
    
    PATH_ARG=$(echo "$TOOL_ARGS" | jq -r '.path // empty')
    CONTENT_ARG=$(echo "$TOOL_ARGS" | jq -r '.content // empty')
    COMMAND_ARG=$(echo "$TOOL_ARGS" | jq -r '.command // empty')
    
    case "$TOOL_NAME" in
      "read_file") RESULT=$(read_file "$PATH_ARG") ;;
      "write_file") RESULT=$(write_file "$PATH_ARG" "$CONTENT_ARG") ;;
      "run_shell") RESULT=$(run_shell "$COMMAND_ARG") ;;
      *) RESULT="未知工具" ;;
    esac
    
    # 把工具结果作为新的消息加入对话
    TOOL_RESULT_MSG=$(jq -n \
      --arg role "tool" \
      --arg content "$RESULT" \
      --arg tool_call_id "$(echo "$call" | jq -r '.id')" \
      '{role: $role, content: $content, tool_call_id: $tool_call_id}')
    
    MESSAGES=$(echo "$MESSAGES" | jq --argjson msg "$TOOL_RESULT_MSG" '. + [$msg]')
  done
done

预期结果：运行脚本，输入「读取 test.py 的内容，然后在文件末尾加一行 # 注释」。你会看到：

1. LLM 先调用 read_file 读取文件
2. 拿到内容后，调用 write_file 写入新内容
3. 最后 LLM 告诉你「已完成」

常见问题

Q: API 返回 400 错误，说 tools 格式不对 A: 检查你的 TOOLS JSON 是否合法。可以用 echo "$TOOLS" | jq . 验证。常见错误是参数类型写成了 "string" 而不是 "string"（少引号），或者 required 字段拼写错误。

Q: LLM 不调用工具，直接返回文本 A: 检查 description 是否足够清晰。如果 LLM 觉得可以直接回答，它就不会调用工具。可以试试更明确的指令，比如「请使用 read_file 工具读取 test.py」。

Q: 工具执行结果太长，API 报错 A: 工具结果会被塞进消息里，如果结果太大（比如读取了一个 10MB 的文件），会超过 API 的 token 限制。可以在工具函数里加截断逻辑，比如只返回前 1000 个字符。

小试牛刀

创建一个 test.py：

def hello():
    print("Hello, Agent!")

hello()

然后运行你的 Agent，输入：

读取 test.py，然后执行它，最后在文件末尾加上一行 # Done

如果一切正常，你会看到 Agent 依次调用 read_file → run_shell → write_file，最终 test.py 末尾多了一行 # Done。

---

完成后你已经得到了： 一个能读文件、写文件、执行 Shell 命令的 Agent。它不再只是一个聊天机器人，而是能真正操作你电脑的工具。下一章我们会给它装上「眼睛」，让它能看懂 Git 变更和错误日志。

让 Agent 会「看」：接入 Git diff 和错误日志作为观察输入

一个只会执行命令的 Agent，就像蒙着眼睛修车——能摸到零件，但不知道哪里出了问题。要让 Agent 真正「看见」它工作的环境，我们需要给它装上眼睛：Git diff 让它看到代码的变化，错误日志让它看到运行时的异常。

这一章，我们给上一章写好的 Agent 加上两个观察能力：读取 Git 工作区的变更差异，以及捕获命令执行后的错误输出。做完之后，你的 Agent 就能根据「看到了什么」来决定「下一步做什么」。

前置条件

• 你已经完成了第 2 章，Agent 能执行 Shell 命令和读写文件
• 当前目录是一个 Git 仓库（如果不是，先 git init）
• 有一个可用的 LLM API key（比如 Claude 或 GPT）

第一步：让 Agent 能读取 Git diff

Git diff 是开发者最常用的「发生了什么变化」的观察方式。我们先写一个函数，让 Agent 能调用它来获取当前工作区的变更。

import subprocess
import json

def get_git_diff():
    """获取当前 Git 工作区的变更差异"""
    try:
        result = subprocess.run(
            ["git", "diff", "--unified=3"],
            capture_output=True,
            text=True,
            timeout=10
        )
        if result.returncode == 0 and result.stdout:
            return {"status": "success", "diff": result.stdout}
        else:
            # 如果没有变更，尝试获取未跟踪的文件
            untracked = subprocess.run(
                ["git", "ls-files", "--others", "--exclude-standard"],
                capture_output=True,
                text=True
            )
            if untracked.stdout:
                return {"status": "success", "diff": f"New untracked files:\n{untracked.stdout}"}
            return {"status": "no_changes", "diff": ""}
    except subprocess.TimeoutExpired:
        return {"status": "error", "message": "Git diff timed out"}
    except FileNotFoundError:
        return {"status": "error", "message": "Git not found in PATH"}

预期结果：运行 get_git_diff() 会返回一个字典，包含 diff 内容或状态说明。如果仓库没有变更，会检查是否有未跟踪的新文件。

常见问题：如果你在非 Git 目录运行，会报 fatal: not a git repository。我们稍后会处理这个边界情况。

第二步：让 Agent 能捕获错误日志

Shell 命令执行后，标准错误（stderr）和标准输出（stdout）都可能包含关键信息。我们改造一下第 2 章的 Shell 执行函数，让它同时返回两者。

def run_command(command, timeout=30):
    """执行命令并返回 stdout 和 stderr"""
    try:
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=timeout
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "return_code": result.returncode,
            "success": result.returncode == 0
        }
    except subprocess.TimeoutExpired:
        return {
            "stdout": "",
            "stderr": f"Command timed out after {timeout} seconds",
            "return_code": -1,
            "success": False
        }

预期结果：现在每次执行命令，Agent 都能看到标准输出和标准错误。如果命令失败，stderr 里通常会有错误信息。

小技巧：return_code 是判断命令是否成功的黄金标准。0 表示成功，非 0 表示失败。很多新手只看 stdout 而忽略 stderr，导致 Agent 以为命令成功了。

第三步：把观察结果喂给 LLM

有了观察能力，我们需要让 LLM 能「看到」这些信息。我们写一个简单的 Agent 循环，把观察结果作为上下文传给模型。

import requests
import os

API_KEY = os.environ.get("LLM_API_KEY")
API_URL = "https://api.anthropic.com/v1/messages"  # 以 Claude API 为例

def call_llm(messages):
    """调用 LLM 并返回回复"""
    headers = {
        "x-api-key": API_KEY,
        "anthropic-version": "2023-06-01",
        "content-type": "application/json"
    }
    data = {
        "model": "claude-3-5-sonnet-20241022",
        "max_tokens": 1024,
        "messages": messages
    }
    response = requests.post(API_URL, headers=headers, json=data)
    response.raise_for_status()
    return response.json()["content"][0]["text"]

def agent_loop(task):
    """Agent 主循环：观察 -> 思考 -> 行动"""
    messages = [
        {"role": "user", "content": f"你的任务是：{task}\n\n你可以使用以下工具：\n1. get_git_diff() - 查看代码变更\n2. run_command(command) - 执行 Shell 命令\n3. read_file(path) - 读取文件\n4. write_file(path, content) - 写入文件\n\n请一步步执行，每次只调用一个工具。"}
    ]
    
    max_steps = 5
    for step in range(max_steps):
        print(f"\n--- 步骤 {step + 1} ---")
        
        # 让 LLM 决定下一步做什么
        response = call_llm(messages)
        print(f"Agent 思考：{response}")
        
        # 解析 LLM 的回复，提取工具调用
        # 这里简化处理：如果回复包含特定关键词，就执行对应工具
        if "git diff" in response.lower():
            obs = get_git_diff()
            messages.append({"role": "assistant", "content": f"我查看了 Git diff，结果：{json.dumps(obs)}"})
        elif "run_command" in response.lower() or "执行" in response:
            # 从回复中提取命令（简化处理）
            command = response.split("`")[1] if "`" in response else response
            obs = run_command(command)
            messages.append({"role": "assistant", "content": f"我执行了命令，结果：stdout={obs['stdout']}, stderr={obs['stderr']}, 返回码={obs['return_code']}"})
        else:
            # 如果 LLM 说任务完成了，就退出
            if "完成" in response or "done" in response.lower():
                print("Agent 认为任务已完成")
                break
            messages.append({"role": "assistant", "content": response})
    
    return messages

预期结果：运行 agent_loop("检查当前代码仓库有什么变更，如果有错误就修复它")，Agent 会先调用 get_git_diff() 查看变更，然后根据看到的内容决定下一步。

注意：上面的解析逻辑非常简陋，实际项目中你会用更结构化的方式（比如 JSON 格式的工具调用）。这里只是为了演示观察-行动循环。

第四步：一个完整的实战例子

让我们用一个真实场景串起来：假设你刚改了一个 Python 文件，但运行报错了。Agent 需要先看 diff 知道改了哪里，然后运行程序看错误，最后修复。

# 模拟场景：先创建一个有 bug 的文件
write_file("hello.py", """
def greet(name):
    print("Hello, " + name)

greet("World")
greet(42)  # 这里会报错
""")

# 然后让 Agent 处理
result = agent_loop("""
1. 查看当前 Git diff，了解代码变更
2. 运行 hello.py，捕获错误
3. 根据错误修复代码
4. 再次运行确认修复成功
""")

print("\n最终对话记录：")
for msg in result:
    print(f"{msg['role']}: {msg['content'][:100]}...")

预期流程：

1. Agent 调用 get_git_diff()，看到新增了 hello.py
2. Agent 调用 run_command("python hello.py")，看到 stderr 里有 TypeError: can only concatenate str (not "int") to str
3. Agent 根据错误信息，调用 read_file("hello.py") 查看内容
4. Agent 决定修改代码，调用 write_file() 把 greet(42) 改成 greet("42")
5. Agent 再次运行，确认没有错误

第五步：处理边界情况

观察能力在理想情况下工作得很好，但实际中会遇到各种边界情况。我们加一些保护：

def safe_get_git_diff():
    """安全的 Git diff 获取，处理非 Git 目录"""
    # 先检查是否在 Git 仓库中
    check = subprocess.run(
        ["git", "rev-parse", "--git-dir"],
        capture_output=True,
        text=True,
        timeout=5
    )
    if check.returncode != 0:
        return {"status": "not_a_git_repo", "diff": ""}
    
    return get_git_diff()

def safe_run_command(command):
    """安全的命令执行，限制危险操作"""
    dangerous = ["rm -rf", "sudo", "> /dev/sda", "dd if="]
    for pattern in dangerous:
        if pattern in command:
            return {
                "stdout": "",
                "stderr": f"Blocked dangerous command pattern: {pattern}",
                "return_code": -1,
                "success": False
            }
    return run_command(command)

预期结果：在非 Git 目录下，safe_get_git_diff() 不会报错，而是返回友好的状态信息。safe_run_command() 会阻止明显危险的命令。

阶段性小结

完成后你已经得到了：

• 一个能读取 Git diff 的观察工具，让 Agent 看到代码变更
• 一个能捕获标准错误输出的命令执行器，让 Agent 看到运行错误
• 一个简单的观察-行动循环，把「看」和「做」连起来
• 边界情况处理，让观察能力更健壮

现在你的 Agent 不再是蒙着眼睛修车了——它能先看看哪里出了问题，再动手修理。下一章，我们会给 Agent 装上「记忆」，让它能记住之前看到的东西，不会在长对话中迷失。

为什么 Agent 会「失忆」

你有没有遇到过这种情况：跟 Agent 聊了十几轮，它开始忘记你十分钟前说过的话。你让它改一个函数，它改了，但改完又把之前修好的 bug 重新引入。你问它"刚才那个错误日志里说了什么"，它一脸茫然。

这不是 Agent 笨，是它的上下文窗口满了。

每个 LLM 都有固定的上下文长度——Claude 的 200K token 听起来很多，但一次代码审查、几轮对话、加上项目文件，很快就撑爆了。一旦超出，最早的内容就会被"挤出去"，Agent 就失忆了。

这一章我们要解决两个问题：

1. 上下文压缩：在不丢失关键信息的前提下，把对话历史压缩到原来的十分之一
2. 子 Agent 隔离：让不同任务使用独立的上下文空间，互不污染

完成后，你的 Agent 就能记住整个项目的来龙去脉，而不是只记得最近五分钟的事。

---

前置条件

• 你已经完成了第 3 章，Agent 能读取 Git diff 和错误日志
• 项目根目录下有 agent/ 文件夹，里面是 Agent 的核心代码
• 你有一个可用的 LLM API key（我们继续用 Claude API）

---

第一步：测量你的上下文使用量

在动手优化之前，先搞清楚现状。我们加一个简单的上下文计数器。

在 agent/ 下新建 context_manager.py：

import tiktoken

class ContextTracker:
    """追踪当前对话的 token 使用量"""
    
    def __init__(self, model="claude-3-opus-20240229"):
        # 用 cl100k_base 近似计算 Claude 的 token
        self.encoder = tiktoken.get_encoding("cl100k_base")
        self.messages = []
        self.total_tokens = 0
    
    def add_message(self, role, content):
        """添加一条消息并更新 token 计数"""
        self.messages.append({"role": role, "content": content})
        tokens = len(self.encoder.encode(content))
        self.total_tokens += tokens
        return tokens
    
    def get_usage(self):
        """返回当前使用情况"""
        # Claude 的上下文上限是 200K，我们留 20% 余量
        limit = 200000
        threshold = int(limit * 0.8)  # 160K 触发警告
        return {
            "total_tokens": self.total_tokens,
            "message_count": len(self.messages),
            "usage_pct": round(self.total_tokens / limit * 100, 1),
            "near_limit": self.total_tokens > threshold
        }

预期结果：运行 python -c "from agent.context_manager import ContextTracker; ct = ContextTracker(); print(ct.get_usage())" 应该输出 {'total_tokens': 0, 'message_count': 0, 'usage_pct': 0.0, 'near_limit': False}。

---

第二步：实现上下文压缩

当上下文接近上限时，我们需要把历史对话压缩成一份"摘要"，而不是直接丢弃。压缩的关键是保留决策轨迹而不是逐字对话。

在 context_manager.py 中添加压缩逻辑：

import json

class ContextCompressor:
    """把长对话压缩成结构化摘要"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    def compress(self, messages, target_ratio=0.1):
        """
        压缩对话历史到原来的 target_ratio
        
        返回: 压缩后的摘要字符串
        """
        # 只压缩用户和助手消息，保留系统提示
        system_messages = [m for m in messages if m["role"] == "system"]
        conversation = [m for m in messages if m["role"] in ("user", "assistant")]
        
        # 如果对话太短，不需要压缩
        if len(conversation) < 10:
            return messages
        
        # 让 LLM 自己提取关键信息
        compression_prompt = f"""你是一个对话压缩器。下面是一段 Agent 与用户的对话历史。
请提取出以下关键信息，用 JSON 格式返回：

1. completed_tasks: 已经完成的任务列表（每个任务一句话）
2. current_state: 当前项目状态（文件修改、bug 修复等）
3. pending_decisions: 尚未决定的待办事项
4. key_context: 对后续对话至关重要的上下文（变量名、API 端点、配置值等）

对话历史：
{self._format_conversation(conversation)}

只返回 JSON，不要其他文字。"""
        
        compressed = self.llm.complete(compression_prompt)
        
        try:
            summary = json.loads(compressed)
        except json.JSONDecodeError:
            # 如果 LLM 返回格式不对，用简单的截断
            return self._truncate_fallback(messages, target_ratio)
        
        # 把摘要格式化成系统消息
        summary_text = f"""【上下文摘要 - 压缩率 {int((1-target_ratio)*100)}%】
已完成任务：{', '.join(summary.get('completed_tasks', []))}
当前状态：{summary.get('current_state', '未知')}
待办事项：{summary.get('pending_decisions', [])}
关键上下文：{summary.get('key_context', '无')}"""
        
        # 保留最近的 3 条消息（避免丢失刚说的内容）
        recent = conversation[-3:] if len(conversation) >= 3 else conversation
        
        return system_messages + [
            {"role": "system", "content": summary_text}
        ] + recent
    
    def _format_conversation(self, messages):
        """把消息列表格式化成文本"""
        lines = []
        for msg in messages:
            role = "用户" if msg["role"] == "user" else "Agent"
            # 只取前 200 字符，避免压缩本身消耗太多 token
            content = msg["content"][:200]
            lines.append(f"{role}: {content}")
        return "\n".join(lines[-50:])  # 最多看最近 50 条
    
    def _truncate_fallback(self, messages, ratio):
        """兜底方案：直接截断"""
        total = len(messages)
        keep = max(int(total * ratio), 5)  # 至少保留 5 条
        return messages[:1] + messages[-keep:]  # 保留系统提示 + 最近几条

注意：这里用 self.llm.complete() 调用 LLM 来压缩——让 AI 自己总结自己的对话，效果远好于任何算法。代价是一次额外的 API 调用，但相比节省的上下文空间，这笔交易很划算。

---

第三步：实现子 Agent 隔离

上下文压缩解决了"对话太长"的问题，但还有一个更隐蔽的问题：不同任务互相污染。

比如你让 Agent 先修 bug A，再实现功能 B。修 bug A 时产生的中间文件、临时变量、失败的尝试，都会留在上下文里。当 Agent 开始做功能 B 时，这些垃圾信息会干扰它的判断。

解决方案：每个任务启动一个独立的子 Agent，任务完成后只把最终结果汇报给主 Agent。

在 agent/ 下新建 sub_agent.py：

import uuid
from .context_manager import ContextTracker, ContextCompressor

class SubAgent:
    """一个隔离的子 Agent，拥有独立的上下文"""
    
    def __init__(self, name, llm_client, tools=None):
        self.id = str(uuid.uuid4())[:8]
        self.name = name
        self.llm = llm_client
        self.tools = tools or []
        self.tracker = ContextTracker()
        self.compressor = ContextCompressor(llm_client)
        self.task_result = None
    
    def assign_task(self, task_description, context=None):
        """给子 Agent 分配一个任务"""
        system_prompt = f"""你是子 Agent "{self.name}"（ID: {self.id}）。
你的任务是：{task_description}

你只能关注这个任务。忽略任何与任务无关的信息。
完成任务后，用以下格式汇报结果：
TASK_COMPLETE: [一句话总结]
RESULT: [详细结果，包括文件路径、关键数据等]
ERROR: [如果有错误，描述错误]"""
        
        self.tracker.add_message("system", system_prompt)
        if context:
            self.tracker.add_message("system", f"参考上下文：{context}")
        
        return system_prompt
    
    def run(self, user_input):
        """执行一步操作"""
        # 检查上下文是否接近上限
        usage = self.tracker.get_usage()
        if usage["near_limit"]:
            # 压缩上下文
            self.tracker.messages = self.compressor.compress(
                self.tracker.messages
            )
            print(f"[{self.name}] 上下文已压缩，当前使用率: {usage['usage_pct']}%")
        
        # 添加用户输入
        self.tracker.add_message("user", user_input)
        
        # 调用 LLM
        response = self.llm.complete(
            messages=self.tracker.messages,
            tools=self.tools
        )
        
        # 记录响应
        self.tracker.add_message("assistant", response)
        
        # 检查是否完成任务
        if "TASK_COMPLETE" in response:
            self.task_result = response
        
        return response
    
    def get_result(self):
        """获取任务结果"""
        return self.task_result
    
    def cleanup(self):
        """清理子 Agent（释放上下文）"""
        self.tracker.messages = []
        self.tracker.total_tokens = 0
        print(f"[{self.name}] 已清理，上下文已释放")

---

第四步：把子 Agent 集成到主 Agent 中

现在让主 Agent 学会"生"子 Agent。在 agent/main.py 中（如果你没有这个文件，创建一个）：

from .sub_agent import SubAgent
from .context_manager import ContextTracker, ContextCompressor

class MainAgent:
    """主 Agent，可以派生子 Agent 处理独立任务"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.tracker = ContextTracker()
        self.compressor = ContextCompressor(llm_client)
        self.sub_agents = {}  # name -> SubAgent
    
    def delegate_task(self, task_name, task_description, context=None):
        """把任务委托给子 Agent"""
        print(f"[主 Agent] 委托任务: {task_name}")
        
        sub = SubAgent(
            name=task_name,
            llm_client=self.llm,
            tools=self._get_available_tools()
        )
        sub.assign_task(task_description, context)
        self.sub_agents[task_name] = sub
        
        return sub
    
    def collect_result(self, task_name):
        """收集子 Agent 的结果"""
        sub = self.sub_agents.get(task_name)
        if not sub:
            return None
        
        result = sub.get_result()
        
        # 把结果压缩后加入主 Agent 的上下文
        if result:
            summary = f"[子任务 '{task_name}' 完成] {result[:500]}"
            self.tracker.add_message("system", summary)
        
        # 清理子 Agent
        sub.cleanup()
        del self.sub_agents[task_name]
        
        return result
    
    def _get_available_tools(self):
        """返回当前可用的工具列表"""
        # 这里返回你在第 2 章实现的工具
        return [
            {"name": "read_file", "description": "读取文件内容"},
            {"name": "write_file", "description": "写入文件"},
            {"name": "run_shell", "description": "执行 shell 命令"},
        ]
    
    def process(self, user_input):
        """处理用户输入，自动判断是否需要子 Agent"""
        # 检查上下文
        usage = self.tracker.get_usage()
        if usage["near_limit"]:
            print(f"[主 Agent] 上下文使用率 {usage['usage_pct']}%，开始压缩...")
            self.tracker.messages = self.compressor.compress(
                self.tracker.messages
            )
        
        # 判断是否需要创建子 Agent
        delegation_prompt = f"""用户说: {user_input}

当前上下文使用率: {usage['usage_pct']}%
活跃子 Agent: {list(self.sub_agents.keys())}

如果这个请求可以分解为独立子任务，请输出：
DELEGATE: [子任务名称]
DESCRIPTION: [子任务描述]

否则输出：
PROCESS_DIRECTLY

只输出以上格式之一。"""
        
        decision = self.llm.complete(delegation_prompt)
        
        if decision.startswith("DELEGATE"):
            # 提取任务信息
            lines = decision.split("\n")
            task_name = lines[0].replace("DELEGATE: ", "").strip()
            task_desc = lines[1].replace("DESCRIPTION: ", "").strip()
            
            sub = self.delegate_task(task_name, task_desc)
            # 让子 Agent 开始工作
            result = sub.run(f"请执行任务：{task_desc}")
            
            # 如果子 Agent 立即完成，收集结果
            if sub.get_result():
                return self.collect_result(task_name)
            
            return f"已创建子 Agent '{task_name}' 处理该任务，请稍候..."
        
        # 直接处理
        self.tracker.add_message("user", user_input)
        response = self.llm.complete(
            messages=self.tracker.messages,
            tools=self._get_available_tools()
        )
        self.tracker.add_message("assistant", response)
        return response

---

第五步：用一个真实场景跑通全流程

假设你在开发一个 Web 应用，需要同时做两件事：修复登录页面的 bug，以及添加用户注册功能。这两个任务应该隔离执行。

创建一个测试脚本 test_memory.py：

from agent.main import MainAgent

# 模拟 LLM 客户端（实际使用时替换为真实 API）
class MockLLM:
    def complete(self, messages=None, prompt=None, tools=None, **kwargs):
        # 这里只是演示结构，实际会调用 Claude API
        return "模拟响应"

def main():
    agent = MainAgent(llm_client=MockLLM())
    
    # 模拟一个长对话后的上下文状态
    for i in range(20):
        agent.tracker.add_message("user", f"这是第 {i+1

第5章：为 Agent 设定「知识库」：按需加载项目文档与 API 规范

你的 Agent 已经能读写文件、执行命令、观察 Git 变化。但它还缺一样东西：它不知道你的项目在做什么。

想象一下：你让 Agent 写一个支付模块，它不知道你们用的支付网关是 Stripe 还是支付宝，不知道 API 签名规则，不知道错误码含义。它只能靠训练数据里的通用知识瞎猜——结果大概率是错的。

这一章我们要解决的就是这个问题：让 Agent 在需要的时候，能读到它需要的文档。不是把所有文档一股脑塞给它（那会撑爆上下文），而是按需加载——就像你写代码时不会从头到尾重读一遍文档，而是遇到问题才去查。

前置条件

• 你已经完成了第4章，Agent 能管理上下文和子 Agent 隔离
• 项目里有一些文档文件（README、API 规范、设计文档等）
• 你的 Agent 代码在 agent.py 或类似文件中

第一步：设计知识库的索引结构

先想清楚一个问题：Agent 怎么知道它需要哪份文档？它不能像人一样浏览目录。所以我们需要一个索引——把文档的标题、用途、关键词列出来，让 Agent 先看索引，再决定加载哪份。

创建一个 knowledge_index.json：

{
  "documents": [
    {
      "id": "api-payment",
      "title": "支付 API 规范",
      "path": "docs/api/payment.md",
      "keywords": ["payment", "pay", "charge", "refund", "stripe"],
      "summary": "支付接口的请求/响应格式、签名算法、错误码"
    },
    {
      "id": "style-guide",
      "title": "代码风格指南",
      "path": "docs/style-guide.md",
      "keywords": ["style", "lint", "format", "naming"],
      "summary": "项目使用的代码风格、命名规范、Lint 规则"
    },
    {
      "id": "deploy",
      "title": "部署流程",
      "path": "docs/deploy.md",
      "keywords": ["deploy", "release", "ci", "cd", "docker"],
      "summary": "从开发到生产的部署步骤和环境配置"
    }
  ]
}

预期结果：项目根目录下有一个 knowledge_index.json，列出了所有可用的文档。这个文件本身很小，可以一直放在 Agent 的上下文中。

第二步：实现知识库加载器

现在写一个函数，让 Agent 能根据关键词搜索并加载文档。这个函数要做的：

1. 接收一个查询（比如"支付签名"）
2. 在索引里匹配关键词
3. 返回匹配文档的内容

创建 knowledge_loader.py：

import json
import os
from pathlib import Path

class KnowledgeLoader:
    def __init__(self, index_path="knowledge_index.json"):
        with open(index_path, "r") as f:
            self.index = json.load(f)
        self.cache = {}  # 避免重复读取文件
    
    def search(self, query: str, top_k: int = 3) -> list:
        """根据查询关键词返回最相关的文档内容"""
        query_lower = query.lower()
        scores = []
        
        for doc in self.index["documents"]:
            # 简单的关键词匹配评分
            score = 0
            for kw in doc["keywords"]:
                if kw in query_lower:
                    score += 1
            # 标题匹配加分
            if doc["title"].lower() in query_lower:
                score += 2
            scores.append((score, doc))
        
        # 按评分排序，取 top_k
        scores.sort(key=lambda x: x[0], reverse=True)
        results = []
        
        for score, doc in scores[:top_k]:
            if score > 0:  # 只返回有匹配的
                content = self._load_doc(doc["path"])
                results.append({
                    "title": doc["title"],
                    "content": content,
                    "relevance": score
                })
        
        return results
    
    def _load_doc(self, path: str) -> str:
        """加载文档内容（带缓存）"""
        if path not in self.cache:
            full_path = Path(path)
            if full_path.exists():
                with open(full_path, "r") as f:
                    self.cache[path] = f.read()
            else:
                self.cache[path] = f"[文件不存在: {path}]"
        return self.cache[path]

预期结果：运行 python -c "from knowledge_loader import KnowledgeLoader; kl = KnowledgeLoader(); print(kl.search('支付签名'))" 应该能看到匹配的文档内容。

第三步：把知识库集成到 Agent 的工具集中

光有加载器还不够，Agent 需要能主动调用它。我们在 Agent 的工具集里加一个 read_knowledge 工具：

# 在 agent.py 或 tools.py 中添加

def read_knowledge(query: str) -> str:
    """
    根据查询关键词加载相关的项目文档。
    当你需要了解某个模块、API、规范或流程时使用。
    
    参数:
        query: 描述你需要了解的内容的关键词或问题
    返回:
        匹配文档的内容，如果没有匹配则返回提示信息
    """
    loader = KnowledgeLoader()
    results = loader.search(query)
    
    if not results:
        return "未找到匹配的文档。请尝试其他关键词。"
    
    output = []
    for r in results:
        output.append(f"## {r['title']} (相关度: {r['relevance']})")
        output.append(r['content'])
        output.append("---")
    
    return "\n".join(output)

关键点：这个工具的描述（docstring）要写清楚什么时候用。Agent 会根据描述决定是否调用它。如果描述太模糊，Agent 可能不会主动使用。

第四步：让 Agent 在需要时自动查询

现在 Agent 有了知识库工具，但它怎么知道什么时候该用？两种方式：

方式一：显式指令

在 Agent 的系统提示里加上一段话：

当遇到以下情况时，请使用 read_knowledge 工具：
- 需要调用某个 API 或函数，但不清楚参数格式
- 需要遵循项目特定的规范或约定
- 需要了解某个模块的设计意图
- 不确定某个错误码的含义

方式二：自动触发（进阶）

在 Agent 的观察层（第3章的内容）里加一个检测：当 Agent 的输出包含 API 调用、函数名或特定关键词时，自动查询知识库并把结果注入上下文。

def auto_knowledge_inject(agent_output: str) -> str:
    """检测 Agent 输出中是否涉及已知文档主题，自动注入相关知识"""
    loader = KnowledgeLoader()
    
    # 从输出中提取可能的关键词（简化版）
    keywords = extract_potential_topics(agent_output)
    
    extra_context = ""
    for kw in keywords:
        results = loader.search(kw, top_k=1)
        if results:
            extra_context += f"\n[自动加载知识: {results[0]['title']}]\n"
            extra_context += results[0]['content'][:500] + "\n"  # 只取前500字符
    
    return extra_context

预期结果：当 Agent 说"我需要调用支付接口"时，它会自动触发知识库查询，把支付 API 规范加载到上下文中。

第五步：处理文档更新

文档不是一成不变的。如果项目文档更新了，Agent 的知识库需要同步。最简单的做法：每次调用时重新加载索引文件。

class KnowledgeLoader:
    def __init__(self, index_path="knowledge_index.json", cache_enabled=True):
        self.index_path = index_path
        self.cache_enabled = cache_enabled
        self.cache = {}
        self._load_index()
    
    def _load_index(self):
        with open(self.index_path, "r") as f:
            self.index = json.load(f)
    
    def refresh(self):
        """手动刷新索引和缓存"""
        self._load_index()
        self.cache.clear()

在 Agent 的每次对话开始时调用 refresh()，确保知识是最新的。

一个完整的例子

假设你的项目有一个支付模块，文档 docs/api/payment.md 内容如下：

# 支付 API

## 创建支付订单
POST /api/v1/payments
请求体: { "amount": number, "currency": "USD"|"CNY", "description": string }
响应: { "id": string, "status": "pending"|"completed"|"failed", "url": string }

## 签名算法
所有请求需要在 Header 中携带 X-Signature:
1. 将请求体 JSON 序列化
2. 拼接 secret_key + body
3. 取 SHA256 哈希

现在 Agent 收到任务："创建一个 100 美元的支付订单"。没有知识库时，Agent 可能会猜测 API 地址和参数。有了知识库，它会：

1. 调用 read_knowledge("支付订单创建")
2. 读到文档，知道 POST /api/v1/payments，参数是 amount、currency、description
3. 正确构造请求

常见问题

Q: 文档太多怎么办？ A: 索引文件只存元数据，不存全文。文档内容按需加载。如果文档真的很大（比如几千行），考虑拆分或只加载关键段落。

Q: 关键词匹配不准怎么办？ A: 这是最简单的方案。如果不够用，可以升级为 TF-IDF 或向量检索（用 sentence-transformers 之类的库），但大多数项目用关键词匹配就够了。

Q: Agent 不主动调用知识库工具怎么办？ A: 检查工具描述是否清晰。在系统提示里明确举例说明什么时候该用。如果还不行，用自动注入的方式。

阶段性小结

完成后你已经得到了：

• 一个 knowledge_index.json 索引文件，列出了所有可用文档
• 一个 KnowledgeLoader 类，能根据关键词搜索并加载文档
• 一个 read_knowledge 工具，Agent 可以在需要时调用
• 可选的自动注入机制，让 Agent 在相关场景下自动获取知识

你的 Agent 现在不再是"盲写"了——它能在需要的时候查阅项目文档，就像你写代码时翻文档一样。下一章我们会给这个知识库加上权限控制，防止 Agent 读到不该读的内容。

控制 Agent 的「权限」：沙箱隔离与审批工作流

你的 Agent 已经能读写文件、执行 Shell 命令、联网获取信息了。但有个问题：如果它误删了生产环境的数据库，或者不小心执行了 rm -rf /，谁来负责？

这不是杞人忧天。LLM 模型会犯错，会误解你的意图，甚至在某些情况下被 prompt injection 攻击诱导执行危险操作。本章要解决的问题就是：如何让 Agent 有力量，但又有约束。

我们会做两件事：

1. 把 Agent 关进一个「沙箱」——限制它能访问的文件系统和网络
2. 给危险操作加一道「审批」——让人类在关键步骤上把关

前置条件

• 你已经完成了第 5 章，Agent 能按需加载项目文档
• 项目根目录下有一个 tools/ 文件夹，里面放着你的工具函数
• 你的 Agent 核心代码在 agent.py 或类似文件中

第一步：理解权限模型

先想清楚我们要控制什么。一个 Agent 的权限可以分为三个层次：

层级 1：文件系统权限 —— 能读/写/执行哪些路径
层级 2：网络权限 —— 能访问哪些域名/端口
层级 3：操作权限 —— 哪些命令需要人工确认

我们按这个顺序，从最基础的开始搭。

第二步：实现文件系统沙箱

最简单的沙箱：白名单路径。Agent 只能操作指定目录下的文件。

在 tools/ 下新建 sandbox.py：

import os
from pathlib import Path

class FileSandbox:
    """文件系统沙箱，限制 Agent 能访问的路径"""
    
    def __init__(self, allowed_paths: list[str]):
        # 把所有允许的路径转成绝对路径
        self.allowed = [os.path.abspath(p) for p in allowed_paths]
    
    def is_path_allowed(self, target_path: str) -> bool:
        """检查目标路径是否在允许范围内"""
        abs_target = os.path.abspath(target_path)
        for allowed in self.allowed:
            # 检查目标路径是否以某个允许路径开头
            if abs_target.startswith(allowed):
                return True
        return False
    
    def resolve_path(self, target_path: str) -> str:
        """解析路径，如果不在允许范围内则抛出异常"""
        if not self.is_path_allowed(target_path):
            raise PermissionError(
                f"路径 '{target_path}' 不在允许的沙箱范围内。"
                f"允许的路径: {self.allowed}"
            )
        return os.path.abspath(target_path)

预期结果：你可以这样测试：

sandbox = FileSandbox(["/home/user/project", "/tmp/agent_workspace"])
sandbox.resolve_path("/home/user/project/main.py")  # 正常返回
sandbox.resolve_path("/etc/passwd")  # 抛出 PermissionError

第三步：把沙箱集成到文件工具中

现在修改你的文件读写工具，让它们使用沙箱。假设你原来有个 read_file 函数：

# 原来的版本（无权限控制）
def read_file(path: str) -> str:
    with open(path, 'r') as f:
        return f.read()

# 加了沙箱的版本
def read_file(path: str, sandbox: FileSandbox) -> str:
    safe_path = sandbox.resolve_path(path)
    with open(safe_path, 'r') as f:
        return f.read()

同理，写文件、执行脚本的工具都要加上沙箱检查。

常见报错：如果你忘了把沙箱传给工具函数，会得到 AttributeError: 'NoneType' object has no attribute 'resolve_path'。解决方案：在 Agent 初始化时就创建沙箱实例，作为全局对象传递。

第四步：实现审批工作流

沙箱解决了「能不能做」的问题，但有些操作即使路径合法，也需要人类确认。比如：删除文件、修改配置文件、执行网络请求。

我们设计一个简单的审批系统：

import json
from typing import Callable

class ApprovalWorkflow:
    """操作审批工作流"""
    
    def __init__(self, approver: Callable[[str, dict], bool]):
        """
        approver: 一个函数，接收操作描述和参数，返回 True(批准) 或 False(拒绝)
        """
        self.approver = approver
    
    def request_approval(self, operation: str, params: dict) -> bool:
        """请求审批"""
        print(f"\n⚠️  需要审批的操作: {operation}")
        print(f"   参数: {json.dumps(params, indent=2, ensure_ascii=False)}")
        return self.approver(operation, params)

最简单的审批者就是让用户在终端输入 y/n：

def human_approver(operation: str, params: dict) -> bool:
    response = input("是否批准此操作? (y/n): ").strip().lower()
    return response == 'y'

approval = ApprovalWorkflow(human_approver)

第五步：定义危险操作清单

不是所有操作都需要审批。我们定义一个规则表：

DANGEROUS_OPERATIONS = {
    'delete_file': {
        'description': '删除文件',
        'require_approval': True,
        'risk_level': 'high'
    },
    'modify_config': {
        'description': '修改配置文件',
        'require_approval': True,
        'risk_level': 'medium'
    },
    'execute_shell': {
        'description': '执行 Shell 命令',
        'require_approval': lambda cmd: 'rm' in cmd or 'sudo' in cmd,
        'risk_level': 'high'
    },
    'network_request': {
        'description': '发起网络请求',
        'require_approval': True,
        'risk_level': 'medium'
    },
    'read_file': {
        'description': '读取文件',
        'require_approval': False,
        'risk_level': 'low'
    }
}

注意 execute_shell 的 require_approval 是一个函数——只有包含危险命令时才需要审批。

第六步：把审批集成到工具调用中

现在把沙箱和审批结合起来。我们创建一个「安全工具包装器」：

class SecureToolWrapper:
    """给工具加上权限控制层"""
    
    def __init__(self, sandbox: FileSandbox, approval: ApprovalWorkflow):
        self.sandbox = sandbox
        self.approval = approval
        self.operation_rules = DANGEROUS_OPERATIONS
    
    def call_tool(self, tool_name: str, params: dict) -> str:
        """安全地调用工具"""
        # 1. 检查操作规则
        rule = self.operation_rules.get(tool_name)
        if rule is None:
            raise ValueError(f"未知操作: {tool_name}")
        
        # 2. 检查是否需要审批
        need_approval = rule['require_approval']
        if callable(need_approval):
            # 如果是函数，用参数判断
            need_approval = need_approval(params.get('command', ''))
        
        if need_approval:
            approved = self.approval.request_approval(
                rule['description'],
                params
            )
            if not approved:
                return "操作已被用户拒绝"
        
        # 3. 执行实际工具（这里假设你有一个工具注册表）
        return self._execute_tool(tool_name, params)
    
    def _execute_tool(self, tool_name: str, params: dict) -> str:
        """实际执行工具（简化版）"""
        # 这里应该调用你真正的工具函数
        # 比如：return tools[tool_name](**params)
        return f"执行 {tool_name}，参数: {params}"

第七步：用真实场景串起来

假设你的 Agent 要执行这样一个任务：「读取项目配置文件，然后修改数据库连接字符串」。

# 初始化安全层
sandbox = FileSandbox(["/home/user/my_project"])
approval = ApprovalWorkflow(human_approver)
secure = SecureToolWrapper(sandbox, approval)

# Agent 的思考过程（模拟）
# 第一步：读取配置（不需要审批）
result1 = secure.call_tool('read_file', {
    'path': '/home/user/my_project/config.json'
})
print(result1)  # 直接返回文件内容

# 第二步：修改配置（需要审批）
result2 = secure.call_tool('modify_config', {
    'path': '/home/user/my_project/config.json',
    'changes': {'db_host': 'new-server.example.com'}
})
# 终端会显示：
# ⚠️  需要审批的操作: 修改配置文件
#    参数: {
#      "path": "/home/user/my_project/config.json",
#      "changes": {
#        "db_host": "new-server.example.com"
#      }
#    }
# 是否批准此操作? (y/n): 

预期结果：用户输入 y 则继续执行，输入 n 则返回拒绝信息。

第八步：进阶——更智能的审批策略

每次都让用户手动确认太烦人了。我们可以加一些智能规则：

class SmartApprovalWorkflow(ApprovalWorkflow):
    """带缓存的审批，同一操作短时间内不再重复询问"""
    
    def __init__(self, approver, cache_timeout=300):
        super().__init__(approver)
        self.approval_cache = {}
        self.cache_timeout = cache_timeout  # 秒
    
    def request_approval(self, operation: str, params: dict) -> bool:
        cache_key = f"{operation}:{json.dumps(params, sort_keys=True)}"
        
        # 检查缓存
        if cache_key in self.approval_cache:
            cached_time, result = self.approval_cache[cache_key]
            if time.time() - cached_time < self.cache_timeout:
                print(f"  (使用上次审批结果: {'批准' if result else '拒绝'})")
                return result
        
        # 正常审批
        result = super().request_approval(operation, params)
        self.approval_cache[cache_key] = (time.time(), result)
        return result

这样，如果 Agent 连续多次执行同一个操作（比如批量读取文件），你只需要审批一次。

常见问题与排查

问题 1：Agent 绕过了沙箱

• 现象：Agent 直接调用了 open() 而不是你的安全工具
• 原因：工具注册时没有替换原始函数
• 解决：确保 Agent 只能通过 SecureToolWrapper.call_tool 访问文件系统

问题 2：审批弹窗太多

• 现象：每个操作都要确认，用户烦了直接全部批准
• 原因：危险操作清单太宽泛
• 解决：细化规则，只对真正危险的操作要求审批（比如 rm -rf 而不是所有 rm）

问题 3：沙箱路径解析不一致

• 现象：同样的路径，有时能通过有时不能
• 原因：相对路径和符号链接导致
• 解决：在沙箱初始化时就解析所有路径，用 os.path.realpath() 处理符号链接

完成后你已经得到了…

• 一个文件系统沙箱，把 Agent 限制在指定目录内
• 一个审批工作流，让人类在关键操作上把关
• 一个安全工具包装器，把沙箱和审批整合到工具调用中
• 一个智能审批缓存，减少重复确认

你的 Agent 现在有了「权限意识」——它知道哪些地方能去、哪些操作需要请示。下一章，我们会让 Agent 能「联网」，但有了这章的权限控制，你可以放心地给它开放网络能力。

让 Agent 能「联网」：实现 HTTP 请求与浏览器控制工具

你的 Agent 现在能读写文件、执行 Shell 命令、看 Git diff、管理记忆——但它还活在一个封闭的孤岛上。它不知道外面发生了什么，不能查 API 文档，不能抓取网页，不能调用外部服务。

这一章，我们要给 Agent 装上「网络神经」。完成之后，你的 Agent 就能：

• 发送 HTTP 请求获取数据
• 控制浏览器访问网页
• 把网络信息作为决策依据

前置条件

• 已完成第 2 章的工具系统（Tool 基类和注册机制）
• 已完成第 3 章的观察输入系统（Observation 接口）
• 项目中有 tools/ 目录存放工具实现
• Python 3.8+，安装了 requests 和 playwright 库

第一步：实现 HTTP GET 工具

先做最简单的：让 Agent 能发送 GET 请求获取网页内容或 API 响应。

在 tools/ 下新建 http_tools.py：

import requests
from typing import Dict, Any, Optional
from .base import Tool

class HttpGetTool(Tool):
    """发送 HTTP GET 请求获取资源"""
    
    name = "http_get"
    description = "发送 HTTP GET 请求到指定 URL，返回响应内容"
    
    parameters = {
        "url": {
            "type": "string",
            "description": "要请求的完整 URL，包含协议头（如 https://api.example.com/data）"
        },
        "headers": {
            "type": "object",
            "description": "自定义请求头，可选",
            "default": {}
        },
        "timeout": {
            "type": "integer",
            "description": "超时时间（秒）",
            "default": 30
        }
    }
    
    def execute(self, args: Dict[str, Any]) -> Dict[str, Any]:
        url = args["url"]
        headers = args.get("headers", {})
        timeout = args.get("timeout", 30)
        
        try:
            response = requests.get(url, headers=headers, timeout=timeout)
            return {
                "status_code": response.status_code,
                "headers": dict(response.headers),
                "content": response.text[:10000],  # 限制长度，避免撑爆上下文
                "truncated": len(response.text) > 10000
            }
        except requests.exceptions.Timeout:
            return {"error": f"请求超时（{timeout}秒）"}
        except requests.exceptions.ConnectionError:
            return {"error": f"无法连接到 {url}"}
        except Exception as e:
            return {"error": f"请求失败: {str(e)}"}

关键设计点：

• 限制返回内容长度（10000 字符）—— 网页可能很大，Agent 的上下文窗口有限
• 返回状态码和响应头 —— Agent 需要知道请求是否成功
• 区分超时和连接错误 —— 不同错误需要不同处理策略

预期结果： 注册这个工具后，Agent 可以调用 http_get 获取网页内容。比如让它查天气，它会发 GET 请求到天气 API。

第二步：实现 HTTP POST 工具

GET 只能读数据，POST 才能写数据——提交表单、调用 API、发送消息。

在同一个文件里加上 HttpPostTool：

class HttpPostTool(Tool):
    """发送 HTTP POST 请求提交数据"""
    
    name = "http_post"
    description = "发送 HTTP POST 请求到指定 URL，提交 JSON 或表单数据"
    
    parameters = {
        "url": {
            "type": "string",
            "description": "目标 URL"
        },
        "data": {
            "type": "object",
            "description": "要发送的数据（JSON 格式）"
        },
        "headers": {
            "type": "object",
            "description": "自定义请求头，可选",
            "default": {}
        },
        "timeout": {
            "type": "integer",
            "description": "超时时间（秒）",
            "default": 30
        }
    }
    
    def execute(self, args: Dict[str, Any]) -> Dict[str, Any]:
        url = args["url"]
        data = args.get("data", {})
        headers = args.get("headers", {})
        timeout = args.get("timeout", 30)
        
        # 默认使用 JSON 格式
        if "Content-Type" not in headers:
            headers["Content-Type"] = "application/json"
        
        try:
            response = requests.post(url, json=data, headers=headers, timeout=timeout)
            return {
                "status_code": response.status_code,
                "headers": dict(response.headers),
                "content": response.text[:10000],
                "truncated": len(response.text) > 10000
            }
        except requests.exceptions.Timeout:
            return {"error": f"请求超时（{timeout}秒）"}
        except requests.exceptions.ConnectionError:
            return {"error": f"无法连接到 {url}"}
        except Exception as e:
            return {"error": f"请求失败: {str(e)}"}

注意： 这里默认用 JSON 格式发送数据。如果你的 Agent 需要提交表单（application/x-www-form-urlencoded），可以再加一个参数控制格式。

第三步：实现浏览器控制工具

HTTP 请求能拿到原始数据，但很多网站依赖 JavaScript 渲染内容——直接 GET 只能拿到空壳 HTML。这时候需要真正的浏览器。

我们用 Playwright 控制 Chromium 浏览器。先安装：

pip install playwright
playwright install chromium

新建 browser_tools.py：

from playwright.sync_api import sync_playwright
from typing import Dict, Any, Optional
from .base import Tool

class BrowserNavigateTool(Tool):
    """打开浏览器访问网页，获取渲染后的内容"""
    
    name = "browser_navigate"
    description = "用无头浏览器打开网页，返回渲染后的文本内容"
    
    parameters = {
        "url": {
            "type": "string",
            "description": "要访问的完整 URL"
        },
        "wait_seconds": {
            "type": "integer",
            "description": "页面加载后等待的秒数（用于等待 JS 渲染）",
            "default": 2
        },
        "viewport_width": {
            "type": "integer",
            "description": "浏览器视口宽度",
            "default": 1280
        },
        "viewport_height": {
            "type": "integer",
            "description": "浏览器视口高度",
            "default": 720
        }
    }
    
    def execute(self, args: Dict[str, Any]) -> Dict[str, Any]:
        url = args["url"]
        wait_seconds = args.get("wait_seconds", 2)
        width = args.get("viewport_width", 1280)
        height = args.get("viewport_height", 720)
        
        try:
            with sync_playwright() as p:
                browser = p.chromium.launch(headless=True)
                page = browser.new_page(viewport={"width": width, "height": height})
                page.goto(url, wait_until="networkidle")
                page.wait_for_timeout(wait_seconds * 1000)
                
                # 提取页面文本内容
                content = page.inner_text("body")
                title = page.title()
                current_url = page.url
                
                browser.close()
                
                return {
                    "title": title,
                    "url": current_url,
                    "content": content[:10000],
                    "truncated": len(content) > 10000
                }
        except Exception as e:
            return {"error": f"浏览器访问失败: {str(e)}"}

为什么需要浏览器工具？

• 很多现代网站是 SPA（单页应用），内容由 JavaScript 动态生成
• API 文档网站（如 Swagger UI）需要 JS 渲染
• 登录后的页面状态需要浏览器 Cookie 维持

性能注意： 每次启动浏览器大约需要 1-2 秒。如果 Agent 频繁访问网页，可以考虑复用浏览器实例（后面第 11 章会讲缓存策略）。

第四步：添加浏览器交互能力

光能看还不够——Agent 可能需要点击按钮、填写表单、滚动页面。加一个更通用的浏览器交互工具：

class BrowserActionTool(Tool):
    """在浏览器中执行交互操作"""
    
    name = "browser_action"
    description = "在已打开的页面上执行点击、输入、滚动等操作"
    
    parameters = {
        "action": {
            "type": "string",
            "description": "操作类型：click（点击）、fill（填写）、scroll（滚动）、screenshot（截图）"
        },
        "selector": {
            "type": "string",
            "description": "CSS 选择器，指定操作目标元素"
        },
        "value": {
            "type": "string",
            "description": "填写操作时的输入值，可选"
        },
        "url": {
            "type": "string",
            "description": "要操作的页面 URL（如果浏览器未打开，会先导航到此 URL）"
        }
    }
    
    def execute(self, args: Dict[str, Any]) -> Dict[str, Any]:
        action = args["action"]
        selector = args.get("selector", "")
        value = args.get("value", "")
        url = args.get("url", "")
        
        try:
            with sync_playwright() as p:
                browser = p.chromium.launch(headless=True)
                page = browser.new_page()
                
                if url:
                    page.goto(url, wait_until="networkidle")
                
                if action == "click":
                    page.click(selector)
                elif action == "fill":
                    page.fill(selector, value)
                elif action == "scroll":
                    page.evaluate(f"window.scrollBy(0, {value or 500})")
                elif action == "screenshot":
                    page.screenshot(path="screenshot.png")
                    return {"message": "截图已保存为 screenshot.png"}
                else:
                    return {"error": f"不支持的操作: {action}"}
                
                # 操作后获取页面状态
                content = page.inner_text("body")
                current_url = page.url
                browser.close()
                
                return {
                    "url": current_url,
                    "content": content[:10000],
                    "truncated": len(content) > 10000
                }
        except Exception as e:
            return {"error": f"浏览器操作失败: {str(e)}"}

实用场景： Agent 可以自动登录网站、搜索信息、抓取需要交互才能看到的数据。

第五步：注册工具并测试

把新工具注册到工具管理器里。假设你已经有 ToolRegistry：

# 在 main.py 或 agent 初始化代码中
from tools.http_tools import HttpGetTool, HttpPostTool
from tools.browser_tools import BrowserNavigateTool, BrowserActionTool

registry = ToolRegistry()
registry.register(HttpGetTool())
registry.register(HttpPostTool())
registry.register(BrowserNavigateTool())
registry.register(BrowserActionTool())

写一个简单的测试，看看 Agent 能不能用这些工具：

# test_network.py
from tools.http_tools import HttpGetTool

tool = HttpGetTool()
result = tool.execute({"url": "https://httpbin.org/get"})
print(f"状态码: {result['status_code']}")
print(f"内容前200字: {result['content'][:200]}")

预期输出：

状态码: 200
内容前200字: {
  "args": {}, 
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Host": "httpbin.org",
    ...
  },
  ...
}

常见问题与排查

问题 1：请求被网站屏蔽

• 症状：返回 403 或 429 状态码
• 原因：网站检测到非浏览器请求
• 解决：添加 User-Agent 请求头，模拟真实浏览器

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}

问题 2：Playwright 启动失败

• 症状：playwright install chromium 后仍报错
• 原因：缺少系统依赖（Linux 上常见）
• 解决：运行 playwright install-deps chromium

问题 3：页面内容为空

• 症状：浏览器返回的 content 是空字符串
• 原因：页面完全依赖 JS 渲染，wait_seconds 太短
• 解决：增加等待时间，或使用 wait_for_selector 等待特定元素出现

问题 4：内存泄漏

• 症状：多次调用浏览器工具后内存飙升
• 原因：每次调用都启动新浏览器实例
• 解决：实现浏览器实例池（第 11 章会讲）

真实场景串联

假设你的 Agent 要完成这样一个任务："查一下 Python requests 库的最新版本，然后告诉我它的发布日期"

Agent 的思考过程：

1. 调用 http_get 访问 PyPI API：https://pypi.org/pypi/requests/json
2. 解析返回的 JSON，找到最新版本号
3. 调用 browser_navigate 访问 PyPI 页面：https://pypi.org/project/requests/
4. 提取发布日期信息
5. 汇总结果返回

如果 API 返回的数据足够完整，Agent 可能只用第一步就完成任务。如果 API 没有发布日期，它就会用浏览器去抓取页面。

安全注意事项

网络工具给了 Agent 强大的能力，也带来了风险：

• 不要访问内网地址：Agent 可能被诱导访问 http://localhost:8080 或 http://192.168.1.1
• 限制请求频率：防止 Agent 变成 DDoS 工具
• 敏感信息过滤：响应内容中可能包含 API Key、密码等
• HTTPS 验证：默认启用，不要关闭 SSL 验证

可以在工具执行前加一层安全检查：

def _validate_url(self, url: str):
    """检查 URL 是否安全"""
    from urllib.parse import urlparse
    parsed = urlparse(url)
    
    # 禁止内网地址
    forbidden_hosts = ["localhost", "127

第 8 章：设计可组合的工具：原子化工具与清晰描述规范

你之前给 Agent 装上了「手」——文件读写和 Shell 执行。但如果你写过几个工具，就会发现一个尴尬的问题：工具之间互相重叠，描述模糊，Agent 经常选错工具，或者用工具的方式完全出乎你的意料。

比如你写了一个 read_file 和一个 search_code，结果 Agent 为了找一行代码，先 read_file 读了整个文件，再自己手动搜索——明明 search_code 就能直接搞定。问题出在哪？工具设计得不够「原子化」，描述也不够清晰。

这一章我们要解决的就是这个问题：把工具拆成最小可组合的单元，配上让 Agent 一眼就能看懂的描述。完成后，你的工具集不再是杂乱的功能列表，而是一套可以灵活组合的「乐高积木」。

前置条件

• 你已经完成了第 2 章，有 read_file、write_file、run_shell 这三个基础工具
• 你的项目里有一个 tools/ 目录，每个工具是一个独立的 Python 文件
• 你熟悉 Claude Code 的工具注册方式（tools 列表里声明 function 对象）

第一步：识别「不原子」的工具

先看一个反面例子。假设你有一个工具叫 analyze_code：

# tools/analyze_code.py — 反面教材
def analyze_code(file_path: str, task: str):
    """分析代码文件：可以统计行数、查找函数、检查语法错误"""
    if task == "count_lines":
        with open(file_path) as f:
            return len(f.readlines())
    elif task == "find_functions":
        # 正则匹配函数定义
        ...
    elif task == "check_syntax":
        # 调用编译器检查
        ...

这个工具有三个问题：

1. 职责不单一：一个工具干了三件事，Agent 需要额外指定 task 参数
2. 组合困难：如果我想先统计行数再查找函数，必须调用两次 analyze_code，但第二次还得传 file_path
3. 描述模糊：「分析代码文件」太宽泛，Agent 不知道具体能做什么

原子化原则：一个工具只做一件事，并且把这件事做到极致。

第二步：拆解成原子工具

把上面的 analyze_code 拆成三个独立工具：

# tools/count_lines.py
def count_lines(file_path: str) -> int:
    """统计文件行数（包括空行和注释）"""
    with open(file_path, 'r') as f:
        lines = f.readlines()
    return len(lines)

# tools/find_functions.py
import re

def find_functions(file_path: str) -> list[dict]:
    """查找 Python 文件中的所有函数定义，返回函数名和行号"""
    with open(file_path, 'r') as f:
        content = f.read()
    
    pattern = r'^def\s+(\w+)\s*\('
    matches = re.finditer(pattern, content, re.MULTILINE)
    
    return [
        {"name": m.group(1), "line": content[:m.start()].count('\n') + 1}
        for m in matches
    ]

# tools/check_syntax.py
import ast

def check_syntax(file_path: str) -> dict:
    """检查 Python 文件语法，返回是否合法及错误信息"""
    with open(file_path, 'r') as f:
        content = f.read()
    
    try:
        ast.parse(content)
        return {"valid": True, "error": None}
    except SyntaxError as e:
        return {"valid": False, "error": str(e)}

现在每个工具只做一件事。Agent 可以自由组合它们：先 check_syntax 确认文件合法，再 find_functions 定位函数，最后用 count_lines 看总行数。每个步骤都清晰可控。

第三步：给工具写「让 Agent 秒懂」的描述

工具的描述（description 字段）是 Agent 选择工具的唯一依据。写不好，Agent 就会乱用。

坏描述的例子：

{
    "name": "read_file",
    "description": "读取文件内容",
    "parameters": {...}
}

「读取文件内容」——读什么文件？返回什么格式？能读大文件吗？Agent 拿到这个描述，可能会用它来读 1GB 的日志文件，然后超时崩溃。

好描述的标准：

1. 说明输入：文件路径的格式、编码要求
2. 说明输出：返回什么、什么格式
3. 说明限制：文件大小上限、支持的编码
4. 说明适用场景：什么时候该用这个工具，什么时候不该用

{
    "name": "read_file",
    "description": "读取文本文件内容，返回字符串。适用于配置文件、源代码、日志片段（<1MB）。不支持二进制文件。如果文件过大（>1MB），请使用 tail_file 或 head_file 只读首尾。路径必须是绝对路径。",
    "parameters": {
        "type": "object",
        "properties": {
            "file_path": {
                "type": "string",
                "description": "文件的绝对路径，例如 /home/user/project/main.py"
            }
        },
        "required": ["file_path"]
    }
}

注意描述里的细节：

• 明确说了「<1MB」的限制
• 提示了替代方案（tail_file 或 head_file）
• 说明了路径格式要求

再看一个更复杂的例子——run_shell 的描述：

{
    "name": "run_shell",
    "description": "在项目根目录执行 Shell 命令。适用于：编译、运行测试、安装依赖、Git 操作。注意：命令会阻塞等待完成，耗时命令（>30s）可能超时。不支持交互式命令（如 vim、top）。每次调用执行一个命令，如需多条命令请用 && 连接。",
    "parameters": {
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "要执行的 Shell 命令，例如 'npm test' 或 'git status'"
            }
        },
        "required": ["command"]
    }
}

这里告诉 Agent：

• 什么场景用（编译、测试、Git）
• 什么场景不用（交互式命令）
• 有什么限制（超时、单命令）
• 怎么绕过限制（用 && 连接）

第四步：设计「可组合」的参数

原子工具的参数也要遵循「最小化」原则——只暴露必要的参数，不要提前设计「以后可能会用到」的参数。

反面例子：

def search_text(file_path: str, pattern: str, case_sensitive: bool = True, 
                regex: bool = False, max_results: int = 100, encoding: str = "utf-8",
                include_comments: bool = False):

参数太多，Agent 每次调用都要纠结要不要传那些可选参数。而且 include_comments 这种参数明显是另一个工具的责任。

正面例子：

# tools/search_text.py
def search_text(file_path: str, pattern: str) -> list[dict]:
    """在文件中搜索文本，返回匹配行及其行号。区分大小写，不支持正则。"""
    matches = []
    with open(file_path, 'r', encoding='utf-8') as f:
        for line_no, line in enumerate(f, 1):
            if pattern in line:
                matches.append({
                    "line": line_no,
                    "content": line.rstrip('\n')
                })
    return matches

如果 Agent 需要正则搜索，那就再写一个 search_regex 工具。两个工具可以组合使用：先用 search_regex 找到模式，再用 search_text 确认上下文。

第五步：用一个真实场景串起来

假设你在调试一个 Web 项目，报错说 config.py 第 42 行有语法错误。Agent 会这样组合你的原子工具：

1. 先用 check_syntax 确认语法错误位置
2. 用 read_file 读取 config.py 第 40-50 行看上下文
3. 用 find_functions 查看第 42 行在哪个函数里
4. 用 run_shell 执行 python -c "import config" 验证修复

每一步都是独立的原子工具，但组合起来完成了一个复杂的调试任务。这就是可组合性的力量。

常见错误与排查

错误 1：Agent 总是选错工具

• 检查描述是否清晰，有没有和其他工具的描述重叠
• 比如 read_file 和 search_text 的描述都提到「读取文件内容」，Agent 就会混淆
• 解决方案：在描述里明确区分「读取全部内容」和「搜索特定内容」

错误 2：Agent 传了不存在的参数

• 检查 parameters 定义是否完整，required 字段是否正确
• 常见遗漏：忘记在 properties 里声明所有参数

错误 3：工具返回格式不一致

• 有的工具返回字符串，有的返回 JSON，有的返回 None
• Agent 处理不一致的返回格式容易出错
• 解决方案：统一返回格式，要么全部返回 dict，要么全部返回字符串

工具设计清单

写完一个工具后，对照这个清单检查：

• 这个工具只做一件事吗？
• 描述里说明了输入、输出、限制、适用场景吗？
• 参数最少化了吗？有没有「以后可能用到」的参数？
• 返回格式和已有工具一致吗？
• 工具名称能准确反映功能吗？（不要用 do_stuff 这种名字）
• 如果 Agent 误用这个工具，最坏后果是什么？能在描述里预防吗？

完成后你已经得到了…

一套原子化的工具集，每个工具职责单一、描述清晰、参数精简。Agent 可以像搭积木一样自由组合它们，完成复杂的任务。更重要的是，你有了一个可复用的工具设计范式——以后每加一个新工具，都按照「原子化 + 清晰描述」的标准来，工具集越大越有序，而不是越混乱。

下一章，我们会让这些工具在长对话中保持可用——即使 Agent 已经聊了上千轮，也不会忘记怎么调用它们。

让 Agent 记住上次聊到哪：任务系统让目标跨会话持久化

你有没有遇到过这种情况：跟 Agent 聊了半小时，它帮你改了一堆代码，结果你不小心关掉了终端——再打开时，Agent 一脸茫然地问你“你好，有什么可以帮你的？”之前讨论的 bug、改到一半的函数、下一阶段计划，全没了。

这不是 Agent 笨，是它天生没有“记忆”。每次对话都是全新的开始。但真实项目开发不是一次会话能搞定的——你可能会在开会时想到一个新需求，下班回家后继续改，第二天早上再 review。如果 Agent 每次都要从头理解上下文，那它就不是助手，是累赘。

这一章，我们要给 Agent 装上一个“任务系统”。让它可以：

• 把当前目标保存下来，下次启动时自动恢复
• 记住未完成的任务和进度
• 跨会话保持同一个工作方向

前置条件

在开始之前，确保你已经有：

• 一个能调用 LLM 的脚本（第 1 章的内容）
• 文件读写工具（第 2 章的内容）
• 一个项目目录，里面有一些代码文件

如果你是从零开始，先创建一个测试项目：

mkdir -p ~/agent-task-demo
cd ~/agent-task-demo
echo "def add(a, b):\n    return a + b" > math.py
echo "def greet(name):\n    return f'Hello, {name}'" > hello.py

第一步：设计任务文件格式

任务系统需要一个地方来存“记忆”。最简单的方案是用一个 JSON 文件。每次 Agent 启动时读它，每次有进展时更新它。

我们先定义任务文件的结构。打开终端，创建一个 task_manager.py：

import json
import os
from datetime import datetime

TASK_FILE = ".agent_tasks.json"

def load_tasks():
    """读取任务文件，如果不存在就返回空列表"""
    if not os.path.exists(TASK_FILE):
        return []
    with open(TASK_FILE, "r") as f:
        return json.load(f)

def save_tasks(tasks):
    """保存任务列表到文件"""
    with open(TASK_FILE, "w") as f:
        json.dump(tasks, f, indent=2, ensure_ascii=False)

def add_task(description, priority="medium"):
    """添加一个新任务"""
    tasks = load_tasks()
    task = {
        "id": len(tasks) + 1,
        "description": description,
        "status": "pending",
        "priority": priority,
        "created_at": datetime.now().isoformat(),
        "updated_at": datetime.now().isoformat()
    }
    tasks.append(task)
    save_tasks(tasks)
    return task

def update_task(task_id, status=None, note=None):
    """更新任务状态或添加备注"""
    tasks = load_tasks()
    for task in tasks:
        if task["id"] == task_id:
            if status:
                task["status"] = status
            if note:
                task.setdefault("notes", []).append(note)
            task["updated_at"] = datetime.now().isoformat()
            save_tasks(tasks)
            return task
    return None

def get_active_tasks():
    """获取所有未完成的任务"""
    tasks = load_tasks()
    return [t for t in tasks if t["status"] in ("pending", "in_progress")]

预期结果：运行 python -c "from task_manager import *; print(add_task('实现加法函数'))" 应该输出一个包含 id、description、status 等字段的字典。同时当前目录下会生成一个 .agent_tasks.json 文件。

第二步：让 Agent 启动时加载任务

现在我们要修改 Agent 的启动流程。每次 Agent 被调用时，先检查有没有未完成的任务，然后把它们作为上下文的一部分传给 LLM。

创建一个 agent_with_tasks.py：

import sys
import json
from task_manager import load_tasks, get_active_tasks, add_task, update_task

def build_system_prompt():
    """构建包含任务上下文的系统提示"""
    active_tasks = get_active_tasks()
    
    base_prompt = """你是一个编程助手。你有以下能力：
- 读取和修改文件
- 执行 Shell 命令
- 管理任务列表

当前会话开始。请先查看未完成的任务，然后继续推进。"""
    
    if active_tasks:
        task_context = "\n\n## 未完成的任务\n"
        for t in active_tasks:
            task_context += f"- [{t['id']}] {t['description']} (状态: {t['status']}, 优先级: {t['priority']})\n"
            if "notes" in t:
                for note in t["notes"]:
                    task_context += f"  - 备注: {note}\n"
        base_prompt += task_context
        base_prompt += "\n请从这些任务中继续工作。完成一个任务后，记得更新它的状态。"
    
    return base_prompt

def process_user_input(user_input):
    """处理用户输入，识别任务相关操作"""
    # 这里简化处理，实际应该由 LLM 判断
    # 但我们可以预定义一些快捷命令
    if user_input.startswith("/task add "):
        desc = user_input[10:]
        task = add_task(desc)
        return f"已添加任务 #{task['id']}: {desc}"
    elif user_input.startswith("/task done "):
        try:
            task_id = int(user_input[11:])
            task = update_task(task_id, status="completed")
            if task:
                return f"任务 #{task_id} 已标记为完成"
            return f"未找到任务 #{task_id}"
        except ValueError:
            return "用法: /task done <任务ID>"
    elif user_input == "/tasks":
        tasks = get_active_tasks()
        if not tasks:
            return "没有未完成的任务"
        result = "未完成的任务:\n"
        for t in tasks:
            result += f"  [{t['id']}] {t['description']} ({t['status']})\n"
        return result
    return None  # 不是任务命令，交给 LLM 处理

def main():
    print("Agent 启动中...")
    system_prompt = build_system_prompt()
    print("系统提示已加载，包含任务上下文。")
    print("输入 /help 查看命令，输入 exit 退出。")
    
    # 这里模拟对话循环
    # 实际项目中，这里会调用 LLM API
    while True:
        user_input = input("\n> ").strip()
        if user_input.lower() in ("exit", "quit"):
            break
        if user_input == "/help":
            print("可用命令:")
            print("  /task add <描述>  - 添加任务")
            print("  /task done <ID>   - 完成任务")
            print("  /tasks            - 查看未完成任务")
            print("  exit              - 退出")
            continue
        
        result = process_user_input(user_input)
        if result:
            print(result)
        else:
            # 这里应该是调用 LLM
            print(f"[模拟 LLM 响应] 收到: {user_input}")
            print(f"当前上下文: {system_prompt[:100]}...")

if __name__ == "__main__":
    main()

预期结果：运行 python agent_with_tasks.py，输入 /task add 重构math.py，然后输入 /tasks，应该能看到刚添加的任务。退出再重新运行，输入 /tasks，任务应该还在。

第三步：让 LLM 自动管理任务

手动输入命令虽然能用，但不够智能。真正的 Agent 应该能自己判断什么时候该创建任务、什么时候该更新状态。

我们需要在系统提示里告诉 LLM 一个“任务管理协议”。修改 build_system_prompt 函数，加入以下内容：

def build_system_prompt():
    active_tasks = get_active_tasks()
    
    base_prompt = """你是一个编程助手。你有以下能力：
- 读取和修改文件
- 执行 Shell 命令
- 管理任务列表

## 任务管理规则
1. 当用户提出一个需要多步完成的目标时，自动创建任务
2. 每完成一个子步骤，更新对应任务的状态
3. 如果用户提到之前的工作，先查看未完成任务
4. 任务状态包括: pending(待处理), in_progress(进行中), completed(已完成), blocked(阻塞)
5. 完成任务后，总结做了什么

## 任务操作格式
要操作任务，使用以下格式：
- 创建任务: [TASK_ADD] 描述 [优先级: high/medium/low]
- 更新状态: [TASK_UPDATE id] 新状态 [备注]
- 查看任务: [TASK_LIST]

我会自动解析这些标记并执行对应操作。"""
    
    if active_tasks:
        task_context = "\n\n## 未完成的任务\n"
        for t in active_tasks:
            task_context += f"- [{t['id']}] {t['description']} (状态: {t['status']}, 优先级: {t['priority']})\n"
            if "notes" in t:
                for note in t["notes"]:
                    task_context += f"  - 备注: {note}\n"
        base_prompt += task_context
        base_prompt += "\n请从这些任务中继续工作。"
    
    return base_prompt

然后在 process_user_input 里添加解析这些标记的逻辑：

def parse_task_markers(text):
    """解析 LLM 响应中的任务标记"""
    import re
    
    # 查找 [TASK_ADD] 标记
    add_pattern = r'\[TASK_ADD\]\s*(.+?)(?:\s*\[优先级:\s*(high|medium|low)\])?'
    for match in re.finditer(add_pattern, text, re.IGNORECASE):
        desc = match.group(1).strip()
        priority = match.group(2) if match.group(2) else "medium"
        task = add_task(desc, priority)
        text = text.replace(match.group(0), f"✅ 已创建任务 #{task['id']}: {desc}")
    
    # 查找 [TASK_UPDATE] 标记
    update_pattern = r'\[TASK_UPDATE\s+(\d+)\]\s*(completed|in_progress|pending|blocked)?\s*(.*?)(?=\[|$)'
    for match in re.finditer(update_pattern, text, re.IGNORECASE):
        task_id = int(match.group(1))
        status = match.group(2) if match.group(2) else None
        note = match.group(3).strip() if match.group(3) else None
        task = update_task(task_id, status, note)
        if task:
            replacement = f"✅ 任务 #{task_id} 已更新"
            if status:
                replacement += f" → {status}"
            text = text.replace(match.group(0), replacement)
    
    # 查找 [TASK_LIST] 标记
    if "[TASK_LIST]" in text:
        tasks = get_active_tasks()
        if tasks:
            list_text = "\n当前未完成任务:\n"
            for t in tasks:
                list_text += f"  [{t['id']}] {t['description']} ({t['status']})\n"
            text = text.replace("[TASK_LIST]", list_text)
        else:
            text = text.replace("[TASK_LIST]", "没有未完成的任务")
    
    return text

预期结果：现在 LLM 的响应里如果包含 [TASK_ADD] 优化性能 [优先级: high]，系统会自动创建任务并替换成确认信息。

第四步：实现会话恢复

任务系统最核心的功能是“跨会话恢复”。当 Agent 重新启动时，它应该能告诉用户之前做到哪了。

添加一个 summarize_session 函数：

def summarize_session():
    """生成会话摘要，用于下次启动时恢复上下文"""
    tasks = load_tasks()
    if not tasks:
        return "这是第一次使用任务系统。"
    
    completed = [t for t in tasks if t["status"] == "completed"]
    active = [t for t in tasks if t["status"] != "completed"]
    
    summary = f"任务系统状态:\n"
    summary += f"- 已完成任务: {len(completed)} 个\n"
    summary += f"- 进行中任务: {len(active)} 个\n"
    
    if active:
        summary += "\n继续推进以下任务:\n"
        for t in active:
            summary += f"  [{t['id']}] {t['description']}\n"
            if "notes" in t and t["notes"]:
                summary += f"    上次备注: {t['notes'][-1]}\n"
    
    return summary

修改 main 函数，启动时显示摘要：

def main():
    print("Agent 启动中...")
    print(summarize_session())
    system_prompt = build_system_prompt()
    # ... 其余代码不变

预期结果：退出 Agent 后再启动，应该能看到类似“进行中任务: 2 个”的摘要，并列出具体任务。

第五步：实战演练——跨会话完成一个功能

让我们用一个小例子来验证整个系统。假设你要实现一个“计算器”功能，需要分两步完成。

第一次会话：

python agent_with_tasks.py

模拟对话：

> 我们需要给项目添加一个计算器模块，支持加减乘除
[模拟 LLM 响应] 收到: 我们需要给项目添加一个计算器模块，支持加减乘除
当前上下文: 你是一个编程助手...

这里 LLM 应该自动创建任务。我们手动添加：

> /task add 创建 calculator.py 实现四则运算
已添加任务 #1: 创建 calculator.py 实现四则运算
> /task add 编写测试用例
已添加任务 #2: 编写测试用例
> exit

第二次会话（模拟关闭终端后重新打开）：

python agent_with_tasks.py

输出应该是：

Agent 启动中...
任务系统状态:
- 已完成任务: 0 个
- 进行中任务: 2 个

继续推进以下任务:
  [1] 创建 calculator.py 实现四则运算

调试 Agent 行为：日志记录与工具调用追踪

你的 Agent 已经能读写文件、执行命令、联网查询——但它像个黑箱。你告诉它“做 X”，它做了，但你怎么知道它为什么那么做？当它出错时，你只能猜是模型理解错了、工具调用失败了、还是上下文被截断了？

这一章我们给 Agent 装上“行车记录仪”。每一条思考、每一次工具调用、每一步决策都会被记录下来。调试不再是猜谜，而是回放录像。

前置条件

• 已完成第 2 章（文件读写与 Shell 执行工具）
• 有一个能工作的 Agent 入口脚本（比如 agent.py 或 agent.sh）
• Python 3.8+ 环境

第一步：给每次对话分配唯一 ID

没有 ID 的日志就像没有编号的证据——你根本不知道哪条记录对应哪次对话。我们先让 Agent 启动时生成一个 session_id：

import uuid
from datetime import datetime

def start_session():
    session_id = str(uuid.uuid4())[:8]  # 取前8位，够用且易读
    timestamp = datetime.now().isoformat()
    print(f"[SESSION] {session_id} started at {timestamp}")
    return session_id, timestamp

预期结果：每次运行 Agent，控制台会输出类似 [SESSION] a3f1c2b9 started at 2025-03-15T14:22:31 的一行。

第二步：设计日志结构

日志不能是散乱的文本。我们需要结构化记录，方便后续查询和分析。定义一个简单的日志条目格式：

import json

class AgentLogger:
    def __init__(self, session_id):
        self.session_id = session_id
        self.entries = []
    
    def log(self, level, source, message, metadata=None):
        entry = {
            "session_id": self.session_id,
            "timestamp": datetime.now().isoformat(),
            "level": level,        # "info", "warn", "error", "debug"
            "source": source,      # "model", "tool", "system"
            "message": message,
            "metadata": metadata or {}
        }
        self.entries.append(entry)
        # 同时输出到控制台，方便实时观察
        print(f"[{level.upper()}] [{source}] {message}")
        return entry
    
    def to_json(self):
        return json.dumps(self.entries, indent=2, ensure_ascii=False)

预期结果：调用 logger.log("info", "tool", "read_file executed", {"path": "config.json"}) 会在控制台输出 [INFO] [tool] read_file executed，同时内部保存了完整结构。

第三步：拦截工具调用——记录每次“出手”

Agent 的“手”是工具函数。我们要在工具执行前后插入日志。最简单的方式是用装饰器包装每个工具：

import functools

def logged_tool(logger):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 记录调用开始
            call_id = str(uuid.uuid4())[:6]
            logger.log("info", "tool", f"{func.__name__} called", {
                "call_id": call_id,
                "args": args,
                "kwargs": kwargs
            })
            
            try:
                result = func(*args, **kwargs)
                # 记录调用成功
                logger.log("info", "tool", f"{func.__name__} succeeded", {
                    "call_id": call_id,
                    "result_preview": str(result)[:200]  # 只截取前200字符，避免日志爆炸
                })
                return result
            except Exception as e:
                # 记录调用失败
                logger.log("error", "tool", f"{func.__name__} failed", {
                    "call_id": call_id,
                    "error": str(e)
                })
                raise
        return wrapper
    return decorator

使用方式：

logger = AgentLogger(session_id)

@logged_tool(logger)
def read_file(path):
    with open(path, 'r') as f:
        return f.read()

@logged_tool(logger)
def run_command(cmd):
    import subprocess
    return subprocess.check_output(cmd, shell=True).decode()

预期结果：每次调用 read_file("config.json")，日志会记录：

[INFO] [tool] read_file called
[INFO] [tool] read_file succeeded

如果文件不存在，会记录：

[INFO] [tool] read_file called
[ERROR] [tool] read_file failed

第四步：记录模型的“思考过程”

工具调用只是结果。更关键的是模型为什么决定调用这个工具。我们需要捕获模型的完整思考链（Chain of Thought）。

假设你的 Agent 循环大致是：

def agent_loop(user_input, logger):
    messages = [{"role": "user", "content": user_input}]
    
    while True:
        # 调用模型
        response = call_llm(messages)
        
        # 记录模型的思考
        logger.log("info", "model", "response received", {
            "content": response["content"],
            "finish_reason": response.get("finish_reason", "unknown")
        })
        
        # 解析工具调用
        tool_calls = parse_tool_calls(response)
        if not tool_calls:
            # 模型决定直接回复，结束循环
            logger.log("info", "system", "agent finished", {
                "final_response": response["content"][:200]
            })
            return response["content"]
        
        # 执行每个工具调用
        for tc in tool_calls:
            logger.log("info", "model", f"decided to call {tc['name']}", {
                "arguments": tc["arguments"]
            })
            result = execute_tool(tc["name"], tc["arguments"])
            messages.append({"role": "tool", "content": result})

关键点：在模型返回后、工具执行前记录“决策”，在工具执行后记录“结果”。这样你就能看到完整的因果链：模型看到了什么 → 决定做什么 → 结果是什么。

第五步：保存日志到文件

日志存在内存里，程序一退出就没了。我们需要持久化：

def save_logs(logger, filename=None):
    if filename is None:
        filename = f"agent_log_{logger.session_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    
    with open(filename, 'w') as f:
        f.write(logger.to_json())
    
    print(f"[SYSTEM] Logs saved to {filename}")
    return filename

预期结果：每次 Agent 运行结束，当前目录会生成一个类似 agent_log_a3f1c2b9_20250315_142231.json 的文件。

第六步：做一个简单的日志查看器

JSON 文件可以直接看，但不够直观。我们写一个简单的命令行查看器：

def view_logs(log_file):
    with open(log_file, 'r') as f:
        entries = json.load(f)
    
    for entry in entries:
        level = entry["level"].upper()
        source = entry["source"].ljust(6)  # 对齐
        msg = entry["message"]
        ts = entry["timestamp"][11:19]  # 只取时间部分
        
        # 用颜色区分级别（终端支持的话）
        color = ""
        reset = "\033[0m"
        if level == "ERROR":
            color = "\033[91m"  # 红色
        elif level == "WARN":
            color = "\033[93m"  # 黄色
        elif level == "INFO":
            color = "\033[92m"  # 绿色
        
        print(f"{color}[{ts}] [{level}] [{source}] {msg}{reset}")
        
        # 如果有元数据，展开显示
        meta = entry.get("metadata", {})
        if meta:
            for key, value in meta.items():
                print(f"       {key}: {value}")

预期结果：运行 view_logs("agent_log_a3f1c2b9_20250315_142231.json") 会看到带颜色、有时间戳的日志回放。

第七步：实战——调试一个“迷路”的 Agent

假设你的 Agent 被要求“读取 config.json 并修改其中的 timeout 为 30”，但它却去执行了 ls -la。没有日志时你只能挠头。有了日志，你可以：

1. 打开日志文件，看到模型的思考内容：

   [14:22:35] [INFO] [model] response received
        content: "我需要先看看当前目录有什么文件..."
        finish_reason: "stop"

2. 接着看到工具调用：

   [14:22:35] [INFO] [model] decided to call run_command
        arguments: {"cmd": "ls -la"}

3. 问题清楚了：模型不知道 config.json 在哪里，所以先探索目录。这是提示词的问题——你没有告诉它项目结构。修复提示词后，模型会直接调用 read_file("config.json")。

常见问题与排查

日志文件太大怎么办？

• 在 logged_tool 装饰器中，对 result_preview 只截取前 200 字符
• 可以加一个 max_entries 参数，超过后自动滚动删除旧条目
• 或者按时间分片：每小时生成一个新日志文件

日志里出现乱码？

• 确保所有日志消息都是字符串。如果传入了 bytes 或对象，先 str() 或 repr() 处理
• JSON 序列化时，ensure_ascii=False 可以保留中文

想搜索特定类型的日志？

• 日志是 JSON 格式，直接用 jq 或 Python 脚本过滤：

  # 找出所有错误
  jq '.[] | select(.level == "error")' agent_log_*.json

完成后你已经得到了...

• 一个完整的日志系统：每次对话有唯一 ID，每条记录有时间戳、级别、来源
• 工具调用的自动拦截：每次“出手”都被记录，包括参数和结果
• 模型思考的捕获：你能看到模型为什么做某个决定
• 日志持久化和查看器：可以回放任何一次 Agent 运行的全过程

下一章，我们将利用这些日志来优化 Agent 性能——通过分析工具调用模式，找到可以并行执行的任务，让 Agent 跑得更快。

让 Agent 跑得更快：并行工具调用与缓存策略

你的 Agent 现在能读写文件、联网、看 Git diff，但你可能已经注意到了——当它需要连续调用多个工具时，每一步都要等前一步完成，像排队过安检一样慢。一个典型的场景：Agent 要审查代码，先读文件 A，再读文件 B，再查 Git diff，再调用 Shell 跑测试……每一步都是串行的，总时间 = 所有步骤时间之和。

这一章我们要解决两个问题：让 Agent 同时做多件事（并行工具调用），以及记住已经做过的结果（缓存策略）。完成后，你的 Agent 处理复杂任务的速度能提升 3-5 倍。

前置条件

• 你已经完成了第 10 章，Agent 有日志记录和工具调用追踪能力
• 你的项目里有一个 tools/ 目录，里面放着各个工具的实现
• 你的 Agent 核心循环在 agent.py 或类似文件中

第一步：理解为什么串行调用慢

先看一个典型的串行调用流程：

用户: "审查这个 PR 的改动"
Agent 思考 → 调用 git diff → 等待结果 → 思考 → 读文件 A → 等待结果 → 思考 → 读文件 B → 等待结果 → 思考 → 写审查报告

每个工具调用都包含：LLM 思考时间 + 工具执行时间 + 网络延迟。如果 LLM 一次能决定"我需要同时读文件 A、B、C"，然后一次性拿到所有结果，就能省掉中间的多次思考轮次。

第二步：让 LLM 一次返回多个工具调用

大多数 LLM API 支持在一次响应中返回多个工具调用。关键是在你的 Agent 循环中处理这种情况。

修改你的 agent.py，在调用 LLM 后检查是否返回了多个工具调用：

# agent.py 中的核心循环
def run_agent_cycle(user_input):
    messages = [{"role": "user", "content": user_input}]
    
    while True:
        response = client.chat.completions.create(
            model="claude-3-opus-20240229",
            messages=messages,
            tools=all_tools,  # 你的工具定义列表
            tool_choice="auto"
        )
        
        message = response.choices[0].message
        
        # 检查是否有多个工具调用
        if message.tool_calls:
            # 并行执行所有工具调用
            results = parallel_execute_tools(message.tool_calls)
            
            # 将所有结果一次性加入消息
            for tool_call, result in zip(message.tool_calls, results):
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
            
            # 继续循环，让 LLM 处理所有结果
            continue
        
        # 没有工具调用，返回最终回复
        return message.content

预期结果：Agent 现在能一次接收多个工具调用，而不是每调用一个工具就等一轮 LLM 思考。

第三步：实现并行执行器

parallel_execute_tools 函数是核心。我们用 Python 的 concurrent.futures 来实现：

# tools/executor.py
from concurrent.futures import ThreadPoolExecutor, as_completed
import json

def parallel_execute_tools(tool_calls):
    """并行执行多个工具调用"""
    results = [None] * len(tool_calls)
    
    def execute_single(index, tool_call):
        try:
            # 解析工具名称和参数
            tool_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            
            # 查找并执行工具
            tool_func = get_tool_function(tool_name)
            result = tool_func(**arguments)
            return index, result
        except Exception as e:
            return index, f"Error: {str(e)}"
    
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(execute_single, i, tc)
            for i, tc in enumerate(tool_calls)
        ]
        
        for future in as_completed(futures):
            index, result = future.result()
            results[index] = result
    
    return results

注意：不是所有工具都能并行。文件读写可以，但 Shell 命令如果修改了共享状态（比如切换目录、设置环境变量）就可能冲突。我们后面会处理这个问题。

第四步：标记可并行的工具

在工具定义中加一个 parallel_safe 字段，告诉执行器哪些工具可以并行：

# tools/definitions.py
TOOL_DEFINITIONS = {
    "read_file": {
        "name": "read_file",
        "description": "读取文件内容",
        "parallel_safe": True,  # 可以并行
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string"}
            }
        }
    },
    "run_shell": {
        "name": "run_shell",
        "description": "执行 Shell 命令",
        "parallel_safe": False,  # 不能并行，可能有副作用
        "parameters": {
            "type": "object",
            "properties": {
                "command": {"type": "string"}
            }
        }
    },
    "git_diff": {
        "name": "git_diff",
        "description": "查看 Git 改动",
        "parallel_safe": True,
        "parameters": {
            "type": "object",
            "properties": {}
        }
    }
}

然后修改执行器，只并行执行 parallel_safe 的工具：

def parallel_execute_tools(tool_calls):
    results = [None] * len(tool_calls)
    serial_indices = []
    
    # 第一轮：并行执行 safe 的工具
    safe_indices = [
        i for i, tc in enumerate(tool_calls)
        if is_parallel_safe(tc.function.name)
    ]
    
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = {
            executor.submit(execute_single, i, tool_calls[i]): i
            for i in safe_indices
        }
        for future in as_completed(futures):
            index, result = future.result()
            results[index] = result
    
    # 第二轮：串行执行 unsafe 的工具
    unsafe_indices = [
        i for i, tc in enumerate(tool_calls)
        if not is_parallel_safe(tc.function.name)
    ]
    for i in unsafe_indices:
        _, result = execute_single(i, tool_calls[i])
        results[i] = result
    
    return results

第五步：引入缓存——别重复做同样的事

缓存是另一个巨大的性能提升点。如果 Agent 在同一个会话中多次读取同一个文件，或者重复查询同一个 Git diff，缓存能直接跳过执行。

实现一个简单的内存缓存：

# tools/cache.py
import hashlib
import json
from datetime import datetime, timedelta

class ToolCache:
    def __init__(self, ttl_seconds=300):
        self.cache = {}
        self.ttl = timedelta(seconds=ttl_seconds)
    
    def _make_key(self, tool_name, arguments):
        """生成缓存键"""
        raw = f"{tool_name}:{json.dumps(arguments, sort_keys=True)}"
        return hashlib.md5(raw.encode()).hexdigest()
    
    def get(self, tool_name, arguments):
        key = self._make_key(tool_name, arguments)
        if key in self.cache:
            entry = self.cache[key]
            if datetime.now() - entry['time'] < self.ttl:
                return entry['result']
            else:
                del self.cache[key]  # 过期了
        return None
    
    def set(self, tool_name, arguments, result):
        key = self._make_key(tool_name, arguments)
        self.cache[key] = {
            'result': result,
            'time': datetime.now()
        }
    
    def invalidate(self, tool_name=None):
        """清除缓存，可选指定工具"""
        if tool_name:
            self.cache = {
                k: v for k, v in self.cache.items()
                if not k.startswith(tool_name)
            }
        else:
            self.cache.clear()

# 全局缓存实例
tool_cache = ToolCache(ttl_seconds=300)

第六步：把缓存接入执行器

修改 execute_single 函数，先查缓存再执行：

def execute_single(index, tool_call):
    tool_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    
    # 先查缓存
    cached = tool_cache.get(tool_name, arguments)
    if cached is not None:
        return index, cached
    
    # 执行工具
    try:
        tool_func = get_tool_function(tool_name)
        result = tool_func(**arguments)
        
        # 写入缓存（只缓存读操作，不缓存写操作）
        if is_read_only(tool_name):
            tool_cache.set(tool_name, arguments, result)
        
        return index, result
    except Exception as e:
        return index, f"Error: {str(e)}"

关键点：只缓存读操作（read_file、git_diff、list_directory 等），不缓存写操作（write_file、run_shell 等）。写操作的结果不应该被缓存，因为下一次调用可能产生不同的结果。

第七步：处理缓存失效

有些操作会让缓存失效。比如，如果 Agent 写了一个文件，那么之前缓存的该文件内容就过时了：

# 在工具执行后处理缓存失效
def after_tool_execution(tool_name, arguments):
    if tool_name == "write_file":
        # 使该文件的缓存失效
        file_path = arguments.get("path")
        tool_cache.invalidate("read_file")  # 简单做法：清除所有 read_file 缓存
        # 更精确的做法：只清除特定文件的缓存
        # 但需要更复杂的键匹配逻辑
    
    elif tool_name == "run_shell":
        # Shell 命令可能改变任何东西，清除所有缓存
        tool_cache.invalidate()

第八步：一个完整的例子

让我们把这一切串起来，看一个实际的代码审查场景：

# 用户请求
user_input = "审查 src/main.py 和 src/utils.py 的改动，检查是否有潜在 bug"

# 传统串行流程（无缓存、无并行）：
# 1. LLM 思考 → 调用 git_diff → 等待 → 思考 → 读 main.py → 等待 → 思考 → 读 utils.py → 等待 → 思考 → 输出报告
# 总时间 ≈ 4 轮 LLM 思考 + 3 次工具执行

# 优化后流程（并行 + 缓存）：
# 1. LLM 思考 → 同时调用 git_diff、read_file(main.py)、read_file(utils.py)
# 2. 并行执行三个工具（如果 git_diff 之前调用过，直接返回缓存）
# 3. LLM 一次性拿到所有结果 → 思考 → 输出报告
# 总时间 ≈ 2 轮 LLM 思考 + 1 次并行执行

# 如果第二次审查同样的文件：
# 1. LLM 思考 → 同时调用三个工具
# 2. 全部命中缓存，零等待
# 3. LLM 直接输出报告
# 总时间 ≈ 1 轮 LLM 思考

常见问题与排查

问题 1：并行执行时出现竞态条件

错误：两个工具同时写同一个文件，内容互相覆盖

解决：对写操作使用锁，或者强制写操作串行执行（我们已经通过 parallel_safe 标记处理了）。

问题 2：缓存导致返回过时数据

场景：Agent 先读文件，然后修改文件，再读文件——第二次读返回了缓存中的旧内容

解决：在写操作后清除相关缓存。如果问题持续，可以缩短 TTL 时间，或者对频繁修改的文件不启用缓存。

问题 3：并行执行太多工具导致 API 限流

错误：429 Too Many Requests

解决：限制 max_workers 的值，或者实现一个简单的速率限制器：

import time
from threading import Lock

class RateLimiter:
    def __init__(self, max_per_second=10):
        self.max_per_second = max_per_second
        self.tokens = max_per_second
        self.last_refill = time.time()
        self.lock = Lock()
    
    def acquire(self):
        with self.lock:
            now = time.time()
            elapsed = now - self.last_refill
            self.tokens = min(self.max_per_second, 
                            self.tokens + elapsed * self.max_per_second)
            self.last_refill = now
            
            if self.tokens < 1:
                wait_time = (1 - self.tokens) / self.max_per_second
                time.sleep(wait_time)
                self.tokens = 0
            else:
                self.tokens -= 1

性能对比

用一个简单的测试来验证优化效果：

# test_performance.py
import time

def test_serial_vs_parallel():
    # 模拟 5 个文件读取 + 1 个 git diff
    tasks = [
        ("read_file", {"path": "file1.py"}),
        ("read_file", {"path": "file2.py"}),
        ("read_file", {"path": "file3.py"}),
        ("read_file", {"path": "file4.py"}),
        ("read_file", {"path": "file5.py"}),
        ("git_diff", {}),
    ]
    
    # 串行执行
    start = time.time()
    for name, args in tasks:
        execute_single(0, MockToolCall(name, args))
    serial_time = time.time() - start
    
    # 并行执行
    start = time.time()
    tool_calls = [MockToolCall(name, args) for name, args in tasks]
    parallel_execute_tools(tool_calls)
    parallel_time = time.time() - start
    
    print(f"串行: {serial_time:.2f}s")
    print

第12章：把 Agent 接入你的项目：一个完整的代码审查助手实战

前11章我们搭了Agent的骨架、装了手、给了眼睛、配了记忆——现在该让它在真实项目里干活了。本章的目标是：把一个能读代码、能跑测试、能提建议的审查助手，直接嵌入你的日常开发流程。你提交PR时它自动审查，你改代码时它实时反馈，而不是一个需要你手动唤醒的玩具。

前置条件

• 已完成第1-11章的所有工具和模块（文件读写、Shell执行、Git diff观察、上下文压缩、权限控制、日志追踪）
• 有一个真实项目（Git仓库，有测试用例，有PR或分支）
• 安装了Python 3.9+和Node.js 16+（用于示例项目）
• 有LLM API密钥（OpenAI或Anthropic）

第一步：定义审查助手的「工作流」

审查不是一次性的问答，而是一个流程：拉取代码 → 分析变更 → 运行测试 → 生成报告 → 提交反馈。我们先把这个流程写成可复用的函数。

创建一个 review_assistant.py，从第9章的任务系统继承：

# review_assistant.py
from task_system import TaskSystem  # 第9章实现
from tools import FileTool, ShellTool, GitTool  # 第2、3章实现
from observation import GitDiffObserver  # 第3章实现
from permissions import Sandbox  # 第6章实现
from logger import ToolLogger  # 第10章实现

class CodeReviewAssistant:
    def __init__(self, project_path: str, api_key: str):
        self.project_path = project_path
        self.task_system = TaskSystem()
        self.file_tool = FileTool()
        self.shell_tool = ShellTool(sandbox=Sandbox(project_path))
        self.git_tool = GitTool(project_path)
        self.observer = GitDiffObserver(project_path)
        self.logger = ToolLogger("review_assistant")
        self.api_key = api_key
        
    def review_pr(self, pr_number: int):
        """审查一个PR的完整流程"""
        self.logger.log("start", f"Reviewing PR #{pr_number}")
        
        # 1. 获取PR的变更
        diff = self.git_tool.get_pr_diff(pr_number)
        if not diff:
            self.logger.log("error", "No diff found")
            return None
            
        # 2. 分析变更文件
        changed_files = self._parse_changed_files(diff)
        
        # 3. 对每个文件执行审查
        results = []
        for file_path in changed_files:
            result = self._review_file(file_path, diff)
            results.append(result)
            
        # 4. 运行受影响测试
        test_results = self._run_affected_tests(changed_files)
        
        # 5. 生成综合报告
        report = self._generate_report(results, test_results)
        
        self.logger.log("complete", f"PR #{pr_number} review done")
        return report

预期结果：你有了一个能接收PR编号、自动走完审查流程的类。运行 python -c "from review_assistant import CodeReviewAssistant; print('OK')" 应该不报错。

第二步：实现「变更分析」——让Agent看懂改了什么

审查的核心是理解变更。我们不能把整个diff扔给LLM——上下文窗口会炸。需要先解析出：改了什么文件、改了哪些行、变更类型（新增/修改/删除）。

def _parse_changed_files(self, diff: str) -> list:
    """从diff文本中提取变更文件列表"""
    files = []
    current_file = None
    
    for line in diff.split('\n'):
        if line.startswith('diff --git a/'):
            # 提取文件名
            parts = line.split(' b/')
            if len(parts) > 1:
                current_file = parts[1].strip()
                files.append({
                    'path': current_file,
                    'additions': 0,
                    'deletions': 0,
                    'chunks': []
                })
        elif current_file and line.startswith('@@'):
            # 解析hunk信息
            import re
            match = re.match(r'@@ -\d+(?:,\d+)? \+(\d+)(?:,\d+)? @@', line)
            if match:
                start_line = int(match.group(1))
                files[-1]['chunks'].append({
                    'start_line': start_line,
                    'lines': []
                })
        elif current_file and files[-1]['chunks']:
            if line.startswith('+') and not line.startswith('+++'):
                files[-1]['additions'] += 1
                files[-1]['chunks'][-1]['lines'].append(('add', line[1:]))
            elif line.startswith('-') and not line.startswith('---'):
                files[-1]['deletions'] += 1
                files[-1]['chunks'][-1]['lines'].append(('del', line[1:]))
                
    return files

预期结果：调用 _parse_changed_files 传入一个git diff字符串，返回结构化的文件列表，每个文件包含新增行数、删除行数和变更块。

常见问题：如果diff格式不对（比如从GitHub API获取的diff和本地git diff格式有差异），解析会失败。加个fallback：

def _parse_changed_files(self, diff: str) -> list:
    try:
        return self._parse_git_diff(diff)
    except Exception:
        # fallback: 简单按文件名分割
        return [{'path': line.split()[-1], 'additions': 0, 'deletions': 0, 'chunks': []} 
                for line in diff.split('\n') if line.startswith('diff --git')]

第三步：实现「文件级审查」——让Agent逐文件提建议

现在有了变更文件列表，对每个文件，我们要让LLM做三件事：检查代码质量、发现潜在bug、给出改进建议。

def _review_file(self, file_path: str, full_diff: str) -> dict:
    """审查单个文件的变更"""
    # 提取该文件的diff片段
    file_diff = self._extract_file_diff(full_diff, file_path)
    
    # 读取文件当前内容（用于上下文）
    current_content = self.file_tool.read(f"{self.project_path}/{file_path}")
    
    # 构建审查prompt
    prompt = f"""你是一个代码审查助手。请审查以下文件的变更：

文件路径：{file_path}

变更内容（diff）：
```diff
{file_diff}

当前文件内容：

{current_content}

请从以下维度审查：

1. 正确性：变更是否可能引入bug？边界条件是否处理？
2. 可维护性：代码是否清晰？是否有重复逻辑？
3. 安全性：是否有SQL注入、XSS、路径遍历等风险？
4. 性能：是否有不必要的计算或资源泄漏？
5. 风格：是否符合项目规范？

对每个问题，请给出：

• 严重程度（critical/warning/info）
• 具体位置（行号）
• 问题描述
• 改进建议

如果没有问题，回复"无问题"。"""

# 调用LLM
response = self._call_llm(prompt)

# 解析响应
issues = self._parse_review_response(response)

return {
    'file': file_path,
    'issues': issues,
    'raw_response': response
}

**预期结果**：对每个变更文件，LLM返回结构化的审查意见，包含严重程度、位置和建议。

**实用技巧**：`_infer_language` 函数根据文件扩展名返回语言标识：

```python
def _infer_language(self, file_path: str) -> str:
    ext_map = {
        '.py': 'python',
        '.js': 'javascript',
        '.ts': 'typescript',
        '.jsx': 'jsx',
        '.tsx': 'tsx',
        '.java': 'java',
        '.go': 'go',
        '.rs': 'rust',
        '.md': 'markdown',
        '.json': 'json',
        '.yaml': 'yaml',
        '.yml': 'yaml',
    }
    _, ext = os.path.splitext(file_path)
    return ext_map.get(ext, 'text')

第四步：实现「测试影响分析」——让Agent跑测试并关联结果

光看代码不够，还要跑测试。但全量跑太慢，我们只跑受影响的测试。

def _run_affected_tests(self, changed_files: list) -> dict:
    """运行受变更影响的测试"""
    test_files = self._find_affected_tests(changed_files)
    
    results = {}
    for test_file in test_files:
        self.logger.log("test", f"Running {test_file}")
        
        # 使用第2章的ShellTool执行测试
        result = self.shell_tool.run(
            f"cd {self.project_path} && pytest {test_file} -v --tb=short 2>&1",
            timeout=120  # 2分钟超时
        )
        
        results[test_file] = {
            'passed': result['exit_code'] == 0,
            'output': result['stdout'],
            'failed_tests': self._parse_failed_tests(result['stdout'])
        }
        
        self.logger.log("test_result", 
                       f"{test_file}: {'PASS' if results[test_file]['passed'] else 'FAIL'}")
    
    return results

def _find_affected_tests(self, changed_files: list) -> list:
    """找出受变更影响的测试文件"""
    # 策略1：同名测试文件
    affected = []
    for file_info in changed_files:
        file_path = file_info['path']
        # 假设测试文件在 tests/ 目录下，命名规则 test_<module>.py
        base_name = os.path.splitext(os.path.basename(file_path))[0]
        test_file = f"tests/test_{base_name}.py"
        if os.path.exists(f"{self.project_path}/{test_file}"):
            affected.append(test_file)
    
    # 策略2：使用git blame找出最近修改的测试
    if not affected:
        result = self.shell_tool.run(
            f"cd {self.project_path} && git diff --name-only HEAD~1 -- 'tests/'",
            timeout=10
        )
        affected = [f.strip() for f in result['stdout'].split('\n') if f.strip()]
    
    return affected[:5]  # 最多跑5个测试文件

预期结果：运行 _run_affected_tests 后，返回每个测试文件的运行结果，包含通过/失败状态和失败测试的详细信息。

常见问题：如果项目没有测试，或者测试框架不是pytest，会报错。加个检测：

def _detect_test_framework(self) -> str:
    """检测项目使用的测试框架"""
    if os.path.exists(f"{self.project_path}/pytest.ini") or \
       os.path.exists(f"{self.project_path}/setup.cfg"):
        return "pytest"
    elif os.path.exists(f"{self.project_path}/jest.config.js"):
        return "jest"
    elif os.path.exists(f"{self.project_path}/go.mod"):
        return "go test"
    else:
        return None  # 没有检测到测试框架

第五步：生成「审查报告」——把结果变成可操作的反馈

审查结果不能是一堆JSON，要变成人类可读的报告，同时也要能机器解析（用于自动评论PR）。

def _generate_report(self, review_results: list, test_results: dict) -> str:
    """生成综合审查报告"""
    report_parts = []
    
    # 统计
    total_issues = sum(len(r['issues']) for r in review_results)
    critical_issues = sum(
        1 for r in review_results for i in r['issues'] 
        if i['severity'] == 'critical'
    )
    warning_issues = sum(
        1 for r in review_results for i in r['issues'] 
        if i['severity'] == 'warning'
    )
    
    # 报告头部
    report_parts.append(f"""# 代码审查报告

## 概览
- 审查文件数：{len(review_results)}
- 发现问题：{total_issues}（严重：{critical_issues}，警告：{warning_issues}，提示：{total_issues - critical_issues - warning_issues}）
- 测试结果：{sum(1 for r in test_results.values() if r['passed'])}/{len(test_results)} 通过
""")
    
    # 严重问题摘要
    if critical_issues > 0:
        report_parts.append("## 🚨 严重问题\n")
        for result in review_results:
            for issue in result['issues']:
                if issue['severity'] == 'critical':
                    report_parts.append(
                        f"- **{result['file']}:{issue['line']}** - {issue['description']}\n"
                        f"  - 建议：{issue['suggestion']}\n"
                    )
    
    # 测试失败详情
    failed_tests = {k: v for k, v in test_results.items() if not v['passed']}
    if failed_tests:
        report_parts.append("## ❌ 测试失败\n")
        for test_file, result in failed_tests.items():
            report_parts.append(f"### {test_file}\n")
            for failed in result['failed_tests']:
                report_parts.append(f"- `{failed['name']}`: {failed['error']}\n")
    
    # 详细审查结果
    report_parts.append("## 详细审查\n")
    for result in review_results:
        report_parts.append(f"### {result['file']}\n")
        if not result['issues']:
            report_parts.append("无问题。\n")
        else:
            for issue in result['issues']:
                emoji = {'critical': '🔴', 'warning': '🟡', 'info': '🔵'}.get(issue['severity'], '⚪')
                report_parts.append(
                    f"{emoji}

Learn Claude Code 常见问题（FAQ）

项目简介

learn-claude-code 是一个从零构建的、类似 Claude Code 的轻量级「Agent 工具架」（Harness），核心思想是：智能来自模型训练，而非代码编排。本项目教你如何为模型构建运行环境，而非试图用代码“制造”智能。

---

问题 1：安装时提示 bash: ./setup.sh: Permission denied 怎么办？

解答：
这是文件权限问题。运行以下命令为脚本添加执行权限：

chmod +x setup.sh
./setup.sh

如果依然报错，请确认你使用的是 Bash（而非 Zsh 或 Fish），且系统已安装 curl、git 等基础工具。

---

问题 2：运行后报错 command not found: claude 或 API key not set

解答：
本项目依赖 Anthropic 的 Claude API，需要正确配置环境变量：

1. 确保已安装 Claude CLI（参考 Anthropic 官方文档）
2. 设置 API Key：

   export ANTHROPIC_API_KEY="your-api-key-here"

3. 建议将上述命令添加到 ~/.bashrc 或 ~/.zshrc 中永久生效

---

问题 3：为什么我的 Agent 总是执行错误命令或产生幻觉？

解答：
这是最常见的使用误区。请理解本项目的核心哲学：

• 智能来自模型：Agent 的感知、推理和行动能力来自 Claude 模型本身，而非代码
• 工具架（Harness）只提供环境：代码只负责提供文件 I/O、Shell 执行、权限控制等基础设施
• 模型质量决定行为：如果模型输出错误，说明需要更好的提示工程（Prompt Engineering）或模型微调，而非修改工具架代码

建议：检查你的系统提示（System Prompt）是否清晰定义了任务边界和约束条件。

---

问题 4：与 LangChain、AutoGPT 等框架相比，本项目有什么不同？

解答：
核心区别在于设计哲学：

特性	learn-claude-code	LangChain / AutoGPT
核心思想	智能来自模型训练	智能来自代码编排
架构	轻量工具架（~500 行 Bash）	重型框架（数万行代码）
复杂度	极简，可读性强	高度抽象，学习曲线陡峭
灵活性	直接暴露模型能力	通过抽象层间接调用
适用场景	理解 Agent 本质、教学演示	生产级复杂工作流

本项目更像一个“教学工具”，帮助你理解 Agent 的真正工作原理，而非生产级框架。

---

问题 5：如何添加自定义工具（如数据库查询、API 调用）？

解答：
本项目设计为可扩展的工具架。添加自定义工具的步骤：

1. 在 tools/ 目录下创建新的 Bash 脚本（如 db_query.sh）
2. 在系统提示（System Prompt）中注册该工具的描述和调用方式
3. 确保工具脚本遵循标准输入/输出格式

示例工具结构：

#!/bin/bash
# tool: db_query
# description: 执行 SQL 查询并返回结果
# usage: db_query "SELECT * FROM users LIMIT 5"

echo "Executing: $1"
# 实际数据库查询逻辑

---

问题 6：为什么我的 Agent 无法访问网络或文件系统？

解答：
这是权限控制机制在起作用。本项目默认采用沙箱隔离：

• 文件系统：默认只允许读取项目目录下的文件
• 网络访问：默认禁用，防止 Agent 意外执行危险操作
• Shell 命令：需要显式授权（通过 allow_command 列表）

如需放宽限制，修改 config.sh 中的权限配置：

# 允许网络访问
ALLOW_NETWORK=true

# 允许访问的目录白名单
ALLOWED_DIRS=("/home/user/projects" "/tmp")

---

问题 7：运行 ./agent.sh 后没有任何输出，卡住了怎么办？

解答：
这通常由以下原因导致：

1. API 调用超时：检查网络连接，或增加超时设置：

   export CLAUDE_TIMEOUT=120  # 秒

2. 模型响应过长：设置最大 token 限制：

   export MAX_TOKENS=4096

3. 死循环：Agent 可能陷入了无限推理循环。添加最大迭代次数：

   export MAX_ITERATIONS=20

4. 调试模式：启用详细日志查看具体卡在哪一步：

   DEBUG=true ./agent.sh

---

问题 8：我可以用这个项目做生产环境部署吗？

解答：
不建议直接用于生产环境。本项目的主要定位是：

• ✅ 教学工具：理解 Agent 工作原理
• ✅ 原型验证：快速测试想法
• ✅ 学习资源：研究工具架设计模式
• ❌ 生产部署：缺少错误处理、日志、监控、安全加固等生产级特性

如需生产级方案，建议参考：

• Claude Code（官方产品）
• 基于本项目的学习成果，自行构建更健壮的实现

---

核心提示：记住本项目的核心理念——智能来自模型训练，而非代码编排。工具架只是车辆，模型才是驾驶员。不要试图用复杂的代码逻辑来“制造”智能，而是为模型提供清晰、安全、高效的操作环境。