多Agent系统与实战¶

"""
层级模式：Manager Agent分配任务给Worker Agent
"""

from openai import OpenAI
import json
from typing import Any
from dataclasses import dataclass

client = OpenAI()

@dataclass
class AgentProfile:
    """Agent角色定义"""
    name: str
    role: str
    capabilities: list[str]
    system_prompt: str

class HierarchicalSystem:
    """层级式多Agent系统"""

    def __init__(self):
        self.manager = AgentProfile(
            name="项目经理",
            role="manager",
            capabilities=["任务分解", "任务分配", "结果整合"],
            system_prompt="""你是一个项目经理Agent。你的职责是：
1. 分析用户的复杂任务
2. 将任务分解为子任务
3. 分配给合适的团队成员
4. 整合各成员的结果

你的团队成员：
- 研究员：擅长信息搜索和分析
- 工程师：擅长编写代码和技术方案
- 写作者：擅长报告和文档撰写

请以JSON格式返回任务分配计划：
{"subtasks": [{"assignee": "研究员/工程师/写作者", "task": "具体任务", "priority": 1-3}]}"""
        )

        self.workers = {
            "研究员": AgentProfile(
                name="研究员",
                role="researcher",
                capabilities=["搜索", "分析", "总结"],
                system_prompt="你是一个专业研究员。请深入研究指定主题，提供详细的分析和数据支持。",
            ),
            "工程师": AgentProfile(
                name="工程师",
                role="engineer",
                capabilities=["编码", "架构设计", "技术选型"],
                system_prompt="你是一个高级软件工程师。请提供高质量的技术方案和代码实现。",
            ),
            "写作者": AgentProfile(
                name="写作者",
                role="writer",
                capabilities=["写作", "编辑", "排版"],
                system_prompt="你是一个专业文档写作者。请将信息整理成清晰、专业的文档。",
            ),
        }

    def _call_llm(self, system_prompt: str, user_message: str) -> str:
        """调用LLM"""
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_message},
            ],
            temperature=0,
        )
        return response.choices[0].message.content

    def decompose_task(self, task: str) -> list[dict]:
        """Manager分解任务"""
        print(f"📋 [Manager] 开始分解任务: {task}")

        response = self._call_llm(
            self.manager.system_prompt,
            f"请分解以下任务: {task}"
        )

        # 尝试解析JSON
        try:
            # 提取JSON部分
            json_str = response
            if "```json" in response:
                json_str = response.split("```json")[1].split("```")[0]
            elif "```" in response:
                json_str = response.split("```")[1].split("```")[0]

            plan = json.loads(json_str)
            subtasks = plan.get("subtasks", [])
        except (json.JSONDecodeError, IndexError):
            # 回退方案
            subtasks = [
                {"assignee": "研究员", "task": f"研究关于'{task}'的信息", "priority": 1},
                {"assignee": "工程师", "task": f"为'{task}'提供技术方案", "priority": 2},
                {"assignee": "写作者", "task": f"撰写'{task}'的最终报告", "priority": 3},
            ]

        print(f"  ✅ 分解为 {len(subtasks)} 个子任务")
        for st in subtasks:
            print(f"    → [{st['assignee']}] {st['task']}")

        return subtasks

    def execute_subtask(self, subtask: dict) -> str:
        """Worker执行子任务"""
        assignee = subtask["assignee"]
        task = subtask["task"]

        worker = self.workers.get(assignee)
        if not worker:
            return f"❌ 未找到Agent: {assignee}"

        print(f"\n🔧 [{worker.name}] 执行任务: {task}")
        result = self._call_llm(worker.system_prompt, task)
        print(f"  ✅ [{worker.name}] 完成")

        return result

    def integrate_results(self, task: str, results: list[dict]) -> str:
        """Manager整合结果"""
        print(f"\n📊 [Manager] 整合所有结果...")

        results_text = "\n\n".join([
            f"### {r['assignee']}的工作结果:\n{r['result']}"
            for r in results
        ])

        integration_prompt = f"""作为项目经理，请整合以下团队成员的工作结果，生成一份完整的最终报告。

原始任务: {task}

团队成员的工作结果:
{results_text}

请生成一份结构化的最终报告，包含：摘要、详细内容、结论和建议。"""

        final_report = self._call_llm(self.manager.system_prompt, integration_prompt)
        return final_report

    def run(self, task: str) -> str:
        """运行完整的多Agent工作流"""
        print("=" * 60)
        print(f"🚀 启动多Agent系统")
        print(f"📌 任务: {task}")
        print("=" * 60)

        # 1. 任务分解
        subtasks = self.decompose_task(task)

        # 2. 按优先级排序并执行
        subtasks.sort(key=lambda x: x.get("priority", 99))  # sort原地排序，lambda+dict.get提供默认优先级99（无优先级的排最后）

        results = []
        for subtask in subtasks:
            result = self.execute_subtask(subtask)
            results.append({
                "assignee": subtask["assignee"],
                "task": subtask["task"],
                "result": result,
            })

        # 3. 整合结果
        final_report = self.integrate_results(task, results)

        print("\n" + "=" * 60)
        print("📋 最终报告:")
        print("=" * 60)

        return final_report

