Agents-as-Tools 模式:专家 Agent 包装成工具
解决什么问题
旅行助手经常需要调用"专家"——签证顾问、汇率计算器、当地美食推荐——但不想让主 Agent 把所有领域知识塞进系统 prompt。 Agents-as-Tools 把每个专家 Agent 包装成一个 Tool,主 Agent 像调函数一样调专家,拿到结果就走。
| 场景 | 为什么选它 |
|---|---|
| 旅行助手调签证专家 | 签证规则复杂,单独封装更好维护 |
| 编程助手调代码审查专家 | 审查逻辑跟生成逻辑完全不同 |
| 客服调退款/物流专家 | 各业务线独立迭代 |
复杂度
⭐⭐ 中低。本质就是 Tool 的回调函数里跑另一个 Agent。
和其他模式的关系
| 模式 | 谁决定下一步 | 什么时候用 |
|---|---|---|
| Agents-as-Tools | 主 Agent 在对话中按需 tool_call | 专家能力需要封装复用,主 Agent 保持控制权 |
| Manager-Worker | Manager 先拆任务再分发 | 子任务已知、可并行批量处理 |
| Handoff | Router 把控制权整个交出去 | 用户意图切换,新 Agent 直接面对用户 |
Agents-as-Tools 比 Manager-Worker 更灵活(不用预先拆解),比 Handoff 更集中(主 Agent 始终持有回复权)。
角色关系
┌──────────────────────┐
│ 主 Agent │
│ (旅行规划师) │
└──┬──────┬──────┬─────┘
│ │ │ tool_call(...)
▼ ▼ ▼
┌─────┐┌─────┐┌─────┐
│签证 ││汇率 ││美食 │ ← 每个都是独立的 Agent
│专家 ││专家 ││专家 │ 被包成 Tool 暴露给主 Agent
└─────┘└─────┘└─────┘
- 主 Agent 通过 tool_call 调用专家
- 专家 Agent 内部可有自己的 prompt、记忆、甚至 sub-tool
- 专家返回结构化结果,主 Agent 拼装最终回复
完整 Python 实现
"""
Agents-as-Tools 旅行助手示例
主 Agent (旅行规划师) 把签证专家、汇率专家、美食专家包装成 Tool。
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Callable
# ── MockLLM ──────────────────────────────────────────────
class MockLLM:
"""模拟 LLM,根据 role 和 prompt 关键词返回预设回复。"""
def __init__(self):
self.call_log: list[dict] = []
def chat(self, role: str, prompt: str) -> str:
self.call_log.append({"role": role, "prompt": prompt[:80]})
# 主 Agent 决定调哪些工具
if role == "planner" and "需要哪些工具" in prompt:
return json.dumps({"tool_calls": ["visa_expert", "currency_expert", "food_expert"]})
# 主 Agent 合并结果
if role == "planner" and "整合" in prompt:
return "【日本7日游完整建议】签证: 需单次旅游签 7工作日 | 预算: ¥8400≈17.5万日元 | 必吃: 寿司大→一蘭拉面→烤肉亭"
# 签证专家 Agent
if role == "visa_expert":
return json.dumps({
"国家": "日本",
"签证类型": "单次旅游签",
"办理周期": "7个工作日",
"所需材料": ["护照", "在职证明", "银行流水", "照片2张"],
})
# 汇率专家 Agent
if role == "currency_expert":
return json.dumps({
"人民币_日元": 20.85,
"建议换汇金额": "¥8400 ≈ ¥175,140 JPY",
"换汇渠道": "出发前银行预约,到了用 ATM 取现",
})
# 美食专家 Agent
if role == "food_expert":
return json.dumps({
"必吃清单": [
{"店": "寿司大", "地点": "筑地", "人均": "¥300"},
{"店": "一蘭拉面", "地点": "新宿", "人均": "¥80"},
{"店": "烤肉亭", "地点": "涩谷", "人均": "¥200"},
],
})
return "{}"
# ── Expert Agent (被包装成 Tool) ─────────────────────────
@dataclass
class ExpertAgent:
"""一个专家 Agent,拥有自己的 system_prompt 和 LLM 调用逻辑。"""
name: str
role: str
system_prompt: str
llm: MockLLM = field(repr=False)
def run(self, query: str) -> dict:
full_prompt = f"[系统] {self.system_prompt}\n[用户] {query}"
raw = self.llm.chat(self.role, full_prompt)
try:
return json.loads(raw)
except json.JSONDecodeError:
return {"raw": raw}
# ── Tool 注册表 ──────────────────────────────────────────
@dataclass
class ToolRegistry:
"""把 ExpertAgent 注册为 Tool,供主 Agent 按名字调用。"""
tools: dict[str, Callable[[str], dict]] = field(default_factory=dict)
def register(self, name: str, agent: ExpertAgent):
self.tools[name] = agent.run
def call(self, name: str, query: str) -> dict:
fn = self.tools.get(name)
if fn is None:
return {"error": f"未知工具: {name}"}
return fn(query)
# ── 主 Agent (旅行规划师) ────────────────────────────────
@dataclass
class PlannerAgent:
llm: MockLLM = field(repr=False)
registry: ToolRegistry = field(repr=False)
def plan(self, user_request: str) -> str:
print(f"[Planner] 收到: {user_request}")
# Step 1: 决定需要调哪些专家工具
decision_raw = self.llm.chat("planner", f"用户想去日本,需要哪些工具?{user_request}")
tool_calls = json.loads(decision_raw).get("tool_calls", [])
print(f"[Planner] 决定调用工具: {tool_calls}")
# Step 2: 依次调用每个专家 Tool
tool_results: dict[str, dict] = {}
for tool_name in tool_calls:
result = self.registry.call(tool_name, user_request)
tool_results[tool_name] = result
print(f" [Tool:{tool_name}] → {json.dumps(result, ensure_ascii=False)[:60]}...")
# Step 3: 让 LLM 整合所有工具结果
merge_prompt = f"请整合以下工具结果: {json.dumps(tool_results, ensure_ascii=False)}"
final = self.llm.chat("planner", merge_prompt)
print(f"[Planner] 最终回复: {final}")
return final
# ── 运行入口 ─────────────────────────────────────────────
def main():
llm = MockLLM()
# 创建专家 Agents
visa = ExpertAgent("签证专家", "visa_expert", "你是签证政策专家,熟悉各国签证要求。", llm)
currency = ExpertAgent("汇率专家", "currency_expert", "你是外汇专家,提供实时汇率建议。", llm)
food = ExpertAgent("美食专家", "food_expert", "你是日本美食达人,推荐当地必吃餐厅。", llm)
# 注册为 Tool
registry = ToolRegistry()
registry.register("visa_expert", visa)
registry.register("currency_expert", currency)
registry.register("food_expert", food)
# 主 Agent 运行
planner = PlannerAgent(llm=llm, registry=registry)
result = planner.plan("帮我规划日本7日游,预算8000-10000元")
# 验证
assert "签证" in result
assert "寿司" in result or "拉面" in result
assert len(llm.call_log) == 5 # planner决策 + 3专家 + planner合并
print(f"\n✓ 共 {len(llm.call_log)} 次 LLM 调用,测试通过")
if __name__ == "__main__":
main()
运行记录
[Planner] 收到: 帮我规划日本7日游,预算8000-10000元
[Planner] 决定调用工具: ['visa_expert', 'currency_expert', 'food_expert']
[Tool:visa_expert] → {"国家": "日本", "签证类型": "单次旅游签", "办理周期": "7个工作日"...
[Tool:currency_expert] → {"人民币_日元": 20.85, "建议换汇金额": "¥8400 ≈ ¥175,14...
[Tool:food_expert] → {"必吃清单": [{"店": "寿司大", "地点": "筑地", "人均": "¥300"}...
[Planner] 最终回复: 【日本7日游完整建议】签证: 需单次旅游签 7工作日 | 预算: ¥8400≈17.5万日元 | 必吃: 寿司大→一蘭拉面→烤肉亭
✓ 共 5 次 LLM 调用,测试通过
踩坑记录
| 坑 | 现象 | trace 信号 | 修法 |
|---|---|---|---|
| 专家返回太长 | 主 Agent 上下文爆了 | tool_call 返回 token 数远超主 Agent 上下文窗口 | 专家返回前做摘要,或限制 max_tokens |
| 递归调用 | 专家 A 调专家 B 调专家 A | call_log 中出现 A→B→A 的循环调用链 | 设最大调用深度,禁止循环依赖 |
| 工具选择错误 | 问机票却调了美食专家 | tool_calls 列表中出现与用户意图无关的工具名 | 在 tool 描述里写清楚适用范围 |
| 延迟叠加 | 3 个专家串行调用很慢 | 总耗时 ≈ 各专家耗时之和,无并发 | 无依赖的专家可以并行调用 (asyncio) |
工程备忘
- Tool 描述要精准:主 Agent 靠描述选工具,描述写模糊 = 选错工具。
- 专家 Agent 可以有 sub-tool:比如签证专家内部可以查数据库,主 Agent 不需要知道。
- 隔离上下文:每个专家 Agent 的对话历史互相独立,不要共享 chat_history。
- fallback 机制:如果专家 Agent 超时或返回垃圾,主 Agent 要能优雅降级。
- 与 Manager-Worker 的区别:Manager-Worker 是 Manager 主动拆任务;Agents-as-Tools 是主 Agent 在对话过程中按需调用,更灵活。
- 成本控制:每调一个专家就是一次 LLM 调用,注意监控 token 消耗。
- 版本管理:专家 Agent 的 prompt 独立管理,可以单独 A/B 测试。
读完以后
如果专家不只是提供数据、而是要接管整个对话,看 Handoff。 如果子任务可以预先拆解、批量并行执行,看 Manager-Worker。 如果专家之间需要讨论协商,看 Group Chat。