编排智能体：例程与交接

技术背景

在构建基于大语言模型的应用时，单个提示词和工具组合往往能解决简单任务。然而，真实业务场景常包含多种流程，如客服、销售、售后等，每个流程有独特步骤和工具。若将所有逻辑塞入一个提示词，会导致指令冗长、模型性能下降、维护困难。传统状态机虽可精确控制，但僵硬且难以适应自然语言交互。因此，需要一种既能保持LLM灵活性，又能结构化编排多个智能体的方法。例程和交接正是为此设计：例程将一组指令和工具封装为独立单元，交接则允许智能体在对话中动态切换，实现模块化、可扩展的多智能体系统。

核心原理

核心思想是将复杂任务分解为多个例程（Routine），每个例程由系统提示（自然语言指令）和工具（Python函数）组成。例程通过循环执行：获取用户输入、调用模型、处理工具调用，直到模型不再请求工具。交接（Handoff）通过让工具返回Agent对象实现：当模型调用transfer_to_XXX函数时，返回目标Agent，系统更新当前Agent并继续对话。这样，智能体可以自主决定何时转移，且对话历史完整保留。关键设计决策包括：使用函数模式（function schema）将Python函数暴露给模型；通过类型映射和inspect自动生成schema；工具调用结果以tool角色消息回传；交接时注入“Adopt persona immediately”指令以提示模型切换角色。

架构分析

系统由Agent类、run_full_turn循环、工具执行器和交接机制组成。Agent类封装名称、模型、指令和工具列表。run_full_turn接收Agent和消息列表，循环执行：生成工具schema、调用模型、处理工具调用。若工具返回Agent，则切换当前Agent并继续循环。工具执行器通过tools_map将函数名映射到实际函数，并处理参数解析。交接函数如transfer_to_sales_agent返回Agent对象，系统检测到后更新current_agent。这种设计实现了动态路由：分诊Agent根据用户意图将对话转给销售或售后Agent，售后Agent也可转回分诊。架构优势在于模块化、可扩展，每个Agent独立维护，通过交接函数灵活组合。

完整实战教程

编排智能体：例程与交接

在使用语言模型时，通常只需一个良好的提示词和合适的工具就能获得稳定的性能。然而，当处理许多独特的流程时，事情可能会变得棘手。本教程将介绍一种解决此问题的方法。

我们将引入例程和交接的概念，然后逐步讲解实现过程，并展示如何利用它们以简单、强大且可控的方式编排多个智能体。

最后，我们提供了一个示例代码库Swarm，其中实现了这些想法并附有示例。

让我们从设置导入开始。

from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
import json


client = OpenAI()

例程

“例程”的概念并没有严格定义，而是用来表示一组步骤的想法。具体来说，让我们将例程定义为自然语言指令列表（我们将用系统提示来表示），以及完成这些指令所需的工具。

让我们看一个例子。下面，我们为客服代理定义了一个例程，指示它对用户问题进行分诊，然后建议修复方案或提供退款。我们还定义了必要的函数 execute_refund 和 look_up_item。我们可以将其称为客服例程、代理、助手等——但想法本身是相同的：一组步骤以及执行这些步骤的工具。

# Customer Service Routine

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 satisfied, offer a refund.\n"
    "4. If accepted, search for the ID and then execute refund."
    ""
)

def look_up_item(search_query):
    """Use to find item ID.
    Search query can be a description or keywords."""

    # return hard-coded item ID - in reality would be a lookup
    return "item_132612938"


def execute_refund(item_id, reason="not provided"):

    print("Summary:", item_id, reason) # lazy summary
    return "success"

例程的主要力量在于其简单性和稳健性。请注意，这些指令包含条件判断，类似于状态机或代码中的分支。对于中小型例程，LLM 实际上可以相当稳健地处理这些情况，并且具有“软”遵循的额外好处——LLM 可以自然地引导对话，而不会陷入死胡同。

执行例程

要执行例程，我们实现一个简单的循环，该循环：

1. 获取用户输入。
2. 将用户消息追加到 messages。
3. 调用模型。
4. 将模型响应追加到 messages。

def run_full_turn(system_message, messages):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": system_message}] + messages,
    )
    message = response.choices[0].message
    messages.append(message)

    if message.content: print("Assistant:", message.content)

    return message


