1
┌─────────────────────────────────────────────────────────────┐
2
│                    Multi-Agent 优势                           │
3
├─────────────────────────────────────────────────────────────┤
4
│                                                             │
5
│  1. 专家分工                                                 │
6
│     ├── 每个 Agent 专注一个领域                               │
7
│     ├── 提示词更精准、输出更稳定                              │
8
│     └── 类似人类团队的专业分工                                │
9
│                                                             │
10
│  2. 相互审核                                                 │
11
│     ├── 一个 Agent 输出，另一个 Agent 检查                    │
12
│     ├── 减少错误、提升质量                                    │
13
│     └── 类似 Code Review 机制                                │
14
│                                                             │
15
│  3. 并行处理                                                 │
16
│     ├── 多个 Agent 同时工作                                   │
17
│     ├── 缩短整体执行时间                                      │
18
│     └── 适合可并行化的子任务                                  │
19
│                                                             │
20
│  4. 灵活扩展                                                 │
21
│     ├── 新增能力只需添加新 Agent                              │
22
│     ├── 不影响现有 Agent                                     │
23
│     └── 模块化、可维护性强                                    │
24
│                                                             │
25
└─────────────────────────────────────────────────────────────┘

1
class SyncCommunication:
2
    def __init__(self, agents: dict):
3
        self.agents = agents
4

5
    def call(self, agent_name: str, message: str) -> str:
6
        """同步调用其他 Agent"""
7
        agent = self.agents[agent_name]
8
        # 阻塞等待返回
9
        response = agent.process(message)
10
        return response
11

12
# 使用示例
13
comm = SyncCommunication({
14
    "researcher": ResearcherAgent(),
15
    "writer": WriterAgent()
16
})
17

18
# 同步调用：等待 researcher 返回后再继续
19
research_result = comm.call("researcher", "研究 AI Agent 发展")
20
final_output = comm.call("writer", f"基于以下内容写文章：{research_result}")

1
import asyncio
2
from dataclasses import dataclass
3
from typing import Callable, Awaitable
4

5
@dataclass
6
class Message:
7
    sender: str
8
    receiver: str
9
    content: str
10
    callback: Callable[[str], Awaitable[None]]
11

12
class AsyncMessageQueue:
13
    def __init__(self):
14
        self.queue = asyncio.Queue()
15
        self.handlers = {}
16

17
    def register(self, agent_name: str, handler):
18
        self.handlers[agent_name] = handler
19

20
    async def send(self, message: Message):
21
        await self.queue.put(message)
22

23
    async def start_processing(self):
24
        while True:
25
            message = await self.queue.get()
26
            handler = self.handlers.get(message.receiver)
27
            if handler:
28
                result = await handler(message.content)
29
                if message.callback:
30
                    await message.callback(result)
31

32
# 使用示例
33
async def main():
34
    mq = AsyncMessageQueue()
35
    mq.register("researcher", researcher_handler)
36
    mq.register("writer", writer_handler)
37

38
    # 发送消息后立即返回，不等待结果
39
    await mq.send(Message(
40
        sender="coordinator",
41
        receiver="researcher",
42
        content="研究 AI Agent 发展",
43
        callback=lambda r: print(f"收到结果：{r}")
44
    ))

1
@dataclass
2
class AgentMessage:
3
    """标准消息格式"""
4
    # 元信息
5
    message_id: str          # 消息唯一标识
6
    timestamp: float         # 时间戳
7

8
    # 发送方信息
9
    sender: str              # 发送方 Agent ID
10
    sender_role: str         # 发送方角色
11

12
    # 接收方信息
13
    receiver: str            # 接收方 Agent ID（或 "broadcast"）
14

15
    # 消息内容
16
    message_type: str        # 消息类型：request/response/notification
17
    content: str             # 消息内容
18
    metadata: dict           # 额外元数据
19

20
    # 任务关联
21
    task_id: str             # 关联的任务 ID
22
    parent_message_id: str   # 父消息 ID（用于追踪对话链）

1
class StarArchitecture:
2
    def __init__(self, coordinator, workers: dict):
3
        self.coordinator = coordinator
