MarkTechPost(RSS)
精选
72AI 编辑部评分,满分 100

Patter SDK 教程:构建餐厅预订电话智能体,支持动态变量、护栏、延迟仪表盘与评估检查

2026-07-16 15:42· 23天前· Sana Hassan
AI 导读

Patter SDK 发布教程,演示如何构建一个餐厅预订场景的语音智能体工作流。该流程支持动态调用变量、注册可调用工具、应用输出护栏(如PII脱敏、脏话过滤、话题范围限制),并可在无需实时电话凭证的情况下运行脚本化通话模拟。教程还涵盖延迟与成本指标追踪、回归式评估检查,以及将智能体逻辑、工具调用、安全检查和通话模拟整合为单一结构化管线。

推荐理由

这个教程把 Patter SDK 的语音代理从模拟到部署跑了一遍,代码能直接抄,想上手电话 AI 的开发者可以当样板,不过工具本身还比较新,生态有待验证。

正文 · AI 翻译

在本教程中,我们将通过构建一个语音智能体工作流来探索 Patter SDK,该工作流模拟了 AI 电话助手在真实对话中的行为方式。我们以一个餐厅预订用例为例,在其中定义动态呼叫者变量、注册可调用工具、应用输出护栏、模拟语音转文本和文本转语音行为,并在无需实时电话凭证的情况下运行完整的脚本化呼叫流程。我们还会在可用时检查已安装的 Patter API,创建一个确定性的智能体大脑,跟踪建模的延迟和成本指标,并通过回归式评估来验证系统。最后,我们将了解 Patter SDK 如何将智能体逻辑、工具使用、安全检查、呼叫模拟和实际部署模式整合到一个结构化的语音智能体流水线中。

设置 Patter SDK、工具和餐厅后端

from __future__ import annotations
import sys, subprocess, importlib, inspect, time, json, re, random, textwrap, os
from dataclasses import dataclass, field
from statistics import median
def _try_install(pkg: str) -> None:
   try:
       subprocess.run([sys.executable, "-m", "pip", "install", "-q", pkg],
                      check=False, timeout=600)
   except Exception:
       pass
def _load_patter():
   """Return the real getpatter module if importable, else None."""
   for name in ("patter", "getpatter"):
       try:
           return importlib.import_module(name)
       except Exception:
           continue
   return None
_PATTER = _load_patter()
if _PATTER is None:
   _try_install("getpatter")
   _PATTER = _load_patter()
