单Agent + 工具调用模式
概述与核心特点
单Agent + 工具调用模式是最基础且广泛应用的LLM Agent架构模式。该模式通过单一的智能体（Agent）协调多个外部工具或API，实现复杂任务的自动化处理。其核心特征包括：
- 统一控制点：由单一Agent负责所有决策和编排
- 工具驱动执行：通过标准化的工具接口调用外部服务
- 轻量级架构：相比多Agent系统，架构复杂度大幅降低
- 快速响应：减少Agent间通信开销，执行效率高
架构设计原理
核心组件架构
graph TB
A[用户输入] --> B[单Agent]
B --> C[意图识别]
C --> D[工具选择器]
D --> E[工具调用器]
E --> F[外部工具/API]
F --> G[结果聚合器]
G --> H[响应生成器]
H --> I[用户输出]
J[工具注册表] --> D
K[上下文管理器] --> C
L[状态存储] --> G工作流程设计
class SingleAgentWithTools:
def __init__(self):
self.tools = ToolRegistry()
self.context_manager = ContextManager()
self.planner = SimplePlanner()
async def process_request(self, user_input: str) -> str:
# 1. 意图识别与上下文理解
intent = await self.understand_intent(user_input)
# 2. 简单规划：确定所需工具和执行顺序
plan = self.planner.create_simple_plan(intent)
# 3. 按序执行工具调用
results = []
for step in plan.steps:
tool_result = await self.execute_tool(step)
results.append(tool_result)
# 4. 结果聚合与响应生成
response = self.aggregate_results(results)
return response工具调用标准化设计
工具接口规范
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from dataclasses import dataclass
@dataclass
class ToolCall:
tool_name: str
parameters: Dict[str, Any]
timeout: Optional[int] = 30
retry_count: int = 3
@dataclass
class ToolResult:
success: bool
data: Any
error: Optional[str] = None
execution_time: float = 0.0
cost: Optional[float] = None
class BaseTool(ABC):
"""标准化工具基类"""
@abstractmethod
async def execute(self, **params) -> ToolResult:
pass
@property
@abstractmethod
def schema(self) -> Dict[str, Any]:
"""工具参数Schema定义"""
pass
@property
@abstractmethod
def description(self) -> str:
"""工具功能描述"""
pass
class WebSearchTool(BaseTool):
"""网页搜索工具示例"""
async def execute(self, query: str, limit: int = 10) -> ToolResult:
try:
start_time = time.time()
# 实际搜索逻辑
results = await self._search_web(query, limit)
execution_time = time.time() - start_time
return ToolResult(
success=True,
data=results,
execution_time=execution_time
)
except Exception as e:
return ToolResult(
success=False,
data=None,
error=str(e)
)
@property
def schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"limit": {"type": "integer", "default": 10, "description": "结果数量限制"}
},
"required": ["query"]
}
@property
def description(self) -> str:
return "进行网页搜索，返回相关结果"工具注册与发现机制
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, BaseTool] = {}
self._schemas: Dict[str, Dict] = {}
def register_tool(self, tool: BaseTool, name: str = None):
"""注册工具"""
tool_name = name or tool.__class__.__name__
self._tools[tool_name] = tool
self._schemas[tool_name] = tool.schema
def get_tool(self, name: str) -> Optional[BaseTool]:
"""获取工具实例"""
return self._tools.get(name)
def list_available_tools(self) -> List[str]:
"""列出所有可用工具"""
return list(self._tools.keys())
def get_tool_schema(self, name: str) -> Optional[Dict]:
"""获取工具Schema"""
return self._schemas.get(name)
def match_tool(self, description: str) -> Optional[str]:
"""基于描述匹配最适合的工具"""
# 简单的关键词匹配，实际可使用更复杂的算法
for name, tool in self._tools.items():
if description.lower() in tool.description.lower():
return name
return None
# 工具使用示例
registry = ToolRegistry()
registry.register_tool(WebSearchTool(), "search_web")
registry.register_tool(CalculatorTool(), "calculate")
registry.register_tool(WeatherTool(), "get_weather")智能体决策逻辑
意图识别与工具选择
class IntentBasedToolSelector:
def __init__(self, registry: ToolRegistry, llm_client):
self.registry = registry
self.llm = llm_client
async def select_tool(self, user_intent: str, context: Dict) -> str:
"""基于用户意图选择最适合的工具"""
# 构建工具描述列表
tools_info = []
for name in self.registry.list_available_tools():
tool = self.registry.get_tool(name)
tools_info.append(f"- {name}: {tool.description}")
prompt = f"""
用户意图: {user_intent}
可用工具:
{chr(10).join(tools_info)}
请选择最合适的工具名称，只返回工具名称。
"""
response = await self.llm.chat_completion(prompt)
selected_tool = response.strip()
# 验证选择是否有效
if selected_tool not in self.registry.list_available_tools():
# 回退到基于关键词的匹配
return self.registry.match_tool(user_intent) or "search_web"
return selected_tool
class SimplePlanner:
def __init__(self):
self.complexity_threshold = 3 # 复杂任务阈值
def create_simple_plan(self, intent: str) -> "ExecutionPlan":
"""创建简单的执行计划"""
# 基于意图判断任务复杂度
if self._is_simple_task(intent):
return ExecutionPlan([ExecutionStep("single_tool", intent)])
else:
# 复杂任务的多步骤计划
return ExecutionPlan([
ExecutionStep("analyze", f"分析任务: {intent}"),
ExecutionStep("execute", intent),
ExecutionStep("validate", "验证结果")
])
def _is_simple_task(self, intent: str) -> bool:
"""判断是否为简单任务"""
simple_indicators = [
"查询", "搜索", "计算", "获取", "今天", "现在"
]
return any(indicator in intent for indicator in simple_indicators)
@dataclass
class ExecutionStep:
step_type: str
description: str
required_tools: List[str] = None
def __post_init__(self):
if self.required_tools is None:
self.required_tools = []
@dataclass
class ExecutionPlan:
steps: List[ExecutionStep]
total_estimated_time: float = 0.0
def add_step(self, step: ExecutionStep):
self.steps.append(step)上下文与状态管理
上下文管理器设计
class ContextManager:
def __init__(self, max_history: int = 10):
self.conversation_history = []
self.working_memory = {}
self.max_history = max_history
async def add_interaction(self, user_input: str, agent_response: str):
"""添加一轮对话到历史记录"""
self.conversation_history.append({
"user": user_input,
"agent": agent_response,
"timestamp": time.time()
})
# 保持历史记录
if len(self.conversation_history) > self.max_history:
self.conversation_history = self.conversation_history[-self.max_history:]规划—执行—评估的ReAct/Plan-Execute模式
模式概述
ReAct（Reasoning and Acting）与Plan-Execute模式是当前LLM Agent架构中最为基础且广泛应用的设计模式之一。该模式将复杂任务的处理过程明确划分为三个核心阶段：规划阶段负责任务分解与策略制定，执行阶段负责按计划逐步实施工具调用，评估阶段负责结果验证、错误纠正与策略调整。
这种分层架构设计带来了显著优势：通过将推理与行动分离，模式能够更好地处理多步骤复杂任务；通过引入评估机制，实现了自我纠错和策略迭代；通过标准化的工作流程，提升了系统的可预测性和可维护性。
三阶段核心机制
1. 规划阶段（Planning）
规划阶段的核心是将复杂的用户请求转化为可执行的行动序列。这一过程通常包含以下关键步骤：
任务分解：将高层目标拆解为具体的子任务序列
依赖分析：识别任务间的依赖关系和执行顺序
策略制定：选择合适的工具调用策略和参数配置
预算规划：预估资源消耗和时间成本
在技术实现上，规划阶段通常采用层级规划（Hierarchical Planning）方法：
class TaskPlanner:
def __init__(self, llm_client):
self.llm_client = llm_client
self.tools = self._register_available_tools()
def plan(self, user_request, context=None):
# 生成初始规划
initial_plan = self._generate_initial_plan(user_request, context)
# 任务分解
subtasks = self._decompose_tasks(initial_plan)
# 依赖分析
dependency_graph = self._analyze_dependencies(subtasks)
# 优化执行序列
optimized_sequence = self._optimize_execution_order(dependency_graph)
return {
'tasks': optimized_sequence,
'dependencies': dependency_graph,
'estimated_cost': self._estimate_cost(optimized_sequence)
}
def _generate_initial_plan(self, request, context):
prompt = f"""
作为一个任务规划助手，请将以下任务分解为具体的执行步骤：
用户请求: {request}
上下文信息: {context}
请提供：
1. 主要任务分解
2. 每个步骤的预期输出
3. 所需工具类型
4. 优先级评估
响应格式为JSON，包含tasks数组和metadata对象。
"""
response = self.llm_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
return json.loads(response.choices[0].message.content)2. 执行阶段（Execute）
执行阶段负责按照规划序列逐步调用工具和子Agent，并维护执行状态：
class TaskExecutor:
def __init__(self, planner, tools, memory_manager):
self.planner = planner
self.tools = tools
self.memory = memory_manager
self.execution_state = {}
def execute(self, plan):
results = []
for task in plan['tasks']:
# 检查前置条件
if not self._check_prerequisites(task, results):
# 触发重新规划
plan = self.planner.replan(task, results)
continue
# 执行任务
result = self._execute_single_task(task)
# 更新记忆
self.memory.update(task, result)
# 存储结果
results.append({
'task': task,
'result': result,
'status': 'completed'
})
# 更新执行状态
self._update_execution_state(task, result)
return results
def _execute_single_task(self, task):
tool_name = task['required_tool']
tool_params = task['parameters']
# 动态工具选择
selected_tool = self._select_tool(tool_name, tool_params)
try:
result = selected_tool.execute(tool_params)
return self._normalize_result(result, task)
except Exception as e:
# 错误处理和回退策略
return self._handle_execution_error(task, e)3. 评估阶段（Evaluate）
评估阶段承担质量控制和自我纠错的重要职责：
class TaskEvaluator:
def __init__(self, llm_client, quality_metrics):
self.llm_client = llm_client
self.metrics = quality_metrics
def evaluate(self, plan, execution_results, original_request):
evaluations = []
for result in execution_results:
# 多维度评估
quality_score = self._assess_quality(result)
consistency_check = self._check_consistency(result, execution_results)
completeness_check = self._verify_completeness(result, original_request)
evaluation = {
'task_id': result['task']['id'],
'quality_score': quality_score,
'consistency_passed': consistency_check,
'completeness_passed': completeness_check,
'issues': self._identify_issues(result),
'needs_correction': quality_score < 0.8 or not consistency_check
}
evaluations.append(evaluation)
# 整体评估
overall_quality = self._calculate_overall_quality(evaluations)
if overall_quality < 0.7:
# 触发重新规划
return {
'status': 'failed',
'evaluations': evaluations,
'replan_suggestions': self._generate_replan_suggestions(evaluations)
}
else:
return {
'status': 'passed',
'evaluations': evaluations,
'quality_score': overall_quality
}