Routing Agent — Python (implementación completa con LLM)

examples/
└── agent-patterns/
    └── routing-agent/
        └── python/
            ├── main.py           # Route -> Delegate -> (optional reroute) -> Finalize
            ├── llm.py            # router + final synthesis
            ├── gateway.py        # policy boundary: route validation + delegation control
            ├── tools.py          # deterministic specialists (billing/technical/sales)
            └── requirements.txt

git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd agentpatterns

cd examples/agent-patterns/routing-agent/python
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

export OPENAI_API_KEY="sk-..."
# optional:
# export OPENAI_MODEL="gpt-4.1-mini"
# export OPENAI_TIMEOUT_SECONDS="60"

python main.py

from __future__ import annotations

from typing import Any

USERS = {
    42: {"id": 42, "name": "Anna", "country": "US", "tier": "pro"},
    7: {"id": 7, "name": "Max", "country": "US", "tier": "free"},
}

BILLING = {
    42: {
        "currency": "USD",
        "plan": "pro_monthly",
        "price_usd": 49.0,
        "days_since_first_payment": 10,
    },
    7: {
        "currency": "USD",
        "plan": "free",
        "price_usd": 0.0,
        "days_since_first_payment": 120,
    },
}


def _extract_user_id(ticket: str) -> int:
    if "user_id=7" in ticket:
        return 7
    return 42


def _contains_any(text: str, keywords: list[str]) -> bool:
    lowered = text.lower()
    return any(keyword in lowered for keyword in keywords)


def billing_specialist(ticket: str) -> dict[str, Any]:
    if not _contains_any(ticket, ["refund", "charge", "billing", "invoice"]):
        return {
            "status": "needs_reroute",
            "reason": "ticket_not_billing",
            "domain": "billing",
        }

    user_id = _extract_user_id(ticket)
    user = USERS.get(user_id)
    billing = BILLING.get(user_id)
    if not user or not billing:
        return {"status": "done", "domain": "billing", "error": "user_not_found"}

    is_refundable = (
        billing["plan"] == "pro_monthly" and billing["days_since_first_payment"] <= 14
    )
    refund_amount = billing["price_usd"] if is_refundable else 0.0

    return {
        "status": "done",
        "domain": "billing",
        "result": {
            "user_name": user["name"],
            "plan": billing["plan"],
            "currency": billing["currency"],
            "refund_eligible": is_refundable,
            "refund_amount_usd": refund_amount,
            "reason": "Pro monthly subscriptions are refundable within 14 days.",
        },
    }


def technical_specialist(ticket: str) -> dict[str, Any]:
    if not _contains_any(ticket, ["error", "bug", "incident", "api", "latency"]):
        return {
            "status": "needs_reroute",
            "reason": "ticket_not_technical",
            "domain": "technical",
        }

    return {
        "status": "done",
        "domain": "technical",
        "result": {
            "incident_id": "INC-4021",
            "service": "public-api",
            "state": "mitigated",
            "next_update_in_minutes": 30,
        },
    }


def sales_specialist(ticket: str) -> dict[str, Any]:
    if not _contains_any(ticket, ["price", "pricing", "quote", "plan", "discount"]):
        return {
            "status": "needs_reroute",
            "reason": "ticket_not_sales",
            "domain": "sales",
        }

    return {
        "status": "done",
        "domain": "sales",
        "result": {
            "recommended_plan": "team_plus",
            "currency": "USD",
            "monthly_price_usd": 199.0,
            "reason": "Best fit for teams that need priority support and usage controls.",
        },
    }

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from typing import Any, Callable


class StopRun(Exception):
    def __init__(self, reason: str):
        super().__init__(reason)
        self.reason = reason


@dataclass(frozen=True)
class Budget:
    max_route_attempts: int = 3
    max_delegations: int = 3
    max_seconds: int = 60


def _stable_json(value: Any) -> str:
    if value is None or isinstance(value, (bool, int, float, str)):
        return json.dumps(value, ensure_ascii=True, sort_keys=True)
    if isinstance(value, list):
        return "[" + ",".join(_stable_json(item) for item in value) + "]"
    if isinstance(value, dict):
        parts = []
        for key in sorted(value):
            parts.append(
                json.dumps(str(key), ensure_ascii=True) + ":" + _stable_json(value[key])
            )
        return "{" + ",".join(parts) + "}"
    return json.dumps(str(value), ensure_ascii=True)


