第2課:給Agent加工具 —— dispatch map模式詳解

系列導讀

這是《12課拆解Claude Code架構》系列的第 2 課。

上一課我們用 20 行 Python 造出了一個能操作真實世界的 Agent:一個 while True 循環(huán) + 一個 Bash 工具。那個循環(huán)是所有 Agent 的骨架,后面 11 課都不會碰它一行。

第 2 課的格言:

"加一個工具,只加一個 handler" —— 循環(huán)不用動,新工具注冊進 dispatch map 就行。


Bash 萬能,但不夠好

第 1 課只有一個 bash 工具。理論上 Bash 能做一切——讀文件用 cat,寫文件用 echo >,編輯文件用 sed。但 "能做" 和 "做得好" 之間差著三個坑:

坑一:cat 截斷不可預測。 文件太大時,模型收到的是截斷后的內容,但它不知道被截了多少。它以為看到了全部,實際只看了一半。后續(xù)決策建立在不完整的信息上。

坑二:sed 遇特殊字符就崩。 文件內容里有 /、&、\ 這些字符時,模型需要在腦子里做正則轉義——這不是 LLM 擅長的事。一個簡單的文本替換,經常因為轉義問題失敗兩三輪。

坑三:每次 bash 調用都是不受約束的安全面。 Bash 能做任何事,也意味著它能做任何危險的事。模型可以讀 /etc/passwd,可以寫 ~/.ssh/authorized_keys,可以 curl 一個惡意腳本然后執(zhí)行。你的防線只有一個字符串黑名單。

專用工具解決這三個問題:read_file 可以精確控制截斷并告知模型;edit_file 用精確字符串匹配替代 sed 正則;路徑沙箱 safe_path() 讓工具只能訪問工作目錄。

關鍵洞察:加工具不需要改循環(huán)。

核心架構:dispatch map

+--------+      +-------+      +------------------+
|  User  | ---> |  LLM  | ---> | Tool Dispatch    |
| prompt |      |       |      | {                |
+--------+      +---+---+      |   bash: run_bash |
                    ^           |   read: run_read |
                    |           |   write: run_wr  |
                    +-----------+   edit: run_edit |
                    tool_result | }                |
                                +------------------+

對比第 1 課的架構圖,唯一變化是右邊:從一個固定的 run_bash 函數(shù),變成了一個字典查找。 模型返回工具名,字典返回對應的處理函數(shù)。

這個模式叫 dispatch map。它的威力在于:無論你有 4 個工具還是 40 個,循環(huán)里的代碼都是同一行:

handler = TOOL_HANDLERS.get(block.name)

四步拆解:從 Bash-only 到多工具 Agent

第一步:路徑沙箱 safe_path()

所有文件操作工具的第一道防線:

WORKDIR = Path.cwd()

def safe_path(p: str) -> Path:
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"Path escapes workspace: {p}")
    return path

三行代碼,做了一件事:確保所有路徑操作都在工作目錄內。

模型傳來 ../../etc/passwd?resolve() 會把它解析為絕對路徑,然后 is_relative_to 檢查發(fā)現(xiàn)它不在 WORKDIR 下,直接拒絕。

這比 Bash 的字符串黑名單強得多。黑名單是 "列舉你不能做什么",沙箱是 "定義你只能在哪做"。前者永遠有漏網(wǎng)之魚,后者天然封閉。

第二步:四個工具處理函數(shù)

每個工具一個函數(shù),各司其職:

def run_bash(command: str) -> str:
    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
    if any(d in command for d in dangerous):
        return "Error: Dangerous command blocked"
    try:
        r = subprocess.run(command, shell=True, cwd=WORKDIR,
                           capture_output=True, text=True, timeout=120)
        out = (r.stdout + r.stderr).strip()
        return out[:50000] if out else "(no output)"
    except subprocess.TimeoutExpired:
        return "Error: Timeout (120s)"

def run_read(path: str, limit: int = None) -> str:
    try:
        text = safe_path(path).read_text()
        lines = text.splitlines()
        if limit and limit < len(lines):
            lines = lines[:limit] + [f"... ({len(lines) - limit} more lines)"]
        return "\n".join(lines)[:50000]
    except Exception as e:
        return f"Error: {e}"

def run_write(path: str, content: str) -> str:
    try:
        fp = safe_path(path)
        fp.parent.mkdir(parents=True, exist_ok=True)
        fp.write_text(content)
        return f"Wrote {len(content)} bytes to {path}"
    except Exception as e:
        return f"Error: {e}"

def run_edit(path: str, old_text: str, new_text: str) -> str:
    try:
        fp = safe_path(path)
        content = fp.read_text()
        if old_text not in content:
            return f"Error: Text not found in {path}"
        fp.write_text(content.replace(old_text, new_text, 1))
        return f"Edited {path}"
    except Exception as e:
        return f"Error: {e}"

