AI Agent的工具调用机制架构设计：从单工具调用到多工具智能编排的完整方案

一、Tool Calling的架构困境——当Agent不再是单一LLM推理

LLM驱动的AI Agent与传统聊天机器人的核心差异在于能力——Agent不输出文本就结束交互，而是像人类一样使用工具、调用API、操作外部系统来完成多步骤任务。这种"推理+行动"的ReAct模式，将Tool Calling推到了Agent架构的核心位置。

然而现实中的Tool Calling远不止OpenAI Function Calling文档中展示的简单示例。一个生产级的AI Agent可能面对30+工具、跨域权限控制、并行调用优化、失败重试策略、上下文窗口管理等问题。这些问题在单工具场景下不明显，一旦进入多工具编排阶段，架构挑战就集中爆发。

当前业界Tool Calling实现处于早期阶段。LangChain/LlamaIndex用AgentExecutor串联tool+LLM，但不处理工具冲突、调用预算、长链回退。OpenAI/Anthropic的官方SDK提供了函数调用的API规范，但将编排逻辑完全交给开发者。生产环境中常见的痛点包括：

• 工具过多导致Prompt膨胀，挤占推理空间
• 并行调用时缺乏事务性保证，部分成功部分失败难以回滚
• 工具输出 token 远超输入 token，成本不可控
• 工具调用链过长导致上下文丢失

这些问题的本质是：Tool Calling 不只是 LLM 的一个能力接口，它是一套完整的分布式计算编排系统。我们需要用对待微服务编排的态度来设计它。

二、五层架构——从协议层到编排层的工具调用基础设施

经过多个 Agent 项目的工程实践，我归纳出一套五层 Tool Calling 架构：

该架构自下而上分为五个核心层级，层层递进且相互协作：

1. L1 协议层：定义工具的输入输出契约。
2. L2 注册与发现层：解决工具规模带来的管理问题。
3. L3 执行层：管理并发度、收集结果、处理部分失败。
4. L4 编排层：将工具调用升级为多轮迭代的执行计划。
5. L5 观测层：提供调用链路追踪与成本核算等保障。
   其中，L5 观测层的数据会反馈优化 L4 编排层与 L2 注册层，形成闭环。

L1 协议层定义工具的输入输出契约。单靠 JSON Schema 不足以表达完整语义，需要附加参数间的互斥/依赖关系、副作用声明（只读/写操作）、幂等性标志。一个典型的 Tool Schema 应包含 name、description、parameters(JSON Schema)、returns、side_effects、is_idempotent 六个字段。

L2 注册与发现层解决工具规模带来的管理问题。当工具数量超过 10 个，就需要自动化的工具筛选——LLM 不应看到全部工具定义，而应根据当前意图动态检索 Top-K 相关工具。这里可以借鉴 RAG（检索增强生成）的思路：将 Tool 的 description 和参数 Schema 做 Embedding，与用户意图做语义匹配，只注入相关性最高的 3-5 个工具到 Prompt 中。

L3 执行层的复杂度在并行调用场景中集中体现。当 Plan 阶段生成多个可并行的工具调用（如同时查天气、查日历、查新闻），L3 需要管理并发度、收集结果、处理部分失败。关键设计是保证每个工具调用的隔离性——一个工具的失败不应导致其他工具结果丢失。

L4 编排层是最具挑战的一层。它将工具调用从单个 LLM 请求的点式交互，升级为多轮迭代的执行计划。经典的设计模式是 Plan-then-Execute：第一步让 LLM 生成执行计划（DAG 图），第二步按 DAG 执行并收集结果，第三步根据结果决定是否需要 Re-plan。

L5 观测层是所有保障的基石。每个工具调用必须具备：唯一 Trace ID、耗时、输入 Token 数、输出 Token 数、成功/失败状态、错误详情。这些数据用于实时监控和事后优化——哪些工具调用频次高但成功率低、哪些工具输出过长导致 Token 浪费、哪些工具组合经常一起出现。

三、多工具编排的核心引擎——Plan-Execute-Replan的设计与实现

以下是多工具编排引擎的核心实现代码：

"""
AI Agent 多工具编排引擎
支持 Plan → Execute → Replan 的循环执行模式
"""

from __future__ import annotations

import asyncio
import json
import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Optional


class ToolSideEffect(str, Enum):
    READ_ONLY = "read_only"
    WRITE = "write"
    DESTRUCTIVE = "destructive"


class PlanStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    PARTIAL_FAILURE = "partial_failure"
    FAILED = "failed"


