DecisionModel

from swarms import DecisionModel

# Reads TYPESAFE_API_KEY and calls TypeSafe's jev-latest.
model = DecisionModel()

urgency = model.noul(
    state="Checkout has been failing for every customer for the last hour.",
    instructions="Is this support request urgent?",
)
print(f"Urgency: {urgency:.2f}")

from swarms import DecisionModel, get_decision_models

DecisionModel(
    model_name: str = "jev-latest",
    api_key: Optional[str] = None,
    api_key_env: Optional[str] = None,
    base_url: Optional[str] = None,
    endpoint: Optional[str] = None,
    timeout: float = 30.0,
    max_retries: int = 3,
    headers: Optional[Dict[str, str]] = None,
    extra_body: Optional[Dict[str, Any]] = None,
)

questions = {
    "department": {
        "type": "choice",
        "instructions": "Which team should handle this ticket?",
        "criteria": {
            "billing": "Payment or subscription issues",
            "technical": "Bugs or integration problems",
        },
    },
    "frustration": {
        "type": "score",
        "instructions": "How frustrated does the customer appear?",
        "criteria": ["Calm", "Frustrated but civil", "Very angry"],
    },
    "is_urgent": {
        "type": "noul",
        "instructions": "The message conveys urgency or time-sensitivity.",
    },
}

def run(
    state: Union[str, Dict[str, Any], List[Any]],
    questions: Dict[str, Dict[str, Any]],
) -> Dict[str, Any]

result = model.run(state={"message": "Help!", "plan": "enterprise"}, questions=questions)

department = result["answers"]["department"]
print(department["choice"], department["confidence"], department["probabilities"])
print(result["answers"]["is_urgent"]["noul"])

async def arun(
    state: Union[str, Dict[str, Any], List[Any]],
    questions: Dict[str, Dict[str, Any]],
) -> Dict[str, Any]

def choice(
    state: Union[str, Dict[str, Any], List[Any]],
    instructions: Any,
    criteria: Dict[str, Any],
) -> Dict[str, Any]

answer = model.choice(
    state="I was charged twice for my March invoice.",
    instructions="Which team should handle this?",
    criteria={"billing": None, "technical": None, "sales": None},
)
print(answer["choice"])

def score(
    state: Union[str, Dict[str, Any], List[Any]],
    instructions: Any,
    criteria: List[Any],
) -> Dict[str, Any]

def noul(
    state: Union[str, Dict[str, Any], List[Any]],
    instructions: Any,
    criteria: Optional[Dict[str, Any]] = None,
) -> float

p = model.noul(
    state="Our webhook endpoint returns 500 since your API update.",
    instructions="Is this request time-sensitive?",
    criteria={"true": "Blocks the customer now", "false": "Can wait"},
)

def list_models() -> List[str]

def close() -> None

def build_headers() -> Dict[str, str]

def build_payload(
    state: Union[str, Dict[str, Any], List[Any]],
    questions: Dict[str, Dict[str, Any]],
) -> Dict[str, Any]

def parse_response(
    data: Dict[str, Any],
    questions: Dict[str, Dict[str, Any]],
) -> Dict[str, Any]

from swarms import DecisionModel


class RankingProvider(DecisionModel):
    def build_headers(self):
        return {"X-Api-Key": self._api_key}

    def build_payload(self, state, questions):
        return {"input": state, "schema": questions}

    def parse_response(self, data, questions):
        return {"answers": data["output"]}


model = RankingProvider(
    model_name="rank-1",
    api_key="...",
    base_url="https://decisions.example.com",
    endpoint="/decide",
)

def get_decision_models() -> List[str]

from swarms import get_decision_models

print(get_decision_models())
# ['jev-latest', 'jev-preview', 'jev-1.13.0', 'clef', 'clef-flash']

from swarms import Agent, DecisionModel

agents = [
    Agent(
        agent_name="Billing-Agent",
        agent_description="Refunds, invoices, failed payments and subscription changes.",
        model_name="gpt-5.4",
        max_loops=1,
    ),
    Agent(
        agent_name="Technical-Agent",
        agent_description="Bugs, API errors, integrations and outages.",
        model_name="gpt-5.4",
        max_loops=1,
    ),
]
agents_by_name = {agent.agent_name: agent for agent in agents}
router = DecisionModel(model_name="jev-latest")


def route(task: str) -> str:
    answers = router.run(
        state=task,
        questions={
            "agent": {
                "type": "choice",
                "instructions": "Which agent should handle this request?",
                "criteria": {a.agent_name: a.agent_description for a in agents},
            },
            "in_scope": {
                "type": "noul",
                "instructions": "This is a request a software company's support team should handle.",
            },
        },
    )["answers"]

    if answers["in_scope"]["noul"] < 0.5:
        return "Declined: out of scope for support."
    decision = answers["agent"]
    if decision["confidence"] < 0.5:
        return f"Escalated to a human: {decision['probabilities']}"
    return agents_by_name[decision["choice"]].run(task)


print(route("Our webhook endpoint returns 500 since your API update this morning."))

import asyncio

from swarms import DecisionModel

model = DecisionModel()
questions = {
    "python": {"type": "noul", "instructions": "The resume shows Python in production."},
    "seniority": {
        "type": "score",
        "instructions": "How senior is this candidate?",
        "criteria": ["Junior", "Mid-level", "Senior", "Staff or above"],
    },
}
resumes = ["Resume text 1...", "Resume text 2...", "Resume text 3..."]


async def screen_all() -> list:
    limit = asyncio.Semaphore(20)

    async def screen(resume: str) -> dict:
        async with limit:
            response = await model.arun(state=resume, questions=questions)
            return response["answers"]

    return await asyncio.gather(*[screen(r) for r in resumes])


for answers in asyncio.run(screen_all()):
    print(answers["python"]["noul"], answers["seniority"]["score"])