第七章：模型集成与输出治理¶

from enum import Enum
from dataclasses import dataclass
from typing import Optional, List

class ModelProviderType(Enum):
    CLAUDE = "claude"
    OPENAI = "openai"
    DEEPSEEK = "deepseek"
    GEMINI = "gemini"
    OLLAMA = "ollama"

@dataclass
class ModelConfig:
    """模型配置对象"""
    provider: ModelProviderType
    model_id: str
    api_key: Optional[str] = None
    api_endpoint: Optional[str] = None
    timeout: int = 30
    max_tokens: int = 4096
    temperature: Optional[float] = None  # 默认不传；部分 Claude 模型拒绝非默认采样参数

@dataclass
class ModelSelectionPolicy:
    """模型选择策略"""
    primary: ModelConfig
    fallback_chain: List[ModelConfig] = None
    cost_threshold: float = None  # 成本上限,超过则切换
    latency_threshold: int = None  # 延迟上限,超过则切换

import time

class CircuitBreaker:
    """熔断器,跟踪模型健康状态"""
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = "closed"  # closed, open, half-open
        self.last_failure_time = None

    def record_success(self):
        self.failure_count = 0
        self.state = "closed"

    def record_failure(self):
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            self.last_failure_time = time.time()

    def is_available(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open 允许尝试

from typing import Protocol, List, Any, Optional

class Message:
    def __init__(self, role: str, content: str):
        self.role = role  # "user", "assistant"
        self.content = content

class ProviderResponse:
    def __init__(self, content: str, tokens_used: int, model: str):
        self.content = content
        self.tokens_used = tokens_used
        self.model = model

class ModelProvider(Protocol):
    """LLM 供应商的统一接口"""

    def complete(
        self,
        messages: List[Message],
        temperature: Optional[float] = None,
        max_tokens: int = 4096,
    ) -> ProviderResponse:
        """完整的模型调用(无流式)"""
        ...

    def stream(
        self,
        messages: List[Message],
        temperature: Optional[float] = None,
        max_tokens: int = 4096,
    ):
        """流式模型调用,逐块返回"""
        ...

    def estimate_tokens(self, text: str) -> int:
        """估算文本的 token 数"""
        ...

    def validate_config(self) -> bool:
        """验证配置有效性(API 密钥等)"""
        ...

import anthropic
from typing import Generator, Optional

class ClaudeProvider:
    def __init__(self, config: ModelConfig):
        self.config = config
        self.client = anthropic.Anthropic(api_key=config.api_key)

    def complete(
        self,
        messages: List[Message],
        temperature: Optional[float] = None,
        max_tokens: int = 4096,
    ) -> ProviderResponse:
        """Claude 完整调用"""
        api_messages = [
            {"role": msg.role, "content": msg.content}
            for msg in messages
        ]
        kwargs = {
            "model": self.config.model_id,
            "max_tokens": max_tokens,
            "messages": api_messages,
        }
        if temperature is not None:
            kwargs["temperature"] = temperature
        response = self.client.messages.create(**kwargs)
        return ProviderResponse(
            content=response.content[0].text,
            tokens_used=(
                response.usage.input_tokens
                + getattr(response.usage, "cache_creation_input_tokens", 0)
                + getattr(response.usage, "cache_read_input_tokens", 0)
                + response.usage.output_tokens
            ),
            model=self.config.model_id,
        )

    def stream(
        self,
        messages: List[Message],
        temperature: Optional[float] = None,
        max_tokens: int = 4096,
    ) -> Generator[str, None, None]:
        """Claude 流式调用"""
        api_messages = [
            {"role": msg.role, "content": msg.content}
            for msg in messages
        ]
        kwargs = {
            "model": self.config.model_id,
            "max_tokens": max_tokens,
            "messages": api_messages,
        }
        if temperature is not None:
            kwargs["temperature"] = temperature
        with self.client.messages.stream(**kwargs) as stream:
            for text in stream.text_stream:
                yield text

    def estimate_tokens(self, text: str) -> int:
        """Claude token 计数"""
        response = self.client.messages.count_tokens(
            model=self.config.model_id,
            messages=[{"role": "user", "content": text}]
        )
        return response.input_tokens

    def validate_config(self) -> bool:
        try:
            self.estimate_tokens("test")
            return True
        except:
            return False

class ModelSelectionEngine:
    def __init__(self, policy: ModelSelectionPolicy):
        self.policy = policy
        self.breakers = {}
        self._init_breakers()

    def _init_breakers(self):
        for config in [self.policy.primary] + (
            self.policy.fallback_chain or []
        ):
            self.breakers[config.model_id] = CircuitBreaker()

    def select_model(self) -> ModelProvider:
        """根据健康状态选择可用模型"""
        candidates = [self.policy.primary] + (
            self.policy.fallback_chain or []
        )
        for config in candidates:
            breaker = self.breakers[config.model_id]
            if breaker.is_available():
                return self._create_provider(config)
        raise Exception("所有模型不可用")

    def mark_failure(self, model_id: str):
        """记录模型故障"""
        if model_id in self.breakers:
            self.breakers[model_id].record_failure()

    def mark_success(self, model_id: str):
        """记录模型成功"""
        if model_id in self.breakers:
            self.breakers[model_id].record_success()

    def _create_provider(self, config: ModelConfig):
        if config.provider == ModelProviderType.CLAUDE:
            return ClaudeProvider(config)
        # ... 其他 provider 实现

from dataclasses import dataclass, field

@dataclass
class CapabilityProfile:
    """描述一个模型能稳定承担什么，而不只是能否调通"""
    context_tokens: int
    structured_output: str        # high | moderate | limited
    tool_calling: str             # high | moderate | limited
    evidence_reconciliation: str  # high | moderate | limited

@dataclass
class AuthorityProfile:
    """该能力档位下允许的最高权限"""
    may_recommend: bool = True
    may_auto_approve: bool = False
    may_execute_external: bool = False

# 与 ModelConfig 一一对应：能力档位决定权限上限
CAPABILITY_AUTHORITY = {
    "primary": AuthorityProfile(
        may_recommend=True, may_auto_approve=True, may_execute_external=True
    ),
    "fallback": AuthorityProfile(
        may_recommend=True, may_auto_approve=False, may_execute_external=False
    ),
}

class AuthorityAwareSelectionEngine(ModelSelectionEngine):
    def select_with_authority(self):
        """返回 (provider, authority, degraded)，让上层无法忽略降级事实"""
        primary = self.policy.primary
        if self.breakers[primary.model_id].is_available():
            return self._create_provider(primary), CAPABILITY_AUTHORITY["primary"], False

        for config in self.policy.fallback_chain or []:
            if self.breakers[config.model_id].is_available():
                # 关键：换模型的同时换权限，二者不能分开决策
                return self._create_provider(config), CAPABILITY_AUTHORITY["fallback"], True

        raise RuntimeError("所有模型不可用")

from dataclasses import dataclass
from enum import Enum
from typing import Optional, List, Any

class ContentBlockType(Enum):
    TEXT = "text"
    TOOL_USE = "tool_use"
    THINKING = "thinking"

@dataclass
class TextBlock:
    """文本内容块"""
    type: str = "text"
    text: str = ""

@dataclass
class ToolUseBlock:
    """工具调用块"""
    type: str = "tool_use"
    id: str = ""
    name: str = ""
    input: dict = None  # JSON 输入参数

@dataclass
class ThinkingBlock:
    """思考过程块(Adaptive Thinking)"""
    type: str = "thinking"
    thinking: str = ""

ContentBlock = TextBlock | ToolUseBlock | ThinkingBlock

@dataclass
class ParsedMessage:
    """解析后的完整消息"""
    content_blocks: List[ContentBlock]
    stop_reason: str  # "end_turn", "tool_use", "max_tokens"
    tokens_used: int

    def text_content(self) -> str:
        """提取所有文本内容"""
        texts = [
            block.text for block in self.content_blocks
            if isinstance(block, TextBlock)
        ]
        return "".join(texts)

    def tool_calls(self) -> List[ToolUseBlock]:
        """提取所有工具调用"""
        return [
            block for block in self.content_blocks
            if isinstance(block, ToolUseBlock)
        ]

    def thinking_content(self) -> Optional[str]:
        """提取思考过程"""
        for block in self.content_blocks:
            if isinstance(block, ThinkingBlock):
                return block.thinking
        return None

import json
from typing import Dict, Any

class ResponseParser:
    """API 响应解析器"""

    @staticmethod
    def parse_response(raw_response: Dict[str, Any]) -> ParsedMessage:
        """解析完整的 API 响应"""
        content_blocks = []

        # 遍历响应的 content 数组
        for block in raw_response.get("content", []):
            block_type = block.get("type")

            if block_type == "text":
                content_blocks.append(TextBlock(
                    type="text",
                    text=block.get("text", "")
                ))

            elif block_type == "tool_use":
                # 工具调用:需要解析 JSON 输入
                input_data = block.get("input", {})
                if isinstance(input_data, str):
                    try:
                        input_data = json.loads(input_data)
                    except json.JSONDecodeError as e:
                        # 捕获 JSON 解析失败,记录错误并使用空对象
                        print(f"警告: JSON 解析失败: {e}, 使用空参数")
                        input_data = {}

                content_blocks.append(ToolUseBlock(
                    type="tool_use",
                    id=block.get("id", ""),
                    name=block.get("name", ""),
                    input=input_data
                ))

            elif block_type == "thinking":
                content_blocks.append(ThinkingBlock(
                    type="thinking",
                    thinking=block.get("thinking", "")
                ))

        return ParsedMessage(
            content_blocks=content_blocks,
            stop_reason=raw_response.get("stop_reason", "end_turn"),
            tokens_used=(
                raw_response.get("usage", {}).get("input_tokens", 0) +
                raw_response.get("usage", {}).get("output_tokens", 0)
            )
        )

import json
from enum import Enum
from typing import Dict, Any, Optional

class StreamEventType(Enum):
    MESSAGE_START = "message_start"
    CONTENT_BLOCK_START = "content_block_start"
    CONTENT_BLOCK_DELTA = "content_block_delta"
    CONTENT_BLOCK_STOP = "content_block_stop"
    MESSAGE_DELTA = "message_delta"
    MESSAGE_STOP = "message_stop"

class StreamingParser:
    """流式响应解析器"""

    def __init__(self):
        self.content_blocks = []
        self.current_block = None
        self.block_index = 0
        self.stop_reason = "end_turn"
        self.tokens_used = 0
        self.partial_json_buffer = ""  # 累积工具参数的 JSON 分片

    def process_event(self, event: Dict[str, Any]) -> Optional[ParsedMessage]:
        """处理单个流式事件"""
        event_type = event.get("type")

        if event_type == "message_start":
            # 消息开始,重置状态
            self.content_blocks = []
            self.current_block = None
            self.block_index = 0
            usage = event.get("message", {}).get("usage", {})
            self.tokens_used = (
                usage.get("input_tokens", 0)
                + usage.get("cache_creation_input_tokens", 0)
                + usage.get("cache_read_input_tokens", 0)
            )

        elif event_type == "content_block_start":
            # 新块开始
            self.block_index = event.get("index", 0)
            block_info = event.get("content_block", {})
            block_type = block_info.get("type")

            if block_type == "text":
                self.current_block = TextBlock(type="text", text="")
            elif block_type == "tool_use":
                self.current_block = ToolUseBlock(
                    type="tool_use",
                    id=block_info.get("id", ""),
                    name=block_info.get("name", ""),
                    input={}
                )
                self.partial_json_buffer = ""  # 参数 JSON 跨多个 delta 到达,需累积
            elif block_type == "thinking":
                self.current_block = ThinkingBlock(type="thinking", thinking="")

            # 确保数组足够长
            while len(self.content_blocks) <= self.block_index:
                self.content_blocks.append(None)

        elif event_type == "content_block_delta":
            # 块内容增量更新
            delta = event.get("delta", {})
            delta_type = delta.get("type")

            if delta_type == "text_delta":
                if isinstance(self.current_block, TextBlock):
                    self.current_block.text += delta.get("text", "")

            elif delta_type == "input_json_delta":
                # 工具调用的 JSON 增量:单个分片(如 '{"query": "Py')通常不是
                # 合法 JSON,只能累积,待 content_block_stop 时一次性解析
                if isinstance(self.current_block, ToolUseBlock):
                    self.partial_json_buffer += delta.get("partial_json", "")

            elif delta_type == "thinking_delta":
                if isinstance(self.current_block, ThinkingBlock):
                    self.current_block.thinking += delta.get("thinking", "")

        elif event_type == "content_block_stop":
            # 块完成:工具块在此刻才拥有完整的参数 JSON
            if isinstance(self.current_block, ToolUseBlock) and self.partial_json_buffer:
                try:
                    self.current_block.input = json.loads(self.partial_json_buffer)
                except json.JSONDecodeError:
                    self.current_block.input = {}
                self.partial_json_buffer = ""
            # 保存到数组
            if self.current_block is not None:
                self.content_blocks[self.block_index] = self.current_block
                self.current_block = None

        elif event_type == "message_delta":
            # 消息完成
            delta = event.get("delta", {})
            self.stop_reason = delta.get("stop_reason", "end_turn")
            usage = event.get("usage", {})
            self.tokens_used += usage.get("output_tokens", 0)

            # 返回完整的解析结果
            return ParsedMessage(
                content_blocks=[b for b in self.content_blocks if b],
                stop_reason=self.stop_reason,
                tokens_used=self.tokens_used
            )

        return None  # 消息未完成

import json

class StreamIterator:
    """流式迭代器,隐藏解析复杂性"""

    def __init__(self, api_stream):
        self.api_stream = api_stream
        self.parser = StreamingParser()

    def __iter__(self):
        for event_data in self.api_stream:
            try:
                event = json.loads(event_data)
            except json.JSONDecodeError as e:
                # JSON 解析失败,记录错误并跳过该事件
                print(f"警告: 流式事件 JSON 解析失败: {e}, 跳过该事件")
                continue
            result = self.parser.process_event(event)
            if result is not None:
                yield result

# 使用示例:api_stream 应是“逐条 JSON 字符串”的原始事件流(如 SSE 的 data: 行)。
# 注意:anthropic SDK 的 client.messages.stream() 产出的是已解析的事件对象,
# 直接遍历即可,不应再传入本迭代器做 json.loads。
for parsed_message in StreamIterator(raw_sse_data_lines):
    print(f"接收到消息: {parsed_message.text_content()}")
    for tool_call in parsed_message.tool_calls():
        print(f"  工具调用: {tool_call.name}({tool_call.input})")

from pydantic import BaseModel, ValidationError, field_validator
from typing import Optional

class SearchToolInput(BaseModel):
    """搜索工具的输入参数"""
    query: str
    max_results: int = 10
    language: str = "en"

    @field_validator("max_results")
    @classmethod
    def validate_max_results(cls, v):
        if v < 1 or v > 100:
            raise ValueError("max_results 必须在 1-100 之间")
        return v

class ToolCallValidator:
    """工具调用参数验证"""

    TOOL_SCHEMAS = {
        "search": SearchToolInput,
        "database_query": "DatabaseQueryInput",
        # ... 其他工具
    }

    @staticmethod
    def validate_tool_call(tool_call: ToolUseBlock) -> bool:
        """验证工具调用的合法性"""
        tool_name = tool_call.name
        if tool_name not in ToolCallValidator.TOOL_SCHEMAS:
            raise ValueError(f"未知的工具: {tool_name}")

        schema = ToolCallValidator.TOOL_SCHEMAS[tool_name]
        try:
            schema(**tool_call.input)
            return True
        except ValidationError as e:
            # 捕获 Pydantic 验证错误,提供详细错误信息
            error_msg = "; ".join([f"{err['loc']}: {err['msg']}" for err in e.errors()])
            raise ValueError(f"工具参数验证失败: {error_msg}")
        except (TypeError, AttributeError) as e:
            # 捕获参数类型错误
            raise ValueError(f"工具调用格式错误: {e}")

# 使用
tool_call = ToolUseBlock(
    id="call_123",
    name="search",
    input={"query": "Python async", "max_results": 10}
)
try:
    if ToolCallValidator.validate_tool_call(tool_call):
        print("工具调用有效")
except ValueError as e:
    print(f"验证失败: {e}")

class ParsingPipeline:
    """完整的解析管道"""

    def __init__(self):
        self.parser = ResponseParser()
        self.validator = ToolCallValidator()

    def parse_and_validate(
        self, raw_response: Dict[str, Any]
    ) -> ParsedMessage:
        """解析并验证响应"""
        # 步骤 1:解析
        parsed = self.parser.parse_response(raw_response)

        # 步骤 2:验证工具调用
        for tool_call in parsed.tool_calls():
            self.validator.validate_tool_call(tool_call)

        # 步骤 3:检查完整性
        if parsed.stop_reason == "max_tokens":
            raise ValueError("响应因 token 限制被截断")

        return parsed

from dataclasses import dataclass
from enum import Enum
from typing import Optional, List

class ValidationResult(Enum):
    PASS = "pass"
    FAIL = "fail"
    WARN = "warn"

@dataclass
class ValidationReport:
    result: ValidationResult
    errors: List[str]
    warnings: List[str]
    suggestion: Optional[str] = None

class FormatValidator:
    """基础格式验证"""

    @staticmethod
    def validate_tool_use_block(tool_use) -> ValidationReport:
        """验证工具调用块的结构完整性"""
        errors = []
        warnings = []

        # 检查必填字段
        if not tool_use.id:
            errors.append("tool_use.id 不能为空")
        if not tool_use.name:
            errors.append("tool_use.name 不能为空")
        if tool_use.input is None:
            errors.append("tool_use.input 不能为空")

        # 检查 ID 格式(Claude 的 tool_use.id 以 "toolu_" 开头)
        if tool_use.id and not tool_use.id.startswith("toolu_"):
            warnings.append(
                f"非标准 tool_use.id: {tool_use.id}"
            )

        # 检查名称格式(应为 snake_case)
        if tool_use.name and not FormatValidator._is_valid_name(tool_use.name):
            warnings.append(
                f"工具名格式非标准: {tool_use.name}"
            )

        if errors:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=errors,
                warnings=warnings,
                suggestion="检查 API 响应是否正确解析"
            )

        if warnings:
            return ValidationReport(
                result=ValidationResult.WARN,
                errors=[],
                warnings=warnings
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    @staticmethod
    def _is_valid_name(name: str) -> bool:
        """检查工具名是否符合 snake_case"""
        return name.islower() and name.replace("_", "").isalnum()

class ToolRegistry:
    """工具注册表与存在性检查"""

    def __init__(self):
        self.tools = {}

    def register(self, name: str, handler, schema):
        """注册工具"""
        self.tools[name] = {
            "handler": handler,
            "schema": schema,
            "enabled": True
        }

    def is_tool_available(self, name: str) -> bool:
        """检查工具是否可用"""
        if name not in self.tools:
            return False
        return self.tools[name]["enabled"]

    def get_tool_schema(self, name: str):
        """获取工具的参数 schema"""
        if name not in self.tools:
            return None
        return self.tools[name]["schema"]

class ExistenceValidator:
    """工具存在性验证"""

    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    def validate(self, tool_name: str) -> ValidationReport:
        """检查工具是否存在且可用"""
        errors = []
        suggestions = []

        if tool_name not in self.registry.tools:
            errors.append(f"工具 '{tool_name}' 未注册")
            # 生成相似工具建议
            similar = self._find_similar_tools(tool_name)
            if similar:
                suggestions.append(
                    f"您是想用 '{similar[0]}' 吗?"
                )

        elif not self.registry.is_tool_available(tool_name):
            errors.append(f"工具 '{tool_name}' 已禁用")

        if errors:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=errors,
                warnings=[],
                suggestion=suggestions[0] if suggestions else None
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    def _find_similar_tools(self, name: str) -> List[str]:
        """使用编辑距离找相似工具名"""
        from difflib import get_close_matches
        available = [n for n in self.registry.tools.keys()
                     if self.registry.is_tool_available(n)]
        return get_close_matches(name, available, n=1, cutoff=0.6)

from pydantic import BaseModel, ValidationError, field_validator
from typing import Optional, Any

class FileReadInput(BaseModel):
    """文件读取工具的输入"""
    file_path: str
    encoding: str = "utf-8"
    max_lines: Optional[int] = None

    @field_validator("file_path")
    @classmethod
    def validate_file_path(cls, v):
        if not isinstance(v, str) or len(v) == 0:
            raise ValueError("file_path 必须是非空字符串")
        return v

    @field_validator("max_lines")
    @classmethod
    def validate_max_lines(cls, v):
        if v is not None and (v < 1 or v > 100000):
            raise ValueError("max_lines 必须在 1-100000 之间")
        return v

class ParameterValidator:
    """参数验证器"""

    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    def validate(self, tool_name: str, parameters: dict) -> ValidationReport:
        """验证工具参数"""
        schema = self.registry.get_tool_schema(tool_name)
        if schema is None:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"无法找到工具 {tool_name} 的 schema"],
                warnings=[]
            )

        errors = []
        warnings = []

        try:
            # 使用 Pydantic 验证参数
            validated = schema(**parameters)
            return ValidationReport(
                result=ValidationResult.PASS,
                errors=[],
                warnings=[]
            )
        except ValidationError as e:
            for error in e.errors():
                field = ".".join(str(x) for x in error["loc"])
                msg = error["msg"]
                errors.append(f"参数 {field}: {msg}")

        if errors:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=errors,
                warnings=warnings,
                suggestion="请检查工具的参数定义"
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=warnings
        )

