Agent 开发实战（三）：工具系统深入——从 function calling 到工具编排

第（一）篇我们搭好了最小 Agent，第（二）篇把它接到数据库、表格、文档、生信等多源数据上——工具一下子变多。但那时工具的接法很"原始"：工具串行执行、出错就直接崩、没有权限区分、也没有缓存。demo 阶段没问题，真要落到生产，这几条都会咬人。本篇把工具系统做扎实：集中注册、并行调用、失败重试、权限分级、结果缓存、超时控制。

1. 多源接入里工具系统的三个短板

回看（二）的 dispatch：

def dispatch(name, args):
    if name == "query_table": return {"rows": db.safe_query(**args)}
    ...

它至少有三个问题：

串行且不可控：模型一次可能返回多个 tool_calls（比如"同时查用户表和订单表"），原实现是 for 循环逐个跑，慢。
无容错：某个工具抛异常，整个 Agent 循环就挂了。
无权限/无缓存：任何调用方都能触发"写库"类工具（如果以后有），且相同查询会被反复执行、反复烧 token。

下面逐条补上。

2. 工具注册表：集中管理

与其用一堆 if，不如用一个注册表把"工具元数据 + 执行函数 + 权限 + 缓存策略"绑在一起：

# tools/registry.py
from dataclasses import dataclass, field
from typing import Callable, Awaitable

@dataclass
class Tool:
    name: str
    description: str
    parameters: dict
    func: Callable
    scope: str = "public"        # public / admin
    cache_ttl: int = 0            # 秒，0 表示不缓存

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool):
        self._tools[tool.name] = tool

    def get(self, name: str) -> Tool:
        return self._tools.get(name)

    def openai_schemas(self, scope="public"):
        """按调用方权限返回可见的工具 schema。"""
        out = []
        for t in self._tools.values():
            if scope == "public" and t.scope == "admin":
                continue
            out.append({
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.parameters,
                },
            })
        return out

用装饰器注册更省事：

registry = ToolRegistry()

def tool(name, description, parameters, scope="public", cache_ttl=0):
    def deco(func):
        registry.register(Tool(name, description, parameters, func, scope, cache_ttl))
        return func
    return deco

@tool("query_table", "查询业务库…", {"type":"object","properties":{
    "table":{"type":"string","enum":["users","orders","posts"]},
    "columns":{"type":"array","items":{"type":"string"}}},"required":["table","columns"]})
def query_table(table, columns, where=None, limit=100):
    return {"rows": db.safe_query(table, columns, where, limit)}

把 scope 放在注册信息里，模型看到的工具列表就随调用方身份变化——这就是权限的第一道闸。

3. 并行调用

当模型一次返回多个 tool_calls，且彼此无依赖时，应当并发执行：

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

def _run_one(tool: Tool, call):
    args = json.loads(call.function.arguments)
    return call.id, tool.func(**args)

def execute_calls(calls, registry: ToolRegistry, max_workers=5):
    tools = [registry.get(c.function.name) for c in calls]
    with ThreadPoolExecutor(max_workers=max_workers) as ex:
        results = list(ex.map(lambda tc: _run_one(*tc), zip(tools, calls)))
    return {tid: res for tid, res in results}

注意：只有无依赖的工具才并行。如果工具之间有先后（B 依赖 A 的输出），必须串行。本期我们先假设同批次工具相互独立——绝大多数"同时查几张表"的场景都满足。

4. 失败重试与降级

工具可能超时、可能临时抽风。加一层重试 + 降级：

import time
from functools import wraps

def with_retry(retries=2, delay=0.5):
    def deco(func):
        @wraps(func)
        def wrapper(*a, **k):
            last = None
            for i in range(retries + 1):
                try:
                    return func(*a, **k)
                except Exception as e:
                    last = e
                    if i < retries:
                        time.sleep(delay * (i + 1))
            # 全部失败 → 返回结构化错误，让 LLM 自己决定怎么跟用户解释
            return {"__error__": f"{func.__name__} 失败: {last}"}
        return wrapper

