多工具智能体 – 光头颜说IT

面向初学者与工程实践者，本文给出一套可直接运行的“多工具智能体”最小实现示例程序（完整源代码见文末），重点覆盖以下要素：

（1）通过DeepSeek的Function Calling（OpenAI兼容SDK）由模型自动选择并调用多个工具函数；

（2）以Pydantic v2建模工具参数，实现类型转换与校验，并提供“温和”的参数修复；

（3）对瞬时故障采用“指数退避 + 抖动”重试；

（4）输出“JSON Lines”结构化日志，便于观测、排障与回放；

（5）以.env文件集中管理配置（python-dotenv自动加载）；

（6）支持“多回合工具调用”，直到模型不再返回“tool_calls”为止。

一、总体架构与执行流

示例程序执行流采用“两轮对话+可能的多回合工具调用”构成。

第一轮提交系统消息（系统提示词）、用户消息（用户提示词）与工具清单（名称、描述、参数 JSON Schema）等信息。如果大模型推理分析后需要调用工具，则将返回一个或多个工具调用指令。随后智能体逐个执行工具，并把每次的结果以“role=”tool””的消息形式回灌到对话历史。若大模型仍有未完成的子任务，可在下一回合继续返回新的工具调用指令。当模型不再返回工具调用指令时，发起第二轮生成最终自然语言回答（或直接使用第一轮已给出的内容）。

这一流转确保在“一个输入包含多个子意图”的情况下，模型既可以一次返回多个工具调用，也可以分回合逐步调用，直至完成任务。

二、运行环境与配置

运行环境建议 Python 3.10+，并安装openai、pydantic、python-dotenv等依赖。配置通过.env文件集中管理（工程内附.env.example，复制为.env并填写API KEY）。

DEEPSEEK_API_KEY=sk-xxxxxx
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1
DEEPSEEK_MODEL=deepseek-chat

如果在.env文件中未设置“DEEPSEEK_API_KEY”则示例程序将进入“MOCK（离线）模式”，该模式下将以启发式规则模拟工具调用指令与最终回答，便于演示与调试；而设置“DEEPSEEK_API_KEY”则示例程序将进入“LIVE（直连 DeepSeek）模式”，即程序将实际调用大模型并依赖其推理分析能力生成工具调用指令。

三、工具设计：参数模型、业务实现与错误分层

示例程序工具以“参数模型 + 业务函数”的形式定义。参数模型由Pydantic v2提供，既用于运行时校验与类型转换，也用于对模型暴露同源的JSON Schema，这种设计确保了实现与约束一致。示例程序提供了两个工具，一个是用于执行加法运算的加法工具，一个是用于执行内容检索的搜索工具。这两个工具定义在源代码文件tools.py中，代码节选如下。

# ---- 示例工具：加法 ----
class AddArgs(BaseModel):
    a: float
    b: float

def add_tool(args: AddArgs) -> float:
    return args.a + args.b


# ---- 示例工具：迷你“搜索” ----
CORPUS = [
    "向量数据库用于相似度搜索，常见有FAISS、Milvus、PGVector。",
    "检索增强生成（RAG）通过检索外部知识提升回答的可靠性。",
    "多智能体（Multi-Agent）强调角色分工与协作。",
    "MCP 标准化了工具与资源的暴露与发现。",
    "ReAct 让模型先思考再行动，提升可追踪性。"
]

class SearchArgs(BaseModel):
    query: str = Field(min_length=1, description="关键词")
    top_k: int = Fielddefault=3, ge=1, le=5, description="返回条数（1~5）")

def search_tool(args: SearchArgs) -> List[str]:
    q = args.query.strip().lower()
    hits = [t for t in CORPUS if q in t.lower()]
    return hits[ args.top_k]

由于大模型基于概率的生成机制以及用户需求描述中潜在的不合理性，大模型工具调用指令所提供的参数可能存在错误，示例程序对此提供了修复机制。不过对“参数修复”应保持克制，仅处理可机械判定的格式与范围，例如对搜索参数query去除空白，对另一搜索参数top_k将字符串形式的参数值”10″转换为整数并夹逼到允许区间。repair_search_args()函数实现了这些规则下的参数的修复。

