第十一章：容错与可靠性工程¶

from typing import Dict, Any
from datetime import datetime
from collections import defaultdict
import time

class MetricsCollector:
    """指标收集器：计数器、仪表、直方图"""

    def __init__(self):
        self.counters: Dict[str, int] = defaultdict(int)
        self.gauges: Dict[str, float] = {}
        self.histograms: Dict[str, list] = defaultdict(list)
        self.start_time = time.time()

    def increment_counter(self, name: str, value: int = 1, tags: Dict = None) -> None:
        """增加计数器"""
        key = self._make_key(name, tags)
        self.counters[key] += value

    def set_gauge(self, name: str, value: float, tags: Dict = None) -> None:
        """设置仪表值"""
        key = self._make_key(name, tags)
        self.gauges[key] = value

    def record_histogram(self, name: str, value: float, tags: Dict = None) -> None:
        """记录直方图值"""
        key = self._make_key(name, tags)
        self.histograms[key].append({
            "value": value,
            "timestamp": datetime.now().isoformat(),
        })

    def _make_key(self, name: str, tags: Dict) -> str:
        """生成带标签的指标键"""
        if not tags:
            return name
        tag_str = ",".join(f"{k}={v}" for k, v in sorted(tags.items()))
        return f"{name}{{{tag_str}}}"

    def get_percentile(self, name: str, percentile: int) -> float:
        """获取百分位数(如 P50、P99)"""
        values = sorted(self._histogram_values(name))
        if not values:
            return 0.0
        index = int(len(values) * percentile / 100)
        return values[index] if index < len(values) else 0.0

    def _counter_sum(self, name: str) -> int:
        """聚合某个指标名下的所有标签维度"""
        return sum(
            value
            for key, value in self.counters.items()
            if key == name or key.startswith(f"{name}{{")
        )

    def _histogram_values(self, name: str) -> list[float]:
        """聚合某个指标名下的所有标签维度样本"""
        values = []
        for key, samples in self.histograms.items():
            if key == name or key.startswith(f"{name}{{"):
                values.extend(sample["value"] for sample in samples)
        return values

    def get_summary(self) -> Dict[str, Any]:
        """获取指标摘要"""
        return {
            "counters": dict(self.counters),
            "gauges": dict(self.gauges),
            "histograms_summary": {
                name: {
                    "count": len(values),
                    "min": min(v["value"] for v in values),
                    "max": max(v["value"] for v in values),
                    "p50": self.get_percentile(name, 50),
                    "p99": self.get_percentile(name, 99),
                }
                for name, values in self.histograms.items()
            },
            "uptime_seconds": time.time() - self.start_time,
        }

class AgentMetrics:
    """Agent相关的指标"""

    def __init__(self):
        self.collector = MetricsCollector()

    def record_tool_call(
        self,
        tool_name: str,
        duration_ms: float,
        success: bool,
        tokens_used: int = 0,
    ) -> None:
        """记录单次工具调用：调用次数、延迟、Token、成功率"""
        tags = {"tool": tool_name, "success": str(success)}
        self.collector.increment_counter("tool_calls_total", tags=tags)
        self.collector.record_histogram("tool_call_duration_ms", duration_ms, tags=tags)
        if tokens_used > 0:
            self.collector.record_histogram("tool_tokens_used", tokens_used, tags=tags)
        if success:
            self.collector.increment_counter("tool_calls_success", tags=tags)
        else:
            self.collector.increment_counter("tool_calls_failed", tags=tags)

    def record_agent_iteration(
        self,
        agent_id: str,
        duration_ms: float,
        tokens_used: int,
        tools_called: int,
    ) -> None:
        """记录 Agent 一次迭代的统计数据"""
        tags = {"agent": agent_id}
        # ... (省略辅助方法)
        self.collector.record_histogram("agent_iteration_ms", duration_ms, tags=tags)
        self.collector.record_histogram("agent_tokens_per_iteration", tokens_used, tags=tags)
        self.collector.record_histogram("agent_tools_per_iteration", tools_called, tags=tags)

    def get_dashboard_data(self) -> Dict[str, Any]:
        """获取仪表板数据"""
        summary = self.collector.get_summary()
        return {
            "total_tool_calls": self.collector._counter_sum("tool_calls_total"),
            "tool_success_rate": self._calculate_success_rate(summary),
            "avg_tool_latency_ms": self._calculate_avg_latency(summary),
            "p99_latency_ms": self.collector.get_percentile("tool_call_duration_ms", 99),
            "total_tokens": sum(self.collector._histogram_values("tool_tokens_used")),
        }

    def _calculate_success_rate(self, summary: Dict) -> float:
        """计算成功率百分比"""
        success = self.collector._counter_sum("tool_calls_success")
        failed = self.collector._counter_sum("tool_calls_failed")
        total = success + failed
        return (success / total * 100) if total > 0 else 0.0

    def _calculate_avg_latency(self, summary: Dict) -> float:
        """计算平均延迟"""
        values = self.collector._histogram_values("tool_call_duration_ms")
        return (sum(values) / len(values)) if values else 0.0

