什么是大模型的工具调用框架？

一个完整的工具调用框架，通常分为以下八层。

1️⃣ Tool Definition Layer（工具定义层）

作用

定义工具的“契约”。

包含

  • name
  • description
  • 参数 schema（JSON Schema / Pydantic）
  • 返回格式
  • 权限标签
  • 幂等性声明（可选）

为什么重要

没有 schema，工具调用就是字符串拼接。

生产环境不允许。

2️⃣ Tool Registry（工具注册与边界层）

作用

  • 管理可用工具列表
  • 做存在性检查
  • 做权限过滤
  • 支持插件化加载

解决的问题

  • 模型调用不存在工具怎么办？
  • 不同用户能用不同工具怎么办？
  • 动态扩展工具怎么办？

这是系统边界。

3️⃣ Tool Routing Layer（工具选择层）

作用

决定“调用哪个工具”。

常见实现

  • LLM function calling
  • ReAct
  • Embedding 路由
  • 规则匹配
  • 混合策略

解决的问题

  • 工具很多时如何选？
  • 避免误触发
  • 控制调用成本

4️⃣ Argument Generation & Validation（参数层）

作用

处理模型生成的参数。

包含

  • JSON 解析
  • 类型校验
  • 必填字段检查
  • 默认值填充
  • 自动纠错
  • 重试机制

解决的问题

  • 参数缺失
  • 类型错误
  • 拼写错误
  • 幻觉字段

这是第一道安全护栏。

5️⃣ Execution Control Layer（执行控制层）🔥最核心

这是框架真正的复杂度来源。

包含中间件链

1
2
3
4
5
6
7
8
9
Execution Layer
 ├── Timeout
 ├── Retry
 ├── RateLimit
 ├── Auth
 ├── Audit
 ├── Cache
 ├── CostTracking
 └── CircuitBreaker

解决的问题

  • 工具超时怎么办？
  • API 失败怎么办？
  • 防止重复执行
  • 限制危险操作
  • 记录谁调用了什么
  • 控制成本

没有这层，就不是工程系统。

6️⃣ State Management（状态管理层）

Agent 不只是单步调用。

包含

  • 当前任务状态
  • 每步结果 artifacts
  • 依赖关系
  • 状态机（pending / running / failed）

解决的问题

  • 多步任务如何串联？
  • 第3步失败怎么办？
  • 是否支持回滚？
  • 是否支持并行？

很多人完全忽略这层。

7️⃣ Memory & Context Layer（记忆与上下文层）

工具调用必须结合上下文管理。

包含

  • 短期记忆（会话）
  • 长期记忆（向量库）
  • 结果摘要
  • 上下文裁剪
  • Token 预算控制

解决的问题

  • 上下文爆炸
  • 结果过大
  • 历史信息污染
  • 记忆写入策略

工具调用和 RAG 在生产系统里是耦合的。

8️⃣ Grounding & Observation Layer（结果注入层）

工具执行完成后：

  • 注入多少结果给模型？
  • 注入摘要还是全文？
  • 是否存储引用 ID？
  • 如何防止 hallucination？

这是很多人忽视的一层。

9️⃣ Observability & Evaluation（可观测与评估层）

这是生产级加分项。

包含

  • Trace（调用链）
  • 延迟统计
  • 成功率
  • 成本统计
  • 自动化测试场景

你可以把完整框架理解成这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
User
  ↓
Planner / Router
  ↓
ToolCall (schema)
  ↓
Registry (存在性/权限)
  ↓
Argument Validation
  ↓
Execution Middleware
  ↓
External System
  ↓
Grounding
  ↓
State / Memory
  ↓
Next Step / Final Answer

现在我们来逐层解析每层的内容。

首先我们讲解：

1. Tool Definition Layer（工具定义层）

这一层首先我们要定义一个基本的工具类。
这里我们采用一个pydantic的库进行参数校验。
采用pydantic的库的好处是

通过pydantic的库，我们可以定义工具的参数,不同的工具可以自动进行参数校验。
这个basemodel具有以下几个功能：

1. 类型检查：如果我们需要的参数是str,但是我们传入的参数是int，那么这个库会自动修正

2. 缺失报错，如果我们缺少了参数，那么这个库会报错，我们可以把错误传给ai，让ai重新生成调用工具的参数。

3. 他会根据这个工具我们定义的参数格式，自动将json的需求拼入prompt中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any, Dict, Generic, Type, TypeVar

from pydantic import BaseModel

ArgsT = TypeVar("ArgsT", bound=BaseModel)


class Tool(ABC, Generic[ArgsT]):
    """
    Day1: Tool + ArgsModel（Pydantic）
    - Tool.name / Tool.description
    - Tool.ArgsModel: 参数schema
    - run(args): 执行逻辑
    """
    name: str
    description: str
    ArgsModel: Type[ArgsT]

    def validate(self, raw_args: Dict[str, Any]) -> ArgsT:
        # 统一用 schema 强校验（这就是后面“参数错怎么办”的入口）
        return self.ArgsModel.model_validate(raw_args)

    @abstractmethod
    def run(self, args: ArgsT) -> Any:
        raise NotImplementedError

# 结构限定：每个工具必须关联一个具体的 ArgsModel（比如：查天气工具必须有 city 字段，且必须是 str 类型）。

# 状态限定：在 run 方法执行之前，validate 方法会将原始的、杂乱的字典（raw_args）转化成一个经过验证、类型确定的 BaseModel 实例。

3. 工具注册与边界层

这一层我们用一个代码完成工具的定义操作
我们实现
注册工具、找工具、列出所有的工具、检查工具是否存在

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Dict, List

from toollab.tools.base import Tool


