Patter SDK Guide to Building a Restaurant Booking Phone Agent with Dynamic Variables, Guardrails, Latency Dashboards, and Eval Checks

from __future__ import annotations
import sys, subprocess, importlib, examine, time, json, re, random, textwrap, os
from dataclasses import dataclass, discipline
from statistics import median
def _try_install(pkg: str) -> None:
   attempt:
       subprocess.run([sys.executable, "-m", "pip", "install", "-q", pkg],
                      examine=False, timeout=600)
   besides Exception:
       cross
def _load_patter():
   """Return the actual getpatter module if importable, else None."""
   for title in ("patter", "getpatter"):
       attempt:
           return importlib.import_module(title)
       besides Exception:
           proceed
   return None
_PATTER = _load_patter()
if _PATTER is None:
   _try_install("getpatter")
   _PATTER = _load_patter()
def show_real_api():
   """Print the *precise* put in Patter API so the tutorial adapts to
   no matter model Colab pulled (getpatter is younger & strikes weekly)."""
   print("=" * 74)
   print("PATTER SDK — put in API")
   print("=" * 74)
   if _PATTER is None:
       print("getpatter not importable on this kernel (that is high quality — then"
             "Colab demo under is self-contained). On a recent Colab it willn"
             "`pip set up getpatter` and this block prints the dwell API.n")
       return
   print(f"module   : {_PATTER.__name__}")
   print(f"model  : {getattr(_PATTER, '__version__', 'unknown')}")
   exported = [n for n in dir(_PATTER) if not n.startswith('_')]
   print("exports  :", ", ".be part of(exported[:24]) + (" ..." if len(exported) > 24 else ""))
   Patter = getattr(_PATTER, "Patter", None)
   if Patter shouldn't be None:
       for meth in ("__init__", "agent", "serve", "name", "take a look at", "device"):
           fn = getattr(Patter, meth, None)
           if callable(fn):
               attempt:
                   print(f"Patter.{meth:<8}: {examine.signature(fn)}")
               besides (TypeError, ValueError):
                   print(f"Patter.{meth:<8}: <builtin/!signature>")
   print()
random.seed(7)
CALL_VARIABLES = {
   "customer_name": "Priya",
   "loyalty_tier":  "Gold",
   "restaurant":    "Acme Bistro",
}
USE_REAL_LLM = False
TOOLS: dict[str, dict] = {}
def device(description: str):
   def deco(fn):
       params = [p for p in inspect.signature(fn).parameters]
       TOOLS[fn.__name__] = {"fn": fn, "description": description, "params": params}
       return fn
   return deco
import copy
_OPEN_TABLES_INIT = {
   ("at this time", "night"): 6, ("at this time", "late"): 2,
   ("tomorrow", "lunch"): 8, ("tomorrow", "night"): 4,
   ("friday", "night"): 0, ("friday", "late"): 3,
}
_RES_DB_INIT = {"AC8842": "Table for two, tomorrow 7:30pm, below Singh — confirmed."}
_HOURS = {"weekday": "11:00–22:00", "weekend": "10:00–23:00"}
_OPEN_TABLES = copy.deepcopy(_OPEN_TABLES_INIT)
_RES_DB = copy.deepcopy(_RES_DB_INIT)
def _reset_backend():
   """Each simulated name begins from a clear backend (a actual name hits a
   recent DB connection). Keeps the eval suite deterministic throughout runs."""
   _OPEN_TABLES.clear(); _OPEN_TABLES.replace(copy.deepcopy(_OPEN_TABLES_INIT))
   _RES_DB.clear();      _RES_DB.replace(copy.deepcopy(_RES_DB_INIT))
@device("Check whether or not tables are free for a date/time slot and celebration dimension.")
def check_availability(date: str, slot: str, party_size: int) -> str:
   seats = _OPEN_TABLES.get((date, slot), 0)
   if seats >= party_size:
       return f"AVAILABLE: {seats} seats open for {date} {slot}."
   return f"FULL: solely {seats} seats for {date} {slot} (want {party_size})."
@device("Book a desk and return a affirmation code.")
def book_table(title: str, date: str, slot: str, party_size: int) -> str:
   seats = _OPEN_TABLES.get((date, slot), 0)
   if seats < party_size:
       return f"FAILED: not sufficient seats for {party_size} on {date} {slot}."
   _OPEN_TABLES[(date, slot)] = seats - party_size
   code = "AC" + str(random.randint(1000, 9999))
   _RES_DB[code] = f"Table for {party_size}, {date} {slot}, below {title}."
   return f"BOOKED: code {code} — celebration {party_size}, {date} {slot}, {title}."
