简介这是一套面向电话营销与智能客服场景的AI语音通话销售机器人源码适合具备一定后端与语音开发基础的开发者、创业团队及企业技术选型人员用于搭建可自动外呼、应答与跟进客户的对话系统。源码整合了自然语言处理、语音识别、语音合成与机器学习算法能够理解客户意图与情绪并动态优化对话策略同时支持按业务需求定制话术、对接CRM等内部系统。压缩包共2002个文件约103.6MB以795个js脚本、203个html页面、216个css样式、219个xml配置及51个php文件构成前后端主体另含66个so动态库、49个h头文件与6个sql脚本并附带FreeSWITCH相关组件与语音资源便于理解通话链路与部署结构。资源还包含系统搭建教程覆盖环境配置、源码编译、数据库连接与API接口设置等环节可帮助读者快速跑通系统并完成二次开发。目前已有1239人学习下载适合作为语音外呼项目的实践参考。1. 一套电话销售机器人源码到底在替你干哪几件事很多人第一次听到「AI智能电话语音通话销售机器人源码」脑子里浮现的是一个能跟客户对答如流的数字人。实际拆开看它更像一条流水线把客户号码从数据库里捞出来自动拨出去接通后播放或生成话术识别客户说了什么判断意图再把结果写回数据库。整条链路里真正难的不是「让机器说话」而是「让机器听懂并做出正确分支」。我接触过的几个模拟项目里最常见的诉求是手里有一批历史客户线索人工一天打两百通就到顶了想用机器人把初筛做完人工只跟进有意向的那部分。这个方向适合有稳定线索来源、话术相对标准化的团队比如课程咨询、活动邀约、老客户回访。不适合需要深度定制方案、客户问题高度发散的场景那种情况机器人反而会消耗客户耐心。这套源码通常包含几个模块号码管理、任务调度、语音网关对接、ASR语音识别、NLU意图理解、TTS语音合成、对话状态机、结果回写。搭建教程要解决的核心问题就是把这些模块串起来让一个最小可用的外呼任务跑通。下面按我实际搭过的顺序从环境准备讲到参数调优和排错。2. 从零搭一套外呼系统环境、依赖与最小可跑通链路2.1 先想清楚选云服务还是本地部署这是第一个分叉口选错了后面返工成本很高。云服务方案把 ASR、TTS、语音网关都托管出去你只需要写业务逻辑优点是上线快、并发扩容简单缺点是按量计费通话量大时成本线性上涨而且话术和客户数据要经过第三方。本地部署方案需要自己准备 GPU 服务器跑 ASR 和 TTS 模型语音网关也要自己对接前期投入大但数据不出内网长期单价低。我一般会建议日呼量在 500 通以下、话术还在频繁调整的阶段先用云服务把流程跑通验证转化率等话术稳定、日呼量上到几千通再考虑把 ASR 和 TTS 迁到本地。迁移时业务层代码基本不用动只要换掉适配层。对比项云服务方案本地部署方案上线周期1-3 天2-4 周前期成本低高GPU 服务器单通成本随量线性增长固定成本摊薄后低数据可控性依赖第三方完全内网适合阶段验证期稳定放量期2.2 用 Docker Compose 把基础服务拉起来不管选哪种方案业务层建议用容器跑方便迁移。下面是一个最小化的 compose 文件包含业务服务、Redis存对话状态和 PostgreSQL存号码和结果。ASR/TTS 如果走云服务这里不需要额外容器如果本地部署再加两个模型服务容器。# docker-compose.yml version: 3.8 services: app: build: . ports: - 8080:8080 environment: - REDIS_HOSTredis - DB_HOSTpostgres - ASR_PROVIDERcloud # 本地部署时改成 local - TTS_PROVIDERcloud depends_on: - redis - postgres redis: image: redis:7-alpine ports: - 6379:6379 postgres: image: postgres:15-alpine environment: POSTGRES_DB: callbot POSTGRES_USER: bot POSTGRES_PASSWORD: botpass volumes: - pgdata:/var/lib/postgresql/data volumes: pgdata:逻辑说明app 服务是业务主进程负责调度外呼任务和跑对话状态机。Redis 存每一通电话的实时对话上下文因为对话状态读写频繁且允许丢失通话结束后落库用 Redis 比直接写数据库快得多。PostgreSQL 存号码列表、通话记录和意图标签这些数据需要持久化和查询。ASR_PROVIDER和TTS_PROVIDER两个环境变量是适配层的开关业务代码里根据这两个值决定调用云端 API 还是本地模型服务。参数说明Redis 用 alpine 镜像体积小生产环境建议加密码和持久化配置。PostgreSQL 的密码不要用示例里的换成强密码并通过环境变量注入。depends_on只保证启动顺序不保证服务就绪业务代码里要有重试逻辑。2.3 号码导入与任务表设计号码管理看起来简单实际最容易出问题。我见过把号码存在 Excel 里每次手动导入的也见过用一张大表所有任务混在一起的。推荐按「任务」维度隔离每个外呼任务一张逻辑表或者在同一张表里用 task_id 区分。-- 号码表 CREATE TABLE leads ( id BIGSERIAL PRIMARY KEY, task_id BIGINT NOT NULL, phone VARCHAR(20) NOT NULL, name VARCHAR(50), extra JSONB, -- 自定义字段比如客户等级、上次沟通记录 status VARCHAR(20) DEFAULT pending, -- pending/calling/done/failed attempt_count INT DEFAULT 0, last_call_at TIMESTAMP, created_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_leads_task_status ON leads(task_id, status); -- 通话结果表 CREATE TABLE call_results ( id BIGSERIAL PRIMARY KEY, lead_id BIGINT REFERENCES leads(id), call_id VARCHAR(64), duration INT, -- 通话秒数 intent VARCHAR(50), -- 识别出的意图标签 transcript TEXT, -- 完整对话文本 recording_url TEXT, created_at TIMESTAMP DEFAULT NOW() );逻辑说明leads 表用 status 字段控制调度调度器每次取statuspending且attempt_count 3的记录。extra 用 JSONB 存非结构化字段避免频繁改表结构。call_results 单独一张表因为一通电话可能对应多条结果记录比如中途转人工后又回写也方便后续做转化分析。参数说明attempt_count限制重拨次数一般设 2-3 次太多会被标记骚扰。idx_leads_task_status这个联合索引很关键调度器高频查询这个条件没有索引时数据量上万就会明显变慢。duration字段用来过滤无效通话小于 5 秒的基本是秒挂分析时要排除。2.4 对话状态机的核心循环这是整个系统的心脏。一通电话接通后状态机按预设流程走问候 → 确认身份 → 说明来意 → 处理异议 → 引导下一步。每一步都要等 ASR 返回文本再走 NLU 判断意图然后决定下一句说什么。# dialogue_engine.py import redis import json r redis.Redis(hostredis, port6379, decode_responsesTrue) # 简化的状态转移表 TRANSITIONS { greeting: {confirmed: pitch, denied: end, unknown: greeting_retry}, greeting_retry: {confirmed: pitch, denied: end, unknown: end}, pitch: {interested: qualify, not_interested: end, question: faq}, qualify: {qualified: transfer, unqualified: end}, } def handle_turn(call_id, user_text, current_state): # 1. 调用 NLU 获取意图 intent nlu_predict(user_text) # 返回 confirmed/denied/interested 等 # 2. 查状态转移表 next_state TRANSITIONS.get(current_state, {}).get(intent, end) # 3. 更新 Redis 中的对话上下文 ctx_key fcall:{call_id} r.hset(ctx_key, mapping{state: next_state, last_intent: intent}) r.expire(ctx_key, 3600) # 1小时过期防止僵尸会话 # 4. 根据下一状态生成回复文本 reply generate_reply(next_state, user_text) return next_state, reply def nlu_predict(text): # 实际项目中这里调用 NLU 服务或规则引擎 # 简化示例关键词匹配 if any(w in text for w in [是的, 对, 是我]): return confirmed if any(w in text for w in [不需要, 别打, 挂]): return denied return unknown逻辑说明handle_turn是每一轮对话的入口接收客户说的话和当前状态返回下一状态和机器人要说的文本。状态转移表用字典硬编码方便快速调整话术流程不需要改代码逻辑。Redis 里存的是当前通话的临时状态用expire设置过期时间防止异常情况下会话永远不释放。参数说明TRANSITIONS里的unknown分支要谨慎设计连续两次 unknown 就应该结束通话否则机器人会陷入循环。expire设 3600 秒是保守值实际通话很少超过 10 分钟但留足余量避免误删。nlu_predict在真实项目里应该是一个独立服务这里用关键词匹配只是为了说明调用位置生产环境要用训练好的分类模型。3. 语音链路调优ASR、TTS 与打断处理的参数怎么设3.1 ASR 识别率上不去的三个常见原因ASR 是整条链路里最影响体验的环节。识别错了后面全错。我踩过的坑集中在三处音频采样率不匹配、静音检测阈值太敏感、方言和口音没有针对性优化。采样率问题最隐蔽。电话线路通常是 8kHz 采样而很多 ASR 模型默认按 16kHz 训练。如果直接把 8kHz 音频喂进去识别率会断崖式下降。解决办法是在送入 ASR 之前做重采样或者选择支持 8kHz 的模型。import librosa import soundfile as sf def resample_audio(input_path, output_path, target_sr16000): # 电话录音通常是 8kHzASR 模型可能需要 16kHz y, sr librosa.load(input_path, srNone) if sr ! target_sr: y librosa.resample(y, orig_srsr, target_srtarget_sr) sf.write(output_path, y, target_sr) return output_path逻辑说明librosa.load的srNone表示保持原始采样率读取不做自动重采样。然后判断原始采样率是否等于目标采样率不等才做重采样。最后用 soundfile 写回文件。这一步在实时通话里不能落盘再读要用流式处理但离线测试时这样写最直观。参数说明target_sr根据你用的 ASR 模型文档来定常见是 16000。重采样会引入轻微失真但比采样率不匹配导致的识别错误小得多。如果 ASR 服务支持直接传 8kHz优先用原生支持省掉这一步。静音检测阈值是第二个坑。设得太高客户说话小声时被当成静音机器人抢话设得太低环境噪音被当成语音机器人一直等。我一般用能量加过零率的双门限法能量阈值根据线路底噪动态调整。import numpy as np def is_speech(frame, energy_threshold500, zcr_threshold0.1): # frame 是一帧音频数据长度通常 20-30ms energy np.sum(frame ** 2) / len(frame) zcr np.sum(np.abs(np.diff(np.sign(frame)))) / (2 * len(frame)) # 能量和过零率都超过阈值才认为是语音 return energy energy_threshold and zcr zcr_threshold逻辑说明能量判断声音大小过零率判断频率特性。语音信号能量高且过零率适中噪音通常过零率很高。两个条件同时满足才判定为语音减少误触发。参数说明energy_threshold需要根据实际线路底噪校准建议在正式外呼前录一段静音环境下的线路音频算平均能量然后设为其 3-5 倍。zcr_threshold一般设 0.1-0.2太高会漏掉清辅音太低会把摩擦音当成噪音。3.2 TTS 语音合成的自然度与延迟平衡TTS 有两个指标要同时看自然度和首字延迟。自然度高的模型往往参数多、推理慢首字延迟超过 800ms 客户就会觉得卡顿。我一般会准备两套 TTS一套轻量模型用于实时对话一套高质量模型用于生成固定话术的缓存音频。固定话术比如问候语、公司介绍、常见问题回答这些内容不变可以提前用高质量 TTS 生成好音频文件通话时直接播放零延迟。只有需要动态拼接的内容比如客户姓名、具体金额才走实时 TTS。import hashlib import os CACHE_DIR /app/tts_cache def get_tts_audio(text, voicefemale_01): # 用文本和音色生成缓存 key key hashlib.md5(f{voice}:{text}.encode()).hexdigest() cache_path os.path.join(CACHE_DIR, f{key}.wav) if os.path.exists(cache_path): return cache_path # 命中缓存直接返回 # 未命中调用 TTS 服务生成 audio tts_service.synthesize(text, voicevoice) with open(cache_path, wb) as f: f.write(audio) return cache_path逻辑说明用文本加音色做 MD5 作为缓存文件名保证相同内容只生成一次。缓存目录挂载到宿主机或对象存储容器重启不丢失。命中缓存时直接返回文件路径播放器读取本地文件比等 TTS 服务返回快得多。参数说明缓存 key 里要包含音色参数否则换音色后会读到旧音频。缓存目录要定期清理或者设置 LRU 淘汰策略避免磁盘占满。对于包含变量的话术比如「张先生您好」每个姓氏都会生成一个缓存文件量不大可以接受如果变量组合爆炸就要考虑模板加拼接的方式。3.3 打断处理让机器人学会闭嘴客户在机器人说话时插话系统要能检测到并停止播放否则客户会觉得机器人「聋」。实现方式是双工音频流播放 TTS 的同时持续采集麦克风数据检测到语音活动就发中断信号。import threading class DialogSession: def __init__(self): self.interrupted threading.Event() self.playing False def play_tts(self, audio_path): self.playing True self.interrupted.clear() # 模拟播放实际用音频库播放 for chunk in read_audio_chunks(audio_path): if self.interrupted.is_set(): break # 收到打断信号停止播放 play_chunk(chunk) self.playing False def on_speech_detected(self): if self.playing: self.interrupted.set() # 通知播放线程停止逻辑说明interrupted是一个线程安全的事件标志。播放线程每播放一个音频块就检查一次如果被设置就跳出循环。语音检测线程在采集到有效语音时调用on_speech_detected设置标志。这样播放和检测互不阻塞。参数说明打断检测的灵敏度要调太灵敏会被环境噪音误触发太迟钝客户说了一半机器人才停。建议在语音检测后加一个 200-300ms 的确认窗口连续多帧检测到语音才判定为打断。另外打断后要清空播放缓冲区否则残留音频会继续播出来。4. 避坑与排查外呼系统上线后最容易翻车的五个地方4.1 号码被标记导致接通率骤降现象系统刚上线时接通率还有 30%跑了一周后降到 10% 以下客户反馈显示号码被标记为「骚扰电话」。原因外呼频率过高、被叫方挂断后立即重拨、同一号码短时间内多次呼叫都会触发运营商的标记机制。另外号码没有做实名认证和资质报备也容易被拦截。解决控制单号码日呼量一般建议不超过 50 通两次呼叫间隔至少 30 分钟挂断后不要立即重拨等至少 1 小时。号码要完成企业实名认证外呼内容要能说明合法用途。如果已经被标记换号是最快的办法同时调整呼叫策略避免新号也被标记。4.2 对话状态在并发时串号现象A 客户的通话里突然出现 B 客户的信息或者机器人回复了上一通电话的内容。原因对话上下文用全局变量或单例存储多个通话共享了同一份状态。或者 Redis key 设计有误不同通话用了相同的 key。解决每通电话生成唯一 call_id所有状态读写都带这个 call_id 作为 key 的一部分。检查代码里有没有用类变量或模块级变量存对话状态。Redis 的 key 命名规范建议用call:{call_id}:state这种格式避免冲突。# 错误做法用全局字典 global_context {} # 所有通话共享必然串号 # 正确做法每个 call_id 独立 key def get_context(call_id): return r.hgetall(fcall:{call_id}:state)4.3 ASR 返回空结果导致对话卡死现象客户说了话但机器人没反应几秒后直接挂断。原因ASR 服务超时或返回空字符串状态机没有处理这种情况一直等输入。或者音频流在传输过程中丢失ASR 收到的是静音。解决给 ASR 调用设超时超时后走「没听清请您再说一遍」的兜底话术。连续两次空结果就转人工或结束通话。音频流要加序号和校验发现丢包时请求重传或直接跳过。def safe_asr(audio_stream, timeout3): try: result asr_service.recognize(audio_stream, timeouttimeout) if not result or len(result.strip()) 0: return None # 空结果上层走兜底 return result except TimeoutError: return None4.4 TTS 音频格式与播放器不兼容现象TTS 返回的音频播放出来是杂音或者播放速度不对。原因TTS 服务输出的采样率、位深、声道数与播放器预期不一致。比如 TTS 输出 24kHz 单声道播放器按 8kHz 双声道解析出来的就是噪音。解决在 TTS 适配层统一转成播放器要求的格式。用 ffmpeg 做格式转换是最稳的虽然多一步但兼容性好。# 把 TTS 输出统一转成 8kHz 16bit 单声道 ffmpeg -i tts_output.wav -ar 8000 -ac 1 -sample_fmt s16 play_ready.wav参数说明-ar 8000指定采样率 8kHz-ac 1单声道-sample_fmt s1616 位有符号整数。这三个参数要和你的语音网关文档对齐不同网关要求可能不同。4.5 数据库连接池耗尽导致新通话无法建立现象系统跑了一段时间后新外呼任务创建失败日志里出现连接超时错误。原因每次通话都创建新的数据库连接通话结束后没有正确释放。并发量上来后连接数超过数据库上限。解决用连接池管理数据库连接设置合理的最大连接数和空闲回收时间。确保每个通话结束后连接归还到池里而不是直接关闭。from sqlalchemy import create_engine from sqlalchemy.pool import QueuePool engine create_engine( postgresql://bot:botpasspostgres/callbot, poolclassQueuePool, pool_size10, # 常驻连接数 max_overflow20, # 峰值可额外创建的连接数 pool_recycle1800, # 30分钟回收空闲连接 pool_pre_pingTrue # 使用前检查连接是否有效 )参数说明pool_size根据并发通话数设置一般设为预期峰值的 1.5 倍。max_overflow是峰值时允许临时超出的数量不要设太大否则数据库扛不住。pool_recycle要小于数据库的wait_timeout避免拿到已被服务端关闭的连接。pool_pre_ping会多一次网络往返但能避免大部分「连接已断开」的错误。5. 让机器人越用越准意图标签回流与话术迭代的一个具体做法系统跑通只是开始真正决定转化率的是话术和意图识别的准确度。我一般会在通话结果表里加一个字段记录每一轮对话的 ASR 原文和 NLU 判定结果然后每周导出一次人工标注错误样本回流到 NLU 模型做增量训练。具体做法是在call_results表里加一个turns字段存 JSON 数组每个元素包含user_text、predicted_intent、confidence。置信度低于阈值的样本自动进入待标注队列。ALTER TABLE call_results ADD COLUMN turns JSONB; -- 查询低置信度样本 SELECT id, turns FROM call_results WHERE EXISTS ( SELECT 1 FROM jsonb_array_elements(turns) AS t WHERE (t-confidence)::float 0.6 ) LIMIT 100;拿到这些样本后人工听录音、看文本修正意图标签然后把这些修正后的数据加入训练集。NLU 模型不需要全量重训用增量学习或者只微调最后一层分类头就行。我一般每两周做一次每次标注 200-300 条迭代三四轮后意图识别准确率能从 70% 出头提到 85% 以上。另一个技巧是话术的 A/B 测试。同一批号码随机分成两组一组用旧话术一组用新话术跑一周后对比转化率。注意两组号码的来源和客户等级要尽量一致否则对比结果没有意义。我见过把高等级客户全分到新话术组结果新话术「效果特别好」实际是客户本身质量高。最后说一个我自己的习惯每次调整话术或模型后先拿 50 个测试号码跑一遍人工听录音确认没有明显问题再放量。这个步骤花不了多少时间但能避免话术 bug 直接暴露给真实客户。外呼系统不像网页出了问题客户直接挂断你连修复的机会都没有。希望帮到你。本文还有配套的精品资源点击获取
简介这是一套面向电话营销与智能客服场景的AI语音通话销售机器人源码适合具备一定后端与语音开发基础的开发者、创业团队及企业技术选型人员用于搭建可自动外呼、应答与跟进客户的对话系统。源码整合了自然语言处理、语音识别、语音合成与机器学习算法能够理解客户意图与情绪并动态优化对话策略同时支持按业务需求定制话术、对接CRM等内部系统。压缩包共2002个文件约103.6MB以795个js脚本、203个html页面、216个css样式、219个xml配置及51个php文件构成前后端主体另含66个so动态库、49个h头文件与6个sql脚本并附带FreeSWITCH相关组件与语音资源便于理解通话链路与部署结构。资源还包含系统搭建教程覆盖环境配置、源码编译、数据库连接与API接口设置等环节可帮助读者快速跑通系统并完成二次开发。目前已有1239人学习下载适合作为语音外呼项目的实践参考。1. 一套电话销售机器人源码到底在替你干哪几件事很多人第一次听到「AI智能电话语音通话销售机器人源码」脑子里浮现的是一个能跟客户对答如流的数字人。实际拆开看它更像一条流水线把客户号码从数据库里捞出来自动拨出去接通后播放或生成话术识别客户说了什么判断意图再把结果写回数据库。整条链路里真正难的不是「让机器说话」而是「让机器听懂并做出正确分支」。我接触过的几个模拟项目里最常见的诉求是手里有一批历史客户线索人工一天打两百通就到顶了想用机器人把初筛做完人工只跟进有意向的那部分。这个方向适合有稳定线索来源、话术相对标准化的团队比如课程咨询、活动邀约、老客户回访。不适合需要深度定制方案、客户问题高度发散的场景那种情况机器人反而会消耗客户耐心。这套源码通常包含几个模块号码管理、任务调度、语音网关对接、ASR语音识别、NLU意图理解、TTS语音合成、对话状态机、结果回写。搭建教程要解决的核心问题就是把这些模块串起来让一个最小可用的外呼任务跑通。下面按我实际搭过的顺序从环境准备讲到参数调优和排错。2. 从零搭一套外呼系统环境、依赖与最小可跑通链路2.1 先想清楚选云服务还是本地部署这是第一个分叉口选错了后面返工成本很高。云服务方案把 ASR、TTS、语音网关都托管出去你只需要写业务逻辑优点是上线快、并发扩容简单缺点是按量计费通话量大时成本线性上涨而且话术和客户数据要经过第三方。本地部署方案需要自己准备 GPU 服务器跑 ASR 和 TTS 模型语音网关也要自己对接前期投入大但数据不出内网长期单价低。我一般会建议日呼量在 500 通以下、话术还在频繁调整的阶段先用云服务把流程跑通验证转化率等话术稳定、日呼量上到几千通再考虑把 ASR 和 TTS 迁到本地。迁移时业务层代码基本不用动只要换掉适配层。对比项云服务方案本地部署方案上线周期1-3 天2-4 周前期成本低高GPU 服务器单通成本随量线性增长固定成本摊薄后低数据可控性依赖第三方完全内网适合阶段验证期稳定放量期2.2 用 Docker Compose 把基础服务拉起来不管选哪种方案业务层建议用容器跑方便迁移。下面是一个最小化的 compose 文件包含业务服务、Redis存对话状态和 PostgreSQL存号码和结果。ASR/TTS 如果走云服务这里不需要额外容器如果本地部署再加两个模型服务容器。# docker-compose.yml version: 3.8 services: app: build: . ports: - 8080:8080 environment: - REDIS_HOSTredis - DB_HOSTpostgres - ASR_PROVIDERcloud # 本地部署时改成 local - TTS_PROVIDERcloud depends_on: - redis - postgres redis: image: redis:7-alpine ports: - 6379:6379 postgres: image: postgres:15-alpine environment: POSTGRES_DB: callbot POSTGRES_USER: bot POSTGRES_PASSWORD: botpass volumes: - pgdata:/var/lib/postgresql/data volumes: pgdata:逻辑说明app 服务是业务主进程负责调度外呼任务和跑对话状态机。Redis 存每一通电话的实时对话上下文因为对话状态读写频繁且允许丢失通话结束后落库用 Redis 比直接写数据库快得多。PostgreSQL 存号码列表、通话记录和意图标签这些数据需要持久化和查询。ASR_PROVIDER和TTS_PROVIDER两个环境变量是适配层的开关业务代码里根据这两个值决定调用云端 API 还是本地模型服务。参数说明Redis 用 alpine 镜像体积小生产环境建议加密码和持久化配置。PostgreSQL 的密码不要用示例里的换成强密码并通过环境变量注入。depends_on只保证启动顺序不保证服务就绪业务代码里要有重试逻辑。2.3 号码导入与任务表设计号码管理看起来简单实际最容易出问题。我见过把号码存在 Excel 里每次手动导入的也见过用一张大表所有任务混在一起的。推荐按「任务」维度隔离每个外呼任务一张逻辑表或者在同一张表里用 task_id 区分。-- 号码表 CREATE TABLE leads ( id BIGSERIAL PRIMARY KEY, task_id BIGINT NOT NULL, phone VARCHAR(20) NOT NULL, name VARCHAR(50), extra JSONB, -- 自定义字段比如客户等级、上次沟通记录 status VARCHAR(20) DEFAULT pending, -- pending/calling/done/failed attempt_count INT DEFAULT 0, last_call_at TIMESTAMP, created_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_leads_task_status ON leads(task_id, status); -- 通话结果表 CREATE TABLE call_results ( id BIGSERIAL PRIMARY KEY, lead_id BIGINT REFERENCES leads(id), call_id VARCHAR(64), duration INT, -- 通话秒数 intent VARCHAR(50), -- 识别出的意图标签 transcript TEXT, -- 完整对话文本 recording_url TEXT, created_at TIMESTAMP DEFAULT NOW() );逻辑说明leads 表用 status 字段控制调度调度器每次取statuspending且attempt_count 3的记录。extra 用 JSONB 存非结构化字段避免频繁改表结构。call_results 单独一张表因为一通电话可能对应多条结果记录比如中途转人工后又回写也方便后续做转化分析。参数说明attempt_count限制重拨次数一般设 2-3 次太多会被标记骚扰。idx_leads_task_status这个联合索引很关键调度器高频查询这个条件没有索引时数据量上万就会明显变慢。duration字段用来过滤无效通话小于 5 秒的基本是秒挂分析时要排除。2.4 对话状态机的核心循环这是整个系统的心脏。一通电话接通后状态机按预设流程走问候 → 确认身份 → 说明来意 → 处理异议 → 引导下一步。每一步都要等 ASR 返回文本再走 NLU 判断意图然后决定下一句说什么。# dialogue_engine.py import redis import json r redis.Redis(hostredis, port6379, decode_responsesTrue) # 简化的状态转移表 TRANSITIONS { greeting: {confirmed: pitch, denied: end, unknown: greeting_retry}, greeting_retry: {confirmed: pitch, denied: end, unknown: end}, pitch: {interested: qualify, not_interested: end, question: faq}, qualify: {qualified: transfer, unqualified: end}, } def handle_turn(call_id, user_text, current_state): # 1. 调用 NLU 获取意图 intent nlu_predict(user_text) # 返回 confirmed/denied/interested 等 # 2. 查状态转移表 next_state TRANSITIONS.get(current_state, {}).get(intent, end) # 3. 更新 Redis 中的对话上下文 ctx_key fcall:{call_id} r.hset(ctx_key, mapping{state: next_state, last_intent: intent}) r.expire(ctx_key, 3600) # 1小时过期防止僵尸会话 # 4. 根据下一状态生成回复文本 reply generate_reply(next_state, user_text) return next_state, reply def nlu_predict(text): # 实际项目中这里调用 NLU 服务或规则引擎 # 简化示例关键词匹配 if any(w in text for w in [是的, 对, 是我]): return confirmed if any(w in text for w in [不需要, 别打, 挂]): return denied return unknown逻辑说明handle_turn是每一轮对话的入口接收客户说的话和当前状态返回下一状态和机器人要说的文本。状态转移表用字典硬编码方便快速调整话术流程不需要改代码逻辑。Redis 里存的是当前通话的临时状态用expire设置过期时间防止异常情况下会话永远不释放。参数说明TRANSITIONS里的unknown分支要谨慎设计连续两次 unknown 就应该结束通话否则机器人会陷入循环。expire设 3600 秒是保守值实际通话很少超过 10 分钟但留足余量避免误删。nlu_predict在真实项目里应该是一个独立服务这里用关键词匹配只是为了说明调用位置生产环境要用训练好的分类模型。3. 语音链路调优ASR、TTS 与打断处理的参数怎么设3.1 ASR 识别率上不去的三个常见原因ASR 是整条链路里最影响体验的环节。识别错了后面全错。我踩过的坑集中在三处音频采样率不匹配、静音检测阈值太敏感、方言和口音没有针对性优化。采样率问题最隐蔽。电话线路通常是 8kHz 采样而很多 ASR 模型默认按 16kHz 训练。如果直接把 8kHz 音频喂进去识别率会断崖式下降。解决办法是在送入 ASR 之前做重采样或者选择支持 8kHz 的模型。import librosa import soundfile as sf def resample_audio(input_path, output_path, target_sr16000): # 电话录音通常是 8kHzASR 模型可能需要 16kHz y, sr librosa.load(input_path, srNone) if sr ! target_sr: y librosa.resample(y, orig_srsr, target_srtarget_sr) sf.write(output_path, y, target_sr) return output_path逻辑说明librosa.load的srNone表示保持原始采样率读取不做自动重采样。然后判断原始采样率是否等于目标采样率不等才做重采样。最后用 soundfile 写回文件。这一步在实时通话里不能落盘再读要用流式处理但离线测试时这样写最直观。参数说明target_sr根据你用的 ASR 模型文档来定常见是 16000。重采样会引入轻微失真但比采样率不匹配导致的识别错误小得多。如果 ASR 服务支持直接传 8kHz优先用原生支持省掉这一步。静音检测阈值是第二个坑。设得太高客户说话小声时被当成静音机器人抢话设得太低环境噪音被当成语音机器人一直等。我一般用能量加过零率的双门限法能量阈值根据线路底噪动态调整。import numpy as np def is_speech(frame, energy_threshold500, zcr_threshold0.1): # frame 是一帧音频数据长度通常 20-30ms energy np.sum(frame ** 2) / len(frame) zcr np.sum(np.abs(np.diff(np.sign(frame)))) / (2 * len(frame)) # 能量和过零率都超过阈值才认为是语音 return energy energy_threshold and zcr zcr_threshold逻辑说明能量判断声音大小过零率判断频率特性。语音信号能量高且过零率适中噪音通常过零率很高。两个条件同时满足才判定为语音减少误触发。参数说明energy_threshold需要根据实际线路底噪校准建议在正式外呼前录一段静音环境下的线路音频算平均能量然后设为其 3-5 倍。zcr_threshold一般设 0.1-0.2太高会漏掉清辅音太低会把摩擦音当成噪音。3.2 TTS 语音合成的自然度与延迟平衡TTS 有两个指标要同时看自然度和首字延迟。自然度高的模型往往参数多、推理慢首字延迟超过 800ms 客户就会觉得卡顿。我一般会准备两套 TTS一套轻量模型用于实时对话一套高质量模型用于生成固定话术的缓存音频。固定话术比如问候语、公司介绍、常见问题回答这些内容不变可以提前用高质量 TTS 生成好音频文件通话时直接播放零延迟。只有需要动态拼接的内容比如客户姓名、具体金额才走实时 TTS。import hashlib import os CACHE_DIR /app/tts_cache def get_tts_audio(text, voicefemale_01): # 用文本和音色生成缓存 key key hashlib.md5(f{voice}:{text}.encode()).hexdigest() cache_path os.path.join(CACHE_DIR, f{key}.wav) if os.path.exists(cache_path): return cache_path # 命中缓存直接返回 # 未命中调用 TTS 服务生成 audio tts_service.synthesize(text, voicevoice) with open(cache_path, wb) as f: f.write(audio) return cache_path逻辑说明用文本加音色做 MD5 作为缓存文件名保证相同内容只生成一次。缓存目录挂载到宿主机或对象存储容器重启不丢失。命中缓存时直接返回文件路径播放器读取本地文件比等 TTS 服务返回快得多。参数说明缓存 key 里要包含音色参数否则换音色后会读到旧音频。缓存目录要定期清理或者设置 LRU 淘汰策略避免磁盘占满。对于包含变量的话术比如「张先生您好」每个姓氏都会生成一个缓存文件量不大可以接受如果变量组合爆炸就要考虑模板加拼接的方式。3.3 打断处理让机器人学会闭嘴客户在机器人说话时插话系统要能检测到并停止播放否则客户会觉得机器人「聋」。实现方式是双工音频流播放 TTS 的同时持续采集麦克风数据检测到语音活动就发中断信号。import threading class DialogSession: def __init__(self): self.interrupted threading.Event() self.playing False def play_tts(self, audio_path): self.playing True self.interrupted.clear() # 模拟播放实际用音频库播放 for chunk in read_audio_chunks(audio_path): if self.interrupted.is_set(): break # 收到打断信号停止播放 play_chunk(chunk) self.playing False def on_speech_detected(self): if self.playing: self.interrupted.set() # 通知播放线程停止逻辑说明interrupted是一个线程安全的事件标志。播放线程每播放一个音频块就检查一次如果被设置就跳出循环。语音检测线程在采集到有效语音时调用on_speech_detected设置标志。这样播放和检测互不阻塞。参数说明打断检测的灵敏度要调太灵敏会被环境噪音误触发太迟钝客户说了一半机器人才停。建议在语音检测后加一个 200-300ms 的确认窗口连续多帧检测到语音才判定为打断。另外打断后要清空播放缓冲区否则残留音频会继续播出来。4. 避坑与排查外呼系统上线后最容易翻车的五个地方4.1 号码被标记导致接通率骤降现象系统刚上线时接通率还有 30%跑了一周后降到 10% 以下客户反馈显示号码被标记为「骚扰电话」。原因外呼频率过高、被叫方挂断后立即重拨、同一号码短时间内多次呼叫都会触发运营商的标记机制。另外号码没有做实名认证和资质报备也容易被拦截。解决控制单号码日呼量一般建议不超过 50 通两次呼叫间隔至少 30 分钟挂断后不要立即重拨等至少 1 小时。号码要完成企业实名认证外呼内容要能说明合法用途。如果已经被标记换号是最快的办法同时调整呼叫策略避免新号也被标记。4.2 对话状态在并发时串号现象A 客户的通话里突然出现 B 客户的信息或者机器人回复了上一通电话的内容。原因对话上下文用全局变量或单例存储多个通话共享了同一份状态。或者 Redis key 设计有误不同通话用了相同的 key。解决每通电话生成唯一 call_id所有状态读写都带这个 call_id 作为 key 的一部分。检查代码里有没有用类变量或模块级变量存对话状态。Redis 的 key 命名规范建议用call:{call_id}:state这种格式避免冲突。# 错误做法用全局字典 global_context {} # 所有通话共享必然串号 # 正确做法每个 call_id 独立 key def get_context(call_id): return r.hgetall(fcall:{call_id}:state)4.3 ASR 返回空结果导致对话卡死现象客户说了话但机器人没反应几秒后直接挂断。原因ASR 服务超时或返回空字符串状态机没有处理这种情况一直等输入。或者音频流在传输过程中丢失ASR 收到的是静音。解决给 ASR 调用设超时超时后走「没听清请您再说一遍」的兜底话术。连续两次空结果就转人工或结束通话。音频流要加序号和校验发现丢包时请求重传或直接跳过。def safe_asr(audio_stream, timeout3): try: result asr_service.recognize(audio_stream, timeouttimeout) if not result or len(result.strip()) 0: return None # 空结果上层走兜底 return result except TimeoutError: return None4.4 TTS 音频格式与播放器不兼容现象TTS 返回的音频播放出来是杂音或者播放速度不对。原因TTS 服务输出的采样率、位深、声道数与播放器预期不一致。比如 TTS 输出 24kHz 单声道播放器按 8kHz 双声道解析出来的就是噪音。解决在 TTS 适配层统一转成播放器要求的格式。用 ffmpeg 做格式转换是最稳的虽然多一步但兼容性好。# 把 TTS 输出统一转成 8kHz 16bit 单声道 ffmpeg -i tts_output.wav -ar 8000 -ac 1 -sample_fmt s16 play_ready.wav参数说明-ar 8000指定采样率 8kHz-ac 1单声道-sample_fmt s1616 位有符号整数。这三个参数要和你的语音网关文档对齐不同网关要求可能不同。4.5 数据库连接池耗尽导致新通话无法建立现象系统跑了一段时间后新外呼任务创建失败日志里出现连接超时错误。原因每次通话都创建新的数据库连接通话结束后没有正确释放。并发量上来后连接数超过数据库上限。解决用连接池管理数据库连接设置合理的最大连接数和空闲回收时间。确保每个通话结束后连接归还到池里而不是直接关闭。from sqlalchemy import create_engine from sqlalchemy.pool import QueuePool engine create_engine( postgresql://bot:botpasspostgres/callbot, poolclassQueuePool, pool_size10, # 常驻连接数 max_overflow20, # 峰值可额外创建的连接数 pool_recycle1800, # 30分钟回收空闲连接 pool_pre_pingTrue # 使用前检查连接是否有效 )参数说明pool_size根据并发通话数设置一般设为预期峰值的 1.5 倍。max_overflow是峰值时允许临时超出的数量不要设太大否则数据库扛不住。pool_recycle要小于数据库的wait_timeout避免拿到已被服务端关闭的连接。pool_pre_ping会多一次网络往返但能避免大部分「连接已断开」的错误。5. 让机器人越用越准意图标签回流与话术迭代的一个具体做法系统跑通只是开始真正决定转化率的是话术和意图识别的准确度。我一般会在通话结果表里加一个字段记录每一轮对话的 ASR 原文和 NLU 判定结果然后每周导出一次人工标注错误样本回流到 NLU 模型做增量训练。具体做法是在call_results表里加一个turns字段存 JSON 数组每个元素包含user_text、predicted_intent、confidence。置信度低于阈值的样本自动进入待标注队列。ALTER TABLE call_results ADD COLUMN turns JSONB; -- 查询低置信度样本 SELECT id, turns FROM call_results WHERE EXISTS ( SELECT 1 FROM jsonb_array_elements(turns) AS t WHERE (t-confidence)::float 0.6 ) LIMIT 100;拿到这些样本后人工听录音、看文本修正意图标签然后把这些修正后的数据加入训练集。NLU 模型不需要全量重训用增量学习或者只微调最后一层分类头就行。我一般每两周做一次每次标注 200-300 条迭代三四轮后意图识别准确率能从 70% 出头提到 85% 以上。另一个技巧是话术的 A/B 测试。同一批号码随机分成两组一组用旧话术一组用新话术跑一周后对比转化率。注意两组号码的来源和客户等级要尽量一致否则对比结果没有意义。我见过把高等级客户全分到新话术组结果新话术「效果特别好」实际是客户本身质量高。最后说一个我自己的习惯每次调整话术或模型后先拿 50 个测试号码跑一遍人工听录音确认没有明显问题再放量。这个步骤花不了多少时间但能避免话术 bug 直接暴露给真实客户。外呼系统不像网页出了问题客户直接挂断你连修复的机会都没有。希望帮到你。本文还有配套的精品资源点击获取
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