# 使用示例
system = HierarchicalSystem()
report = system.run("分析AI Agent技术在金融领域的应用前景，并给出具体的落地建议")
print(report)

"""
平等模式：Agent之间直接通信协作
适用于头脑风暴、辩论、多角度分析等场景
"""

from openai import OpenAI
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class PeerAgent:
    """平等协作Agent"""
    name: str
    perspective: str  # 分析视角
    system_prompt: str
    memory: list[dict] = field(default_factory=list)  # dataclass可变默认值必须用field(default_factory=)，避免共享引用

    def respond(self, discussion_context: str) -> str:
        """参与讨论并发表观点"""
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": discussion_context},
        ]

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            temperature=0.7,  # 稍高温度以增加多样性
            max_tokens=500,
        )

        return response.choices[0].message.content

class PeerDiscussion:
    """平等讨论系统"""

    def __init__(self, topic: str, rounds: int = 3):
        self.topic = topic
        self.rounds = rounds
        self.agents: list[PeerAgent] = []
        self.discussion_log: list[dict] = []

    def add_agent(self, agent: PeerAgent):
        self.agents.append(agent)

    def run(self) -> str:
        """运行多轮讨论"""
        print(f"📢 讨论主题: {self.topic}")
        print(f"👥 参与者: {', '.join(a.name for a in self.agents)}")
        print(f"🔄 讨论轮数: {self.rounds}")
        print("=" * 60)

        for round_num in range(1, self.rounds + 1):
            print(f"\n--- 第 {round_num} 轮 ---")

            for agent in self.agents:
                # 构建讨论上下文
                context = f"讨论主题: {self.topic}\n\n"
                if self.discussion_log:
                    context += "之前的讨论:\n"
                    for entry in self.discussion_log[-6:]:  # 最近6条
                        context += f"[{entry['agent']}]: {entry['content']}\n\n"

                if round_num == 1:
                    context += f"请从你的专业角度（{agent.perspective}）发表你对该主题的看法。"
                else:
                    context += f"请基于其他人的观点，从你的角度（{agent.perspective}）进一步补充或反驳。"

                # Agent发言
                response = agent.respond(context)

                self.discussion_log.append({
                    "round": round_num,
                    "agent": agent.name,
                    "content": response,
                })

                print(f"\n🗣️ [{agent.name}]:")
                print(f"  {response[:200]}...")

        # 生成讨论总结
        return self._summarize()

    def _summarize(self) -> str:
        """总结讨论结果"""
        all_discussion = "\n\n".join([
            f"[{entry['agent']} - 第{entry['round']}轮]: {entry['content']}"
            for entry in self.discussion_log
        ])

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个公正的讨论总结者。请综合各方观点，生成一份平衡、全面的总结报告。"},
                {"role": "user", "content": f"讨论主题: {self.topic}\n\n讨论内容:\n{all_discussion}\n\n请生成讨论总结，包含：共识点、分歧点、关键洞察和建议。"},
            ],
        )
        return response.choices[0].message.content

# 使用示例: 多角度分析
discussion = PeerDiscussion("AI Agent是否会取代传统软件开发？", rounds=2)

discussion.add_agent(PeerAgent(
    name="技术乐观派",
    perspective="技术进步和创新",
    system_prompt="你是一个技术乐观主义者，相信AI Agent将极大提升软件开发效率和质量。请用具体案例和数据支持你的观点。",
))

discussion.add_agent(PeerAgent(
    name="务实工程师",
    perspective="工程实践",
    system_prompt="你是一位务实的软件工程师，从实际工程经验出发分析AI Agent的能力边界和局限性。",
))

discussion.add_agent(PeerAgent(
    name="产业分析师",
    perspective="商业和就业影响",
    system_prompt="你是一位产业分析师，关注AI Agent对软件行业商业模式和就业市场的影响。",
))

summary = discussion.run()
print(f"\n{'='*60}\n📝 讨论总结:\n{summary}")

"""
市场模式：基于任务池的动态任务分配
Agent根据能力和负载竞争任务
"""