class BusinessLogicValidator:
    """业务逻辑验证"""

    def __init__(self):
        self.config = {
            "max_api_calls_per_minute": 100,
            "max_file_size_mb": 10,
            "allowed_domains": ["example.com", "api.service.com"],
        }

    def validate_api_call_frequency(self, tool_name: str,
                                    call_history: List[float]) -> ValidationReport:
        """检查 API 调用频率是否超过限制"""
        import time
        now = time.time()
        # 计算最近 1 分钟的调用数
        recent_calls = [t for t in call_history if now - t < 60]

        limit = self.config["max_api_calls_per_minute"]
        if len(recent_calls) >= limit:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"工具 {tool_name} 在 1 分钟内调用已达 {limit} 次限制"],
                warnings=[],
                suggestion="请稍后再试"
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    def validate_file_access(self, file_path: str) -> ValidationReport:
        """检查文件访问权限和大小"""
        import os
        errors = []

        # 检查文件存在性
        if not os.path.exists(file_path):
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"文件不存在: {file_path}"],
                warnings=[]
            )

        # 检查文件大小
        file_size_mb = os.path.getsize(file_path) / (1024 * 1024)
        max_size = self.config["max_file_size_mb"]
        if file_size_mb > max_size:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"文件大小 {file_size_mb:.2f}MB 超过限制 {max_size}MB"],
                warnings=[]
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    def validate_http_request(self, url: str) -> ValidationReport:
        """检查 HTTP 请求的目标合法性"""
        from urllib.parse import urlparse
        parsed = urlparse(url)
        domain = parsed.netloc

        if domain not in self.config["allowed_domains"]:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"不允许访问域名: {domain}"],
                warnings=[],
                suggestion=f"允许的域名: {self.config['allowed_domains']}"
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

