Hierarchical process: a manager for the crew

from crewai.tools import tool

ORDERS = {"A17": "shipped on 3 March", "C40": "waiting for stock"}


@tool
def lookup_order(order_id: str) -> str:
    """Look up an order's shipping status by its id, such as A17."""
    status = ORDERS.get(order_id)
    return f"{order_id} {status}." if status else f"{order_id} is not an order we have."

import json
import os
import re

from crewai import BaseLLM

os.environ["OTEL_SDK_DISABLED"] = "true"
os.environ["CREWAI_DISABLE_TELEMETRY"] = "true"
os.environ["CREWAI_TRACING_ENABLED"] = "false"
os.environ["CREWAI_DISABLE_VERSION_CHECK"] = "true"


class ShopLLM(BaseLLM):
    script: list = []

    def supports_function_calling(self):
        return True

    def call(self, messages, tools=None, **kwargs):
        if isinstance(messages, str):
            messages = [{"role": "user", "content": messages}]
        if self.script:
            return self.script.pop(0)
        names = [t["function"]["name"] for t in tools or []]
        return self.decide(messages, names)

    def decide(self, messages, tools):
        last = messages[-1]
        if last["role"] == "tool":
            return last["content"]
        text = last["content"]
        orders = re.findall(r"\b[A-Z]\d+\b", text)
        wanted = "refund_order" if "refund" in text.lower() else "lookup_order"
        matches = [name for name in tools if name.endswith(wanted)]
        if orders and matches:
            args = json.dumps({"order_id": orders[0]})
            return [{"id": f"call_{orders[0]}", "type": "function",
                     "function": {"name": matches[0], "arguments": args}}]
        if "working with:" in text:
            context = text.split("working with:")[1].strip().split("\n\n")[0]
            return f"Dear customer, {context}"
        if orders:
            return f"I have no way to look up {orders[0]} yet."
        return "Hello. Which order is this about?"

import json
import re

from shop_llm import ShopLLM


class ManagerLLM(ShopLLM):
    def call(self, messages, tools=None, **kwargs):
        found = {t["function"]["name"]: t["function"] for t in tools or []}
        delegate = found.get("delegate_work_to_coworker")
        if delegate and messages[-1]["role"] == "user":
            names = delegate["description"].split("coworkers: ")[1].split("\n")[0]
            coworker = next(n for n in names.split(", ") if "order" in n.lower())
            job = re.search(r"Current Task: (.*)", messages[-1]["content"]).group(1)
            args = {"task": job, "context": "A customer wrote in.", "coworker": coworker}
            return [{"id": "delegate_1", "type": "function", "function": {
                "name": "delegate_work_to_coworker", "arguments": json.dumps(args)}}]
        return super().call(messages, tools=tools, **kwargs)