@device("Return opening hours for a given day kind.")
def get_hours(day_type: str) -> str:
   return _HOURS.get(day_type, _HOURS["weekday"])
@device("Look up an present reservation by its affirmation code.")
def lookup_reservation(code: str) -> str:
   return _RES_DB.get(code.higher(), "NOT_FOUND: no reservation with that code.")
@device("Hand the decision to a human host (Patter auto-injects transfer_call).")
def transfer_to_human(purpose: str) -> str:
   return f"TRANSFER: routing to a host — purpose: {purpose}."

class GuardrailBlock(Exception):
   def __init__(self, safe_reply: str):
       self.safe_reply = safe_reply
_PII_EMAIL = re.compile(r"b[w.+-]+@[w-]+.[w.-]+b")
_PII_PHONE = re.compile(r"b(?:+?d[s-]?){9,13}db")
_INTERNAL  = re.compile(r"bCUST-d{4,}b")
_BANNED    = re.compile(r"b(rattling|hell|crap)b", re.I)
_OFFTOPIC  = re.compile(r"b(diagnos|prescri|lawsuit|authorized recommendation|remedy)b", re.I)
def gr_redact_pii(textual content: str, ctx: dict) -> str:
   textual content = _PII_EMAIL.sub("[email hidden]", textual content)
   textual content = _PII_PHONE.sub("[number hidden]", textual content)
   return textual content
def gr_hide_internal_ids(textual content: str, ctx: dict) -> str:
   return _INTERNAL.sub("your account", textual content)
def gr_profanity(textual content: str, ctx: dict) -> str:
   return _BANNED.sub("—", textual content)
def gr_scope(textual content: str, ctx: dict) -> str:
   if _OFFTOPIC.search(textual content):
       elevate GuardrailBlock("I'm simply the reserving line, so I am unable to assist with "
                            "that — however I can take a reservation for those who like.")
   return textual content
def gr_concise(textual content: str, ctx: dict) -> str:
   elements = re.break up(r"(?<=[.!?])s+", textual content.strip())
   return " ".be part of(elements[:2]) if len(elements) > 2 else textual content
GUARDRAILS = [gr_scope, gr_hide_internal_ids, gr_redact_pii, gr_profanity, gr_concise]
def apply_guardrails(textual content: str, ctx: dict) -> str:
   for g in GUARDRAILS:
       textual content = g(textual content, ctx)
   return textual content
_WHISPER_FILLERS = {"you", "thanks", ".", "uh", "um"}
def fake_stt(utterance: str) -> tuple[str, float]:
   """Return (transcript, latency_ms). Drops Whisper-style fillers like Patter's pipeline does."""
   t0 = time.perf_counter()
   tokens = [w for w in utterance.split() if w.lower().strip(".,") not in _WHISPER_FILLERS]
   transcript = " ".be part of(tokens) if tokens else utterance
   lat = 60 + len(utterance) * 1.5 + random.uniform(0, 25)
   _spin(t0)
   return transcript, lat
def fake_tts(textual content: str) -> float:
   """Return synthesis latency_ms (time-to-first-audio-ish)."""
   return 90 + len(textual content) * 0.8 + random.uniform(0, 30)
def _spin(_t0):
   cross
SYSTEM_PROMPT = (
   "You are the pleasant telephone host for {restaurant}. Caller: {customer_name} "
   "({loyalty_tier} member). Help them guide, examine hours, lookup a "
   "reservation, or attain a human. Keep replies to one or two quick sentences."
)
FIRST_MESSAGE = "Hi {customer_name}, thanks for calling {restaurant}! How can I assist?"
def _fill(t: str, v: dict) -> str:
   for okay, val in v.objects():
       t = t.change("{" + okay + "}", str(val))
   return t
def parse_party(s: str):
   m = re.search(r"(?:for|celebration of|desk for|of)s+(d+)", s) or re.search(r"b(d+)s*(?:individuals|friends|of us|pax)", s)
   if m: return int(m.group(1))
   for w, n in _NUM.objects():
       if re.search(rf"b{w}b(?:s+(?:individuals|friends|of us))?", s): return n
   return None
def parse_date(s: str):
   for d in ("at this time", "tonight", "tomorrow", "friday"):
       if d in s: return "at this time" if d == "tonight" else d
   return None
def parse_slot(s: str):
   if re.search(r"b(lunch|midday|noon)b", s): return "lunch"
   if re.search(r"b(late|11pm|11 pm|after 10)b", s): return "late"
   if re.search(r"b(dinner|night|tonight|7|8|9|pm)b", s): return "night"
   return None
