OpenAI 近期推出了 Swarm 框架，这是 OpenAI 首次开源的项目。Swarm 是一个无状态、轻量级多代理框架，利用 “协程” 和 “交接” 的概念来协调多个 AI 代理。这将复杂的任务分解成更小的、更易管理的单元，从而简化了任务，而且人工智能代理可以在任何时候选择将对话交接给另一个代理。
与单个代理中管理大量提示和不同逻辑相比，Swarm是一种稳健且可扩展的方法。示例库提供了可以轻松实现的实用示例和代码。
关键内容
协程 - 定义一组指令（系统提示），并将工具（功能）与之关联，以处理特定任务。这样就可以进行模块化设计，更容易管理复杂的流程。LLM 遵循这些协程，为更自然的对话提供 “软” 遵循。
# 客户服务协程示例
system_message = (
"You are a customer support agent for ACME Inc."
"Always answer in a sentence or less."
"Follow the following routine with the user:"
"1. First, ask probing questions and understand the user's problem deeper.\n"
" - unless the user has already provided a reason.\n"
"2. Propose a fix (make one up).\n"
"3. ONLY if not satesfied, offer a refund.\n"
"4. If accepted, search for the ID and then execute refund."
""
)
def look_up_item(search_query):
"""用于查询物品ID.
Search query can be a description or keywords."""
# 返回硬编码物品ID - 在真实代码中应该是一次查询
return "item_132612938"
def execute_refund(item_id, reason="not provided"):
print("Summary:", item_id, reason)
return "success"
交接 - 在对话过程中实现不同代理之间的动态交接，允许专门的代理处理交互的特定部分。代理可以调用交接功能来触发交接，为代理保留完整的对话历史记录。
# 交接函数示例
def transfer_to_agent_b():
return agent_b
agent_a = Agent(
name="Agent A",
instructions="You are a helpful agent.",
# 交接给 agent_b
functions=[transfer_to_agent_b],
)
agent_b = Agent(
name="Agent B",
instructions="Only speak in Haikus.",
)
工具调用和执行 - 功能会自动转换为 JSON 模式，以便与 OpenAI 的工具调用功能配合使用。功能执行的结果将返回模型，从而促进动态交互和复杂逻辑。
def greet(context_variables, language):
user_name = context_variables["user_name"]
greeting = "Hola" if language.lower() == "spanish" else "Hello"
print(f"{greeting}, {user_name}!")
return "Done"
agent = Agent(
functions=[greet]
)
{
"type": "function",
"function": {
"name": "greet",
"description": "Greets the user. Make sure to get their name and age before calling.\n\nArgs:\n name: Name of the user.\n age: Age of the user.\n location: Best place on earth.",
"parameters": {
"type": "object",
"properties": {
"name": {"type": "string"},
"age": {"type": "integer"},
"location": {"type": "string"}
},
"required": ["name", "age"]
}
}
}
本地设置 - Swarm 几乎完全在客户端运行，这意味着您可以测试和运行基于代理的系统，而无需担心服务器端的依赖性。
Swarm库（实验性质）--这里有一个方便的 Python 库，它与示例一起实现了这些想法。可以将其视为构建自己的多代理系统的入门套件。在 examples 中包括了大量示例：
basic：设置、函数调用、交接和上下文变量等基础知识的简单示例 triage_agent：设置基本分流步骤以移交给正确代理的简单示例 weather_agent：函数调用的简单示例 airline:：多代理设置，用于处理航空公司中不同的客户服务请求。 support_bot：一个客户服务机器人，包括一个用户界面代理和一个带有多种工具的帮助中心代理 personal_shopper：个人购物代理，可帮助进行销售和退还订单
应用示例
def escalate_to_human(summary):
"""Only call this if explicitly asked to."""
print("Escalating to human agent...")
print("\n=== Escalation Report ===")
print(f"Summary: {summary}")
print("=========================\n")
exit()
def transfer_to_sales_agent():
"""User for anything sales or buying related."""
return sales_agent
def transfer_to_issues_and_repairs():
"""User for issues, repairs, or refunds."""
return issues_and_repairs_agent
def transfer_back_to_triage():
"""Call this if the user brings up a topic outside of your purview,
including escalating to human."""
return triage_agent
triage_agent = Agent(
name="Triage Agent",
instructions=(
"You are a customer service bot for ACME Inc. "
"Introduce yourself. Always be very brief. "
"Gather information to direct the customer to the right department. "
"But make your questions subtle and natural."
),
tools=[transfer_to_sales_agent, transfer_to_issues_and_repairs, escalate_to_human],
)
def execute_order(product, price: int):
"""Price should be in USD."""
print("\n\n=== Order Summary ===")
print(f"Product: {product}")
print(f"Price: ${price}")
print("=================\n")
confirm = input("Confirm order? y/n: ").strip().lower()
if confirm == "y":
print("Order execution successful!")
return "Success"
else:
print("Order cancelled!")
return "User cancelled order."
sales_agent = Agent(
name="Sales Agent",
instructions=(
"You are a sales agent for ACME Inc."
"Always answer in a sentence or less."
"Follow the following routine with the user:"
"1. Ask them about any problems in their life related to catching roadrunners.\n"
"2. Casually mention one of ACME's crazy made-up products can help.\n"
" - Don't mention price.\n"
"3. Once the user is bought in, drop a ridiculous price.\n"
"4. Only after everything, and if the user says yes, "
"tell them a crazy caveat and execute their order.\n"
""
),
tools=[execute_order, transfer_back_to_triage],
)
def look_up_item(search_query):
"""Use to find item ID.
Search query can be a description or keywords."""
item_id = "item_132612938"
print("Found item:", item_id)
return item_id
def execute_refund(item_id, reason="not provided"):
print("\n\n=== Refund Summary ===")
print(f"Item ID: {item_id}")
print(f"Reason: {reason}")
print("=================\n")
print("Refund execution successful!")
return "success"
issues_and_repairs_agent = Agent(
name="Issues and Repairs Agent",
instructions=(
"You are a customer support agent for ACME Inc."
"Always answer in a sentence or less."
"Follow the following routine with the user:"
"1. First, ask probing questions and understand the user's problem deeper.\n"
" - unless the user has already provided a reason.\n"
"2. Propose a fix (make one up).\n"
"3. ONLY if not satesfied, offer a refund.\n"
"4. If accepted, search for the ID and then execute refund."
""
),
tools=[execute_refund, look_up_item, transfer_back_to_triage],
)
上述购物示例包括三个代理，以处理各种客户服务请求：
triage_agent(分流代理)：确定请求类型并转给相应的代理。 sales_agent(销售代理)：处理与下订单相关的操作，需要用户 ID 和产品 ID 才能完成购买。 issues_and_repairs_agent(售后代理)：管理客户退款，需要用户 ID 和产品 ID 才能启动退款。
from swarm.repl import run_demo_loop
if __name__ == "__main__":
run_demo_loop(triage_agent)
使用辅助函数 run_demo_loop，帮助我们创建一个交互式 Swarm 会话。
Swarm 的 run() 函数类似于聊天完成 API 中的 chat.completions.create() 函数--它接收消息并返回消息，调用之间不保存任何状态。但重要的是，它还能处理代理函数的执行、交接、上下文变量引用，并能在返回用户之前进行多次交接。
Swarm 的 client.run() 核心实现了以下循环：
从当前代理获取完成信息 执行工具调用并附加结果 必要时交接代理 必要时更新上下文变量 如果没有新的函数调用，则返回
接下来让我们执行上面的示例，看看会发生什么？
上面体现了代理思考、代理交接、工具执行的完整过程，整个过程非常清晰！
参考文献
https://github.com/openai/swarm https://cookbook.openai.com/examples/orchestrating_agents?utm_source=www.theunwindai.com&utm_medium=referral&utm_campaign=openai-drops-opensource-multi-agent-framework