from enum import Enum

class Permission(Enum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    EXECUTE = "execute"

class SecurityValidator:
    """安全与权限检查"""

    def __init__(self, user_permissions: set):
        self.user_permissions = user_permissions

    def validate_tool_permission(self, tool_name: str) -> ValidationReport:
        """检查用户是否有权调用该工具"""
        # 定义工具所需权限
        tool_permissions = {
            "read_file": {Permission.READ},
            "write_file": {Permission.WRITE},
            "delete_file": {Permission.DELETE},
            "execute_command": {Permission.EXECUTE},
        }

        if tool_name not in tool_permissions:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"未知工具: {tool_name}"],
                warnings=[]
            )

        required = tool_permissions[tool_name]
        if not required.issubset(self.user_permissions):
            missing = required - self.user_permissions
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"缺少权限: {', '.join(p.value for p in missing)}"],
                warnings=[]
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    def validate_parameter_injection(self, parameters: dict) -> ValidationReport:
        """检查参数是否包含注入攻击"""
        dangerous_patterns = [
            "rm -rf",
            "DROP TABLE",
            "<script>",
            "${jndi:",
        ]

        for key, value in parameters.items():
            if isinstance(value, str):
                for pattern in dangerous_patterns:
                    if pattern in value:
                        return ValidationReport(
                            result=ValidationResult.FAIL,
                            errors=[f"参数 {key} 包含危险模式: {pattern}"],
                            warnings=[],
                            suggestion="检查参数是否被恶意注入"
                        )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