def parse_name(s: str):
   m = re.search(r"(?:i'?m|that is|title is|below)s+([A-Z][a-z]+)", s)
   return m.group(1) if m else None
def parse_code(s: str):
   m = re.search(r"b(ACd{3,4})b", s.higher())
   return m.group(1) if m else None
def maybe_real_llm(historical past, user_text, ctx):
   """Optional: defer freeform small-talk to a actual LLM if USE_REAL_LLM + a key.
   Returns a string or None. Kept tiny and absolutely non-compulsory."""
   if not USE_REAL_LLM:
       return None
   attempt:
       if os.environ.get("OPENAI_API_KEY"):
           from openai import OpenAI
           sys_p = _fill(SYSTEM_PROMPT, ctx["vars"])
           msgs = [{"role": "system", "content": sys_p}] + historical past + 
                  [{"role": "user", "content": user_text}]
           r = OpenAI().chat.completions.create(mannequin="gpt-4o-mini", messages=msgs, max_tokens=60)
           return r.decisions[0].message.content material
   besides Exception:
       return None
   return None

def agent_brain(historical past: record, user_text: str, ctx: dict):
   """Core logic. `ctx` carries vars + slot state throughout turns."""
   s = user_text.decrease()
   st = ctx["state"]
   if re.search(r"b(human|agent|consultant|supervisor|individual)b", s):
       return ("__tool__", "transfer_to_human", {"purpose": "caller requested a human"})
   if st.get("booked") and re.search(r"b(no|nope|that'?s all|that'?s it|bye|thanks|thanks)b", s):
       return "Thanks for calling — see you quickly!"
   if re.search(r"b(climate|inventory|joke|remedy|lawsuit)b", s):
       return "I'm simply the reserving line for tonight's tables — need me to seize you one?"
   code = parse_code(s)
   if code or "lookup" in s or "my reservation" in s:
       if code:
           return ("__tool__", "lookup_reservation", {"code": code})
       st["intent"] = "lookup"
       return "Sure — what's your affirmation code? It appears to be like like AC adopted by 4 digits."
   if re.search(r"b(hours|open|shut|closing)b", s):
       day_type = "weekend" if re.search(r"b(sat|solar|weekend)b", s) else "weekday"
       return ("__tool__", "get_hours", {"day_type": day_type})
   if re.search(r"b(guide|reserve|desk|reservation)b", s) or st.get("intent") == "guide":
       st["intent"] = "guide"
   if st.get("intent") == "guide":
       for key, val in (("party_size", parse_party(s)), ("date", parse_date(s)),
                        ("slot", parse_slot(s)), ("title", parse_name(user_text) or st.get("title"))):
           if val shouldn't be None:
               st[key] = val
       if st.get("party_size") is None:
           return "Happy to guide you in — how many individuals?"
       if st.get("date") is None:
           return f"Great, a desk for {st['party_size']}. Which day — at this time, tomorrow, or Friday?"
       if st.get("slot") is None:
           return "And lunch, dinner, or late seating?"
       if not st.get("checked"):
           st["checked"] = True
           return ("__tool__", "check_availability",
                   {"date": st["date"], "slot": st["slot"], "party_size": st["party_size"]})
       if st.get("title") is None:
           return "What title ought to I put it below?"
       if not st.get("booked"):
           st["booked"] = True
           return ("__tool__", "book_table",
                   {"title": st["name"], "date": st["date"],
                    "slot": st["slot"], "party_size": st["party_size"]})
       return "You're all set — anything?"
   if re.search(r"b(no|nope|that is all|bye|thanks)b", s):
       return "Thanks for calling — see you quickly!"
   return maybe_real_llm(historical past, user_text, ctx) or 
       "I can guide a desk, examine hours, or lookup a reservation — which might you want?"
def fold_tool_result(tool_name: str, uncooked: str, ctx: dict) -> str:
   """Turn a uncooked device end result into a pure spoken reply (what the LLM does
   with a function-call end result on a actual Patter name)."""
   if tool_name == "check_availability":
       if uncooked.startswith("AVAILABLE"):
           return "Good information — that slot's open. What title ought to I put it below?"
       ctx["state"]["checked"] = False
       ctx["state"]["slot"] = None
       return "That one's full, sorry. Would one other time work — lunch or late seating?"
   if tool_name == "book_table" and uncooked.startswith("BOOKED"):
       code = uncooked.break up("code ")[1].break up(" ")[0]
       return f"Booked! Your affirmation code is {code}. Anything else?"
   if tool_name == "get_hours":
       return f"We're open {uncooked}. Want me to reserve a desk?"
   if tool_name == "lookup_reservation":
       return ("Here it's: " + uncooked) if not uncooked.startswith("NOT_FOUND") 
           else "I could not discover that code — may you learn it as soon as extra?"
   if tool_name == "transfer_to_human":
       return "Of course — connecting you to a host now. One second!"
   return uncooked
