Build An Agent

Appearance

s04 - 工具路由 ​

上一章给了模型一把螺丝刀。它能读文件了，但只会这一招。 现在你要给它一整个工具箱——关键不是多给几把工具，而是怎么管理它们。 当工具变多，你需要一个"调度员"：模型说用哪个，调度员就递哪个。

这一章要解决什么问题 ​

s03 的代码里，工具调用是写死在循环里的。模型说要用 read_file，代码就直接调 read_file()。如果你要加第二个、第三个工具，就得在循环里写更多的 if/else。

这能跑，但不好扩展。

这一章的目标：建一个工具注册表，让"模型想用什么工具"和"工具怎么实现"彻底分开。

以后加新工具，只需要做三件事：

  1. 写一个 Python 函数
  2. 把它塞进注册表
  3. 给模型看一份说明书

循环？一行都不用改。

代码 ​

新建 agent.py：

python
"""
Build An Agent - s04: 工具路由

三个工具 + 一个注册表。新增工具不需要改循环。
"""

import os
import json
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
)


# ── 工具实现 ──────────────────────────────────────────

def read_file(path: str) -> str:
    """读取文件内容。"""
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"错误：{e}"


def list_files(path: str = ".") -> str:
    """列出目录下的文件和子目录。"""
    try:
        entries = os.listdir(path)
        return "\n".join(sorted(entries))
    except Exception as e:
        return f"错误：{e}"


def search_files(query: str, path: str = ".") -> str:
    """在目录下的文本文件中搜索包含关键词的行。"""
    results = []
    try:
        for root, _, files in os.walk(path):
            for name in files:
                filepath = os.path.join(root, name)
                try:
                    with open(filepath, "r", encoding="utf-8") as f:
                        for i, line in enumerate(f, 1):
                            if query in line:
                                results.append(f"{filepath}:{i}: {line.rstrip()}")
                except (UnicodeDecodeError, PermissionError):
                    continue  # 跳过二进制文件和无权限文件
        if not results:
            return "没有找到匹配的内容。"
        return "\n".join(results[:50])  # 最多返回 50 条
    except Exception as e:
        return f"错误：{e}"


# ── 工具注册表 ─────────────────────────────────────────

TOOL_REGISTRY = {
    "read_file": read_file,
    "list_files": list_files,
    "search_files": search_files,
}


def dispatch_tool(name: str, args: dict) -> str:
    """从注册表中查找工具并执行。"""
    func = TOOL_REGISTRY.get(name)
    if func is None:
        return f"错误：未知工具 {name}"
    return func(**args)


# ── 工具定义（给模型看的说明书） ────────────────────────

tools = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取指定路径的文件内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"},
                },
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "list_files",
            "description": "列出指定目录下的文件和子目录",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "目录路径，默认为当前目录"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_files",
            "description": "在目录下的文件中搜索包含指定关键词的行",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "要搜索的关键词"},
                    "path": {"type": "string", "description": "搜索的目录路径，默认为当前目录"},
                },
                "required": ["query"],
            },
        },
    },
]


# ── 对话循环 ───────────────────────────────────────────

messages = [
    {"role": "system", "content": "你是一个有文件访问能力的助手。可以用工具查看文件、列目录、搜索内容。"},
]

print("Agent 已启动，输入 quit 退出。\n")

while True:
    user_input = input("你：")
    if user_input.strip().lower() == "quit":
        break

    messages.append({"role": "user", "content": user_input})

    while True:
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=messages,
            tools=tools,
        )

        msg = response.choices[0].message
        messages.append(msg)

        # 模型没有调用工具，直接输出回复
        if msg.tool_calls is None:
            print(f"Agent：{msg.content}\n")
            break

        # 模型调用了工具，逐个执行
        for tool_call in msg.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            print(f"  [调用工具] {name}({args})")

            result = dispatch_tool(name, args)
            print(f"  [工具结果] {result[:200]}")

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })

运行：

bash
python agent.py

试着这样对话：

你：当前目录下有哪些文件？
  [调用工具] list_files({'path': '.'})
  [工具结果] agent.py
Agent：当前目录下有一个文件：agent.py。

你：帮我搜一下哪里用了 OpenAI
  [调用工具] search_files({'query': 'OpenAI'})
  [工具结果] /path/to/agent.py:6: from openai import OpenAI
Agent：在 agent.py 的第 6 行找到了，import 了 OpenAI。

