多 Agent 协作

复杂任务
    ↓
┌──────────────┐
│  任务分解     │ → 将大任务拆成子任务
│ (Decompose)  │
└──────┬───────┘
       ↓
┌──────────────┐
│  任务分配     │ → 将子任务匹配给合适的 Agent
│ (Allocate)   │
└──────┬───────┘
       ↓
┌──────────────┐
│  协调执行     │ → 管理依赖、同步、冲突
│ (Coordinate) │
└──────┬───────┘
       ↓
┌──────────────┐
│  结果聚合     │ → 合并各 Agent 的输出
│ (Aggregate)  │
└──────────────┘

import asyncio

class CentralPlanner:
    """中央规划器：分解任务 + 按能力分配 + 按依赖编排"""

    def __init__(self, agents: dict):
        self.agents = agents

    async def plan_and_execute(self, task: str):
        # Step 1: LLM 分解任务为子任务
        subtasks = await self.decompose(task)

        # Step 2: 根据 Agent 能力分配子任务
        assignments = self.assign(subtasks)

        # Step 3: 按依赖关系编排执行
        results = {}
        for batch in self.topological_sort(assignments):
            # 同一批次的任务无依赖，可并行
            batch_results = await asyncio.gather(
                *[self.agents[a.agent_id].execute(a.subtask, results)
                  for a in batch]
            )
            results.update(zip([a.subtask.id for a in batch], batch_results))

        return results

    def assign(self, subtasks):
        """基于 Agent 能力描述的自动匹配"""
        assignments = []
        for subtask in subtasks:
            best_agent = max(
                self.agents.values(),
                key=lambda a: self.capability_match(a, subtask)
            )
            assignments.append(Assignment(subtask, best_agent.id))
        return assignments

class LLMCoordinator:
    async def coordinate(self, task, agents):
        agent_descriptions = "\n".join([
            f"- {a.name}: {a.description}, 擅长: {a.capabilities}"
            for a in agents
        ])

        plan = await self.llm.generate(f"""
        任务: {task}

        可用 Agent:
        {agent_descriptions}

        请分解任务并分配给合适的 Agent。
        输出格式:
        1. [Agent名] → 子任务描述 (依赖: 无/步骤N)
        2. ...
        """)

        return self.parse_and_execute(plan)

class DynamicAllocator:
    """基于亲和度的事件驱动动态分配"""

    def __init__(self, agents):
        self.agents = agents
        self.affinity_scores = {}  # (agent, task_type) → 历史成功率

    async def allocate(self, subtask):
        candidates = []
        for agent in self.agents:
            score = self.compute_score(agent, subtask)
            candidates.append((agent, score))

        best = max(candidates, key=lambda x: x[1])
        return best[0]

    def compute_score(self, agent, subtask):
        return (
            0.4 * self.capability_match(agent, subtask) +   # 能力匹配
            0.3 * self.affinity_scores.get((agent.id, subtask.type), 0.5) +  # 历史表现
            0.2 * (1 - agent.current_load / agent.max_load) +  # 当前负载
            0.1 * self.recency_bonus(agent)  # 最近是否空闲
        )

    def update_affinity(self, agent_id, task_type, success: bool):
        """根据执行结果更新亲和度"""
        key = (agent_id, task_type)
        old = self.affinity_scores.get(key, 0.5)
        self.affinity_scores[key] = old * 0.8 + (1.0 if success else 0.0) * 0.2

# 用 DAG 表示任务依赖
task_graph = {
    "search":    {"depends_on": []},
    "analyze":   {"depends_on": ["search"]},
    "visualize": {"depends_on": ["search"]},      # 和 analyze 并行
    "report":    {"depends_on": ["analyze", "visualize"]},  # 等两者完成
}

# 拓扑排序确定执行顺序
# Batch 1: search（无依赖）
# Batch 2: analyze, visualize（并行）
# Batch 3: report（等待前两个）

def compress_for_handoff(agent_output: str, max_tokens: int = 500):
    """传递摘要而非完整输出，减少 token 冗余"""
    if count_tokens(agent_output) > max_tokens:
        return summarize(agent_output, max_tokens=max_tokens)
    return agent_output