from dataclasses import dataclass, field
import random

@dataclass
class Task:
    """任务定义"""
    id: str
    description: str
    required_skills: list[str]
    complexity: int  # 1-5
    reward: float    # 奖励积分
    status: str = "pending"  # pending/assigned/completed
    assigned_to: str | None = None
    result: str | None = None

@dataclass
class MarketAgent:
    """市场中的Agent"""
    name: str
    skills: list[str]
    capacity: int = 3          # 最大并行任务数
    current_tasks: int = 0
    total_reward: float = 0.0

    def can_handle(self, task: Task) -> bool:
        """判断是否能处理该任务"""
        # 有能力且有空闲
        has_skills = all(s in self.skills for s in task.required_skills)  # all():所有需求技能都在自身技能列表中才返回True
        has_capacity = self.current_tasks < self.capacity
        return has_skills and has_capacity

    def bid(self, task: Task) -> float:
        """对任务出价（置信度）"""
        if not self.can_handle(task):
            return 0.0

        # 技能匹配度
        skill_match = len(set(task.required_skills) & set(self.skills)) / len(task.required_skills)
        # 空闲率
        availability = 1 - (self.current_tasks / self.capacity)

        return skill_match * 0.7 + availability * 0.3

    def execute(self, task: Task) -> str:
        """执行任务（模拟）"""
        self.current_tasks += 1
        result = f"[{self.name}] 已完成任务: {task.description}"
        self.current_tasks -= 1
        self.total_reward += task.reward
        return result

class TaskMarket:
    """任务市场"""

    def __init__(self):
        self.agents: list[MarketAgent] = []
        self.task_queue: list[Task] = []
        self.completed_tasks: list[Task] = []

    def register_agent(self, agent: MarketAgent):
        """注册Agent"""
        self.agents.append(agent)
        print(f"✅ Agent注册: {agent.name} (技能: {', '.join(agent.skills)})")

    def submit_task(self, task: Task):
        """提交任务"""
        self.task_queue.append(task)
        print(f"📌 新任务: [{task.id}] {task.description} (需要: {', '.join(task.required_skills)})")

    def allocate_tasks(self):
        """分配所有待分配的任务"""
        print(f"\n🔄 开始任务分配...")

        for task in self.task_queue:
            if task.status != "pending":
                continue

            # 收集所有出价
            bids = []
            for agent in self.agents:
                bid_value = agent.bid(task)
                if bid_value > 0:
                    bids.append((agent, bid_value))

            if not bids:
                print(f"  ⚠️ 任务 [{task.id}] 无Agent可处理")
                continue

            # 选择出价最高的Agent
            bids.sort(key=lambda x: x[1], reverse=True)  # 按元组第二个元素(出价)降序排序，bids[0]即最高出价
            winner = bids[0][0]

            task.status = "assigned"
            task.assigned_to = winner.name

            print(f"  📋 任务 [{task.id}] → {winner.name} (置信度: {bids[0][1]:.2f})")

    def execute_all(self):
        """执行所有已分配的任务"""
        print(f"\n⚡ 执行任务...")

        for task in self.task_queue:
            if task.status != "assigned":
                continue

            agent = next(a for a in self.agents if a.name == task.assigned_to)  # next+生成器：找第一个匹配的Agent，无匹配时抛StopIteration
            task.result = agent.execute(task)
            task.status = "completed"
            self.completed_tasks.append(task)

            print(f"  ✅ {task.result}")

    def report(self):
        """生成报告"""
        print(f"\n{'='*50}")
        print(f"📊 任务市场报告")
        print(f"  完成任务数: {len(self.completed_tasks)}")
        print(f"  Agent表现:")
        for agent in self.agents:
            print(f"    {agent.name}: 积分={agent.total_reward:.1f}")

# 使用示例
market = TaskMarket()

# 注册Agent
market.register_agent(MarketAgent("搜索专家", ["搜索", "信息提取"]))
market.register_agent(MarketAgent("数据分析师", ["数据分析", "可视化", "统计"]))
market.register_agent(MarketAgent("全栈工程师", ["编码", "架构", "搜索"]))
market.register_agent(MarketAgent("技术写手", ["写作", "编辑", "信息提取"]))

# 提交任务
market.submit_task(Task("T1", "搜索AI Agent最新论文", ["搜索", "信息提取"], 2, 10))
market.submit_task(Task("T2", "分析框架使用趋势数据", ["数据分析", "可视化"], 3, 15))
market.submit_task(Task("T3", "编写Agent系统代码", ["编码", "架构"], 4, 20))
market.submit_task(Task("T4", "撰写技术调研报告", ["写作", "信息提取"], 2, 10))

