简介：这是一套面向电话营销与智能客服场景的AI语音通话销售机器人源码，适合具备一定后端与语音开发基础的开发者、企业技术团队研究部署。系统整合自然语言处理、语音识别、语音合成与机器学习算法，可模拟人声完成外呼、需求识别、意向判断与自动应答，并支持按业务定制对话策略、对接CRM等内部系统。压缩包共2002个文件，约103.6MB，以795个js、203个html、216个css构成前端与管理界面，51个php、49个h及66个so支撑服务端与语音引擎，另含xml配置、sql建库脚本、json接口数据、md说明文档及wav音频样例，目录层次完整。资源附带系统搭建教程，覆盖环境配置、源码编译、数据库连接与API接口设置等环节，并涉及负载均衡、缓存与数据隐私等工程要点。目前已有1239人学习下载，适合希望快速跑通AI外呼流程、理解语音交互链路并做二次开发的读者参考。
1. 从一通被挂断的电话说起：AI语音销售机器人到底能干什么
很多团队第一次接触 AI 智能电话语音通话销售机器人，都是被同一个场景逼出来的：外呼名单越堆越多，人工坐席一天打两百通就到顶，接通率还不到三成，客户听到第二句就挂。于是有人想，能不能让机器先把海量号码筛一遍，把有意向的、愿意多聊两句的挑出来，人工只跟进高价值线索。这个思路本身没问题，问题在于大多数人低估了「语音链路」的复杂度——它不是调一个大模型接口那么简单，而是 ASR、NLU、对话管理、TTS、电话线路五件事串在一起跑，任何一环抖动，客户听到的就是卡顿、答非所问、机械念稿。
这篇笔记面向两类人：一类是拿到一份 AI 电话销售机器人源码、想在自己环境里跑通并改造成业务可用的开发者；另一类是想评估这条路值不值得投入的技术负责人。我会按「先讲清系统由什么组成，再一步步搭起来，最后把踩过的坑摊开」的顺序写，代码和参数都给到能直接抄的程度。需要先说明的是，市面上流传的所谓「源码包」质量参差，本文不针对任何具体仓库，而是讲一套通用可复现的搭建路径，你手里的代码大概率能对上号。
2. 拆开这套系统：语音销售机器人的五个核心模块与选型逻辑
在动手之前，得先知道一通 AI 外呼电话从拨号到挂机，中间到底发生了什么。把它拆成五个模块，每个模块都有明确的输入输出和选型取舍，理解了这层，后面改代码才不会瞎改。
2.1 一通 AI 外呼的完整数据流
典型链路是这样的：任务调度器从名单里取一个号码，交给线路网关发起呼叫；对方接通后，音频流被切成 20ms 到 100ms 的小帧，送进 ASR 做流式识别；识别出的文本进入对话引擎，结合当前对话状态和业务话术，生成回复文本；回复文本交给 TTS 合成音频，再推回线路播放给对方。整个过程是双工的，对方随时可能打断，所以还要有一套打断检测和状态回滚机制。
这条链路里，延迟是生死线。经验值是：从对方说完最后一个字，到机器人开口，中间超过 800ms，客户就会觉得「这机器是不是卡了」；超过 1.5 秒，大概率直接挂断。所以后面所有选型和参数调整，都要围绕「把端到端延迟压到 800ms 以内」这个目标来做。
2.2 ASR 选型：流式识别为什么比整段识别更关键
电话场景的音频是 8kHz 采样、窄带、还带线路噪声，跟你在安静房间录的 16kHz 音频完全不是一回事。选 ASR 时，第一看它支不支持流式（streaming），第二看它有没有针对电话场景做优化。
整段识别（说完一句再识别）延迟天然高一截，因为要等静音检测判定句子结束。流式识别则是边说边出字，能把首字延迟压到 300ms 以内。下面是一个流式识别的调用骨架，用伪代码风格写，你按自己用的引擎替换即可：
# 流式 ASR 调用骨架：音频分帧送入，增量拿结果 import audioop class StreamingASR: def __init__(self, sample_rate=8000, frame_ms=20): self.sample_rate = sample_rate self.frame_bytes = int(sample_rate * frame_ms / 1000) * 2 # 16bit 采样 self.buffer = b"" self.session = self._open_session() def _open_session(self): # 建立流式识别会话，开启中间结果回调 # 不同引擎 API 不同，这里只表达「开一个长连接会话」的语义 return {"interim": True, "vad_silence_ms": 500} def push(self, pcm_chunk: bytes): self.buffer += pcm_chunk # 按固定帧长切分，避免一次送太多导致识别抖动 while len(self.buffer) >= self.frame_bytes: frame, self.buffer = self.buffer[:self.frame_bytes], self.buffer[self.frame_bytes:] self._send(frame) def _send(self, frame: bytes): # 送入引擎，拿中间结果或最终结果 # 中间结果用于「抢跑」，最终结果用于对话决策 pass这段代码的关键参数有两个：frame_ms和vad_silence_ms。frame_ms设 20ms 是电话场景的常见值，太小会增加调用次数和开销，太大则首字延迟上升。vad_silence_ms是静音判定阈值，设 500ms 意味着对方停顿半秒就认为一句话说完了；设太短会把正常换气当成句尾，设太长则机器人反应迟钝。我一般从 500ms 起步，根据实际通话录音再微调。
2.3 对话引擎：状态机和大模型该怎么配合
对话引擎是这套系统的大脑，也是最多人纠结的地方：到底用规则状态机，还是直接上大模型？我的建议是两者结合，而不是二选一。
纯状态机的好处是可控、可预测、延迟低，适合处理「确认身份、介绍产品、约定时间」这类流程固定的环节。纯大模型的好处是能应对开放式提问，但延迟高、还可能胡说。常见做法是：主流程用状态机兜底，在客户问出状态机没覆盖的问题时，把上下文丢给大模型生成回复，同时给大模型加一层输出过滤，防止它承诺业务不允许的内容。
# 对话引擎：状态机为主，大模型兜底 class DialogEngine: def __init__(self, fsm, llm_client, fallback_threshold=0.6): self.fsm = fsm # 状态机，管理主流程 self.llm = llm_client # 大模型客户端 self.fallback_threshold = fallback_threshold def respond(self, user_text, context): intent, score = self.fsm.classify(user_text) if score >= self.fallback_threshold: # 意图明确，走状态机，延迟低且可控 return self.fsm.next_action(intent, context) # 意图不明确，交给大模型，但限制输出长度和话题 prompt = self._build_prompt(user_text, context) reply = self.llm.generate(prompt, max_tokens=80) return self._sanitize(reply) def _sanitize(self, reply): # 过滤敏感承诺、价格、绝对化用语，业务红线在这里拦 banned = ["保证", "一定", "最低价"] for w in banned: if w in reply: return "这个问题我帮您转接人工同事确认一下，好吗？" return replyfallback_threshold是意图置信度阈值，设 0.6 意味着状态机只有六成把握时才让大模型接手。这个值调高，大模型介入变多，灵活但延迟和风险上升；调低则更保守。max_tokens=80是硬约束，防止大模型长篇大论，电话里没人听你说三段话。
2.4 TTS 与线路：音色、打断和回声消除
TTS 这块，电话场景优先选低延迟的流式合成，音色选偏自然、语速稍快的，别用那种一字一顿的播报腔。合成参数里，语速（speed）一般设 1.0 到 1.1，太快听不清，太慢显得呆。
线路侧最容易被忽略的是回声消除（AEC）。如果机器人自己的声音被麦克风（或线路）收回去，ASR 会把机器人说的话也识别成客户的话，导致对话逻辑彻底乱套。所以要么在网关层开 AEC，要么在送 ASR 之前做一轮本地回声抑制。打断处理也要在这里做：检测到客户说话，立刻停止当前 TTS 播放，并把对话状态回滚到「等待客户输入」。
3. 从零搭一套能跑的最小系统：环境、配置与联调步骤
理解了模块划分，接下来把它跑起来。这一章给的是最小可运行路径，目标是让你在本地或一台测试服务器上，打通「拨号—识别—回复—合成」的闭环，先不管并发和稳定性。
3.1 环境准备与依赖清单
先明确依赖。下面这张表是我一般会准备的组件清单，版本按你实际能拿到的稳定版填，不要盲目追新：
| 组件 | 作用 | 选型要点 |
|---|---|---|
| 线路网关 | 发起/接听电话，转发音频流 | 支持 SIP，能拿到双向 PCM 流 |
| ASR 服务 | 语音转文本 | 必须支持流式，电话场景优化 |
| 对话引擎 | 生成回复 | 状态机 + 大模型兜底 |
| TTS 服务 | 文本转语音 | 支持流式合成，低延迟 |
| 任务调度 | 名单管理、并发控制 | 能限速、能重试 |
| 存储 | 通话记录、录音、日志 | 录音和文本分开存，便于排查 |
环境上，Python 3.9 以上、一台能跑音频处理的机器就够起步。别一上来就上集群，先把单通电话跑顺。
3.2 配置文件怎么写：把参数从代码里抽出来
把参数硬编码在代码里是后期维护的灾难。我习惯用一个 YAML 配置文件集中管理，改参数不用动代码：
# config.yaml 核心配置示例 asr: engine: "streaming_asr" sample_rate: 8000 frame_ms: 20 vad_silence_ms: 500 interim_result: true dialog: fallback_threshold: 0.6 max_reply_tokens: 80 max_turns: 12 # 单通电话最多对话轮数，防止死循环 tts: voice: "female_natural" speed: 1.05 streaming: true call: max_concurrent: 20 # 并发上限，按线路和算力调 ring_timeout_s: 30 # 振铃超时 max_call_duration_s: 300 # 单通最长时长，超时强制挂断max_turns和max_call_duration_s是两个保命参数。没有它们，遇到一个爱聊的客户或者一个逻辑死循环，这通电话能打到天荒地老，既占线路又烧算力。max_concurrent要结合你的线路数和 ASR/TTS 的并发能力来定，盲目调高只会让每通电话都卡。
3.3 打通单通电话：联调顺序与验证方法
联调不要五个模块一起上，那样出了问题根本不知道是谁的锅。我一般按这个顺序来：
第一步，只测线路。拨一个测试号码，确认能接通、能收到对方的 PCM 音频流、能把一段本地音频播出去。这一步过了，说明线路通了。
第二步，接 ASR。把收到的音频流送进 ASR，在日志里打印识别结果。拿一段真实通话录音回放，看识别准不准、延迟多少。这一步重点看首字延迟和断句是否合理。
第三步，接对话引擎。先不接 TTS，把 ASR 的文本喂给对话引擎，看它生成的回复文本合不合理。用脚本模拟多轮对话，把状态机的分支都走一遍。
第四步，接 TTS。把回复文本合成音频，播回线路。这时候重点听音色、语速、有没有吞字。
第五步，全链路联调。真实拨号，自己扮演客户，走完整流程，掐表算端到端延迟。
# 联调时用日志观察各阶段耗时，定位瓶颈 # 在关键节点打时间戳，输出到日志 python run_call.py --config config.yaml --phone 10000 --debug 2>&1 | tee call.log # 统计各阶段平均耗时 grep -E "asr_first_char|dialog_done|tts_first_frame" call.log | awk '{print $1, $NF}'联调阶段一定要开 debug 日志，把 ASR 首字延迟、对话生成耗时、TTS 首帧延迟都打出来。哪个环节超了 300ms，就去优化哪个，别凭感觉猜。
4. 避坑指南：语音销售机器人落地时最容易翻车的五个地方
这套系统我见过太多团队在同一个地方摔倒。下面五条都是血泪经验，每条按「现象—原因—解决」写，你对号入座。
4.1 机器人自问自答，对话逻辑全乱
现象：客户明明没说话，机器人却像听到了什么一样继续往下说，甚至自己跟自己对话。
原因：九成是回声问题。机器人播放的 TTS 音频被线路或麦克风收回去，ASR 把它识别成了客户的话。电话线路如果没有做好回声消除，这个问题几乎必然出现。
解决：在网关层开启 AEC；如果网关不支持，就在送 ASR 之前做本地回声抑制，把机器人正在播放的音频作为参考信号减掉。另外，TTS 播放期间可以临时把 ASR 的输入静音，等播放完再恢复，代价是客户在这期间打断会被忽略，需要权衡。
4.2 首字延迟高，客户等不及就挂
现象：客户说完话，机器人要愣一两秒才开口，接通率上不去。
原因：多半是 ASR 用了整段识别，或者对话引擎里调了大模型但没做流式输出，又或者 TTS 是整句合成完才播。
解决：ASR 换流式；对话引擎对固定话术走状态机，别什么都丢给大模型；TTS 用流式合成，边合成边播。三管齐下，端到端延迟能压到 800ms 以内。如果还高，检查是不是网络往返太多，把服务部署得离线路网关近一点。
4.3 大模型胡说，承诺了业务不允许的内容
现象：客户随口一问，机器人给出了不该给的承诺，比如价格、效果保证。
原因：大模型没有业务边界意识，你给它什么上下文它就顺着说。
解决：在输出层加过滤，像 2.3 节代码里那样，命中敏感词就转成「帮您转接人工确认」。更稳妥的做法是给大模型限定话题范围，超出范围的问题一律走兜底话术。别指望靠提示词就能完全管住它，输出过滤这层必须有。
4.4 并发一上来，通话质量断崖式下跌
现象：单通测试很顺，一开二十路并发，各种卡顿、断音、识别错乱。
原因：ASR/TTS 服务有并发上限，算力被摊薄；或者线路带宽不够；或者任务调度没有限速，瞬间打满。
解决：先压测，摸清 ASR/TTS 在保证延迟前提下的最大并发数，把max_concurrent设在这个数以下。任务调度加令牌桶限速，别让呼叫瞬间爆发。线路带宽按每路 8kHz 单声道算，留足余量。
4.5 录音和日志缺失，出了问题没法复盘
现象：客户投诉机器人说错话，你翻遍系统找不到那通电话的录音和对话记录。
原因：只存了文本日志，没存原始音频；或者日志级别太低，关键节点没打点。
解决：录音和文本分开存，录音保留原始 PCM 或压缩后的音频，文本存每轮的识别结果和回复。关键节点（ASR 首字、对话生成、TTS 首帧、挂断原因）都要打时间戳。这套「黑匣子」平时没用，出事时是唯一的后悔药。
5. 让机器人更像人：三个可量化的调优技巧
系统跑通只是起点，真正决定转化率的是「像不像人」。这一章给三个能落地、能验证的调优方向，都是我反复试出来的。
5.1 用打断响应时间衡量「自然度」
自然对话里，人被打断会立刻停下听对方说。机器人也一样，打断响应时间（从检测到客户说话，到机器人停止播放）是衡量自然度的硬指标。我的经验值是压到 200ms 以内，客户基本感觉不到延迟。
实现上，在 TTS 播放线程和 ASR 输入线程之间加一个共享的「打断标志位」，ASR 一旦检测到有效人声（不是噪声），立刻置位，TTS 播放循环每帧检查这个标志，置位就停。注意要区分「有效人声」和「咳嗽、背景噪声」，否则机器人会被环境音频繁打断，反而显得神经质。这个区分靠 VAD 的置信度阈值来调，一般从 0.5 起步。
5.2 话术轮次控制：别让机器人变成话痨
电话销售最忌讳长篇大论。我一般给每轮回复设一个字数上限，比如 40 到 60 字，超过就拆成两轮说，中间留出客户插话的空间。同时用max_turns控制总轮数，超过就引导到「加个联系方式，稍后人工跟进」这类收尾话术。
验证方法很简单：拿真实通话录音，统计每轮机器人说话时长和客户说话时长的比例。健康的值大概在 1:1 到 1.5:1 之间，如果机器人说话时长是客户的三倍以上，说明话术太啰嗦，转化率大概率不好。
5.3 用 A/B 对比验证调优是否真的有效
调优最怕自嗨。同一个话术改动，你觉得更自然了，但转化率可能反而降了。所以每次改动都要做 A/B 对比：把名单随机分成两组，一组用旧话术，一组用新话术，跑够样本量（单组至少几百通），对比接通率、有效对话率、留资率。
# 简单的 A/B 分流与指标统计 import random def assign_group(phone, groups=("A", "B")): # 用号码做稳定哈希，保证同一号码每次分到同一组 return groups[hash(phone) % len(groups)] def summarize(records): # records: [{"group": "A", "connected": 1, "qualified": 0}, ...] stat = {} for r in records: g = r["group"] stat.setdefault(g, {"total": 0, "connected": 0, "qualified": 0}) stat[g]["total"] += 1 stat[g]["connected"] += r["connected"] stat[g]["qualified"] += r["qualified"] for g, s in stat.items(): s["connect_rate"] = s["connected"] / s["total"] s["qualify_rate"] = s["qualified"] / max(s["connected"], 1) return stat分流要用号码哈希而不是随机数，保证同一号码在不同批次里分到同一组，否则数据会互相污染。指标上，接通率反映线路和号码质量，有效对话率（对话超过三轮的比例）反映话术吸引力，留资率才是最终目标。三个指标一起看，才知道改动到底动了哪一环。
我自己的习惯是，任何话术或参数改动，没有 A/B 数据支撑就不上线。这套系统里「感觉」是最不可靠的东西，只有数据能告诉你机器人到底有没有变得更像人。希望这些能帮到你，少走点我当年走过的弯路。
本文还有配套的精品资源，点击获取