4
        self.workers = workers
5

6
    def execute(self, task: str) -> str:
7
        # 1. 协调者分解任务
8
        subtasks = self.coordinator.decompose(task)
9

10
        # 2. 分发给各个 Worker
11
        results = {}
12
        for subtask in subtasks:
13
            worker_name = subtask.assigned_worker
14
            worker = self.workers[worker_name]
15
            results[worker_name] = worker.execute(subtask)
16

17
        # 3. 协调者整合结果
18
        final_result = self.coordinator.integrate(results)
19
        return final_result

1
class ChainArchitecture:
2
    def __init__(self, agents: list):
3
        self.chain = agents  # 按顺序排列的 Agent 列表
4

5
    def execute(self, input_data: str) -> str:
6
        data = input_data
7
        for agent in self.chain:
8
            data = agent.process(data)
9
            # 可选：检查是否需要提前终止
10
            if agent.should_stop():
11
                break
12
        return data
13

14
# 软件开发流水线
15
dev_pipeline = ChainArchitecture([
16
    RequirementAgent(),   # 需求分析
17
    ArchitectAgent(),     # 架构设计
18
    DeveloperAgent(),     # 代码实现
19
    QAAgent(),            # 测试验证
20
    DeployerAgent()       # 部署发布
21
])
22

23
result = dev_pipeline.execute("开发一个用户登录功能")

1
class MeshArchitecture:
2
    def __init__(self, agents: dict):
3
        self.agents = agents
4
        # 每个 Agent 持有对其他 Agent 的引用
5
        for agent in agents.values():
6
            agent.set_peers(agents)
7

8
    def broadcast(self, sender: str, message: str):
9
        """广播消息给所有 Agent"""
10
        for name, agent in self.agents.items():
11
            if name != sender:
12
                agent.receive(sender, message)
13

14
    def direct_message(self, sender: str, receiver: str, message: str):
15
        """点对点通信"""
16
        self.agents[receiver].receive(sender, message)

1
class HierarchicalAgent:
2
    def __init__(self, name: str, role: str):
3
        self.name = name
4
        self.role = role
5
        self.subordinates = []
6
        self.superior = None
7

8
    def add_subordinate(self, agent):
9
        agent.superior = self
10
        self.subordinates.append(agent)
11

12
    def delegate(self, task: str):
13
        """将任务分配给下属"""
14
        results = []
15
        for subordinate in self.subordinates:
16
            result = subordinate.execute(task)
17
            results.append(result)
18
        return self.aggregate(results)
19

20
    def escalate(self, issue: str):
21
        """向上级汇报问题"""
22
        if self.superior:
23
            return self.superior.handle_escalation(self.name, issue)

1
1. 单一职责
2
   - 每个子任务目标明确
3
   - 避免一个任务包含多个不相关目标
4

5
2. 粒度适中
6
   - 太粗：Agent 负担过重
7
   - 太细：协调开销过大
8

9
3. 边界清晰
10
   - 输入输出定义明确
11
   - 减少 Agent 间的耦合
12

13
4. 可验证性
14
   - 每个子任务有明确的完成标准
15
   - 便于质量检查

1
class TaskDecomposer:
2
    def __init__(self, llm):
3
        self.llm = llm
4

5
    def decompose(self, task: str, agents: list) -> list:
6
        """使用 LLM 分解任务"""
7
        prompt = f"""
8
        任务：{task}
9

10
        可用的 Agent 角色：
11
        {self._format_agents(agents)}
12

13
        请将任务分解为子任务，每个子任务指定：
14
        1. 子任务描述
15
        2. 负责的 Agent
16
        3. 输入要求
17
        4. 输出格式
18
        5. 前置依赖
19
        """
20

21
        response = self.llm.generate(prompt)
22
        subtasks = self._parse_response(response)
23
        return subtasks
24

25
    def _format_agents(self, agents):
26
        return "\n".join([
27
            f"- {a.name}: {a.capabilities}"
28
            for a in agents
29
        ])

1
class ResultIntegrator:
2
    def sequential_merge(self, results: list) -> str:
3
        """顺序合并"""
4
        return "\n\n".join(results)
