19 实践课-工具链注册分类与权限控制设计

工具链注册分类与权限控制设计

关联：索引

术语小抄（初学者版）

作业：布置（见文末）

1）工具链四类工具（功能视角）

2）三类调用场景（场景视角）

3）RBAC 角色（权限视角）

4）拒绝策略（安全优先）

AI 工具使用：权限方案设计 / 权限校验代码生成 / 优先级策略说明（学生可直接复制）

使用方法：把你的“工具清单、场景定义、角色定义、风险约束”粘贴到 {你的内容}。要求 AI 输出“方案 + 可复制代码 + 自测用例 + 校验点”。你必须做人工审计与回归测试，不可直接照搬上线。

模板目录：

模板 1：生成工具链注册分类清单

你是工业智能体工具链架构师。请根据我提供的工具列表，生成一个“工具链注册分类清单”，要求：
1）分类：common/data_query/device_control/knowledge_base；
2）每个工具给出元信息：tool_name、category、scenes、required_roles、reliability(0~1)、risk_level(low/medium/high)、description；
3）指出哪些工具必须加“二次确认”或“安全门禁”（例如设备控制）；
4）输出为 JSON 数组，便于我直接保存为 tools_manifest.json。

我的工具列表与约束：{你的内容}

模板 2：生成分拣系统 RBAC 权限分配方案（多场景）

你是工业产线权限控制工程师。请为“分拣系统智能体工具链”设计 RBAC 权限方案，要求：
1）角色：admin/operator/viewer（可扩展，但要说明原因）；
2）场景：sorting_ops/maintenance/audit；
3）输出：每个角色在每个场景下允许调用的工具列表与限制条件（例如只允许查询最近 24h、只允许 speed<=0.6）；
4）给出 5 条“危险操作”示例，并说明如何通过权限与参数校验拦截；
5）最后给一份可以落地的 policy.json（结构清晰，可读可维护）。

我的工具清单与风险约束：{你的内容}

模板 3：生成权限校验核心代码（含审计与自测）

你是 Python 工程师。请根据我给的 policy.json 与工具清单，生成权限校验代码，要求：
1）函数 authorize(user_ctx, scene, tool_name, action, payload) -> (ok, error_code, message)；
2）拒绝时返回结构化错误（含 trace_id），允许时返回 ok=true；
3）记录审计日志（JSONL，每行一条，字段含 ts_ms/user_id/role/scene/tool/action/ok/trace_id/reason）；
4）给出至少 8 条自测用例（不同角色/场景/工具/参数组合），并说明每条的预期结果；
5）只用 Python 标准库，不引入第三方依赖。

我的 policy.json 与工具清单：{你的内容}

模板 4：调用优先级排序实现（按场景 + 可靠性）

请你讲解并给出一个可运行示例：当一个任务在某个场景下有多个候选工具时，如何按“场景优先级 + 工具可靠性 + 风险等级”排序并选择。
要求：
1）给出数据结构（tools_manifest 里要有哪些字段）；
2）给出排序规则（用自然语言 + 伪代码 + Python 可运行代码）；
3）用 2 个场景举例（sorting_ops 与 audit），展示排序结果为什么合理；
4）给出 3 条常见坑（例如只看可靠性导致高风险工具被优先）。
  1. 当工具从 3 个增长到 30 个，为什么“把 tools=[...] 直接塞给 Agent”会变成灾难？
  2. 如果一个操作员误调用了“设备急停解除工具”，后果是什么？你如何证明“权限校验起作用了”？
工具实现（Python Tool 函数/StructuredTool）
  └─ 工具注册表（registry：tool + meta）
        ├─ 分类视图（按 category）
        ├─ 场景视图（按 scene）
        ├─ 权限过滤（RBAC：role + scene + tool）
        └─ 优先级排序（scene priority + reliability）
              ↓
        给 Agent 的工具集（tools_for_this_user_in_this_scene）
              ↓
        审计记录（调用通过/拒绝/失败都记录 trace_id）

关键点解释与自检要点：

目标：用一份清单把“工具是谁、属于哪类、可用于哪些场景、谁能用、可靠不可靠”说清楚。

from __future__ import annotations

import json
import time
import uuid
from dataclasses import dataclass
from typing import Any, Callable, Dict, Iterable, List, Literal, Optional, Tuple

Role = Literal["admin", "operator", "viewer"]
Scene = Literal["sorting_ops", "maintenance", "audit"]
Category = Literal["common", "data_query", "device_control", "knowledge_base"]
RiskLevel = Literal["low", "medium", "high"]

@dataclass(frozen=True)
class UserContext:
    user_id: str
    role: Role

@dataclass(frozen=True)
class ToolMeta:
    tool_name: str
    category: Category
    scenes: List[Scene]
    required_roles: List[Role]
    reliability: float
    risk_level: RiskLevel
    description: str
    scene_priority: Dict[Scene, int]

ToolInvoke = Callable[ [Dict[str, Any]], str ]

