多Agent六大协调范式实战拆解：从蚁迹到黑板的全栈实测

# blackboard-core 范式示例（基于PyPI开源包）
from blackboard_core import Blackboard, KnowledgeSource, Controller

class BlackboardSystem:
    def __init__(self):
        self.bb = Blackboard()
        self.controller = Controller(self.bb)

    def register_agents(self):
        # 注册专家Agent——它们只关心黑板上的特定层级
        self.controller.register(
            KnowledgeSource("researcher", 
                trigger_level="task_spec",
                write_level="research_data",
                llm_callable=self._call_llm)
        )
        self.controller.register(
            KnowledgeSource("analyst",
                trigger_level="research_data", 
                write_level="analysis_result",
                llm_callable=self._call_llm)
        )
        self.controller.register(
            KnowledgeSource("writer",
                trigger_level="analysis_result",
                write_level="final_report",
                llm_callable=self._call_llm)
        )

    def run(self, task_spec: str):
        self.bb.write("task_spec", {"query": task_spec})
        # 控制器循环：监听→触发→写入→再监听
        while not self.bb.is_complete():
            self.controller.step()  # 检查谁被触发，执行推理
        return self.bb.read("final_report")

# Stigmergic协调范式核心实现
import redis
import json
import time
import math

class StigmergicEnvironment:
    """共享环境——所有Agent通过读写此环境间接协调"""
    def __init__(self, redis_client: redis.Redis):
        self.r = redis_client
        self.decay_rate = 0.95  # 信息素蒸发率
        self.ttl = 3600  # 标记生存时间(秒)

    def leave_pheromone(self, agent_id: str, task_type: str, 
                        result_quality: float, coords: dict):
        """Agent完成任务后在环境中留下信息素标记"""
        key = f"pheromone:{task_type}:{coords['x']}:{coords['y']}"
        marker = {
            "agent": agent_id,
            "quality": result_quality,
            "timestamp": time.time(),
            "strength": 1.0  # 初始强度
        }
        self.r.hset(key, mapping=marker)
        self.r.expire(key, self.ttl)

    def read_pheromones(self, task_type: str, 
                        nearby_coords: dict, radius: int = 5):
        """Agent读取附近的信息素，决定是否参与同类任务"""
        pheromones = []
        for dx in range(-radius, radius + 1):
            for dy in range(-radius, radius + 1):
                key = (f"pheromone:{task_type}:"
                       f"{nearby_coords['x']+dx}:"
                       f"{nearby_coords['y']+dy}")
                data = self.r.hgetall(key)
                if data:
                    # 应用蒸发衰减
                    age = time.time() - float(data[b'timestamp'])
                    decayed = float(data[b'strength']) * (
                        self.decay_rate ** (age / 60))
                    if decayed > 0.1:  # 阈值过滤
                        pheromones.append({
                            "coords": {"x": nearby_coords['x']+dx, 
                                       "y": nearby_coords['y']+dy},
                            "strength": decayed,
                            "quality": float(data[b'quality'])
                        })
        return sorted(pheromones, key=lambda p: p["strength"], 
                       reverse=True)

class StigmergicAgent:
    def __init__(self, agent_id: str, env: StigmergicEnvironment):
        self.id = agent_id
        self.env = env

    def decide_task(self, available_tasks):
        """根据信息素浓度决策——高质量区域吸引更多Agent"""
        best_task = None
        best_score = -1
        for task in available_tasks:
            pheromones = self.env.read_pheromones(
                task["type"], task["coords"])
            # 高信息素=已有人成功完成，值得参与
            score = sum(p["strength"] * p["quality"] 
                       for p in pheromones[:3])
            if score > best_score:
                best_score = score
                best_task = task
        return best_task

# 拍卖协调范式——Contract Net Protocol实现
from dataclasses import dataclass, field
from typing import List, Optional
import asyncio