def show_real_api():
   """Print the *actual* installed Patter API so the tutorial adapts to
   whatever version Colab pulled (getpatter is young & moves weekly)."""
   print("=" * 74)
   print("PATTER SDK — installed API")
   print("=" * 74)
   if _PATTER is None:
       print("getpatter not importable in this kernel (that's fine — the\n"
             "Colab demo below is self-contained). On a fresh Colab it will\n"
             "`pip install getpatter` and this block prints the live API.\n")
       return
   print(f"module   : {_PATTER.__name__}")
   print(f"version  : {getattr(_PATTER, '__version__', 'unknown')}")
   exported = [n for n in dir(_PATTER) if not n.startswith('_')]
   print("exports  :", ", ".join(exported[:24]) + (" ..." if len(exported) > 24 else ""))
   Patter = getattr(_PATTER, "Patter", None)
   if Patter is not None:
       for meth in ("__init__", "agent", "serve", "call", "test", "tool"):
           fn = getattr(Patter, meth, None)
           if callable(fn):
               try:
                   print(f"Patter.{meth:<8}: {inspect.signature(fn)}")
               except (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 tool(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 = {
   ("today", "evening"): 6, ("today", "late"): 2,
   ("tomorrow", "lunch"): 8, ("tomorrow", "evening"): 4,
   ("friday", "evening"): 0, ("friday", "late"): 3,
}
_RES_DB_INIT = {"AC8842": "Table for 2, tomorrow 7:30pm, under 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 call starts from a clean backend (a real call hits a
   fresh DB connection). Keeps the eval suite deterministic across runs."""
   _OPEN_TABLES.clear(); _OPEN_TABLES.update(copy.deepcopy(_OPEN_TABLES_INIT))
   _RES_DB.clear();      _RES_DB.update(copy.deepcopy(_RES_DB_INIT))
@tool("Check whether tables are free for a date/time slot and party size.")
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: only {seats} seats for {date} {slot} (need {party_size})."
@tool("Book a table and return a confirmation code.")
def book_table(name: str, date: str, slot: str, party_size: int) -> str:
   seats = _OPEN_TABLES.get((date, slot), 0)
   if seats < party_size:
       return f"FAILED: not enough 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}, under {name}."
   return f"BOOKED: code {code} — party {party_size}, {date} {slot}, {name}."
@tool("Return opening hours for a given day type.")
def get_hours(day_type: str) -> str:
   return _HOURS.get(day_type, _HOURS["weekday"])
@tool("Look up an existing reservation by its confirmation code.")
def lookup_reservation(code: str) -> str:
   return _RES_DB.get(code.upper(), "NOT_FOUND: no reservation with that code.")
@tool("Hand the call to a human host (Patter auto-injects transfer_call).")
def transfer_to_human(reason: str) -> str:
   return f"TRANSFER: routing to a host — reason: {reason}."

我们通过导入所需库、可选地安装 Patter SDK 以及在可用时检查已安装的 API 来设置教程环境。我们定义动态呼叫者变量,创建一个小的工具注册表,并准备一个用于检查可用性、预订、营业时间和转接的内存餐厅后端。我们还会注册核心工具,使我们的模拟电话智能体能够检查餐桌、进行预订、查找确认码以及将呼叫者转接给人工客服。

添加输出护栏和模拟语音层

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}\d\b")
_INTERNAL  = re.compile(r"\bCUST-\d{4,}\b")
_BANNED    = re.compile(r"\b(damn|hell|crap)\b", re.I)
_OFFTOPIC  = re.compile(r"\b(diagnos|prescri|lawsuit|legal advice|medication)\b", re.I)
def gr_redact_pii(text: str, ctx: dict) -> str:
   text = _PII_EMAIL.sub("[email hidden]", text)
   text = _PII_PHONE.sub("[number hidden]", text)
   return text
def gr_hide_internal_ids(text: str, ctx: dict) -> str:
   return _INTERNAL.sub("your account", text)
def gr_profanity(text: str, ctx: dict) -> str:
   return _BANNED.sub("—", text)
def gr_scope(text: str, ctx: dict) -> str:
   if _OFFTOPIC.search(text):
       raise GuardrailBlock("I'm just the booking line, so I can't help with "
                            "that — but I can take a reservation if you like.")
   return text
def gr_concise(text: str, ctx: dict) -> str:
   parts = re.split(r"(?<=[.!?])\s+", text.strip())
   return " ".join(parts[:2]) if len(parts) > 2 else text
GUARDRAILS = [gr_scope, gr_hide_internal_ids, gr_redact_pii, gr_profanity, gr_concise]
def apply_guardrails(text: str, ctx: dict) -> str:
   for g in GUARDRAILS:
       text = g(text, ctx)
   return text
_WHISPER_FILLERS = {"you", "thank you", ".", "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 = " ".join(tokens) if tokens else utterance
   lat = 60 + len(utterance) * 1.5 + random.uniform(0, 25)
   _spin(t0)
   return transcript, lat
def fake_tts(text: str) -> float:
   """Return synthesis latency_ms (time-to-first-audio-ish)."""
   return 90 + len(text) * 0.8 + random.uniform(0, 30)
def _spin(_t0):
   pass
SYSTEM_PROMPT = (
   "You are the friendly phone host for {restaurant}. Caller: {customer_name} "
   "({loyalty_tier} member). Help them book, check hours, look up a "
   "reservation, or reach a human. Keep replies to one or two short sentences."
)
FIRST_MESSAGE = "Hi {customer_name}, thanks for calling {restaurant}! How can I help?"
def _fill(t: str, v: dict) -> str:
   for k, val in v.items():
       t = t.replace("{" + k + "}", str(val))
   return t
def parse_party(s: str):
   m = re.search(r"(?:for|party of|table for|of)\s+(\d+)", s) or re.search(r"\b(\d+)\s*(?:people|guests|of us|pax)", s)
   if m: return int(m.group(1))
   for w, n in _NUM.items():
       if re.search(rf"\b{w}\b(?:\s+(?:people|guests|of us))?", s): return n
   return None
def parse_date(s: str):
   for d in ("today", "tonight", "tomorrow", "friday"):
       if d in s: return "today" if d == "tonight" else d
   return None
def parse_slot(s: str):
   if re.search(r"\b(lunch|noon|midday)\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|evening|tonight|7|8|9|pm)\b", s): return "evening"
   return None
def parse_name(s: str):
   m = re.search(r"(?:i'?m|this is|name is|under)\s+([A-Z][a-z]+)", s)
   return m.group(1) if m else None
def parse_code(s: str):
   m = re.search(r"\b(AC\d{3,4})\b", s.upper())
   return m.group(1) if m else None
def maybe_real_llm(history, user_text, ctx):
   """Optional: defer freeform small-talk to a real LLM if USE_REAL_LLM + a key.
   Returns a string or None. Kept tiny and fully optional."""
   if not USE_REAL_LLM:
       return None
   try:
       if os.environ.get("OPENAI_API_KEY"):
           from openai import OpenAI
           sys_p = _fill(SYSTEM_PROMPT, ctx["vars"])
           msgs = [{"role": "system", "content": sys_p}] + history + \
                  [{"role": "user", "content": user_text}]
           r = OpenAI().chat.completions.create(model="gpt-4o-mini", messages=msgs, max_tokens=60)
           return r.choices[0].message.content
   except Exception:
       return None
   return None

我们构建输出护栏层,使电话助手保持安全、简洁,并适用于预订用例。我们编辑敏感信息、隐藏内部客户 ID、清理不当语言、阻止与主题无关的请求,并保持回复简短以获得更好的电话体验。然后,我们模拟语音转文本和文本转语音行为,定义系统提示词,并添加用于解析人数、日期、时段、姓名和预订码的轻量级解析函数。

构建智能体大脑和呼叫模拟器

def agent_brain(history: list, user_text: str, ctx: dict):
   """Core logic. `ctx` carries vars + slot state across turns."""
   s = user_text.lower()
   st = ctx["state"]
   if re.search(r"\b(human|agent|representative|manager|person)\b", s):
       return ("__tool__", "transfer_to_human", {"reason": "caller requested a human"})
   if st.get("booked") and re.search(r"\b(no|nope|that'?s all|that'?s it|bye|thanks|thank you)\b", s):
       return "Thanks for calling — see you soon!"
   if re.search(r"\b(weather|stock|joke|medication|lawsuit)\b", s):
       return "I'm just the booking line for tonight's tables — want me to grab you one?"
   code = parse_code(s)
   if code or "look up" in s or "my reservation" in s:
       if code:
           return ("__tool__", "lookup_reservation", {"code": code})
       st["intent"] = "lookup"
       return "Sure — what's your confirmation code? It looks like AC followed by four digits."
   if re.search(r"\b(hours|open|close|closing)\b", s):
       day_type = "weekend" if re.search(r"\b(sat|sun|weekend)\b", s) else "weekday"
       return ("__tool__", "get_hours", {"day_type": day_type})
   if re.search(r"\b(book|reserve|table|reservation)\b", s) or st.get("intent") == "book":
       st["intent"] = "book"
   if st.get("intent") == "book":
       for key, val in (("party_size", parse_party(s)), ("date", parse_date(s)),
                        ("slot", parse_slot(s)), ("name", parse_name(user_text) or st.get("name"))):
           if val is not None:
               st[key] = val
       if st.get("party_size") is None:
           return "Happy to book you in — how many people?"
       if st.get("date") is None:
           return f"Great, a table for {st['party_size']}. Which day — today, 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("name") is None:
           return "What name should I put it under?"
       if not st.get("booked"):
           st["booked"] = True
           return ("__tool__", "book_table",
                   {"name": st["name"], "date": st["date"],
                    "slot": st["slot"], "party_size": st["party_size"]})
       return "You're all set — anything else?"
   if re.search(r"\b(no|nope|that's all|bye|thanks)\b", s):
       return "Thanks for calling — see you soon!"
   return maybe_real_llm(history, user_text, ctx) or \
       "I can book a table, check hours, or look up a reservation — which would you like?"
def fold_tool_result(tool_name: str, raw: str, ctx: dict) -> str:
   """Turn a raw tool result into a natural spoken reply (what the LLM does
   with a function-call result on a real Patter call)."""
   if tool_name == "check_availability":
       if raw.startswith("AVAILABLE"):
           return "Good news — that slot's open. What name should I put it under?"
       ctx["state"]["checked"] = False
       ctx["state"]["slot"] = None
       return "That one's full, sorry. Would another time work — lunch or late seating?"
   if tool_name == "book_table" and raw.startswith("BOOKED"):
       code = raw.split("code ")[1].split(" ")[0]
       return f"Booked! Your confirmation code is {code}. Anything else?"
   if tool_name == "get_hours":
       return f"We're open {raw}. Want me to reserve a table?"
   if tool_name == "lookup_reservation":
       return ("Here it is: " + raw) if not raw.startswith("NOT_FOUND") \
           else "I couldn't find that code — could you read it once more?"
   if tool_name == "transfer_to_human":
       return "Of course — connecting you to a host now. One moment!"
   return raw
@dataclass
class Turn:
   speaker: str
   text: str
   stt_ms: float = 0.0
   llm_ms: float = 0.0
   tool_ms: float = 0.0
   tts_ms: float = 0.0
   tool: str | None = None
   @property
   def total_ms(self): return self.stt_ms + self.llm_ms + self.tool_ms + self.tts_ms
@dataclass
class CallResult:
   transcript: list = field(default_factory=list)
   turns: list = field(default_factory=list)
def run_call(caller_lines: list[str], variables: dict, barge_in_at: int | None = None) -> CallResult:
   _reset_backend()
   ctx = {"vars": dict(variables), "state": {}}
   history, res = [], CallResult()
   def speak(text: str, llm_ms: float, tool=None, tool_ms=0.0, truncate=None):
       t0 = time.perf_counter()
       try:
           safe = apply_guardrails(text, ctx)
       except GuardrailBlock as b:
           safe = b.safe_reply
       if truncate:
           safe = safe.split()[:truncate]
           safe = " ".join(safe) + " —"
       gr_ms = (time.perf_counter() - t0) * 1000
       tts = fake_tts(safe)
       res.transcript.append(("agent", safe))
       res.turns.append(Turn("agent", safe, llm_ms=llm_ms + gr_ms,
                             tool=tool, tool_ms=tool_ms, tts_ms=tts))
       history.append({"role": "assistant", "content": safe})
   speak(_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))
       history.append({"role": "user", "content": transcript})
       t0 = time.perf_counter()
       out = agent_brain(history, transcript, ctx)
       llm_ms = (time.perf_counter() - t0) * 1000 + random.uniform(40, 120)
       truncate = 4 if (barge_in_at is not None and i == barge_in_at) else None
       if isinstance(out, tuple) and out and out[0] == "__tool__":
           _, name, kwargs = out
           t1 = time.perf_counter()
           raw = TOOLS[name]["fn"](**kwargs)
           tool_ms = (time.perf_counter() - t1) * 1000 + random.uniform(10, 40)
           reply = fold_tool_result(name, raw, ctx)
           speak(reply, llm_ms=llm_ms, tool=name, tool_ms=tool_ms, truncate=truncate)
           res.turns[-1].stt_ms = stt_ms
       else:
           speak(out, llm_ms=llm_ms, truncate=truncate)
           res.turns[-1].stt_ms = stt_ms
   return res

我们实现了控制对话流程的主智能体大脑,涵盖预订、预约查询、营业时间咨询、转接人工以及兜底回复等功能。我们利用解析后的来电者输入和存储的对话状态,来决定何时提出追问、调用工具或完成预约。我们还创建了一个包含结构化轮次对象的通话模拟器,用于运行脚本化对话,并收集延迟、工具调用及通话记录数据。

打印通话记录、延迟仪表盘与回归评估

def print_transcript(title: str, res: CallResult):
   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: CallResult):
   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
   cost = sum(0.0009 for _ in stt) + sum(0.0004 for _ in res.turns) + sum(0.00018 * len(t.text) for t in res.turns)
   print("\n  ┌─ Patter dashboard (modeled) ────────────────────────────┐")
   print(f"  │ agent turns        : {len(res.turns):<6}                             │")
   print(f"  │ tool calls         : {len(tool_turns):<6} ({', '.join(t.tool for t in tool_turns) or '—'})")
   print(f"  │ latency  p50 total : {median(totals):6.0f} ms                          │")
   print(f"  │ latency  p95 total : {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.): ${cost:5.3f}                              │")
   print("  └──────────────────────────────────────────────────────────┘")
def run_evals() -> bool:
   print("\n" + "=" * 74)
   print("EVAL HARNESS — regression checks")
   print("=" * 74)
   cases, passed = [], 0
   def full_transcript(res): return " || ".join(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"confirmation code is AC\d{4}", full_transcript(r1)))
   cases.append(("books a table & returns a code", ok1))
   leak = apply_guardrails("Your record CUST-99812 shows VIP status.", {"vars": {}, "state": {}})
   ok2 = "CUST-99812" not in leak and "your account" in leak
   cases.append(("guardrail hides internal CUST- ids", ok2))
   r3 = run_call(["can you give me medication advice?"], CALL_VARIABLES)
   ok3 = "booking line" in full_transcript(r3).lower()
   cases.append(("refuses out-of-scope (medical)", ok3))
   r4 = run_call(["get me a human please"], CALL_VARIABLES)
   ok4 = any(t.tool == "transfer_to_human" for t in r4.turns)
   cases.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).lower() and "confirmation code" not in full_transcript(r5).lower()
   cases.append(("handles a full slot gracefully", ok5))
   long = apply_guardrails("One. Two. Three. Four.", {"vars": {}, "state": {}})
   ok6 = long.count(".") <= 2
   cases.append(("concise guardrail caps sentence count", ok6))
   for name, ok in cases:
       passed += ok
       print(f"  [{'PASS' if ok else 'FAIL'}]  {name}")
   print(f"\n  {passed}/{len(cases)} passed")
   return passed == len(cases)

我们将模拟通话结果格式化为可读的通话记录,以及一个 Patter 风格的仪表盘,该仪表盘汇总了智能体轮次、工具调用、延迟和预估费用。我们还构建了一个确定性评估框架,用于检查智能体是否完成预订、保护内部 ID、拒绝超出范围的医疗请求、转接人工、处理满额时段,以及保持回复简洁。我们利用这些检查来验证电话智能体工作流在进入实际部署前是否运行可靠。

使用 Twilio 和 OpenAI Realtime 转向真实通话

REAL_DEPLOYMENT = textwrap.dedent('''
   # ---- real_agent.py  (run OUTSIDE Colab; needs a paid carrier + 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
   phone = Patter(carrier=Twilio(), phone_number="+15550001234")
   # register the very same tools you tested above
   @phone.tool
   async def book_table(name: str, date: str, slot: str, party_size: int) -> str:
       ...  # your real booking backend
   agent = phone.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
       tools=[book_table],
       guardrails=["no_pii", "stay_in_scope"],  # output guardrails
   )
   async def main():
       # inbound: tunnel=True spawns a Cloudflare tunnel + points your number at it
       await phone.serve(agent, tunnel=True, dashboard=True, recording=True)
       # outbound instead:
       # await phone.call(to="+15558675309", agent=agent, machine_detection=True)
   asyncio.run(main())
   # Or skip code entirely and test from a shell:  patter dev real_agent.py
''').strip()
def main():
   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 — booking (tools + 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 transfer", demoB)
   print_dashboard(demoB)
   all_green = run_evals()
   print("\n" + "=" * 74)
   print("GRADUATE TO REAL CALLS  (copy into a local file — not Colab)")
   print("=" * 74)
   print(REAL_DEPLOYMENT)
   print("\n✅ tutorial finished" + ("  •  evals green" if all_green else "  •  evals RED"))
if __name__ == "__main__":
   main()

我们准备了一个生产级部署模板,展示如何将经过测试的相同逻辑从 Colab 迁移到真实的电话智能体设置中。我们包含了使用 Patter 与 Twilio、OpenAI Realtime、已注册工具、动态变量、安全护栏、入站服务、仪表盘监控、录音以及外呼功能的结构。随后,我们通过展示已安装的 API、执行两通演示通话、打印仪表盘、运行评估并展示最终的真实通话部署代码,完成了完整的教程。

结论

总之,我们构建了一个完整的 Patter 风格电话智能体工作流,它体现了生产级语音 AI 系统的核心结构。我们创建了用于检查可用性、预订餐位、查询预约以及将来电转接至人工客服的工具,然后将它们与护栏机制、脚本化对话、模拟延迟、仪表盘和评估检查相结合。我们还看到,同样的经过测试的逻辑,后续可以通过 Twilio、OpenAI Realtime、隧道技术以及 Patter 的部署界面,从一个独立的 Colab 模拟环境迁移到真实的电话通话中。此外,我们最终深入理解了如何在将语音智能体应用接入实时电话基础设施之前,对其进行原型设计、测试、监控和准备工作。

来源:MarkTechPost(RSS) · marktechpost.com

Patter SDK 教程:构建餐厅预订电话智能体,支持动态变量、护栏、延迟仪表盘与评估检查

MarkTechPost(RSS)·2026-07-16 15:42·23天前·Sana Hassan
AI 导读

Patter SDK 发布教程,演示如何构建一个餐厅预订场景的语音智能体工作流。该流程支持动态调用变量、注册可调用工具、应用输出护栏(如PII脱敏、脏话过滤、话题范围限制),并可在无需实时电话凭证的情况下运行脚本化通话模拟。教程还涵盖延迟与成本指标追踪、回归式评估检查,以及将智能体逻辑、工具调用、安全检查和通话模拟整合为单一结构化管线。

正文 · AI 翻译

在本教程中,我们将通过构建一个语音智能体工作流来探索 Patter SDK,该工作流模拟了 AI 电话助手在真实对话中的行为方式。我们以一个餐厅预订用例为例,在其中定义动态呼叫者变量、注册可调用工具、应用输出护栏、模拟语音转文本和文本转语音行为,并在无需实时电话凭证的情况下运行完整的脚本化呼叫流程。我们还会在可用时检查已安装的 Patter API,创建一个确定性的智能体大脑,跟踪建模的延迟和成本指标,并通过回归式评估来验证系统。最后,我们将了解 Patter SDK 如何将智能体逻辑、工具使用、安全检查、呼叫模拟和实际部署模式整合到一个结构化的语音智能体流水线中。

设置 Patter SDK、工具和餐厅后端

from __future__ import annotations
import sys, subprocess, importlib, inspect, time, json, re, random, textwrap, os
from dataclasses import dataclass, field
from statistics import median
def _try_install(pkg: str) -> None:
   try:
       subprocess.run([sys.executable, "-m", "pip", "install", "-q", pkg],
                      check=False, timeout=600)
   except Exception:
       pass
def _load_patter():
   """Return the real getpatter module if importable, else None."""
   for name in ("patter", "getpatter"):
       try:
           return importlib.import_module(name)
       except Exception:
           continue
   return None
_PATTER = _load_patter()
if _PATTER is None:
   _try_install("getpatter")
   _PATTER = _load_patter()
def show_real_api():
   """Print the *actual* installed Patter API so the tutorial adapts to
   whatever version Colab pulled (getpatter is young & moves weekly)."""
   print("=" * 74)
   print("PATTER SDK — installed API")
   print("=" * 74)
   if _PATTER is None:
       print("getpatter not importable in this kernel (that's fine — the\n"
             "Colab demo below is self-contained). On a fresh Colab it will\n"
             "`pip install getpatter` and this block prints the live API.\n")
       return
   print(f"module   : {_PATTER.__name__}")
   print(f"version  : {getattr(_PATTER, '__version__', 'unknown')}")
   exported = [n for n in dir(_PATTER) if not n.startswith('_')]
   print("exports  :", ", ".join(exported[:24]) + (" ..." if len(exported) > 24 else ""))
   Patter = getattr(_PATTER, "Patter", None)
   if Patter is not None:
       for meth in ("__init__", "agent", "serve", "call", "test", "tool"):
           fn = getattr(Patter, meth, None)
           if callable(fn):
               try:
                   print(f"Patter.{meth:<8}: {inspect.signature(fn)}")
               except (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 tool(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 = {
   ("today", "evening"): 6, ("today", "late"): 2,
   ("tomorrow", "lunch"): 8, ("tomorrow", "evening"): 4,
   ("friday", "evening"): 0, ("friday", "late"): 3,
}
_RES_DB_INIT = {"AC8842": "Table for 2, tomorrow 7:30pm, under 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 call starts from a clean backend (a real call hits a
   fresh DB connection). Keeps the eval suite deterministic across runs."""
   _OPEN_TABLES.clear(); _OPEN_TABLES.update(copy.deepcopy(_OPEN_TABLES_INIT))
   _RES_DB.clear();      _RES_DB.update(copy.deepcopy(_RES_DB_INIT))
@tool("Check whether tables are free for a date/time slot and party size.")
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: only {seats} seats for {date} {slot} (need {party_size})."
@tool("Book a table and return a confirmation code.")
def book_table(name: str, date: str, slot: str, party_size: int) -> str:
   seats = _OPEN_TABLES.get((date, slot), 0)
   if seats < party_size:
       return f"FAILED: not enough 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}, under {name}."
   return f"BOOKED: code {code} — party {party_size}, {date} {slot}, {name}."
@tool("Return opening hours for a given day type.")
def get_hours(day_type: str) -> str:
   return _HOURS.get(day_type, _HOURS["weekday"])
@tool("Look up an existing reservation by its confirmation code.")
def lookup_reservation(code: str) -> str:
   return _RES_DB.get(code.upper(), "NOT_FOUND: no reservation with that code.")
@tool("Hand the call to a human host (Patter auto-injects transfer_call).")
def transfer_to_human(reason: str) -> str:
   return f"TRANSFER: routing to a host — reason: {reason}."

我们通过导入所需库、可选地安装 Patter SDK 以及在可用时检查已安装的 API 来设置教程环境。我们定义动态呼叫者变量,创建一个小的工具注册表,并准备一个用于检查可用性、预订、营业时间和转接的内存餐厅后端。我们还会注册核心工具,使我们的模拟电话智能体能够检查餐桌、进行预订、查找确认码以及将呼叫者转接给人工客服。

添加输出护栏和模拟语音层

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}\d\b")
_INTERNAL  = re.compile(r"\bCUST-\d{4,}\b")
_BANNED    = re.compile(r"\b(damn|hell|crap)\b", re.I)
_OFFTOPIC  = re.compile(r"\b(diagnos|prescri|lawsuit|legal advice|medication)\b", re.I)
def gr_redact_pii(text: str, ctx: dict) -> str:
   text = _PII_EMAIL.sub("[email hidden]", text)
   text = _PII_PHONE.sub("[number hidden]", text)
   return text
def gr_hide_internal_ids(text: str, ctx: dict) -> str:
   return _INTERNAL.sub("your account", text)
def gr_profanity(text: str, ctx: dict) -> str:
   return _BANNED.sub("—", text)
def gr_scope(text: str, ctx: dict) -> str:
   if _OFFTOPIC.search(text):
       raise GuardrailBlock("I'm just the booking line, so I can't help with "
                            "that — but I can take a reservation if you like.")
   return text
def gr_concise(text: str, ctx: dict) -> str:
   parts = re.split(r"(?<=[.!?])\s+", text.strip())
   return " ".join(parts[:2]) if len(parts) > 2 else text
GUARDRAILS = [gr_scope, gr_hide_internal_ids, gr_redact_pii, gr_profanity, gr_concise]
def apply_guardrails(text: str, ctx: dict) -> str:
   for g in GUARDRAILS:
       text = g(text, ctx)
   return text
_WHISPER_FILLERS = {"you", "thank you", ".", "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 = " ".join(tokens) if tokens else utterance
   lat = 60 + len(utterance) * 1.5 + random.uniform(0, 25)
   _spin(t0)
   return transcript, lat
def fake_tts(text: str) -> float:
   """Return synthesis latency_ms (time-to-first-audio-ish)."""
   return 90 + len(text) * 0.8 + random.uniform(0, 30)
def _spin(_t0):
   pass
SYSTEM_PROMPT = (
   "You are the friendly phone host for {restaurant}. Caller: {customer_name} "
   "({loyalty_tier} member). Help them book, check hours, look up a "
   "reservation, or reach a human. Keep replies to one or two short sentences."
)
FIRST_MESSAGE = "Hi {customer_name}, thanks for calling {restaurant}! How can I help?"
def _fill(t: str, v: dict) -> str:
   for k, val in v.items():
       t = t.replace("{" + k + "}", str(val))
   return t
def parse_party(s: str):
   m = re.search(r"(?:for|party of|table for|of)\s+(\d+)", s) or re.search(r"\b(\d+)\s*(?:people|guests|of us|pax)", s)
   if m: return int(m.group(1))
   for w, n in _NUM.items():
       if re.search(rf"\b{w}\b(?:\s+(?:people|guests|of us))?", s): return n
   return None
def parse_date(s: str):
   for d in ("today", "tonight", "tomorrow", "friday"):
       if d in s: return "today" if d == "tonight" else d
   return None
def parse_slot(s: str):
   if re.search(r"\b(lunch|noon|midday)\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|evening|tonight|7|8|9|pm)\b", s): return "evening"
   return None
def parse_name(s: str):
   m = re.search(r"(?:i'?m|this is|name is|under)\s+([A-Z][a-z]+)", s)
   return m.group(1) if m else None
def parse_code(s: str):
   m = re.search(r"\b(AC\d{3,4})\b", s.upper())
   return m.group(1) if m else None
def maybe_real_llm(history, user_text, ctx):
   """Optional: defer freeform small-talk to a real LLM if USE_REAL_LLM + a key.
   Returns a string or None. Kept tiny and fully optional."""
   if not USE_REAL_LLM:
       return None
   try:
       if os.environ.get("OPENAI_API_KEY"):
           from openai import OpenAI
           sys_p = _fill(SYSTEM_PROMPT, ctx["vars"])
           msgs = [{"role": "system", "content": sys_p}] + history + \
                  [{"role": "user", "content": user_text}]
           r = OpenAI().chat.completions.create(model="gpt-4o-mini", messages=msgs, max_tokens=60)
           return r.choices[0].message.content
   except Exception:
       return None
   return None

我们构建输出护栏层,使电话助手保持安全、简洁,并适用于预订用例。我们编辑敏感信息、隐藏内部客户 ID、清理不当语言、阻止与主题无关的请求,并保持回复简短以获得更好的电话体验。然后,我们模拟语音转文本和文本转语音行为,定义系统提示词,并添加用于解析人数、日期、时段、姓名和预订码的轻量级解析函数。

构建智能体大脑和呼叫模拟器

def agent_brain(history: list, user_text: str, ctx: dict):
   """Core logic. `ctx` carries vars + slot state across turns."""
   s = user_text.lower()
   st = ctx["state"]
   if re.search(r"\b(human|agent|representative|manager|person)\b", s):
       return ("__tool__", "transfer_to_human", {"reason": "caller requested a human"})
   if st.get("booked") and re.search(r"\b(no|nope|that'?s all|that'?s it|bye|thanks|thank you)\b", s):
       return "Thanks for calling — see you soon!"
   if re.search(r"\b(weather|stock|joke|medication|lawsuit)\b", s):
       return "I'm just the booking line for tonight's tables — want me to grab you one?"
   code = parse_code(s)
   if code or "look up" in s or "my reservation" in s:
       if code:
           return ("__tool__", "lookup_reservation", {"code": code})
       st["intent"] = "lookup"
       return "Sure — what's your confirmation code? It looks like AC followed by four digits."
   if re.search(r"\b(hours|open|close|closing)\b", s):
       day_type = "weekend" if re.search(r"\b(sat|sun|weekend)\b", s) else "weekday"
       return ("__tool__", "get_hours", {"day_type": day_type})
   if re.search(r"\b(book|reserve|table|reservation)\b", s) or st.get("intent") == "book":
       st["intent"] = "book"
   if st.get("intent") == "book":
       for key, val in (("party_size", parse_party(s)), ("date", parse_date(s)),
                        ("slot", parse_slot(s)), ("name", parse_name(user_text) or st.get("name"))):
           if val is not None:
               st[key] = val
       if st.get("party_size") is None:
           return "Happy to book you in — how many people?"
       if st.get("date") is None:
           return f"Great, a table for {st['party_size']}. Which day — today, 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("name") is None:
           return "What name should I put it under?"
       if not st.get("booked"):
           st["booked"] = True
           return ("__tool__", "book_table",
                   {"name": st["name"], "date": st["date"],
                    "slot": st["slot"], "party_size": st["party_size"]})
       return "You're all set — anything else?"
   if re.search(r"\b(no|nope|that's all|bye|thanks)\b", s):
       return "Thanks for calling — see you soon!"
   return maybe_real_llm(history, user_text, ctx) or \
       "I can book a table, check hours, or look up a reservation — which would you like?"
def fold_tool_result(tool_name: str, raw: str, ctx: dict) -> str:
   """Turn a raw tool result into a natural spoken reply (what the LLM does
   with a function-call result on a real Patter call)."""
   if tool_name == "check_availability":
       if raw.startswith("AVAILABLE"):
           return "Good news — that slot's open. What name should I put it under?"
       ctx["state"]["checked"] = False
       ctx["state"]["slot"] = None
       return "That one's full, sorry. Would another time work — lunch or late seating?"
   if tool_name == "book_table" and raw.startswith("BOOKED"):
       code = raw.split("code ")[1].split(" ")[0]
       return f"Booked! Your confirmation code is {code}. Anything else?"
   if tool_name == "get_hours":
       return f"We're open {raw}. Want me to reserve a table?"
   if tool_name == "lookup_reservation":
       return ("Here it is: " + raw) if not raw.startswith("NOT_FOUND") \
           else "I couldn't find that code — could you read it once more?"
   if tool_name == "transfer_to_human":
       return "Of course — connecting you to a host now. One moment!"
   return raw
@dataclass
class Turn:
   speaker: str
   text: str
   stt_ms: float = 0.0
   llm_ms: float = 0.0
   tool_ms: float = 0.0
   tts_ms: float = 0.0
   tool: str | None = None
   @property
   def total_ms(self): return self.stt_ms + self.llm_ms + self.tool_ms + self.tts_ms
@dataclass
class CallResult:
   transcript: list = field(default_factory=list)
   turns: list = field(default_factory=list)
def run_call(caller_lines: list[str], variables: dict, barge_in_at: int | None = None) -> CallResult:
   _reset_backend()
   ctx = {"vars": dict(variables), "state": {}}
   history, res = [], CallResult()
   def speak(text: str, llm_ms: float, tool=None, tool_ms=0.0, truncate=None):
       t0 = time.perf_counter()
       try:
           safe = apply_guardrails(text, ctx)
       except GuardrailBlock as b:
           safe = b.safe_reply
       if truncate:
           safe = safe.split()[:truncate]
           safe = " ".join(safe) + " —"
       gr_ms = (time.perf_counter() - t0) * 1000
       tts = fake_tts(safe)
       res.transcript.append(("agent", safe))
       res.turns.append(Turn("agent", safe, llm_ms=llm_ms + gr_ms,
                             tool=tool, tool_ms=tool_ms, tts_ms=tts))
       history.append({"role": "assistant", "content": safe})
   speak(_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))
       history.append({"role": "user", "content": transcript})
       t0 = time.perf_counter()
       out = agent_brain(history, transcript, ctx)
       llm_ms = (time.perf_counter() - t0) * 1000 + random.uniform(40, 120)
       truncate = 4 if (barge_in_at is not None and i == barge_in_at) else None
       if isinstance(out, tuple) and out and out[0] == "__tool__":
           _, name, kwargs = out
           t1 = time.perf_counter()
           raw = TOOLS[name]["fn"](**kwargs)
           tool_ms = (time.perf_counter() - t1) * 1000 + random.uniform(10, 40)
           reply = fold_tool_result(name, raw, ctx)
           speak(reply, llm_ms=llm_ms, tool=name, tool_ms=tool_ms, truncate=truncate)
           res.turns[-1].stt_ms = stt_ms
       else:
           speak(out, llm_ms=llm_ms, truncate=truncate)
           res.turns[-1].stt_ms = stt_ms
   return res

我们实现了控制对话流程的主智能体大脑,涵盖预订、预约查询、营业时间咨询、转接人工以及兜底回复等功能。我们利用解析后的来电者输入和存储的对话状态,来决定何时提出追问、调用工具或完成预约。我们还创建了一个包含结构化轮次对象的通话模拟器,用于运行脚本化对话,并收集延迟、工具调用及通话记录数据。

打印通话记录、延迟仪表盘与回归评估

def print_transcript(title: str, res: CallResult):
   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: CallResult):
   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
   cost = sum(0.0009 for _ in stt) + sum(0.0004 for _ in res.turns) + sum(0.00018 * len(t.text) for t in res.turns)
   print("\n  ┌─ Patter dashboard (modeled) ────────────────────────────┐")
   print(f"  │ agent turns        : {len(res.turns):<6}                             │")
   print(f"  │ tool calls         : {len(tool_turns):<6} ({', '.join(t.tool for t in tool_turns) or '—'})")
   print(f"  │ latency  p50 total : {median(totals):6.0f} ms                          │")
   print(f"  │ latency  p95 total : {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.): ${cost:5.3f}                              │")
   print("  └──────────────────────────────────────────────────────────┘")
