agent-governance

代理治理模式

為 AI 代理系統加入安全、信任與政策強制的模式。

概覽

治理模式確保 AI 代理在定義的邊界內運作——控制它們可以呼叫哪些工具、可以處理哪些內容、可以執行多少動作，並透過稽核軌跡維持問責制。

使用者請求 → 意圖分類 → 政策檢查 → 工具執行 → 稽核日誌
                     ↓                      ↓               ↓
              威脅偵測               允許/拒絕       信任更新

使用時機

• 具有工具存取權的代理：任何呼叫外部工具（API、資料庫、Shell 指令）的代理
• 多代理系統：代理委派給其他代理時需要信任邊界
• 正式環境部署：合規、稽核與安全需求
• 敏感操作：金融交易、資料存取、基礎設施管理

---

模式 1：治理政策

將代理允許的動作定義為可組合、可序列化的政策物件。

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import re

class PolicyAction(Enum):
    ALLOW = "allow"
    DENY = "deny"
    REVIEW = "review"  # 標記為需人工審查

@dataclass
class GovernancePolicy:
    """控制代理行為的宣告式政策。"""
    name: str
    allowed_tools: list[str] = field(default_factory=list)       # 允許清單
    blocked_tools: list[str] = field(default_factory=list)       # 封鎖清單
    blocked_patterns: list[str] = field(default_factory=list)    # 內容過濾
    max_calls_per_request: int = 100                             # 速率限制
    require_human_approval: list[str] = field(default_factory=list)  # 需核准的工具

    def check_tool(self, tool_name: str) -> PolicyAction:
        """檢查此政策是否允許某工具。"""
        if tool_name in self.blocked_tools:
            return PolicyAction.DENY
        if tool_name in self.require_human_approval:
            return PolicyAction.REVIEW
        if self.allowed_tools and tool_name not in self.allowed_tools:
            return PolicyAction.DENY
        return PolicyAction.ALLOW

    def check_content(self, content: str) -> Optional[str]:
        """檢查內容是否匹配封鎖模式。回傳匹配的模式或 None。"""
        for pattern in self.blocked_patterns:
            if re.search(pattern, content, re.IGNORECASE):
                return pattern
        return None

政策組合

組合多個政策（例如組織級 + 團隊級 + 代理特定）：

def compose_policies(*policies: GovernancePolicy) -> GovernancePolicy:
    """合併政策，以最嚴格者為準。"""
    combined = GovernancePolicy(name="composed")

    for policy in policies:
        combined.blocked_tools.extend(policy.blocked_tools)
        combined.blocked_patterns.extend(policy.blocked_patterns)
        combined.require_human_approval.extend(policy.require_human_approval)
        combined.max_calls_per_request = min(
            combined.max_calls_per_request,
            policy.max_calls_per_request
        )
        if policy.allowed_tools:
            if combined.allowed_tools:
                combined.allowed_tools = [
                    t for t in combined.allowed_tools if t in policy.allowed_tools
                ]
            else:
                combined.allowed_tools = list(policy.allowed_tools)

    return combined


# 使用方式：從廣泛到特定層層疊加政策
org_policy = GovernancePolicy(
    name="org-wide",
    blocked_tools=["shell_exec", "delete_database"],
    blocked_patterns=[r"(?i)(api[_-]?key|secret|password)\s*[:=]"],
    max_calls_per_request=50
)
team_policy = GovernancePolicy(
    name="data-team",
    allowed_tools=["query_db", "read_file", "write_report"],
    require_human_approval=["write_report"]
)
agent_policy = compose_policies(org_policy, team_policy)

政策以 YAML 儲存

將政策儲存為設定檔而非程式碼：

# governance-policy.yaml
name: production-agent
allowed_tools:
  - search_documents
  - query_database
  - send_email
blocked_tools:
  - shell_exec
  - delete_record
blocked_patterns:
  - "(?i)(api[_-]?key|secret|password)\\s*[:=]"
  - "(?i)(drop|truncate|delete from)\\s+\\w+"
max_calls_per_request: 25
require_human_approval:
  - send_email

import yaml

