stateDiagram-v2
    [*] --> S1_Precise : 用户请求
    S1_Precise --> S1_Precise : 成功循环
    S1_Precise --> S2_ParamHallu : 模糊参数推断
    S1_Precise --> S4_OverInvoke : 不确定性过度补偿
    S2_ParamHallu --> S7_SelfRepair : 错误检测触发
    S2_ParamHallu --> S6_ChainCollapse : 错误级联
    S3_ToolMismatch --> S2_ParamHallu : 工具边界模糊
    S4_OverInvoke --> S5_Refusal : 资源耗尽
    S5_Refusal --> S8_EmergentNovel : 替代路径发现
    S6_ChainCollapse --> S7_SelfRepair : 外部干预
    S7_SelfRepair --> S1_Precise : 修复成功
    S7_SelfRepair --> S6_ChainCollapse : 修复失败
    S8_EmergentNovel --> S1_Precise : 验证通过
    S8_EmergentNovel --> S6_ChainCollapse : 验证失败

# 概念示例：基于内部表征的工具调用置信度评估
class ToolCallConfidence:
    """利用模型输出中的元信息评估工具调用可靠性"""

    def evaluate(self, tool_call: dict, schema: dict) -> dict:
        checks = {
            "param_exists": self._check_param_existence(tool_call, schema),
            "type_match": self._check_type_conformance(tool_call, schema),
            "enum_valid": self._check_enum_values(tool_call, schema),
            "semantic_plausibility": self._check_semantics(tool_call),
        }
        confidence = sum(checks.values()) / len(checks)

        if confidence < 0.5:
            return {"action": "reject", "confidence": confidence, "reason": checks}
        elif confidence < 0.8:
            return {"action": "confirm_with_user", "confidence": confidence}
        return {"action": "execute", "confidence": confidence}

    def _check_enum_values(self, call, schema):
        """枚举参数幻觉是最常见的失败模式"""
        for prop, spec in schema.get("properties", {}).items():
            if "enum" in spec and prop in call.get("arguments", {}):
                if call["arguments"][prop] not in spec["enum"]:
                    return 0.0
        return 1.0

class ToolCircuitBreaker:
    """工具调用断路器 - 三态保护"""

    def __init__(self, tool_name, failure_threshold=3, recovery_timeout=30):
        self.tool_name = tool_name
        self.state = "CLOSED"  # CLOSED -> OPEN -> HALF_OPEN
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = None

    async def call(self, tool_fn, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"  # 允许一次试探性调用
            else:
                return {"error": "circuit_open", "fallback": self._get_fallback()}

        try:
            result = await tool_fn(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"  # 试探成功，恢复
            self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
                logger.warning(f"Circuit OPEN for {self.tool_name}")
            return {"error": str(e), "fallback": self._get_fallback()}

flowchart TD
    A[工具调用失败] --> B{错误类型判断}
    B -->|参数错误| C[参数自纠错]
    B -->|工具不可用| D[工具降级切换]
    B -->|逻辑错误| E[策略重规划]
    C --> F{修复验证}
    D --> F
    E --> F
    F -->|通过| G[S7: 自修复成功]
    F -->|再次失败| H{重试次数}
    H -->|< 阈值| A
    H -->|>= 阈值| I[S6: 链式崩溃]
    A --> J{替代路径发现}
    J -->|有效| K[S8: 涌现创新]
    J -->|无效| I

// MCP工具调用请求（JSON-RPC 2.0）
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": {
      "sql": "SELECT * FROM users WHERE status = 'active' LIMIT 10",
      "database": "production"
    }
  }
}

// MCP工具调用响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"id\": 1, \"name\": \"Alice\", \"status\": \"active\"}]"
      }
    ]
  }
}

from jsonschema import validate, ValidationError

class SchemaGuard:
    """第一道防线：调用前Schema校验"""

    def guard(self, tool_call: dict, tool_schema: dict) -> bool:
        try:
            validate(instance=tool_call["arguments"], schema=tool_schema)
            return True
        except ValidationError as e:
            logger.warning(
                f"Schema violation in {tool_call['name']}: "
                f"path={list(e.path)}, msg={e.message}"
            )
            # 将错误信息注入上下文，触发S7自修复
            return False

class SemanticGuard:
    """第二道防线：语义合理性检查"""

    # 规则示例：日期合理性、ID格式、数值范围
    RULES = {
        "date_plausibility": lambda v: not _is_impossible_date(v),
        "id_format": lambda v: re.match(r'^[a-zA-Z0-9-]+$', str(v)),
        "limit_range": lambda v: 1 <= int(v) <= 10000,
    }

    def check(self, tool_call: dict, context: dict) -> list:
        violations = []
        for param, value in tool_call.get("arguments", {}).items():
            if rule := self.RULES.get(param):
                if not rule(value):
                    violations.append(f"Semantic violation: {param}={value}")
        return violations

class BehaviorMonitor:
    """第三道防线：行为异常检测"""

    def __init__(self):
        self.call_history = []
        self.tool_counters = defaultdict(int)

    def record(self, tool_call: dict, result: dict):
        self.call_history.append({
            "tool": tool_call["name"],
            "args_hash": hash(frozenset(tool_call.get("arguments", {}).items())),
            "success": "error" not in result,
            "timestamp": time.time()
        })
        self.tool_counters[tool_call["name"]] += 1

    def detect_anomalies(self) -> list:
        anomalies = []

        # 检测重试风暴（S4 -> S6路径）
        recent = self.call_history[-10:]
        if len(recent) >= 3:
            same_call_count = sum(
                1 for i in range(len(recent)-2)
                if recent[i]["args_hash"] == recent[i+1]["args_hash"] == recent[i+2]["args_hash"]
            )
            if same_call_count > 0:
                anomalies.append("retry_storm_detected")

        # 检测工具过载（S4状态）
        total = sum(self.tool_counters.values())
        for tool, count in self.tool_counters.items():
            if count / total > 0.6:
                anomalies.append(f"tool_overuse: {tool} ({count}/{total})")

        return anomalies

graph TB
    subgraph "Phase 1: Schema硬校验"
        A1[JSON Schema验证] --> A2[类型检查]
        A1 --> A3[枚举校验]
        A1 --> A4[必填项检查]
    end
    subgraph "Phase 2: 语义软校验"
        B1[日期合理性] --> B2[ID格式验证]
        B1 --> B3[数值范围检查]
        B1 --> B4[跨参数一致性]
    end
    subgraph "Phase 3: 行为异常检测"
        C1[重试风暴检测] --> C2[工具过载预警]
        C1 --> C3[断路器保护]
        C1 --> C4[上下文污染隔离]
    end
    subgraph "Phase 4: 涌现行为治理"
        D1[八态转移监控] --> D2[自修复率追踪]
        D1 --> D3[创新行为审核]
        D1 --> D4[模型对比A/B测试]
    end
    A4 --> B1
    B4 --> C1
    C4 --> D1
    style A1 fill:#4CAF50,color:white
    style B1 fill:#2196F3,color:white
    style C1 fill:#FF9800,color:white
    style D1 fill:#9C27B0,color:white