简化的助手Agent，使用函数式工具
| 22 | |
| 23 | |
| 24 | class MyAgent(BaseAgent): |
| 25 | """简化的助手Agent，使用函数式工具""" |
| 26 | |
| 27 | def __init__(self, model: BaseModel, config: Dict[str, Any]): |
| 28 | super().__init__(model, config) |
| 29 | |
| 30 | # 获取工具注册表 |
| 31 | self.tool_registry = get_registry() |
| 32 | |
| 33 | logger.info(f"MyAgent initialized with {len(self.tool_registry)} tools: {self.tool_registry.get_all_names()}") |
| 34 | |
| 35 | async def _execute_tool(self, function_name: str, function_args: Dict[str, Any]) -> Any: |
| 36 | """ |
| 37 | 执行工具函数 |
| 38 | |
| 39 | Args: |
| 40 | function_name: 工具名称 |
| 41 | function_args: 工具参数 |
| 42 | |
| 43 | Returns: |
| 44 | 工具执行结果 |
| 45 | """ |
| 46 | logger.info(f"🔧 Executing: {function_name}") |
| 47 | logger.info(f"📝 Args: {json.dumps(function_args, ensure_ascii=False)}") |
| 48 | |
| 49 | # 通过注册表执行工具 |
| 50 | result = await self.tool_registry.execute(function_name, **function_args) |
| 51 | |
| 52 | logger.info(f"✅ Result: {result}") |
| 53 | return result |
| 54 | |
| 55 | def get_registered_tools(self) -> list: |
| 56 | """获取所有工具的OpenAI格式定义""" |
| 57 | return self.tool_registry.get_all_definitions() |
| 58 | |
| 59 | async def execute(self, context: Dict[str, Any], params: Dict[str, Any]) -> Dict[str, Any]: |
| 60 | """ |
| 61 | 执行Agent任务 |
| 62 | |
| 63 | Args: |
| 64 | context: {"query": "用户问题"} |
| 65 | params: {"temperature": 0.7} |
| 66 | """ |
| 67 | query = context.get("query", "") |
| 68 | |
| 69 | if not query: |
| 70 | return {"status": "error", "error": "No query provided"} |
| 71 | |
| 72 | logger.info(f"📥 Query: {query}") |
| 73 | |
| 74 | try: |
| 75 | # 获取所有工具定义 |
| 76 | tools = self.get_registered_tools() |
| 77 | |
| 78 | # 调用模型（假设base_agent中有这个方法） |
| 79 | response = await self._call_model_with_tools( |
| 80 | prompt=query, |
| 81 | tools=tools, |