pip install git+https://github.com/openai/swarm.git

from swarm import Agent

support_agent = Agent(
    name="Support Agent",
    instructions="You are a helpful customer support agent. Answer general questions about the product.",
    functions=[get_product_info, get_order_status]
)

def get_product_info(product_id: str) -> str:
    """Get information about a product.

    Args:
        product_id: The product identifier
    """
    # Look up product in database
    return f"Product {product_id}: Widget Pro, $49.99"

def transfer_to_billing():
    """Transfer the conversation to the billing department."""
    return billing_agent

support_agent = Agent(
    name="Support Agent",
    instructions="...",
    functions=[transfer_to_billing]
)

from swarm import Swarm, Agent

# Initialize Swarm client
client = Swarm()

# Database simulation
ORDERS = {
    "ORD-123": {"status": "shipped", "tracking": "1Z999AA10123"},
    "ORD-456": {"status": "processing", "tracking": None}
}

ACCOUNTS = {
    "ACC-789": {"balance": 49.99, "plan": "Pro Monthly"}
}

# Tool functions
def get_order_status(order_id: str) -> str:
    """Look up the status of an order."""
    order = ORDERS.get(order_id)
    if not order:
        return f"Order {order_id} not found."
    return f"Order {order_id}: {order['status']}, Tracking: {order['tracking'] or 'Not yet available'}"


def get_account_balance(account_id: str) -> str:
    """Get the current account balance."""
    account = ACCOUNTS.get(account_id)
    if not account:
        return f"Account {account_id} not found."
    return f"Account {account_id}: ${account['balance']:.2f}, Plan: {account['plan']}"


def process_refund(order_id: str, reason: str) -> str:
    """Process a refund for an order."""
    if order_id not in ORDERS:
        return f"Cannot refund: Order {order_id} not found."
    return f"Refund initiated for {order_id}. Reason: {reason}. Amount will be credited in 3-5 days."


def create_support_ticket(issue: str, priority: str = "normal") -> str:
    """Create a support ticket for technical issues."""
    ticket_id = "TKT-" + str(hash(issue))[:6]
    return f"Created ticket {ticket_id} with {priority} priority. A technician will respond within 24 hours."


# Agent definitions
def transfer_to_tech_support():
    """Transfer to technical support for product issues."""
    return tech_support_agent


def transfer_to_billing():
    """Transfer to billing for payment and subscription issues."""
    return billing_agent


def transfer_to_triage():
    """Transfer back to triage for re-routing."""
    return triage_agent


# Triage Agent - First point of contact
triage_agent = Agent(
    name="Triage Agent",
    instructions="""You are the first point of contact for customer service.
    Your job is to understand the customer's issue and route them to the right agent.

    - For technical issues (product not working, bugs, how-to questions): transfer to tech support
    - For billing issues (payments, refunds, subscriptions): transfer to billing
    - For general questions: answer directly if possible

    Always be polite and confirm the transfer before making it.
    """,
    functions=[transfer_to_tech_support, transfer_to_billing]
)

# Tech Support Agent
tech_support_agent = Agent(
    name="Tech Support Agent",
    instructions="""You are a technical support specialist.
    Help customers with product issues, troubleshooting, and how-to questions.

    If you can't resolve the issue, create a support ticket.
    If the customer has a billing question, transfer to billing.
    """,
    functions=[create_support_ticket, get_order_status, transfer_to_billing, transfer_to_triage]
)

# Billing Agent
billing_agent = Agent(
    name="Billing Agent",
    instructions="""You are a billing specialist.
    Help customers with payments, refunds, and subscription management.

    You can:
    - Look up account balances
    - Process refunds for valid orders
    - Explain billing policies

    If the customer has a technical issue, transfer to tech support.
    """,
    functions=[get_account_balance, process_refund, get_order_status, transfer_to_tech_support, transfer_to_triage]
)

def run_customer_service():
    """Run an interactive customer service session."""
    print("Customer Service System")
    print("=" * 40)
    print("Type 'quit' to exit\n")

    messages = []
    agent = triage_agent

    while True:
        user_input = input("Customer: ").strip()
        if user_input.lower() == 'quit':
            break

        messages.append({"role": "user", "content": user_input})

        response = client.run(
            agent=agent,
            messages=messages
        )

        # Update messages with response
        messages = response.messages

        # Check if agent changed (handoff occurred)
        if response.agent != agent:
            print(f"\n[Transferred to {response.agent.name}]\n")
            agent = response.agent

        # Print agent response
        last_message = response.messages[-1]
        print(f"{agent.name}: {last_message['content']}\n")


if __name__ == "__main__":
    run_customer_service()

checkout_agent = Agent(
    name="Checkout Agent",
    instructions="""You guide customers through checkout.

    Follow these steps IN ORDER:
    1. Confirm the items in their cart
    2. Collect shipping address
    3. Collect payment information
    4. Confirm the order

    Do not skip steps. If the customer wants to modify their cart,
    start over from step 1.
    """,
    functions=[get_cart, update_shipping, process_payment, confirm_order]
)

from swarm import Agent

def get_customer_context(context_variables):
    """Access customer info from context."""
    customer_id = context_variables.get("customer_id")
    return f"Customer ID: {customer_id}"

support_agent = Agent(
    name="Support",
    instructions="You have access to customer context.",
    functions=[get_customer_context]
)

# Run with context
response = client.run(
    agent=support_agent,
    messages=[{"role": "user", "content": "Who am I?"}],
    context_variables={"customer_id": "CUST-12345", "plan": "Premium"}
)

def safe_get_order(order_id: str) -> str:
    """Get order status with error handling."""
    try:
        order = database.get_order(order_id)
        return format_order(order)
    except OrderNotFoundError:
        return f"Order {order_id} was not found. Please check the order ID."
    except DatabaseError:
        return "I'm having trouble accessing order information right now. Please try again."

response = client.run(
    agent=agent,
    messages=messages,
    stream=True
)

for chunk in response:
    if chunk.get("content"):
        print(chunk["content"], end="", flush=True)