class ToolRegistry:
    def __init__(self) -> None:
        self._tools: Dict[str, ToolInvoke] = {}
        self._meta: Dict[str, ToolMeta] = {}

    def register(self, tool: ToolInvoke, meta: ToolMeta) -> None:
        name = meta.tool_name.strip()
        if not name:
            raise ValueError("tool_name is empty")
        if name in self._tools:
            raise ValueError(f"tool already registered: {name}")
        if not (0.0 <= meta.reliability <= 1.0):
            raise ValueError("reliability must be in [0, 1]")
        self._tools[name] = tool
        self._meta[name] = meta

    def list_meta(self) -> List[ToolMeta]:
        return list(self._meta.values())

    def list_by_category(self, category: Category) -> List[ToolMeta]:
        return [m for m in self._meta.values() if m.category == category]

    def list_by_scene(self, scene: Scene) -> List[ToolMeta]:
        return [m for m in self._meta.values() if scene in m.scenes]

    def get(self, tool_name: str) -> Tuple[ToolInvoke, ToolMeta]:
        return self._tools[tool_name], self._meta[tool_name]

def _now_ms() -> int:
    return int(time.time() * 1000)

def _json_ok(data: Dict[str, Any], trace_id: str) -> str:
    return json.dumps({"ok": True, "data": data, "trace_id": trace_id}, ensure_ascii=False)

def _json_err(code: str, message: str, trace_id: str) -> str:
    return json.dumps({"ok": False, "error": {"code": code, "message": message}, "trace_id": trace_id}, ensure_ascii=False)

def make_demo_tools() -> Dict[str, ToolInvoke]:
    def common_ping(payload: Dict[str, Any]) -> str:
        trace_id = uuid.uuid4().hex[:8]
        return _json_ok({"pong": True, "ts_ms": _now_ms(), "payload_preview": str(payload)[:120]}, trace_id)

    def data_query_batch_stats(payload: Dict[str, Any]) -> str:
        trace_id = uuid.uuid4().hex[:8]
        batch_no = str(payload.get("batch_no", "")).strip()
        if not batch_no:
            return _json_err("INPUT_INVALID", "batch_no is required", trace_id)
        return _json_ok({"batch_no": batch_no, "total": 1200, "grade_a": 860}, trace_id)

    def device_control_arm_home(payload: Dict[str, Any]) -> str:
        trace_id = uuid.uuid4().hex[:8]
        device_id = str(payload.get("device_id", "")).strip() or "arm_01"
        return _json_ok({"cmd": "arm.home", "device_id": device_id, "accepted": True}, trace_id)

    def kb_qa(payload: Dict[str, Any]) -> str:
        trace_id = uuid.uuid4().hex[:8]
        query = str(payload.get("query", "")).strip()
        if not query:
            return _json_err("INPUT_EMPTY", "query is empty", trace_id)
        return _json_ok({"answer": f"示例回答：{query}", "citations": ["doc:sorting_spec#safety"]}, trace_id)

    return {
        "common_ping": common_ping,
        "data_query_batch_stats": data_query_batch_stats,
        "device_control_arm_home": device_control_arm_home,
        "kb_qa": kb_qa,
    }

def build_demo_registry() -> ToolRegistry:
    tools = make_demo_tools()
    r = ToolRegistry()

    r.register(
        tools["common_ping"],
        ToolMeta(
            tool_name="common_ping",
            category="common",
            scenes=["sorting_ops", "maintenance", "audit"],
            required_roles=["admin", "operator", "viewer"],
            reliability=0.99,
            risk_level="low",
            description="连通性探针，用于验证工具链与输出口径是否正常。",
            scene_priority={"sorting_ops": 50, "maintenance": 50, "audit": 50},
        ),
    )
    r.register(
        tools["data_query_batch_stats"],
        ToolMeta(
            tool_name="data_query_batch_stats",
            category="data_query",
            scenes=["sorting_ops", "maintenance", "audit"],
            required_roles=["admin", "operator", "viewer"],
            reliability=0.9,
            risk_level="medium",
            description="查询批次统计（示例）。真实实现中应使用只读账号与参数化查询。",
            scene_priority={"sorting_ops": 70, "maintenance": 60, "audit": 90},
        ),
    )
    r.register(
        tools["device_control_arm_home"],
        ToolMeta(
            tool_name="device_control_arm_home",
            category="device_control",
            scenes=["sorting_ops", "maintenance"],
            required_roles=["admin", "operator"],
            reliability=0.8,
            risk_level="high",
            description="机械臂回零（示例）。高风险控制动作，必须权限校验与审计。",
            scene_priority={"sorting_ops": 80, "maintenance": 95, "audit": 0},
        ),
    )
    r.register(
        tools["kb_qa"],
        ToolMeta(
            tool_name="kb_qa",
            category="knowledge_base",
            scenes=["sorting_ops", "maintenance", "audit"],
            required_roles=["admin", "operator", "viewer"],
            reliability=0.85,
            risk_level="low",
            description="知识库问答（示例）。真实实现应返回命中证据与来源字段。",
            scene_priority={"sorting_ops": 60, "maintenance": 70, "audit": 80},
        ),
    )
    return r

