Tool Harness：模型只能提议动作，执行权必须被隔离

[ 用户输入 ] -> [ 大模型 ] -> (输出 JSON) -> [ 应用自动解析并无脑调用目标 API ] -> (返回结果给模型)

import json
from typing import Dict, Any, Callable, Tuple
from pydantic import BaseModel, Field

class ToolResponse(BaseModel):
    success: bool
    data: Any
    error: str = ""

class BaseTool:
    name: str
    description: str
    args_schema: type[BaseModel]
    risk_level: str  # "Tier-1", "Tier-2", "Tier-3", "Tier-4"

    def __init__(self, execution_func: Callable[..., Any]):
        self.execution_func = execution_func

    @property
    def schema(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self.args_schema.model_json_schema()
        }

    def execute(self, **kwargs) -> ToolResponse:
        try:
            # 进行输入值强类型校验
            validated_args = self.args_schema(**kwargs)
            result = self.execution_func(**validated_args.model_dump())
            return ToolResponse(success=True, data=result)
        except Exception as e:
            return ToolResponse(success=False, data=None, error=str(e))

class TransferFundsSchema(BaseModel):
    recipient_account: str = Field(..., description="收款人账号，必须为 10 位纯数字")
    amount: float = Field(..., description="转账金额，单位：元")
    currency: str = Field("CNY", description="币种，默认 CNY")

def real_transfer_execution(recipient_account: str, amount: float, currency: str) -> str:
    # 这里是真实的转账 API 交互代码
    return f"Successfully transferred {amount} {currency} to {recipient_account}"

# 注册工具，标注为 Tier-4 高危动作
transfer_tool = BaseTool(execution_func=real_transfer_execution)
transfer_tool.name = "transfer_funds"
transfer_tool.description = "向指定的收款人账户转账。此操作不可逆。"
transfer_tool.args_schema = TransferFundsSchema
transfer_tool.risk_level = "Tier-4"

import hashlib

class IdempotencyManager:
    def __init__(self):
        self._seen_keys = set()

    def generate_key(self, session_id: str, tool_name: str, arguments: Dict[str, Any]) -> str:
        """
        将 Session ID、工具名和参数排序序列化后，生成唯一的 MD5 签名作为幂等键
        """
        serialized_args = json.dumps(arguments, sort_keys=True)
        raw_str = f"{session_id}:{tool_name}:{serialized_args}"
        return hashlib.md5(raw_str.encode("utf-8")).hexdigest()

    def is_duplicate(self, key: str) -> bool:
        return key in self._seen_keys

    def mark_executed(self, key: str):
        self._seen_keys.add(key)

class ToolHarnessEngine:
    def __init__(self):
        self.registry: Dict[str, BaseTool] = {}
        self.idempotency = IdempotencyManager()

    def register_tool(self, tool: BaseTool):
        self.registry[tool.name] = tool

    def handle_proposal(
        self, 
        session_id: str, 
        tool_name: str, 
        arguments: Dict[str, Any], 
        human_approved: bool = False
    ) -> Tuple[str, Any]:
        """
        处理大模型的工具调用提议。
        返回值状态码：
        - "SCHEM_ERROR": 参数结构错误
        - "DUPLICATE_BLOCKED": 幂等拦截
        - "PENDING_APPROVAL": 等待人工审批
        - "EXECUTED": 执行成功
        - "EXECUTION_FAILED": 执行期崩溃
        """
        # 1. 查找工具是否存在
        if tool_name not in self.registry:
            return "SCHEM_ERROR", f"Tool '{tool_name}' is not registered."

        tool = self.registry[tool_name]

        # 2. Schema 校验
        try:
            tool.args_schema(**arguments)
        except Exception as schema_err:
            # 拦截校验错误，格式化后作为 System 消息反哺给 LLM 重试
            return "SCHEM_ERROR", f"Invalid arguments format: {str(schema_err)}. Please correct your parameters."

        # 3. 幂等校验
        idem_key = self.idempotency.generate_key(session_id, tool_name, arguments)
        if self.idempotency.is_duplicate(idem_key):
            return "DUPLICATE_BLOCKED", {
                "msg": "This action was already executed previously. Blocked duplication to prevent double-spending.",
                "idempotency_key": idem_key
            }

        # 4. 风险控制与人工拦截
        if tool.risk_level == "Tier-4" and not human_approved:
            return "PENDING_APPROVAL", {
                "msg": f"Tool '{tool_name}' requires human confirmation due to High-Risk tier.",
                "idempotency_key": idem_key,
                "details": arguments
            }

        # 5. 执行真实物理动作
        response = tool.execute(**arguments)
        if response.success:
            self.idempotency.mark_executed(idem_key)
            return "EXECUTED", response.data
        else:
            return "EXECUTION_FAILED", response.error