@dataclass
class ToolDefinition:
    """工具定义"""

    name: str
    description: str
    parameters: dict  # JSON Schema
    returns: Optional[dict] = None
    side_effect: ToolSideEffect = ToolSideEffect.READ_ONLY
    is_idempotent: bool = True
    timeout_seconds: int = 30
    max_retries: int = 2
    handler: Optional[Callable] = None


@dataclass
class PlanStep:
    """执行计划中的一个步骤"""

    step_id: str
    tool_name: str
    arguments: dict
    depends_on: list[str] = field(default_factory=list)
    status: PlanStatus = PlanStatus.PENDING
    result: Any = None
    error: Optional[str] = None
    start_time: Optional[float] = None
    end_time: Optional[float] = None


@dataclass
class ExecutionContext:
    """执行上下文：跨步骤共享的变量空间"""

    variables: dict = field(default_factory=dict)
    conversation_history: list[dict] = field(default_factory=list)
    token_usage: dict[str, int] = field(default_factory=lambda: {
        "prompt_tokens": 0,
        "completion_tokens": 0,
        "tool_output_tokens": 0,
    })


class ToolRegistry:
    """工具注册中心"""

    def __init__(self, tools: Optional[list[ToolDefinition]] = None):
        self._tools: dict[str, ToolDefinition] = {}
        if tools:
            for tool in tools:
                self.register(tool)

    def register(self, tool: ToolDefinition) -> None:
        if tool.name in self._tools:
            raise ValueError(f"Tool '{tool.name}' 已注册")
        self._tools[tool.name] = tool

    def get(self, name: str) -> Optional[ToolDefinition]:
        return self._tools.get(name)

    def list_for_intent(
        self, intent: str, top_k: int = 5
    ) -> list[ToolDefinition]:
        """根据意图检索相关工具（简化版：关键词匹配）"""
        intent_lower = intent.lower()
        scored = []
        for tool in self._tools.values():
            score = 0
            desc_lower = tool.description.lower()
            for word in intent_lower.split():
                if word in desc_lower or word in tool.name.lower():
                    score += 1
            if score > 0:
                scored.append((score, tool))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [tool for _, tool in scored[:top_k]]


class ToolOrchestrator:
    """多工具编排引擎"""

    def __init__(
        self,
        registry: ToolRegistry,
        max_parallel: int = 5,
        max_total_steps: int = 20,
    ):
        self.registry = registry
        self.max_parallel = max_parallel
        self.max_total_steps = max_total_steps
        self._semaphore = asyncio.Semaphore(max_parallel)

    async def execute_plan(
        self, steps: list[PlanStep], ctx: ExecutionContext
    ) -> list[PlanStep]:
        """执行规划好的步骤列表，支持DAG依赖"""
        completed: dict[str, PlanStep] = {}
        failed_steps: list[PlanStep] = []
        all_steps = {s.step_id: s for s in steps}

        while len(completed) + len(failed_steps) < len(steps):
            ready = []
            for s in steps:
                if s.step_id in completed or s.step_id in failed_steps:
                    continue
                if all(
                    dep in completed and completed[dep].status == PlanStatus.SUCCESS
                    for dep in s.depends_on
                ):
                    ready.append(s)

            if not ready:
                # 检查是否有死锁
                remaining = {
                    s.step_id for s in steps
                    if s.step_id not in completed and s.step_id not in failed_steps
                }
                unresolved = {
                    sid for s in steps
                    if s.step_id in remaining
                    for sid in s.depends_on
                    if sid in remaining or sid in failed_steps
                }
                if not unresolved:
                    break
                raise RuntimeError(f"执行计划死锁: {unresolved}")

            tasks = [self._execute_step(step, ctx) for step in ready]
            results = await asyncio.gather(*tasks, return_exceptions=True)

            for step, result in zip(ready, results):
                if isinstance(result, Exception):
                    step.status = PlanStatus.FAILED
                    step.error = str(result)
                    failed_steps.append(step)
                else:
                    completed[step.step_id] = step

            if failed_steps:
                print(f"警告: {len(failed_steps)} 个步骤执行失败")

            if len(completed) + len(failed_steps) >= self.max_total_steps:
                break

        return list(completed.values()) + failed_steps

    async def _execute_step(
        self, step: PlanStep, ctx: ExecutionContext
    ) -> PlanStep:
        """执行单个步骤"""
        async with self._semaphore:
            step.start_time = time.time()
            step.status = PlanStatus.RUNNING

            tool = self.registry.get(step.tool_name)
            if tool is None:
                step.status = PlanStatus.FAILED
                step.error = f"工具 '{step.tool_name}' 未注册"
                return step

            if tool.handler is None:
                step.status = PlanStatus.FAILED
                step.error = f"工具 '{step.tool_name}' 无执行器"
                return step

            for attempt in range(tool.max_retries + 1):
                try:
                    def resolve_args(args: dict, ctx_: ExecutionContext) -> dict:
                        """解析参数中的变量引用 {$prev_step.result.field}"""
                        resolved = {}
                        for k, v in args.items():
                            if isinstance(v, str) and v.startswith("{$"):
                                path = v[2:-1].strip()
                                parts = path.split(".")
                                val = ctx_.variables
                                for p in parts:
                                    if isinstance(val, dict):
                                        val = val.get(p)
                                    else:
                                        val = getattr(val, p, None)
                                resolved[k] = val
                            else:
                                resolved[k] = v
                        return resolved

                    resolved_args = resolve_args(step.arguments, ctx)
                    result = await asyncio.wait_for(
                        tool.handler(**resolved_args),
                        timeout=tool.timeout_seconds,
                    )
                    step.result = result
                    step.status = PlanStatus.SUCCESS
                    ctx.variables[step.step_id] = result
                    return step

                except asyncio.TimeoutError:
                    if attempt < tool.max_retries:
                        await asyncio.sleep(2 ** attempt)
                        continue
                    step.error = f"超时({tool.timeout_seconds}s)"
                except Exception as e:
                    if attempt < tool.max_retries:
                        await asyncio.sleep(2 ** attempt)
                        continue
                    step.error = f"执行异常: {type(e).__name__}: {e}"

            step.status = PlanStatus.FAILED
            return step