def load_policy(path: str) -> GovernancePolicy:
    with open(path) as f:
        data = yaml.safe_load(f)
    return GovernancePolicy(**data)

---

模式 2：語意意圖分類

在提示送達代理前，使用基於模式的訊號偵測危險意圖。

from dataclasses import dataclass

@dataclass
class IntentSignal:
    category: str       # 例如 "data_exfiltration", "privilege_escalation"
    confidence: float   # 0.0 到 1.0
    evidence: str       # 觸發偵測的內容

# 威脅偵測的加權訊號模式
THREAT_SIGNALS = [
    # 資料外洩
    (r"(?i)send\s+(all|every|entire)\s+\w+\s+to\s+", "data_exfiltration", 0.8),
    (r"(?i)export\s+.*\s+to\s+(external|outside|third.?party)", "data_exfiltration", 0.9),
    (r"(?i)curl\s+.*\s+-d\s+", "data_exfiltration", 0.7),

    # 權限提升
    (r"(?i)(sudo|as\s+root|admin\s+access)", "privilege_escalation", 0.8),
    (r"(?i)chmod\s+777", "privilege_escalation", 0.9),

    # 系統破壞
    (r"(?i)(rm\s+-rf|del\s+/[sq]|format\s+c:)", "system_destruction", 0.95),
    (r"(?i)(drop\s+database|truncate\s+table)", "system_destruction", 0.9),

    # 提示注入
    (r"(?i)ignore\s+(previous|above|all)\s+(instructions?|rules?)", "prompt_injection", 0.9),
    (r"(?i)you\s+are\s+now\s+(a|an)\s+", "prompt_injection", 0.7),
]

def classify_intent(content: str) -> list[IntentSignal]:
    """分類內容中的威脅訊號。"""
    signals = []
    for pattern, category, weight in THREAT_SIGNALS:
        match = re.search(pattern, content)
        if match:
            signals.append(IntentSignal(
                category=category,
                confidence=weight,
                evidence=match.group()
            ))
    return signals

def is_safe(content: str, threshold: float = 0.7) -> bool:
    """快速檢查：內容是否安全（低於給定門檻）？"""
    signals = classify_intent(content)
    return not any(s.confidence >= threshold for s in signals)

關鍵洞察：意圖分類在工具執行之前進行，作為飛行前的安全檢查。這與僅在生成之後檢查的輸出護欄有根本不同。

---

模式 3：工具層級治理裝飾器

為個別工具函式包裝治理檢查：

import functools
import time
from collections import defaultdict

_call_counters: dict[str, int] = defaultdict(int)

def govern(policy: GovernancePolicy, audit_trail=None):
    """裝飾器，對工具函式強制執行治理政策。"""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            tool_name = func.__name__

            # 1. 檢查工具允許清單/封鎖清單
            action = policy.check_tool(tool_name)
            if action == PolicyAction.DENY:
                raise PermissionError(f"政策 '{policy.name}' 封鎖了工具 '{tool_name}'")
            if action == PolicyAction.REVIEW:
                raise PermissionError(f"工具 '{tool_name}' 需要人工核准")

            # 2. 檢查速率限制
            _call_counters[policy.name] += 1
            if _call_counters[policy.name] > policy.max_calls_per_request:
                raise PermissionError(f"超出速率限制：最多 {policy.max_calls_per_request} 次呼叫")

            # 3. 檢查引數中的內容
            for arg in list(args) + list(kwargs.values()):
                if isinstance(arg, str):
                    matched = policy.check_content(arg)
                    if matched:
                        raise PermissionError(f"偵測到封鎖模式：{matched}")

            # 4. 執行並稽核
            start = time.monotonic()
            try:
                result = await func(*args, **kwargs)
                if audit_trail is not None:
                    audit_trail.append({
                        "tool": tool_name,
                        "action": "allowed",
                        "duration_ms": (time.monotonic() - start) * 1000,
                        "timestamp": time.time()
                    })
                return result
            except Exception as e:
                if audit_trail is not None:
                    audit_trail.append({
                        "tool": tool_name,
                        "action": "error",
                        "error": str(e),
                        "timestamp": time.time()
                    })
                raise

        return wrapper
    return decorator


