问题

这一章主要目的是为了体会不同tool是如何被使用的。

解决方案

+--------+      +-------+      +------------------+

|  User  | ---> |  LLM  | ---> | Tool Dispatch    |

| prompt |      |       |      | {                |

+--------+      +---+---+      |   bash: run_bash |

                    ^           |   read: run_read |

                    |           |   write: run_wr  |

                    +-----------+   edit: run_edit |

                    tool_result | }                |

                                +------------------+

  

The dispatch map is a dict: {tool_name: handler_function}.

One lookup replaces any if/elif chain.

工作原理

1. 每个工具有一个处理函数。路径沙箱防止逃逸工作区。

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

  

def run_read(path: str, limit: int = None) -> str:

    text = safe_path(path).read_text()

    lines = text.splitlines()

    if limit and limit < len(lines):

        lines = lines[:limit]

    return "\n".join(lines)[:50000]

1. 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"]),

}

这种就是匿名函数lambda
简单理解就是** 解析传入的词典后扔到对应的函数对应的参数中。

1. 循环中按名称查找处理函数。循环体本身与 s01 完全一致。

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,

        })

加工具 = 加 handler + 加 schema。循环永远不变。

笔记

这一节主要就是理解工具是怎么注册的。
我们从上到下理解，首先我们的Agent需要非常多的tool，那我们怎么让AI辨别这些tool呢，其实很简单，就是告诉AI我们有哪些tool的名称以及对应的参数吗，然后AI会根据函数名以及参数名"猜“出来这个函数并返回对应的格式。

示例：
当我运行之后，我输入use bash ls the current file
LLM会收到如下信息：

        response = client.messages.create(

            model=MODEL, system=SYSTEM, messages=messages,

            tools=TOOLS, max_tokens=8000,

        )

这些参数都是claude code要求的参数

同时，我们会收到LLM传回的response，其中包含很多个消息块（我们可以在debug中看到response中content参数），对应代码中的block
如果block.type == "tool_use":的时候,我们来看一下对应block的详细信息

ToolUseBlock(id='toolu_vrtx_01GgcQiZyHQaFYf8STUJjw5a', caller=None, input={'command': 'ls -la'}, name='bash', type='tool_use')

需要重点关注如下几个参数，name就是使用工具的名称，input就是我们使用的参数对应的词典，以及type就是当前block的类型。
继续往下，就是我们通过代码，也就是现在最火的所谓的Harness来解析LLM返回的参数

                handler = TOOL_HANDLERS.get(block.name)

                output = handler(**block.input)

然后首先我们通过TOOL_HANDLERS也就是前文我们注册的工具，获取当前block的name，当前handler就变成了

    "bash":       lambda **kw: run_bash(kw["command"]),

尽管在实际编译器不是这样子的，但是我们可以理解成这个函数。我们向前文注册的工具，对应于这个例子就是bash（对应前文的name，输入了参数{'command': 'ls -la'}，最终由run_bash这个函数执行

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)"

具体执行函数是subprocess.run，并得到output为

drwxr-xr-x 13 root root  4096 Apr  2 10:07 .
drwx------ 12 root root  4096 Apr  2 11:45 ..
-rw-r--r--  1 root root  2665 Apr  2 11:45 .env
drwxr-xr-x  8 root root  4096 Mar 27 11:19 .git
......

所以整个流程说起来就是我们向LLM发送了我们的prompt（输入栏输入的信息）以及注册的工具信息以及其需要的参数（TOOLS），LLM通过解析我们的prompt，决定使用bash这个工具，并决定command是ls -la，最终我们接受了该response的信息并解析得到输出output