通过 Swarm 构建模块化、可扩展的多代理应用程序

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],
)

上述购物示例包括三个代理，以处理各种客户服务请求：

1. triage_agent(分流代理)：确定请求类型并转给相应的代理。
2. sales_agent(销售代理)：处理与下订单相关的操作，需要用户 ID 和产品 ID 才能完成购买。
3. 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() 核心实现了以下循环：

1. 从当前代理获取完成信息
2. 执行工具调用并附加结果
3. 必要时交接代理
4. 必要时更新上下文变量
5. 如果没有新的函数调用，则返回

接下来让我们执行上面的示例，看看会发生什么？

上面体现了代理思考、代理交接、工具执行的完整过程，整个过程非常清晰！

参考文献

1. https://github.com/openai/swarm
2. https://cookbook.openai.com/examples/orchestrating_agents?utm_source=www.theunwindai.com&utm_medium=referral&utm_campaign=openai-drops-opensource-multi-agent-framework