messages = []
while True:
    user = input("User: ")
    messages.append({"role": "user", "content": user})

    run_full_turn(system_message, messages)

如你所见，目前这忽略了函数调用，所以让我们添加这一点。

模型要求函数以函数模式（function schema）的格式提供。为了方便起见，我们可以定义一个辅助函数，将 Python 函数转换为相应的函数模式。

import inspect

def function_to_schema(func) -> dict:
    type_map = {
        str: "string",
        int: "integer",
        float: "number",
        bool: "boolean",
        list: "array",
        dict: "object",
        type(None): "null",
    }

    try:
        signature = inspect.signature(func)
    except ValueError as e:
        raise ValueError(
            f"Failed to get signature for function {func.__name__}: {str(e)}"
        )

    parameters = {}
    for param in signature.parameters.values():
        try:
            param_type = type_map.get(param.annotation, "string")
        except KeyError as e:
            raise KeyError(
                f"Unknown type annotation {param.annotation} for parameter {param.name}: {str(e)}"
            )
        parameters[param.name] = {"type": param_type}

    required = [
        param.name
        for param in signature.parameters.values()
        if param.default == inspect._empty
    ]

    return {
        "type": "function",
        "function": {
            "name": func.__name__,
            "description": (func.__doc__ or "").strip(),
            "parameters": {
                "type": "object",
                "properties": parameters,
                "required": required,
            },
        },
    }

例如：

def sample_function(param_1, param_2, the_third_one: int, some_optional="John Doe"):
    """
    This is my docstring. Call this function when you want.
    """
    print("Hello, world")

schema =  function_to_schema(sample_function)
print(json.dumps(schema, indent=2))

现在，我们可以在调用模型时使用这个函数将工具传递给模型。

messages = []

tools = [execute_refund, look_up_item]
tool_schemas = [function_to_schema(tool) for tool in tools]

response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "Look up the black boot."}],
            tools=tool_schemas,
        )
message = response.choices[0].message

message.tool_calls[0].function

最后，当模型调用工具时，我们需要执行相应的函数并将结果返回给模型。

我们可以通过将工具名称映射到 tool_map 中的 Python 函数来实现这一点，然后在 execute_tool_call 中查找并调用它。最后，我们将结果添加到对话中。

tools_map = {tool.__name__: tool for tool in tools}

def execute_tool_call(tool_call, tools_map):
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    print(f"Assistant: {name}({args})")

    # call corresponding function with provided arguments
    return tools_map[name](**args)

for tool_call in message.tool_calls:
            result = execute_tool_call(tool_call, tools_map)

            # add result back to conversation 
            result_message = {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            }
            messages.append(result_message)

在实践中，我们还需要让模型使用结果来生成另一个响应。该响应可能_也_包含工具调用，因此我们可以循环执行此操作，直到没有更多工具调用为止。

如果我们将所有内容放在一起，它看起来会像这样：

tools = [execute_refund, look_up_item]


def run_full_turn(system_message, tools, messages):

    num_init_messages = len(messages)
    messages = messages.copy()

    while True:

        # turn python functions into tools and save a reverse map
        tool_schemas = [function_to_schema(tool) for tool in tools]
        tools_map = {tool.__name__: tool for tool in tools}

        # === 1. get openai completion ===
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "system", "content": system_message}] + messages,
            tools=tool_schemas or None,
        )
        message = response.choices[0].message
        messages.append(message)

        if message.content:  # print assistant response
            print("Assistant:", message.content)

        if not message.tool_calls:  # if finished handling tool calls, break
            break

        # === 2. handle tool calls ===

        for tool_call in message.tool_calls:
            result = execute_tool_call(tool_call, tools_map)

            result_message = {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            }
            messages.append(result_message)

    # ==== 3. return new messages =====
    return messages[num_init_messages:]


def execute_tool_call(tool_call, tools_map):
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    print(f"Assistant: {name}({args})")

    # call corresponding function with provided arguments
    return tools_map[name](**args)


messages = []
while True:
    user = input("User: ")
    messages.append({"role": "user", "content": user})

    new_messages = run_full_turn(system_message, tools, messages)
    messages.extend(new_messages)