# 分配和执行
market.allocate_tasks()
market.execute_all()
market.report()

"""
Mixture-of-Agents (MoA) 简化实现

核心思想：多个LLM分层协作，逐层精炼
参考论文：Together AI "Mixture-of-Agents Enhances Large Language Model Capabilities" (2024)
"""

from openai import OpenAI
from typing import List, Dict
import asyncio

class MixtureOfAgents:
    """
    Mixture-of-Agents 推理框架

    架构：多层Agent，每层Agent可看到上一层所有输出
    """

    def __init__(self, layers: List[List[Dict]]):
        """
        初始化MoA

        Args:
            layers: 每层Agent的配置列表（本实现仅使用OpenAI客户端）
                    [
                        [{"model": "gpt-4o", "role": "分析师"}, ...],  # Layer 1
                        [{"model": "gpt-4o", "role": "精炼者"}, ...],  # Layer 2
                        [{"model": "gpt-4o", "role": "聚合者"}],  # Final layer
                    ]
        """
        self.layers = layers
        self.client = OpenAI()

    def _call_agent(self, model: str, role: str, prompt: str,
                    prev_responses: List[str] = None) -> str:
        """
        调用单个Agent

        如果有前一层的输出，将其作为参考信息注入提示词
        """
        messages = [{"role": "system", "content": f"你是一个{role}。"}]

        if prev_responses:
            # 将前一层所有Agent的输出作为参考
            ref_text = "\n\n---\n\n".join(
                [f"【参考回答 {i+1}】\n{r}" for i, r in enumerate(prev_responses)]
            )
            messages.append({
                "role": "user",
                "content": (
                    f"以下是其他专家对同一问题的回答，请参考并给出你的改进版回答：\n\n"
                    f"{ref_text}\n\n"
                    f"---\n原始问题：{prompt}\n\n"
                    f"请综合以上参考，给出更准确、更全面的回答。"
                )
            })
        else:
            messages.append({"role": "user", "content": prompt})

        response = self.client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=0.7
        )
        return response.choices[0].message.content

    def run(self, query: str) -> str:
        """
        执行MoA多层推理

        Returns:
            最终聚合后的回答
        """
        prev_responses = []

        for layer_idx, layer_agents in enumerate(self.layers):
            current_responses = []
            print(f"\n🔄 Layer {layer_idx + 1} ({len(layer_agents)} agents)...")

            # 注意：同一层内的Agent在此实现中是顺序执行的。
            # 若需真正并行，可改为 asyncio.gather() 并发调用，显著降低层内延迟。
            for agent_cfg in layer_agents:
                response = self._call_agent(
                    model=agent_cfg["model"],
                    role=agent_cfg["role"],
                    prompt=query,
                    prev_responses=prev_responses if prev_responses else None
                )
                current_responses.append(response)
                print(f"  ✅ {agent_cfg['role']} ({agent_cfg['model']}) 完成")

            prev_responses = current_responses

        # 返回最后一层（聚合层）的输出
        return prev_responses[-1]

# === 使用示例 ===
moa = MixtureOfAgents(layers=[
    # Layer 1：多模型并行生成（利用模型多样性）
    [
        {"model": "gpt-4o", "role": "技术分析师"},
        {"model": "gpt-4o-mini", "role": "通俗解释者"},
        {"model": "gpt-4o", "role": "批判性审稿人"},
    ],
    # Layer 2：精炼（综合Layer 1输出）
    [
        {"model": "gpt-4o", "role": "综合精炼专家"},
        {"model": "gpt-4o", "role": "事实验证者"},
    ],
    # Layer 3：最终聚合
    [
        {"model": "gpt-4o", "role": "首席聚合官"},
    ],
])

result = moa.run("比较 Transformer 和 Mamba 架构的优劣势")
print(f"\n📋 MoA最终回答:\n{result}")

"""
Subagent三种模式的完整实现

1. Agent-as-Tool: 将Agent注册为Tool Calling中的工具
2. Dynamic Spawn: 运行时动态创建子Agent
3. Handoff: 任务转交（OpenAI Agents SDK风格）

参考: Anthropic "Building Effective Agents" (2025) — Orchestrator-Workers
"""

from openai import OpenAI
import json
from typing import Callable

client = OpenAI()

# ============================================================
# 模式1: Agent-as-Tool — 将Agent包装为OpenAI Tool Calling的工具
# ============================================================

