Patter SDK で電話予約エージェント構築ガイド
MarkTechPost は、Patter SDK を活用してリアルタイムの電話予約エージェントを構築する包括的なチュートリアルを提供し、動的変数、ガードレール、遅延ダッシュボード、評価チェックといった実運用に必要な要素を実装可能にしている。
キーポイント
完全なシミュレーション環境の構築
実際の電話回線や認証情報を必要とせず、Patter SDK を用いて音声通話のフローを完全にシミュレートするワークフローを提供している。
運用に必要な機能の実装ガイド
動的な顧客変数の定義、呼び出し可能なツールの登録、出力ガードレール(安全性チェック)、および音声合成・認識の動作シミュレーション方法を詳述している。
パフォーマンスと品質の評価体系
モデル化された遅延(レイテンシ)とコストメトリクスの追跡、回帰テスト形式によるシステム検証を通じて、本番環境での信頼性を確保する手法を示している。
構造化パイプラインの統合
エージェントロジック、ツール使用、安全チェック、シミュレーション機能を単一の構造化された音声エージェントパイプラインとして統合する方法を解説している。
動的ツール定義と自動登録
Pythonのデコレーター(@tool)を使用して関数を定義し、パラメータや説明を自動的に抽出してツールカタログに登録する仕組みを実装しています。
シミュレーション環境の完全なリセット機能
_reset_backend 関数により、各通話シナリオ開始前にデータベースとテーブル状態を初期値に強制リセットし、評価テストの決定論的な実行を保証しています。
バージョン非依存な API 検出ロジック
SDK のインストール状況を確認し、存在しない場合は自動インストールを試みることで、異なる環境や SDK バージョン間での互換性を確保する柔軟な初期化処理を実装しています。
重要な引用
simulate speech-to-text and text-to-speech behavior, and run a complete scripted call flow without requiring live telephony credentials
track modeled latency and cost metrics, and validate the system through regression-style evaluations
integrates agent logic, tool use, safety checks, call simulation, and real-world deployment patterns into a single structured voice-agent pipeline
"Each simulated call starts from a clean backend (a real call hits a fresh DB connection). Keeps the eval suite deterministic across runs."
Print the *actual* installed Patter API so the tutorial adapts to whatever version Colab pulled (getpatter is young & moves weekly).
We define the dynamic caller variables, create a small tool registry, and prepare an in-memory restaurant backend for availability, reservations, hours, and transfers.
影響分析・編集コメントを表示
影響分析
この記事は、音声 AI エージェントの実装における「開発から本番運用までのギャップ」を埋める実践的なガイドラインを提供しています。特に、セキュリティやパフォーマンス評価といった実務で直面する課題に対する具体的な解決策を示すことで、開発者がより信頼性の高い自律型エージェントを短期間で構築・展開できる環境を整えることに寄与します。
編集コメント
音声 AI エージェントの実装において、単なるプロトタイプ作成だけでなく、セキュリティやパフォーマンス評価まで含めた本番運用レベルのガイドが提供されている点は非常に価値が高いです。特に、実際の電話回線に接続しなくてもテスト可能なシミュレーション機能は、開発サイクルを大幅に短縮する要因となるでしょう。
本チュートリアルでは、Patter SDK を活用して、AI 電話アシスタントが実際の会話でどのように振る舞うかをシミュレートする音声エージェントワークフローの構築方法を解説します。
ここではレストラン予約というユースケースを取り上げます。動的な通話者変数の定義、呼び出し可能なツールの登録、出力ガードレールの適用、音声認識(STT)と音声合成(TTS)の挙動シミュレーション、そして生きた電話回線の認証情報なしで完結するスクリプトベースの通話フローの実行を行います。
さらに、利用可能な場合はインストールされた Patter API の確認、決定論的なエージェント・ブレインの作成、モデル化されたレイテンシとコストメトリクスの追跡、そして回帰テスト形式の評価によるシステムの検証も実施します。最後に、Patter SDK がどのようにしてエージェントロジック、ツール活用、安全チェック、通話シミュレーション、そして実世界での展開パターンを、単一の構造化された音声エージェント・パイプラインに統合しているのかを理解します。
Patter SDK、ツールのセットアップとレストランバックエンドの構築
{
"status": "success",
"message": "SDK setup complete"
}ブラウザを変更してコードをコピーしてください。
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():
"""import 可能な場合は実体の getpatter モジュールを返し、そうでなければ 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():
"""チュートリアルが Colab で取得したバージョンに合わせて適応できるよう、実際にインストールされている Patter API を表示する。"""
print("=" * 74)
print("PATTER SDK — installed API")
print("=" * 74)
if _PATTER is None:
print("getpatter がこのカーネルで import できないようです(問題ありません。以下の Colab デモは完結しています)。" "\n" "新鮮な Colab 環境では pip install getpatter を実行し、このブロックがライブ 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}")
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():
"""各シミュレーションされた通話ではバックエンドをクリーンな状態から開始します(実際の通話では新しい DB 接続が使用されます)。これにより、評価スイートの実行結果を一定に保ちます。"""
_OPEN_TABLES.clear(); _OPEN_TABLES.update(copy.deepcopy(_OPEN_TABLES_INIT))
_RES_DB.clear(); _RES_DB.update(copy.deepcopy(_RES_DB_INIT))
@tool("日付・時間帯と人数に対してテーブルが空いているか確認する")
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("テーブルを予約し、確認コードを返す")
def book_table(name: str, date: str, slot: str, party_size: int) -> str:
seats = _OPEN_TABLES.get((date, slot), 0)
if seats >= party_size:
code = f"{random.randint(100000, 999999)}"
_RES_DB[code] = f"Table for {party_size}, {date} {slot}, under {name}."
_OPEN_TABLES[(date, slot)] -= party_size
return f"CONFIRMED: {code}"
return "FULL: not enough seats available."
def get_hours(day_type: str) -> str:
return _HOURS.get(day_type, _HOURS["weekday"])
@tool("確認コードを使って既存の予約を検索する")
def lookup_reservation(code: str) -> str:
return _RES_DB.get(code.upper(), "NOT_FOUND: no reservation with that code.")
@tool("人間オペレーターに転送する(Patter が自動的に transfer_call を注入します)")
def transfer_to_human(reason: str) -> str:
return f"TRANSFER: routing to a host — reason: {reason}."
チュートリアル環境のセットアップでは、必要なライブラリのインポートに加え、Patter SDK のインストール(任意)と、利用可能な場合の API 確認を行います。次に、動的な通話者変数を定義し、簡易なツールレジストリを作成。さらに、在庫状況の確認や予約、営業時間照会、人間担当者への転送などを処理するインメモリ型のレストランバックエンドを準備します。
これにより、シミュレーション上の電話エージェントがテーブル確認、予約実行、確認コードの検索、そして必要に応じて人間担当者に接続といった一連の操作を行えるようになります。
出力ガードレールと音声レイヤーの追加
Copy CodeCopiedUse a different Browser
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"(? 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 の隠蔽、不要な言語の除去、本題からの逸脱要求のブロックを行い、応答を短く保つことで通話体験を向上させます。その後、音声認識(STT)とテキスト読み上げ(TTS)の挙動をシミュレーションし、システムプロンプトを定義して、人数、日付、時間帯、名前、予約コードといった項目を軽量なパーサー関数で解析できるようにします。
エージェントの脳と通話シミュレータの構築
Copy CodeCopiedUse a different Browser
def agent_brain(history: list, user_text: str, ctx: dict):
"""コアロジック。ctx には変数と会話ごとのステートが保持されます。"""
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:
"""ツール結果を自然な口語応答に変換します(Patter 呼び出しで関数呼び出しの結果を受け取った LLM の動作)"""
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"
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"
image {title}")
print("─" * 74)
for who, txt in res.transcript:
tag = "
image agent " if who == "agent" else "
image 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): 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)
原文を表示
In this tutorial, we explore the Patter SDK by building a voice-agent workflow that simulates how an AI phone assistant behaves during real conversations. We work with a restaurant booking use case in which we define dynamic caller variables, register callable tools, apply output guardrails, simulate speech-to-text and text-to-speech behavior, and run a complete scripted call flow without requiring live telephony credentials. We also inspect the installed Patter API when available, create a deterministic agent brain, track modeled latency and cost metrics, and validate the system through regression-style evaluations. Finally, we understand how the Patter SDK integrates agent logic, tool use, safety checks, call simulation, and real-world deployment patterns into a single structured voice-agent pipeline.
Setting Up the Patter SDK, Tools, and Restaurant Backend
Copy CodeCopiedUse a different Browser
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}."
We set up the tutorial environment by importing the required libraries, optionally installing the Patter SDK, and inspecting the installed API when it is available. We define the dynamic caller variables, create a small tool registry, and prepare an in-memory restaurant backend for availability, reservations, hours, and transfers. We also register the core tools that allow our simulated phone agent to check tables, book reservations, look up confirmation codes, and route callers to a human.
Adding Output Guardrails and Simulated Speech Layers
Copy CodeCopiedUse a different Browser
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
We build the output guardrail layer that keeps the phone assistant safe, concise, and appropriate for the booking use case. We redact sensitive information, hide internal customer IDs, clean unwanted language, block off-topic requests, and keep responses short for a better phone experience. We then simulate speech-to-text and text-to-speech behavior, define the system prompt, and add lightweight parsing functions for party size, date, slot, name, and reservation code.
Building the Agent Brain and Call Simulator
Copy CodeCopiedUse a different Browser
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"
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
We implement the main agent brain that controls the conversation flow across booking, reservation lookup, opening-hours questions, human transfer, and fallback responses. We use the parsed caller input and stored conversation state to decide when to ask follow-up questions, call tools, or complete a reservation. We also create a call simulator with structured turn objects to run scripted conversations and collect latency, tool, and transcript data.
Printing Transcripts, Latency Dashboards, and Regression Evals
Copy CodeCopiedUse a different Browser
def print_transcript(title: str, res: CallResult):
print("\n" + "─" * 74)
print(f"
image {title}")
print("─" * 74)
for who, txt in res.transcript:
tag = "
image agent " if who == "agent" else "
image 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)
関連記事
今日のまとめ
AI日報で今日の重要ニュースをまとめ読み