def _normalize_for_hash(value: Any) -> Any:
    if isinstance(value, str):
        return " ".join(value.strip().split())
    if isinstance(value, list):
        return [_normalize_for_hash(item) for item in value]
    if isinstance(value, dict):
        return {str(key): _normalize_for_hash(value[key]) for key in sorted(value)}
    return value


def _normalize_ticket(value: str) -> str:
    return " ".join(value.strip().split())


def args_hash(args: dict[str, Any]) -> str:
    normalized = _normalize_for_hash(args or {})
    raw = _stable_json(normalized)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:12]


def validate_route_action(
    action: Any,
    *,
    allowed_routes: set[str],
    previous_target: str | None = None,
    previous_status: str | None = None,
) -> dict[str, Any]:
    if not isinstance(action, dict):
        raise StopRun("invalid_route:not_object")

    kind = action.get("kind")
    if kind == "invalid":
        raise StopRun("invalid_route:non_json")
    if kind != "route":
        raise StopRun("invalid_route:bad_kind")

    allowed_keys = {"kind", "target", "args"}
    if set(action.keys()) - allowed_keys:
        raise StopRun("invalid_route:extra_keys")

    target = action.get("target")
    if not isinstance(target, str) or not target.strip():
        raise StopRun("invalid_route:missing_target")
    target = target.strip()
    if target not in allowed_routes:
        raise StopRun(f"invalid_route:route_not_allowed:{target}")

    args = action.get("args", {})
    if args is None:
        args = {}
    if not isinstance(args, dict):
        raise StopRun("invalid_route:bad_args")
    ticket = args.get("ticket")
    if not isinstance(ticket, str) or not ticket.strip():
        raise StopRun("invalid_route:missing_ticket")
    ticket = _normalize_ticket(ticket)
    normalized_args = {**args, "ticket": ticket}

    if previous_status == "needs_reroute" and target == previous_target:
        raise StopRun("invalid_route:repeat_target_after_reroute")

    return {"kind": "route", "target": target, "args": normalized_args}


class RouteGateway:
    def __init__(
        self,
        *,
        allow: set[str],
        registry: dict[str, Callable[..., dict[str, Any]]],
        budget: Budget,
    ):
        self.allow = set(allow)
        self.registry = registry
        self.budget = budget
        self.delegations = 0
        self.seen_routes: set[str] = set()

    def call(self, target: str, args: dict[str, Any]) -> dict[str, Any]:
        self.delegations += 1
        if self.delegations > self.budget.max_delegations:
            raise StopRun("max_delegations")

        if target not in self.allow:
            raise StopRun(f"route_denied:{target}")

        worker = self.registry.get(target)
        if worker is None:
            raise StopRun(f"route_missing:{target}")

        signature = f"{target}:{args_hash(args)}"
        if signature in self.seen_routes:
            raise StopRun("loop_detected")
        self.seen_routes.add(signature)

        try:
            return worker(**args)
        except TypeError as exc:
            raise StopRun(f"route_bad_args:{target}") from exc
        except Exception as exc:
            raise StopRun(f"route_error:{target}") from exc

from __future__ import annotations

import json
import os
from typing import Any

from openai import APIConnectionError, APITimeoutError, OpenAI

MODEL = os.getenv("OPENAI_MODEL", "gpt-4.1-mini")
LLM_TIMEOUT_SECONDS = float(os.getenv("OPENAI_TIMEOUT_SECONDS", "60"))


class LLMTimeout(Exception):
    pass


class LLMEmpty(Exception):
    pass


ROUTER_SYSTEM_PROMPT = """
You are a routing decision engine.
Return only one JSON object in this exact shape:
{"kind":"route","target":"<route_name>","args":{"ticket":"..."}}

Rules:
- Choose exactly one target from available_routes.
- Never choose targets from forbidden_targets.
- Keep args minimal and valid for that target.
- If previous attempts failed with needs_reroute, choose a different target.
- Respect routing budgets and avoid unnecessary retries.
- Do not answer the user directly.
- Never output markdown or extra keys.
""".strip()

FINAL_SYSTEM_PROMPT = """
You are a support response assistant.
Write a short final answer in English for a US customer.
Use only evidence from delegated specialist observation.
Include: selected specialist, final decision, and one reason.
For billing refunds, include amount in USD when available.
""".strip()

ROUTE_CATALOG = [
    {
        "name": "billing_specialist",
        "description": "Handle refunds, charges, invoices, and billing policy",
        "args": {"ticket": "string"},
    },
    {
        "name": "technical_specialist",
        "description": "Handle errors, incidents, API issues, and outages",
        "args": {"ticket": "string"},
    },
    {
        "name": "sales_specialist",
        "description": "Handle pricing, plan recommendations, and quotes",
        "args": {"ticket": "string"},
    },
]