5

6
    def voting_decision(self, results: list, voters: list) -> str:
7
        """投票决策"""
8
        votes = {}
9
        for voter in voters:
10
            chosen = voter.vote(results)
11
            votes[chosen] = votes.get(chosen, 0) + 1
12
        return max(votes, key=votes.get)
13

14
    def weighted_merge(self, results: list, weights: dict) -> str:
15
        """权重融合"""
16
        total_weight = sum(weights.values())
17
        weighted_result = ""
18
        for agent_name, result in results.items():
19
            weight = weights.get(agent_name, 1)
20
            # 根据权重决定内容的比重
21
            weighted_result += result * weight
22
        return weighted_result
23

24
    def llm_synthesis(self, results: dict, llm) -> str:
25
        """LLM 综合整合"""
26
        prompt = f"""
27
        多个 Agent 的输出结果如下：
28
        {self._format_results(results)}
29

30
        请综合以上信息，生成一个完整、连贯的最终结果。
31
        """
32
        return llm.generate(prompt)

1
class PriorityResolver:
2
    def __init__(self, priorities: dict):
3
        self.priorities = priorities  # Agent 优先级配置
4

5
    def resolve(self, conflict):
6
        """按优先级解决冲突"""
7
        agent_a, agent_b = conflict.parties
8
        priority_a = self.priorities.get(agent_a, 0)
9
        priority_b = self.priorities.get(agent_b, 0)
10

11
        winner = agent_a if priority_a >= priority_b else agent_b
12
        return conflict.proposals[winner]

1
class VotingResolver:
2
    def resolve(self, proposals: dict, voters: list) -> str:
3
        """投票解决冲突"""
4
        votes = {k: 0 for k in proposals.keys()}
5

6
        for voter in voters:
7
            # 每个 Agent 根据自己的判断投票
8
            choice = voter.evaluate(proposals)
9
            votes[choice] += 1
10

11
        return max(votes, key=votes.get)

1
class DebateResolver:
2
    def __init__(self, judge_agent):
3
        self.judge = judge_agent
4

5
    def resolve(self, proposals: dict, rounds: int = 3):
6
        """辩论解决冲突"""
7
        arguments = {k: [] for k in proposals.keys()}
8

9
        for _ in range(rounds):
10
            for agent_name, proposal in proposals.items():
11
                # 每个 Agent 提出论据
12
                arg = agent.generate_argument(
13
                    proposal,
14
                    arguments  # 可以参考其他 Agent 的论据
15
                )
16
                arguments[agent_name].append(arg)
17

18
        # 裁决者做出最终决定
19
        decision = self.judge.decide(arguments)
20
        return decision

1
┌─────────────────────────────────────────────────────────────┐
2
│                    共识协议流程                               │
3
├─────────────────────────────────────────────────────────────┤
4
│                                                             │
5
│  1. 提案阶段                                                 │
6
│     Agent A 提出方案 → 广播给所有 Agent                       │
7
│                                                             │
8
│  2. 预投票阶段                                               │
9
│     各 Agent 表达初步态度（赞成/反对/弃权）                    │
10
│                                                             │
11
│  3. 讨论阶段                                                 │
12
│     反对者提出修改建议 → 提案者修改方案                        │
13
│                                                             │
14
│  4. 正式投票                                                 │
15
│     达到多数同意（如 2/3）→ 方案通过                          │
16
│                                                             │
17
│  5. 执行阶段                                                 │
18
│     所有 Agent 按共识结果执行                                 │
19
│                                                             │
20
└─────────────────────────────────────────────────────────────┘

1
from crewai import Agent, Task, Crew, Process
2

3
# 定义 Agent 角色
4
researcher = Agent(
5
    role="研究员",
6
    goal="深入研究主题，收集全面信息",
7
    backstory="你是一位经验丰富的研究员，擅长信息收集和分析",
8
    allow_delegation=False,
9
    verbose=True
10
)
11

12
writer = Agent(
13
    role="作家",
14
    goal="撰写高质量、引人入胜的文章",
15
    backstory="你是一位专业作家，擅长将复杂信息转化为易懂的文章",
16
    allow_delegation=True,
17
    verbose=True
18
)
19