@dataclass
class Bid:
    agent_id: str
    task_id: str
    capability_score: float  # 能力匹配度 0-1
    estimated_time: float    # 预估完成时间(秒)
    current_load: float      # 当前负载 0-1
    bid_price: float = 0.0   # 竞标价（系统自动计算）

    def __post_init__(self):
        # 出价公式：能力高×负载低×速度快 = 高出价
        self.bid_price = (
            self.capability_score * 0.5 +
            (1 - self.current_load) * 0.3 +
            (1 / max(self.estimated_time, 1)) * 0.2
        )

class AuctioneerAgent:
    """拍卖师Agent——负责任务发布、收集标书、决标"""
    def __init__(self):
        self.registered_agents = {}
        self.auction_log = []

    def register(self, agent_id: str, capabilities: dict):
        self.registered_agents[agent_id] = capabilities

    async def run_auction(self, task: dict) -> Optional[str]:
        """执行一轮拍卖"""
        # 阶段1：广播任务公告(Call for Proposals)
        proposals = await self._broadcast_cfp(task)

        # 阶段2：收集标书
        if not proposals:
            return None

        # 阶段3：评标——综合出价与历史表现
        winner = self._evaluate_bids(proposals, task)

        # 阶段4：派标
        self.auction_log.append({
            "task": task["id"], "winner": winner.agent_id,
            "price": winner.bid_price
        })
        return winner.agent_id

    async def _broadcast_cfp(self, task: dict) -> List[Bid]:
        bids = []
        for agent_id, caps in self.registered_agents.items():
            # 每个Agent计算自己的出价
            cap_match = sum(1 for s in task["required_skills"] 
                          if s in caps["skills"]) / len(task["required_skills"])
            bid = Bid(
                agent_id=agent_id,
                task_id=task["id"],
                capability_score=cap_match,
                estimated_time=caps["avg_task_time"],
                current_load=caps["current_load"]
            )
            bids.append(bid)
        return bids

    def _evaluate_bids(self, bids: List[Bid], task: dict) -> Bid:
        """评标策略：最高出价者中标，但排除过载Agent"""
        valid = [b for b in bids if b.current_load < 0.9]
        return max(valid, key=lambda b: b.bid_price)

# CrewAI分层协调——三层委派模型
from crewai import Agent, Task, Crew, Process

# 第一层：规划Agent
planner = Agent(
    role="项目总监",
    goal="将复杂项目分解为可执行的子任务链",
    backstory="你是一位资深项目经理，擅长任务分解和优先级排序",
    llm="gpt-4o",
    allow_delegation=True  # 关键：允许向下委派
)

# 第二层：协调Agent（领域主管）
research_lead = Agent(
    role="研究主管",
    goal="协调研究团队完成市场和技术调研",
    backstory="你管理一个研究团队，负责分配调研任务",
    llm="gpt-4o",
    allow_delegation=True
)

writing_lead = Agent(
    role="内容主管", 
    goal="协调写作团队产出高质量报告",
    backstory="你管理一个写作团队，负责内容编排和质量把控",
    llm="gpt-4o",
    allow_delegation=True
)

# 第三层：执行Agent（工人）
researcher = Agent(
    role="初级研究员",
    goal="完成分配的具体调研任务",
    backstory="你执行调研任务并返回结构化结果",
    llm="gpt-4o-mini",  # 底层用便宜模型
    allow_delegation=False  # 底层不再委派
)

writer = Agent(
    role="初级写手",
    goal="根据大纲撰写报告章节",
    backstory="你按照要求撰写指定章节",
    llm="gpt-4o-mini",
    allow_delegation=False
)

# 组建分层Crew
crew = Crew(
    agents=[planner, research_lead, writing_lead, researcher, writer],
    tasks=[
        Task(description="调研AI Agent市场现状并撰写报告", 
             agent=planner, expected_output="完整报告")
    ],
    process=Process.hierarchical,  # 启用分层模式
    manager_llm="gpt-4o",  # 主管的推理模型
    verbose=True
)