class PlanGenerator:
    """执行计划生成器——与LLM交互生成Plan"""

    def build_plan_prompt(
        self, user_intent: str, tools: list[ToolDefinition]
    ) -> str:
        """构建Plan生成Prompt"""
        tool_descriptions = []
        for t in tools:
            params_str = json.dumps(t.parameters, ensure_ascii=False, indent=2)
            tool_descriptions.append(
                f"- **{t.name}**: {t.description}\n"
                f"  参数Schema: {params_str}\n"
                f"  {'只读' if t.side_effect == ToolSideEffect.READ_ONLY else '写操作'}"
            )
        tools_block = "\n\n".join(tool_descriptions)

        return f"""你是一个任务规划器。根据用户意图和可用工具，生成执行计划。

## 用户意图
{user_intent}

## 可用工具
{tools_block}

## 输出格式（JSON）
{{
  "plan": [
    {{
      "step_id": "step_1",
      "tool_name": "工具名称",
      "arguments": {{"参数名": "参数值"}},
      "depends_on": [],
      "reason": "为什么这一步是必要的"
    }}
  ],
  "explanation": "整体执行思路说明"
}}

## 要求
1. 每一步必须有明确的 step_id (step_1, step_2, ...)
2. 如果后续步骤依赖前置步骤的输出，用 {$step_id.field} 引用
3. 优先并行安排无依赖关系的步骤
4. 步骤数量控制在10个以内
5. 只输出JSON，不要其他内容"""


# ===== 使用示例 =====
async def example_weather_travel_agent():
    """示例：天气预报+行程规划Agent"""
    import random

    async def get_weather(city: str) -> dict:
        await asyncio.sleep(0.3)
        temps = ["晴 25°C", "多云 22°C", "小雨 18°C", "晴 30°C"]
        return {"city": city, "weather": random.choice(temps)}

    async def search_flights(from_city: str, to_city: str) -> dict:
        await asyncio.sleep(0.5)
        return {
            "flights": [
                {"flight": "CA1234", "price": 890, "duration": "2h15m"},
                {"flight": "MU5678", "price": 750, "duration": "2h40m"},
            ]
        }

    async def recommend_activities(city: str, weather: str) -> list:
        await asyncio.sleep(0.2)
        return [f"{city}室内博物馆", f"{city}特色美食街"]

    registry = ToolRegistry()
    registry.register(
        ToolDefinition(
            name="get_weather",
            description="查询城市天气",
            parameters={
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
            handler=get_weather,
        )
    )
    registry.register(
        ToolDefinition(
            name="search_flights",
            description="搜索航班",
            parameters={
                "type": "object",
                "properties": {
                    "from_city": {"type": "string"},
                    "to_city": {"type": "string"},
                },
                "required": ["from_city", "to_city"],
            },
            handler=search_flights,
        )
    )
    registry.register(
        ToolDefinition(
            name="recommend_activities",
            description="根据天气推荐活动",
            parameters={
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "weather": {"type": "string"},
                },
                "required": ["city", "weather"],
            },
            handler=recommend_activities,
        )
    )

    orchestrator = ToolOrchestrator(registry, max_parallel=3)

    plan = [
        PlanStep(
            step_id="step_1",
            tool_name="get_weather",
            arguments={"city": "北京"},
        ),
        PlanStep(
            step_id="step_2",
            tool_name="get_weather",
            arguments={"city": "上海"},
        ),
        PlanStep(
            step_id="step_3",
            tool_name="search_flights",
            arguments={"from_city": "北京", "to_city": "上海"},
        ),
        PlanStep(
            step_id="step_4",
            tool_name="recommend_activities",
            arguments={
                "city": "上海",
                "weather": "{$step_2.weather}",
            },
            depends_on=["step_2"],
        ),
    ]

    ctx = ExecutionContext()
    results = await orchestrator.execute_plan(plan, ctx)

    for step in results:
        print(f"[{step.status.value}] {step.tool_name}: {step.result or step.error}")