20
editor = Agent(
21
    role="编辑",
22
    goal="审核文章质量，确保准确性和可读性",
23
    backstory="你是一位严谨的编辑，注重细节和质量",
24
    allow_delegation=False,
25
    verbose=True
26
)
27

28
# 定义任务
29
research_task = Task(
30
    description="研究 AI Agent 的最新发展趋势",
31
    expected_output="详细的研究报告",
32
    agent=researcher
33
)
34

35
write_task = Task(
36
    description="基于研究报告撰写一篇技术文章",
37
    expected_output="结构完整的技术文章",
38
    agent=writer,
39
    context=[research_task]  # 依赖研究任务
40
)
41

42
edit_task = Task(
43
    description="审核并改进文章",
44
    expected_output="最终版本的技术文章",
45
    agent=editor,
46
    context=[write_task]  # 依赖写作任务
47
)
48

49
# 组建团队
50
crew = Crew(
51
    agents=[researcher, writer, editor],
52
    tasks=[research_task, write_task, edit_task],
53
    process=Process.sequential  # 顺序执行
54
)
55

56
# 执行
57
result = crew.kickoff()

1
from langgraph.graph import StateGraph, END
2
from typing import TypedDict, Annotated
3
import operator
4

5
class State(TypedDict):
6
    messages: Annotated[list, operator.add]
7
    next_agent: str
8

9
def researcher_node(state: State) -> dict:
10
    # 研究员处理
11
    result = researcher_agent.invoke(state["messages"])
12
    return {"messages": [result], "next_agent": "writer"}
13

14
def writer_node(state: State) -> dict:
15
    # 作家处理
16
    result = writer_agent.invoke(state["messages"])
17
    return {"messages": [result], "next_agent": "reviewer"}
18

19
def reviewer_node(state: State) -> dict:
20
    # 审核员处理
21
    result = reviewer_agent.invoke(state["messages"])
22

23
    # 判断是否需要修改
24
    if needs_revision(result):
25
        return {"messages": [result], "next_agent": "writer"}
26
    return {"messages": [result], "next_agent": END}
27

28
# 构建状态图
29
workflow = StateGraph(State)
30
workflow.add_node("researcher", researcher_node)
31
workflow.add_node("writer", writer_node)
32
workflow.add_node("reviewer", reviewer_node)
33

34
# 定义边
35
workflow.set_entry_point("researcher")
36
workflow.add_edge("researcher", "writer")
37
workflow.add_edge("writer", "reviewer")
38
workflow.add_conditional_edges(
39
    "reviewer",
40
    lambda s: s["next_agent"],
41
    {"writer": "writer", END: END}
42
)
43

44
# 编译并执行
45
app = workflow.compile()
46
result = app.invoke({"messages": ["写一篇关于 AI Agent 的文章"], "next_agent": "researcher"})

1
┌─────────────────────────────────────────────────────────────┐
2
│                    A2A 协议核心概念                           │
3
├─────────────────────────────────────────────────────────────┤
4
│                                                             │
5
│  1. Agent Card（Agent 名片）                                 │
6
│     - 描述 Agent 的能力、接口、认证方式                        │
7
│     - 类似 OpenAPI 规范，但针对 Agent                         │
8
│                                                             │
9
│  2. Message（消息）                                          │
10
│     - 标准化的消息格式                                        │
11
│     - 支持文本、文件、结构化数据                               │
12
│                                                             │
13
│  3. Task（任务）                                             │
14
│     - 任务状态管理                                           │
15
│     - 状态：pending → running → completed/failed             │
16
│                                                             │
17
│  4. Artifact（产物）                                         │
18
│     - 任务执行产生的输出                                      │
19
│     - 支持多种格式                                           │
20
│                                                             │
21
└─────────────────────────────────────────────────────────────┘

1
{
2
  "jsonrpc": "2.0",
3
  "method": "tasks/send",
4
  "params": {
5
    "task": {
6
      "id": "task-001",
7
      "status": {
8
        "state": "running"
9
      },
10
      "history": [
11
        {
12
          "role": "user",
13
          "parts": [{ "text": "分析最近的销售数据" }]
14
        }
15
      ]
16
    }
17
  }
18
}

