"""动态 Handoff 工具创建器 展示如何在运行时动态创建 Agent 切换工具，并将其注入到 LLM 的工具列表中 """ import json from typing import Dict, Any, List, Optional, Callable from pydantic import BaseModel, Field from app.core.logging_config import get_business_logger logger = get_business_logger() class DynamicHandoffToolCreator: """动态 Handoff 工具创建器 核心功能： 1. 根据可用 Agent 动态生成工具定义 2. 将工具转换为 LLM 可理解的 schema 3. 处理工具调用并执行 handoff """ def __init__(self, current_agent_id: str, available_agents: Dict[str, Any]): """初始化工具创建器 Args: current_agent_id: 当前 Agent ID available_agents: 可用的 Agent 字典 """ self.current_agent_id = current_agent_id self.available_agents = available_agents self.tools = [] self.tool_handlers = {} # 动态创建工具 self._create_handoff_tools() def _create_handoff_tools(self): """动态创建所有 handoff 工具""" for agent_id, agent_data in self.available_agents.items(): if agent_id == self.current_agent_id: continue # 不创建切换到自己的工具 # 创建工具 tool_def = self._create_single_tool(agent_id, agent_data) self.tools.append(tool_def) # 创建工具处理器 handler = self._create_tool_handler(agent_id, agent_data) self.tool_handlers[tool_def["function"]["name"]] = handler logger.info( f"为 Agent {self.current_agent_id} 创建了 {len(self.tools)} 个 handoff 工具" ) def _create_single_tool(self, target_agent_id: str, agent_data: Dict[str, Any]) -> Dict[str, Any]: """创建单个 handoff 工具定义 Args: target_agent_id: 目标 Agent ID agent_data: Agent 数据 Returns: 工具定义（OpenAI function calling 格式） """ agent_info = agent_data.get("info", {}) name = agent_info.get("name", "未命名") role = agent_info.get("role", "") capabilities = agent_info.get("capabilities", []) # 生成工具名称（符合函数命名规范） tool_name = f"transfer_to_{self._sanitize_name(target_agent_id)}" # 生成描述 description = f"切换到 {name}" if role: description += f"（{role}）" if capabilities: cap_str = "、".join(capabilities[:3]) description += f"。擅长: {cap_str}" description += "。当用户的问题更适合该 Agent 处理时调用此工具。" # 构建工具定义（OpenAI function calling 格式） tool_def = { "type": "function", "function": { "name": tool_name, "description": description, "parameters": { "type": "object", "properties": { "reason": { "type": "string", "description": "为什么要切换到该 Agent？请简要说明原因。" }, "context_summary": { "type": "string", "description": "需要传递给目标 Agent 的上下文摘要（可选）。例如：之前的计算结果、用户的具体需求等。" } }, "required": ["reason"] } } } return tool_def def _create_tool_handler( self, target_agent_id: str, agent_data: Dict[str, Any] ) -> Callable: """创建工具处理器函数 Args: target_agent_id: 目标 Agent ID agent_data: Agent 数据 Returns: 工具处理器函数 """ def handler(reason: str, context_summary: Optional[str] = None) -> Dict[str, Any]: """处理 handoff 工具调用 Args: reason: 切换原因 context_summary: 上下文摘要 Returns: Handoff 请求 """ agent_info = agent_data.get("info", {}) logger.info( f"Handoff 工具被调用: {self.current_agent_id} → {target_agent_id}", extra={ "reason": reason, "has_context": bool(context_summary) } ) return { "type": "handoff", "target_agent_id": target_agent_id, "target_agent_name": agent_info.get("name", ""), "reason": reason, "context_summary": context_summary, "from_agent_id": self.current_agent_id } return handler def _sanitize_name(self, name: str) -> str: """清理名称，使其符合函数命名规范 Args: name: 原始名称 Returns: 清理后的名称 """ # 替换特殊字符为下划线 sanitized = name.replace("-", "_").replace(" ", "_") # 移除其他非法字符 sanitized = "".join(c for c in sanitized if c.isalnum() or c == "_") return sanitized.lower() def get_tools_for_llm(self) -> List[Dict[str, Any]]: """获取用于 LLM 的工具列表 Returns: 工具定义列表（OpenAI function calling 格式） """ return self.tools def handle_tool_call(self, tool_name: str, arguments: Dict[str, Any]) -> Optional[Dict[str, Any]]: """处理 LLM 的工具调用 Args: tool_name: 工具名称 arguments: 工具参数 Returns: Handoff 请求或 None """ handler = self.tool_handlers.get(tool_name) if not handler: logger.warning(f"未找到工具处理器: {tool_name}") return None try: return handler(**arguments) except Exception as e: logger.error(f"工具调用失败: {tool_name}, 错误: {str(e)}") return None def get_tool_names(self) -> List[str]: """获取所有工具名称 Returns: 工具名称列表 """ return [tool["function"]["name"] for tool in self.tools] # ==================== 使用示例 ==================== def example_usage(): """展示如何使用动态工具创建器""" # 1. 准备可用的 Agent 信息 available_agents = { "math-agent-uuid": { "info": { "name": "数学助手", "role": "数学专家", "capabilities": ["数学计算", "方程求解", "几何问题"] } }, "creative-agent-uuid": { "info": { "name": "创意助手", "role": "创意专家", "capabilities": ["写作", "诗歌", "故事创作"] } }, "code-agent-uuid": { "info": { "name": "代码助手", "role": "编程专家", "capabilities": ["代码编写", "调试", "代码审查"] } } } # 2. 为当前 Agent 创建工具 current_agent_id = "general-agent-uuid" tool_creator = DynamicHandoffToolCreator(current_agent_id, available_agents) # 3. 获取工具定义（用于 LLM） tools = tool_creator.get_tools_for_llm() print("=" * 60) print("动态创建的 Handoff 工具:") print("=" * 60) for tool in tools: print(f"\n工具名称: {tool['function']['name']}") print(f"描述: {tool['function']['description']}") print(f"参数: {json.dumps(tool['function']['parameters'], indent=2, ensure_ascii=False)}") # 4. 模拟 LLM 调用工具 print("\n" + "=" * 60) print("模拟 LLM 工具调用:") print("=" * 60) # LLM 决定切换到数学 Agent tool_call = { "name": "transfer_to_math_agent_uuid", "arguments": { "reason": "用户问题涉及数学计算", "context_summary": "用户想解方程 x^2 + 5x + 6 = 0" } } print(f"\nLLM 调用: {tool_call['name']}") print(f"参数: {json.dumps(tool_call['arguments'], indent=2, ensure_ascii=False)}") # 5. 处理工具调用 handoff_request = tool_creator.handle_tool_call( tool_call["name"], tool_call["arguments"] ) if handoff_request: print(f"\n✓ Handoff 请求已创建:") print(f" 类型: {handoff_request['type']}") print(f" 从: {handoff_request['from_agent_id']}") print(f" 到: {handoff_request['target_agent_id']} ({handoff_request['target_agent_name']})") print(f" 原因: {handoff_request['reason']}") print(f" 上下文: {handoff_request['context_summary']}") # ==================== 与 LLM 集成示例 ==================== async def integrate_with_llm_example(): """展示如何将动态工具集成到 LLM 调用中""" from app.core.models import RedBearLLM from app.core.models.base import RedBearModelConfig # 1. 准备 Agent 信息 available_agents = { "math-agent": { "info": { "name": "数学助手", "role": "数学专家", "capabilities": ["计算", "方程"] } } } # 2. 创建工具 tool_creator = DynamicHandoffToolCreator("general-agent", available_agents) tools = tool_creator.get_tools_for_llm() # 3. 构建 LLM 配置 model_config = RedBearModelConfig( model_name="gpt-4", provider="openai", api_key="your-api-key", extra_params={ "temperature": 0.7, "tools": tools, # 注入动态创建的工具 "tool_choice": "auto" # 让 LLM 自动决定是否调用工具 } ) # 4. 创建 LLM 实例 llm = RedBearLLM(model_config) # 5. 构建消息（包含系统提示） system_prompt = """你是一个智能助手。当用户的问题超出你的能力范围时，你可以使用工具切换到专业的 Agent。 可用的切换工具： - transfer_to_math_agent: 当遇到数学问题时使用 请根据用户问题判断是否需要切换。""" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": "帮我计算 3*6+15 的结果"} ] # 6. 调用 LLM response = await llm.ainvoke(messages) # 7. 检查是否有工具调用 if hasattr(response, 'tool_calls') and response.tool_calls: for tool_call in response.tool_calls: print(f"\n✓ LLM 调用了工具: {tool_call.name}") print(f" 参数: {tool_call.arguments}") # 处理工具调用 handoff_request = tool_creator.handle_tool_call( tool_call.name, tool_call.arguments ) if handoff_request: print(f"\n✓ 执行 Handoff:") print(f" 切换到: {handoff_request['target_agent_name']}") print(f" 原因: {handoff_request['reason']}") # 这里可以执行实际的 Agent 切换逻辑 # await execute_handoff(handoff_request) else: print(f"\n✓ LLM 直接回复: {response.content}") # ==================== 完整的 Agent 执行流程 ==================== class AgentExecutorWithHandoffs: """支持 Handoffs 的 Agent 执行器""" def __init__(self, agent_id: str, agent_config: Any, available_agents: Dict[str, Any]): """初始化执行器 Args: agent_id: 当前 Agent ID agent_config: Agent 配置 available_agents: 可用的其他 Agent """ self.agent_id = agent_id self.agent_config = agent_config self.available_agents = available_agents # 创建工具创建器 self.tool_creator = DynamicHandoffToolCreator(agent_id, available_agents) async def execute( self, message: str, conversation_history: List[Dict[str, Any]] = None ) -> Dict[str, Any]: """执行 Agent（支持 handoff） Args: message: 用户消息 conversation_history: 会话历史 Returns: 执行结果，可能包含 handoff_request """ from app.core.models import RedBearLLM from app.core.models.base import RedBearModelConfig # 1. 获取动态工具 tools = self.tool_creator.get_tools_for_llm() # 2. 构建系统提示（包含工具说明） system_prompt = self._build_system_prompt_with_tools() # 3. 构建消息 messages = [{"role": "system", "content": system_prompt}] # 添加历史消息 if conversation_history: messages.extend(conversation_history) # 添加当前消息 messages.append({"role": "user", "content": message}) # 4. 配置 LLM（注入工具） model_config = RedBearModelConfig( model_name=self.agent_config.model_name, provider=self.agent_config.provider, api_key=self.agent_config.api_key, extra_params={ "temperature": 0.7, "tools": tools, # 动态工具 "tool_choice": "auto" } ) llm = RedBearLLM(model_config) # 5. 调用 LLM response = await llm.ainvoke(messages) # 6. 处理响应 result = { "message": response.content if hasattr(response, 'content') else str(response), "agent_id": self.agent_id } # 7. 检查工具调用 if hasattr(response, 'tool_calls') and response.tool_calls: for tool_call in response.tool_calls: # 检查是否是 handoff 工具 if tool_call.name in self.tool_creator.get_tool_names(): # 处理 handoff handoff_request = self.tool_creator.handle_tool_call( tool_call.name, tool_call.arguments ) if handoff_request: result["handoff_request"] = handoff_request result["is_final_answer"] = False break else: # 处理其他业务工具 pass return result def _build_system_prompt_with_tools(self) -> str: """构建包含工具说明的系统提示""" base_prompt = self.agent_config.system_prompt or "你是一个智能助手。" # 添加工具说明 tool_names = self.tool_creator.get_tool_names() if tool_names: tools_desc = "\n\n你可以使用以下工具切换到专业的 Agent：\n" for tool in self.tool_creator.get_tools_for_llm(): tools_desc += f"- {tool['function']['name']}: {tool['function']['description']}\n" tools_desc += "\n当用户的问题超出你的能力范围时，请使用相应的工具切换到专业 Agent。" return base_prompt + tools_desc return base_prompt if __name__ == "__main__": # 运行示例 print("\n🚀 动态 Handoff 工具创建示例\n") example_usage() print("\n\n" + "=" * 60) print("完成！查看上面的输出了解工具创建流程。") print("=" * 60)