def repair_search_args(raw: Dict[str, Any]) -> Dict[str, Any]:
    """示例性“参数修复”：top_k 转换/夹逼，query 清洗。"""
    fixed = dict(raw)
    if "top_k" in fixed:
        try:
            fixed["top_k"] = int(float(fixed["top_k"]))
        except Exception:
            fixed["top_k"] = 3
        fixed["top_k"] = max(1, min(5, fixed["top_k"]))
    if "query" in fixed and isinstance(fixed["query"], str):
        fixed["query"] = fixed["query"].strip()
    return fixed

四、工具注册与元数据

在支持多工具调用的场景下，应用程序应建立对工具的管理机制，例如通过工具注册机制使用注册表实现对工具元数据的管理。
统一的注册表集中管理工具的名称、参数模型、实现函数、描述、最大重试次数以及可选的参数修复钩子等元数据。这些元数据在数据类ToolSpec中被定义。

@dataclass
class ToolSpec:
    name: str
    params_model: Type[BaseModel]
    func: Callable[[BaseModel], Any]
    desc: str = ""
    max_retries: int = 1
    repair: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None

而实现工具注册的类是ToolRegistry，该注册既用于构造tools列表供模型基于函数调用机制理解并返回工具调用指令，也服务于运行时路由，架设工具调用指令与工具执行之间的桥梁。ToolRegistry类的实现代码如下。

class ToolRegistry:
    def __init__(self) -> None:
        self._tools: Dict[str, ToolSpec] = {}

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

    def get(self, name: str) -> ToolSpec:
        if name not in self._tools:
            raise KeyError(f"Tool not found: {name}")
        return self._tools[name]

    def list(self) -> List[ToolSpec]:
        return list(self._tools.values())

build_registry() 函数实现了对本示例程序两个工具的注册，其代码如下。

def build_registry() -> ToolRegistry:
    reg = ToolRegistry()
    reg.register(ToolSpec(
        name="add",
        params_model=AddArgs,
        func=add_tool,
        desc="加法计算",
        max_retries=0,
    ))
    reg.registerToolSpec(
        name="search",
        params_model=SearchArgs,
        func=search_tool,
        desc="关键词检索（内置语料）",
        max_retries=2,
        repair=repair_search_args,
    ))
    return reg

五、执行器：校验→修复→执行→指数退避重试→结构化日志

执行器对每个工具调用进行统一处理，它通过execute()函数实现。

它首先根据参数所传入的工具名称取得工具的元数据；然后调用params_model(**arg_dict)完成参数类型转换与校验，如果校验失败且工具提供了参数修复能力则代码尝试执行“可推断”的修复并修复后的参数再次校验；通过校验后则调用执行工具名称所对应的业务函数；执行业务函数时若抛出TransientToolError异常则按“指数退避+抖动”策略重试并限制其重试次数，若业务函数抛出FatalToolError异常或其他未知异常则直接失败。

execute()函数的代码如下。

    def execute(self, tool_name: str, arg_dict: Dict[str, Any]) -> Tuple[bool, Any, str]:

        try:
            spec = self.registry.get(tool_name)
        except KeyError as e:
            return False, str(e), "\n".join(logs)

        # 校验
        try:
            params = spec.params_model(**arg_dict)
        except ValidationError as ve:
            if spec.repair:
                fixed = spec.repair(arg_dict)
                try:
                    params = spec.params_model(**fixed)
                    arg_dict = fixed
                except ValidationError as ve2:
                    return False, f"参数无效：{ve2.errors()}", "\n".join(logs)
            else:
                return False, f"参数无效：{ve.errors()}", "\n".join(logs)

        # 执行 + 指数退避重试
        attempts = 0
        while True:
            attempts += 1
            try:
                result = spec.func(params)  # type: ignore[arg-type]
                return True, result, "\n".join(logs)
            except TransientToolError as te:
                if attempts > spec.max_retries:
                    return False, f"工具暂时不可用：{te}", "\n".join(logs)
                # 指数退避 + 抖动
                delay = self.backoff_base * (2 ** (attempts - 1)) + random.uniform(0, self.jitter_max)
                time.sleep(delay)
                continue
            except FatalToolError as fe:
                return False, f"工具执行失败：{fe}", "\n".join(logs)
            except Exception as e:
                return False, f"工具异常：{e}", "\n".join(logs)

为便于调试与跟踪工具执行情况，在execute()函数中还添加了JSON Lines结构化日志（上述代码中已省略，完整代码可参见源文件），日志中包含每步生成的一行事件记录，包含时间戳、trace_id、事件名与关键字段。它不是多工具智能体的必须组成部分，可参考源代码理解，本文不再赘述。