from datetime import datetime
from collections import defaultdict

class MessageBus:
    """点对点 + 广播消息传递"""

    def __init__(self):
        self.queues = defaultdict(list)  # agent_id → message_queue

    async def send(self, from_agent: str, to_agent: str, message: dict):
        """发送消息给特定 Agent"""
        msg = {
            "from": from_agent,
            "to": to_agent,
            "content": message,
            "timestamp": datetime.now()
        }
        self.queues[to_agent].append(msg)

    async def broadcast(self, from_agent: str, message: dict):
        """广播消息给所有 Agent"""
        for agent_id in self.queues:
            if agent_id != from_agent:
                self.queues[agent_id].append({
                    "from": from_agent,
                    "content": message
                })

import asyncio
from typing import TypedDict


class SharedState(TypedDict):
    """共享状态的结构定义"""
    messages: list[str]
    research_data: dict
    draft: str
    review_comments: list[str]
    status: str


class SafeSharedState:
    """带并发控制的共享状态"""

    def __init__(self):
        self.state: SharedState = {
            "messages": [], "research_data": {},
            "draft": "", "review_comments": [], "status": "init",
        }
        self.lock = asyncio.Lock()

    async def update(self, agent_id: str, updates: dict):
        async with self.lock:  # 确保原子更新
            self.state.update(updates)
            self.state["last_updated_by"] = agent_id

from datetime import datetime

class Blackboard:
    def __init__(self):
        self.public_space = []   # 公共区域：所有 Agent 可见
        self.private_spaces = {} # 私有区域：各 Agent 独立空间

    def post(self, agent_id: str, content: dict, visibility="public"):
        entry = {
            "author": agent_id,
            "content": content,
            "timestamp": datetime.now()
        }
        if visibility == "public":
            self.public_space.append(entry)
        else:
            self.private_spaces.setdefault(agent_id, []).append(entry)

    def read(self, agent_id: str) -> list:
        """Agent 读取黑板上与自己相关的内容"""
        return self.public_space + self.private_spaces.get(agent_id, [])

class BlackboardSystem:
    def __init__(self, blackboard, agents, controller):
        self.blackboard = blackboard
        self.agents = agents
        self.controller = controller  # 控制器（决定谁行动）

    async def run(self, task):
        self.blackboard.post("system", {"task": task})

        while not self.is_solved():
            # 控制器选择下一个行动的 Agent
            active_agent = self.controller.select(
                self.blackboard, self.agents
            )
            # Agent 读取黑板、处理、写回结果
            result = await active_agent.process(
                self.blackboard.read(active_agent.id)
            )
            self.blackboard.post(active_agent.id, result)

from agents import Agent, handoff

# 定义专家 Agent
refund_agent = Agent(
    name="Refund Agent",
    instructions="你负责处理退款请求...",
    tools=[execute_refund, check_refund_status],
)

shipping_agent = Agent(
    name="Shipping Agent",
    instructions="你负责处理物流问题...",
    tools=[track_package, update_address],
)

# 路由 Agent 有 Handoff 能力
triage_agent = Agent(
    name="Triage Agent",
    instructions="根据用户需求将请求路由给合适的专家",
    handoffs=[
        # 简单写法：直接传 Agent
        refund_agent,
        # 完整写法：通过 handoff(...) 自定义工具名 / 回调 / 输入过滤
        handoff(
            agent=shipping_agent,
            tool_name_override="route_to_shipping",
            tool_description_override="将对话交给物流专家处理快递相关问题",
            on_handoff=lambda ctx: log_handoff(ctx),
            input_filter=keep_last_n_messages(5),
            input_type=ShippingHandoffInput,  # Pydantic 模型校验 Handoff 入参
        ),
    ],
    # 内部自动生成工具：transfer_to_refund_agent, route_to_shipping
)

from typing import TypedDict
from langgraph.graph import StateGraph, END

class SupervisorState(TypedDict):
    messages: list
    current_agent: str