class QualityGate:
    """完整的质量门控引擎"""

    def __init__(self, registry: ToolRegistry, user_permissions: set):
        self.format_validator = FormatValidator()
        self.existence_validator = ExistenceValidator(registry)
        self.parameter_validator = ParameterValidator(registry)
        self.business_validator = BusinessLogicValidator()
        self.security_validator = SecurityValidator(user_permissions)

    def validate_tool_call(self, tool_call, call_history: List[float] = None):
        """执行完整的质量门控"""
        results = []

        # 门控 1: 格式检查
        r1 = self.format_validator.validate_tool_use_block(tool_call)
        results.append(("格式检查", r1))
        if r1.result == ValidationResult.FAIL:
            return self._report_failure(results)

        # 门控 2: 工具存在性
        r2 = self.existence_validator.validate(tool_call.name)
        results.append(("工具存在性", r2))
        if r2.result == ValidationResult.FAIL:
            return self._report_failure(results)

        # 门控 3: 参数验证
        r3 = self.parameter_validator.validate(
            tool_call.name, tool_call.input
        )
        results.append(("参数验证", r3))
        if r3.result == ValidationResult.FAIL:
            return self._report_failure(results)

        # 门控 4: 业务逻辑(对文件类参数执行业务规则)
        file_path = (tool_call.input or {}).get("file_path")
        if file_path:
            r4 = self.business_validator.validate_file_access(file_path)
            results.append(("业务逻辑", r4))
            if r4.result == ValidationResult.FAIL:
                return self._report_failure(results)

        # 门控 5: 权限检查
        r5 = self.security_validator.validate_tool_permission(tool_call.name)
        results.append(("权限检查", r5))
        if r5.result == ValidationResult.FAIL:
            return self._report_failure(results)

        # 门控 6: 参数注入检测
        r6 = self.security_validator.validate_parameter_injection(
            tool_call.input or {}
        )
        results.append(("注入检测", r6))
        if r6.result == ValidationResult.FAIL:
            return self._report_failure(results)

        return {
            "passed": True,
            "details": results
        }

    def _report_failure(self, results):
        return {
            "passed": False,
            "details": results
        }