import json
from enum import Enum
from datetime import datetime

class LogLevel(Enum):
    DEBUG = 10
    INFO = 20
    WARNING = 30
    ERROR = 40
    CRITICAL = 50

class StructuredLogger:
    """结构化日志记录器"""

    def __init__(self, service_name: str):
        self.service_name = service_name
        self.logs = []

    def log(self, level: LogLevel, message: str, **kwargs) -> None:
        """记录结构化日志为 JSON"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "level": level.name,
            "service": self.service_name,
            "message": message,
            **kwargs,
        }
        self.logs.append(log_entry)
        print(json.dumps(log_entry))

    # 便捷方法
    def debug(self, message: str, **kwargs) -> None:
        self.log(LogLevel.DEBUG, message, **kwargs)

    def info(self, message: str, **kwargs) -> None:
        self.log(LogLevel.INFO, message, **kwargs)

    def warning(self, message: str, **kwargs) -> None:
        self.log(LogLevel.WARNING, message, **kwargs)

    def error(self, message: str, **kwargs) -> None:
        self.log(LogLevel.ERROR, message, **kwargs)

    def critical(self, message: str, **kwargs) -> None:
        self.log(LogLevel.CRITICAL, message, **kwargs)

# 日志记录的智能体专用方法
class StructuredLogger:
    # ... (其他方法如上)

    def log_tool_call(
        self,
        tool_name: str,
        agent_id: str,
        duration_ms: float,
        success: bool,
        error: str = None,
    ) -> None:
        """记录工具调用的性能数据"""
        self.info(
            f"Tool call: {tool_name}",
            tool_name=tool_name,
            agent_id=agent_id,
            duration_ms=duration_ms,
            success=success,
            error=error,
        )

    def log_agent_decision(
        self,
        agent_id: str,
        decision: str,
        reasoning: str,
        confidence: float,
    ) -> None:
        """记录 Agent 决策及其推理"""
        self.info(
            f"Agent decision: {decision}",
            agent_id=agent_id,
            decision=decision,
            reasoning=reasoning,
            confidence=confidence,
        )

import uuid
import time
from dataclasses import dataclass
from typing import Dict, Any, List
from collections import defaultdict

@dataclass
class Span:
    """追踪跨度：表示一个操作的开始、结束和时间"""
    trace_id: str
    span_id: str
    parent_span_id: str
    operation_name: str
    start_time: float
    end_time: float = None
    duration_ms: float = None
    tags: Dict[str, Any] = None
    status: str = "ok"

    def finish(self) -> None:
        """完成 Span 的记录"""
        self.end_time = time.time()
        self.duration_ms = (self.end_time - self.start_time) * 1000

class Tracer:
    """分布式追踪记录器"""

    def __init__(self, service_name: str):
        self.service_name = service_name
        self.traces: Dict[str, list] = defaultdict(list)
        self.current_trace_id: str = None
        self.current_span_id: str = None

    def start_trace(self, request_id: str = None) -> str:
        """开始新的追踪(对应一个请求)"""
        self.current_trace_id = request_id or str(uuid.uuid4())
        self.current_span_id = None
        return self.current_trace_id

    def start_span(
        self,
        operation_name: str,
        parent_span_id: str = None,
        tags: Dict = None,
    ) -> Span:
        """开始新的 Span(子操作)"""
        span_id = str(uuid.uuid4())
        parent = parent_span_id or self.current_span_id
        span = Span(
            trace_id=self.current_trace_id,
            span_id=span_id,
            parent_span_id=parent,
            operation_name=operation_name,
            start_time=time.time(),
            tags=tags or {},
        )
        self.traces[self.current_trace_id].append(span)
        self.current_span_id = span_id
        return span

    def get_trace(self, trace_id: str) -> Dict[str, Any]:
        """获取完整的追踪数据(包括所有 Span)"""
        spans = self.traces.get(trace_id, [])
        root_spans = [s for s in spans if s.parent_span_id is None]
        total_duration = sum(s.duration_ms for s in spans if s.duration_ms)
        return {
            "trace_id": trace_id,
            "service": self.service_name,
            "spans": len(spans),
            "total_duration_ms": total_duration,
            "root_operations": [s.operation_name for s in root_spans],
        }

# 集成使用：指标 + 日志 + 追踪
logger = StructuredLogger("MiniHarness")
metrics = AgentMetrics()
tracer = Tracer("MiniHarness")

async def execute_tool_with_observability(
    tool_name: str,
    arguments: Dict,
    agent_id: str,
) -> Dict:
    """执行工具时记录完整的可观测信息"""
    trace_id = tracer.start_trace()
    span = tracer.start_span(f"tool.{tool_name}", tags={"tool": tool_name, "agent": agent_id})
    start_time = time.time()

    try:
        logger.debug(f"Calling tool: {tool_name}", tool=tool_name)
        # ... (省略实际工具调用逻辑)
        success, result, error = await registry.call_tool(tool_name, arguments, agent_id)
        duration_ms = (time.time() - start_time) * 1000

        if success:
            span.status = "ok"
            logger.info(f"Tool call succeeded", tool=tool_name, duration_ms=int(duration_ms))
            metrics.record_tool_call(tool_name, duration_ms, True)
        else:
            span.status = "error"
            logger.error(f"Tool call failed", tool=tool_name, error=error)
            metrics.record_tool_call(tool_name, duration_ms, False)

        return {"success": success, "result": result, "error": error}

    except Exception as e:
        span.status = "error"
        logger.critical(f"Tool call exception", tool=tool_name, exception=str(e))
        raise
    finally:
        span.finish()

class ObservabilityDashboard:
    """可观测性仪表板：聚合所有可观测数据"""

    def __init__(self, metrics: AgentMetrics, logger: StructuredLogger, tracer: Tracer):
        self.metrics = metrics
        self.logger = logger
        self.tracer = tracer

    def get_dashboard(self) -> Dict[str, Any]:
        """获取完整的仪表板数据(KPI + 日志 + 追踪)"""
        return {
            "metrics": self.metrics.get_dashboard_data(),
            "recent_logs": self.logger.logs[-100:],
            "active_traces": len([l for l in self.logger.logs if "trace_id" in l]),
            "error_rate_percent": self._calculate_error_rate(),
            "health_status": self._determine_health_status(),
        }

    def _calculate_error_rate(self) -> float:
        """计算错误日志百分比"""
        total_logs = len(self.logger.logs)
        error_logs = sum(1 for l in self.logger.logs if l.get("level") in ["ERROR", "CRITICAL"])
        return (error_logs / total_logs * 100) if total_logs > 0 else 0.0

    def _determine_health_status(self) -> str:
        """根据错误率判断系统健康状态"""
        error_rate = self._calculate_error_rate()
        if error_rate > 5:
            return "critical"
        elif error_rate > 1:
            return "warning"
        else:
            return "healthy"

from enum import Enum
from dataclasses import dataclass
from typing import Any, Dict

class RiskLevel(Enum):
    """风险等级"""
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4

@dataclass
class RiskAssessment:
    """风险评估结果"""
    level: RiskLevel
    score: float  # 0-100
    factors: Dict[str, float]
    explanation: str
    recommended_action: str
    requires_approval: bool

class RiskEvaluator:
    """风险评估器：加权因子模型"""

    def __init__(self):
        self.weights = {
            'data_sensitivity': 0.3,
            'financial_impact': 0.25,
            'irreversibility': 0.2,
            'confidence': 0.15,
            'user_authorization': 0.1
        }

    def assess(
        self,
        action_type: str,
        params: Dict[str, Any],
        context: Dict[str, Any],
        model_confidence: float = 0.8
    ) -> RiskAssessment:
        """评估操作的风险等级(0-100分)"""
        factors = {}
        factors['data_sensitivity'] = self._assess_data_sensitivity(params, context)
        factors['financial_impact'] = self._assess_financial_impact(action_type, params)
        factors['irreversibility'] = self._assess_irreversibility(action_type)
        factors['confidence'] = max(0, 1.0 - model_confidence)
        factors['user_authorization'] = self._assess_authorization(context)

        # 加权求和得到风险分数
        score = sum(factors.get(f, 0) * weight for f, weight in self.weights.items()) * 100

        # 分数映射到风险等级
        if score < 20:
            level = RiskLevel.LOW
        elif score < 50:
            level = RiskLevel.MEDIUM
        elif score < 80:
            level = RiskLevel.HIGH
        else:
            level = RiskLevel.CRITICAL

        return RiskAssessment(
            level=level,
            score=score,
            factors=factors,
            explanation=self._explain_risk(level, factors),
            recommended_action=self._recommend_action(level),
            requires_approval=level in (RiskLevel.HIGH, RiskLevel.CRITICAL)
        )

# RiskEvaluator 的评分方法
class RiskEvaluator:
    # ... (初始化如上)

    def _assess_data_sensitivity(self, params: dict, context: dict) -> float:
        """检查参数中是否包含敏感字段(密码、令牌等)"""
        sensitivity_keywords = ['password', 'token', 'secret', 'key', 'ssn', 'credit_card']
        for key in params:
            if any(keyword in key.lower() for keyword in sensitivity_keywords):
                return 1.0
        user_level = context.get('user_permission_level', 0)
        return user_level / 10

    def _assess_financial_impact(self, action_type: str, params: dict) -> float:
        """评估财务操作的风险和金额"""
        financial_actions = {
            'transfer': 0.9, 'charge': 0.8, 'refund': 0.7, 'delete_transaction': 0.85
        }
        base_score = financial_actions.get(action_type, 0.0)
        amount = params.get('amount', 0)
        if amount > 10000:
            base_score = min(1.0, base_score + 0.2)
        return base_score

    def _assess_irreversibility(self, action_type: str) -> float:
        """评估操作是否不可逆"""
        irreversible_actions = {
            'delete': 1.0, 'archive': 0.8, 'disable': 0.6, 'publish': 0.4, 'modify': 0.3
        }
        return irreversible_actions.get(action_type, 0.2)

    def _assess_authorization(self, context: dict) -> float:
        """检查用户是否已明确授权"""
        has_explicit_consent = context.get('has_explicit_consent', False)
        return 0.0 if has_explicit_consent else 1.0

    def _explain_risk(self, level: RiskLevel, factors: dict) -> str:
        """生成人类可读的风险解释"""
        explanations = {
            RiskLevel.LOW: "Low risk: Action is safe to execute automatically.",
            RiskLevel.MEDIUM: "Medium risk: Recommend human review before execution.",
            RiskLevel.HIGH: "High risk: Requires explicit human approval.",
            RiskLevel.CRITICAL: "Critical risk: Requires manager approval and logging."
        }
        return explanations[level]

    def _recommend_action(self, level: RiskLevel) -> str:
        """根据风险等级推荐执行方式"""
        actions = {
            RiskLevel.LOW: "auto_execute",
            RiskLevel.MEDIUM: "request_approval",
            RiskLevel.HIGH: "require_approval",
            RiskLevel.CRITICAL: "escalate_to_manager"
        }
        return actions[level]

from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import List, Optional, Callable
import uuid

class ApprovalStatus(Enum):
    """审批请求的状态"""
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    NEEDS_CLARIFICATION = "needs_clarification"
    EXPIRED = "expired"

@dataclass
class ApprovalRequest:
    """审批请求对象"""
    id: str
    action_type: str
    action_params: dict
    requester_id: str
    request_time: datetime
    required_approvals: int
    approval_timeout_minutes: int = 60
    approvals: List[dict] = field(default_factory=list)
    rejections: List[dict] = field(default_factory=list)
    status: ApprovalStatus = ApprovalStatus.PENDING
    notes: str = ""

    def is_approved(self) -> bool:
        """检查是否已获得所有必要的批准"""
        if self.status == ApprovalStatus.APPROVED:
            return True
        if self.status in (ApprovalStatus.REJECTED, ApprovalStatus.EXPIRED):
            return False
        return len(self.approvals) >= self.required_approvals

    def is_expired(self) -> bool:
        """检查请求是否已超时"""
        expiry_time = self.request_time + timedelta(minutes=self.approval_timeout_minutes)
        return datetime.now() > expiry_time

    def get_remaining_approvals(self) -> int:
        """获取还需要的批准数"""
        return max(0, self.required_approvals - len(self.approvals))

# 审批工作流管理器
class ApprovalWorkflow:
    """管理审批请求的生命周期"""

    def __init__(self):
        self.requests: dict[str, ApprovalRequest] = {}
        self.callbacks: dict[str, List[Callable]] = {}

    def create_request(
        self,
        action_type: str,
        action_params: dict,
        requester_id: str,
        required_approvals: int = 1,
        timeout_minutes: int = 60
    ) -> ApprovalRequest:
        """创建新的审批请求"""
        request_id = str(uuid.uuid4())
        request = ApprovalRequest(
            id=request_id,
            action_type=action_type,
            action_params=action_params,
            requester_id=requester_id,
            request_time=datetime.now(),
            required_approvals=required_approvals,
            approval_timeout_minutes=timeout_minutes
        )
        self.requests[request_id] = request
        self._trigger_callbacks('on_request_created', request)
        return request

    def approve(self, request_id: str, approver_id: str, comments: str = "") -> bool:
        """批准请求"""
        if request_id not in self.requests:
            return False
        request = self.requests[request_id]
        if request.is_expired():
            request.status = ApprovalStatus.EXPIRED
            self._trigger_callbacks('on_request_expired', request)
            return False
        request.approvals.append({'approver_id': approver_id, 'time': datetime.now(), 'comments': comments})
        if request.is_approved():
            request.status = ApprovalStatus.APPROVED
            self._trigger_callbacks('on_request_approved', request)
        return True

    def reject(self, request_id: str, reviewer_id: str, reason: str) -> bool:
        """拒绝请求"""
        if request_id not in self.requests:
            return False
        request = self.requests[request_id]
        request.status = ApprovalStatus.REJECTED
        request.rejections.append({'reviewer_id': reviewer_id, 'time': datetime.now(), 'reason': reason})
        self._trigger_callbacks('on_request_rejected', request)
        return True

    def request_clarification(self, request_id: str, clarifier_id: str, question: str) -> bool:
        """请求澄清"""
        if request_id not in self.requests:
            return False
        request = self.requests[request_id]
        request.status = ApprovalStatus.NEEDS_CLARIFICATION
        request.notes = question
        self._trigger_callbacks('on_clarification_requested', request)
        return True

    def get_request(self, request_id: str) -> Optional[ApprovalRequest]:
        """获取单个请求"""
        return self.requests.get(request_id)

    def get_pending_requests(self) -> List[ApprovalRequest]:
        """获取所有待审批的请求"""
        return [r for r in self.requests.values() if r.status == ApprovalStatus.PENDING and not r.is_expired()]

    def register_callback(self, event: str, callback: Callable):
        """为事件注册回调处理器"""
        if event not in self.callbacks:
            self.callbacks[event] = []
        self.callbacks[event].append(callback)

    def _trigger_callbacks(self, event: str, request: ApprovalRequest):
        """触发事件回调"""
        if event in self.callbacks:
            for callback in self.callbacks[event]:
                callback(request)

from anthropic import Anthropic
from dataclasses import dataclass
from typing import List
import json

@dataclass
class PermissionDeclaration:
    """权限声明的解释结果"""
    user_input: str
    interpreted_permissions: List[str]
    confidence: float
    explanation: str

class PermissionInterpreter:
    """自然语言权限解释器"""

    def __init__(self):
        self.client = Anthropic()

    def interpret(self, user_declaration: str) -> PermissionDeclaration:
        """将用户的自然语言声明转换为权限列表"""
        system_prompt = """You are a permission interpreter for AI agents.
