Python AI Agent Swarm编排实战：多智能体协作系统的5个核心模式
AI与大数据
单Agent撑不住了？多智能体编排才是正解
你精心调优的Agent，面对"分析竞品+生成报告+发送邮件"这种复合任务就歇菜了。单Agent的四大痛点：能力边界有限——一个模型不可能精通所有领域；任务复杂度超限——多步骤任务容易中途跑偏；串行执行效率低——能并行的步骤只能排队等；专业领域深度不足——通用模型做专业判断总差一口气。
2026年，Agent Swarm（多智能体群体）成为AI工程的主流范式。OpenAI开源了Swarm框架，LangGraph推出Multi-Agent模块，AutoGen进化到0.4版本——多智能体编排不再是实验室玩具，而是生产级系统的刚需。
核心概念速查
| 概念 | 说明 | 典型实现 |
|---|---|---|
| Agent Swarm | 多个Agent协作完成复杂任务的群体 | OpenAI Swarm、CrewAI |
| Handoff | Agent之间转移控制权的机制 | 函数调用返回Agent引用 |
| Orchestrator | 负责任务分解和分配的中央调度器 | Supervisor模式 |
| Worker Agent | 执行具体子任务的专业Agent | 研究Agent、写作Agent |
| Supervisor | 监督Worker执行并汇总结果的Agent | LangGraph Supervisor |
| Group Chat | 多Agent平等参与讨论的协作模式 | AutoGen GroupChat |
| Agent路由 | 根据意图将请求分发到对应Agent | 意图分类+条件路由 |
| 上下文传递 | Agent间共享和传递对话上下文 | 消息历史+变量注入 |
问题分析：多智能体编排的5大挑战
| # | 挑战 | 具体表现 | 影响 |
|---|---|---|---|
| 1 | Agent间上下文传递 | Handoff时丢失前序对话关键信息 | 后续Agent缺乏完整背景，输出质量下降 |
| 2 | 任务分解与分配 | 复杂任务拆分不合理，Agent职责模糊 | 重复执行或遗漏步骤，结果不完整 |
| 3 | 死锁与循环 | Agent A转给B，B又转回A | 无限循环消耗Token，系统卡死 |
| 4 | 结果聚合 | 多Agent输出格式不统一，难以合并 | 最终结果混乱，无法直接使用 |
| 5 | 错误传播 | 上游Agent出错，下游基于错误信息继续 | 错误放大，整个Swarm输出偏离预期 |
分步实操：5个核心编排模式
模式1：OpenAI Swarm基础Handoff
最轻量的多Agent协作——Agent通过Handoff函数将控制权交给另一个Agent。OpenAI Swarm的核心理念：Agent = Instructions + Functions + Handoff。
from dataclasses import dataclass, field
from typing import Callable
from openai import OpenAI
@dataclass
class Agent:
name: str
instructions: str
functions: list[Callable] = field(default_factory=list)
class SwarmRunner:
def __init__(self, api_key: str, model: str = "gpt-4o-mini"):
self.client = OpenAI(api_key=api_key)
self.model = model
def _handoff(self, target_agent: Agent) -> dict:
return {"transfer_to": target_agent.name}
def run(self, agent: Agent, messages: list[dict],
max_turns: int = 10) -> list[dict]:
for _ in range(max_turns):
system_msg = {"role": "system", "content": agent.instructions}
response = self.client.chat.completions.create(
model=self.model,
messages=[system_msg] + messages,
tools=self._build_tools(agent),
)
choice = response.choices[0]
if choice.finish_reason == "tool_calls":
for tool_call in choice.message.tool_calls:
fn_name = tool_call.function.name
if fn_name.startswith("transfer_to_"):
target = fn_name.replace("transfer_to_", "")
agent = self._find_agent(target)
messages.append({
"role": "assistant",
"content": f"Transferring to {target}...",
})
break
else:
for fn in agent.functions:
if fn.__name__ == fn_name:
result = fn()
messages.append({
"role": "tool",
"content": str(result),
"tool_call_id": tool_call.id,
})
else:
messages.append({
"role": "assistant",
"content": choice.message.content,
})
break
return messages
def _build_tools(self, agent: Agent) -> list[dict]:
tools = []
for fn in agent.functions:
tools.append({
"type": "function",
"function": {
"name": fn.__name__,
"description": fn.__doc__ or "",
"parameters": {"type": "object", "properties": {}},
},
})
return tools
def _find_agent(self, name: str) -> Agent:
return self.agents_map.get(name, self.agents_map["default"])
def register_agents(self, agents: list[Agent]):
self.agents_map = {a.name: a for a in agents}
def get_weather() -> str:
"""Get current weather information"""
return "Beijing: 28°C, sunny"
def get_news() -> str:
"""Get latest tech news"""
return "OpenAI releases Swarm 2.0 with built-in monitoring"
weather_agent = Agent(
name="weather",
instructions="You are a weather assistant. Transfer to news agent for non-weather queries.",
functions=[get_weather],
)
news_agent = Agent(
name="news",
instructions="You are a news assistant. Transfer to weather agent for weather queries.",
functions=[get_news],
)
runner = SwarmRunner(api_key="your-api-key")
runner.register_agents([weather_agent, news_agent])
适用场景：客服分流、简单多领域问答，Agent数量 < 5。
模式2：Supervisor编排模式
一个Supervisor Agent负责任务分解、分配和汇总。这是多智能体编排中最常用的生产模式。
from dataclasses import dataclass, field
from enum import Enum
from openai import OpenAI
class TaskStatus(Enum):
PENDING = "pending"
RUNNING = "running"
DONE = "done"
FAILED = "failed"
@dataclass
class SubTask:
task_id: str
description: str
assigned_agent: str
status: TaskStatus = TaskStatus.PENDING
result: str = ""
class SupervisorOrchestrator:
def __init__(self, api_key: str, model: str = "gpt-4o"):
self.client = OpenAI(api_key=api_key)
self.model = model
self.workers: dict[str, Agent] = {}
self.task_history: list[dict] = []
def register_worker(self, agent: Agent):
self.workers[agent.name] = agent
def decompose(self, user_request: str) -> list[SubTask]:
prompt = (
f"Decompose the following request into sub-tasks.\n"
f"Available agents: {list(self.workers.keys())}\n"
f"Request: {user_request}\n\n"
f"Output JSON array: [{{\"task_id\": \"t1\", "
f"\"description\": \"...\", \"assigned_agent\": \"...\"}}]"
)
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
)
import json
data = json.loads(response.choices[0].message.content)
tasks = data.get("tasks", [])
return [
SubTask(
task_id=t["task_id"],
description=t["description"],
assigned_agent=t["assigned_agent"],
)
for t in tasks
]
def execute_task(self, task: SubTask) -> str:
worker = self.workers.get(task.assigned_agent)
if not worker:
task.status = TaskStatus.FAILED
return f"Agent '{task.assigned_agent}' not found"
task.status = TaskStatus.RUNNING
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": worker.instructions},
{"role": "user", "content": task.description},
],
)
task.result = response.choices[0].message.content
task.status = TaskStatus.DONE
return task.result
def run(self, user_request: str) -> str:
subtasks = self.decompose(user_request)
results = []
for task in subtasks:
result = self.execute_task(task)
results.append(f"[{task.task_id}] {result}")
self.task_history.append({
"task_id": task.task_id,
"agent": task.assigned_agent,
"status": task.status.value,
})
summary_prompt = (
f"Combine the following sub-task results into a coherent response:\n"
+ "\n".join(results)
)
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": summary_prompt}],
)
return response.choices[0].message.content
researcher = Agent(
name="researcher",
instructions="You research topics thoroughly. Provide detailed findings.",
)
writer = Agent(
name="writer",
instructions="You write clear, engaging content based on research.",
)
supervisor = SupervisorOrchestrator(api_key="your-api-key")
supervisor.register_worker(researcher)
supervisor.register_worker(writer)
适用场景：报告生成、内容创作、研究+写作组合任务。
模式3：Group Chat讨论模式
多个Agent平等参与讨论，由GroupChat Manager决定下一个发言者。适合需要多视角碰撞的场景。
from dataclasses import dataclass, field
from openai import OpenAI
@dataclass
class ChatAgent:
name: str
role: str
instructions: str
class GroupChatManager:
def __init__(self, api_key: str, model: str = "gpt-4o",
max_rounds: int = 6):
self.client = OpenAI(api_key=api_key)
self.model = model
self.max_rounds = max_rounds
self.agents: list[ChatAgent] = []
self.chat_history: list[dict] = []
def add_agent(self, agent: ChatAgent):
self.agents.append(agent)
def _select_speaker(self, context: str) -> ChatAgent:
agent_names = [a.name for a in self.agents]
prompt = (
f"Based on the conversation context, select the next speaker.\n"
f"Available speakers: {agent_names}\n"
f"Context: {context[-500:]}\n\n"
f"Reply with ONLY the speaker name."
)
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
max_tokens=20,
)
name = response.choices[0].message.content.strip()
for agent in self.agents:
if agent.name.lower() in name.lower():
return agent
return self.agents[0]
def _agent_respond(self, agent: ChatAgent,
messages: list[dict]) -> str:
system_msg = (
f"You are {agent.name}, role: {agent.role}. "
f"{agent.instructions}"
)
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "system", "content": system_msg}] + messages,
)
return response.choices[0].message.content
def discuss(self, topic: str) -> list[dict]:
self.chat_history = []
self.chat_history.append({
"role": "user",
"content": f"Discussion topic: {topic}",
})
for round_num in range(self.max_rounds):
context = "\n".join(
f"{m['role']}: {m['content'][:100]}"
for m in self.chat_history[-6:]
)
speaker = self._select_speaker(context)
response = self._agent_respond(speaker, self.chat_history)
self.chat_history.append({
"role": "assistant",
"content": f"[{speaker.name}]: {response}",
})
return self.chat_history
pm = ChatAgent(
name="PM",
role="Product Manager",
instructions="Focus on user needs and business value.",
)
dev = ChatAgent(
name="Dev",
role="Senior Developer",
instructions="Focus on technical feasibility and architecture.",
)
designer = ChatAgent(
name="Designer",
role="UX Designer",
instructions="Focus on user experience and interaction design.",
)
chat = GroupChatManager(api_key="your-api-key", max_rounds=6)
chat.add_agent(pm)
chat.add_agent(dev)
chat.add_agent(designer)
适用场景：需求评审、方案讨论、头脑风暴。
模式4：流水线Pipeline模式
Agent按固定顺序依次处理，前一个Agent的输出作为后一个的输入。适合确定性流程。
from dataclasses import dataclass
from openai import OpenAI
@dataclass
class PipelineStep:
name: str
agent: Agent
input_key: str
output_key: str
class PipelineOrchestrator:
def __init__(self, api_key: str, model: str = "gpt-4o-mini"):
self.client = OpenAI(api_key=api_key)
self.model = model
self.steps: list[PipelineStep] = []
def add_step(self, step: PipelineStep):
self.steps.append(step)
def run(self, initial_input: dict) -> dict:
context = dict(initial_input)
for step in self.steps:
input_text = context.get(step.input_key, "")
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": step.agent.instructions},
{"role": "user", "content": str(input_text)},
],
)
context[step.output_key] = response.choices[0].message.content
return context
collector = Agent(
name="data_collector",
instructions="Collect and organize raw data. Output structured findings.",
)
analyzer = Agent(
name="analyzer",
instructions="Analyze the collected data. Identify key patterns and insights.",
)
reporter = Agent(
name="reporter",
instructions="Write a professional report based on the analysis. Use markdown format.",
)
pipeline = PipelineOrchestrator(api_key="your-api-key")
pipeline.add_step(PipelineStep("collect", collector, "query", "raw_data"))
pipeline.add_step(PipelineStep("analyze", analyzer, "raw_data", "analysis"))
pipeline.add_step(PipelineStep("report", reporter, "analysis", "report"))
result = pipeline.run({"query": "2026 AI Agent market trends"})
适用场景：数据处理流水线、报告生成、ETL流程。
模式5：生产级Swarm框架（含监控）
整合Handoff、Supervisor、Pipeline，加入监控、重试和熔断。这是生产级多智能体编排的完整方案。
import time
import logging
from dataclasses import dataclass, field
from enum import Enum
from openai import OpenAI
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("SwarmMonitor")
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
@dataclass
class AgentMetrics:
name: str
call_count: int = 0
success_count: int = 0
fail_count: int = 0
total_latency_ms: float = 0.0
last_error: str = ""
@property
def success_rate(self) -> float:
return self.success_count / max(self.call_count, 1)
@property
def avg_latency_ms(self) -> float:
return self.total_latency_ms / max(self.call_count, 1)
class CircuitBreaker:
def __init__(self, failure_threshold: int = 3,
recovery_timeout: float = 30.0):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time: float = 0
def allow(self) -> bool:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
return True
return False
return True
def record_success(self):
self.failure_count = 0
self.state = CircuitState.CLOSED
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
class ProductionSwarm:
def __init__(self, api_key: str, model: str = "gpt-4o",
max_retries: int = 2):
self.client = OpenAI(api_key=api_key)
self.model = model
self.max_retries = max_retries
self.agents: dict[str, Agent] = {}
self.metrics: dict[str, AgentMetrics] = {}
self.circuit_breakers: dict[str, CircuitBreaker] = {}
def register(self, agent: Agent):
self.agents[agent.name] = agent
self.metrics[agent.name] = AgentMetrics(name=agent.name)
self.circuit_breakers[agent.name] = CircuitBreaker()
def call_agent(self, agent_name: str, prompt: str) -> str:
agent = self.agents.get(agent_name)
if not agent:
raise ValueError(f"Agent '{agent_name}' not found")
cb = self.circuit_breakers[agent_name]
metrics = self.metrics[agent_name]
if not cb.allow():
return f"[CIRCUIT_OPEN] Agent {agent_name} is unavailable"
for attempt in range(self.max_retries + 1):
start = time.time()
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": agent.instructions},
{"role": "user", "content": prompt},
],
)
result = response.choices[0].message.content
latency = (time.time() - start) * 1000
metrics.call_count += 1
metrics.success_count += 1
metrics.total_latency_ms += latency
cb.record_success()
logger.info(
f"[{agent_name}] success in {latency:.0f}ms "
f"(attempt {attempt + 1})"
)
return result
except Exception as e:
latency = (time.time() - start) * 1000
metrics.call_count += 1
metrics.fail_count += 1
metrics.total_latency_ms += latency
metrics.last_error = str(e)
cb.record_failure()
logger.warning(
f"[{agent_name}] failed: {e} (attempt {attempt + 1})"
)
return f"[FAILED] Agent {agent_name} after {self.max_retries} retries"
def health_check(self) -> dict:
return {
name: {
"success_rate": f"{m.success_rate:.1%}",
"avg_latency": f"{m.avg_latency_ms:.0f}ms",
"circuit": self.circuit_breakers[name].state.value,
"last_error": m.last_error[:80],
}
for name, m in self.metrics.items()
}
swarm = ProductionSwarm(api_key="your-api-key", max_retries=2)
swarm.register(Agent(
name="planner",
instructions="Break down complex tasks into actionable steps.",
))
swarm.register(Agent(
name="executor",
instructions="Execute the given task step and return results.",
))
swarm.register(Agent(
name="reviewer",
instructions="Review the execution results for quality and completeness.",
))
plan = swarm.call_agent("planner", "Design a microservice for user auth")
result = swarm.call_agent("executor", f"Execute: {plan}")
review = swarm.call_agent("reviewer", f"Review: {result}")
print(swarm.health_check())
适用场景：企业级AI助手、需要高可用的多Agent生产系统。
避坑指南：5个常见陷阱
陷阱1：Handoff时丢失上下文
❌ 错误做法：
def transfer_to_agent_b():
return agent_b
✅ 正确做法：
def transfer_to_agent_b(context: dict):
agent_b.instructions += f"\nPrevious context: {context}"
return agent_b
陷阱2：没有循环检测导致死锁
❌ 错误做法：
def handoff(target_agent):
return target_agent
✅ 正确做法：
class LoopDetector:
def __init__(self, max_handoffs: int = 10):
self.handoff_chain: list[str] = []
self.max_handoffs = max_handoffs
def check(self, agent_name: str) -> bool:
if len(self.handoff_chain) >= self.max_handoffs:
return False
recent = self.handoff_chain[-3:]
if recent.count(agent_name) >= 2:
return False
self.handoff_chain.append(agent_name)
return True
陷阱3：所有Agent用同一个模型
❌ 错误做法：
all_agents = [Agent(name=n, instructions=i, model="gpt-4o") for n, i in specs]
✅ 正确做法：
router_agent = Agent(name="router", instructions="...", model="gpt-4o-mini")
reasoning_agent = Agent(name="reasoning", instructions="...", model="o3-mini")
creative_agent = Agent(name="creative", instructions="...", model="gpt-4o")
陷阱4：忽略Agent执行超时
❌ 错误做法：
result = client.chat.completions.create(model=model, messages=msgs)
✅ 正确做法：
import signal
class TimeoutError(Exception):
pass
def run_with_timeout(fn, timeout: float = 30.0):
result = [None]
def wrapper():
result[0] = fn()
thread = threading.Thread(target=wrapper)
thread.start()
thread.join(timeout=timeout)
if thread.is_alive():
raise TimeoutError(f"Agent call exceeded {timeout}s")
return result[0]
陷阱5：结果聚合不做格式校验
❌ 错误做法：
final = "\n".join(agent_results)
✅ 正确做法：
from pydantic import BaseModel, ValidationError
class AgentOutput(BaseModel):
agent_name: str
status: str
content: str
confidence: float
def aggregate(results: list[str]) -> list[AgentOutput]:
outputs = []
for r in results:
try:
outputs.append(AgentOutput.model_validate_json(r))
except ValidationError as e:
outputs.append(AgentOutput(
agent_name="unknown", status="parse_error",
content=r, confidence=0.0,
))
return outputs
报错排查：10个常见错误
| # | 错误信息 | 原因 | 解决方案 |
|---|---|---|---|
| 1 | Handoff loop detected |
Agent间互相Handoff形成循环 | 添加LoopDetector限制Handoff链长度 |
| 2 | Agent not found in registry |
Handoff目标Agent未注册 | 检查register_agents是否包含所有Agent |
| 3 | Context window exceeded in group chat |
多Agent讨论历史过长 | 限制chat_history长度，定期摘要 |
| 4 | Supervisor decomposition failed |
LLM返回的JSON格式不正确 | 使用response_format=json_object + 重试 |
| 5 | Circuit breaker open for agent |
Agent连续失败触发熔断 | 检查API配额、模型可用性，调整阈值 |
| 6 | Pipeline step timeout |
某步骤执行超时 | 设置per-step超时，添加降级策略 |
| 7 | Token cost spike in swarm |
Agent数量*轮次导致Token暴增 | 限制max_turns，用mini模型做路由 |
| 8 | Inconsistent output format |
不同Agent输出格式不统一 | 定义Pydantic输出Schema并强制校验 |
| 9 | Deadlock in parallel execution |
并行Agent争抢共享资源 | 使用异步锁或消息队列解耦 |
| 10 | Memory leak in long-running swarm |
长时间运行的Swarm上下文累积 | 定期清理chat_history，设置上下文窗口上限 |
进阶优化：4个关键技巧
1. Agent路由优化——用小模型做意图分类
from openai import OpenAI
class AgentRouter:
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
self.route_map: dict[str, str] = {}
def add_route(self, intent: str, agent_name: str):
self.route_map[intent] = agent_name
def route(self, query: str) -> str:
prompt = (
f"Classify the intent of this query into one of: "
f"{list(self.route_map.keys())}\n"
f"Query: {query}\nReply with ONLY the intent name."
)
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
max_tokens=20,
)
intent = response.choices[0].message.content.strip().lower()
return self.route_map.get(intent, "default")
2. 异步并行执行——多个Worker同时运行
import asyncio
from openai import AsyncOpenAI
class AsyncSwarmRunner:
def __init__(self, api_key: str, model: str = "gpt-4o-mini"):
self.client = AsyncOpenAI(api_key=api_key)
self.model = model
async def call_agent(self, agent: Agent, prompt: str) -> str:
response = await self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": agent.instructions},
{"role": "user", "content": prompt},
],
)
return response.choices[0].message.content
async def run_parallel(self, tasks: list[tuple[Agent, str]]) -> list[str]:
coros = [self.call_agent(agent, prompt) for agent, prompt in tasks]
return await asyncio.gather(*coros)
3. 上下文窗口管理——滑动窗口+摘要
class SwarmContextManager:
def __init__(self, max_messages: int = 20):
self.max_messages = max_messages
self.summary: str = ""
def add(self, message: dict):
self.history.append(message)
if len(self.history) > self.max_messages:
self._summarize_old()
def get_context(self) -> list[dict]:
ctx = []
if self.summary:
ctx.append({"role": "system", "content": f"Summary: {self.summary}"})
ctx.extend(self.history)
return ctx
4. 成本监控——Token用量追踪
class TokenTracker:
def __init__(self, budget_per_request: float = 0.5):
self.budget = budget_per_request
self.usage: dict[str, float] = {}
def track(self, agent_name: str, tokens: int, cost_per_1k: float):
cost = tokens / 1000 * cost_per_1k
self.usage[agent_name] = self.usage.get(agent_name, 0) + cost
def check_budget(self, agent_name: str) -> bool:
return self.usage.get(agent_name, 0) < self.budget
def report(self) -> dict:
total = sum(self.usage.values())
return {"per_agent": self.usage, "total_usd": total}
对比分析：4大多智能体框架
| 维度 | OpenAI Swarm | LangGraph | AutoGen | CrewAI |
|---|---|---|---|---|
| 核心理念 | 极简Handoff | 图状态机 | 对话驱动 | 角色扮演 |
| 学习曲线 | ★☆☆ | ★★★ | ★★☆ | ★★☆ |
| 灵活性 | ★★★★ | ★★★★★ | ★★★ | ★★☆ |
| 生产就绪 | ★★☆ | ★★★★ | ★★★ | ★★★ |
| 内置监控 | ✗ | ✓ | △ | △ |
| 流式输出 | ✓ | ✓ | ✓ | △ |
| 持久化 | ✗ | ✓ | ✓ | ✓ |
| Human-in-loop | ✗ | ✓ | ✓ | ✓ |
| 适用场景 | 快速原型 | 复杂工作流 | 多Agent讨论 | 团队协作 |
| Token效率 | ★★★★ | ★★★ | ★★☆ | ★★☆ |
★越多表示该维度表现越好；✓支持 △部分支持 ✗不支持
总结展望
Agent Swarm（多智能体编排）正在从"实验性探索"走向"工程化落地"。2026年的关键趋势：
- 原生Swarm支持：OpenAI Swarm 2.0将内置监控和持久化，LangGraph Multi-Agent成为标配
- 自适应编排：根据任务复杂度自动选择Handoff/Supervisor/Group Chat模式
- Agent市场：可复用的专业Agent组件，像npm包一样组合
- 多模态Swarm：视觉Agent、语音Agent、代码Agent协同工作
- 安全与治理：Agent行为审计、权限控制、成本上限成为生产刚需
选择编排模式的原则：从简单开始，按需升级。2-3个Agent用Handoff，需要中央调度上Supervisor，多视角碰撞用Group Chat，确定性流程走Pipeline。别一上来就搞全套——先跑通最小可用路径，再逐步增加复杂度。
在线工具推荐
本站提供浏览器本地工具，免注册即可试用 →
#Agent Swarm#多智能体编排#OpenAI Swarm#Agent协作#群体智能#2026#AI与大数据