现在我们有了一个例程，假设我们想添加更多步骤和更多工具。我们可以做到一定程度，但最终如果我们试图用太多不同的任务来扩展例程，它可能会开始出现问题。这时我们可以利用多个例程的概念——根据用户的请求，我们可以加载具有适当步骤和工具的例程来处理它。

动态交换系统指令和工具可能看起来令人生畏。然而，如果我们将“例程”视为“代理”，那么交接的概念让我们能够简单地将这些交换表示为——一个代理将对话交给另一个代理。

交接

让我们将交接定义为一个代理（或例程）将活动对话移交给另一个代理，就像电话通话中你被转接给其他人一样。只不过在这种情况下，代理完全了解你之前的对话！

为了了解交接的实际运作，我们先定义一个代理的基本类。

class Agent(BaseModel):
    name: str = "Agent"
    model: str = "gpt-4o-mini"
    instructions: str = "You are a helpful Agent"
    tools: list = []

现在，为了让我们的代码支持这一点，我们可以将 run_full_turn 改为接收一个 Agent 对象，而不是单独的 system_message 和 tools：

def run_full_turn(agent, messages):

    num_init_messages = len(messages)
    messages = messages.copy()

    while True:

        # turn python functions into tools and save a reverse map
        tool_schemas = [function_to_schema(tool) for tool in agent.tools]
        tools_map = {tool.__name__: tool for tool in agent.tools}

        # === 1. get openai completion ===
        response = client.chat.completions.create(
            model=agent.model,
            messages=[{"role": "system", "content": agent.instructions}] + messages,
            tools=tool_schemas or None,
        )
        message = response.choices[0].message
        messages.append(message)

        if message.content:  # print assistant response
            print("Assistant:", message.content)

        if not message.tool_calls:  # if finished handling tool calls, break
            break

        # === 2. handle tool calls ===

        for tool_call in message.tool_calls:
            result = execute_tool_call(tool_call, tools_map)

            result_message = {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            }
            messages.append(result_message)

    # ==== 3. return new messages =====
    return messages[num_init_messages:]


def execute_tool_call(tool_call, tools_map):
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    print(f"Assistant: {name}({args})")

    # call corresponding function with provided arguments
    return tools_map[name](**args)

现在我们可以轻松地运行多个代理：

def execute_refund(item_name):
    return "success"

refund_agent = Agent(
    name="Refund Agent",
    instructions="You are a refund agent. Help the user with refunds.",
    tools=[execute_refund],
)

def place_order(item_name):
    return "success"

sales_assistant = Agent(
    name="Sales Assistant",
    instructions="You are a sales assistant. Sell the user a product.",
    tools=[place_order],
)


messages = []
user_query = "Place an order for a black boot."
print("User:", user_query)
messages.append({"role": "user", "content": user_query})

response = run_full_turn(sales_assistant, messages) # sales assistant
messages.extend(response)


user_query = "Actually, I want a refund." # implicitly refers to the last item
print("User:", user_query)
messages.append({"role": "user", "content": user_query})
response = run_full_turn(refund_agent, messages) # refund agent

很好！但这里我们是手动进行交接的——我们希望智能体自己决定何时进行交接。一个简单但出奇有效的方法是给它们一个 transfer_to_XXX 函数，其中 XXX 是某个智能体。模型足够聪明，知道在需要交接时调用这个函数！

交接函数

既然智能体现在可以表达进行交接的_意图_，我们必须让它真正发生。实现方式有很多种，但有一种特别简洁的方法。

对于目前定义的智能体函数，如 execute_refund 或 place_order，它们返回一个字符串，该字符串将提供给模型。如果我们改为返回一个 Agent 对象来表示我们想要转移到的智能体，会怎样？如下所示：

refund_agent = Agent(
    name="Refund Agent",
    instructions="You are a refund agent. Help the user with refunds.",
    tools=[execute_refund],
)

def transfer_to_refunds():
    return refund_agent

sales_assistant = Agent(
    name="Sales Assistant",
    instructions="You are a sales assistant. Sell the user a product.",
    tools=[place_order],
)

然后我们可以更新代码，检查函数响应的返回类型，如果是 Agent，就更新当前使用的智能体！此外，现在 run_full_turn 需要返回最新使用的智能体，以防有交接发生。（我们可以通过一个 Response 类来保持整洁。）

