class PlannerExecutorValidator:
    """PEV 三元组模式的核心实现"""

    def __init__(self, planner_llm, validator_llm, tool_registry):
        self.planner = planner_llm
        self.validator = validator_llm
        self.tools = tool_registry

    async def run(self, user_task: str, max_retries: int = 3) -> str:
        # Step 1: Planner 分解任务
        plan = await self.planner.plan(user_task)
        # plan = [
        #   {"id": "search", "depends_on": [], "tool": "rag_search", "args": {...}},
        #   {"id": "code_gen", "depends_on": ["search"], "tool": "code_writer", "args": {...}},
        #   {"id": "test", "depends_on": ["code_gen"], "tool": "test_runner", "args": {...}},
        #   {"id": "report", "depends_on": ["test", "search"], "tool": "report_gen", "args": {...}},
        # ]

        results = {}
        for retry in range(max_retries):
            # Step 2: 按 DAG 拓扑序执行
            for step in self._topological_sort(plan):
                if step["id"] in results:
                    continue  # 已执行过（重试时跳过）
                result = await self.tools.call(step["tool"], step["args"])
                results[step["id"]] = result

            # Step 3: Validator 检查
            validation = await self.validator.validate(results)
            if validation["passed"]:
                break
            else:
                # 反馈给 Planner 调整
                plan = await self.planner.replan(
                    user_task, results, validation["failures"]
                )
                results = {}  # 重设结果，从头再来

        if not validation["passed"]:
            raise MaxRetriesExceeded(f"重试 {max_retries} 次后仍未通过验证")

        # Step 4: 合并结果
        return await self._merge_results(results, plan)

# Validator 的已知偏差（基于 500 次实测）
VALIDATOR_BIASES = {
    "false_positive": 12.3%,   # 误判：正确结果被标记为失败
    "false_negative": 5.7%,    # 漏判：错误结果被标记为通过
    "length_bias": True,       # 更长的输出更容易被判为「好」
    "confidence_calibration": 0.31,  # 校准度低（自信但不准确）
}

class RoutingOrchestrator:
    """路由分发模式——智能路由器"""

    def __init__(self, llm, agent_registry: dict[str, 'SubAgent']):
        self.llm = llm          # 用于路由决策的轻量级 LLM
        self.agents = agent_registry  # {"rag": RAGAgent, "code": CodeAgent, "sql": SQLAgent}

    async def route(self, user_query: str) -> str:
        # Step 1: 意图识别（用轻量模型，不要用旗舰）
        intent = await self._classify_intent(user_query,
            categories=["检索", "编码", "数据库查询", "数据分析", "报告生成"]
        )
        # 实测：使用 GPT-4o-mini 级别模型即可达到 94% 准确率
        # 用旗舰模型做路由决策是资源浪费

        # Step 2: 路由
        agent = self.agents.get(intent["category"])
        if not agent:
            # 未知类型 → 回退到通用 Agent
            agent = self.agents["general"]

        # Step 3: 执行
        result = await agent.run(user_query, intent.get("params", {}))

        return result

    async def _classify_intent(self, query: str, categories: list[str]) -> dict:
        """路由意图分类——关键优化点"""
        prompt = f"""Classify the following user query into exactly one category:
Categories: {', '.join(categories)}
Query: {query}

Return JSON: {{"category": "<one of the categories>", "params": {{...}}}}
"""
        response = await self.llm.complete(prompt)
        return json.loads(response)

class CachedRouter:
    """带缓存的路由器"""

    def __init__(self, llm, agents: dict, cache_ttl: int = 3600):
        self.llm = llm
        self.agents = agents
        self.cache = {}  # embedding_hash -> (category, expires_at)

    async def route(self, query: str) -> str:
        # Step 1: 快速哈希匹配
        query_hash = self._semantic_hash(query)
        if query_hash in self.cache:
            cached = self.cache[query_hash]
            if cached["expires_at"] > time.time():
                return await self.agents[cached["category"]].run(query, {})

        # Step 2: LLM 路由
        category = await self._llm_route(query)

        # Step 3: 缓存结果
        self.cache[query_hash] = {
            "category": category,
            "expires_at": time.time() + self.cache_ttl,
        }
        return await self.agents[category].run(query, {})

    def _semantic_hash(self, query: str) -> str:
        """用 embedding 做语义哈希"""
        # 用 SentenceTransformer / OpenAI Embeddings
        emb = self.embedding_model.encode(query)
        # 量化为 8-bit 哈希（精度 vs 空间权衡）
        return hashlib.md5(emb.tobytes()).hexdigest()[:16]

class SagaOrchestrator:
    """基于 Saga 模式的路由编排"""

    def __init__(self):
        self.saga_log = []  # 事务日志

    async def execute_with_saga(self, steps: list[SagaStep]) -> dict:
        results = {}
        completed = []

        for step in steps:
            try:
                result = await step.action()
                results[step.name] = result
                completed.append(step)
                self.saga_log.append({"step": step.name, "status": "done"})
            except Exception as e:
                # 补偿：逆序执行已完成步骤的补偿操作
                for done_step in reversed(completed):
                    await done_step.compensate()
                    self.saga_log.append({
                        "step": done_step.name, "status": "compensated"
                    })
                raise SagaFailed(f"Step {step.name} failed: {e}")

        return results

