Patter SDK で電話予約エージェント構築ガイド

本チュートリアルでは、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)