class ToolNotFoundError(KeyError):
    pass


@dataclass
class ToolRegistry:
    """
    Day1: ToolRegistry（存在性检查）
    - register / get / list
    - 工具不存在时给出可用列表（这就是面试题1：工具名错怎么办）
    """
    _tools: Dict[str, Tool] = field(default_factory=dict)

    def register(self, tool: Tool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"Tool already registered: {tool.name}")
        self._tools[tool.name] = tool

    def get(self, name: str) -> Tool:
        if name not in self._tools:
            raise ToolNotFoundError(
                f"Tool not found: {name}. Available tools: {sorted(self._tools.keys())}"
            )
        return self._tools[name]

    def list(self) -> List[str]:
        return sorted(self._tools.keys())

2. Execution Control Layer（执行控制层）🔥

这一层我们首先限定工具的调用模式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from __future__ import annotations

from typing import Any, Dict, Optional
from pydantic import BaseModel, Field


class ToolCall(BaseModel):
    """
    这是“模型输出”的结构化结果（今天先固定输出，后续再随机出错）。
    """
    tool_name: str = Field(..., description="Name of the tool to call")
    arguments: Dict[str, Any] = Field(default_factory=dict)
    call_id: str = Field("call_001", description="For tracing/debugging")


class ToolObservation(BaseModel):
    """
    工具执行后的结构化观测结果。
    """
    call_id: str
    tool_name: str
    ok: bool = True
    result: Any = None
    error: Optional[str] = None

通过这样两个类指定模型输出的结构化结构和工具执行的结构化结果，我们可以确保模型输出的工具调用是符合我们定义的格式的，同时也可以确保工具执行的结果是符合我们定义的格式的。

3. Execution Control Layer（执行控制层 / 中间件链）

执行层的主要目的是失败不会让系统崩、有重试策略、有退避策略、有 trace 可复盘。
通过这一串代码，我们可以优雅的实现，工具存在性校验、参数校验、超时等待、全程观测等功能。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
from __future__ import annotations

import time
import math
from dataclasses import dataclass
from typing import Any, Dict, Optional

from pydantic import ValidationError

from contracts import ToolCall, ToolObservation
from tools.registry import ToolRegistry, ToolNotFoundError


@dataclass
class ExecutionConfig:
    timeout_seconds: float = 1.0
    max_attempts: int = 3
    backoff_base_seconds: float = 0.2  # 0.2, 0.4, 0.8...


@dataclass
class ToolExecutor:
    """
    Day3：执行控制层最小实现
    - registry 存在性检查
    - schema 参数校验
    - timeout（用 tool 自己的超时/或框架超时）
    - retry（指数退避）
    - trace（attempt、耗时、错误）
    """
    registry: ToolRegistry
    config: ExecutionConfig = ExecutionConfig()

    def execute(self, call: ToolCall) -> ToolObservation:
        # 1) tool existence
        try:
            tool = self.registry.get(call.tool_name)
        except ToolNotFoundError as e:
            return ToolObservation(
                call_id=call.call_id,
                tool_name=call.tool_name,
                ok=False,
                error=str(e),
            )

        # 2) arg validation
        try:
            args = tool.validate(call.arguments)
        except ValidationError as e:
            return ToolObservation(
                call_id=call.call_id,
                tool_name=call.tool_name,
                ok=False,
                error=f"Argument validation failed: {e}",
            )
        except Exception as e:
            return ToolObservation(
                call_id=call.call_id,
                tool_name=call.tool_name,
                ok=False,
                error=f"Argument validation failed: {type(e).__name__}: {e}",
            )

        # 3) execute with retry + backoff
        last_err: Optional[str] = None
        attempts_meta = []

        for attempt in range(1, self.config.max_attempts + 1):
            start = time.time()
            try:
                # 注意：我们 Day1 的 tool.run 是 sync 的
                # timeout 的实现方式：让工具自己尊重 timeout（今天先用工具内部实现）
                result = tool.run(args)
                latency_ms = int((time.time() - start) * 1000)

                return ToolObservation(
                    call_id=call.call_id,
                    tool_name=call.tool_name,
                    ok=True,
                    result=result,
                    error=None,
                ).model_copy(update={
                    # 把 trace 信息塞进 result meta（最小可观测）
                    "result": {
                        "data": result,
                        "trace": {
                            "attempts": attempt,
                            "latency_ms": latency_ms,
                            "attempt_logs": attempts_meta,
                        }
                    }
                })

            except TimeoutError as e:
                latency_ms = int((time.time() - start) * 1000)
                last_err = f"TimeoutError: {e}"
                attempts_meta.append({"attempt": attempt, "latency_ms": latency_ms, "error": last_err})

            except Exception as e:
                latency_ms = int((time.time() - start) * 1000)
                last_err = f"{type(e).__name__}: {e}"
                attempts_meta.append({"attempt": attempt, "latency_ms": latency_ms, "error": last_err})

            # backoff before retry (except after last attempt)
            if attempt < self.config.max_attempts:
                sleep_s = self.config.backoff_base_seconds * (2 ** (attempt - 1))
                time.sleep(sleep_s)

        return ToolObservation(
            call_id=call.call_id,
            tool_name=call.tool_name,
            ok=False,
            result={"trace": {"attempts": self.config.max_attempts, "attempt_logs": attempts_meta}},
            error=f"Failed after {self.config.max_attempts} attempts. Last error: {last_err}",
        )
Knowledge > Agent
#Knowledge #LLM #Data #Agent #数据人快速学 AI
什么是大模型的工具调用框架？
https://norushcoder.com/2026/02/23/scheme-for-tool-calling-20260223/
作者
RichyLiu
发布于
2026年2月23日
许可协议