multi-agent-orchestration by qodex-ai/ai-agent-skills
npx skills add https://github.com/qodex-ai/ai-agent-skills --skill multi-agent-orchestration设计和编排复杂的多智能体系统,让具备专业知识的智能体协同工作以解决复杂问题,融合不同的专业知识和视角。
通过示例和实用工具开始多智能体实现:
示例 : 查看 examples/ 目录获取完整实现:
orchestration_patterns.py - 顺序、并行、分层和共识编排framework_implementations.py - CrewAI、AutoGen、LangGraph 和 Swarm 的模板实用工具 : 查看 scripts/ 目录获取辅助模块:
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
agent_communication.py - 消息代理、共享内存和通信协议workflow_management.py - 工作流执行、优化和监控benchmarking.py - 团队性能和智能体有效性指标多智能体系统将复杂问题分解为专门的子任务,将每个子任务分配给具有相关专业知识的智能体,然后协调它们的工作以实现统一目标。
User Request
↓
Orchestrator
├→ Agent 1 (Specialist) → Task 1
├→ Agent 2 (Specialist) → Task 2
├→ Agent 3 (Specialist) → Task 3
↓
Result Aggregator
↓
Final Response
智能体由以下要素定义:
Coordinator Agent
├→ Market Analyst Agent
│ ├ Tools: Market data API, financial news
│ └ Task: Analyze market conditions
├→ Financial Analyst Agent
│ ├ Tools: Financial statements, ratio calculations
│ └ Task: Analyze company financials
├→ Risk Manager Agent
│ ├ Tools: Risk models, scenario analysis
│ └ Task: Assess investment risks
└→ Report Writer Agent
├ Tools: Document generation
└ Task: Synthesize findings into report
Case Manager Agent (Coordinator)
├→ Contract Analyzer Agent
│ └ Task: Review contract terms
├→ Precedent Research Agent
│ └ Task: Find relevant case law
├→ Risk Assessor Agent
│ └ Task: Identify legal risks
└→ Document Drafter Agent
└ Task: Prepare legal documents
Support Coordinator
├→ Issue Classifier Agent
│ └ Task: Categorize customer issue
├→ Knowledge Base Agent
│ └ Task: Find relevant documentation
├→ Escalation Agent
│ └ Task: Determine if human escalation needed
└→ Solution Synthesizer Agent
└ Task: Prepare comprehensive response
最适合 : 具有清晰角色和分层结构的团队
from crewai import Agent, Task, Crew
# Define agents
analyst = Agent(
role="Financial Analyst",
goal="Analyze financial data and provide insights",
backstory="Expert in financial markets with 10+ years experience"
)
researcher = Agent(
role="Market Researcher",
goal="Research market trends and competition",
backstory="Data-driven researcher specializing in market analysis"
)
# Define tasks
analysis_task = Task(
description="Analyze Q3 financial results for {company}",
agent=analyst,
tools=[financial_tool, data_tool]
)
research_task = Task(
description="Research competitive landscape in {market}",
agent=researcher,
tools=[web_search_tool, industry_data_tool]
)
# Create crew and execute
crew = Crew(
agents=[analyst, researcher],
tasks=[analysis_task, research_task],
process=Process.sequential
)
result = crew.kickoff(inputs={"company": "TechCorp", "market": "AI"})
最适合 : 复杂的多轮对话和协商
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# Define agents
analyst = AssistantAgent(
name="analyst",
system_message="You are a financial analyst..."
)
researcher = AssistantAgent(
name="researcher",
system_message="You are a market researcher..."
)
# Create group chat
groupchat = GroupChat(
agents=[analyst, researcher],
messages=[],
max_round=10,
speaker_selection_method="auto"
)
# Manage group conversation
manager = GroupChatManager(groupchat=groupchat)
# User proxy to initiate conversation
user = UserProxyAgent(name="user")
# Have conversation
user.initiate_chat(
manager,
message="Analyze if Company X should invest in Y market"
)
最适合 : 具有状态管理的复杂工作流
from langgraph.graph import Graph, StateGraph
from langgraph.prebuilt import create_agent_executor
# Define state
class AgentState:
research_findings: str
analysis: str
recommendations: str
# Create graph
graph = StateGraph(AgentState)
# Add nodes for each agent
graph.add_node("researcher", research_agent)
graph.add_node("analyst", analyst_agent)
graph.add_node("writer", writer_agent)
# Define edges (workflow)
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
# Set entry/exit points
graph.set_entry_point("researcher")
graph.set_finish_point("writer")
# Compile and run
workflow = graph.compile()
result = workflow.invoke({"topic": "AI trends"})
最适合 : 简单的智能体交接和对话工作流
from swarm import Agent, Swarm
# Define agents
triage_agent = Agent(
name="Triage Agent",
instructions="Determine which specialist to route the customer to"
)
billing_agent = Agent(
name="Billing Specialist",
instructions="Handle billing and payment questions"
)
technical_agent = Agent(
name="Technical Support",
instructions="Handle technical issues"
)
# Define handoff functions
def route_to_billing(reason: str):
return billing_agent
def route_to_technical(reason: str):
return technical_agent
# Add tools to triage agent
triage_agent.functions = [route_to_billing, route_to_technical]
# Execute swarm
client = Swarm()
response = client.run(
agent=triage_agent,
messages=[{"role": "user", "content": "I have a billing question"}]
)
智能体按顺序执行任务,每个任务都基于前一个结果:
# Task 1: Research
research_output = research_agent.work("Analyze AI market trends")
# Task 2: Analysis (uses research output)
analysis = analyst_agent.work(f"Analyze these findings: {research_output}")
# Task 3: Report (uses analysis)
report = writer_agent.work(f"Write report on: {analysis}")
何时使用 : 步骤有依赖关系,每个步骤都基于前一个步骤
多个智能体同时工作,结果合并:
import asyncio
async def parallel_teams():
# All agents work in parallel
market_task = market_agent.work_async("Analyze market")
technical_task = tech_agent.work_async("Analyze technology")
user_task = user_agent.work_async("Analyze user needs")
# Wait for all to complete
market_results, tech_results, user_results = await asyncio.gather(
market_task, technical_task, user_task
)
# Synthesize results
return synthesize(market_results, tech_results, user_results)
何时使用 : 独立分析、需要快速结果、希望多样化
管理者智能体协调专家:
manager_agent.orchestrate({
"market_analysis": {
"agents": [competitor_analyst, trend_analyst],
"task": "Comprehensive market analysis"
},
"technical_evaluation": {
"agents": [architecture_agent, security_agent],
"task": "Technical feasibility assessment"
},
"synthesis": {
"agents": [strategy_agent],
"task": "Create strategic recommendations"
}
})
何时使用 : 清晰的分层结构、不同的团队、复杂的协调
多个智能体讨论并达成共识:
agents = [bull_agent, bear_agent, neutral_agent]
question = "Should we invest in this startup?"
# Debate round 1
arguments = {agent: agent.argue(question) for agent in agents}
# Debate round 2 (respond to others)
counter_arguments = {
agent: agent.respond(arguments) for agent in agents
}
# Reach consensus
consensus = mediator_agent.synthesize_consensus(counter_arguments)
何时使用 : 复杂决策、需要多视角、风险评估
智能体直接相互传递消息:
agent_a.send_message(agent_b, {
"type": "request",
"action": "analyze_document",
"document": doc_content,
"context": {"deadline": "urgent"}
})
智能体使用共享工具/数据库:
# Agent A writes to shared memory
shared_memory.write("findings", {"market_size": "$5B", "growth": "20%"})
# Agent B reads from shared memory
findings = shared_memory.read("findings")
中央协调器管理智能体通信:
manager.broadcast("update_all_agents", {
"new_deadline": "tomorrow",
"priority": "critical"
})
解决方案 :
解决方案 :
解决方案 :
解决方案 :
团队性能 :
智能体有效性 :
智能体自主决定角色和工作流:
# Agents negotiate roles based on task
agents = [agent1, agent2, agent3]
task = "complex financial analysis"
# Agents determine best structure
negotiated_structure = self_organize(agents, task)
# Returns optimal workflow for this task
工作流根据进度变化:
# Monitor progress
if progress < expected_rate:
# Increase resources
workflow.add_agent(specialist_agent)
elif quality < threshold:
# Increase validation
workflow.insert_review_step()
智能体从彼此的工作中学习:
# After team execution
execution_trace = crew.get_execution_trace()
# Extract learnings
learnings = extract_patterns(execution_trace)
# Update agent knowledge
for agent, learning in learnings.items():
agent.update_knowledge(learning)
每周安装量
800
代码库
GitHub 星标数
5
首次出现
2026年1月22日
安全审计
安装于
opencode726
codex725
gemini-cli724
github-copilot696
cursor670
kimi-cli641
Design and orchestrate sophisticated multi-agent systems where specialized agents collaborate to solve complex problems, combining different expertise and perspectives.
Get started with multi-agent implementations in the examples and utilities:
Examples : See examples/ directory for complete implementations:
orchestration_patterns.py - Sequential, parallel, hierarchical, and consensus orchestrationframework_implementations.py - Templates for CrewAI, AutoGen, LangGraph, and SwarmUtilities : See scripts/ directory for helper modules:
agent_communication.py - Message broker, shared memory, and communication protocolsworkflow_management.py - Workflow execution, optimization, and monitoringbenchmarking.py - Team performance and agent effectiveness metricsMulti-agent systems decompose complex problems into specialized sub-tasks, assigning each to an agent with relevant expertise, then coordinating their work toward a unified goal.
User Request
↓
Orchestrator
├→ Agent 1 (Specialist) → Task 1
├→ Agent 2 (Specialist) → Task 2
├→ Agent 3 (Specialist) → Task 3
↓
Result Aggregator
↓
Final Response
An agent is defined by:
Coordinator Agent
├→ Market Analyst Agent
│ ├ Tools: Market data API, financial news
│ └ Task: Analyze market conditions
├→ Financial Analyst Agent
│ ├ Tools: Financial statements, ratio calculations
│ └ Task: Analyze company financials
├→ Risk Manager Agent
│ ├ Tools: Risk models, scenario analysis
│ └ Task: Assess investment risks
└→ Report Writer Agent
├ Tools: Document generation
└ Task: Synthesize findings into report
Case Manager Agent (Coordinator)
├→ Contract Analyzer Agent
│ └ Task: Review contract terms
├→ Precedent Research Agent
│ └ Task: Find relevant case law
├→ Risk Assessor Agent
│ └ Task: Identify legal risks
└→ Document Drafter Agent
└ Task: Prepare legal documents
Support Coordinator
├→ Issue Classifier Agent
│ └ Task: Categorize customer issue
├→ Knowledge Base Agent
│ └ Task: Find relevant documentation
├→ Escalation Agent
│ └ Task: Determine if human escalation needed
└→ Solution Synthesizer Agent
└ Task: Prepare comprehensive response
Best For : Teams with clear roles and hierarchical structure
from crewai import Agent, Task, Crew
# Define agents
analyst = Agent(
role="Financial Analyst",
goal="Analyze financial data and provide insights",
backstory="Expert in financial markets with 10+ years experience"
)
researcher = Agent(
role="Market Researcher",
goal="Research market trends and competition",
backstory="Data-driven researcher specializing in market analysis"
)
# Define tasks
analysis_task = Task(
description="Analyze Q3 financial results for {company}",
agent=analyst,
tools=[financial_tool, data_tool]
)
research_task = Task(
description="Research competitive landscape in {market}",
agent=researcher,
tools=[web_search_tool, industry_data_tool]
)
# Create crew and execute
crew = Crew(
agents=[analyst, researcher],
tasks=[analysis_task, research_task],
process=Process.sequential
)
result = crew.kickoff(inputs={"company": "TechCorp", "market": "AI"})
Best For : Complex multi-turn conversations and negotiations
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# Define agents
analyst = AssistantAgent(
name="analyst",
system_message="You are a financial analyst..."
)
researcher = AssistantAgent(
name="researcher",
system_message="You are a market researcher..."
)
# Create group chat
groupchat = GroupChat(
agents=[analyst, researcher],
messages=[],
max_round=10,
speaker_selection_method="auto"
)
# Manage group conversation
manager = GroupChatManager(groupchat=groupchat)
# User proxy to initiate conversation
user = UserProxyAgent(name="user")
# Have conversation
user.initiate_chat(
manager,
message="Analyze if Company X should invest in Y market"
)
Best For : Complex workflows with state management
from langgraph.graph import Graph, StateGraph
from langgraph.prebuilt import create_agent_executor
# Define state
class AgentState:
research_findings: str
analysis: str
recommendations: str
# Create graph
graph = StateGraph(AgentState)
# Add nodes for each agent
graph.add_node("researcher", research_agent)
graph.add_node("analyst", analyst_agent)
graph.add_node("writer", writer_agent)
# Define edges (workflow)
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
# Set entry/exit points
graph.set_entry_point("researcher")
graph.set_finish_point("writer")
# Compile and run
workflow = graph.compile()
result = workflow.invoke({"topic": "AI trends"})
Best For : Simple agent handoffs and conversational workflows
from swarm import Agent, Swarm
# Define agents
triage_agent = Agent(
name="Triage Agent",
instructions="Determine which specialist to route the customer to"
)
billing_agent = Agent(
name="Billing Specialist",
instructions="Handle billing and payment questions"
)
technical_agent = Agent(
name="Technical Support",
instructions="Handle technical issues"
)
# Define handoff functions
def route_to_billing(reason: str):
return billing_agent
def route_to_technical(reason: str):
return technical_agent
# Add tools to triage agent
triage_agent.functions = [route_to_billing, route_to_technical]
# Execute swarm
client = Swarm()
response = client.run(
agent=triage_agent,
messages=[{"role": "user", "content": "I have a billing question"}]
)
Agents execute tasks in sequence, each building on previous results:
# Task 1: Research
research_output = research_agent.work("Analyze AI market trends")
# Task 2: Analysis (uses research output)
analysis = analyst_agent.work(f"Analyze these findings: {research_output}")
# Task 3: Report (uses analysis)
report = writer_agent.work(f"Write report on: {analysis}")
When to Use : Steps have dependencies, each builds on previous
Multiple agents work simultaneously, results combined:
import asyncio
async def parallel_teams():
# All agents work in parallel
market_task = market_agent.work_async("Analyze market")
technical_task = tech_agent.work_async("Analyze technology")
user_task = user_agent.work_async("Analyze user needs")
# Wait for all to complete
market_results, tech_results, user_results = await asyncio.gather(
market_task, technical_task, user_task
)
# Synthesize results
return synthesize(market_results, tech_results, user_results)
When to Use : Independent analyses, need quick results, want diversity
Manager agent coordinates specialists:
manager_agent.orchestrate({
"market_analysis": {
"agents": [competitor_analyst, trend_analyst],
"task": "Comprehensive market analysis"
},
"technical_evaluation": {
"agents": [architecture_agent, security_agent],
"task": "Technical feasibility assessment"
},
"synthesis": {
"agents": [strategy_agent],
"task": "Create strategic recommendations"
}
})
When to Use : Clear hierarchy, different teams, complex coordination
Multiple agents discuss and reach consensus:
agents = [bull_agent, bear_agent, neutral_agent]
question = "Should we invest in this startup?"
# Debate round 1
arguments = {agent: agent.argue(question) for agent in agents}
# Debate round 2 (respond to others)
counter_arguments = {
agent: agent.respond(arguments) for agent in agents
}
# Reach consensus
consensus = mediator_agent.synthesize_consensus(counter_arguments)
When to Use : Complex decisions, need multiple perspectives, risk assessment
Agents pass messages directly to each other:
agent_a.send_message(agent_b, {
"type": "request",
"action": "analyze_document",
"document": doc_content,
"context": {"deadline": "urgent"}
})
Agents use shared tools/databases:
# Agent A writes to shared memory
shared_memory.write("findings", {"market_size": "$5B", "growth": "20%"})
# Agent B reads from shared memory
findings = shared_memory.read("findings")
Central coordinator manages agent communication:
manager.broadcast("update_all_agents", {
"new_deadline": "tomorrow",
"priority": "critical"
})
Solutions :
Solutions :
Solutions :
Solutions :
Team Performance :
Agent Effectiveness :
Agents autonomously decide roles and workflow:
# Agents negotiate roles based on task
agents = [agent1, agent2, agent3]
task = "complex financial analysis"
# Agents determine best structure
negotiated_structure = self_organize(agents, task)
# Returns optimal workflow for this task
Workflow changes based on progress:
# Monitor progress
if progress < expected_rate:
# Increase resources
workflow.add_agent(specialist_agent)
elif quality < threshold:
# Increase validation
workflow.insert_review_step()
Agents learn from each other's work:
# After team execution
execution_trace = crew.get_execution_trace()
# Extract learnings
learnings = extract_patterns(execution_trace)
# Update agent knowledge
for agent, learning in learnings.items():
agent.update_knowledge(learning)
Weekly Installs
800
Repository
GitHub Stars
5
First Seen
Jan 22, 2026
Security Audits
Gen Agent Trust HubFailSocketPassSnykPass
Installed on
opencode726
codex725
gemini-cli724
github-copilot696
cursor670
kimi-cli641
agent-browser 浏览器自动化工具 - Vercel Labs 命令行网页操作与测试
136,300 周安装