class PGEAgent:
    """规划-生成-评估三角色分离"""

    def __init__(self, planner_model, generator_model, evaluator_model):
        self.planner = planner_model       # 高推理预算
        self.generator = generator_model   # 优化吞吐量
        self.evaluator = evaluator_model   # 独立验证

    async def execute(self, user_request: str) -> ExecutionResult:
        """执行PGE流程"""

        # 第一步:规划
        print("[Phase 1] Planning...")
        plan = await self.planner.analyze_and_plan(user_request)
        print(f"Plan: {plan.steps}")

        # 第二步:生成
        print("[Phase 2] Generating...")
        output = await self.generator.execute_plan(plan)

        # 第三步:评估 - 关键:使用独立的评估者
        print("[Phase 3] Evaluating...")
        evaluation = await self.evaluate_output(
            output=output,
            plan=plan,
            evaluator=self.evaluator  # 独立的模型
        )

        if evaluation.passed:
            return output

        # 失败时的重试逻辑
        if evaluation.retry_generator:
            # 让生成者修正(基于评估反馈)
            output_v2 = await self.generator.refine(
                original_output=output,
                feedback=evaluation.feedback,
                plan=plan
            )
            evaluation_v2 = await self.evaluate_output(
                output=output_v2,
                plan=plan,
                evaluator=self.evaluator
            )
            if evaluation_v2.passed:
                return output_v2

        if evaluation.retry_planner:
            # 让规划者重新规划(当多次生成都失败时)
            plan_v2 = await self.planner.replan(
                user_request=user_request,
                failure_reason=evaluation.failure_analysis
            )
            # 重新执行生成
            return await self.execute_with_plan(plan_v2)

        raise ExecutionError(f"Failed after retries: {evaluation.feedback}")

    async def execute_with_plan(self, plan) -> ExecutionResult:
        """跳过规划阶段,按给定计划执行生成与评估(重新规划后的入口)"""
        output = await self.generator.execute_plan(plan)
        evaluation = await self.evaluate_output(
            output=output, plan=plan, evaluator=self.evaluator
        )
        if evaluation.passed:
            return output
        raise ExecutionError(f"Replanned execution failed: {evaluation.feedback}")

    async def evaluate_output(
        self,
        output: str,
        plan: Plan,
        evaluator: Model
    ) -> EvaluationResult:
        """独立评估输出"""

        # 构造评估提示 - 这是关键:评估者不应该看到生成过程
        evaluation_prompt = f"""
Please evaluate the following output against the original requirements.

Original Requirements:
{plan.requirements}

Verification Criteria:
{plan.verification_criteria}

Output to Evaluate:
{output}

Answer:
1. Does the output meet ALL requirements? (yes/no)
2. Are there any errors or issues?
3. If failed, what specific improvements are needed?
4. Confidence level: (high/medium/low)
"""

        evaluation_text = await evaluator.infer(evaluation_prompt)
        return self._parse_evaluation(evaluation_text, plan)

    def _parse_evaluation(self, eval_text: str, plan: Plan) -> EvaluationResult:
        """解析评估结果"""
        # 简化实现:检查评估是否包含肯定
        passed = "yes" in eval_text.lower()

        # 判断是重试生成还是重新规划
        retry_generator = "small" in eval_text.lower() or "adjust" in eval_text.lower()
        retry_planner = "fundamental" in eval_text.lower() or "rethink" in eval_text.lower()

        return EvaluationResult(
            passed=passed,
            retry_generator=retry_generator,
            retry_planner=retry_planner,
            feedback=eval_text
        )

# 配置示例
# 注意：Opus 4.7 起已移除 temperature/top_p/top_k，传入会返回 400；
# 推理深度改用 adaptive thinking + effort 控制。Sonnet 4.6 仍接受采样参数。
PLANNER_CONFIG = {
    "model": "claude-opus-4-7",  # 最强模型
    "adaptive_thinking": True,   # 高推理预算
    "effort": "high",            # 取代 temperature 的确定性/深度旋钮
}

GENERATOR_CONFIG = {
    "model": "claude-sonnet-4-6", # 平衡成本/性能
    "adaptive_thinking": False,   # 轻量级,专注执行
    "temperature": 0.1,           # 4.6 仍支持采样参数
}

EVALUATOR_CONFIG = {
    "model": "claude-opus-4-7",  # 同样强大的模型
    # 但使用完全不同的上下文和提示
    "adaptive_thinking": True,
    "effort": "high",            # 同上：4.7 不接受 temperature
}