关键点：重试失败不要抛异常中断 Agent，而是返回一个带 __error__ 标记的字典。这样 LLM 能看到"这个工具挂了"，可以在最终回答里如实说明，而不是整个流程崩掉。

5. 权限分级

执行前做一次校验：

def authorize(tool: Tool, caller_scope: str) -> bool:
    if tool.scope == "admin" and caller_scope != "admin":
        return False
    return True

在 Agent 循环里：

for call in msg.tool_calls:
    t = registry.get(call.function.name)
    if not t or not authorize(t, caller_scope):
        messages.append({"role":"tool","tool_call_id":call.id,
                         "content": json.dumps({"__error__":"无权限调用该工具"}, ensure_ascii=False)})
        continue
    ...

原则：模型永远不该看到它没权限的工具，更不该能调用。白名单 + 执行前校验双保险。

6. 结果缓存

相同参数的查询（比如"今日概览"被反复问）没必要每次都跑库：

import hashlib, time

_cache: dict[str, tuple[float, object]] = {}

def cached_call(tool: Tool, call):
    if tool.cache_ttl <= 0:
        return tool.func(**json.loads(call.function.arguments))
    key = hashlib.md5((tool.name + call.function.arguments).encode()).hexdigest()
    now = time.time()
    if key in _cache and now - _cache[key][0] < tool.cache_ttl:
        return _cache[key][1]
    res = tool.func(**json.loads(call.function.arguments))
    _cache[key] = (now, res)
    return res

生产环境把 _cache 换成 Redis（带 TTL），多实例共享。缓存 key 用 工具名 + 参数 的 hash，简单且够用。

7. 超时控制

防止某个工具卡死拖垮整个请求：

from concurrent.futures import ThreadPoolExecutor, TimeoutError as TErr

def with_timeout(seconds=10):
    def deco(func):
        @wraps(func)
        def wrapper(*a, **k):
            with ThreadPoolExecutor(1) as ex:
                fut = ex.submit(func, *a, **k)
                try:
                    return fut.result(timeout=seconds)
                except TErr:
                    return {"__error__": f"{func.__name__} 超时({seconds}s)"}
        return wrapper
    return deco

8. 组合起来：一个更稳的执行器

def execute(calls, registry, caller_scope, max_workers=5):
    results = {}
    with ThreadPoolExecutor(max_workers=max_workers) as ex:
        futures = {}
        for call in calls:
            t = registry.get(call.function.name)
            if not t or not authorize(t, caller_scope):
                results[call.id] = {"__error__": "无权限调用该工具"}
                continue
            futures[ex.submit(cached_call, t, call)] = call
        for fut in futures:
            call = futures[fut]
            try:
                results[call.id] = fut.result()
            except Exception as e:
                results[call.id] = {"__error__": str(e)}
    return results

它在一次提交里完成了：权限校验 → 缓存命中 → 并行执行 → 异常兜底，返回 {tool_call_id: result}，直接回灌给模型。

9. 实战示例

if __name__ == "__main__":
    q = "把 users 表和 orders 表各查前 5 行，再搜一下《退款政策》里关于渠道的条款。"
    print(run_agent(q))   # 模型会并行触发 query_table + query_table + search_documents

你会发现三个工具同时跑完，而不是一个一个等。对于"对比多张表/多文件"的高频分析场景，这一步能把响应时间砍掉一大截。

10. 小结

本篇把工具系统从"能调"升级到"稳、快、可控"：

注册表让工具元数据、权限、缓存策略集中管理；
并行解决多工具慢的问题；
重试/降级/超时保证单点故障不拖垮整体；
权限分级按调用方收敛可见与可执行范围；
缓存减少重复查询与 token 浪费。

这些都不是"炫技"，而是 Agent 从玩具走向可用的必经之路。下一篇我们谈记忆——让 Agent 不只是"一次性问答机"。

工具越多，越要收敛。每加一个工具，先问自己：它该对谁可见？要不要缓存？失败了我希望 Agent 怎么表现？