User -> Tool Harness Engine
Status: SCHEM_ERROR
Payload: "Invalid arguments format: 1 validation error for TransferFundsSchema \n recipient_account \n String should have 10 characters..."

pip install pydantic

# test_tool_harness.py
from pydantic import BaseModel, Field
from typing import Dict, Any
import sys

# 1. 导入或贴入上文定义的 BaseTool, ToolHarnessEngine 等基础类
# (此处省略上文已给出的 BaseTool, IdempotencyManager, ToolHarnessEngine 的类声明)
# [请确保把前面的类代码复制到此处]

# 2. 定义高危的删除工具
class DeleteUserDataSchema(BaseModel):
    user_id: int = Field(..., description="要删除的用户的唯一数字 ID")
    confirm_permanent: bool = Field(..., description="是否永久删除，必须为 True 才会触发真实删除")

def real_delete_data(user_id: int, confirm_permanent: bool) -> str:
    return f"DATABASE ALTERED: User {user_id} data has been purged permanently."

delete_tool = BaseTool(execution_func=real_delete_data)
delete_tool.name = "delete_user_data"
delete_tool.description = "危险操作：从生产数据库永久擦除用户数据。"
delete_tool.args_schema = DeleteUserDataSchema
delete_tool.risk_level = "Tier-4"

# 3. 补全执行与审批循环
def run_harness_loop(engine: ToolHarnessEngine, session_id: str, tool_name: str, arguments: Dict[str, Any]):
    # 首次尝试调用
    status, result = engine.handle_proposal(session_id, tool_name, arguments)
    
    if status == "PENDING_APPROVAL":
        print(f"\n[🚨 WARNING] 检测到高危操作！")
        print(f"操作名称: {tool_name}")
        print(f"操作明细: {json.dumps(result['details'], indent=2)}")
        print(f"幂等键: {result['idempotency_key']}")
        
        # 【练习任务】：在此处模拟人工干预命令行提示
        # 提示用户输入 y/n
        user_input = input("是否批准该操作执行？(y/N): ").strip().lower()
        
        if user_input == 'y':
            # 如果人工同意，再次调用 handle_proposal 并在参数中传入 human_approved=True
            status, final_res = engine.handle_proposal(session_id, tool_name, arguments, human_approved=True)
            print(f"\n[✅ 执行结果] 状态: {status} -> 反馈给模型的数据: {final_res}")
        else:
            print(f"\n[❌ 拒绝执行] 该操作已被安全管理员驳回。")
    else:
        print(f"\n[INFO] 操作直接执行。状态: {status} -> 结果: {result}")

if __name__ == "__main__":
    # 初始化网关
    engine = ToolHarnessEngine()
    engine.register_tool(delete_tool)
    
    session_id = "user_session_999"
    
    # 测试用例 1: 错误的参数类型（user_id 应为 int 却传入了 string "abc"）
    print("--- 测试用例 1: 格式校验拦截 ---")
    run_harness_loop(engine, session_id, "delete_user_data", {"user_id": "abc", "confirm_permanent": True})
    
    # 测试用例 2: 正常高危调用（触发人工干预流程）
    print("\n--- 测试用例 2: 人工干预流程 ---")
    run_harness_loop(engine, session_id, "delete_user_data", {"user_id": 42, "confirm_permanent": True})