Escalating to hosted reasoning

The embedded runtime answers questions that local rules can settle. Some decisions need evidence, multi-step reasoning, or a calibrated estimate of whether the answer is right. Those go to the hosted Helixor reasoning API.

Embedded or hosted?#

A common pattern is to run the embedded decision first, on everything, and escalate only the cases it cannot settle. The example below uses the built-in example pack (data protection) as that first step, so only its redacted text ever leaves your process.

The decide request#

POST/v1/decide takes a goal, the questions to answer, and the evidence to answer them from:

{
  "goal": "Route the support ticket",
  "questions": [
    {
      "question_id": "q1",
      "family_id": "support.ticket.route",
      "kind": "choice",
      "prompt": "Which team should handle this ticket?",
      "options": [
        {"option_id": "billing", "criteria": "Charges, invoices, refunds"},
        {"option_id": "technical", "criteria": "Errors, outages, integrations"}
      ]
    }
  ],
  "evidence": [
    {"id": "ticket", "text": "I was charged twice for [REDACTED_CARD_PAN]", "source": "helpdesk"}
  ],
  "require": {"proof": true}
}

Question kind is choice, score or boolean. require can also set min_confidence and max_latency_ms. Authenticate with your API key in the X-API-Key header. The endpoint URL and key come with your hosted API access.

The response#

{
  "verdicts": {
    "q1": {"verdict": "answer", "value": "billing", "p_correct": 0.97,
           "proof": {"...": "..."}, "reason": "..."}
  },
  "composed": {"verdict": "answer", "rule_ids": [], "band": "..."},
  "proof_id": "...",
  "latency_ms": 42
}

Each question gets a verdict of answer, abstain or refuse. Act only on answer, and treat abstain as "a person decides". Keep proof_id with your record, as you keep receipt_hash for embedded decisions.

Combining the two#

import os
import httpx
from helixor_runtime import HelixorEngine

ENGINE = HelixorEngine()
API = httpx.Client(
    base_url=os.environ["HELIXOR_API_URL"],        # your reasoning API endpoint
    headers={"X-API-Key": os.environ["HELIXOR_API_KEY"]},
    timeout=5.0,
)

def route_ticket(text: str) -> str:
    guard = ENGINE.evaluate(text)
    if guard.action.startswith("block_") or any(t.severity == "FATAL" for t in guard.triggers):
        return "manual_review"                      # never escalate blocked content
    evidence = guard.remedy.clean_text               # only redacted text leaves the process
    resp = API.post("/v1/decide", json=build_request(evidence))
    resp.raise_for_status()                          # an error is not an answer
    verdict = resp.json()["verdicts"]["q1"]
    return verdict["value"] if verdict["verdict"] == "answer" else "manual_review"