☰
AI智能电话销售机器人源码搭建教程:从环境部署到对话状态机与ASR/TTS调优
2026/10/20 8:31:43 网站建设 项目流程

简介:这是一套面向电话营销与智能客服场景的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_HOST=redis - DB_HOST=postgres - ASR_PROVIDER=cloud # 本地部署时改成 local - TTS_PROVIDER=cloud 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 字段控制调度,调度器每次取status='pending'且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(host='redis', port=6379, decode_responses=True) # 简化的状态转移表 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 = f"call:{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_sr=16000): # 电话录音通常是 8kHz,ASR 模型可能需要 16kHz y, sr = librosa.load(input_path, sr=None) if sr != target_sr: y = librosa.resample(y, orig_sr=sr, target_sr=target_sr) sf.write(output_path, y, target_sr) return output_path

逻辑说明:librosa.load的sr=None表示保持原始采样率读取,不做自动重采样。然后判断原始采样率是否等于目标采样率,不等才做重采样。最后用 soundfile 写回文件。这一步在实时通话里不能落盘再读,要用流式处理,但离线测试时这样写最直观。

参数说明:target_sr根据你用的 ASR 模型文档来定,常见是 16000。重采样会引入轻微失真,但比采样率不匹配导致的识别错误小得多。如果 ASR 服务支持直接传 8kHz,优先用原生支持,省掉这一步。

静音检测阈值是第二个坑。设得太高,客户说话小声时被当成静音,机器人抢话;设得太低,环境噪音被当成语音,机器人一直等。我一般用能量加过零率的双门限法,能量阈值根据线路底噪动态调整。

import numpy as np def is_speech(frame, energy_threshold=500, zcr_threshold=0.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, voice="female_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, voice=voice) 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(f"call:{call_id}:state")

4.3 ASR 返回空结果导致对话卡死

现象:客户说了话,但机器人没反应,几秒后直接挂断。

原因:ASR 服务超时或返回空字符串,状态机没有处理这种情况,一直等输入。或者音频流在传输过程中丢失,ASR 收到的是静音。

解决:给 ASR 调用设超时,超时后走「没听清,请您再说一遍」的兜底话术。连续两次空结果就转人工或结束通话。音频流要加序号和校验,发现丢包时请求重传或直接跳过。

def safe_asr(audio_stream, timeout=3): try: result = asr_service.recognize(audio_stream, timeout=timeout) if not result or len(result.strip()) == 0: return None # 空结果,上层走兜底 return result except TimeoutError: return None

4.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:botpass@postgres/callbot", poolclass=QueuePool, pool_size=10, # 常驻连接数 max_overflow=20, # 峰值可额外创建的连接数 pool_recycle=1800, # 30分钟回收空闲连接 pool_pre_ping=True # 使用前检查连接是否有效 )

参数说明: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 直接暴露给真实客户。外呼系统不像网页,出了问题客户直接挂断,你连修复的机会都没有。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询