class DebateConsensus:
    """辩论式共识——多 Agent 讨论并达成一致"""

    CONSENSUS_STRATEGIES = {
        "majority": lambda votes: Counter(votes).most_common(1)[0][0],
        "weighted": lambda votes, weights: max(
            set(votes), key=lambda v: sum(w for vi, w in zip(votes, weights) if vi == v)
        ),
        "debate": "multi_round",  # 需要多轮交互
    }

    async def debate(self, question: str,
                     agents: list[SubAgent],
                     max_rounds: int = 5,
                     strategy: str = "debate") -> str:
        # Round 0: 初始答案
        statements = [
            await agent.answer(question)
            for agent in agents
        ]

        for round_num in range(1, max_rounds):
            # 每轮：每个 Agent 可以看到其他 Agent 的答案并修正
            new_statements = []
            for i, agent in enumerate(agents):
                others = [s for j, s in enumerate(statements) if j != i]
                revised = await agent.revise(question, statements[i], others)
                new_statements.append(revised)

            # 检查是否达成共识
            if self._has_converged(new_statements, round_num):
                return self._merge_consensus(new_statements, strategy)

            statements = new_statements

        # 未达成共识 → 退化为加权投票
        return self._merge_consensus(statements, "weighted",
                                     [a.confidence for a in agents])

    def _has_converged(self, statements: list[str], round_num: int) -> bool:
        """检查是否收敛——基于语义相似度"""
        if round_num >= 3:  # 至少 3 轮后再检查收敛
            similarities = []
            for i in range(len(statements)):
                for j in range(i + 1, len(statements)):
                    sim = self._semantic_similarity(statements[i], statements[j])
                    similarities.append(sim)
            return sum(similarities) / len(similarities) > 0.85
        return False

class GossipDebateNode:
    """基于 Gossip 协议的 Debate 节点"""

    def __init__(self, agent_id: str, llm, peers: list[str]):
        self.id = agent_id
        self.llm = llm
        self.peers = peers
        self.viewpoint = None  # 当前 Agent 的观点
        self.viewpoints_heard = {}  # {peer_id: viewpoint, round}

    async def gossip_round(self, question: str) -> bool:
        """一轮 Gossip 通信"""
        # 1. 随机选择 2-3 个邻居
        targets = random.sample(self.peers, min(3, len(self.peers)))

        # 2. 发送我的观点
        for target in targets:
            await self._send(target, {"from": self.id, "viewpoint": self.viewpoint})

        # 3. 接收邻居的观点
        received = await self._receive_all(timeout=2.0)

        # 4. 综合所有观点，更新我的观点
        all_viewpoints = [self.viewpoint] + list(received.values())
        self.viewpoint = await self.llm.synthesize(question, all_viewpoints)

        # 5. 检查是否收敛
        return self._detect_convergence(all_viewpoints)

# MCP 协议将 Sub-Agent 暴露为 Tool 接口
class MCPAgentTool:
    """通过 MCP 协议暴露 Sub-Agent 为 Tool"""

    def __init__(self, name: str, mcp_server_url: str):
        self.name = name
        self.server_url = mcp_server_url

    @property
    def mcp_tool_definition(self) -> dict:
        """MCP Tool 描述——Orchestrator 通过这个了解 Sub-Agent 的能力"""
        return {
            "name": self.name,
            "description": f"Sub-Agent: {self.name} - 通过 MCP 协议接入",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "task": {"type": "string", "description": "任务描述"},
                    "context": {"type": "string", "description": "上下文/前一步结果"},
                },
                "required": ["task"],
            },
        }

    async def execute(self, task: str, context: str = "") -> str:
        """通过 MCP 协议远程调用 Sub-Agent"""
        async with MCPClient(self.server_url) as client:
            result = await client.call_tool(self.name, {
                "task": task,
                "context": context,
            })
            return result.content[0].text

# A2A 通信：Agent → Agent 直接对话
class A2AMessage:
    """A2A 协议消息体"""

    def __init__(self, agent_id: str, task_id: str, msg_type: str, payload: dict):
        self.agent_id = agent_id
        self.task_id = task_id
        self.type = msg_type  # "request" | "progress" | "response" | "query"
        self.payload = payload

class A2ADebateChannel:
    """基于 A2A 协议的 Debate 通信通道"""

    async def broadcast(self, message: A2AMessage, recipients: list[str]):
        """向所有参与者广播消息"""
        responses = []
        for recipient in recipients:
            response = await self._send_a2a(recipient, message)
            responses.append(response)
        return responses

    async def _send_a2a(self, target_agent: str, msg: A2AMessage) -> A2AMessage:
        """发送 A2A 消息并等待响应"""
        # 通过 A2A Agent Card 发现目标 Agent 的端点
        agent_card = await self._resolve_agent_card(target_agent)
        # 发送任务（带 stream 选项）
        async with A2AClient(agent_card.url) as client:
            result = await client.send_task(msg.to_dict(), stream=True)
            return A2AMessage.from_dict(result)