1
from a2a import A2AClient, AgentCard
2

3
# 定义 Agent 名片
4
my_agent_card = AgentCard(
5
    name="SalesAnalyzer",
6
    description="分析销售数据的专家 Agent",
7
    capabilities=["data_analysis", "report_generation"],
8
    endpoints={
9
        "task": "https://my-agent.example.com/a2a/task"
10
    }
11
)
12

13
# 创建 A2A 客户端
14
client = A2AClient()
15

16
# 发现其他 Agent
17
other_agent = client.discover("ReportGenerator")
18

19
# 发送任务请求
20
response = client.send_task(
21
    agent=other_agent,
22
    task={
23
        "type": "generate_report",
24
        "data": sales_data,
25
        "format": "pdf"
26
    }
27
)

1
class MemoryStream:
2
    def __init__(self):
3
        self.memories = []  # 按时间顺序存储
4

5
    def add(self, event: str, importance: float):
6
        """添加新记忆"""
7
        memory = {
8
            "content": event,
9
            "timestamp": time.now(),
10
            "importance": importance,
11
            "access_count": 0
12
        }
13
        self.memories.append(memory)
14

15
    def retrieve(self, query: str, top_k: int = 10):
16
        """检索相关记忆"""
17
        scores = []
18
        for mem in self.memories:
19
            # 综合评分：相关性 + 时近性 + 重要性
20
            recency = self._recency_score(mem)
21
            importance = mem["importance"]
22
            relevance = self._relevance_score(query, mem)
23
            score = 0.5 * relevance + 0.3 * recency + 0.2 * importance
24
            scores.append((mem, score))
25

26
        return sorted(scores, key=lambda x: -x[1])[:top_k]

1
def reflect(agent, memory_stream):
2
    """生成反思"""
3
    # 1. 检索最近的记忆
4
    recent_memories = memory_stream.retrieve_recent(n=100)
5

6
    # 2. 提取关键问题
7
    questions = llm.generate_questions(recent_memories)
8

9
    # 3. 对每个问题生成洞察
10
    insights = []
11
    for q in questions:
12
        relevant = memory_stream.retrieve(q)
13
        insight = llm.synthesize(q, relevant)
14
        insights.append(insight)
15

16
    # 4. 存储反思结果
17
    for insight in insights:
18
        memory_stream.add(insight, importance=0.9)

1
规划层级：
2
┌─────────────────────────────────────┐
3
│  长期计划（年/月）                     │
4
│  "成为一名受尊敬的教授"                │
5
├─────────────────────────────────────┤
6
│  中期计划（周/天）                     │
7
│  "完成论文初稿"                       │
8
├─────────────────────────────────────┤
9
│  短期计划（小时）                      │
10
│  "下午 2 点去图书馆查资料"              │
11
├─────────────────────────────────────┤
12
│  即时行动                             │
13
│  "走向图书馆"                         │
14
└─────────────────────────────────────┘

1
1. 记忆架构的重要性
2
   - 单一记忆不足以支撑复杂行为
3
   - 需要分层记忆系统
4

5
2. 反思提升智能
6
   - 不只是反应式执行
7
   - 定期总结形成更高层次认知
8

9
3. 规划需要层级化
10
   - 从长期目标到即时行动
11
   - 动态调整计划
12

13
4. 社交行为的涌现
14
   - 简单规则可以产生复杂行为
15
   - Agent 间交互是智能的重要来源

1
1. 最小化 Agent 数量
2
   - 从最少 Agent 开始
3
   - 只在必要时增加
4

5
2. 明确职责边界
6
   - 每个 Agent 只做一件事
7
   - 避免职责重叠
8

9
3. 标准化通信
10
   - 使用统一的消息格式
11
   - 定义清晰的接口
12

13
4. 可观测性
14
   - 记录所有 Agent 间通信
15
   - 支持执行链追踪
16

17
5. 容错设计
18
   - 单个 Agent 失败不影响整体
19
   - 设置超时和重试机制