def _get_client() -> OpenAI:
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        raise EnvironmentError(
            "OPENAI_API_KEY is not set. Run: export OPENAI_API_KEY='sk-...'"
        )
    return OpenAI(api_key=api_key)


def _build_state_summary(history: list[dict[str, Any]]) -> dict[str, Any]:
    routes_used = [
        step.get("route", {}).get("target")
        for step in history
        if isinstance(step, dict)
        and isinstance(step.get("route"), dict)
        and step.get("route", {}).get("kind") == "route"
    ]
    routes_used_unique = list(dict.fromkeys(route for route in routes_used if route))
    last_route_target = routes_used[-1] if routes_used else None
    last_observation = history[-1].get("observation") if history else None
    last_observation_status = (
        last_observation.get("status") if isinstance(last_observation, dict) else None
    )
    return {
        "attempts_completed": len(history),
        "routes_used_unique": routes_used_unique,
        "last_route_target": last_route_target,
        "last_observation_status": last_observation_status,
        "last_observation": last_observation,
    }


def decide_route(
    goal: str,
    history: list[dict[str, Any]],
    *,
    max_route_attempts: int,
    remaining_attempts: int,
    forbidden_targets: list[str],
) -> dict[str, Any]:
    recent_history = history[-3:]
    payload = {
        "goal": goal,
        "budgets": {
            "max_route_attempts": max_route_attempts,
            "remaining_attempts": remaining_attempts,
        },
        "forbidden_targets": forbidden_targets,
        "state_summary": _build_state_summary(history),
        "recent_history": recent_history,
        "available_routes": ROUTE_CATALOG,
    }

    client = _get_client()
    try:
        completion = client.chat.completions.create(
            model=MODEL,
            temperature=0,
            timeout=LLM_TIMEOUT_SECONDS,
            response_format={"type": "json_object"},
            messages=[
                {"role": "system", "content": ROUTER_SYSTEM_PROMPT},
                {"role": "user", "content": json.dumps(payload, ensure_ascii=True)},
            ],
        )
    except (APITimeoutError, APIConnectionError) as exc:
        raise LLMTimeout("llm_timeout") from exc

    text = completion.choices[0].message.content or "{}"
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"kind": "invalid", "raw": text}


def compose_final_answer(
    goal: str, selected_route: str, history: list[dict[str, Any]]
) -> str:
    payload = {
        "goal": goal,
        "selected_route": selected_route,
        "history": history,
    }

    client = _get_client()
    try:
        completion = client.chat.completions.create(
            model=MODEL,
            temperature=0,
            timeout=LLM_TIMEOUT_SECONDS,
            messages=[
                {"role": "system", "content": FINAL_SYSTEM_PROMPT},
                {"role": "user", "content": json.dumps(payload, ensure_ascii=True)},
            ],
        )
    except (APITimeoutError, APIConnectionError) as exc:
        raise LLMTimeout("llm_timeout") from exc

    text = completion.choices[0].message.content or ""
    text = text.strip()
    if not text:
        raise LLMEmpty("llm_empty")
    return text

from __future__ import annotations

import json
import time
from typing import Any

from gateway import Budget, RouteGateway, StopRun, args_hash, validate_route_action
from llm import LLMEmpty, LLMTimeout, compose_final_answer, decide_route
from tools import billing_specialist, sales_specialist, technical_specialist

GOAL = (
    "User Anna (user_id=42) asks: Can I get a refund for my pro_monthly subscription "
    "charged 10 days ago? Route to the correct specialist and provide a short final answer."
)

BUDGET = Budget(max_route_attempts=3, max_delegations=3, max_seconds=60)

ROUTE_REGISTRY = {
    "billing_specialist": billing_specialist,
    "technical_specialist": technical_specialist,
    "sales_specialist": sales_specialist,
}

ALLOWED_ROUTE_TARGETS_POLICY = {
    "billing_specialist",
    "technical_specialist",
    "sales_specialist",
}

ALLOWED_ROUTE_TARGETS_EXECUTION = {
    "billing_specialist",
    "technical_specialist",
    "sales_specialist",
}