class AgentAsTool:
    """将一个Agent包装为另一个Agent可调用的Tool"""

    def __init__(self, name: str, description: str, system_prompt: str):
        self.name = name
        self.description = description
        self.system_prompt = system_prompt

    def run(self, task: str) -> str:
        """Agent自主执行任务（内部可能有多轮推理）"""
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": task},
            ],
            temperature=0,
        )
        return response.choices[0].message.content

    def as_tool_schema(self) -> dict:
        """导出为OpenAI Tool Calling格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {
                        "task": {
                            "type": "string",
                            "description": "要交给该Agent执行的任务描述",
                        }
                    },
                    "required": ["task"],
                },
            },
        }

# 定义Subagent
code_agent = AgentAsTool(
    name="code_expert",
    description="编程专家Agent，擅长代码编写、调试和代码审查",
    system_prompt="你是一个高级Python工程师。收到任务后，独立完成代码编写和测试。输出完整代码和简要说明。",
)

research_agent = AgentAsTool(
    name="research_expert",
    description="研究专家Agent，擅长搜索信息、分析论文、生成调研报告",
    system_prompt="你是一个资深研究员。收到课题后，进行深入分析，输出结构化的研究报告。",
)

# 编排者Agent使用Subagent作为工具
subagents = {"code_expert": code_agent, "research_expert": research_agent}
tools = [agent.as_tool_schema() for agent in subagents.values()]

def orchestrator_with_subagents(user_request: str, max_steps: int = 5) -> str:
    """编排者Agent：通过Tool Calling调用Subagent"""
    messages = [
        {"role": "system", "content": """你是一个项目经理Agent。你手下有多个专家Agent可以调用。
分析用户需求，将子任务委派给合适的专家Agent，最后汇总结果给用户。
每个专家Agent都是独立运行的，你只需要描述清楚任务即可。"""},
        {"role": "user", "content": user_request},
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            temperature=0,
        )
        message = response.choices[0].message

        # 无工具调用 → 编排者已得出最终结论
        if not message.tool_calls:
            return message.content

        messages.append(message)

        # 调用Subagent
        for tool_call in message.tool_calls:
            agent_name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            task = args.get("task", "")

            print(f"  🤖 [Orchestrator → {agent_name}] {task[:60]}...")
            result = subagents[agent_name].run(task)
            print(f"  ✅ [{agent_name}] 返回 {len(result)} 字符")

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })

    return "达到最大步骤数"

# ============================================================
# 模式2: Dynamic Spawn — 运行时动态创建子Agent
# ============================================================

def dynamic_subagent(task: str, context: str = "") -> str:
    """动态创建一个临时Subagent执行特定任务

    与预定义Agent不同，Dynamic Spawn根据任务需求
    在运行时构造专属的system_prompt和参数配置。
    适用于编排者无法预知子任务类型的场景。
    """

    # 让LLM分析任务需求，动态生成Subagent的配置
    config_resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": """分析任务需求，生成最适合的Agent配置。