class ResultMerger:
    """结果合并器"""

    async def merge(self, results: dict[str, str], strategy: str) -> str:
        if strategy == "sequential":
            return self._sequential_merge(results)
        elif strategy == "intelligent":
            return await self._intelligent_merge(results)
        elif strategy == "voting":
            return await self._voting_merge(results)

    def _sequential_merge(self, results: dict[str, str]) -> str:
        """顺序拼接——按 Plan 中定义的 DAG 拓扑序输出"""
        return "\n\n---\n\n".join(
            f"## {step_name}\n{content}"
            for step_name, content in results.items()
        )

    async def _intelligent_merge(self, results: dict[str, str]) -> str:
        """智能合并——让 LLM 把多个结果融合成一篇流畅的输出"""
        merge_prompt = "你是一个结果合并器。合并以下多个 Sub-Agent 的输出为一篇连贯的报告：\n\n"
        for name, content in results.items():
            merge_prompt += f"=== {name} 的输出 ===\n{content}\n\n"
        merge_prompt += "请合并输出（保持事实准确、删除冗余、统一风格）："
        return await self.llm.complete(merge_prompt)

    async def _voting_merge(self, results: dict[str, str]) -> str:
        """投票融合——多个 Agent 对同一问题输出不同答案时，投票选最优"""
        # 先做语义去重（相同答案的不同表述）
        unique_results = self._deduplicate(results.values())
        if len(unique_results) == 1:
            return unique_results[0]

        # 投票
        vote_prompt = "以下是多个 Agent 对同一问题的回答。请选出最准确的一个：\n\n"
        for i, content in enumerate(unique_results):
            vote_prompt += f"--- 方案{i+1} ---\n{content}\n\n"
        vote_prompt += "输出格式：方案编号"
        winner = await self.llm.complete(vote_prompt)
        return unique_results[int(winner.strip()) - 1]

class HierarchicalArbitrator:
    """分层仲裁——仲裁 Agent 拥有最终决定权"""

    def __init__(self, arbiter_llm, arbitrator_instructions: str):
        self.arbiter = arbiter_llm
        self.instructions = arbitrator_instructions

    async def arbitrate(self, question: str,
                        answers: list[dict]) -> dict:
        """
        answers: [{"agent": "RAG", "answer": "...", "confidence": 0.92, "evidence": [...]}, ...]
        """
        # 检查是否有明显分歧
        if self._all_agree(answers):
            return {"winner": answers[0], "confidence": "high"}

        # 构建仲裁上下文
        context = f"问题: {question}\n\n各 Agent 的回答:\n"
        for a in answers:
            context += f"\nAgent [{a['agent']}] (置信度: {a['confidence']}):\n"
            context += f"答案: {a['answer']}\n"
            context += f"证据: {json.dumps(a.get('evidence', []), ensure_ascii=False)}\n"

        # 仲裁 Agent 做决策
        decision = await self.arbiter.complete(
            f"{self.instructions}\n\n{context}\n\n请决定哪个答案最可靠，并解释原因。"
        )
        # 解析仲裁结果
        ...

    def _all_agree(self, answers: list[dict]) -> bool:
        """检查所有 Agent 是否本质上一致"""
        texts = [a["answer"] for a in answers]
        return all(self._semantic_similarity(texts[0], t) > 0.9 for t in texts)

class HybridOrchestrator:
    """混合编排——路由分发 + PEV + Debate 三级组合"""

    def __init__(self):
        # Level 1: 路由分发（顶层）
        self.router = RoutingOrchestrator(...)
        # Level 2: PEV（中层）
        self.pev = PlannerExecutorValidator(...)
        # Level 3: Debate（底层，关键决策用）
        self.debate = DebateConsensus(...)

    async def run(self, task: str) -> str:
        # 1. 路由分发：判断任务主类型
        intent = await self.router.classify(task)

        if intent["category"] == "research":
            # 2. 研究型任务 → PEV（检索 + 分析 + 验证）
            return await self.pev.run(task)

        elif intent["category"] == "decision":
            # 3. 决策型任务 → Debate（多 Agent 讨论后决定）
            return await self.debate.debate(
                task, agents=self.debate_agents, max_rounds=3
            )

        elif intent["category"] == "coding":
            # 4. 编码型任务 → 先 PEV 再局部 Debate
            plan = await self.pev.plan(task)
            if any("architectural_decision" in s for s in plan):
                # 架构决策需要 Debate
                decision = await self.debate.debate(
                    "选择最佳实现方案",
                    agents=self.arch_agents, max_rounds=2
                )
                plan = await self.pev.replan_with_decision(task, decision)
            return await self.pev.execute_plan(plan)

        # 默认：通用 PEV
        return await self.pev.run(task)