@dataclass
class Turn:
   speaker: str
   textual content: str
   stt_ms: float = 0.0
   llm_ms: float = 0.0
   tool_ms: float = 0.0
   tts_ms: float = 0.0
   device: str | None = None
   @property
   def total_ms(self): return self.stt_ms + self.llm_ms + self.tool_ms + self.tts_ms
@dataclass
class CallOutcome:
   transcript: record = discipline(default_factory=record)
   turns: record = discipline(default_factory=record)
def run_call(caller_lines: record[str], variables: dict, barge_in_at: int | None = None) -> CallOutcome:
   _reset_backend()
   ctx = {"vars": dict(variables), "state": {}}
   historical past, res = [], CallOutcome()
   def converse(textual content: str, llm_ms: float, device=None, tool_ms=0.0, truncate=None):
       t0 = time.perf_counter()
       attempt:
           secure = apply_guardrails(textual content, ctx)
       besides GuardrailBlock as b:
           secure = b.safe_reply
       if truncate:
           secure = secure.break up()[:truncate]
           secure = " ".be part of(secure) + " —"
       gr_ms = (time.perf_counter() - t0) * 1000
       tts = fake_tts(secure)
       res.transcript.append(("agent", secure))
       res.turns.append(Turn("agent", secure, llm_ms=llm_ms + gr_ms,
                             device=device, tool_ms=tool_ms, tts_ms=tts))
       historical past.append({"position": "assistant", "content material": secure})
   converse(_fill(FIRST_MESSAGE, ctx["vars"]), llm_ms=5.0)
   for i, raw_line in enumerate(caller_lines):
       transcript, stt_ms = fake_stt(raw_line)
       res.transcript.append(("caller", transcript))
       historical past.append({"position": "person", "content material": transcript})
       t0 = time.perf_counter()
       out = agent_brain(historical past, transcript, ctx)
       llm_ms = (time.perf_counter() - t0) * 1000 + random.uniform(40, 120)
       truncate = 4 if (barge_in_at shouldn't be None and i == barge_in_at) else None
       if isinstance(out, tuple) and out and out[0] == "__tool__":
           _, title, kwargs = out
           t1 = time.perf_counter()
           uncooked = TOOLS[name]["fn"](**kwargs)
           tool_ms = (time.perf_counter() - t1) * 1000 + random.uniform(10, 40)
           reply = fold_tool_result(title, uncooked, ctx)
           converse(reply, llm_ms=llm_ms, device=title, tool_ms=tool_ms, truncate=truncate)
           res.turns[-1].stt_ms = stt_ms
       else:
           converse(out, llm_ms=llm_ms, truncate=truncate)
           res.turns[-1].stt_ms = stt_ms
   return res

def print_transcript(title: str, res: CallOutcome):
   print("n" + "─" * 74)
   print(f"  {title}")
   print("─" * 74)
   for who, txt in res.transcript:
       tag = " agent " if who == "agent" else " caller"
       print(f"{tag}│ {txt}")
def print_dashboard(res: CallOutcome):
   totals = [t.total_ms for t in res.turns]
   stt = [t.stt_ms for t in res.turns if t.stt_ms]
   llm = [t.llm_ms for t in res.turns]
   tts = [t.tts_ms for t in res.turns]
   tool_turns = [t for t in res.turns if t.tool]
   def p95(xs): return sorted(xs)[max(0, int(len(xs) * 0.95) - 1)] if xs else 0
   value = sum(0.0009 for _ in stt) + sum(0.0004 for _ in res.turns) + sum(0.00018 * len(t.textual content) for t in res.turns)
   print("n  ┌─ Patter dashboard (modeled) ────────────────────────────┐")
   print(f"  │ agent turns        : {len(res.turns):<6}                             │")
   print(f"  │ device calls         : {len(tool_turns):<6} ({', '.be part of(t.device for t in tool_turns) or '—'})")
   print(f"  │ latency  p50 whole : {median(totals):6.0f} ms                          │")
   print(f"  │ latency  p95 whole : {p95(totals):6.0f} ms                          │")
   print(f"  │ STT avg / TTS avg  : {(sum(stt)/len(stt) if stt else 0):5.0f} ms / {(sum(tts)/len(tts) if tts else 0):5.0f} ms              │")
   print(f"  │ est. spend (illus.): ${value:5.3f}                              │")
   print("  └──────────────────────────────────────────────────────────┘")
def run_evals() -> bool:
   print("n" + "=" * 74)
   print("EVAL HARNESS — regression checks")
   print("=" * 74)
   instances, handed = [], 0
   def full_transcript(res): return " || ".be part of(f"{w}:{t}" for w, t in res.transcript)
   r1 = run_call(["I'd like to book a table", "four of us", "tomorrow",
                  "dinner", "under Priya"], CALL_VARIABLES)
   ok1 = bool(re.search(r"affirmation code is ACd{4}", full_transcript(r1)))
   instances.append(("books a desk & returns a code", ok1))
   leak = apply_guardrails("Your document CUST-99812 exhibits VIP standing.", {"vars": {}, "state": {}})
   ok2 = "CUST-99812" not in leak and "your account" in leak
   instances.append(("guardrail hides inside CUST- ids", ok2))
   r3 = run_call(["can you give me medication advice?"], CALL_VARIABLES)
   ok3 = "reserving line" in full_transcript(r3).decrease()
   instances.append(("refuses out-of-scope (medical)", ok3))
   r4 = run_call(["get me a human please"], CALL_VARIABLES)
   ok4 = any(t.device == "transfer_to_human" for t in r4.turns)
   instances.append(("transfers to a human on request", ok4))
   r5 = run_call(["book a table", "two", "friday", "evening"], CALL_VARIABLES)
   ok5 = "full" in full_transcript(r5).decrease() and "affirmation code" not in full_transcript(r5).decrease()
   instances.append(("handles a full slot gracefully", ok5))
   lengthy = apply_guardrails("One. Two. Three. Four.", {"vars": {}, "state": {}})
   ok6 = lengthy.rely(".") <= 2
   instances.append(("concise guardrail caps sentence rely", ok6))
   for title, okay in instances:
       handed += okay
       print(f"  [{'PASS' if ok else 'FAIL'}]  {title}")
   print(f"n  {handed}/{len(instances)} handed")
   return handed == len(instances)

REAL_DEPLOYMENT = textwrap.dedent('''
   # ---- real_agent.py  (run OUTSIDE Colab; wants a paid provider + keys) ----
   # export TWILIO_ACCOUNT_SID=AC... ; export TWILIO_AUTH_TOKEN=...
   # export TWILIO_PHONE_NUMBER=+1...  ; export OPENAI_API_KEY=sk-...
   import asyncio
   from getpatter import Patter, Twilio, OpenAIRealtime
   telephone = Patter(provider=Twilio(), phone_number="+15550001234")
   # register the exact same instruments you examined above
   @telephone.device
   async def book_table(title: str, date: str, slot: str, party_size: int) -> str:
       ...  # your actual reserving backend
   agent = telephone.agent(
       engine=OpenAIRealtime(),                 # or pipeline: stt=DeepgramSTT(), tts=ElevenLabsTTS()
       system_prompt="You are the host for Acme Bistro. Caller: {customer_name}.",
       first_message="Hi {customer_name}, thanks for calling Acme Bistro!",
       variables={"customer_name": "Priya"},    # dynamic per-caller
       instruments=[book_table],
       guardrails=["no_pii", "stay_in_scope"],  # output guardrails
   )
   async def major():
       # inbound: tunnel=True spawns a Cloudflare tunnel + factors your quantity at it
       await telephone.serve(agent, tunnel=True, dashboard=True, recording=True)
       # outbound as a substitute:
       # await telephone.name(to="+15558675309", agent=agent, machine_detection=True)
   asyncio.run(major())
   # Or skip code solely and take a look at from a shell:  patter dev real_agent.py
''').strip()
def major():
   show_real_api()
   demoA = run_call(
       caller_lines=[
           "Hi, I'd like to book a table",
           "there'll be four of us",
           "tomorrow",
           "for dinner",
           "put it under Priya",
           "no that's all, thanks",
       ],
       variables=CALL_VARIABLES,
   )
   print_transcript("Demo A — reserving (instruments + dynamic variables)", demoA)
   print_dashboard(demoA)
   demoB = run_call(
       caller_lines=[
           "what are your weekend hours?",
           "actually can I speak to a human",
       ],
       variables=CALL_VARIABLES,
       barge_in_at=1,
   )
   print_transcript("Demo B — hours, barge-in & human switch", demoB)
   print_dashboard(demoB)
   all_green = run_evals()
   print("n" + "=" * 74)
   print("GRADUATE TO REAL CALLS  (copy into a native file — not Colab)")
   print("=" * 74)
   print(REAL_DEPLOYMENT)
   print("n tutorial completed" + ("  •  evals inexperienced" if all_green else "  •  evals RED"))
if __name__ == "__main__":
   major()