返回JSON: {"role": "角色名", "system_prompt": "专家提示词", "temperature": 0.0-1.0}"""},
            {"role": "user", "content": f"任务: {task}\n上下文: {context}"},
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
    config = json.loads(config_resp.choices[0].message.content)

    print(f"  🔧 动态创建Subagent: {config.get('role', 'unknown')}")

    # 用动态生成的配置创建并运行子Agent
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": config.get("system_prompt", "你是一个有用的助手。")},
            {"role": "user", "content": task},
        ],
        temperature=config.get("temperature", 0),
    )
    return response.choices[0].message.content

# ============================================================
# 模式3: Handoff — 任务转交（类似OpenAI Agents SDK的handoffs）
# ============================================================

class HandoffAgent:
    """支持任务转交的Agent

    当Agent判断任务超出自身能力范围时，
    主动将对话转交给更合适的Agent继续处理。
    人类用户无感知，对话上下文无损传递。
    """

    def __init__(self, name: str, system_prompt: str, handoff_targets: dict[str, 'HandoffAgent'] = None):
        self.name = name
        self.system_prompt = system_prompt
        self.handoff_targets = handoff_targets or {}

    def run(self, messages: list[dict], max_handoffs: int = 3) -> str:
        """运行Agent，支持自动转交"""
        current_agent = self
        handoff_count = 0

        while handoff_count < max_handoffs:
            # 构造当前Agent的系统提示（含转交能力声明）
            handoff_info = ""
            if current_agent.handoff_targets:
                targets = ", ".join(
                    f"{name}（{a.system_prompt[:30]}...）"
                    for name, a in current_agent.handoff_targets.items()
                )
                handoff_info = f"\n\n你可以将对话转交给其他专家: {targets}\n转交时回复格式: HANDOFF:专家名称"

            full_messages = [
                {"role": "system", "content": current_agent.system_prompt + handoff_info},
            ] + messages

            response = client.chat.completions.create(
                model="gpt-4o",
                messages=full_messages,
                temperature=0,
            )
            reply = response.choices[0].message.content

            # 检查是否请求转交
            if reply.strip().startswith("HANDOFF:"):
                target_name = reply.strip().split("HANDOFF:")[1].strip()
                if target_name in current_agent.handoff_targets:
                    print(f"  🔄 [{current_agent.name}] → 转交给 [{target_name}]")
                    current_agent = current_agent.handoff_targets[target_name]
                    handoff_count += 1
                    continue

            return f"[{current_agent.name}] {reply}"

        return f"[{current_agent.name}] 达到最大转交次数"

# === 使用示例 ===

# 模式1: Agent-as-Tool
print("=== 模式1: Agent-as-Tool ===")
result = orchestrator_with_subagents("帮我调研目前主流的Agent框架，然后写一个LangGraph的Hello World示例")
print(f"结果: {result[:300]}...\n")

# 模式2: Dynamic Spawn
print("=== 模式2: Dynamic Spawn ===")
result = dynamic_subagent("分析Transformer注意力机制的计算复杂度", context="面向大三本科生的教学文档")
print(f"结果: {result[:300]}...\n")

# 模式3: Handoff
print("=== 模式3: Handoff ===")
billing_agent = HandoffAgent("账单专家", "你是账单和支付问题的专家。只回答账单相关问题，其他问题请转交。")
tech_agent = HandoffAgent("技术支持", "你是技术支持专家。解决软件使用和技术故障问题。")
triage_agent = HandoffAgent(
    "分诊Agent",
    "你是客服分诊Agent。判断用户问题类型，转交给对应的专家。不要自己回答具体问题。",
    handoff_targets={"账单专家": billing_agent, "技术支持": tech_agent},
)

result = triage_agent.run([{"role": "user", "content": "我的软件一直崩溃，怎么办？"}])
print(f"结果: {result[:300]}...")

"""
A2A通信概念实现
定义Agent间的标准化消息格式
"""

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Callable
import json
import uuid

class MessageType(Enum):
    """消息类型"""
    REQUEST = "request"        # 任务请求
    RESPONSE = "response"      # 任务响应
    STATUS = "status"          # 状态更新
    DELEGATE = "delegate"      # 任务委托
    NOTIFY = "notify"          # 通知

@dataclass
class AgentCard:
    """Agent名片 - 描述Agent的能力"""
    agent_id: str
    name: str
    description: str
    capabilities: list[str]
    input_schema: dict      # 接受的输入格式
    output_schema: dict     # 输出格式
    endpoint: str           # 通信端点

@dataclass
class A2AMessage:
    """Agent间通信消息"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))  # default_factory接收lambda，每次实例化生成新UUID
    type: MessageType = MessageType.REQUEST
    sender: str = ""         # 发送方Agent ID
    receiver: str = ""       # 接收方Agent ID
    content: dict = field(default_factory=dict)
    metadata: dict = field(default_factory=dict)
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    reply_to: str | None = None  # 回复的消息ID

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "type": self.type.value,
            "sender": self.sender,
            "receiver": self.receiver,
            "content": self.content,
            "metadata": self.metadata,
            "timestamp": self.timestamp,
            "reply_to": self.reply_to,
        }

    def to_json(self) -> str:
        return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)

class A2ARouter:
    """Agent间消息路由器"""

    def __init__(self):
        self.agents: dict[str, AgentCard] = {}
        self.message_handlers: dict[str, Callable] = {}
        self.message_log: list[A2AMessage] = []

    def register(self, card: AgentCard, handler: Callable):
        """注册Agent"""
        self.agents[card.agent_id] = card
        self.message_handlers[card.agent_id] = handler
        print(f"📝 注册Agent: {card.name} ({card.agent_id})")

    def discover(self, required_capabilities: list[str]) -> list[AgentCard]:
        """发现具有指定能力的Agent"""
        results = []
        for card in self.agents.values():
            # all()判断所有要求的能力是否都在该Agent的能力列表中（子集关系检查）
            if all(cap in card.capabilities for cap in required_capabilities):
                results.append(card)
        return results

    def send(self, message: A2AMessage) -> A2AMessage | None:
        """路由消息"""
        self.message_log.append(message)

        if message.receiver not in self.message_handlers:
            print(f"❌ 目标Agent不存在: {message.receiver}")
            return None

        handler = self.message_handlers[message.receiver]
        response = handler(message)

        if response:
            self.message_log.append(response)

        return response

