AWS开源Dogwood：用时序策略语言治理AI代理的工具调用链

# Dogwood策略示例：速率限制
# 当代理在1小时内向外部API发起超过10次请求时拒绝
permit(
    principal == agent::"payment_agent",
    action == tool::"call_external_api",
    resource == external_service::"stripe_api"
) when {
    count_within(duration("PT1H"), action) <= 10
};

# Dogwood策略示例：审批门控
# 在执行高敏感操作前必须获得人类审批
permit(
    principal == agent::"data_agent",
    action == tool::"modify_production_config",
    resource == config::"prod_database"
) when temporal {
    formerly(duration("PT24H"), action == tool::"request_human_approval")
};

# Dogwood策略示例：数据隔离
# 访问机密数据后24小时内禁止外部调用
forbid(
    principal == agent::"research_agent",
    action == tool::"call_external_service",
    resource == external_service::"webhook"
) when temporal {
    formerly(duration("PT24H"), action == tool::"access_confidential_data")
};

# 场景：三个并发的$2000转账请求
# 时间线：
# T0: 请求1 ($2000) 到达
# T1: 请求2 ($2000) 到达  
# T2: 请求3 ($2000) 到达
# T3: 响应1 返回
# T4: 响应2 返回
# T5: 响应3 返回

# --- 错误策略：基于响应事件计算累计金额 ---
# 在T0时，策略查看响应历史：无已完成转账，累计$0
# 在T1时，策略查看响应历史：仍无已完成转账，累计$0
# 在T2时，策略查看响应历史：仍无已完成转账，累计$0
# 结果：三笔$2000全部通过，累计$6000超出$5000限额！

permit(
    principal == agent::"transfer_agent",
    action == tool::"execute_transfer",
    resource == account::"checking"
) when temporal {
    sum_within(duration("PT24H"), response.amount) < 5000
};

# --- 正确策略：基于请求事件计算累计金额 ---
# 在T0时，策略查看请求历史：无请求，累计$0，允许
# 在T1时，策略查看请求历史：1个请求$2000，累计$2000，允许
# 在T2时，策略查看请求历史：2个请求共$4000，累计$4000+$2000=$6000>$5000，拒绝！

permit(
    principal == agent::"transfer_agent",
    action == tool::"execute_transfer",
    resource == account::"checking"
) when temporal {
    sum_within(duration("PT24H"), request.amount) < 5000
};

[AI代理] → [工具调用请求] → [AgentCore Policy引擎]
                                    |
                              [事件历史存储]
                                    |
                           [Dogwood时序评估]
                                    |
                        [允许 / 拒绝 / 需审批]
                                    |
                              [执行或阻止]
                                    |
                           [记录事件到历史]

import boto3
from datetime import datetime, timedelta

class AgentGovernanceManager:
    """AI代理时序治理管理器"""
    
    def __init__(self, policy_store_id):
        self.agentcore = boto3.client('agentcore-policy')
        self.policy_store_id = policy_store_id
        self.event_store = EventStore()  # 事件存储抽象
    
    def check_tool_call(self, agent_id, tool_name, tool_input):
        """检查代理工具调用是否被策略允许"""
        # 获取代理的事件历史
        event_history = self.event_store.get_events(
            agent_id=agent_id,
            time_window=timedelta(hours=24)
        )
        
        # 评估策略
        response = self.agentcore.evaluate(
            policyStoreId=self.policy_store_id,
            principal={
                'entityType': 'agent',
                'entityId': agent_id
            },
            action={
                'actionType': 'tool',
                'actionId': tool_name
            },
            resource={
                'resourceType': 'service',
                'resourceId': tool_input.get('target_service', 'default')
            },
            context={
                'eventHistory': event_history,
                'currentInput': tool_input,
                'currentTimestamp': datetime.utcnow().isoformat()
            }
        )
        
        decision = response.get('decision', 'DENY')
        
        if decision == 'ALLOW':
            # 记录请求事件
            self.event_store.record_event(
                agent_id=agent_id,
                action=tool_name,
                input_data=tool_input,
                event_type='request',
                timestamp=datetime.utcnow()
            )
            return True, None
        elif decision == 'DENY':
            reason = response.get('deniedReason', 'Policy denied')
            return False, reason
        else:
            # 需要人类审批
            return False, 'HUMAN_APPROVAL_REQUIRED'
    
    def record_tool_response(self, agent_id, tool_name, response_data):
        """记录工具调用响应"""
        self.event_store.record_event(
            agent_id=agent_id,
            action=tool_name,
            input_data=response_data,
            event_type='response',
            timestamp=datetime.utcnow()
        )


class EventStore:
    """事件历史存储"""
    
    def __init__(self):
        # 实际实现应使用DynamoDB、Aurora或专门的审计日志
        self._events = []  # 简化示例
    
    def record_event(self, agent_id, action, input_data, event_type, timestamp):
        self._events.append({
            'agent_id': agent_id,
            'action': action,
            'input': input_data,
            'type': event_type,
            'timestamp': timestamp
        })
    
    def get_events(self, agent_id, time_window):
        cutoff = datetime.utcnow() - time_window
        return [e for e in self._events 
                if e['agent_id'] == agent_id and e['timestamp'] >= cutoff]