Orchestrating Agents: Routines and Handoffs

When working with language models, achieving solid performance often requires a good prompt and the right tools. However, managing multiple unique flows can become complex. This guide introduces the concepts of routines and handoffs, demonstrating how they can be used to orchestrate multiple agents in a simple, powerful, and controllable way.

Setting Up the Environment

To begin, ensure you have the necessary imports and client setup.

from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
import json
 
client = OpenAI()

Understanding Routines

A "routine" is a set of steps defined in natural language, represented by a system prompt, along with the tools necessary to complete them. For example, a customer service routine might involve triaging a user issue, suggesting a fix, or providing a refund.

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.
"
    " - unless the user has already provided a reason.
"
    "2. Propose a fix (make one up).
"
    "3. ONLY if not satisfied, offer a refund.
"
    "4. If accepted, search for the ID and then execute refund."
    ""
)

Executing Routines

To execute a routine, implement a loop that handles user input, appends messages, and calls the model.

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

Handling Function Calls

Models require functions to be formatted as a function schema. Define a helper function to convert Python functions into the corresponding function schema.

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,
            },
        },
    }

Implementing Handoffs

Handoffs allow one agent to transfer an active conversation to another agent, similar to being transferred during a phone call. Define a basic class for an Agent and modify the code to support agent handoffs.

class Agent(BaseModel):
    name: str = "Agent"
    model: str = "gpt-4o-mini"
    instructions: str = "You are a helpful Agent"
    tools: list = []
 
def run_full_turn(agent, messages):
    current_agent = agent
    num_init_messages = len(messages)
    messages = messages.copy()
 
    while True:
        tool_schemas = [function_to_schema(tool) for tool in current_agent.tools]
        tools = {tool.__name__: tool for tool in current_agent.tools}
 
        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
 
        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"Transferred to {current_agent.name}. Adopt persona immediately."
                )
 
            result_message = {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            }
            messages.append(result_message)
 
    return Response(agent=current_agent, messages=messages[num_init_messages:])

Conclusion

By implementing routines and handoffs, you can effectively manage multiple agents, allowing for dynamic task handling and improved efficiency. This approach provides a robust framework for orchestrating complex workflows.

Reference: Orchestrating Agents: Routines and Handoffs by Ilan Bigio, OpenAI.