并行调度的核心设计：

• 使用 asyncio.Semaphore 控制并发度，防止工具并发撑爆下游API
• 通过 depends_on 字段表达步骤间的DAG依赖关系
• 每一步支持独立的超时和重试配置
• {$step_id.field} 语法实现步骤间数据传递

Re-plan机制：Plan生成不是一次性的。当上一步工具返回意外结果时，编排引擎需要将中间结果和当前状态重新提交给LLM，让LLM生成修正后的执行计划。Re-plan的触发条件包括：工具执行失败、返回空结果、返回数据的置信度低于阈值。

四、上下文窗口管理——工具调用中的Token预算控制

工具调用最隐蔽的成本在于上下文窗口。一个典型的场景：Agent调用搜索工具返回10000字结果，然后调用翻译工具翻译这10000字，接着调用摘要工具……每一步都在累加，上下文窗口快速膨胀。

上下文管理策略需要分层设计：

工具输出压缩（针对L4编排层的上下文管理）。不是将工具原始输出全量注入Prompt，而是根据后续步骤的需要做信息提取：

• 搜索工具→只保留Top-3结果的核心摘要
• 数据库查询工具→只保留聚合统计和非空字段
• API调用工具→截断到N tokens并标注截断信息

分层上下文（Short-term vs Long-term Memory）。将工具调用的完整记录存储在向量数据库中作为长期记忆，当前Prompt中只保留当前步骤的调用链摘要。当后续推理需要完整上下文时，通过RAG检索相关历史记录。

Token预算API设计：

class TokenBudget:
    """Token预算管理器"""

    def __init__(self, max_tokens: int = 8000, tool_result_ratio: float = 0.3):
        self.max_tokens = max_tokens
        self.tool_cap = int(max_tokens * tool_result_ratio)
        self.used_prompt = 0
        self.used_tool_output = 0

    def can_inject_tool_result(self, token_count: int) -> bool:
        return (self.used_tool_output + token_count) <= self.tool_cap

    def record_tool_result(self, token_count: int) -> None:
        self.used_tool_output += token_count

    def is_exhausted(self) -> bool:
        total = self.used_prompt + self.used_tool_output
        return total > self.max_tokens * 0.9

实践中，工具输出通常限制在500-1000 tokens，超出部分通过LLM摘要压缩。这不是完美的方案，但能满足80%的生产场景。

五、总结

AI Agent的工具调用架构本质上是一套分布式任务编排系统。五层架构（协议层→注册发现层→执行层→编排层→观测层）提供了一个可扩展的设计蓝图。

核心工程要点：

1. Tool Schema需要超出普通JSON Schema，附加副作用声明和幂等性标志，这是后续并行编排和错误恢复的基础
2. 多工具编排使用Plan-Execute-Replan循环，用DAG表达步骤依赖，用Semaphore控制并发度
3. 上下文窗口管理是隐性性能瓶颈——工具输出必须经过压缩再注入Prompt，否则长链工具调用会迅速耗尽上下文
4. 观测层的全链路追踪和成本核算不是可选项：Agent每次推理+工具调用的token消耗和耗时，应当像微服务的APM一样监控

从单工具到多工具的跨越，不只是LLM能力的问题，更是系统架构的工程挑战。当Agent开始像一个初级工程师一样使用多种工具解决复杂任务时，这些架构设计就从"锦上添花"变成了"必须品"。