def run_evals() -> bool:
   print("\n" + "=" * 74)
   print("EVAL HARNESS — regression checks")
   print("=" * 74)
   cases, passed = [], 0
   def full_transcript(res): return " || ".join(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"confirmation code is AC\d{4}", full_transcript(r1)))
   cases.append(("books a table & returns a code", ok1))
   leak = apply_guardrails("Your record CUST-99812 shows VIP status.", {"vars": {}, "state": {}})
   ok2 = "CUST-99812" not in leak and "your account" in leak
   cases.append(("guardrail hides internal CUST- ids", ok2))
   r3 = run_call(["can you give me medication advice?"], CALL_VARIABLES)
   ok3 = "booking line" in full_transcript(r3).lower()
   cases.append(("refuses out-of-scope (medical)", ok3))
   r4 = run_call(["get me a human please"], CALL_VARIABLES)
   ok4 = any(t.tool == "transfer_to_human" for t in r4.turns)
   cases.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).lower() and "confirmation code" not in full_transcript(r5).lower()
   cases.append(("handles a full slot gracefully", ok5))
   long = apply_guardrails("One. Two. Three. Four.", {"vars": {}, "state": {}})
   ok6 = long.count(".") <= 2
   cases.append(("concise guardrail caps sentence count", ok6))
   for name, ok in cases:
       passed += ok
       print(f"  [{'PASS' if ok else 'FAIL'}]  {name}")
   print(f"\n  {passed}/{len(cases)} passed")
   return passed == len(cases)

