Authorization权限判断在获得Tool工具后执行工具前进行权限判断即在 for item in tool_calls:中。for item in tool_calls: print(\n Tool Call ) print(工具:, item.name) print(参数:, item.arguments) arguments json.loads(item.arguments) # 从 Registry 获取 tool tool_registry.get(item.name) if tool is None: raise ValueError( f不存在的工具: {item.name} ) # 执行 # 执行前先检查权限 if tool.requires_confirmation: print(\n⚠️ 检测到需要授权的敏感操作) print(工具:, tool.name) print(参数:, arguments) confirm input(是否允许执行(y/n)).strip().lower() if confirm ! y: result 操作被拒绝未获得用户授权 print(❌ 用户拒绝执行) else: print(✅ 用户已确认执行) try: result tool.execute_with_retry(arguments,max_retries2) print(Tool Result:, result) except Exception as e: result f工具执行失败{type(e).__name__}: {e} print(Tool Error:, result) else: try: result tool.execute_with_retry(arguments,max_retries2) print(Tool Result:, result) except Exception as e: result f工具执行失败{type(e).__name__}: {e} print(Tool Error:, result) print(Tool Result:, result) # # 把 Tool Result 放回上下文 # input_items.append( { type: function_call_output, call_id: item.call_id, output: str(result) } ) else: print(❌ Agent 达到最大执行轮数停止运行。)定义工具的时候加入字段requires_confirmation表示是否需要进行权限判断。calculator_tool Tool( namecalculator, description计算两个数字的乘积, args_modelCalculatorArgs, functioncalculator, requires_confirmationFalse #权限 ) secret_tool Tool( nameget_secret, description获取内部系统中的敏感信息, args_modelSecretArgs, functionget_secret, requires_confirmationTrue )异常捕获在 Agent Loop 中找到真正执行工具的位置(就是tool.execute)try: result tool.execute(arguments) print(Tool Result:, result) except Exception as e: result f工具执行失败{type(e).__name__}: {e} print(Tool Error:, result)增加工具重试机制在你的.py中找到Tool类在execute()方法下面增加一个方法def execute_with_retry(self, arguments, max_retries2): for attempt in range(max_retries 1): try: return self.execute(arguments) except Exception as e: print( f[Tool Error] 第 {attempt 1} 次执行失败 f{type(e).__name__}: {e} ) if attempt max_retries: raise注意这个方法需要定义在class Tool:内部与execute()保持相同的缩进层级。这里的max_retries2表示第 1 次正常执行。第 2 次第一次重试。第 3 次第二次重试。仍然失败抛出异常由 Agent 外层捕获。接下来在 Agent Loop 中把原来的result tool.execute(arguments)替换成result tool.execute_with_retry( arguments, max_retries2 )保留外层的try/except这样最终失败时仍然可以将错误回传给 LLM。最终代码import json from openai import OpenAI from pydantic import BaseModel client OpenAI() # # 1. 参数模型 # class CalculatorArgs(BaseModel): a: float b: float class SecretArgs(BaseModel): reason: str # # 2. Python 工具函数 # def calculator(a, b): print(f[Python] calculator({a}, {b})) if b 0: raise ValueError(第二个参数不能为 0) return a * b def get_secret(reason): print(f[Python] get_secret(reason{reason})) # 模拟一个敏感资源 return 这是模拟的内部信息SECRET-123456 # # 3. Tool 类 # class Tool: def __init__(self, name, description, args_model, function,requires_confirmationFalse): self.name name self.description description self.args_model args_model self.function function self.requires_confirmation requires_confirmation def schema(self): return { type: function, name: self.name, description: self.description, parameters: self.args_model.model_json_schema() } def execute(self, arguments): args self.args_model(**arguments) return self.function(**args.model_dump()) def execute_with_retry(self, arguments, max_retries2): for attempt in range(max_retries 1): try: return self.execute(arguments) except Exception as e: print( f[Tool Error] 第 {attempt 1} 次执行失败 f{type(e).__name__}: {e} ) if attempt max_retries: raise # # 4. 创建 Tool # calculator_tool Tool( namecalculator, description计算两个数字的乘积, args_modelCalculatorArgs, functioncalculator, requires_confirmationFalse #权限 ) secret_tool Tool( nameget_secret, description获取内部系统中的敏感信息, args_modelSecretArgs, functionget_secret, requires_confirmationTrue ) # # 5. Tool Registry # tool_registry { calculator_tool.name: calculator_tool, secret_tool.name: secret_tool } # # 6. Tool Schema # tools [ tool.schema() for tool in tool_registry.values() ] # # 7. 模拟恶意用户 # user_input 计算 123 * 0。 print( User ) print(user_input) # # 8. Agent Loop # input_items [ { role: user, content: user_input } ] MAX_STEPS 5 for step in range(MAX_STEPS): print(\n Calling LLM ) response client.responses.create( modelgpt-6.1-sol, inputinput_items, toolstools, tool_choicerequired ) print(LLM 返回) # # 把 LLM 的输出加入上下文 # input_items.extend(response.output) # # 找出 Tool Call # tool_calls [ item for item in response.output if item.type function_call ] # # 没有 Tool Call → Agent 完成 # if not tool_calls: print(\n Final Answer ) print(response.output_text) break # # 执行所有 Tool # for item in tool_calls: print(\n Tool Call ) print(工具:, item.name) print(参数:, item.arguments) arguments json.loads(item.arguments) # 从 Registry 获取 tool tool_registry.get(item.name) if tool is None: raise ValueError( f不存在的工具: {item.name} ) # 执行 # 执行前先检查权限 if tool.requires_confirmation: print(\n⚠️ 检测到需要授权的敏感操作) print(工具:, tool.name) print(参数:, arguments) confirm input(是否允许执行(y/n)).strip().lower() if confirm ! y: result 操作被拒绝未获得用户授权 print(❌ 用户拒绝执行) else: print(✅ 用户已确认执行) try: result tool.execute_with_retry(arguments,max_retries2) print(Tool Result:, result) except Exception as e: result f工具执行失败{type(e).__name__}: {e} print(Tool Error:, result) else: try: result tool.execute_with_retry(arguments,max_retries2) print(Tool Result:, result) except Exception as e: result f工具执行失败{type(e).__name__}: {e} print(Tool Error:, result) print(Tool Result:, result) # # 把 Tool Result 放回上下文 # input_items.append( { type: function_call_output, call_id: item.call_id, output: str(result) } ) else: print(❌ Agent 达到最大执行轮数停止运行。)总结此时的Agent 架构已经开始成型┌─────────────────────────────┐│ Agent ││ ││ LLM ←──────────────┐ ││ ↓ │ ││ Tool Selection │ ││ ↓ │ ││ Authorization │ ││ ↓ │ ││ Pydantic Validation │ ││ ↓ │ ││ Tool Execution │ ││ ↓ │ ││ Error / Result ──────┘ ││ ││ MAX_STEPS │└─────────────────────────────┘Prompt 层LLM 应该怎么做Tool Calling 层LLM 想调用什么Pydantic 层参数是否合法Authorization 层有没有权限Error Handling 层工具失败怎么办Agent Loop 层失败之后能不能继续