# 群涌协调范式——OpenAI Swarm风格实现
from openai import Swarm  # pip install swarm
from dataclasses import dataclass

# 每个Agent只定义极简的转移规则
def triage_agent(context):
    """分流Agent：只做一件事——判断并转移"""
    query = context["query"]
    if "代码" in query or "bug" in query:
        return {"transfer": "coding_agent"}
    elif "文档" in query or "报告" in query:
        return {"transfer": "doc_agent"}  
    else:
        return {"transfer": "general_agent"}

def coding_agent(context):
    """编码Agent：只管写代码，完成后交给审查"""
    code = generate_code(context["query"])
    return {"result": code, "transfer": "review_agent"}

def review_agent(context):
    """审查Agent：只管Review，通过则结束，否则打回"""
    review = review_code(context["result"])
    if review["approved"]:
        return {"result": context["result"], "done": True}
    else:
        return {"transfer": "coding_agent",  # 打回重做
                "feedback": review["comments"]}

def doc_agent(context):
    """文档Agent：只管写文档"""
    doc = generate_doc(context["query"])
    return {"result": doc, "done": True}

# Swarm运行时——无中心编排器
client = Swarm()
client.register("triage", triage_agent)
client.register("coding", coding_agent)
client.register("review", review_agent)
client.register("doc", doc_agent)

# 执行：Agent自行决定链路
result = client.run(
    starting_agent="triage",
    context={"query": "写一个排序函数的Python代码并出文档"}
)
# 群涌路径：triage → coding → review → (可能回coding) → done
# 没有编排器，每个Agent自行判断"下一步去哪"

# LangGraph事件驱动协调范式
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    current_task: str
    artifacts: dict
    events: List[str]  # 事件日志

def researcher_node(state: AgentState) -> dict:
    """研究Agent——完成研究后发出'research_done'事件"""
    research = call_llm(f"Research: {state['current_task']}")
    return {
        "messages": [{"role": "researcher", "content": research}],
        "artifacts": {"research": research},
        "events": ["research_done"]  # 发出事件
    }

def analyst_node(state: AgentState) -> dict:
    """分析Agent——监听'research_done'后自动触发"""
    research = state["artifacts"]["research"]
    analysis = call_llm(f"Analyze: {research}")
    return {
        "messages": [{"role": "analyst", "content": analysis}],
        "artifacts": {"analysis": analysis},
        "events": ["analysis_done"]
    }

def reviewer_node(state: AgentState) -> dict:
    """审查Agent——监听'analysis_done'后触发"""
    analysis = state["artifacts"]["analysis"]
    review = call_llm(f"Review quality: {analysis}")
    if "REJECT" in review:
        return {
            "messages": [{"role": "reviewer", "content": "驳回，需重新研究"}],
            "events": ["revision_needed"]
        }
    return {
        "messages": [{"role": "reviewer", "content": "通过"}],
        "artifacts": {"final": analysis},
        "events": ["approved"]
    }

# 构建事件驱动的转移图
graph = StateGraph(AgentState)

# 注册Agent节点
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node) 
graph.add_node("reviewer", reviewer_node)

# 定义事件驱动的转移条件
graph.add_conditional_edges(
    "reviewer",
    lambda state: "researcher" 
        if "revision_needed" in state.get("events", []) 
        else END,
    {"researcher": "researcher", END: END}
)
graph.add_edge("researcher", "analyst")   # research_done → analyst
graph.add_edge("analyst", "reviewer")     # analysis_done → reviewer

graph.set_entry_point("researcher")
app = graph.compile()

