Guard a model call

You put a decision between your application and a language model: every prompt, and every reply, is evaluated against a pack before it moves on, and your logs record a receipt instead of the prompt. With the built-in example pack (data protection), prompts that carry fatal data never leave your process and prompts with contact details go out redacted.

What you'll build#

A PromptGateway class with one method, ask(prompt, request_id). It:

• raises PromptBlocked for a fatal action, without calling the model;
• sends remedy.clean_text to the model for permit and redact actions;
• checks the model's reply on the way back;
• writes one structured log line per decision with the action, rule IDs and receipt_hash, and never the prompt.

The model call is a stub, call_model(). Replace it with your own client.

Prerequisites#

• The runtime installed. See Installation.
• Make your first decision, or equivalent familiarity with the three outcome classes.

Steps#

1. Stub the model call

   Keep the model client behind one function. That makes it obvious that the only way a prompt reaches the network is through the gateway.

   import json
   import logging
   from dataclasses import dataclass
   
   from helixor_runtime import HelixorEngine
   
   logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
   log = logging.getLogger("prompt_gateway")
   
   
   def call_model(prompt: str) -> str:
       """Stand-in for your language model client."""
       return f"(model reply to {len(prompt)} chars)"
   

2. Define what a caller gets back

   A blocked prompt is an exception, not a return value, so a caller cannot forward it by mistake. The exception carries the action, rule IDs and receipt, which is enough to explain the refusal without echoing the data.

   class PromptBlocked(Exception):
       def __init__(self, action: str, rule_ids: list[str], receipt_hash: str):
           super().__init__(f"prompt blocked by {action}")
           self.action = action
           self.rule_ids = rule_ids
           self.receipt_hash = receipt_hash
   
   
   @dataclass
   class GatewayReply:
       text: str
       forwarded_prompt: str
       receipt_hash: str
   

3. Decide, log, then forward

   Evaluate the prompt, log the decision, and branch on the action. For permit actions remedy.clean_text equals the input, so you can forward it in both non-blocking cases without a second branch.

   class PromptGateway:
       def __init__(self) -> None:
           self.engine = HelixorEngine()
   
       def ask(self, prompt: str, request_id: str) -> GatewayReply:
           decision = self.engine.evaluate(prompt)
           rule_ids = [t.rule_id for t in decision.triggers]
   
           # One structured log line per decision. Never log the prompt,
           # the clean text, or trigger.matched_items.
           log.info(json.dumps({
               "request_id": request_id,
               "pack_id": decision.pack_id,
               "action": decision.action,
               "rule_ids": rule_ids,
               "receipt_hash": decision.receipt_hash,
               "latency_us": decision.latency_us,
           }))
   
           if decision.action.startswith("block"):
               raise PromptBlocked(decision.action, rule_ids, decision.receipt_hash)
   
           # permit_* returns the text unchanged; redact_* returns it repaired.
           forwarded = decision.remedy.clean_text
           reply = call_model(forwarded)
   
           # Check the reply on the way back, too.
           outbound = self.engine.evaluate(reply)
           if outbound.action.startswith("block"):
               raise PromptBlocked(outbound.action, [t.rule_id for t in outbound.triggers], outbound.receipt_hash)
           return GatewayReply(outbound.remedy.clean_text, forwarded, decision.receipt_hash)
   

   Never log matched_items

   trigger.matched_items holds the raw values the pack found; with the example pack, the full SSN. Treat it like the payload itself: use it in memory if you must, never write it to a log, a metric label or an error message.

4. Drive it with three prompts

   if __name__ == "__main__":
       gateway = PromptGateway()
       prompts = [
           ("req-1", "Summarize the Q3 planning notes in three bullets."),
           ("req-2", "Draft a reply to dana.reyes@example.com and offer a call at (415) 555-0199."),
           ("req-3", "Check eligibility for applicant with SSN 123-45-6789."),
       ]
       for request_id, prompt in prompts:
           try:
               reply = gateway.ask(prompt, request_id)
               print(f"{request_id} forwarded: {reply.forwarded_prompt!r}")
               print(f"{request_id} reply    : {reply.text}")
           except PromptBlocked as exc:
               print(f"{request_id} BLOCKED  : {exc.action} {exc.rule_ids} receipt={exc.receipt_hash}")
   

   python gateway.py
   

   INFO {"request_id": "req-1", "pack_id": "compliance.regulatory_pii_guard.v1", "action": "permit_clean_payload", "rule_ids": [], "receipt_hash": "hx_proof_17c28207c786fa419103a0db", "latency_us": 47.54}
   req-1 forwarded: 'Summarize the Q3 planning notes in three bullets.'
   req-1 reply    : (model reply to 49 chars)
   INFO {"request_id": "req-2", "pack_id": "compliance.regulatory_pii_guard.v1", "action": "redact_and_permit_contact_pii", "rule_ids": ["RULE-GDPR-EMAIL-REDACT", "RULE-TCPA-PHONE-REDACT"], "receipt_hash": "hx_proof_c8e6f52e5613d15c5b33b3bc", "latency_us": 81.04}
   req-2 forwarded: 'Draft a reply to [REDACTED_EMAIL] and offer a call at [REDACTED_PHONE].'
   req-2 reply    : (model reply to 71 chars)
   INFO {"request_id": "req-3", "pack_id": "compliance.regulatory_pii_guard.v1", "action": "block_glba_ssn_leakage", "rule_ids": ["RULE-GLBA-SSN-BLOCK"], "receipt_hash": "hx_proof_2478f4a3a462b6198eb21498", "latency_us": 24.58}
   req-3 BLOCKED  : block_glba_ssn_leakage ['RULE-GLBA-SSN-BLOCK'] receipt=hx_proof_2478f4a3a462b6198eb21498
   

   The latency_us values vary by machine; the first call in a process is the slowest. The model was called twice: req-3 stopped at the gateway.

How it works#

The gateway is a policy enforcement point in your own process. The runtime makes the decision locally: tokens_spent and egress_bytes are always 0, so screening the prompt adds no model call and sends nothing over the network. Only after the decision does your code choose whether anything leaves.

The log line is your audit trail. receipt_hash is a fingerprint of the pack, the action, the rules that fired and a SHA-256 of the input. If you later need to show what the gateway decided for a request, you can recompute the receipt from the original prompt and compare it with the one you logged, without ever having stored the prompt. See Receipts for exactly what it does and does not prove.

Checking the reply matters because a model can produce content that was never in the prompt; with the example pack, sensitive data. The same three-way branch applies: block, or forward the clean text.

Variations#

• Streaming replies. If your model streams tokens, evaluate the stream instead of the final string. See Decide on a stream as it arrives.
• Other languages. If the calling service is not Python, run the gateway decision as a local service. See Serve decisions to other languages.
• Hosted escalation. Some decisions need more than a local pack. See Hosted escalation.

Next steps#