# === 使用示例 ===

router = A2ARouter()

# 注册Agent
def research_handler(msg: A2AMessage) -> A2AMessage:
    print(f"  🔬 [研究Agent] 收到请求: {msg.content.get('task', '')}")
    return A2AMessage(
        type=MessageType.RESPONSE,
        sender="agent-research",
        receiver=msg.sender,
        content={"result": "找到5篇相关论文", "papers": ["论文A", "论文B"]},
        reply_to=msg.id,
    )

router.register(
    AgentCard(
        agent_id="agent-research",
        name="研究Agent",
        description="学术研究和文献搜索",
        capabilities=["搜索", "文献分析"],
        input_schema={"type": "object", "properties": {"task": {"type": "string"}}},
        output_schema={"type": "object", "properties": {"result": {"type": "string"}}},
        endpoint="local://agent-research",
    ),
    research_handler,
)

# 发送消息
msg = A2AMessage(
    type=MessageType.REQUEST,
    sender="agent-manager",
    receiver="agent-research",
    content={"task": "搜索关于ReAct的最新论文"},
)

response = router.send(msg)
if response:
    print(f"  ✅ 收到响应: {response.content}")

"""ANP Agent发现与通信概念演示"""

from dataclasses import dataclass, field
import json

@dataclass
class AgentDescriptor:
    """Agent描述符 —— 类似ANP的Agent Description文件"""
    agent_id: str           # DID格式: did:anp:agent-xxx
    name: str
    description: str
    capabilities: list[str]
    endpoint: str           # 通信端点
    protocols: list[str] = field(default_factory=lambda: ["anp/1.0"])  # lambda返回新列表实例，避免可变默认值陷阱
    trust_score: float = 0.0  # 信任评分

class AgentRegistry:
    """简化的Agent注册与发现中心"""

    def __init__(self):
        self.agents: dict[str, AgentDescriptor] = {}

    def register(self, descriptor: AgentDescriptor):
        """注册Agent到网络"""
        self.agents[descriptor.agent_id] = descriptor
        print(f"  ✅ Agent注册成功: {descriptor.name} ({descriptor.agent_id})")

    def discover(self, capability: str) -> list[AgentDescriptor]:
        """按能力发现Agent"""
        return [
            a for a in self.agents.values()
            if capability in a.capabilities
        ]

    def verify_identity(self, agent_id: str) -> bool:
        """验证Agent身份（简化版，实际应使用DID/VC验证）"""
        return agent_id in self.agents

# 示例：Agent发现与跨网络调用
registry = AgentRegistry()

# 注册不同组织的Agent
registry.register(AgentDescriptor(
    agent_id="did:anp:travel-agent-001",
    name="携程旅行助手",
    description="提供机票、酒店查询和预订服务",
    capabilities=["机票查询", "酒店预订", "行程规划"],
    endpoint="https://api.ctrip.com/agent/v1",
))

registry.register(AgentDescriptor(
    agent_id="did:anp:weather-agent-001",
    name="天气服务Agent",
    description="提供全球天气预报",
    capabilities=["天气查询", "气象预警"],
    endpoint="https://api.weather.com/agent/v1",
))

# 发现具有"机票查询"能力的Agent
travel_agents = registry.discover("机票查询")
for agent in travel_agents:
    print(f"  🔍 发现Agent: {agent.name} @ {agent.endpoint}")
    if registry.verify_identity(agent.agent_id):
        print(f"     ✅ 身份验证通过")

"""
基于LangGraph的多Agent工作流编排
实现: 研究 → 分析 → 写作 → 审核 的流水线
"""

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

# === 1. 定义全局状态 ===

# TypedDict定义结构化字典类型，LangGraph用作工作流状态容器
class WorkflowState(TypedDict):
    """工作流共享状态"""
    # Annotated[list, add_messages]：add_messages是归约函数，新消息追加而非覆盖
    messages: Annotated[list, add_messages]
    task: str                    # 原始任务
    research_result: str         # 研究结果
    analysis_result: str         # 分析结果
    draft: str                   # 写作草稿
    review_feedback: str         # 审核反馈
    final_output: str            # 最终输出
    revision_count: int          # 修改次数
    status: str                  # 工作流状态

# === 2. 定义Agent节点 ===

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def researcher_node(state: WorkflowState) -> WorkflowState:
    """研究Agent: 搜索和收集信息"""
    task = state["task"]

    response = llm.invoke([
        SystemMessage(content="""你是一个专业研究员。请针对给定主题进行深入研究。
输出格式：
1. 关键发现（3-5条）
2. 数据/证据支持
3. 相关参考来源"""),
        HumanMessage(content=f"请研究: {task}"),
    ])

    return {
        "research_result": response.content,
        "status": "researched",
        "messages": [AIMessage(content=f"[研究员] {response.content[:200]}...")],
    }