六、.env加载与DeepSeek Function Calling（OpenAI 兼容 SDK）

Python项目可将所需的环境变量保存在.env文件中，程序运行时先通过python-dotenv加载.env文件内容获取其中所保存的API Key等环境变量。

获取.env文件中保存的环境变量代码如下。

# 加载项目根目录下的 .env（若存在）
load_dotenv()
# 注意：DeepSeek 兼容 OpenAI SDK 协议，需设置 base_url 与 api_key
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1")
DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "")

DeepSeek提供与OpenAI兼容的SDK调用方式。这种兼容性，一方面表现在.env文件中如果使用OpenAI的SDK所默认的环境变量名称（即OPENAI_API_KEY和OPENAI_BASE_URL）存储相关值，则在构建客户端实例（即实例化SDK所提供的OpenAI类）时不再需要显式传递API Key与Base URL；另一方面DeepSeek使用与OpenAI相同的会话API，包含其工具列表描述参数的形式与要求。

示例程序通过build_tools_payload()函数构建符合SDK要求的工具列表，其代码如下。

def pydantic_to_json_schema(model: type[BaseModel]) -> Dict[str, Any]:
    return model.model_json_schema()

def build_tools_payload(registry: ToolRegistry) -> List[Dict[str, Any]]:
    tools = []
    for t in registry.list():
        tools.append({
            "type": "function",
            "function": {
                "name": t.name,
                "description": t.desc or t.name,
                "parameters": pydantic_to_json_schema(t.params_model)
            }
        })
    return tools

上述代码在构造工具列表时，将“params_model.model_json_schema()”直接作为parameters字段传入，确保了参数约束与实现同源。

在一个智能体中，与大模型进行的交互通常会同时涉及使用工具的交互与不需要工具的交互，示例程序对这两种情况下的交互分别进行了封装。

deepseek_chat()函数对需要使用工具的场景进行封装，它在调用大模型会话API时传入了tools和tool_choice参数，并在大模型返回的响应中抽取其中的工具执行指令形成一个字典对象供上层使用。

_client = None
def _client_live() -> OpenAI:
    global _client
    if _client is None:
        _client = OpenAI(base_url=DEEPSEEK_BASE_URL, api_key=DEEPSEEK_API_KEY)
    return _client

def deepseek_chat(messages: List[Dict[str, Any]], tools: List[Dict[str, Any]]) -> Dict[str, Any]:
    """第一轮：让模型生成 tool_calls。"""

    client = _client_live()
    resp = client.chat.completions.create(
        model=DEEPSEEK_MODEL,
        messages=messages,
        tools=tools,
        tool_choice="auto",
        temperature=0.1,
    )
    # 统一返回一个 dict，结构与 OpenAI HTTP 响应近似，便于上层复用
    return {"choices": [{"message": {
        "role": resp.choices[0].message.role,
        "content": resp.choices[0].message.content,
        "tool_calls": [
            {
                "id": tc.id,
                "type": tc.type,
                "function": {"name": tc.function.name, "arguments": tc.function.arguments}
            } for tc in (resp.choices[0].message.tool_calls or [])
        ]
    }}]}

deepseek_chat_no_tools()函数则对无需使用工具的场景进行封装，它没有tools和tool_choice参数，并且抽取大模型返回响应中的角色和内容构成一个字典返回供上层使用。