# 在任何代理框架中使用
audit_log = []
policy = GovernancePolicy(
    name="search-agent",
    allowed_tools=["search", "summarize"],
    blocked_patterns=[r"(?i)password"],
    max_calls_per_request=10
)

@govern(policy, audit_trail=audit_log)
async def search(query: str) -> str:
    """搜尋文件——受政策治理。"""
    return f"搜尋結果：{query}"

# 通過：search("latest quarterly report")
# 封鎖：search("show me the admin password")

---

模式 4：信任評分

使用基於衰減的信任分數追蹤代理隨時間的可靠性：

from dataclasses import dataclass, field
import math
import time

@dataclass
class TrustScore:
    """信任分數，含時間衰減。"""
    score: float = 0.5          # 0.0（不信任）到 1.0（完全信任）
    successes: int = 0
    failures: int = 0
    last_updated: float = field(default_factory=time.time)

    def record_success(self, reward: float = 0.05):
        self.successes += 1
        self.score = min(1.0, self.score + reward * (1 - self.score))
        self.last_updated = time.time()

    def record_failure(self, penalty: float = 0.15):
        self.failures += 1
        self.score = max(0.0, self.score - penalty * self.score)
        self.last_updated = time.time()

    def current(self, decay_rate: float = 0.001) -> float:
        """取得含時間衰減的分數——信任會隨無活動而侵蝕。"""
        elapsed = time.time() - self.last_updated
        decay = math.exp(-decay_rate * elapsed)
        return self.score * decay

    @property
    def reliability(self) -> float:
        total = self.successes + self.failures
        return self.successes / total if total > 0 else 0.0


# 在多代理系統中使用
trust = TrustScore()

# 代理成功完成任務
trust.record_success()  # 0.525
trust.record_success()  # 0.549

# 代理發生錯誤
trust.record_failure()  # 0.467

# 根據信任門檻決定敏感操作
if trust.current() >= 0.7:
    # 允許自主操作
    pass
elif trust.current() >= 0.4:
    # 允許但需人工監督
    pass
else:
    # 拒絕或要求明確核准
    pass

多代理信任：在代理委派給其他代理的系統中，每個代理維護其委派對象的信任分數：

class AgentTrustRegistry:
    def __init__(self):
        self.scores: dict[str, TrustScore] = {}

    def get_trust(self, agent_id: str) -> TrustScore:
        if agent_id not in self.scores:
            self.scores[agent_id] = TrustScore()
        return self.scores[agent_id]

    def most_trusted(self, agents: list[str]) -> str:
        return max(agents, key=lambda a: self.get_trust(a).current())

    def meets_threshold(self, agent_id: str, threshold: float) -> bool:
        return self.get_trust(agent_id).current() >= threshold

---

模式 5：稽核軌跡

所有代理動作的僅附加稽核日誌——對合規與除錯至關重要：

from dataclasses import dataclass, field
import json
import time

@dataclass
class AuditEntry:
    timestamp: float
    agent_id: str
    tool_name: str
    action: str           # "allowed", "denied", "error"
    policy_name: str
    details: dict = field(default_factory=dict)

class AuditTrail:
    """代理治理事件的僅附加稽核軌跡。"""
    def __init__(self):
        self._entries: list[AuditEntry] = []

    def log(self, agent_id: str, tool_name: str, action: str,
            policy_name: str, **details):
        self._entries.append(AuditEntry(
            timestamp=time.time(),
            agent_id=agent_id,
            tool_name=tool_name,
            action=action,
            policy_name=policy_name,
            details=details
        ))

    def denied(self) -> list[AuditEntry]:
        """取得所有被拒絕的動作——有助於安全審查。"""
        return [e for e in self._entries if e.action == "denied"]

    def by_agent(self, agent_id: str) -> list[AuditEntry]:
        return [e for e in self._entries if e.agent_id == agent_id]

    def export_jsonl(self, path: str):
        """匯出為 JSON Lines 格式，供日誌聚合系統使用。"""
        with open(path, "w") as f:
            for entry in self._entries:
                f.write(json.dumps({
                    "timestamp": entry.timestamp,
                    "agent_id": entry.agent_id,
                    "tool": entry.tool_name,
                    "action": entry.action,
                    "policy": entry.policy_name,
                    **entry.details
                }) + "\n")