def supervisor_node(state):
    """Supervisor 决定下一个处理的 Agent"""
    response = supervisor_llm.invoke(
        f"当前对话：{state['messages']}\n"
        f"可用专家：researcher, coder, writer\n"
        f"谁应该处理下一步？回复 agent 名称或 'FINISH'"
    )
    return {"current_agent": response.agent_name}

def route(state):
    if state["current_agent"] == "FINISH":
        return END
    return state["current_agent"]

graph = StateGraph(SupervisorState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("coder", coder_node)
graph.add_conditional_edges("supervisor", route)
# 每个 worker 完成后回到 supervisor
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")

# 工具函数返回 Agent 对象触发 Handoff
def handle_refund(order_id: str):
    """处理退款请求"""
    status = check_order(order_id)
    if status == "delivered":
        return refund_agent  # 返回 Agent 对象 = 触发 Handoff
    else:
        return f"订单 {order_id} 尚未送达，无法退款"

triage_agent = Agent(
    name="Triage",
    tools=[handle_refund],  # 函数可能返回 Agent
)

from agents import handoff

def filter_for_billing(handoff_input):
    """只传递与计费相关的上下文给 Billing Agent"""
    filtered = []
    for msg in handoff_input.messages:
        if is_billing_related(msg) or msg == handoff_input.messages[-1]:
            filtered.append(msg)
    handoff_input.messages = filtered
    return handoff_input

billing_handoff = handoff(
    agent=billing_agent,
    input_filter=filter_for_billing,
)

class ReliableHandoff:
    async def execute(self, from_agent, to_agent, context):
        # 1. 验证目标 Agent 可用
        if not to_agent.is_healthy():
            return await self.fallback(from_agent, context)

        # 2. 结构化上下文打包
        payload = self.pack_context(context)

        # 3. 执行 Handoff
        try:
            result = await to_agent.invoke(payload)
        except Exception as e:
            # 4. 失败回退
            return await from_agent.handle_handoff_failure(e, context)

        # 5. 记录 Handoff 日志
        self.log_handoff(from_agent.id, to_agent.id, payload, result)
        return result

import json
import uuid
from dataclasses import dataclass, field
from enum import Enum


class TaskState(Enum):
    """A2A Task 状态枚举"""
    SUBMITTED = "submitted"
    WORKING = "working"
    INPUT_REQUIRED = "input-required"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class AgentCard:
    """Agent Card — 声明 Agent 的能力，发布于 /.well-known/agent.json"""
    name: str
    description: str
    url: str
    skills: list[dict] = field(default_factory=list)
    capabilities: dict = field(default_factory=lambda: {
        "streaming": True, "pushNotifications": False,
    })

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


@dataclass
class Task:
    """A2A Task — 协议核心工作单元，带状态机校验"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    state: TaskState = TaskState.SUBMITTED
    messages: list[dict] = field(default_factory=list)
    artifacts: list[dict] = field(default_factory=list)

    # 合法状态转换表
    _transitions = {
        TaskState.SUBMITTED: {TaskState.WORKING, TaskState.FAILED},
        TaskState.WORKING: {TaskState.COMPLETED, TaskState.FAILED, TaskState.INPUT_REQUIRED},
        TaskState.INPUT_REQUIRED: {TaskState.WORKING, TaskState.FAILED},
    }

    def transition(self, new_state: TaskState):
        allowed = self._transitions.get(self.state, set())
        if new_state not in allowed:
            raise ValueError(f"非法状态转换: {self.state.value} → {new_state.value}")
        self.state = new_state


class A2AServer:
    """模拟 A2A 服务端——接收 JSON-RPC 请求、管理 Task 生命周期"""

    def __init__(self, card: AgentCard):
        self.card = card
        self.tasks: dict[str, Task] = {}

    def handle_request(self, request: dict) -> dict:
        """路由 JSON-RPC 2.0 请求"""
        method = request["method"]
        params = request.get("params", {})
        req_id = request["id"]

        if method == "message/send":
            result = self._send(params)
        elif method == "tasks/get":
            task = self.tasks[params["id"]]
            result = {"id": task.id, "state": task.state.value}
        else:
            return {"jsonrpc": "2.0", "id": req_id,
                    "error": {"code": -32601, "message": "方法不存在"}}

        return {"jsonrpc": "2.0", "id": req_id, "result": result}

    def _send(self, params: dict) -> dict:
        task_id = params.get("id", str(uuid.uuid4()))
        if task_id in self.tasks:
            task = self.tasks[task_id]
            task.messages.append(params["message"])
            task.transition(TaskState.WORKING)
        else:
            task = Task(id=task_id, messages=[params["message"]])
            self.tasks[task_id] = task
            task.transition(TaskState.WORKING)

        # 模拟业务逻辑
        text = task.messages[-1]["parts"][0]["text"]
        if "分析" in text and "时间" not in text:
            task.transition(TaskState.INPUT_REQUIRED)
            return {"id": task.id, "state": task.state.value,
                    "messages": [{"role": "agent",
                                  "parts": [{"type": "text", "text": "请指定分析时间范围"}]}]}

        task.transition(TaskState.COMPLETED)
        task.artifacts.append({
            "name": "report",
            "parts": [{"type": "text", "text": f"基于 [{text}] 的分析报告..."}]
        })
        return {"id": task.id, "state": task.state.value, "artifacts": task.artifacts}

// Supervisor 把 Agent 当作工具调用
const tools = [researchAgent, codeAgent, reviewAgent];
const supervisor = createReActAgent(model, tools);

import asyncio

async def fan_out_fan_in(task, agents):
    # Fan-out：并行分发
    results = await asyncio.gather(
        *[agent.execute(task) for agent in agents]
    )
    # Fan-in：汇聚结果
    return aggregate(results)  # 投票、合并或 LLM 综合

async def maker_checker_loop(task, maker, checker, max_rounds=3):
    for round in range(max_rounds):
        output = await maker.generate(task)
        review = await checker.review(output)
        if review.approved:
            return output
        task = f"{task}\n\n上一轮反馈：{review.feedback}"
    return output  # 达到最大轮次

import asyncio

class VotingResolver:
    """投票式冲突解决"""

    async def resolve_binary(self, agents, question) -> str:
        """二元投票：是/否"""
        votes = await asyncio.gather(
            *[agent.vote(question) for agent in agents]
        )
        yes_count = sum(1 for v in votes if v == "yes")
        return "yes" if yes_count > len(agents) / 2 else "no"

    async def resolve_weighted(self, agents, question) -> str:
        """加权投票：按 Agent 专业度加权"""
        weighted_votes = {}
        for agent in agents:
            vote = await agent.vote(question)
            weight = agent.expertise_score
            weighted_votes[vote] = weighted_votes.get(vote, 0) + weight
        return max(weighted_votes, key=weighted_votes.get)

    async def resolve_ranked(self, agents, options) -> str:
        """排序投票：Borda 计分法"""
        rankings = await asyncio.gather(
            *[agent.rank(options) for agent in agents]
        )
        scores = {opt: 0 for opt in options}
        for ranking in rankings:
            for i, opt in enumerate(ranking):
                scores[opt] += len(options) - i
        return max(scores, key=scores.get)

class ConsensusProtocol:
    """多轮共识协议"""

    def __init__(self, agents, max_rounds=5, threshold=0.8):
        self.agents = agents
        self.max_rounds = max_rounds
        self.threshold = threshold  # 80% 一致才算共识

    async def reach_consensus(self, topic):
        proposals = {}

        for round_num in range(self.max_rounds):
            for agent in self.agents:
                context = {
                    "topic": topic,
                    "round": round_num,
                    "other_proposals": {
                        a.id: p for a, p in proposals.items() if a != agent
                    }
                }
                proposals[agent] = await agent.propose(context)

            # 检查是否达成共识
            agreement_rate = self.calculate_agreement(proposals)
            if agreement_rate >= self.threshold:
                return self.merge_proposals(proposals)

            # 动态调整阈值（任务越紧急，越容易通过）
            self.threshold *= 0.95

        # 达到最大轮次仍未共识 → 降级到投票或人工裁决
        return await self.fallback_to_voting(proposals)

class MediatorResolver:
    def __init__(self, mediator_agent):
        self.mediator = mediator_agent

    async def resolve(self, conflicting_outputs: dict):
        prompt = f"""
        以下是不同 Agent 对同一问题的分析结果，它们存在冲突：

        {self._format_conflicts(conflicting_outputs)}

        请分析每个 Agent 的推理过程，指出各自的优缺点，
        然后给出你的最终裁决和理由。
        """
        decision = await self.mediator.generate(prompt)
        return decision

class AntiSycophancyProtocol:
    def __init__(self, max_rounds=3):
        self.max_rounds = max_rounds

    async def debate(self, agents, topic):
        for round in range(self.max_rounds):
            for agent in agents:
                # 收集其他 Agent 的观点
                other_views = await self.collect_other_views(agents, agent, topic)

                response = await agent.generate(f"""
                第一步：独立分析（忽略其他 Agent 的观点）
                {topic}

                第二步：考虑其他观点后，明确说明你是否改变立场
                如果改变，解释具体哪个论据说服了你
                如果不改变，解释为什么你认为自己的分析更正确

                其他 Agent 的观点：{other_views}
                """)

class ResourceConflictResolver:
    def __init__(self):
        self.locks = {}  # 资源锁

    async def request_resource(self, agent_id, resource_id):
        if resource_id in self.locks:
            holder = self.locks[resource_id]
            if self.priority(agent_id) > self.priority(holder):
                # 高优先级 Agent 可以抢占
                await self.preempt(holder, resource_id)
                self.locks[resource_id] = agent_id
                return True
            else:
                # 排队等待
                await self.wait_queue(resource_id, agent_id)
                return False
        else:
            self.locks[resource_id] = agent_id
            return True

agent_prompt = """
你是团队中的一员，正在解决一个复杂问题。

在回答之前，请考虑：
1. 其他 Agent 可能拥有什么信息？
2. 你的独特贡献是什么？（不要重复他人已有的分析）
3. 如何让你的输出与团队的整体目标互补？

只有 Theory-of-Mind prompt 条件才能产生"身份关联的差异化
和目标导向的互补性"——Agent 作为一个整合的、目标导向的
单元运作。
"""

class EmergentBehaviorMonitor:
    """监控多 Agent 系统的涌现行为"""

    def __init__(self, agents, baseline_behaviors, threshold=0.3):
        self.agents = agents
        self.baseline = baseline_behaviors
        self.threshold = threshold

    def detect_anomaly(self, agent_outputs):
        for agent_id, output in agent_outputs.items():
            # 检测：输出是否偏离基线行为
            deviation = self.compute_deviation(output, self.baseline[agent_id])
            if deviation > self.threshold:
                self.alert(f"Agent {agent_id} 行为偏离基线: {deviation}")

            # 检测：Agent 是否出现自引用循环
            if self.detect_self_reference(output):
                self.interrupt(agent_id, "检测到自引用循环")

            # 检测：级联错误模式
            if self.detect_cascade(agent_outputs):
                self.halt_all("检测到级联错误，暂停系统")

from collections import Counter

class BFTConsensus:
    """拜占庭容错共识：即使部分 Agent 异常，系统仍能正确决策"""

    def decide(self, agent_outputs: dict) -> str:
        # 需要 2/3 + 1 的 Agent 达成一致
        n = len(agent_outputs)
        threshold = (2 * n // 3) + 1

        votes = Counter(agent_outputs.values())
        for answer, count in votes.most_common():
            if count >= threshold:
                return answer

        return "NO_CONSENSUS"  # 需要人工介入

完全可控              平衡点              完全自主
│←────────────────────┼────────────────────→│
│ 确定性工作流        │ 引导式涌现          │ 无约束涌现
│ 无涌现              │ 有监控的自主性      │ 不可预测
│ Pipeline 模式       │ Hybrid 拓扑         │ 全连接对话
│ 安全但无创新        │ 最佳实践            │ 创新但危险

type Phase = "INIT" | "PLAN" | "EXEC" | "VERIFY" | "DONE";

const ALLOWED_TRANSITIONS: Record<Phase, Phase[]> = {
  INIT: ["PLAN"],
  PLAN: ["EXEC"],
  EXEC: ["VERIFY"],
  VERIFY: ["DONE", "EXEC"], // 不通过可打回重做
  DONE: [],
};

function transition(current: Phase, next: Phase): Phase {
  if (!ALLOWED_TRANSITIONS[current].includes(next)) {
    throw new Error(`非法状态转换: ${current} -> ${next}`);
  }
  return next;
}

Trace: "帮我分析竞品A的定价策略"（总耗时 8.2s）
│
├── Span: LLM 调用 - 理解任务（1.2s）
│   └── 输出: "需要搜索竞品A的定价信息"
│
├── Span: 工具调用 - web_search（2.5s）
│   └── 输出: [搜索结果 5 条]
│
├── Span: LLM 调用 - 分析结果（3.1s）
│   └── 输出: "竞品A采用阶梯定价..."
│
└── Span: LLM 调用 - 生成报告（1.4s）
    └── 输出: 最终分析报告

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

tracer = trace.get_tracer("multi-agent-system")

async def agent_execute(agent_name, task):
    with tracer.start_as_current_span(f"agent.{agent_name}") as span:
        span.set_attribute("agent.name", agent_name)
        span.set_attribute("task", task)

        # LLM 调用
        with tracer.start_as_current_span("llm.invoke") as llm_span:
            response = await llm.invoke(task)
            llm_span.set_attribute("tokens.input", response.input_tokens)
            llm_span.set_attribute("tokens.output", response.output_tokens)
            llm_span.set_attribute("model", response.model)

        # 工具调用
        for tool_call in response.tool_calls:
            with tracer.start_as_current_span(f"tool.{tool_call.name}") as tool_span:
                result = await execute_tool(tool_call)
                tool_span.set_attribute("tool.success", result.success)

        span.set_attribute("agent.status", "completed")
        return response

from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse()

@observe(name="triage-agent")
async def triage_agent(user_message):
    # 自动记录输入/输出/延迟/token
    response = await llm.invoke(user_message)

    if response.needs_handoff:
        # 嵌套 Span：Handoff
        with langfuse.span(name="handoff", metadata={"to": response.target_agent}):
            return await handoff(response.target_agent, response.context)

    return response

@observe(name="refund-agent")
async def refund_agent(context):
    # 嵌套 Span 自动形成调用树
    order = await get_order(context["order_id"])
    refund = await process_refund(order)
    return refund

1. 问题发现
   └→ 告警触发（延迟飙升/错误率上升/成本异常）

2. 问题定位
   └→ 查看 Trace 列表，筛选异常 Trace
      └→ 展开 Trace 的 Span 树
         └→ 定位到具体的 Agent/工具/LLM 调用

3. 根因分析
   └→ 检查该 Span 的输入/输出/prompt
      └→ 是 LLM 选错了工具？参数错误？工具超时？

4. 修复验证
   └→ 调整 prompt/工具描述/超时设置
      └→ 在 Playground 中用原始输入重放
         └→ 确认修复后部署

import json

class HandoffDebugger:
    def on_handoff(self, from_agent, to_agent, context):
        self.logger.info(
            f"Handoff: {from_agent} → {to_agent}",
            extra={
                "context_size": len(json.dumps(context)),
                "context_keys": list(context.keys()),
                "messages_count": len(context.get("messages", [])),
            }
        )

class DecisionLogger:
    def log_decision(self, agent_name, prompt, response, chosen_tool):
        self.store({
            "agent": agent_name,
            "input_summary": summarize(prompt),
            "tool_chosen": chosen_tool,
            "confidence": response.confidence,
            "alternatives_considered": response.alternatives,
        })

class TraceReplayer:
    def replay(self, trace_id):
        trace = self.load_trace(trace_id)
        for span in trace.spans:
            print(f"[{span.agent}] {span.operation}: {span.input[:100]}...")
            print(f"  → {span.output[:100]}...")
            print(f"  ⏱ {span.duration_ms}ms, 💰 {span.cost}")