def main() -> None:
    r = build_demo_registry()
    print("all tools:", [m.tool_name for m in r.list_meta()])
    print("device_control:", [m.tool_name for m in r.list_by_category("device_control")])
    print("audit scene:", [m.tool_name for m in r.list_by_scene("audit")])

if __name__ == "__main__":
    main()

解释与自检要点：

运行方式（Windows/通用）：

python tool_registry_demo.py

关键点解释：

  1. 最小权限：默认不允许，按需授权；尤其是 device_control 工具。
  2. 角色清晰：角色数量要少且可解释，避免“角色爆炸”导致无法维护。
  3. 场景分层：同一角色在不同场景权限不同（生产操作 vs 审计）。
  4. 双重门禁：高风险工具除了角色权限，还要做参数级校验（例如速度上限、危险动作白名单）。
  5. 证据链：放行与拒绝都记录审计日志；每次调用都有 trace_id。

目标：把“是否允许调用”变成一段可运行、可测试的代码。

import json
import time
import uuid
from typing import Any, Dict, List, Tuple

def now_ms() -> int:
    return int(time.time() * 1000)

def new_trace_id() -> str:
    return uuid.uuid4().hex[:8]

def audit_line(
    *,
    user: UserContext,
    scene: Scene,
    tool_name: str,
    action: str,
    ok: bool,
    trace_id: str,
    reason: str,
) -> str:
    rec = {
        "ts_ms": now_ms(),
        "user_id": user.user_id,
        "role": user.role,
        "scene": scene,
        "tool": tool_name,
        "action": action,
        "ok": ok,
        "trace_id": trace_id,
        "reason": reason,
    }
    return json.dumps(rec, ensure_ascii=False)

def authorize(
    *,
    user: UserContext,
    scene: Scene,
    tool_required_roles: List[Role],
    tool_risk_level: RiskLevel,
    action: str,
    payload: Dict[str, Any],
) -> Tuple[bool, str]:
    if user.role not in tool_required_roles:
        return False, "ROLE_DENY"

    if tool_risk_level == "high":
        if scene == "audit":
            return False, "SCENE_DENY"
        if user.role != "admin":
            need_confirm = bool(payload.get("confirm", False))
            if not need_confirm:
                return False, "CONFIRM_REQUIRED"

    return True, "ALLOW"

解释与自检要点：

目标：给 Agent/业务层的不是“全量工具”，而是“当前用户在当前场景可用且排序后”的工具集。

from typing import List, Tuple

def select_tools_for_user_and_scene(
    *,
    registry: ToolRegistry,
    user: UserContext,
    scene: Scene,
) -> List[ToolMeta]:
    candidates = registry.list_by_scene(scene)

    allowed: List[ToolMeta] = []
    for m in candidates:
        ok, _reason = authorize(
            user=user,
            scene=scene,
            tool_required_roles=m.required_roles,
            tool_risk_level=m.risk_level,
            action="invoke",
            payload={},
        )
        if ok:
            allowed.append(m)

    def score(meta: ToolMeta) -> Tuple[int, float, int]:
        scene_p = int(meta.scene_priority.get(scene, 0))
        reliability = float(meta.reliability)
        risk_penalty = 0 if meta.risk_level == "low" else 1 if meta.risk_level == "medium" else 2
        return (scene_p, reliability, -risk_penalty)

    return sorted(allowed, key=score, reverse=True)

解释与自检要点：

把下面测试代码追加到 main() 里，验证“不同角色/场景”下的工具集合差异与排序效果。

def demo_permissions_and_priority() -> None:
    registry = build_demo_registry()
    users = [
        UserContext(user_id="u_admin", role="admin"),
        UserContext(user_id="u_op", role="operator"),
        UserContext(user_id="u_view", role="viewer"),
    ]
    scenes: List[Scene] = ["sorting_ops", "maintenance", "audit"]

    for s in scenes:
        print("scene:", s)
        for u in users:
            metas = select_tools_for_user_and_scene(registry=registry, user=u, scene=s)
            print(" ", u.role, [m.tool_name for m in metas])

def main() -> None:
    demo_permissions_and_priority()

解释与自检要点：

运行方式：

python tool_registry_demo.py

项目工坊主题：工具链注册分类与整体架构梳理 + RBAC 权限控制设计与校验实现 + 调用优先级排序与验证。

步骤（教师演示 + 学生分组）：

  1. 演示：把全量工具塞给智能体时的风险（越权尝试、误选高风险工具、难审计）。
  2. 演示：统一 manifest + registry 后，如何按场景筛选与按分类展示工具链。
  3. 演示：RBAC 方案设计，把“谁能用什么”写成 policy，并实现校验与审计记录。
  4. 演示：调用优先级排序对结果的影响，并用测试日志证明排序生效。
  5. 学生分组：用你们组已有工具替换演示工具，完成注册、分类、权限、排序与测试证据链。

九、课程思政融入点（工业设备操作安全与工程伦理）

作业：布置

1）提交工具链注册分类清单及架构梳理图，附分类说明。
2）提交权限控制方案文档（含角色定义、权限分配）及权限校验代码。
3）提交不同角色调用工具的测试日志（含权限校验通过 / 拒绝截图）。