注意幾個設計選擇:

  • run_readlimit 參數(shù):模型可以只讀前 N 行,避免大文件撐爆上下文。截斷時主動告知還有多少行沒顯示。
  • run_write 自動創(chuàng)建父目錄mkdir(parents=True) 讓模型不需要先 mkdir -p 再寫文件。
  • run_edit 用精確字符串匹配old_text not in content 直接判斷,不需要正則。替換只替換第一個匹配(replace(..., 1)),避免誤傷。
  • 所有路徑操作都經過 safe_path():Bash 保留了獨立的黑名單防護,文件工具用沙箱防護。

第三步:dispatch map 注冊

TOOL_HANDLERS = {
    "bash":       lambda **kw: run_bash(kw["command"]),
    "read_file":  lambda **kw: run_read(kw["path"], kw.get("limit")),
    "write_file": lambda **kw: run_write(kw["path"], kw["content"]),
    "edit_file":  lambda **kw: run_edit(kw["path"], kw["old_text"],
                                        kw["new_text"]),
}

一個字典,四個條目。新增工具 = 新增一個函數(shù) + 字典里加一行。 不需要碰循環(huán),不需要加 if-elif,不需要改任何已有代碼。

配套的工具定義(schema)告訴模型每個工具長什么樣:

TOOLS = [
    {"name": "bash", "description": "Run a shell command.",
     "input_schema": {"type": "object",
                      "properties": {"command": {"type": "string"}},
                      "required": ["command"]}},
    {"name": "read_file", "description": "Read file contents.",
     "input_schema": {"type": "object",
                      "properties": {"path": {"type": "string"},
                                     "limit": {"type": "integer"}},
                      "required": ["path"]}},
    {"name": "write_file", "description": "Write content to file.",
     "input_schema": {"type": "object",
                      "properties": {"path": {"type": "string"},
                                     "content": {"type": "string"}},
                      "required": ["path", "content"]}},
    {"name": "edit_file", "description": "Replace exact text in file.",
     "input_schema": {"type": "object",
                      "properties": {"path": {"type": "string"},
                                     "old_text": {"type": "string"},
                                     "new_text": {"type": "string"}},
                      "required": ["path", "old_text", "new_text"]}},
]

第四步:循環(huán)集成

循環(huán)本身和第 1 課一模一樣,唯一變化是工具執(zhí)行那幾行:

def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            return
        results = []
        for block in response.content:
            if block.type == "tool_use":
                handler = TOOL_HANDLERS.get(block.name)
                output = handler(**block.input) if handler \
                    else f"Unknown tool: {block.name}"
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})

對比第 1 課,變化只有一處:

# s01: 硬編碼
output = run_bash(block.input["command"])

# s02: 字典查找
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"

兩行替換一行。循環(huán)結構、退出條件、消息管理,全部不變。

架構決策:為什么這樣設計

為什么用 dict 不用 if-elif

# 不要這樣寫
if block.name == "bash":
    output = run_bash(block.input["command"])
elif block.name == "read_file":
    output = run_read(block.input["path"])
elif block.name == "write_file":
    output = run_write(block.input["path"], block.input["content"])
# ... 每加一個工具,循環(huán)里多兩行

if-elif 的問題不是性能,是耦合。每次加工具,你要改循環(huán)代碼。循環(huán)是 Agent 的核心,改核心就要重新測試核心。dispatch map 把 "有哪些工具" 和 "怎么執(zhí)行循環(huán)" 解耦了。

這也是 Claude Code 真實的架構選擇??此脑创a,工具注冊和循環(huán)執(zhí)行是兩個完全獨立的模塊。

為什么要路徑沙箱

你可能覺得 "教學代碼,安全不重要"。但想一下這個場景:

你在 ~/projects/my-app/ 跑 Agent,讓它 "把 config.json 里的數(shù)據(jù)庫密碼改成新密碼"。模型決定先讀文件看看當前內容,調用 read_file("../../.ssh/id_rsa")——

沒有沙箱?它就讀到了你的 SSH 私鑰。

沙箱不是 "生產級安全",而是最低限度的隔離。即使在教學場景,也不應該讓模型訪問工作目錄以外的任何東西。

為什么 edit 用精確匹配不用正則

Claude Code 的 Edit 工具也是精確字符串替換,不是正則。原因很簡單:

  1. LLM 不擅長寫正則。讓模型構造正則表達式,然后用這個正則去改你的代碼——這是在不擅長的事情上疊加風險。
  2. 精確匹配可預測。old_text 在文件里要么存在要么不存在,沒有 "部分匹配" 的歧義。
  3. 只替換第一個replace(old_text, new_text, 1) 避免批量誤傷。如果文件里有多處相同文本,模型需要提供更長的上下文來精確定位。

五分鐘跑起來

# 克隆倉庫
git clone https://github.com/shareAI-lab/learn-claude-code
cd learn-claude-code

# 安裝依賴
pip install -r requirements.txt

# 配置 API Key
cp .env.example .env
# 編輯 .env,填入你的 ANTHROPIC_API_KEY 和 MODEL_ID

# 啟動第二課
python agents/s02_tool_use.py

啟動后你會看到 s02 >> 提示符。以下是真實的運行記錄:

任務 1:讀取文件

s02 >> Read the file requirements.txt
> read_file:
anthropic>=0.25.0
python-dotenv>=1.0.0
pyyaml>=6.0
The `requirements.txt` file contains three dependencies:

1. **anthropic>=0.25.0** - The Anthropic Python SDK (for interacting with Claude API)
2. **python-dotenv>=1.0.0** - For loading environment variables from `.env` files
3. **pyyaml>=6.0** - For parsing and working with YAML files

模型直接選擇了 read_file 而不是 bash → cat。給了專用工具,模型自然會用。

任務 2:創(chuàng)建文件

s02 >> Create a file called greet.py with a greet(name) function
> write_file:
Wrote 46 bytes to greet.py
The file `greet.py` has been created with a `greet(name)` function that takes
a name as input and returns a greeting string like `"Hello, {name}!"`.

任務 3:編輯文件

s02 >> Edit greet.py to add a docstring to the function
> edit_file:
Edited greet.py
Done! A docstring has been added to the `greet` function describing its purpose.

注意模型沒有先 read_fileedit_file——它在上一輪創(chuàng)建文件時已經知道內容,直接精確定位并插入 docstring。Agent 有記憶,不做多余的事。

任務 4:驗證修改

s02 >> Read greet.py to verify the edit worked
> read_file:
def greet(name):
    """Return a greeting message for the given name."""
    return f"Hello, {name}!"
The edit worked correctly. The `greet.py` file now contains the `greet(name)` function
with the docstring included.

對比第 1 課的行為變化:s01 里模型會用 bash → cat 讀文件、bash → sed 編輯文件?,F(xiàn)在它直接選擇專用工具——更精確、更安全、更省 token。你不需要告訴它 "用 read_file 而不是 cat"。

總結:從 s01 到 s02 變了什么

組件 之前 (s01) 之后 (s02)
工具數(shù)量 1 (僅 bash) 4 (bash, read, write, edit)
工具分發(fā) 硬編碼 run_bash() 調用 TOOL_HANDLERS 字典查找
路徑安全 safe_path() 沙箱
文件讀取 bash → cat(截斷不可控) read_file(精確截斷 + 行數(shù)提示)
文件編輯 bash → sed(正則易崩) edit_file(精確匹配替換)
Agent loop while True + stop_reason 不變
新增工具成本 改循環(huán)代碼 加一個函數(shù) + 字典加一行

核心信息:循環(huán)是穩(wěn)定的,工具是可擴展的。 dispatch map 讓這兩件事徹底解耦。

下一課預告

現(xiàn)在 Agent 有了多個工具,能更精確地讀寫文件。但它仍然是 "走到哪算哪"——沒有計劃,不知道自己完成了多少,容易在復雜任務中迷失方向。

第 3 課:TodoWrite —— 給 Agent 一個規(guī)劃系統(tǒng)。核心是一個 TodoManager,讓 Agent 先列步驟再動手。循環(huán)里加一個 nag reminder,催促 Agent 按計劃推進。

# 預告:s03 的 TodoManager
class TodoManager:
    def add(self, task: str, priority: str) -> str: ...
    def complete(self, task_id: str) -> str: ...
    def get_summary(self) -> str: ...  # "3/7 done"

沒有計劃的 Agent 走哪算哪。有了 TodoWrite,完成率直接翻倍。


這是《12課拆解Claude Code架構:從零掌握Agent Harness工程》系列的第 2 課。關注Claw開發(fā)者,不錯過后續(xù)更新。

完整代碼和交互式學習平臺:github.com/shareAI-lab/learn-claude-code

如果這篇文章對你有幫助,歡迎轉發(fā)給你的技術團隊。

系列目錄

  • 第1課:用20行Python造出你的第一個AI Agent
  • 第2課:給Agent加工具 —— dispatch map模式詳解(本文)
  • 第3課:TodoWrite —— 讓Agent先想后做:規(guī)劃系統(tǒng)
  • 第4課:Subagent —— 拆解大任務,上下文隔離
  • 第5課:按需加載領域知識——Skill機制
  • 第6課:無限對話——上下文壓縮三層策略
  • 第7課:任務持久化——文件級DAG任務圖
  • 第8課:后臺執(zhí)行——異步任務與通知隊列
  • 第9課:Agent Teams——多Agent協(xié)作:團隊與郵箱系統(tǒng)
  • 第10課:團隊協(xié)議——狀態(tài)機驅動的協(xié)商
  • 第11課:自治Agent——自組織任務認領
  • 第12課:終極隔離——Worktree并行執(zhí)行

?? 本文原始鏈接第2課:給Agent加工具 —— dispatch map模式詳解

?? 更多 AI Agent 開發(fā)實戰(zhàn)教程,訪問 HuanCode

?? 完整代碼倉庫:github.com/shareAI-lab/learn-claude-code

?著作權歸作者所有,轉載或內容合作請聯(lián)系作者
【社區(qū)內容提示】社區(qū)部分內容疑似由AI輔助生成,瀏覽時請結合常識與多方信息審慎甄別。
平臺聲明:文章內容(如有圖片或視頻亦包括在內)由作者上傳并發(fā)布,文章內容僅代表作者本人觀點,簡書系信息發(fā)布平臺,僅提供信息存儲服務。

相關閱讀更多精彩內容

友情鏈接更多精彩內容