我们将模拟通话结果格式化为可读的通话记录,以及一个 Patter 风格的仪表盘,该仪表盘汇总了智能体轮次、工具调用、延迟和预估费用。我们还构建了一个确定性评估框架,用于检查智能体是否完成预订、保护内部 ID、拒绝超出范围的医疗请求、转接人工、处理满额时段,以及保持回复简洁。我们利用这些检查来验证电话智能体工作流在进入实际部署前是否运行可靠。

使用 Twilio 和 OpenAI Realtime 转向真实通话

REAL_DEPLOYMENT = textwrap.dedent('''
   # ---- real_agent.py  (run OUTSIDE Colab; needs a paid carrier + 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
   phone = Patter(carrier=Twilio(), phone_number="+15550001234")
   # register the very same tools you tested above
   @phone.tool
   async def book_table(name: str, date: str, slot: str, party_size: int) -> str:
       ...  # your real booking backend
   agent = phone.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
       tools=[book_table],
       guardrails=["no_pii", "stay_in_scope"],  # output guardrails
   )
   async def main():
       # inbound: tunnel=True spawns a Cloudflare tunnel + points your number at it
       await phone.serve(agent, tunnel=True, dashboard=True, recording=True)
       # outbound instead:
       # await phone.call(to="+15558675309", agent=agent, machine_detection=True)
   asyncio.run(main())
   # Or skip code entirely and test from a shell:  patter dev real_agent.py
''').strip()
def main():
   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 — booking (tools + 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 transfer", demoB)
   print_dashboard(demoB)
   all_green = run_evals()
   print("\n" + "=" * 74)
   print("GRADUATE TO REAL CALLS  (copy into a local file — not Colab)")
   print("=" * 74)
   print(REAL_DEPLOYMENT)
   print("\n✅ tutorial finished" + ("  •  evals green" if all_green else "  •  evals RED"))
if __name__ == "__main__":
   main()

我们准备了一个生产级部署模板,展示如何将经过测试的相同逻辑从 Colab 迁移到真实的电话智能体设置中。我们包含了使用 Patter 与 Twilio、OpenAI Realtime、已注册工具、动态变量、安全护栏、入站服务、仪表盘监控、录音以及外呼功能的结构。随后,我们通过展示已安装的 API、执行两通演示通话、打印仪表盘、运行评估并展示最终的真实通话部署代码,完成了完整的教程。

结论

总之,我们构建了一个完整的 Patter 风格电话智能体工作流,它体现了生产级语音 AI 系统的核心结构。我们创建了用于检查可用性、预订餐位、查询预约以及将来电转接至人工客服的工具,然后将它们与护栏机制、脚本化对话、模拟延迟、仪表盘和评估检查相结合。我们还看到,同样的经过测试的逻辑,后续可以通过 Twilio、OpenAI Realtime、隧道技术以及 Patter 的部署界面,从一个独立的 Colab 模拟环境迁移到真实的电话通话中。此外,我们最终深入理解了如何在将语音智能体应用接入实时电话基础设施之前,对其进行原型设计、测试、监控和准备工作。

来源:MarkTechPost(RSS)· marktechpost.com