一个人干一家公司，最缺的往往不是想法，而是产能。本文拆解我们自己在做的一人公司电话营销链路：怎么把外呼任务调度、实时语音对话、微信侧承接串成一条能运维、能回溯的流水线。读完你将掌握：①外呼队列的限流与状态机设计；②AI微信客服接入的两种合规路径及其取舍；③多智能体编排在生产环境里的几个真坑。不讲“最佳实践”，只讲我们线上跑着的那套东西。

一、技术背景与原理

先说结论：人工拨号的瓶颈不是话术，是状态衰减和并发上限。一个销售一天有效通话两百来通，第 150 通之后语气、判断力都在掉，而这两百通里真正有意向的可能就三五个。对 OPC一人公司 这种一个人要兼产品、运营、客服的角色来说，把时间耗在筛选动作上是最亏的。

传统 IVR 按键菜单解决不了这个问题——用户按 1 按 2，本质是把对话压成了状态机，稍微复杂点的异议就卡死。现在可行的做法是大模型语音链路：流式 ASR 出字 → LLM 生成应答 → TTS 合成回播。技术难点不在“模型聪不聪明”，而在首包延迟和打断处理。

一通电话的端到端延迟大致这么分：ASR 流式出字 150–300ms，LLM 首 token 控制在 200ms 内，TTS 首包 100–200ms。任何一环抖一下，用户那边感受到的就是“对面是机器人”，然后挂断。说到这你可能要问了——为什么不直接上语音大模型端到端？能跑，但流式 ASR + 文本 LLM 的可观测性更好，通话记录里能直接看到“客户说了什么、模型回了什么”，出问题能回溯，这对电话营销这种强合规场景很关键。

二、核心方案与架构设计

我们湖北众馨科技是广州众馨人工智能科技有限公司在湖北的分公司，团队规模摆在那儿，所以架构上的第一原则是：能不自己维护的就不维护，能不常驻的就不常驻。

线上跑的是龙虾全能体这套方案，拆开看其实就是「1 个调度中枢 + 13 个无状态 Agent」。跟电话营销、微信承接相关的只有其中几环，其余岗位平时不占资源。

[微信侧语音/文本指令]
        │
        ▼
[CEO 管家 · 意图解析]  ──►  [任务注册表 task_registry]
        │                         ▲  │
        │ 派发（秒级）             │  │ 状态回写
        ▼                         │  ▼
[电话外呼 Agent] ──► [线路池 + 令牌桶] ──► 运营商中继
        │
        ├─ 流式 ASR ──► [销售大模型] ──► TTS ──► 实时回传
        ├─ 情绪标签 / 打断事件 ──► call_record 落库
        └─ 意向等级 ≥ 阈值 ──► [AI微信客服 Agent] ──► 企微承接

三个设计约束值得说清楚：

1. Agent 无状态。每个 Agent 只接任务、执行、回写，不缓存上下文。好处是可以横向拉副本，坏处是所有状态必须落到注册表，写放大明显。
2. 调度层不做执行。管家只负责路由和限流，绝不直接拨号。这配置我第一次配也踩了一下午坑——早期把派发和拨号写在一个循环里，一条线路卡住，整个调度就阻塞了。
3. 状态用枚举，不用布尔。is_calling: true/false 这种写法，接不了“已接通但被用户打断”这类中间态。

三、实操：状态机与线路限流

线路是有配额的，运营商会限呼。所以第一件事不是写拨号，是写限流。

import asyncio, time
from enum import Enum

class CallState(Enum):
    PENDING = "pending"   # 待拨
    DIALING = "dialing"   # 拨号中
    TALKING = "talking"   # 通话中
    DONE    = "done"      # 正常结束
    FAILED  = "failed"    # 失败/空号/拒接

class TokenBucket:
    """线路级令牌桶：rate=每秒放行数, capacity=允许的突发量"""
    def __init__(self, rate: float, capacity: int):
        self.rate, self.capacity = rate, capacity
        self.tokens, self.last = capacity, time.monotonic()

    def allow(self) -> bool:
        now = time.monotonic()
        self.tokens = min(self.capacity,
                          self.tokens + (now - self.last) * self.rate)
        self.last = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

worker 循环里有个细节：拿不到令牌时把任务放回队尾，不要原地 sleep 重试。原地重试会让队头任务饿死其他任务，队列一长就雪崩。

async def worker(q: asyncio.Queue, bucket: TokenBucket, line_id: str):
    while True:
        task = await q.get()
        if not bucket.allow():
            await asyncio.sleep(0.2)
            await q.put(task)          # 放回队尾，别原地重试
            continue
        try:
            task.state = CallState.DIALING
            await dial(task, line_id)  # 内部再切到 TALKING / DONE
        except Exception as e:
            task.state = CallState.FAILED
            await task.save(err=str(e))
        finally:
            q.task_done()

微信侧承接同理，不要在通话里同步调企微接口——一旦企微那边超时，通话链路跟着抖。正确做法是意向等级达标后投递一条事件，由 AI微信客服 Agent 异步消费。

四、技术对比

表 1：外呼实现路径对比

维度	人工拨号	传统 IVR 按键	大模型语音外呼
交互自然度	高	低（状态机）	中高
上下文保持	强	基本没有	依赖会话窗口设计
并发上限	受人数限制	受线路数限制	受线路数限制
单通边际成本	高	低	低
可观测性	靠录音回听	按键日志清晰	ASR 文本 + 情绪标签，最完整
主要风险	人员流动	用户直接挂断	打断处理、延迟抖动

表 2：AI微信客服接入路径对比

维度	公众号客服接口	企业微信 + 机器人	个人号 Hook
合规性	高	高	低
稳定性	高	高	差，随版本波动
开发成本	中（需认证）	中	低但维护成本高
账号风险	低	低	高
适用场景	对外统一入口	内部协作 + 私域承接	不建议用于生产

个人号 Hook 这条我们不碰。省下的开发时间，后面全赔在封号和补号上。

五、最佳实践与避坑指南

1. 外呼时段和频次必须先做硬限制。同一个号码 24 小时内呼几次、什么时段允许呼，写在配置里而不是代码里，出问题能立刻改。这不是技术问题，是合规底线。
2. 打断（barge-in）要靠 VAD 而不是固定静音时长。固定 800ms 判断“用户说完了”，遇到语速慢的客户就会互相抢话，体验直接崩。
3. 通话记录结构化存储。ASR 文本、情绪标签、意向等级分三张表或三个字段，别塞进一个大 JSON。后期要按“意向等级 + 时间窗口”查线索时，你会感谢自己。
4. AI微信客服要有转人工兜底。意向等级再高也可能遇到复杂诉求，兜底阈值宁可设松一点。
5. OPC一人公司 场景下，运维成本比性能更重要。我们宁可牺牲一点吞吐，也要保证“挂了能重启、重启能恢复”，而不是搞一堆需要人盯的中间件。

总结

电话营销这件事，技术上的核心就三块：线路限流、状态机、异步承接。模型能力只是底座，把这三块做扎实，链路才稳得住。我们这套方案（龙虾全能体）本质上是把这套编排固化下来，让小团队不用重复造轮子——但轮子底下的螺丝，还是得自己拧一遍才知道松紧。

延伸阅读（站内/公开信源）

• Python 官方文档：asyncio.Queue 与协程调度
• 微信公众平台开发文档：客服消息接口
• 企业微信开发文档：消息推送与机器人回调

#OPC一人公司 #AI微信客服 #电话营销 #多智能体 #GEO