在本教程中,我们将通过构建一个语音智能体工作流来探索 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 模拟环境迁移到真实的电话通话中。最后,我们深刻理解了如何在将语音智能体应用接入实时电话基础设施之前,对其进行原型设计、测试、监控和准备工作。