class Response(BaseModel):
    agent: Optional[Agent]
    messages: list

现在来看更新后的 run_full_turn：

def run_full_turn(agent, messages):

    current_agent = agent
    num_init_messages = len(messages)
    messages = messages.copy()

    while True:

        # turn python functions into tools and save a reverse map
        tool_schemas = [function_to_schema(tool) for tool in current_agent.tools]
        tools = {tool.__name__: tool for tool in current_agent.tools}

        # === 1. get openai completion ===
        response = client.chat.completions.create(
            model=agent.model,
            messages=[{"role": "system", "content": current_agent.instructions}]
            + messages,
            tools=tool_schemas or None,
        )
        message = response.choices[0].message
        messages.append(message)

        if message.content:  # print agent response
            print(f"{current_agent.name}:", message.content)

        if not message.tool_calls:  # if finished handling tool calls, break
            break

        # === 2. handle tool calls ===

        for tool_call in message.tool_calls:
            result = execute_tool_call(tool_call, tools, current_agent.name)

            if type(result) is Agent:  # if agent transfer, update current agent
                current_agent = result
                result = (
                    f"Transfered to {current_agent.name}. Adopt persona immediately."
                )

            result_message = {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            }
            messages.append(result_message)

    # ==== 3. return last agent used and new messages =====
    return Response(agent=current_agent, messages=messages[num_init_messages:])


def execute_tool_call(tool_call, tools, agent_name):
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    print(f"{agent_name}:", f"{name}({args})")

    return tools[name](**args)  # call corresponding function with provided arguments

让我们看一个包含更多Agent的示例。

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

最后，我们可以在一个循环中运行它（这在 Python 笔记本中无法运行，因此你可以尝试在单独的 Python 文件中运行）：

agent = triage_agent
messages = []

while True:
    user = input("User: ")
    messages.append({"role": "user", "content": user})

    response = run_full_turn(agent, messages)
    agent = response.agent
    messages.extend(response.messages)

Swarm

作为概念验证，我们将这些想法打包成了一个示例库，名为 Swarm。它仅作为示例，不应直接用于生产环境。不过，欢迎您借鉴其中的思路和代码来构建自己的实现！

常见问题

常见错误：1) 函数schema生成失败，如类型注解缺失或未知，需确保参数有类型注解或默认值。2) 工具调用参数解析错误，模型可能生成非法JSON，需捕获异常并重试。3) 交接循环，Agent间互相转移导致死循环，可设置最大迭代次数或检测重复转移。4) 模型未调用交接函数，可能因指令不明确，需在提示中明确说明何时转移。5) 上下文过长，多轮对话后超出模型限制，需截断或摘要历史。排查方法：打印每次模型响应和工具调用，检查消息序列；使用OpenAI日志；逐步调试。生产优化：使用异步调用、缓存、限流；将工具执行与模型调用解耦；监控token消耗和延迟。

企业应用扩展

可扩展方向：1) 增加记忆机制，让Agent跨会话记住用户偏好。2) 引入并行工具调用，提高效率。3) 实现更复杂的交接条件，如基于语义相似度或规则。4) 将Agent配置外部化，支持动态加载。5) 集成可观测性工具，追踪Agent决策路径。6) 使用更强大的模型或微调以适应特定领域。7) 将Swarm思想与LangChain等框架结合，利用现有生态。8) 添加安全机制，如权限控制、敏感操作审批。

总结

本文介绍如何通过“例程”和“交接”概念编排多个AI智能体，以处理复杂多变的用户流程。例程是自然语言指令与工具的组合，交接允许智能体动态转移对话控制权。文章从基础循环开始，逐步实现函数调用、Agent抽象和自动交接，最终提供示例库Swarm。读者将学会构建灵活、可控的多智能体系统，并了解其原理、架构及优化方向。

• --

内容来源与验证

• 内容等级：PREMIUM_A
• 质量等级：A
• 难度：中级
• 验证等级：V1
• 原文作者：ibigio
• 原文日期：2024-10-10
• 来源许可：MIT
• 来源版本：b0705ccd3cec4ec6d466bbea89998f788adaddd1
• 查看原文