Analyze the user's declaration and extract the implied permissions.
Return a JSON object with:
{
  "permissions": ["permission1", "permission2"],
  "confidence": 0.95,
  "explanation": "The user is granting permission to X because Y"
}

Common patterns: "You can X" → grant X; "Don't X" → deny X; "Ask before Y" → conditional Y
"""
        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=500,
            system=system_prompt,
            messages=[{"role": "user", "content": f"Interpret: {user_declaration}"}]
        )
        result = json.loads(response.content[0].text)
        return PermissionDeclaration(
            user_input=user_declaration,
            interpreted_permissions=result['permissions'],
            confidence=result['confidence'],
            explanation=result['explanation']
        )

    def ask_for_clarification(self, declaration: str, context: str) -> str:
        """在权限解释不清晰时请求澄清"""
        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=300,
            messages=[{"role": "user", "content": f"""User said: "{declaration}"\nContext: {context}\nSuggest clarifying questions."""}]
        )
        return response.content[0].text

# 权限解释器使用示例
if __name__ == "__main__":
    interpreter = PermissionInterpreter()

    # 简单授权:删除缓存文件
    decl1 = interpreter.interpret("You can delete old cache files")
    print(f"Permissions: {decl1.interpreted_permissions}")
    print(f"Confidence: {decl1.confidence}")

    # 条件授权:数据库迁移但要求在删除表前确认
    decl2 = interpreter.interpret("Go ahead with migration, but confirm before dropping tables")
    print(f"Permissions: {decl2.interpreted_permissions}")

    # 请求澄清:对于含糊的权限声明
    clarification = interpreter.ask_for_clarification(
        "You can handle my emails",
        "Agent considering: send, delete, move"
    )
    print(f"Clarification questions:\n{clarification}")

from enum import Enum
from typing import Optional
from datetime import datetime, timedelta

class PermissionLevel(Enum):
    """6 级信任模型"""
    MANUAL_ONLY = "Manual-only"
    APPROVE_ALWAYS = "Approve-always"
    APPROVE_ONCE = "Approve-once"
    ASK_FIRST = "Ask-first"
    AUTO_WITH_NOTIFICATION = "Auto-with-notification"
    FULL_TRUST = "Full-trust"

class AskFirstManager:
    """Ask-First 权限管理器"""

    def __init__(self):
        self.user_decisions = {}  # {user_id: {action_type: (decision, timestamp)}}

    def should_ask(
        self,
        user_id: str,
        action_type: str,
        permission_level: PermissionLevel,
        context_changed: bool = False
    ) -> bool:
        """判断是否需要询问用户"""
        if permission_level == PermissionLevel.FULL_TRUST:
            return False
        if permission_level in (PermissionLevel.MANUAL_ONLY, PermissionLevel.APPROVE_ALWAYS, PermissionLevel.APPROVE_ONCE):
            return True

        if permission_level == PermissionLevel.ASK_FIRST:
            # 检查是否已有用户决策
            if user_id in self.user_decisions:
                action_key = self._normalize_action(action_type)
                if action_key in self.user_decisions[user_id]:
                    decision, timestamp = self.user_decisions[user_id][action_key]
                    # 24 小时后或上下文变化时重新询问
                    if context_changed or (datetime.now() - timestamp > timedelta(hours=24)):
                        return True
                    return not decision
            return True  # 首次询问

        if permission_level == PermissionLevel.AUTO_WITH_NOTIFICATION:
            return False
        return False

    def record_decision(
        self,
        user_id: str,
        action_type: str,
        approved: bool,
        remember_for_hours: int = 24
    ):
        """记录用户决定以便下次记忆"""
        if user_id not in self.user_decisions:
            self.user_decisions[user_id] = {}
        action_key = self._normalize_action(action_type)
        self.user_decisions[user_id][action_key] = (approved, datetime.now())

    def clear_decisions(self, user_id: str):
        """清除用户的决定缓存"""
        if user_id in self.user_decisions:
            del self.user_decisions[user_id]

    @staticmethod
    def _normalize_action(action_type: str) -> str:
        """规范化操作类型为小写"""
        return action_type.lower().strip()

# Ask-First 对话生成和解析
class AskFirstDialog:
    """Ask-First 对话生成和响应解析"""

    @staticmethod
    def generate_prompt(action_type: str, action_description: str, impact: str) -> str:
        """生成询问用户的提示"""
        return f"""The AI agent wants to perform the following action:

Action: {action_type}
Description: {action_description}
Impact: {impact}

Should I proceed? Please reply with:
- "yes" or "approve" to allow this action
- "no" or "deny" to block this action
- "ask next time" if you want to be asked again next time
- Or provide modified parameters"""

    @staticmethod
    def parse_response(user_response: str) -> dict:
        """解析用户的响应"""
        response_lower = user_response.lower().strip()

        if response_lower in ('yes', 'approve', 'ok', 'proceed'):
            return {'approved': True, 'remember': True}
        elif response_lower in ('no', 'deny', 'block', 'cancel'):
            return {'approved': False, 'remember': True}
        elif 'ask next time' in response_lower or 'ask again' in response_lower:
            return {'approved': None, 'remember': False}
        else:
            return {'approved': None, 'modified_params': user_response}