npx skills add https://github.com/daffy0208/ai-dev-standards --skill 'Multi-Agent Architect'设计由多个专业智能体协作解决复杂问题的系统。
将复杂任务分配给专业智能体,每个智能体在其领域内都是专家,通过清晰的通信模式进行协调。
使用场景:多步骤工作流,每个智能体在前一个的基础上工作
User Query → Researcher → Analyst → Writer → Editor → Output
示例:研究报告生成
优点:依赖关系清晰,易于调试 缺点:顺序执行(无并行化),存在瓶颈
使用场景:复杂任务分解为并行子任务
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
Manager Agent
/ | \
Worker 1 Worker 2 Worker 3
(Search) (Analyze) (Summarize)
\ | /
Aggregator Agent
示例:跨竞争对手的市场研究
优点:并行化,专业化 缺点:管理者复杂度高,协调开销大
使用场景:多重视角提高质量
Coder ↔ Reviewer ↔ Tester
↓ ↓ ↓
Consensus
示例:带评审的代码生成
优点:通过评审保证质量,自我纠正 缺点:可能无法收敛,成本高(多次 LLM 调用)
使用场景:多个智能体独立探索解决方案空间
Agent 1 → Candidate Solution 1
Agent 2 → Candidate Solution 2
Agent 3 → Candidate Solution 3
↓
Selector (pick best)
示例:创意头脑风暴
优点:探索性,创造性 缺点:成本高(N 个智能体),可能产生相似的解决方案
shared_state = {
"research_findings": [],
"current_task": "analyze_competitors",
"decisions": []
}
# All agents read/write to shared state
researcher.execute(shared_state)
analyst.execute(shared_state)
优点:简单,所有智能体都能看到完整上下文 缺点:竞态条件,难以调试谁更改了什么
# Agent A sends message to Agent B
message = {
"from": "researcher",
"to": "analyst",
"content": research_findings,
"metadata": {"confidence": 0.9}
}
message_queue.send(message)
优点:通信流程清晰,可追溯 缺点:实现更复杂
# Agents subscribe to events
event_bus.subscribe("research_complete", analyst.on_research_complete)
event_bus.subscribe("analysis_complete", writer.on_analysis_complete)
# Agent publishes event when done
event_bus.publish("research_complete", research_data)
优点:松耦合,可扩展 缺点:更难跟踪执行流程
预定义序列,无动态决策
workflow = [
("researcher", gather_info),
("analyst", analyze_data),
("writer", create_report)
]
for agent_name, task in workflow:
result = agents[agent_name].execute(task, context)
context.update(result)
使用场景:可预测的任务,清晰的依赖关系
管理者根据上下文决定下一个智能体
class ManagerAgent:
def route_task(self, task, context):
if requires_technical_expertise(task):
return tech_specialist
elif requires_creative_input(task):
return creative_agent
else:
return generalist
使用场景:任务差异显著,需要灵活性
智能体投票或达成一致
proposals = [agent.propose_solution(task) for agent in agents]
scores = [agent.evaluate(proposals) for agent in agents]
best = proposals[argmax(mean(scores))]
使用场景:高风险决策,质量至关重要
CrewAI 模式(基于角色的团队):
from crewai import Agent, Task, Crew
# Define specialized agents
researcher = Agent(
role="Research Specialist",
goal="Gather comprehensive information on {topic}",
backstory="Expert researcher with 10 years experience",
tools=[search_tool, scrape_tool]
)
analyst = Agent(
role="Data Analyst",
goal="Synthesize research findings into insights",
backstory="Data scientist specialized in trend analysis",
tools=[analysis_tool]
)
writer = Agent(
role="Technical Writer",
goal="Create clear, compelling reports",
backstory="Professional writer with technical expertise",
tools=[writing_tool]
)
# Define tasks
research_task = Task(
description="Research {topic} thoroughly",
agent=researcher,
expected_output="Comprehensive research findings with sources"
)
analysis_task = Task(
description="Analyze research findings for key insights",
agent=analyst,
context=[research_task], # Depends on research_task
expected_output="List of key insights and trends"
)
writing_task = Task(
description="Write executive summary based on analysis",
agent=writer,
context=[research_task, analysis_task],
expected_output="500-word executive summary"
)
# Create crew and execute
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
verbose=True
)
result = crew.kickoff(inputs={"topic": "AI market trends"})
LangGraph 模式(状态机):
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
input: str
research: str
analysis: str
output: str
def research_node(state):
research = researcher_agent.run(state["input"])
return {"research": research}
def analysis_node(state):
analysis = analyst_agent.run(state["research"])
return {"analysis": analysis}
def writing_node(state):
output = writer_agent.run(state["analysis"])
return {"output": output}
# Build graph
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("writing", writing_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", "writing")
workflow.add_edge("writing", END)
app = workflow.compile()
# Execute
result = app.invoke({"input": "Analyze AI market trends"})
每个智能体应具有特定的专业知识和职责
智能体越多 = 协调开销越大。从简单开始。
智能体应可重启而无副作用
为智能体故障设计(重试、回退、跳过)
记录智能体决策,跟踪执行流程
跟踪每个智能体的令牌使用情况,优化昂贵的调用
❌ 智能体过多 → 从 2-3 个开始,仅在需要时添加 ❌ 职责不明确 → 定义明确的角色 ❌ 无故障处理 → 一个智能体故障导致整个系统崩溃 ❌ 同步瓶颈 → 并行化独立的智能体 ❌ 忽略成本 → N 个智能体 = N× LLM 调用 ❌ 过度设计 → 单个智能体通常就足够了
Task Complexity?
│
├─ Simple, linear → Single Agent
│
├─ Complex, requires specialization?
│ │
│ ├─ Sequential steps → Pipeline Pattern
│ ├─ Parallel subtasks → Hierarchical Pattern
│ ├─ Need review → Peer Collaboration
│ └─ Explore solutions → Swarm Pattern
│
└─ Uncertain → Start with Single Agent, refactor to Multi if needed
# Track agent execution
class TrackedAgent(Agent):
def execute(self, task, context):
start = time.time()
logger.info(f"{self.name} starting: {task}")
result = super().execute(task, context)
duration = time.time() - start
logger.info(f"{self.name} completed in {duration}s")
metrics.record({
"agent": self.name,
"task": task,
"duration": duration,
"tokens": result.token_count,
"cost": result.cost
})
return result
关键指标:
相关技能:
rag-implementer - 用于基于知识的智能体knowledge-graph-builder - 用于智能体知识库api-designer - 用于智能体通信 API相关模式:
META/DECISION-FRAMEWORK.md - 框架选择(CrewAI vs LangGraph)STANDARDS/architecture-patterns/multi-agent-pattern.md - 智能体架构(创建时)相关操作手册:
PLAYBOOKS/deploy-multi-agent-system.md - 部署指南(创建时)PLAYBOOKS/debug-agent-workflows.md - 调试流程(创建时)每周安装数
0
代码仓库
GitHub 星标数
18
首次出现
Jan 1, 1970
安全审计
Design systems where multiple specialized agents collaborate to solve complex problems.
Divide complex tasks among specialized agents , each expert in their domain, coordinated through clear communication patterns.
Use : Multi-step workflow where each agent builds on previous
User Query → Researcher → Analyst → Writer → Editor → Output
Example : Research report generation
Pros : Clear dependencies, easy to debug Cons : Sequential (no parallelization), bottlenecks
Use : Complex task broken into parallel subtasks
Manager Agent
/ | \
Worker 1 Worker 2 Worker 3
(Search) (Analyze) (Summarize)
\ | /
Aggregator Agent
Example : Market research across competitors
Pros : Parallelization, specialization Cons : Manager complexity, coordination overhead
Use : Multiple perspectives improve quality
Coder ↔ Reviewer ↔ Tester
↓ ↓ ↓
Consensus
Example : Code generation with review
Pros : Quality through review, self-correction Cons : May not converge, expensive (multiple LLM calls)
Use : Many agents explore solution space independently
Agent 1 → Candidate Solution 1
Agent 2 → Candidate Solution 2
Agent 3 → Candidate Solution 3
↓
Selector (pick best)
Example : Creative brainstorming
Pros : Exploration, creativity Cons : Cost (N agents), may produce similar solutions
shared_state = {
"research_findings": [],
"current_task": "analyze_competitors",
"decisions": []
}
# All agents read/write to shared state
researcher.execute(shared_state)
analyst.execute(shared_state)
Pros : Simple, all agents see full context Cons : Race conditions, hard to debug who changed what
# Agent A sends message to Agent B
message = {
"from": "researcher",
"to": "analyst",
"content": research_findings,
"metadata": {"confidence": 0.9}
}
message_queue.send(message)
Pros : Clear communication flow, traceable Cons : More complex to implement
# Agents subscribe to events
event_bus.subscribe("research_complete", analyst.on_research_complete)
event_bus.subscribe("analysis_complete", writer.on_analysis_complete)
# Agent publishes event when done
event_bus.publish("research_complete", research_data)
Pros : Loose coupling, scalable Cons : Harder to follow execution flow
Predefined sequence, no dynamic decisions
workflow = [
("researcher", gather_info),
("analyst", analyze_data),
("writer", create_report)
]
for agent_name, task in workflow:
result = agents[agent_name].execute(task, context)
context.update(result)
Use : Predictable tasks, clear dependencies
Manager decides next agent based on context
class ManagerAgent:
def route_task(self, task, context):
if requires_technical_expertise(task):
return tech_specialist
elif requires_creative_input(task):
return creative_agent
else:
return generalist
Use : Tasks vary significantly, need flexibility
Agents vote or reach agreement
proposals = [agent.propose_solution(task) for agent in agents]
scores = [agent.evaluate(proposals) for agent in agents]
best = proposals[argmax(mean(scores))]
Use : High-stakes decisions, quality critical
CrewAI Pattern (Role-based teams):
from crewai import Agent, Task, Crew
# Define specialized agents
researcher = Agent(
role="Research Specialist",
goal="Gather comprehensive information on {topic}",
backstory="Expert researcher with 10 years experience",
tools=[search_tool, scrape_tool]
)
analyst = Agent(
role="Data Analyst",
goal="Synthesize research findings into insights",
backstory="Data scientist specialized in trend analysis",
tools=[analysis_tool]
)
writer = Agent(
role="Technical Writer",
goal="Create clear, compelling reports",
backstory="Professional writer with technical expertise",
tools=[writing_tool]
)
# Define tasks
research_task = Task(
description="Research {topic} thoroughly",
agent=researcher,
expected_output="Comprehensive research findings with sources"
)
analysis_task = Task(
description="Analyze research findings for key insights",
agent=analyst,
context=[research_task], # Depends on research_task
expected_output="List of key insights and trends"
)
writing_task = Task(
description="Write executive summary based on analysis",
agent=writer,
context=[research_task, analysis_task],
expected_output="500-word executive summary"
)
# Create crew and execute
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
verbose=True
)
result = crew.kickoff(inputs={"topic": "AI market trends"})
LangGraph Pattern (State machines):
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
input: str
research: str
analysis: str
output: str
def research_node(state):
research = researcher_agent.run(state["input"])
return {"research": research}
def analysis_node(state):
analysis = analyst_agent.run(state["research"])
return {"analysis": analysis}
def writing_node(state):
output = writer_agent.run(state["analysis"])
return {"output": output}
# Build graph
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("writing", writing_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", "writing")
workflow.add_edge("writing", END)
app = workflow.compile()
# Execute
result = app.invoke({"input": "Analyze AI market trends"})
Each agent should have specific expertise and responsibilities
More agents = more coordination overhead. Start simple.
Agents should be restartable without side effects
Design for agent failures (retry, fallback, skip)
Log agent decisions, trace execution flow
Track token usage per agent, optimize expensive calls
❌ Too many agents → Start with 2-3, add only if needed ❌ Unclear responsibilities → Define explicit roles ❌ No failure handling → One agent failure breaks entire system ❌ Synchronous bottlenecks → Parallelize independent agents ❌ Ignoring costs → N agents = N× LLM calls ❌ Over-engineering → Single agent often sufficient
Task Complexity?
│
├─ Simple, linear → Single Agent
│
├─ Complex, requires specialization?
│ │
│ ├─ Sequential steps → Pipeline Pattern
│ ├─ Parallel subtasks → Hierarchical Pattern
│ ├─ Need review → Peer Collaboration
│ └─ Explore solutions → Swarm Pattern
│
└─ Uncertain → Start with Single Agent, refactor to Multi if needed
# Track agent execution
class TrackedAgent(Agent):
def execute(self, task, context):
start = time.time()
logger.info(f"{self.name} starting: {task}")
result = super().execute(task, context)
duration = time.time() - start
logger.info(f"{self.name} completed in {duration}s")
metrics.record({
"agent": self.name,
"task": task,
"duration": duration,
"tokens": result.token_count,
"cost": result.cost
})
return result
Key Metrics :
Related Skills :
rag-implementer - For knowledge-grounded agentsknowledge-graph-builder - For agent knowledge basesapi-designer - For agent communication APIsRelated Patterns :
META/DECISION-FRAMEWORK.md - Framework selection (CrewAI vs LangGraph)STANDARDS/architecture-patterns/multi-agent-pattern.md - Agent architectures (when created)Related Playbooks :
PLAYBOOKS/deploy-multi-agent-system.md - Deployment guide (when created)PLAYBOOKS/debug-agent-workflows.md - Debugging procedures (when created)Weekly Installs
0
Repository
GitHub Stars
18
First Seen
Jan 1, 1970
Security Audits
AI Elements:基于shadcn/ui的AI原生应用组件库,快速构建对话界面
60,400 周安装