你：读一下这个文件的前 10 行
  [调用工具] read_file({'path': 'agent.py'})
  [工具结果] """
  Build An Agent - s04: 工具路由...
Agent：文件开头是一个 docstring，说明这是 s04 的工具路由章节...

发生了什么 ​

跟 s03 比，代码多了三个东西：两个新工具函数，一个注册表。

三个工具函数 ​

python
def read_file(path: str) -> str: ...
def list_files(path: str = ".") -> str: ...
def search_files(query: str, path: str = ".") -> str: ...

每个函数只做一件事，只接收参数、返回字符串。没有副作用，不依赖全局状态。这是"工具"的基本形态——越简单越好。

注册表：名字到函数的映射 ​

python
TOOL_REGISTRY = {
    "read_file": read_file,
    "list_files": list_files,
    "search_files": search_files,
}

一个普通的字典。key 是工具名（字符串），value 是对应的函数。

模型返回的工具调用带着一个 name 字段，比如 "list_files"。你要做的就是用这个名字去字典里查，找到对应的函数，调用它。

dispatch_tool：调度员 ​

python
def dispatch_tool(name: str, args: dict) -> str:
    func = TOOL_REGISTRY.get(name)
    if func is None:
        return f"错误：未知工具 {name}"
    return func(**args)

三行逻辑：

  1. 从注册表里查名字
  2. 没找到就报错
  3. 找到了就用 **args 展开参数调用

**args 是关键。模型返回的参数是一个字典，比如 {"path": "."}。** 把它展开成关键字参数，等价于 func(path=".")。这样不管工具的参数签名是什么，dispatch 都能处理。

对话循环的变化 ​

python
for tool_call in msg.tool_calls:
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    result = dispatch_tool(name, args)  # 就这一行

    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result,
    })

循环里不再有 if name == "read_file" 这样的判断。不管模型调用什么工具，都走同一条路：dispatch_tool(name, args)。

这就是"路由"的意思：根据名字，把请求分发到正确的处理函数。

三层分离 ​

现在代码里有三个独立的层：

层职责改动频率
工具函数具体实现加新工具时改
注册表名字→函数映射加新工具时改
工具说明书告诉模型有哪些工具可用加新工具时改
对话循环发消息、收消息、调度工具几乎不用改

循环只跟 dispatch_tool 打交道，不关心具体有哪些工具。这就是解耦。

加一个新工具 ​

假设你要加一个 write_file 工具。三步：

第一步，写函数：

python
def write_file(path: str, content: str) -> str:
    """写入内容到文件。"""
    try:
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        return f"已写入 {path}"
    except Exception as e:
        return f"错误：{e}"

第二步，加进注册表：

python
TOOL_REGISTRY = {
    "read_file": read_file,
    "list_files": list_files,
    "search_files": search_files,
    "write_file": write_file,  # 加这一行
}

第三步，加一份说明书：

python
{
    "type": "function",
    "function": {
        "name": "write_file",
        "description": "将内容写入指定路径的文件",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "文件路径"},
                "content": {"type": "string", "description": "要写入的内容"},
            },
            "required": ["path", "content"],
        },
    },
},

循环？不用动。dispatch_tool 会自动识别新工具。

试着改改 ​

1. 加一个 calculate 工具

python
def calculate(expression: str) -> str:
    """计算数学表达式。"""
    try:
        # 注意：eval 有安全风险，这里仅作演示
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"错误：{e}"

注册到 TOOL_REGISTRY，加上说明书，然后问模型："帮我算一下 2 的 20 次方。"

2. 给 dispatch_tool 加日志

python
def dispatch_tool(name: str, args: dict) -> str:
    func = TOOL_REGISTRY.get(name)
    if func is None:
        return f"错误：未知工具 {name}"
    print(f"  [路由] {name} -> {func.__name__}")  # 看看实际调用了哪个函数
    return func(**args)

3. 把工具说明书也从注册表自动生成

现在工具的"名字"和"说明书"是分开维护的。你可以把说明书也放进注册表，这样只维护一份数据：

python
# 这是一个更高级的模式，先理解思路
TOOL_REGISTRY = {
    "read_file": {
        "func": read_file,
        "schema": { ... },  # 说明书
    },
    ...
}

教学边界 ​

这一章只做一件事：用注册表实现工具路由，让工具和循环解耦。

不涉及：

  • 动态加载工具（从文件或插件系统加载）
  • 工具权限控制（哪些工具可以用、哪些不行）
  • 工具结果后处理（过滤、摘要、格式化）
  • 并行执行工具（现在是逐个执行）
  • 工具依赖（一个工具的输出作为另一个的输入）

这些都在后面的章节里。现在你只需要理解一个模式：

工具是插头，注册表是插线板。 你不用改墙里的线路（循环），只需要把新插头（工具函数）插到插线板（注册表）上。模型要用电，插线板会找到对应的插头。

一句话记住 ​

Agent 的第三步，是建一个工具注册表——模型说用什么，注册表就递什么，循环永远不用改。