graph TB
    subgraph "六大协调范式全景"
        direction TB

        subgraph "中心化谱系"
            BB[黑板协调<br/>Blackboard]
            AU[拍卖协调<br/>Auction]
            HI[分层协调<br/>Hierarchical]
        end

        subgraph "去中心化谱系"
            ST[蚁迹协调<br/>Stigmergy]
            SW[群涌协调<br/>Swarm]
            ED[事件驱动<br/>Event-Driven]
        end
    end

    subgraph "核心基础设施"
        RC[Redis/Valkey<br/>共享环境/信息素]
        MQ[Kafka/NATS<br/>事件总线]
        DB[PostgreSQL<br/>黑板持久化]
        LLM[LLM API<br/>推理引擎]
    end

    BB --> DB
    BB --> LLM
    ST --> RC
    AU --> DB
    HI --> LLM
    SW --> LLM
    ED --> MQ
    ED --> DB

    subgraph "应用场景"
        KB[企业知识库]
        LG[物流优化]
        RS[资源调度]
        VT[虚拟团队]
        GM[游戏/仿真]
        WF[工作流自动化]
    end

    BB -.-> KB
    ST -.-> LG
    AU -.-> RS
    HI -.-> VT
    SW -.-> GM
    ED -.-> WF

flowchart LR
    subgraph "选型决策树"
        Q1{需要全局<br/>共享状态?}
        Q2{Agent是否<br/>完全自治?}
        Q3{任务是否<br/>可竞价?}
        Q4{是否有明确<br/>组织层级?}
        Q5{是否需要<br/>因果可追溯?}
        Q6{是否需要<br/>弹性扩缩?}

        Q1 -->|是| BB_SEL[黑板范式]
        Q1 -->|否| Q2
        Q2 -->|是| Q3
        Q2 -->|否| Q4
        Q3 -->|是| AU_SEL[拍卖范式]
        Q3 -->|否| Q5
        Q4 -->|是| HI_SEL[分层范式]
        Q4 -->|否| Q6
        Q5 -->|是| ED_SEL[事件驱动范式]
        Q5 -->|否| SW_SEL[群涌范式]
        Q6 -->|是| ST_SEL[蚁迹范式]
        Q6 -->|否| SW_SEL2[群涌范式]
    end

# 混合架构：蚁迹（自组织）+ 事件驱动（可审计）
import asyncio
from redis import Redis

class HybridOrchestrator:
    def __init__(self):
        self.redis = Redis()
        self.event_subscribers = {}

    async def run(self, task: dict):
        # 第一层：蚁迹——Agent自主认领任务
        await self._publish_task_pheromone(task)

        # 等待Agent自发聚集（信息素引导）
        agents = await self._wait_for_agent_convergence(
            task, timeout=30)

        # 第二层：事件驱动——对认领的Agent编排工作流
        result = await self._event_driven_pipeline(agents, task)

        # 反馈：将结果质量写回环境，影响未来信息素
        self._update_pheromone_quality(task, result.quality)
        return result

    async def _publish_task_pheromone(self, task: dict):
        """发布任务信息素——吸引匹配的Agent"""
        self.redis.hset(
            f"task_pheromone:{task['id']}",
            mapping={
                "type": task["type"],
                "priority": task["priority"],
                "timestamp": time.time(),
                "status": "open"
            })

    async def _wait_for_agent_convergence(self, task, timeout):
        """等待Agent被信息素吸引并报到"""
        deadline = time.time() + timeout
        agents = []
        while time.time() < deadline:
            # 检查是否有Agent通过蚁迹机制报到
            reported = self.redis.lrange(
                f"task_volunteers:{task['id']}", 0, -1)
            agents.extend(reported)
            if len(agents) >= task["min_agents"]:
                break
            await asyncio.sleep(1)
        return agents

    async def _event_driven_pipeline(self, agents, task):
        """对聚集的Agent执行事件驱动流水线"""
        # 现在用精确的事件链编排，保证可审计
        pipeline = [
            {"event": "task_assigned", "handler": "planner"},
            {"event": "plan_ready", "handler": "executor"},
            {"event": "execution_done", "handler": "reviewer"},
        ]
        state = {"task": task, "agents": agents}
        for step in pipeline:
            state = await self._dispatch_event(step, state)
        return state["result"]