def analyst_node(state: WorkflowState) -> WorkflowState:
    """分析Agent: 深度分析研究数据"""
    research = state["research_result"]
    task = state["task"]

    response = llm.invoke([
        SystemMessage(content="""你是一个数据分析专家。基于研究数据进行深度分析。
输出格式：
1. 核心洞察
2. 趋势分析
3. 机会与风险
4. 建议"""),
        HumanMessage(content=f"任务: {task}\n\n研究数据:\n{research}"),
    ])

    return {
        "analysis_result": response.content,
        "status": "analyzed",
        "messages": [AIMessage(content=f"[分析师] {response.content[:200]}...")],
    }

def writer_node(state: WorkflowState) -> WorkflowState:
    """写作Agent: 撰写报告"""
    task = state["task"]
    research = state["research_result"]
    analysis = state["analysis_result"]
    feedback = state.get("review_feedback", "")
    revision_count = state.get("revision_count", 0)

    prompt = f"任务: {task}\n\n研究结果:\n{research}\n\n分析结果:\n{analysis}"
    if feedback:
        prompt += f"\n\n审核反馈（请据此修改）:\n{feedback}"

    response = llm.invoke([
        SystemMessage(content="""你是一个专业报告写作者。请撰写一份高质量的分析报告。
报告结构: 摘要 → 背景 → 核心发现 → 详细分析 → 结论与建议"""),
        HumanMessage(content=prompt),
    ])

    return {
        "draft": response.content,
        "revision_count": revision_count + 1,
        "status": "drafted",
        "messages": [AIMessage(content=f"[写作者] 报告草稿已完成（第{revision_count + 1}版）")],
    }

def reviewer_node(state: WorkflowState) -> WorkflowState:
    """审核Agent: 审查报告质量"""
    draft = state["draft"]
    task = state["task"]

    response = llm.invoke([
        SystemMessage(content="""你是一个严格的质量审核专家。请审查报告，评估以下方面：
1. 内容准确性（1-10分）
2. 逻辑连贯性（1-10分）
3. 信息完整性（1-10分）
4. 文字表达（1-10分）

如果总分 >= 32（满分40），回复"APPROVED: [简短评语]"
否则回复"REVISION: [具体修改建议]" """),
        HumanMessage(content=f"原始任务: {task}\n\n待审核报告:\n{draft}"),
    ])

    return {
        "review_feedback": response.content,
        "status": "reviewed",
        "messages": [AIMessage(content=f"[审核者] {response.content[:200]}...")],
    }

# === 3. 路由逻辑 ===

def review_router(state: WorkflowState) -> Literal["writer", "end"]:
    """根据审核结果决定是否需要修改"""
    feedback = state.get("review_feedback", "")
    revision_count = state.get("revision_count", 0)

    # 如果审核通过或修改次数超过3次，结束
    if "APPROVED" in feedback or revision_count >= 3:
        return "end"
    return "writer"

def finalize_node(state: WorkflowState) -> WorkflowState:
    """最终整理输出"""
    return {
        "final_output": state["draft"],
        "status": "completed",
        "messages": [AIMessage(content="✅ 报告已完成！")],
    }

# === 4. 构建工作流图 ===

workflow = StateGraph(WorkflowState)

# 添加节点
workflow.add_node("researcher", researcher_node)
workflow.add_node("analyst", analyst_node)
workflow.add_node("writer", writer_node)
workflow.add_node("reviewer", reviewer_node)
workflow.add_node("finalize", finalize_node)

# 添加边
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "analyst")
workflow.add_edge("analyst", "writer")
workflow.add_edge("writer", "reviewer")

# 条件路由: 审核 → 修改 或 结束
workflow.add_conditional_edges("reviewer", review_router, {
    "writer": "writer",
    "end": "finalize",
})
workflow.add_edge("finalize", END)

# 编译
app = workflow.compile()

# === 5. 运行 ===

result = app.invoke({
    "messages": [],
    "task": "分析2025-2026年AI Agent技术在企业中的落地情况和未来趋势",
    "research_result": "",
    "analysis_result": "",
    "draft": "",
    "review_feedback": "",
    "final_output": "",
    "revision_count": 0,
    "status": "started",
})

print(f"\n{'='*60}")
print(f"📋 最终报告 (修改了{result['revision_count']}次):")
print(f"{'='*60}")
print(result["final_output"])