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 架构：
该架构自下而上分为五个核心层级，层层递进且相互协作：
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的工具调用架构本质上是一套分布式任务编排系统。五层架构（协议层→注册发现层→执行层→编排层→观测层）提供了一个可扩展的设计蓝图。
核心工程要点：
从单工具到多工具的跨越，不只是LLM能力的问题，更是系统架构的工程挑战。当Agent开始像一个初级工程师一样使用多种工具解决复杂任务时，这些架构设计就从"锦上添花"变成了"必须品"。