def deepseek_chat_no_tools(messages: List[Dict[str, Any]]) -> Dict[str, Any]:
    """第二轮：不带 tools，让模型基于 tool 回执生成最终回答。"""

    client = _client_live()
    resp = client.chat.completions.create(
        model=DEEPSEEK_MODEL,
        messages=messages,
        temperature=0.2,
    )
    return {"choices": [{"message": {
        "role": resp.choices[0].message.role,
        "content": resp.choices[0].message.content
    }}]

在本示例智能体中，响应用户的输入时将首先调用deepseek_chat()函数，以支持潜在可能的工具调用需求，然后调用deepseek_chat_no_tools()函数获得大模型的最终回复。当然，在代码实现策略上，也完全可以将deepseek_chat()和deepseek_chat_no_tools()两个函数合并为一个统一的封装函数，即无论什么场景下都向大模型提交工具信息，在大模型返回响应后，对响应的结构进行分析，然后根据是否有工具调用指令分别向上层返回不同结构的字典。

七、多回合工具调用的编排

在包含多个子意图的输入场景下，智能体在调用模型后，模型返回的工具调用指令可能一次包含多个工具调用要求，也可能在多次交互中连续返回工具调用指令，或者在更复杂情况下，可能是这两种情况的组合。

run_agent()函数实现了上述多回合工具调用的编排，它将调用封装函数deepseek_chat()并获取其回复，只要模型在回复中返回“tool_calls”，它就执行其中所包含的所有函数调用并回灌函数调用结果到会话消息列表中。而所有工具调用执行完毕后，它将在一个被限定最大循环次数（max_rounds）的循环中继续调用deepseek_chat()函数，直至大模型不再返回“tool_calls”或已达到最大限定循环次数为止。

run_agent()函数的实现代码如下。

def run_agent(user_text: str, max_rounds: int = 5) -> None:
    registry = build_registry()
    executor = Executor(registry)
    tools_payload = build_tools_payload(registry)

    messages: List[Dict[str, Any]] = [
        {"role": "system", "content": (
            "你是一个会合理使用工具的助理。"
            "当用户需求包含多个子任务时，可以在一次或多次 tool_calls 中逐步完成，"
            "直到所有子任务完成为止。若无需工具则直接给出回答。"
            "最终回答阶段请给出整洁的自然语言输出，不要在最终回答里包含任何与工具相关的标记。"
        )},
        {"role": "user", "content": user_text},
    ]

    # 允许模型跨多回合逐步提出工具调用
    for _ in range(max_rounds):
        resp = deepseek_chat(messages, tools_payload)
        msg = resp["choices"][0]["message"]
        tool_calls = msg.get("tool_calls") or []
        content = (msg.get("content") or "").strip() if msg.get("content") else ""

        if not tool_calls:
            # 没有工具调用时，若已给出内容则直接作为最终回答；否则请求一次无工具的总结
            if content:
                print("=== 最终回答 ===")
                print(content)
                return
            final_resp = deepseek_chat_no_tools(messages)
            final_msg = final_resp["choices"][0]["message"]["content"]
            print("=== 最终回答 ===")
            print(final_msg)
            return

        # 执行本轮所有 tool_calls
        for call in tool_calls:
            name = call["function"]["name"]
            args = json.loads(call["function"]["arguments"] or "{}")
            ok, payload, debug = executor.execute(name, args)

            # 调试输出
            print(f"--- ToolCall {name} ---")
            print(debug)
            print("[result]" if ok else "[failed]", payload, "\n")

            # 把工具调用与结果回灌给模型
            messages.append({
                "role": "assistant",
                "tool_calls": [call],
                "content": None
            })
            messages.append({
                "role": "tool",
                "tool_call_id": call["id"],
                "name": name,
                "content": json.dumps({"ok": ok, "data": payload}, ensure_ascii=False)
            })

    # 达到最大轮次仍未给出最终回答，做一次总结后返回
    final_resp = deepseek_chat_no_tools(messages)
    final_msg = final_resp["choices"][0]["message"]["content"]
    print("=== 最终回答 ===")
    print(final_msg)

至此，具备调用多个工具能力的智能体开发完毕。

八、测试运行

智能体开发完成后，可以添加测试代码测试运行智能体。以下测试示例使用一个同时包含“检索”与“加法”的输入，以便观察多工具调用链路。

if __name__ == "__main__":
    # 示例输入：包含搜索与加法两个意图
    user_text = '帮我搜“向量数据库”，要10条；顺便把 1.2 + "3" 算一下。'
    run_agent(user_text)

运行测试代码，在运行窗口可以依次看到图 1、图 2、图 3所示的输出。

图 1 连续两次调用搜索工具

在图1中可见搜索工具被调用了两次，其原因是测试代码输入的要10条向量数据库的结果（虽然在代码逻辑中它被修正夹逼到5条），但由于第一次搜索时仅搜索出1条结果，因而大模型发出了再次搜索的指令。

此外，在第二次搜索时，输出信息包含exec.retryable、backoff.sleep、0.323s等信息。这是由于在搜索工具中设定了20%概率的瞬时错误（该错误由_flaky()函数实现，正文未提及，可参见完整源代码）以模拟真实场景中可能发生的错误。而发生错误后，示例程序并未直接抛出异常，而是按设计中的“指数退避+抖动”策略重试。