def run_routing(goal: str) -> dict[str, Any]:
    started = time.monotonic()
    trace: list[dict[str, Any]] = []
    history: list[dict[str, Any]] = []

    gateway = RouteGateway(
        allow=ALLOWED_ROUTE_TARGETS_EXECUTION,
        registry=ROUTE_REGISTRY,
        budget=BUDGET,
    )

    for attempt in range(1, BUDGET.max_route_attempts + 1):
        elapsed = time.monotonic() - started
        if elapsed > BUDGET.max_seconds:
            return {
                "status": "stopped",
                "stop_reason": "max_seconds",
                "trace": trace,
                "history": history,
            }

        previous_step = history[-1] if history else None
        previous_observation = (
            previous_step.get("observation")
            if isinstance(previous_step, dict)
            else None
        )
        previous_route = previous_step.get("route") if isinstance(previous_step, dict) else None
        previous_status = (
            previous_observation.get("status")
            if isinstance(previous_observation, dict)
            else None
        )
        previous_target = (
            previous_route.get("target")
            if isinstance(previous_route, dict)
            else None
        )
        forbidden_targets = (
            [previous_target]
            if previous_status == "needs_reroute" and isinstance(previous_target, str)
            else []
        )

        try:
            raw_route = decide_route(
                goal=goal,
                history=history,
                max_route_attempts=BUDGET.max_route_attempts,
                remaining_attempts=(BUDGET.max_route_attempts - attempt + 1),
                forbidden_targets=forbidden_targets,
            )
        except LLMTimeout:
            return {
                "status": "stopped",
                "stop_reason": "llm_timeout",
                "phase": "route",
                "trace": trace,
                "history": history,
            }

        try:
            route_action = validate_route_action(
                raw_route,
                allowed_routes=ALLOWED_ROUTE_TARGETS_POLICY,
                previous_target=previous_target,
                previous_status=previous_status,
            )
        except StopRun as exc:
            return {
                "status": "stopped",
                "stop_reason": exc.reason,
                "phase": "route",
                "raw_route": raw_route,
                "trace": trace,
                "history": history,
            }

        target = route_action["target"]
        route_args = route_action["args"]

        try:
            observation = gateway.call(target, route_args)
            trace.append(
                {
                    "attempt": attempt,
                    "target": target,
                    "args_hash": args_hash(route_args),
                    "ok": True,
                }
            )
        except StopRun as exc:
            trace.append(
                {
                    "attempt": attempt,
                    "target": target,
                    "args_hash": args_hash(route_args),
                    "ok": False,
                    "stop_reason": exc.reason,
                }
            )
            return {
                "status": "stopped",
                "stop_reason": exc.reason,
                "phase": "delegate",
                "route": route_action,
                "trace": trace,
                "history": history,
            }

        history.append(
            {
                "attempt": attempt,
                "route": route_action,
                "observation": observation,
            }
        )

        observation_status = observation.get("status")
        if trace:
            trace[-1]["observation_status"] = observation_status
            if isinstance(observation, dict) and observation.get("domain"):
                trace[-1]["domain"] = observation.get("domain")
        if observation_status == "needs_reroute":
            continue
        if observation_status != "done":
            return {
                "status": "stopped",
                "stop_reason": "route_bad_observation",
                "phase": "delegate",
                "route": route_action,
                "expected_statuses": ["needs_reroute", "done"],
                "received_status": observation_status,
                "bad_observation": observation,
                "trace": trace,
                "history": history,
            }

        try:
            answer = compose_final_answer(
                goal=goal,
                selected_route=target,
                history=history,
            )
        except LLMTimeout:
            return {
                "status": "stopped",
                "stop_reason": "llm_timeout",
                "phase": "finalize",
                "route": route_action,
                "trace": trace,
                "history": history,
            }
        except LLMEmpty:
            return {
                "status": "stopped",
                "stop_reason": "llm_empty",
                "phase": "finalize",
                "route": route_action,
                "trace": trace,
                "history": history,
            }

        return {
            "status": "ok",
            "stop_reason": "success",
            "selected_route": target,
            "answer": answer,
            "trace": trace,
            "history": history,
        }

    return {
        "status": "stopped",
        "stop_reason": "max_route_attempts",
        "trace": trace,
        "history": history,
    }


def main() -> None:
    result = run_routing(GOAL)
    print(json.dumps(result, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

openai==2.21.0

{
  "status": "ok",
  "stop_reason": "success",
  "selected_route": "billing_specialist",
  "answer": "The billing specialist reviewed your request and confirmed that your pro_monthly subscription charged 10 days ago is eligible for a refund. You will receive a refund of $49.00 because pro monthly subscriptions are refundable within 14 days.",
  "trace": [
    {
      "attempt": 1,
      "target": "billing_specialist",
      "args_hash": "5e89...",
      "ok": true,
      "observation_status": "done",
      "domain": "billing"
    }
  ],
  "history": [{...}]
}