---

模式 6：框架整合

PydanticAI

from pydantic_ai import Agent

policy = GovernancePolicy(
    name="support-bot",
    allowed_tools=["search_docs", "create_ticket"],
    blocked_patterns=[r"(?i)(ssn|social\s+security|credit\s+card)"],
    max_calls_per_request=20
)

agent = Agent("openai:gpt-4o", system_prompt="你是一位支援助理。")

@agent.tool
@govern(policy)
async def search_docs(ctx, query: str) -> str:
    """搜尋知識庫——受治理。"""
    return await kb.search(query)

@agent.tool
@govern(policy)
async def create_ticket(ctx, title: str, body: str) -> str:
    """建立支援工單——受治理。"""
    return await tickets.create(title=title, body=body)

CrewAI

from crewai import Agent, Task, Crew

policy = GovernancePolicy(
    name="research-crew",
    allowed_tools=["search", "analyze"],
    max_calls_per_request=30
)

# 在 Crew 層級套用治理
def governed_crew_run(crew: Crew, policy: GovernancePolicy):
    """包裝 Crew 執行，加入治理檢查。"""
    audit = AuditTrail()
    for agent in crew.agents:
        for tool in agent.tools:
            original = tool.func
            tool.func = govern(policy, audit_trail=audit)(original)
    result = crew.kickoff()
    return result, audit

OpenAI Agents SDK

from agents import Agent, function_tool

policy = GovernancePolicy(
    name="coding-agent",
    allowed_tools=["read_file", "write_file", "run_tests"],
    blocked_tools=["shell_exec"],
    max_calls_per_request=50
)

@function_tool
@govern(policy)
async def read_file(path: str) -> str:
    """讀取檔案內容——受治理。"""
    import os
    safe_path = os.path.realpath(path)
    if not safe_path.startswith(os.path.realpath(".")):
        raise ValueError("路徑遍歷已被治理封鎖")
    with open(safe_path) as f:
        return f.read()

---

治理層級

根據風險等級調整治理嚴格度：

層級	控制項	使用案例
開放	僅稽核，無限制	內部開發/測試
標準	工具允許清單 + 內容過濾	一般正式環境代理
嚴格	所有控制項 + 敏感操作需人工核准	金融、醫療、法律
鎖定	僅允許清單，無動態工具，完整稽核	合規關鍵系統

---

最佳實務

實務	理由
政策作為設定	將政策儲存在 YAML/JSON 中，而非寫死——可在不部署的情況下變更
最嚴格者為準	組合政策時，拒絕永遠覆蓋允許
飛行前意圖檢查	在工具執行之前分類意圖，而非之後
信任衰減	信任分數應隨時間衰減——需要持續的良好行為
僅附加稽核	絕不修改或刪除稽核條目——不可變性實現合規
失敗關閉	若治理檢查出錯，拒絕該動作而非允許
政策與邏輯分離	治理強制應獨立於代理業務邏輯

---

快速入門檢查清單

## 代理治理實作檢查清單

### 設定
- [ ] 定義治理政策（允許的工具、封鎖模式、速率限制）
- [ ] 選擇治理層級（開放/標準/嚴格/鎖定）
- [ ] 設定稽核軌跡儲存

### 實作
- [ ] 為所有工具函式加上 @govern 裝飾器
- [ ] 在使用者輸入處理中加入意圖分類
- [ ] 為多代理互動實作信任評分
- [ ] 連接稽核軌跡匯出

### 驗證
- [ ] 測試封鎖的工具是否正確被拒絕
- [ ] 測試內容過濾器是否捕捉敏感模式
- [ ] 測試速率限制行為
- [ ] 驗證稽核軌跡是否捕捉所有事件
- [ ] 測試政策組合（最嚴格者為準）

---

相關資源

• Agent Governance Toolkit — 完整治理框架
• AgentMesh Integrations — 框架特定套件
• OWASP Top 10 for LLM Applications