/* ========================================================= code2.jsx — the listings for v16–v29. Extends the same CODE registry defined in code.jsx. Never write a bare dollar-brace inside these template literals; escape shell variables as "$VAR". ========================================================= */ /* ============ RT1 · v16 — the latency budget ============ */ CODE.v16 = { note: { zh: "延迟要能被测量才能被优化。Python 在链路的六个点打时间戳,算出每一段的耗时并输出 P50/P95——注意它把「顾客说完」这个时刻单独记下来,因为一切都是从那里开始计时的;配置是编排器里那几个真正决定延迟的参数;最后是一段生产环境的查询,用来回答「这周慢在哪一跳」。", en: "Latency has to be measurable before it can be optimised. The Python stamps six points along the pipeline, reports each stage and prints P50/P95 — and note that it records the moment the customer stopped speaking separately, because everything is timed from there. The configuration holds the orchestrator parameters that actually move latency. The last tab is the production query that answers which hop got slow this week.", }, tabs: [ { lang: "Python", k: "py", file: "latency.py", src: `"""Six timestamps. Everything is measured from 'the customer stopped talking'.""" import time, json, statistics from contextlib import contextmanager STAGES = ["network", "endpoint", "asr_final", "llm_first_token", "tts_first_packet", "playout"] class TurnTimer: def __init__(self, turn_id): self.turn_id, self.marks, self.t0 = turn_id, {}, None def speech_ended(self): """Time zero. Not 'request received' — the caller's last syllable.""" self.t0 = time.perf_counter() def mark(self, stage): self.marks[stage] = (time.perf_counter() - self.t0) * 1000 def report(self): prev, out = 0.0, {} for s in STAGES: if s in self.marks: out[s] = round(self.marks[s] - prev, 1) prev = self.marks[s] out["total_ms"] = round(prev, 1) return out @contextmanager def timed(timer, stage): yield timer.mark(stage) def summarise(path="turns.jsonl"): rows = [json.loads(l) for l in open(path)] tot = sorted(r["total_ms"] for r in rows) p = lambda q: tot[max(0, int(len(tot) * q) - 1)] print(f"turns {len(rows)} p50 {p(0.5):.0f} ms p95 {p(0.95):.0f} ms max {tot[-1]:.0f} ms") for s in STAGES: # which hop owns the budget? vals = [r[s] for r in rows if s in r] if vals: print(f" {s:<18} mean {statistics.mean(vals):6.0f} p95 {sorted(vals)[int(len(vals)*.95)-1]:6.0f}") # Typical shape before optimisation: # endpoint 600 | llm 430 | tts 260 | asr 140 | playout 90 | network 45 = 1565 ms # The longest bar is a threshold you chose, not a model you bought.`, }, { lang: "配置 / config", k: "yaml", file: "pipeline.yaml", src: `pipeline: # The three overlaps. All of them are "start earlier", none is "run faster". overlaps: warm_llm_on_partial: true # send partial ASR text to the LLM before the final synthesise_first_sentence: true# start TTS on the model's first streamed sentence play_while_generating: true # never wait for the whole utterance budgets_ms: # alert when a stage exceeds its budget endpoint: 700 asr_final: 200 llm_first_token: 500 tts_first_packet: 300 playout: 100 total_p95: 1500 timeouts_ms: asr: 4000 llm: 6000 tool_call: 3000 tts: 4000 degradation: # what to do instead of hanging on_llm_timeout: "先说一句「稍等一下」,然后重试一次,再不行转人工" on_tool_timeout: "告诉顾客系统有点慢,改成人工回拨" on_tts_timeout: "切换到轻量音色" filler: enabled: true after_ms: 600 # a 300 ms "嗯" buys 800 ms of perceived patience never_before_commitment: true`, }, { lang: "对接 / SQL", k: "sql", file: "latency_report.sql", src: `-- Which hop got slower this week, and did callers notice? SELECT DATE(created_at) AS d, COUNT(*) AS turns, ROUND(AVG(total_ms)) AS mean_ms, ROUND(MAX(CASE WHEN pct_rank <= 0.95 THEN total_ms END)) AS p95_ms, ROUND(AVG(endpoint_ms)) AS endpoint, ROUND(AVG(asr_final_ms)) AS asr, ROUND(AVG(llm_first_token_ms)) AS llm, ROUND(AVG(tts_first_packet_ms)) AS tts FROM ( SELECT t.*, PERCENT_RANK() OVER (PARTITION BY DATE(created_at) ORDER BY total_ms) AS pct_rank FROM turn_latency t WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 14 DAY) ) x GROUP BY d ORDER BY d; -- The question that matters: do slow turns end badly? SELECT CASE WHEN total_ms < 900 THEN 'fast' WHEN total_ms < 1500 THEN 'ok' ELSE 'slow' END AS bucket, COUNT(*) AS turns, ROUND(AVG(caller_hung_up), 3) AS hangup_rate, ROUND(AVG(handed_off), 3) AS handoff_rate FROM turn_latency WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY bucket;`, }, ], }; /* ============ RT2 · v17 — barge-in ============ */ CODE.v17 = { note: { zh: "打断处理的难点不是「检测到声音就停」,而是「别把自己的声音当成顾客」。Python 是一个带三重闸门的打断检测器:回声消除之后的能量、持续时长、以及语义确认;配置是音频侧参数;第三个标签处理最容易被忘的一件事——打断之后怎么恢复状态,已经说过的不重播,已经确认的槽位不丢。", en: "The hard part of barge-in is not stopping when you hear sound; it is not mistaking your own voice for the customer. The Python is an interrupt detector with three gates: post-AEC energy, duration, and a semantic check. The configuration holds the audio-side parameters. The third tab handles the most commonly forgotten part — recovering state after an interrupt, replaying nothing already said and losing no slot already confirmed.", }, tabs: [ { lang: "Python", k: "py", file: "barge_in.py", src: `"""Listen while speaking, without hearing yourself.""" import numpy as np, time class BargeInDetector: def __init__(self, aec, vad, gate_ms=220, energy_db=-32): self.aec, self.vad = aec, vad self.gate_ms, self.energy_db = gate_ms, energy_db self.voiced_ms = 0.0 self.playing = False def on_playback(self, pcm): """Feed the AEC its reference: this is what we are about to emit.""" self.aec.push_reference(pcm) self.playing = True def on_microphone(self, pcm, frame_ms=20): clean = self.aec.process(pcm) # subtract our own audio db = 20 * np.log10(np.sqrt(np.mean(clean.astype(np.float32) ** 2)) + 1e-9) # Gate 1: is anything there at all, after echo cancellation? if db < self.energy_db or not self.vad.is_speech(clean): self.voiced_ms = 0.0 return None # Gate 2: sustained, not a cough, a door chime or a television consonant. self.voiced_ms += frame_ms if self.voiced_ms < self.gate_ms: return None return "interrupt" def semantic_gate(partial_text: str) -> bool: """Gate 3: did they say something, or did the room make a noise?""" if not partial_text or len(partial_text.strip()) < 2: return False noise = {"嗯", "啊", "哦", "呃"} # backchannel: keep talking return partial_text.strip() not in noise def handle_turn(agent, detector): for frame in agent.microphone(): if detector.on_microphone(frame) == "interrupt": partial = agent.asr.partial_text() if semantic_gate(partial): agent.stop_playback() # stop mid-word, like a person agent.remember_spoken_prefix() # <- chapter's most forgotten line return partial`, }, { lang: "配置 / config", k: "yaml", file: "audio.yaml", src: `audio: sample_rate: 16000 frame_ms: 20 aec: enabled: true backend: webrtc # speexdsp | webrtc | vendor SDK filter_length_ms: 200 # longer covers more acoustic delay, costs CPU nonlinear_processing: true # residual echo suppression after the linear filter # Carriers usually apply their own AEC on a phone leg. It is never enough: # delay and non-linear distortion leave residue, and residue triggers interrupts. barge_in: enabled: true # half-duplex (mute while speaking) is worse duration_gate_ms: 220 # sustained speech, not a transient energy_threshold_db: -32 semantic_gate: true # is the recognised text meaningful? ignore_backchannel: ["嗯", "啊", "哦", "对", "好"] # do not stop for agreement noise: suppression: moderate # aggressive suppression eats quiet speakers expected_shop_ambience_db: -38 # television, music, a second customer recovery: replay_unspoken_only: true # never repeat what the caller already heard keep_confirmed_slots: true # an interrupt must not reset the booking`, }, { lang: "对接 / Python", k: "py", file: "recover_state.py", src: `"""After an interrupt: do not start over, and do not repeat yourself.""" class SpeechState: def __init__(self): self.queued_text = "" # what we intended to say self.spoken_chars = 0 # how much actually reached the caller's ear self.confirmed_slots = {} def on_chunk_played(self, chunk: str): self.spoken_chars += len(chunk) def on_interrupted(self): """The caller cut in. Work out what they did and did not hear.""" heard = self.queued_text[: self.spoken_chars] unheard = self.queued_text[self.spoken_chars:] self.queued_text = "" return heard, unheard def resume_text(self, unheard: str, answer_to_interruption: str) -> str: """Answer them first; only then finish the sentence, if it still matters.""" if not unheard.strip(): return answer_to_interruption # Do not simply replay: a person would rephrase, shorter. return f"{answer_to_interruption} 刚才没说完的是,{unheard.strip()}" # A booking that survives an interrupt is a booking the caller trusts. state = SpeechState() state.confirmed_slots = {"service": "肩颈理疗", "day": "2026-09-13"} state.queued_text = "好的,明天下午四点半的肩颈理疗,技师是王师傅,费用一百三十八元。" state.spoken_chars = 14 # they cut in here heard, unheard = state.on_interrupted() print("heard :", heard) print("unheard:", unheard) print("slots survive the interrupt:", state.confirmed_slots)`, }, ], }; /* ============ RT3 · v18 — cascade vs end-to-end ============ */ CODE.v18 = { note: { zh: "两条架构的代码摆在一起,差别一眼就能看出来:级联里有一个「文本」变量,你可以在它身上做任何检查;端到端里没有那个变量。Python 展示两种接法;配置是按环节分流的路由规则(报价走级联、闲聊走端到端);第三个标签是一个对比测试台,用同一批话术跑两条链路并记录延迟与可审计性。", en: "Put the two architectures side by side and the difference is immediately visible: the cascade has a text variable you can inspect, and the end-to-end version does not. The Python shows both wirings. The configuration routes by stage — quotes through the cascade, small talk through end-to-end. The third tab is a comparison harness running the same scripts through both and recording latency and auditability.", }, tabs: [ { lang: "Python", k: "py", file: "two_architectures.py", src: `"""Same job, two shapes. Notice where 'text' exists — and where it does not.""" import asyncio, json, websockets, anthropic client = anthropic.Anthropic() async def cascade(audio_stream, asr, tts, guardrails, tools): """ASR -> LLM -> TTS. Slower, and every hop is inspectable.""" text = await asr.transcribe(audio_stream) # <- a text variable exists msg = client.messages.create( model="claude-sonnet-5", max_tokens=300, temperature=0.3, tools=tools, messages=[{"role": "user", "content": text}]) reply = msg.content[0].text ok, reply, reason = guardrails.check(reply) # <- you can act on it here log_turn(user_text=text, agent_text=reply, blocked=reason) # <- and audit it async for chunk in tts.stream(reply): yield chunk async def end_to_end(audio_stream, session_prompt): """Audio in, audio out. Faster, warmer — and there is no text to check.""" async with websockets.connect("wss://realtime.vendor/v1/speech") as ws: await ws.send(json.dumps({ "type": "session.update", "session": {"instructions": session_prompt, "turn_detection": {"type": "server_vad", "silence_ms": 650}, "voice": "shop_female_warm"}, })) async def send(): async for pcm in audio_stream: await ws.send(json.dumps({"type": "input_audio_buffer.append", "audio": pcm.hex()})) async def recv(): async for raw in ws: ev = json.loads(raw) if ev["type"] == "response.audio.delta": yield bytes.fromhex(ev["delta"]) # straight to the speaker # A transcript event may arrive too — but AFTER the audio is # already on its way to the caller. That is the whole problem. await asyncio.gather(send(), recv())`, }, { lang: "配置 / config", k: "yaml", file: "routing.yaml", src: `# Route by what the stage promises, not by what sounds nicer. architecture_routing: default: cascade rules: - stage: greeting route: end_to_end why: 没有承诺,只有语气 —— 自然度值钱 - stage: small_talk route: end_to_end why: 安抚、寒暄、等待时的闲聊 - stage: intent_capture route: cascade why: 需要结构化槽位 - stage: quote_price route: cascade why: 价格是承诺,必须经过校验与留痕 - stage: booking route: cascade why: 写操作,需要工具调用与幂等 - stage: complaint route: human why: 需要共情与授权,两条链路都不该接 - stage: health_question route: cascade why: 必须过禁语与安全模板,端到端审不住 cascade: asr: funasr-streaming llm: claude-sonnet-5 tts: cosyvoice2 end_to_end: vendor: realtime-speech max_turn_sec: 20 transcript_logging: required # keep whatever transcript you can get`, }, { lang: "对接 / Python", k: "py", file: "compare_harness.py", run: "# 用同一批真实话术跑两条链路,记录延迟与可审计性", src: `"""Run the same scripts through both architectures and compare honestly.""" import asyncio, json, time, statistics SCRIPTS = [ ("greeting", "喂,你好"), ("ask_price", "肩颈理疗多少钱"), ("booking", "明天下午三点约一下,王师傅"), ("health", "我颈椎不太好,能做吗"), ("oob", "你们晚上有别的服务吗"), ] async def run_once(arch, audio): t0 = time.perf_counter() first = None text_available = False async for chunk in arch(audio): if first is None: first = (time.perf_counter() - t0) * 1000 text_available = text_available or getattr(chunk, "text", None) is not None return {"first_audio_ms": round(first), "text_layer": text_available} async def main(): rows = [] for stage, utterance in SCRIPTS: audio = synth_caller_audio(utterance) casc = await run_once(cascade_pipeline, audio) e2e = await run_once(end_to_end_pipeline, audio) rows.append({ "stage": stage, "utterance": utterance, "cascade_ms": casc["first_audio_ms"], "e2e_ms": e2e["first_audio_ms"], "cascade_auditable": casc["text_layer"], "e2e_auditable": e2e["text_layer"], }) print(json.dumps(rows, ensure_ascii=False, indent=2)) print("median gain from end-to-end:", statistics.median(r["cascade_ms"] - r["e2e_ms"] for r in rows), "ms") print("stages where only the cascade is auditable:", [r["stage"] for r in rows if r["cascade_auditable"] and not r["e2e_auditable"]]) asyncio.run(main())`, }, ], }; /* ============ PF1 · v19 — the vendor bake-off ============ */ CODE.v19 = { note: { zh: "厂商选型唯一靠得住的方法是用自己的录音做盲测。Python 是一个统一封装:把同一批门店录音喂给不同厂商,统一算 CER 和槽位准确率;配置列出候选与权重;第三个标签把结果和权重合成一张加权得分表——注意它同时输出「决定性因素」,因为很多时候差距只来自一个维度。", en: "The only trustworthy way to choose a vendor is a blind test on your own recordings. The Python is a thin common wrapper feeding one set of shop recordings to several vendors and scoring CER and slot accuracy identically. The configuration lists the candidates and the weights. The third tab combines results and weights into a scored table — and prints the deciding factor, because the gap often comes from a single dimension.", }, tabs: [ { lang: "Python", k: "py", file: "bakeoff.py", src: `"""One interface, several vendors, your own audio. Anything else is marketing.""" import os, time, json, yaml from evaluate import edit_ops, slot_accuracy CFG = yaml.safe_load(open("vendors.yaml", encoding="utf-8")) def transcribe(vendor: str, wav_path: str) -> dict: t0 = time.perf_counter() if vendor == "aliyun": from alibabacloud_nls import Recognizer text = Recognizer(os.environ["ALI_KEY"]).recognise(wav_path) elif vendor == "tencent": from tencentcloud_asr import recognise text = recognise(wav_path, os.environ["TC_SECRET"]) elif vendor == "azure": import azure.cognitiveservices.speech as speechsdk cfg = speechsdk.SpeechConfig(subscription=os.environ["AZURE_KEY"], region="chinaeast2") cfg.speech_recognition_language = "zh-CN" text = speechsdk.SpeechRecognizer(cfg, speechsdk.AudioConfig(filename=wav_path)).recognize_once().text elif vendor == "funasr": from funasr import AutoModel text = AutoModel(model="paraformer-zh").generate(input=wav_path)[0]["text"] else: raise ValueError(vendor) return {"text": text, "latency_ms": round((time.perf_counter() - t0) * 1000)} def run(testset="testset/reference.jsonl"): refs = [json.loads(l) for l in open(testset, encoding="utf-8")] results = {} for v in CFG["candidates"]: rows = [] for r in refs: out = transcribe(v, r["audio"]) rows.append({**r, "hypothesis": out["text"], "latency_ms": out["latency_ms"]}) cer = sum(edit_ops(r["reference"], r["hypothesis"])["cer"] for r in rows) / len(rows) results[v] = {"cer": round(cer, 4), "slots": slot_accuracy(rows), "p95_latency_ms": sorted(r["latency_ms"] for r in rows)[int(len(rows) * .95) - 1]} json.dump(results, open("bakeoff.json", "w"), ensure_ascii=False, indent=2) return results print(json.dumps(run(), ensure_ascii=False, indent=2))`, }, { lang: "配置 / config", k: "yaml", file: "vendors.yaml", src: `candidates: [aliyun, tencent, iflytek, volcano, azure, funasr] # Your constraints, as weights. Everything else is somebody's brochure. weights: mandarin_telephony: 5 # 8 kHz, the channel you actually run on dialect_accent: 3 # how much local accent do your callers have? hotwords: 4 # therapist and service names must survive streaming: 5 # realtime or batch? voice_cloning: 2 on_premise: 2 # can data leave the building? unit_price: 3 compliance: 4 # data residency, contracts, audit support # The test set that decides it. Build this before you talk to anyone. testset: size: 120 # 120 real calls is enough to separate vendors must_include: - 电话 8 kHz 录音,不是手机录的干净样本 - 至少 20 通带明显口音或方言 - 至少 30 通包含手机号 - 至少 20 通包含技师姓名与项目名 - 至少 10 通有背景噪声(电视、说话声) labelled_by: 人工逐字校对 # the reference must be exact, or nothing means anything notes: | Prices and capabilities change constantly — treat published numbers as indicative and confirm with the vendor. What does not change is the method: your audio, your metric, one blind comparison.`, }, { lang: "对接 / Python", k: "py", file: "score_vendors.py", src: `"""Combine measured results with your weights. Print the deciding factor.""" import json, yaml CFG = yaml.safe_load(open("vendors.yaml", encoding="utf-8")) MEASURED = json.load(open("bakeoff.json", encoding="utf-8")) # Capability scores 0-5 you fill in from docs, trials and the contract talk. CAPS = yaml.safe_load(open("capabilities.yaml", encoding="utf-8")) W = CFG["weights"] def score(vendor): caps = dict(CAPS[vendor]) m = MEASURED.get(vendor) if m: # measurement overrides the brochure caps["mandarin_telephony"] = max(0, min(5, round((0.12 - m["cer"]) / 0.02))) caps["hotwords"] = max(0, min(5, round(m["slots"]["therapist"] * 5))) caps["streaming"] = 5 if m["p95_latency_ms"] < 900 else 3 total = sum(W[k] * caps.get(k, 0) for k in W) return total / (sum(W.values()) * 5), caps ranked = sorted(((score(v)[0], v, score(v)[1]) for v in CAPS), reverse=True) for fit, v, caps in ranked: print(f"{v:<10} fit {fit:.0%}") best, runner = ranked[0], ranked[1] gaps = {k: W[k] * (best[2].get(k, 0) - runner[2].get(k, 0)) for k in W} decisive = max(gaps, key=gaps.get) print(f"\\nwinner: {best[1]} — decided by: {decisive}") print("If that dimension does not actually matter to you, re-weight and re-run.")`, }, ], }; /* ============ PF2 · v20 — build vs buy ============ */ CODE.v20 = { note: { zh: "自建的成本从来不是显卡钱。Python 把两条成本曲线算出来并求交叉点,注意它把运维人天单独列成一项——那是最容易被漏掉、也往往是最大的一笔;配置是容量与单价;部署给出一份自建全家桶的 compose,同时注意最后那段混合部署的逻辑:日常自建、高峰溢出到云,很多连锁最后都落在这里。", en: "The cost of self-hosting was never the card. The Python computes both curves and their crossing, with engineer days broken out as their own line — the one most often forgotten and frequently the largest. The configuration holds capacity and prices. The deployment tab stands up the whole self-hosted stack, and note the hybrid logic at the end: baseline in-house with peak overflow to the cloud, which is where many chains actually land.", }, tabs: [ { lang: "Python", k: "py", file: "tco.py", src: `"""Where does self-hosting start being cheaper? Include the human.""" import math def cloud_cost(minutes_per_month, unit_price_per_min=0.055): return minutes_per_month * unit_price_per_min def build_cost(peak_concurrency, sessions_per_gpu=16, gpu_per_month=1800, power_per_gpu=120, ops_days=2.0, day_rate=1200): cards = max(1, math.ceil(peak_concurrency / sessions_per_gpu)) return { "cards": cards, "gpu": cards * gpu_per_month, "power": cards * power_per_gpu, "ops": ops_days * day_rate, # the line everyone forgets "total": cards * (gpu_per_month + power_per_gpu) + ops_days * day_rate, } def break_even(peak_concurrency, **kw): b = build_cost(peak_concurrency, **kw) unit = kw.get("unit_price_per_min", 0.055) return b["total"] / unit, b for conc in (8, 16, 40, 120): minutes, b = break_even(conc) print(f"peak={conc:3d} cards={b['cards']} build=CNY {b['total']:>7,.0f}/mo " f"(ops {b['ops']:,.0f}) break-even at {minutes:>9,.0f} min/month") # A single shop runs maybe 9,000 minutes a month. It will never cross. # A twenty-shop chain runs ~180,000. It might — if it has the person.`, }, { lang: "配置 / config", k: "yaml", file: "capacity.yaml", src: `capacity: # Measure these on YOUR hardware with YOUR models. Never take the datasheet. asr: model: paraformer-streaming measured_rtf: 0.05 sessions_per_24gb_gpu: 28 tts: model: cosyvoice2-0.5b measured_rtf: 0.06 speaking_duty_cycle: 0.38 sessions_per_24gb_gpu: 43 combined_sessions_per_gpu: 16 # both on one card, with headroom costs: gpu_rent_per_month_cny: 1800 # or depreciation of a purchased card over 36 months power_per_gpu_per_month_cny: 120 ops_engineer_day_rate_cny: 1200 ops_days_per_month: 2.0 # realistically 2-4 once something goes wrong cloud: asr_per_minute_cny: 0.04 tts_per_10k_chars_cny: 3.0 llm_input_per_million_cny: 2.4 llm_output_per_million_cny: 9.6 all_in_per_minute_cny: 0.055 # what it actually adds up to per call minute hard_constraints: data_must_stay_on_premise: false # if true, cost stops being the question peak_concurrency: 16 minutes_per_month: 60000`, }, { lang: "部署 / compose", k: "yaml", file: "docker-compose.stack.yml", run: "# 自建全家桶:ASR + TTS + LLM + 编排,一张卡起步", src: `services: asr: image: funasr/runtime:latest ports: ["10095:10095"] volumes: ["./models:/workspace/models", "./hotwords.txt:/workspace/hotwords.txt:ro"] deploy: { resources: { reservations: { devices: [{ driver: nvidia, count: 1, capabilities: [gpu] }] } } } tts: image: cosyvoice/cosyvoice2:latest ports: ["9880:9880"] volumes: ["./voices:/app/voices:ro"] deploy: { resources: { reservations: { devices: [{ driver: nvidia, count: 1, capabilities: [gpu] }] } } } llm: image: vllm/vllm-openai:latest ports: ["8000:8000"] command: ["--model", "Qwen/Qwen3-8B-Instruct", "--max-model-len", "8192"] deploy: { resources: { reservations: { devices: [{ driver: nvidia, count: 1, capabilities: [gpu] }] } } } agent: build: ./agent ports: ["7860:7860"] environment: ASR_URL: ws://asr:10095 TTS_URL: http://tts:9880 LLM_URL: http://llm:8000/v1 # Hybrid: when local capacity is saturated, spill to the cloud rather # than making callers wait. Most chains end up here. OVERFLOW_ENABLED: "true" OVERFLOW_THRESHOLD_SESSIONS: "14" OVERFLOW_VENDOR: aliyun depends_on: [asr, tts, llm] # Before committing to this: who gets up at 2 a.m. when it dies? # That answer, not the GPU price, decides whether you should build.`, }, ], }; /* ============ PF3 · v21 — per-call cost ============ */ CODE.v21 = { note: { zh: "把成本做成一个函数,才能知道该优化哪一段。Python 按四段计价并输出瀑布;配置是单价表(记得写上抓取日期,这些数字变得很快);SQL 把真实用量按天聚合出来——注意那个「输入 token 占比」列,它通常是最容易砍的一刀:检索从 top-8 降到 top-3、历史做摘要,这一项直接腰斩。", en: "Turn cost into a function and you learn which line to optimise. The Python prices the four segments and prints the waterfall. The configuration is the price table — with the date it was captured, because these numbers move. The SQL aggregates real usage by day, and note the input-token share column: it is usually the easiest cut, since dropping retrieval from top-8 to top-3 and summarising history halves it.", }, tabs: [ { lang: "Python", k: "py", file: "call_cost.py", src: `"""What one call costs, and which line to attack first.""" import yaml P = yaml.safe_load(open("pricing.yaml", encoding="utf-8"))["prices"] def call_cost(duration_min=3.0, turns=9, input_tokens_per_turn=2400, output_tokens_per_turn=140, tts_chars_per_turn=42): asr = duration_min * P["asr_per_min"] tel = duration_min * P["telephony_per_min"] llm = (turns * input_tokens_per_turn / 1e6) * P["llm_input_per_m"] \\ + (turns * output_tokens_per_turn / 1e6) * P["llm_output_per_m"] tts = (turns * tts_chars_per_turn / 10000) * P["tts_per_10k_chars"] parts = {"telephony": tel, "llm": llm, "asr": asr, "tts": tts} return parts, sum(parts.values()) def human_cost(salary=5200, duration_min=3.0, utilisation=0.6, days=22, hours=8): calls_month = days * hours * 60 * utilisation / duration_min return salary * 1.35 / calls_month # 1.35 = contributions and workspace parts, total = call_cost() for k, v in sorted(parts.items(), key=lambda kv: -kv[1]): print(f"{k:<10} CNY {v:.4f} {v / total:5.1%}") print(f"{'TOTAL':<10} CNY {total:.4f} vs human CNY {human_cost():.2f}") # The optimisation that always pays: cut the input context. cheap, cheap_total = call_cost(input_tokens_per_turn=900) # top-3 + summarised history print(f"\\nwith compressed context: CNY {cheap_total:.4f} " f"({(total - cheap_total) / total:.0%} cheaper)")`, }, { lang: "配置 / config", k: "yaml", file: "pricing.yaml", src: `# Indicative unit prices. ALWAYS re-check against the vendor's own page — # these move several times a year and vary with contract volume. captured_on: "2026-09-01" currency: CNY prices: asr_per_min: 0.04 # streaming recognition, per minute of audio telephony_per_min: 0.08 # inbound, landline; mobile and outbound differ llm_input_per_m: 2.4 # per million input tokens llm_output_per_m: 9.6 # per million output tokens tts_per_10k_chars: 3.0 # neural voice; cloned voices often cost more typical_call: duration_min: 3.0 turns: 9 input_tokens_per_turn: 2400 # system prompt + retrieved chunks + history output_tokens_per_turn: 140 tts_chars_per_turn: 42 optimisation_levers: # ranked by payoff / effort - lever: retrieval top_k 8 -> 3 saves: "roughly half the input tokens" - lever: summarise history beyond 6 turns saves: "another 20-30% on long calls" - lever: cache the system prompt saves: "most vendors bill cached input at a fraction" - lever: shorter replies (fewer TTS characters) saves: "small, but also improves the experience"`, }, { lang: "对接 / SQL", k: "sql", file: "cost_report.sql", src: `-- Real spend by day, with the share that is easiest to cut. SELECT DATE(started_at) AS d, COUNT(*) AS calls, ROUND(SUM(duration_sec) / 60.0) AS minutes, ROUND(SUM(duration_sec) / 60.0 * 0.04, 2) AS asr_cny, ROUND(SUM(duration_sec) / 60.0 * 0.08, 2) AS telephony_cny, ROUND(SUM(input_tokens) / 1e6 * 2.4, 2) AS llm_input_cny, ROUND(SUM(output_tokens) / 1e6 * 9.6, 2) AS llm_output_cny, ROUND(SUM(tts_chars) / 1e4 * 3.0, 2) AS tts_cny, ROUND(SUM(input_tokens) / NULLIF(SUM(input_tokens + output_tokens), 0), 3) AS input_share FROM call_usage WHERE started_at >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY d ORDER BY d; -- Cost against outcome. A call that books is worth far more than it costs. SELECT outcome, -- booked | answered | handed_off | abandoned COUNT(*) AS calls, ROUND(AVG(duration_sec)) AS avg_sec, ROUND(AVG(input_tokens)) AS avg_input_tokens, ROUND(AVG(duration_sec)/60*0.12 + AVG(input_tokens)/1e6*2.4, 3) AS approx_cost_cny FROM call_usage WHERE started_at >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY outcome ORDER BY calls DESC;`, }, ], }; /* ============ CH1 · v22 — telephony ============ */ CODE.v22 = { note: { zh: "电话接入的两半:工程和合规。Python 通过 FreeSWITCH 的事件套接字接管来电并把音频分叉给 AI;第二个标签是真实的拨号方案,注意开头那条录音告知——它必须在最前面且要留痕;第三个标签是外呼前的合规检查,同意、拒绝、时段、频次四道闸门,任何一道不过就不拨。", en: "Telephony has two halves, engineering and compliance. The Python takes a call over the FreeSWITCH event socket and forks its audio to the AI. The second tab is a real dialplan — note the recording notice at the very top, which must come first and be logged. The third is the pre-dial compliance check: consent, refusal, time of day and frequency, four gates, and any failure means the call is not placed.", }, tabs: [ { lang: "Python", k: "py", file: "fs_agent.py", src: `"""Answer a call on FreeSWITCH, fork the audio into the AI pipeline.""" import asyncio, greenswitch ESL = greenswitch.InboundESL(host="127.0.0.1", port=8021, password="ClueCon") async def on_new_call(uuid, caller_number): # 1. Notice first. Always. Logged, not just spoken. ESL.send(f"api uuid_broadcast {uuid} notice_recording.wav aleg") await log_consent(uuid, caller_number, notice="recording_notice_v3") # 2. Fork the media to our websocket — this is the audio the AI hears. ESL.send(f"api uuid_audio_fork {uuid} start " f"ws://127.0.0.1:7860/media/{uuid} mono 8000") # 3. Run the agent. It writes audio back over the same socket. result = await run_agent(uuid, caller_number) # 4. Handoff means bridging to a real extension, mid-call, with context. if result["handoff"]: await push_context_card(result) ESL.send(f"api uuid_transfer {uuid} -bleg 2001 XML default") await log_outcome(uuid, result) def on_event(event): if event.headers.get("Event-Name") == "CHANNEL_ANSWER": asyncio.create_task(on_new_call( event.headers["Unique-ID"], event.headers.get("Caller-Caller-ID-Number"))) ESL.connect() ESL.register_handle("all", on_event) ESL.process_events()`, }, { lang: "配置 / dialplan", k: "xml", file: "dialplan/shop.xml", src: ` `, }, { lang: "合规 / Python", k: "py", file: "outbound_guard.py", run: "# 四道闸门,任何一道不过就不拨 —— 号码是门店的资产", src: `"""Never place an outbound call that any of these four gates rejects.""" from datetime import datetime, time, timedelta import sqlite3 db = sqlite3.connect("contacts.db") QUIET_START, QUIET_END = time(20, 30), time(9, 30) # do not call outside these MAX_PER_MONTH = 2 MIN_GAP_DAYS = 10 class OutboundRefused(Exception): pass def may_call(phone: str, now=None) -> bool: now = now or datetime.now() row = db.execute( "SELECT consent, opted_out, last_called_at, calls_this_month " "FROM contacts WHERE phone = ?", (phone,)).fetchone() if row is None: raise OutboundRefused("no record: consent was never obtained") consent, opted_out, last_called_at, calls_this_month = row # Gate 1: prior consent for commercial voice calls. if not consent: raise OutboundRefused("no prior consent") # Gate 2: a refusal is permanent. There is no 'try again next quarter'. if opted_out: raise OutboundRefused("contact opted out — never call again") # Gate 3: time of day. t = now.time() if t >= QUIET_START or t < QUIET_END: raise OutboundRefused(f"quiet hours ({t})") # Gate 4: frequency. if calls_this_month >= MAX_PER_MONTH: raise OutboundRefused("monthly cap reached") if last_called_at and (now - datetime.fromisoformat(last_called_at)) < timedelta(days=MIN_GAP_DAYS): raise OutboundRefused("too soon since the last call") return True def record_opt_out(phone: str, heard: str): """The caller said stop. Honour it immediately and permanently.""" db.execute("UPDATE contacts SET opted_out = 1, opted_out_at = ?, " "opt_out_phrase = ? WHERE phone = ?", (datetime.now().isoformat(), heard, phone)) db.commit()`, }, ], }; /* ============ CH2 · v23 — WeChat and platform IM ============ */ CODE.v23 = { note: { zh: "私域触达的代码里,最重要的不是发消息,是发之前的那几个检查:窗口内吗、频次超了吗、这个人上次是不是刚被打扰过。Python 就是这套检查加发送;配置是消息模板与频次策略;SQL 找出「到期该提醒」的会员——按每个人自己的到店周期,而不是统一群发。", en: "In private-domain code, the important part is not sending but the checks before sending: are we inside the window, is the frequency cap hit, was this person just contacted. The Python is those checks plus the send. The configuration holds templates and the frequency policy. The SQL finds the members who are genuinely due — by each person's own visit cycle rather than one broadcast for everyone.", }, tabs: [ { lang: "Python", k: "py", file: "wecom_push.py", src: `"""Reach a member — only if you are allowed to, and only if it is the moment.""" from datetime import datetime, timedelta import requests, yaml, sqlite3 CFG = yaml.safe_load(open("outreach.yaml", encoding="utf-8")) db = sqlite3.connect("crm.db") def within_service_window(member_id) -> bool: """Official-account service messages: 48 hours after the member interacted.""" row = db.execute("SELECT last_interaction_at FROM members WHERE id = ?", (member_id,)).fetchone() if not row or not row[0]: return False return datetime.now() - datetime.fromisoformat(row[0]) < timedelta(hours=48) def frequency_ok(member_id) -> bool: n = db.execute( "SELECT COUNT(*) FROM outreach_log WHERE member_id = ? " "AND sent_at >= date('now','start of month')", (member_id,)).fetchone()[0] return n < CFG["policy"]["max_per_month"] def send(member_id, template_key, params): if not frequency_ok(member_id): return {"sent": False, "reason": "monthly cap"} channel = "service_message" if within_service_window(member_id) else "wecom_1v1" if channel == "wecom_1v1" and not db.execute( "SELECT wecom_external_id FROM members WHERE id = ?", (member_id,)).fetchone()[0]: return {"sent": False, "reason": "outside window and not on WeCom"} text = CFG["templates"][template_key].format(**params) r = requests.post(CFG["endpoints"][channel], timeout=5, json={"member_id": member_id, "text": text}) db.execute("INSERT INTO outreach_log(member_id, template, channel, sent_at) " "VALUES (?,?,?,?)", (member_id, template_key, channel, datetime.now().isoformat())) db.commit() return {"sent": r.ok, "channel": channel} # Timing beats copywriting: send when THIS member is due, not on Fridays. for m in due_members(): send(m["id"], "due_reminder", {"name": m["name"], "service": m["usual_service"], "days": m["days_since"]})`, }, { lang: "配置 / config", k: "yaml", file: "outreach.yaml", src: `policy: max_per_month: 2 # 4+ and the blocked-contact curve turns up sharply min_gap_days: 10 quiet_hours: ["21:00", "09:30"] stop_on_keywords: ["退订", "别发了", "不要再发", "TD"] # Every message must carry a way out. It is both the rule and good sense. opt_out_footer: "回复 TD 退订" templates: due_reminder: | {name}您好,您上次做{service}是{days}天前啦。 这两天下午还有空位,需要我帮您留一个吗?回复 TD 退订 empty_slot: | {name}您好,今天下午{time}刚空出一位{therapist}的{service}, 要给您留着吗?回复 TD 退订 membership_expiry: | {name}您好,您的会员卡里还有{sessions}次{service},{date}到期。 需要帮您约上吗?回复 TD 退订 endpoints: service_message: https://internal/wechat/service_message wecom_1v1: https://internal/wecom/send channels: official_account: window_hours: 48 # service messages only inside this window outside_window: template_message # restricted, do not abuse mini_program: note: 有独立客服会话,下单场景内触达最自然 wecom: note: 关系长期留存、会话可存档 —— 门店最该经营的资产 platform_im: # Meituan / Dianping / Douyin first_response_target_sec: 30 # first-response time feeds your ranking auto_reply: true escalate_after_turns: 4`, }, { lang: "对接 / SQL", k: "sql", file: "due_members.sql", src: `-- Who is genuinely due, by their own rhythm — not a broadcast list. WITH visits AS ( SELECT member_id, MAX(visit_date) AS last_visit, COUNT(*) AS visits, DATEDIFF(MAX(visit_date), MIN(visit_date)) / NULLIF(COUNT(*) - 1, 0) AS avg_cycle_days FROM visit WHERE visit_date >= DATE_SUB(CURDATE(), INTERVAL 18 MONTH) GROUP BY member_id HAVING visits >= 2 ) SELECT m.id, m.name, m.usual_service, v.last_visit, ROUND(v.avg_cycle_days) AS cycle_days, DATEDIFF(CURDATE(), v.last_visit) AS days_since, ROUND(DATEDIFF(CURDATE(), v.last_visit) / NULLIF(v.avg_cycle_days, 0), 2) AS overdue_ratio FROM members m JOIN visits v ON v.member_id = m.id LEFT JOIN outreach_log o ON o.member_id = m.id AND o.sent_at >= DATE_SUB(CURDATE(), INTERVAL 10 DAY) WHERE m.opted_out = 0 AND o.id IS NULL -- not contacted recently AND DATEDIFF(CURDATE(), v.last_visit) BETWEEN v.avg_cycle_days AND v.avg_cycle_days * 2.5 ORDER BY overdue_ratio DESC LIMIT 200; -- The number to watch monthly: is the reachable pool shrinking? SELECT DATE_FORMAT(opted_out_at, '%Y-%m') AS month, COUNT(*) AS opted_out FROM members WHERE opted_out = 1 GROUP BY month ORDER BY month;`, }, ], }; /* ============ CH3 · v24 — human handoff ============ */ CODE.v24 = { note: { zh: "转人工的代码有两半:什么时候转,以及转过去之后人看到什么。Python 是触发判断,注意顾客直说要找人那一条是无条件立即生效的;配置是四类触发信号与阈值;第三个标签是上下文交接卡的结构——坐席必须在三秒内看懂顾客是谁、要什么、AI 卡在哪,否则顾客要从头讲一遍,前面的效率全被抵消。", en: "Handoff code has two halves: when to transfer, and what the human sees afterwards. The Python is the trigger logic, and note that the caller asking for a person is unconditional and immediate. The configuration holds the four signal classes and their thresholds. The third tab is the structure of the context card — an agent must grasp who, what and where it stuck within three seconds, or the customer repeats everything and every earlier gain is cancelled.", }, tabs: [ { lang: "Python", k: "py", file: "handoff.py", src: `"""Four signals. One of them overrides everything else.""" import re, yaml CFG = yaml.safe_load(open("handoff.yaml", encoding="utf-8")) ASK_HUMAN = re.compile("转人工|找(个)?(真)?人|人工|真人|店长|不要机器") HUMAN_ONLY = set(CFG["human_only_intents"]) NEGATIVE = set(CFG["emotion"]["negative_words"]) def should_hand_off(turn) -> dict: # 1. The caller asked. Immediate, unconditional, no retention attempt. if ASK_HUMAN.search(turn["user_text"]): return {"handoff": True, "priority": "immediate", "reason": "caller asked"} # 2. Intent is on the humans-only list. if turn["intent"] in HUMAN_ONLY: return {"handoff": True, "priority": "high", "reason": f"intent {turn['intent']}"} # 3. Emotion: rate, volume and wording moving together. e = turn["prosody"] negatives = sum(w in turn["user_text"] for w in NEGATIVE) if (e["rate_ratio"] > 1.35 and e["volume_db_delta"] > 6) or negatives >= 2: return {"handoff": True, "priority": "high", "reason": "emotional signal"} # 4. Confidence: two consecutive unclear turns is already one too many. if turn["asr_confidence"] < CFG["thresholds"]["asr_confidence"] \\ and turn["consecutive_low_confidence"] >= 2: return {"handoff": True, "priority": "normal", "reason": "cannot hear the caller"} if turn["retrieval_score"] < CFG["thresholds"]["retrieval_score"]: return {"handoff": True, "priority": "normal", "reason": "no grounded answer"} return {"handoff": False} def transfer(session, decision): card = build_context_card(session) # see the third tab queue = "priority" if decision["priority"] == "immediate" else "normal" agent_desk.push(card, queue=queue) # Say it plainly. Never "let me try one more thing". return "好的,马上帮您转接前台同事,请稍等。"`, }, { lang: "配置 / config", k: "yaml", file: "handoff.yaml", src: `thresholds: asr_confidence: 0.62 # below this twice in a row: stop guessing retrieval_score: 0.62 # no grounded answer available max_turns_before_offer: 8 # a long circling conversation is a failed one human_only_intents: - complaint # needs empathy AND authority - refund - health_condition # never diagnose, never negotiate - out_of_bounds # refuse, end, log, escalate - injury_or_dispute emotion: negative_words: ["投诉", "退钱", "太差", "生气", "垃圾", "骗", "曝光", "消协"] rate_ratio_trigger: 1.35 # speaking 35% faster than their own baseline volume_db_trigger: 6 queue: agents_on_duty: 2 target_wait_sec: 25 # past ~30 s the damage exceeds a wrong AI answer overflow_action: callback # offer a callback rather than holding them never: - 挽留顾客("要不我再帮您试试") - 先问原因再转接 - 转接后让顾客重复一遍已经说过的内容`, }, { lang: "对接 / JSON", k: "json", file: "context_card.json", run: "// 坐席要在三秒内看懂:谁、要什么、卡在哪", src: `{ "card_version": "1.2", "session_id": "c_8f21a0", "handed_off_at": "2026-09-12T20:07:41+08:00", "reason": "caller asked for a person", "priority": "immediate", "caller": { "phone": "138****5768", "member": true, "name": "李女士", "visits_last_90d": 4, "usual_service": "肩颈理疗", "usual_therapist": "王师傅", "last_visit": "2026-08-20" }, "what_they_want": { "intent": "make_booking", "slots_confirmed": { "service": "肩颈理疗", "day": "2026-09-13", "time": "16:30" }, "slots_missing": ["party_size"], "hold_id": "h_44c1", "hold_expires_at": "2026-09-12T20:19:00+08:00" }, "where_it_stuck": { "turn": 8, "question": "团购券能不能和会员卡一起用?", "why": "knowledge base has no entry for voucher stacking", "agent_said": "这个规则我帮您问一下前台" }, "transcript_tail": [ { "who": "caller", "text": "我上次买的团购券能和会员卡一起用吗?" }, { "who": "agent", "text": "这个规则我帮您问一下前台,先把时间定下来好吗?" }, { "who": "caller", "text": "好的。那你让人跟我说一下吧。" } ], "do_not_repeat": ["时间已确认 16:30", "技师已确认 王师傅", "价格已报 138 元"] }`, }, ], }; /* ============ OP1 · v25 — the QA loop ============ */ CODE.v25 = { note: { zh: "全量质检的实现很直接:转写 + 让模型按评分卡逐条打分。真正的工程在后面——把坏案例归因到组件(识别错、知识缺、话术差、工具失败),并且把每一个修复冻进回归集。Python 是打分器;配置是评分卡;SQL 出周报,注意那个置信区间列:样本太小时它会告诉你「这周的改进看不出来」。", en: "Full-coverage QA is straightforward to build: transcribe, then have a model score each call against a rubric. The engineering is what follows — attributing each bad case to a component (misrecognition, missing knowledge, weak script, tool failure) and freezing every fix into a regression set. The Python is the scorer, the configuration the rubric, and the SQL the weekly report — note the confidence-interval column, which tells you honestly when the sample is too small to see this week's improvement.", }, tabs: [ { lang: "Python", k: "py", file: "auto_qa.py", src: `"""Score every call, then attribute every failure to something you can fix.""" import json, yaml, anthropic client = anthropic.Anthropic() RUBRIC = yaml.safe_load(open("rubric.yaml", encoding="utf-8")) PROMPT = """你是一家按摩养生门店的客服质检员。按下面的评分卡给这通电话打分。 评分卡: {rubric} 同时判断:如果这通电话不理想,根因属于哪一类? asr 顾客说的话被识别错了 knowledge 知识库里没有这条,或者答错了 script 话术生硬、答非所问、该转人工没转 tool 查询或下单失败、超时、重复 none 没有问题 只输出 JSON:{{"scores": {{...}}, "outcome": "...", "root_cause": "...", "evidence": "引用出问题的那一句"}} 通话转写: {transcript}""" def score(transcript: str) -> dict: msg = client.messages.create( model="claude-sonnet-5", max_tokens=600, temperature=0, messages=[{"role": "user", "content": PROMPT.format( rubric=yaml.dump(RUBRIC, allow_unicode=True), transcript=transcript[:8000])}]) return json.loads(msg.content[0].text) def run(day): bad = [] with open(f"qa/{day}.jsonl", "w", encoding="utf-8") as out: for line in open(f"transcripts/{day}.jsonl", encoding="utf-8"): call = json.loads(line) r = score(call["text"]) out.write(json.dumps({**call, **r}, ensure_ascii=False) + "\\n") if r["outcome"] != "good": bad.append({**r, "call_id": call["id"]}) # Every bad case becomes a regression test. That is how fixes stay fixed. with open("regression/cases.jsonl", "a", encoding="utf-8") as f: for b in bad: f.write(json.dumps(b, ensure_ascii=False) + "\\n") return bad`, }, { lang: "配置 / rubric", k: "yaml", file: "rubric.yaml", src: `# Score what the shop cares about, not what is easy to measure. dimensions: understood: weight: 3 question: 机器是否正确理解了顾客的意图与关键信息 grounded: weight: 3 question: 所有价格、时间、技师信息是否来自查询而不是编造 completed: weight: 3 question: 顾客的目的是否达成(约上了 / 问清了 / 正确转人工) natural: weight: 2 question: 是否抢话、是否呆滞、读法是否正确、语气是否得体 compliant: weight: 5 question: 是否出现疗效表述、是否正确拒绝越界请求、是否播报了录音告知 handoff: weight: 2 question: 该转人工的是否转了,转的时候是否干脆 outcome_rules: good: 所有维度达标 acceptable: 有小瑕疵但顾客目的达成 bad: 顾客目的未达成,或任一合规维度不达标 alerting: compliant_below: 1.0 # any compliance failure pages someone, same day bad_rate_above: 0.08 sample_floor: 400 # below this, do not draw conclusions from a change`, }, { lang: "对接 / SQL", k: "sql", file: "weekly_qa.sql", src: `-- The weekly report. Note the interval: it tells you when NOT to conclude. SELECT YEARWEEK(call_date) AS wk, COUNT(*) AS calls_scored, ROUND(AVG(outcome = 'good'), 3) AS good_rate, ROUND(AVG(outcome = 'bad'), 3) AS bad_rate, -- 95% interval on the bad rate: sqrt(p(1-p)/n) * 1.96 ROUND(1.96 * SQRT(AVG(outcome='bad') * (1 - AVG(outcome='bad')) / COUNT(*)), 4) AS bad_rate_margin, ROUND(AVG(self_served), 3) AS self_service_rate, ROUND(AVG(handed_off), 3) AS handoff_rate, SUM(root_cause = 'asr') AS cause_asr, SUM(root_cause = 'knowledge') AS cause_knowledge, SUM(root_cause = 'script') AS cause_script, SUM(root_cause = 'tool') AS cause_tool FROM qa_scores WHERE call_date >= DATE_SUB(CURDATE(), INTERVAL 12 WEEK) GROUP BY wk ORDER BY wk; -- This week's fix list, ranked by how much it would actually buy. SELECT root_cause, evidence, COUNT(*) AS occurrences FROM qa_scores WHERE outcome = 'bad' AND call_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY) GROUP BY root_cause, evidence ORDER BY occurrences DESC LIMIT 20;`, }, ], }; /* ============ OP2 · v26 — the compliance gate ============ */ CODE.v26 = { note: { zh: "合规必须写进代码,写在文档里的合规等于没有。Python 是一道上线前的闸门:十四项义务逐条检查,任何一项不过就拒绝发布;配置是检查单本身,每项都注明依据与自查方法;SQL 是留痕与到期删除——告知记录、同意记录、保留期限,以及一个每天跑的清理任务。本章为工程提示,不构成法律意见。", en: "Compliance has to live in code; compliance that lives in a document does not exist. The Python is a pre-release gate walking fourteen duties and refusing the deploy if any fails. The configuration is the checklist itself, each item carrying its basis and a self-check. The SQL is the record and the expiry job — notice logs, consent records, retention periods and a daily cleanup. This chapter is engineering guidance, not legal advice.", }, tabs: [ { lang: "Python", k: "py", file: "compliance_gate.py", src: `"""A release gate. If this fails, the deploy does not happen.""" import sys, yaml from datetime import date CHECKS = yaml.safe_load(open("compliance.yaml", encoding="utf-8"))["checks"] def evaluate(state: dict): """state comes from config, the database and a few live probes.""" failures = [] for c in CHECKS: if c.get("applies_when") and not state.get(c["applies_when"]): continue # e.g. outbound rules when no outbound if not state.get(c["key"]): failures.append(c) return failures def probe_live_system(): """Do not trust the config file — check the running system.""" return { "recording_notice_played": ivr.has_prompt("recording_notice"), "consent_logged": db.exists("consent_log", since_days=1), "voiceprint_separate_consent": not features.voiceprint_enabled or db.exists("biometric_consent"), "minimal_collection": storage.retention_days() <= 90, "retention_disclosed": kb.contains("录音保存期限"), "access_and_deletion": api.route_exists("/privacy/delete"), "synthetic_labelled": tts.config["announce_synthetic"] is True, "voice_authorised": all(v["authorised"] for v in voices.cloned()), "outbound_consent": outbound.checks_consent(), "outbound_optout": outbound.has_optout_path(), "outbound_hours": outbound.quiet_hours_enforced(), "no_efficacy_claims": guardrails.banned_terms_active(), "symptom_template": prompts.contains("不能判断"), "oob_refusal_logged": guardrails.oob_logging_enabled(), # context flags "does_outbound": outbound.enabled, "uses_cloned_voice": bool(voices.cloned()), } if __name__ == "__main__": failures = evaluate(probe_live_system()) for f in failures: print(f"FAIL {f['key']}\\n {f['zh']}\\n 基于:{f['basis']}\\n" f" 自查:{f['self_check']}\\n") if failures: print(f"{len(failures)} compliance checks failed — refusing to deploy.") sys.exit(1) print(f"all {len(CHECKS)} checks passed on {date.today()}")`, }, { lang: "配置 / checklist", k: "yaml", file: "compliance.yaml", src: `# Engineering checklist, not legal advice. Have a lawyer review before launch. checks: - key: recording_notice_played zh: 通话开始时明确告知可能录音 basis: 录音属于个人信息处理,需告知并取得同意 self_check: 随机抽 10 通录音,听开头 5 秒 - key: consent_logged zh: 同意记录可查、可追溯 basis: 需要能证明取得过同意 self_check: 查 consent_log 表最近 24 小时有无写入 - key: voiceprint_separate_consent zh: 声纹等生物识别信息取得单独同意 basis: 属于敏感个人信息 self_check: 若未启用声纹功能则不适用 - key: minimal_collection zh: 最小必要,不为「以后训练」留全量录音 basis: 收集范围应与目的相称 self_check: 查存储保留策略的实际天数 - key: retention_disclosed zh: 明示存储期限并到期删除 basis: 期限届满应删除或匿名化 self_check: 隐私说明里是否写明天数,清理任务是否在跑 - key: access_and_deletion zh: 提供查询、更正与删除通道 basis: 个人信息主体的基本权利 self_check: 调用 /privacy/delete 是否真的删掉 - key: synthetic_labelled zh: 合成语音按规定标识 basis: 生成合成内容标识要求 self_check: 开场白里是否说明这是智能助手 - key: voice_authorised zh: 使用真人音色取得本人书面授权 basis: 《民法典》第 1023 条参照肖像权保护声音 self_check: voices.yaml 里每个克隆音色的授权是否在有效期内 applies_when: uses_cloned_voice - key: outbound_consent zh: 外呼事先取得同意 basis: 商业性语音呼叫的强制要求 self_check: 拨号前是否真的查过 consent 字段 applies_when: does_outbound - key: outbound_optout zh: 提供便捷退订并立即生效 basis: 拒绝后不得再拨 self_check: 说一次「别再打了」,看记录是否立刻置位 applies_when: does_outbound - key: outbound_hours zh: 外呼时段与频次限制 basis: 避免骚扰与投诉 self_check: 20:30 之后尝试拨号应被拒绝 applies_when: does_outbound - key: no_efficacy_claims zh: 禁止治疗 / 疗效 / 根治类表述 basis: 养生服务不是医疗服务 self_check: 跑 test_guardrails.py 的八条试探 - key: symptom_template zh: 症状类咨询走安全模板,不作诊断 basis: 不得替代就医 self_check: 问「我颈椎疼能做吗」,看回复是否邀请到店评估 - key: oob_refusal_logged zh: 越界请求:拒绝、结束、留痕 basis: 正规门店的底线 self_check: 查 guardrail_log 是否记录了拒绝事件`, }, { lang: "对接 / SQL", k: "sql", file: "retention.sql", src: `-- Consent and notice: the record you will be asked for. CREATE TABLE consent_log ( id BIGINT AUTO_INCREMENT PRIMARY KEY, call_id VARCHAR(64) NOT NULL, phone_hash VARCHAR(64) NOT NULL, -- store a hash, not the number notice_id VARCHAR(32) NOT NULL, -- which wording was played played_at TIMESTAMP NOT NULL, consent_type VARCHAR(32) NOT NULL, -- recording | voiceprint | marketing granted TINYINT NOT NULL, evidence_uri VARCHAR(255) NULL, -- the audio segment of the notice UNIQUE KEY uniq_call_type (call_id, consent_type) ); -- Retention: disclosed, enforced by a job, not by good intentions. -- recordings 90 days, transcripts 180, aggregates forever (anonymised). DELETE FROM call_recording WHERE created_at < DATE_SUB(CURDATE(), INTERVAL 90 DAY); UPDATE call_transcript SET phone = NULL, caller_name = NULL, anonymised = 1 WHERE created_at < DATE_SUB(CURDATE(), INTERVAL 180 DAY) AND anonymised = 0; -- A subject deletion request must actually delete. DELETE FROM call_recording WHERE phone_hash = SHA2(?, 256); DELETE FROM call_transcript WHERE phone_hash = SHA2(?, 256); UPDATE members SET deleted = 1, phone = NULL WHERE phone_hash = SHA2(?, 256); -- Guardrail events: refusals are kept deliberately, and reviewed. SELECT DATE(created_at) AS d, kind, COUNT(*) AS n FROM guardrail_log WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY d, kind ORDER BY d DESC, n DESC;`, }, ], }; /* ============ OP3 · v27 — the flywheel ============ */ CODE.v27 = { note: { zh: "飞轮要自动转。Python 每周跑一次:从质检结果里挖出高频坏案例,自动提出三类改进建议(该加的热词、该补的知识条目、该改的话术),并生成一个 PR 草稿;配置是分阶段落地的路线图,每个阶段有明确的准入条件;最后是把这一切串起来的 cron 与准入检查——没达标就不许进下一阶段。", en: "The flywheel should turn itself. The Python runs weekly: mine the QA results for recurring bad cases and propose three kinds of improvement — hotwords to add, knowledge entries to write, script changes to make — and open a draft change. The configuration is the staged roadmap with explicit entry criteria for each phase. The last tab is the cron that ties it together plus the gate that refuses to advance a phase before its criteria are met.", }, tabs: [ { lang: "Python", k: "py", file: "flywheel.py", src: `"""Weekly: turn last week's failures into next week's configuration.""" import json, collections, re, yaml def load_bad_cases(path="qa/last_week.jsonl"): return [json.loads(l) for l in open(path, encoding="utf-8") if json.loads(l)["outcome"] == "bad"] def propose_hotwords(cases, min_hits=3): """Names and service words that the ASR keeps getting wrong.""" counter = collections.Counter() for c in cases: if c["root_cause"] != "asr": continue for token in re.findall(r"[\\u4e00-\\u9fa5]{2,4}(?=师傅|理疗|按摩|推拿)", c["evidence"]): counter[token] += 1 return [w for w, n in counter.items() if n >= min_hits] def propose_kb_entries(cases, min_hits=2): """Questions the shop could not answer. Each one is a missing entry.""" counter = collections.Counter(c["evidence"] for c in cases if c["root_cause"] == "knowledge") return [q for q, n in counter.items() if n >= min_hits] def propose_script_fixes(cases): return [{"evidence": c["evidence"], "call_id": c["call_id"]} for c in cases if c["root_cause"] == "script"][:10] def weekly(): cases = load_bad_cases() report = { "bad_cases": len(cases), "add_hotwords": propose_hotwords(cases), "write_kb_entries": propose_kb_entries(cases), "review_script": propose_script_fixes(cases), "tool_failures": sum(c["root_cause"] == "tool" for c in cases), } # Append the proposals so a human reviews them — never auto-merge a prompt. open("proposals/" + today() + ".json", "w", encoding="utf-8").write( json.dumps(report, ensure_ascii=False, indent=2)) # And freeze every bad case as a regression test, forever. with open("regression/cases.jsonl", "a", encoding="utf-8") as f: for c in cases: f.write(json.dumps(c, ensure_ascii=False) + "\\n") return report print(json.dumps(weekly(), ensure_ascii=False, indent=2))`, }, { lang: "配置 / roadmap", k: "yaml", file: "roadmap.yaml", src: `# Text before voice, read before write. A shop has no budget for trial and error. phases: - id: 1 months: "1-2" scope: 文字渠道问答(微信 + 平台 IM),知识库上线,不碰写操作 entry_criteria: [] exit_criteria: self_service_rate: ">= 0.60" hallucination_rate: "<= 0.02" kb_coverage_smoke_test: ">= 0.90" - id: 2 months: "3-4" scope: 在 IM 上加预约写操作(查空档 / 占位 / 落单) entry_criteria: [phase_1_exit_met] exit_criteria: booking_success_rate: ">= 0.97" duplicate_bookings_per_month: "<= 1" ghost_holds_per_day: "<= 3" - id: 3 months: "5-8" scope: 上电话:实时链路、打断、转人工 entry_criteria: [phase_2_exit_met, compliance_gate_passed] exit_criteria: p95_turn_latency_ms: "<= 1500" false_barge_in_per_min: "<= 0.3" handoff_context_complete: ">= 0.95" - id: 4 months: "9-12" scope: 外呼与主动召回,全量质检制度化 entry_criteria: [phase_3_exit_met, outbound_compliance_passed] exit_criteria: complaint_rate: "<= 0.005" opt_out_honoured: "1.0" weekly_loop_running: true cadence: qa_scoring: daily proposal_review: weekly prompt_change: "只在有回归集通过时才发布" compliance_review: quarterly`, }, { lang: "对接 / shell", k: "sh", file: "weekly.sh", run: "# crontab: 0 9 * * 1 /opt/voice/weekly.sh", src: `#!/usr/bin/env bash # Monday morning: score, propose, gate. Nothing here should need a human until # the proposals land in someone's inbox. set -euo pipefail cd /opt/voice echo "== 1. score everything from last week ==" python auto_qa.py --week last echo "== 2. mine proposals ==" python flywheel.py | tee "proposals/$(date +%F).json" echo "== 3. run the regression set against the CURRENT config ==" if ! pytest regression/ -q; then echo "regression failed — do not ship any config change this week" exit 1 fi echo "== 4. compliance gate ==" python compliance_gate.py || { echo "compliance gate failed"; exit 1; } echo "== 5. phase gate: may we move to the next phase? ==" python - <<'PY' import yaml, json road = yaml.safe_load(open("roadmap.yaml", encoding="utf-8"))["phases"] metrics = json.load(open("metrics/current.json", encoding="utf-8")) cur = int(open("phase.txt").read().strip()) phase = next(p for p in road if p["id"] == cur) unmet = [k for k, rule in phase["exit_criteria"].items() if not eval(f"metrics[{k!r}] {rule.strip('\"')}")] print("phase", cur, "unmet criteria:", unmet or "none — you may advance") PY echo "== done. proposals are waiting for a human to review =="`, }, ], }; /* ============ CS1 · v28 — the chain's ledger ============ */ CODE.v28 = { note: { zh: "把投入产出写成一个函数,老板就能自己改参数。Python 是那本账,注意「被劝退的顾客」是一个显式的负项——不写进去的模型都是在骗自己;配置是这家连锁的真实数字;SQL 从门店的系统里把这些数字取出来,这样这本账就不是拍脑袋,而是可以每个月重算一次的。", en: "Write the investment case as a function and the owner can move the parameters themselves. The Python is the ledger, with deflected customers as an explicit negative line — a model without it is lying to you. The configuration holds this chain's real numbers. The SQL pulls those numbers out of the shop's own systems, so the ledger stops being a guess and becomes something you recompute monthly.", }, tabs: [ { lang: "Python", k: "py", file: "roi.py", src: `"""The ledger. Every line is adjustable; one of them is negative on purpose.""" import yaml C = yaml.safe_load(open("chain.yaml", encoding="utf-8")) def ledger(shops=3, coverage=0.60): s = C["shop"]; e = C["economics"]; c = C["costs"] calls_month = shops * s["calls_per_shop_per_day"] * 30 minutes = calls_month * s["avg_minutes_per_call"] # --- costs ------------------------------------------------------------- platform = minutes * c["platform_per_minute"] ops = c["ops_allocation_per_month"] monthly_cost = platform + ops # --- benefits ---------------------------------------------------------- recovered = (calls_month * s["baseline_blocking"] * coverage * e["booking_conversion"] * e["avg_ticket"] * e["repeat_multiplier"] * 0.5) desk_hours = calls_month * coverage * 0.72 * s["avg_handle_sec"] / 3600 desk_saved = desk_hours * e["front_desk_hourly_cost"] dormant = C["members"]["total"] * C["members"]["dormant_share"] wake_gmv = dormant * C["members"]["reactivation_rate"] * coverage * e["avg_ticket"] # --- the honest negative ---------------------------------------------- # Deflection grows super-linearly: the more you automate, the more of what # is left needs a person, and the worse it goes when it does not get one. deflected = (calls_month * coverage * coverage ** 2.2 * 0.055 * e["avg_ticket"] * 0.9) benefit = recovered + desk_saved + wake_gmv - deflected net = benefit - monthly_cost payback = c["one_off_integration"] / net if net > 0 else float("inf") return { "calls_month": round(calls_month), "monthly_cost": round(monthly_cost), "recovered": round(recovered), "desk_saved": round(desk_saved), "wake_gmv": round(wake_gmv), "deflected": round(deflected), "net": round(net), "payback_months": round(payback, 1), "three_year": round(net * 36 - c["one_off_integration"]), } for cov in (0.3, 0.6, 0.9): r = ledger(coverage=cov) print(f"coverage {cov:.0%}: net CNY {r['net']:>8,} payback {r['payback_months']:>4} mo " f"(deflection cost CNY {r['deflected']:,})") # Notice what happens at 90%: automating more stops paying.`, }, { lang: "配置 / config", k: "yaml", file: "chain.yaml", src: `# A real three-shop tuina chain. Replace every number with your own. shop: count: 3 calls_per_shop_per_day: 70 avg_minutes_per_call: 2.6 avg_handle_sec: 150 baseline_blocking: 0.28 # share of peak calls nobody answered before economics: avg_ticket: 268 booking_conversion: 0.55 repeat_multiplier: 1.8 front_desk_hourly_cost: 32 # fully loaded members: total: 6800 dormant_share: 0.35 # no visit in three months reactivation_rate: 0.055 # of the dormant pool, per month, when contacted well costs: one_off_integration: 38000 # build, tune, knowledge base, go-live platform_per_minute: 0.19 # ASR + TTS + LLM + telephony, all in ops_allocation_per_month: 2600 # part of somebody's time, honestly counted # What the model deliberately does NOT assume: # - that the AI answers everything (coverage is a dial, and 90% is not optimal) # - that saved front-desk hours are free money (they must be redeployed) # - that deflected customers cost nothing (they cost the most, and silently)`, }, { lang: "对接 / SQL", k: "sql", file: "chain_actuals.sql", src: `-- Stop guessing: pull every parameter from the systems you already have. -- 1. calls, blocking and handle time, from the carrier CDR SELECT shop_id, COUNT(*) / 30.0 AS calls_per_day, ROUND(AVG(duration_sec)) AS avg_handle_sec, ROUND(SUM(disposition IN ('NO ANSWER','BUSY')) / COUNT(*), 3) AS blocking FROM cdr WHERE direction = 'inbound' AND start_time >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY shop_id; -- 2. average ticket and repeat behaviour, from the POS SELECT ROUND(AVG(amount)) AS avg_ticket, ROUND(COUNT(*) / COUNT(DISTINCT member_id), 2) AS visits_per_member_90d FROM sale WHERE sold_at >= DATE_SUB(CURDATE(), INTERVAL 90 DAY); -- 3. the dormant pool, which is where the win-back money is SELECT COUNT(*) AS dormant_members FROM members m WHERE m.opted_out = 0 AND NOT EXISTS (SELECT 1 FROM visit v WHERE v.member_id = m.id AND v.visit_date >= DATE_SUB(CURDATE(), INTERVAL 90 DAY)); -- 4. after go-live: did blocking actually fall, and did bookings actually rise? SELECT CASE WHEN start_time < '2026-06-01' THEN 'before' ELSE 'after' END AS period, ROUND(SUM(disposition IN ('NO ANSWER','BUSY')) / COUNT(*), 3) AS blocking, COUNT(*) AS calls, (SELECT COUNT(*) FROM booking b WHERE b.created_at BETWEEN MIN(c.start_time) AND MAX(c.start_time)) AS bookings FROM cdr c WHERE direction = 'inbound' AND start_time >= '2026-03-01' GROUP BY period;`, }, ], }; /* ============ CS2 · v29 — the call trace ============ */ CODE.v29 = { note: { zh: "能复盘的前提是留痕。Python 定义了一条通话的结构化 trace:每一跳记录耗时、文本、置信度、检索命中和工具调用,并支持把整通电话重放一遍——这是修 bug 唯一靠谱的方式;配置是 trace 的字段定义(注意手机号存哈希);SQL 用来找出「哪一类通话最容易崩」,以及那条最有用的查询:失败前的最后一跳是什么。", en: "You cannot review what you did not record. The Python defines a structured trace for one call: every hop keeps its timing, text, confidence, retrieval hits and tool calls, and the whole call can be replayed — the only reliable way to fix a bug. The configuration is the trace schema, with the phone number stored as a hash. The SQL finds which kinds of call break most, plus the single most useful query: what the last hop before a failure was.", }, tabs: [ { lang: "Python", k: "py", file: "trace.py", src: `"""One call, fully traced — and replayable, which is the point.""" import json, time, hashlib from dataclasses import dataclass, field, asdict @dataclass class Hop: idx: int who: str # caller | agent | system started_ms: float duration_ms: float user_text: str = "" agent_text: str = "" asr_confidence: float = None retrieved: list = field(default_factory=list) tool_call: dict = None guardrail: str = None note: str = "" class CallTrace: def __init__(self, call_id, phone): self.call_id = call_id self.phone_hash = hashlib.sha256(phone.encode()).hexdigest()[:32] self.t0 = time.perf_counter() self.hops = [] def hop(self, **kw): h = Hop(idx=len(self.hops) + 1, started_ms=(time.perf_counter() - self.t0) * 1000, **kw) self.hops.append(h) return h def save(self, path="traces/"): payload = {"call_id": self.call_id, "phone_hash": self.phone_hash, "total_ms": sum(h.duration_ms for h in self.hops), "hops": [asdict(h) for h in self.hops]} open(f"{path}{self.call_id}.json", "w", encoding="utf-8").write( json.dumps(payload, ensure_ascii=False, indent=2)) def replay(call_id, inject=None): """Re-run a real call against the current config. inject: 'no_hotwords', 'short_endpoint', 'no_kb' — the three faults from the chapter.""" trace = json.load(open(f"traces/{call_id}.json", encoding="utf-8")) cfg = load_config() if inject == "no_hotwords": cfg["asr"]["hotwords_file"] = None elif inject == "short_endpoint": cfg["endpointing"]["tail_silence_ms"] = 300 elif inject == "no_kb": cfg["knowledge_base"]["min_score"] = 1.1 # nothing will ever match out = [] for h in trace["hops"]: if h["who"] == "caller": out.append(run_pipeline(h["user_text"], cfg)) return out print(replay("c_8f21a0", inject="no_hotwords"))`, }, { lang: "配置 / schema", k: "json", file: "trace_schema.json", src: `{ "call_id": "c_8f21a0", "phone_hash": "9f2c... (never the raw number)", "started_at": "2026-09-12T20:03:11+08:00", "channel": "phone", "total_ms": 137000, "outcome": "booked", "hops": [ { "idx": 5, "who": "agent", "started_ms": 6960, "duration_ms": 1120, "user_text": "16:30 吧,还是王师傅", "asr_confidence": 0.58, "asr_alternatives": ["黄师傅", "王师傅", "汪师傅"], "retrieved": [], "tool_call": { "name": "query_therapist", "args": { "name": "黄师傅" }, "ok": false, "error": "not found" }, "guardrail": null, "note": "hotword list did not contain this name — the call nearly failed here" }, { "idx": 8, "who": "agent", "started_ms": 14280, "duration_ms": 1460, "user_text": "团购券能和会员卡一起用吗", "retrieved": [{ "source": "promotions.md", "score": 0.41 }], "agent_text": "这个规则我帮您问一下前台,先把时间定下来好吗?", "guardrail": "ungrounded_fallback", "note": "retrieval below threshold — fallback fired instead of inventing" } ], "retention": { "audio_days": 90, "trace_days": 180, "anonymise_after_days": 180 } }`, }, { lang: "对接 / SQL", k: "sql", file: "trace_queries.sql", src: `-- Which kinds of call break, and where exactly. SELECT intent, COUNT(*) AS calls, ROUND(AVG(outcome = 'booked'), 3) AS booked_rate, ROUND(AVG(outcome = 'handed_off'), 3) AS handoff_rate, ROUND(AVG(outcome = 'abandoned'), 3) AS abandoned_rate, ROUND(AVG(total_ms) / 1000) AS avg_sec, ROUND(AVG(hop_count), 1) AS avg_hops FROM call_trace WHERE started_at >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY intent ORDER BY abandoned_rate DESC; -- The most useful query in this file: what was the LAST hop before a failure? SELECT h.note, h.guardrail, COUNT(*) AS times, ROUND(AVG(h.asr_confidence), 2) AS avg_conf FROM call_hop h JOIN call_trace t ON t.call_id = h.call_id WHERE t.outcome IN ('abandoned', 'handed_off') AND h.idx = (SELECT MAX(idx) FROM call_hop x WHERE x.call_id = h.call_id) - 1 AND t.started_at >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY h.note, h.guardrail ORDER BY times DESC LIMIT 20; -- Low-confidence hops that a hotword would have fixed. SELECT JSON_EXTRACT(asr_alternatives, '$[0]') AS heard_as, user_text, COUNT(*) AS times FROM call_hop WHERE asr_confidence < 0.65 AND created_at >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY heard_as, user_text HAVING times >= 3 ORDER BY times DESC;`, }, ], }; /* ============ CH4 · v30 — Twilio ============ */ CODE.v30 = { note: { zh: "三个视角对应三档接入。Python 是 Media Streams 那一档的最小骨架:一个返回 TwiML 的 webhook,加一个收发音频帧的 WebSocket——注意帧是 base64 的 8 kHz μ-law,每 20 毫秒 160 字节,回送时用同样的格式塞进 media 消息。TwiML 那一栏把三档的 XML 并排放在一起,一眼就能看出谁负责什么。最后一栏是把号码买下来、把 webhook 指过去,并且把第二十二章那四道外呼闸门接在拨号之前——Twilio 让拨号变得太容易了,闸门必须在代码里,不能在制度里。", en: "Three angles for three tiers. The Python is the minimal skeleton of the Media Streams tier: a webhook returning TwiML plus a WebSocket exchanging audio frames — note the frames are base64 8 kHz mu-law, 160 bytes every 20 ms, and you send audio back in exactly the same shape inside a media message. The TwiML tab puts all three tiers' XML side by side so the division of responsibility is visible at a glance. The last tab buys the number, points the webhook at your server, and puts chapter CH1's four outbound gates in front of the dialler — Twilio makes dialling far too easy, so the gates belong in code rather than in a policy.", }, tabs: [ { lang: "Python", k: "py", file: "twilio_agent.py", run: "# uvicorn twilio_agent:app --port 8080 (公网可达,或用 ngrok)", src: `"""Media Streams: Twilio carries the audio, everything else is yours.""" import base64, json, audioop from fastapi import FastAPI, WebSocket, Request from fastapi.responses import Response app = FastAPI() TWIML = """ 您好,这里是中山路店的智能助手,本次通话可能被录音。 """ @app.post("/voice") async def voice(request: Request): """Twilio hits this on every inbound call and performs what you return.""" return Response(content=TWIML, media_type="application/xml") @app.websocket("/media") async def media(ws: WebSocket): await ws.accept() stream_sid = None asr = StreamingASR() # the client from chapter AS2 async for raw in ws.iter_text(): msg = json.loads(raw) if msg["event"] == "start": stream_sid = msg["start"]["streamSid"] elif msg["event"] == "media": # 8 kHz mu-law, 160 bytes per 20 ms frame, base64 ulaw = base64.b64decode(msg["media"]["payload"]) pcm8 = audioop.ulaw2lin(ulaw, 2) pcm16k, _ = audioop.ratecv(pcm8, 2, 1, 8000, 16000, None) await asr.push(pcm16k) # chapter AS1: upsample, never feed 8 k elif msg["event"] == "stop": break async def speak(ws, stream_sid, pcm16k: bytes): """Send audio back in the same shape. Chunk it — chapter TS4 applies here too.""" pcm8, _ = audioop.ratecv(pcm16k, 2, 1, 16000, 8000, None) ulaw = audioop.lin2ulaw(pcm8, 2) for i in range(0, len(ulaw), 160): # one 20 ms frame at a time await ws.send_text(json.dumps({ "event": "media", "streamSid": stream_sid, "media": {"payload": base64.b64encode(ulaw[i:i + 160]).decode()}, })) async def barge_in(ws, stream_sid): """The caller cut in: drop what Twilio has buffered. Chapter RT2's other half.""" await ws.send_text(json.dumps({"event": "clear", "streamSid": stream_sid}))`, }, { lang: "TwiML / 配置", k: "xml", file: "tiers.xml", src: ` 请问您想预约哪个项目? 没有听到,帮您转前台。 +8613800135768 本次通话可能被录音。 `, }, { lang: "对接 / shell", k: "sh", file: "provision.sh", run: "# 号码有监管绑定:很多国家要先提交地址与证件才能买", src: `#!/usr/bin/env bash set -euo pipefail # 1. Pick an edge near your callers. From Asia a US edge costs about 250 ms # each way and eats the entire budget you saved in chapter RT1. export TWILIO_EDGE=singapore # 2. Find a number. Many countries need a regulatory bundle (address and ID) # before a local number can be issued — check before promising a date. twilio api:core:available-phone-numbers:local:list \ --country-code SG --voice-enabled --limit 5 # 3. Buy it and point voice at your webhook. twilio api:core:incoming-phone-numbers:create \ --phone-number "+6531234567" \ --voice-url "https://your.host/voice" \ --voice-method POST \ --status-callback "https://your.host/status" # 4. Local development: a public URL Twilio can reach. ngrok http 8080 # 5. Outbound. Twilio puts dialling one API call away, which is exactly why # chapter CH1's four gates must run BEFORE this line, in code. python - <<'PY' from outbound_guard import may_call, OutboundRefused from twilio.rest import Client import os client = Client(os.environ["TWILIO_SID"], os.environ["TWILIO_TOKEN"]) for phone in load_due_members(): try: may_call(phone) # consent, refusal, hours, frequency except OutboundRefused as e: print("skip", phone, e) continue client.calls.create(to=phone, from_="+6531234567", url="https://your.host/outbound") PY`, }, ], }; /* ============ BR5 · v31 — 三层应答与多店配置 ============ */ CODE.v31 = { note: { zh: "Python 是那条应答链路本身:三层按顺序降级,每一层要么返回答案,要么把问题交给下一层——注意 RAG 那一层的返回条件是「分数达标」而不是「检索到了」,这一个判断就是幻觉率的开关。配置那一栏是三层继承的合并规则,门店层只写真的不同的几行;注意 locked 列表——护栏与合规告知在全局层被锁死,门店层写了也不生效。最后一栏是闭环:把兜底层记下来的缺口按频次聚合,高频的那几条就是下周该写进 FAQ 的内容,不用猜。", en: "The Python is the answering path itself: three tiers degrading in order, each either returning an answer or passing the question down — and note that the retrieval tier's return condition is score clears the threshold rather than something was retrieved, because that single test is the hallucination switch. The configuration tab is the three-layer merge, where the shop layer carries only what genuinely differs; note the locked list, which pins guardrails and the compliance notice in the global layer so a shop layer that sets them has no effect. The last tab is the loop: aggregate the gaps the fallback tier recorded, and the frequent ones are next week's FAQ entries, no guessing required.", }, tabs: [ { lang: "Python", k: "py", file: "answer_router.py", run: "# 三层降级:能精确命中的不走模型,走模型的必须先有证据", src: `"""One question, three tiers, in order. Each tier answers or passes it down.""" from dataclasses import dataclass FAQ_MIN = 0.86 # exact tier: high bar, because its answers are human-written RAG_MIN = 0.62 # retrieval tier: below this we do NOT call the model at all @dataclass class Answer: text: str tier: str # faq | rag | fallback attributable: bool # human-written, or citing retrieved evidence handoff: bool = False def answer(question: str, store_id: str) -> Answer: cfg = assemble_config(store_id) # see the configuration tab # --- tier 1 · exact ----------------------------------------------- # Curated question to answer, with synonyms merged. 3 ms, no tokens, # and every line is attributable to the person who wrote it. hit = faq_lookup(question, namespace=cfg.namespaces) if hit and hit.score >= FAQ_MIN: return Answer(hit.answer, tier="faq", attributable=True) # --- tier 2 · evidenced ------------------------------------------- docs = retrieve(question, namespace=cfg.namespaces, top_k=cfg.top_k) # The switch. Retrieval ALWAYS returns top_k rows; relevance is the # question. Without this test the model answers from unrelated chunks. if docs and max(d.score for d in docs) >= RAG_MIN: text = generate(question, evidence=docs, system=cfg.prompt) return Answer(text, tier="rag", attributable=True) # --- tier 3 · fallback -------------------------------------------- # Say so, hand off, and write the gap down. The record is what makes # tier 1 grow next week. record_gap(question, store_id) return Answer(cfg.fallback_line, tier="fallback", attributable=True, handoff=True) # Live state never comes from any tier — it is not knowledge. LIVE_ONLY = {"price", "availability", "rota", "member_balance"} def route(question: str, store_id: str) -> Answer: intent = classify(question) if intent in LIVE_ONLY: return tool_call(intent, store_id) # always a live lookup return answer(question, store_id)`, }, { lang: "配置 / config", k: "yaml", file: "layers.yaml", src: `# Three layers, merged shop-last. Adding a shop is adding a block here. global: engines: { asr: paraformer-streaming, tts: cosyvoice2 } prompt_base: prompts/system.txt guardrails: [no_efficacy_claims, symptom_template, out_of_bounds_refusal] compliance: { recording_notice: notice_v3, synthetic_disclosure: true } faq: faq/common.yaml # 问候 / 营业时间 / 停车 — every shop top_k: 4 fallback_line: "这个我帮您问一下前台,先把时间定下来好吗?" # Pinned here. A brand or shop layer that sets these has no effect — # this class of thing is not negotiable, see chapter OP2. locked: [guardrails, compliance] brand: acme_tuina: voice: brand_female_warm faq: faq/brand.yaml # 连锁统一的优惠与会员规则 price_list: kb/brand_prices.md shops: zhongshan: inherits: acme_tuina number: "+6531234567" # 被叫号码 → 这家店,见 CH4 faq: faq/shop_zhongshan.yaml # 只写这家店真的不同的那几条 namespaces: [global, acme_tuina, zhongshan] knowledge: kb/zhongshan/ # 本店价目 · 技师 · 地址 · 营业时间 tiers: [faq, rag, fallback] writes: [booking, reschedule] handoff_to: agent_group_a greeting: "您好,这里是中山路店,本次通话可能被录音。" new_branch: # 试运营:先只读,FAQ only inherits: acme_tuina number: "+6531234568" namespaces: [global, acme_tuina] tiers: [faq, fallback] writes: [] handoff_to: supervisor greeting: "您好,这里是新店,本次通话可能被录音。" # Telephony resolves the shop from the dialled number the webhook carries. # Other channels resolve it differently: a QR parameter, a mini-program # entry, a platform shop id — same identity, three sources. number_map: "+6531234567": zhongshan "+6531234568": new_branch # One number per shop, or one main number plus IVR routing? Per-shop costs # the monthly rental times the shop count and puts the shop's own number on # its sign; a main number is cheaper and adds a keypress, and the number # customers save is the switchboard rather than their shop.`, }, { lang: "闭环 / SQL", k: "sql", file: "gap_loop.sql", run: "-- 兜底层记下来的缺口,就是下周 FAQ 的内容 —— 不用猜", src: `-- Which questions does this system repeatedly fail to answer? -- Anything frequent here is an FAQ entry waiting to be written. SELECT normalised_question, COUNT(*) AS asked, COUNT(DISTINCT store_id) AS shops_affected, MIN(asked_at) AS first_seen, CASE WHEN COUNT(DISTINCT store_id) > 1 THEN 'brand faq' ELSE 'shop faq' END AS write_it_into FROM answer_gap WHERE asked_at >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) GROUP BY normalised_question HAVING asked >= 5 ORDER BY asked DESC LIMIT 40; -- The health of the split, by tier. Watch the last column: answers that -- are neither human-written nor evidence-citing are where risk lives. SELECT store_id, ROUND(AVG(tier = 'faq'), 3) AS faq_share, ROUND(AVG(tier = 'rag'), 3) AS rag_share, ROUND(AVG(tier = 'fallback'), 3) AS fallback_share, ROUND(AVG(attributable = 0), 4) AS unattributable FROM answer_log WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY) GROUP BY store_id ORDER BY unattributable DESC; -- Did a shop layer try to override something pinned in global? -- This should always return zero rows; if it does not, the merge is wrong. SELECT store_id, config_key FROM config_override WHERE config_key IN ('guardrails', 'compliance') AND layer <> 'global';`, }, ], };