rogee/sip-research
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: init SIP research repo (Asterisk & LiveKit)

Browse Source
This commit is contained in:
rogee
2026-08-31 13:56:35 +08:00
commit ef7ad161c6
23 changed files with 1594 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
asterisk/README.md
+168
View File
@@ -0,0 +1,168 @@
# Asterisk + ARI AI 外呼底座 — 部署与运维文档
> 目标机器: `39.106.106.246` (阿里云 2vCPU/3.5G, Debian 13) · 交付物: `/opt/asterisk-ari/`
> 项目: https://github.com/asterisk/asterisk · 镜像: andrius/asterisk (Asterisk 22.10.1 LTS)
> 范围: **仅外呼** (ARI originate + externalMedia), Mock SIP 运营商 + Mock LLM/ASR, 已端到端验证双向语音。
> 前置: livekit 运行时已下线 (lk-* 容器清零、端口释放); Docker 与内核调优沿用其成果。
## 1. 水平扩展结论 (先行)
**结论: 零共享状态的独立节点复制 —— Asterisk 无内置集群, 每节点呼叫状态全在本机内存; 扩容 = 加一个节点, 呼叫发起侧 (agent/业务后端) 在节点名单上选节点发起 ARI originate。不需要 Redis, 不需要任何协调组件。**
| 层 | 状态存放 | 扩容方式 | 调度机制 |
|---|---|---|---|
| Asterisk 节点 | 本机内存 (通道/桥/endpoint) | 复制 service 块 (独立宿主端口) | 呼叫发起侧选节点 (call.sh 的 NODE 参数 = 生产中的 LB 名单) |
| AI Agent (生产) | 无状态 worker | `--scale`/K8s HPA | 每个外呼任务自带节点选择与 ARI/RTP 会话 |
对比 livekit 方案: livekit-sip 需要 Redis 存 trunk/呼叫状态并做派发; Asterisk 出站外呼场景连共享层都不需要 —— 每个呼叫自包含 (originate + bridge + externalMedia 全在一个节点内闭环), 节点间零交互。实测 1→2 节点**零配置变更** (共用同一套 conf), 呼叫按发起侧选择分摊 (见 §7)。
瓶颈与边界: ① 单节点容量受 CPU (RTP 收发 + 桥混音) 与 RTP 端口段限制 — 本文 10000-10800 共 800 端口, 每呼叫 2 腿 (被叫 + externalMedia) 各占 1 对端口 (RTP+RTCP) ≈ 200 并发/节点; ② 无跨节点媒体互通 — 仅外呼场景无影响 (呼叫不跨节点); ③ 未来要呼入/注册时需前置 Kamailio dispatcher (SIP 负载均衡), 纯出站不需要。
## 2. 架构
```
业务后端 / AI Agent (mock-agent, 容器或 host)
│ ① POST /bridges (mixing) ─ ARI REST, 节点可选 (ast1/ast2 轮询 = 调度层)
│ ② POST /channels/externalMedia ─ Asterisk 将桥内混音以 ulaw RTP 发往 agent
│ ③ POST /channels (originate) ─ PJSIP/mock-trunk 外呼被叫
│ ④ 轮询通道至 Up → addChannel 双通道入桥
▼
┌─ docker net: astari ──────────────────────────────────────────────┐
│ ast1 (ARI :8088, PJSIP :5060/udp, RTP 10000-10800) ← 独立节点 │
│ ast2 (ARI :8088→宿主 8089, 其余同上) ← 零共享状态 │
│ │ INVITE / 100/180 / 200 OK (SDP, PCMU) / ACK │
│ ▼ │
│ mock-provider (SIP UAS :5060/udp, RTP :40000, 440Hz 应答音) │
└────────────────────────────────────────────────────────────────────┘
agent ⇄ Asterisk: 混音 RTP 双向 (ulaw)
TX = 440Hz 间歇音 (Mock LLM/TTS) → 被叫听到的 "AI 说话"
RX = RMS 统计 (Mock ASR) → 被叫语音进 "识别"
```
外呼信令流 (仅出站):
```
业务后端 → POST /ari/channels (endpoint=PJSIP/+1510...@mock-trunk, app=outbound)
→ Asterisk 发 INVITE → mock-provider 100/180 → 200 OK (PCMU) → ACK
→ 被叫通道 Up, 与 externalMedia 通道同入 mixing 桥
→ agent 经 RTP 双向收发; 挂断 DELETE 通道 → BYE
```
## 3. 端口规划
| 端口 | 组件 | 说明 |
|---|---|---|
| 8088/tcp | ast1 | ARI REST + 事件 websocket, 宿主映射 8088 |
| 8089/tcp→8088 | ast2 | 第二节点 ARI (profile `scale`) |
| 5060/udp | ast1/ast2 | PJSIP 信令 (容器网内, 未映射宿主) |
| 10000-10800/udp | ast1/ast2 | RTP 媒体 (各容器独立 netns 不冲突) |
| 5060/udp, 40000/udp | mock-provider | Mock 运营商 SIP + RTP |
| 40001/udp | mock-agent (每次呼叫) | externalMedia 目的端口 |
生产外网仅需放行: SIP 5060/udp (+tcp/5061 tls) 与 RTP 段; **ARI 8088/8089 只对 agent 内网放行, 严禁公网裸开**。
## 4. 从零部署步骤 (全部实测)
```bash
# ── 4.1 Docker 与内核调优: 沿用 livekit 项目成果
# (Docker 已装; /etc/sysctl.d/99-livekit-sip.conf 已应用, 见 §8)
# livekit 运行时下线: cd /opt/livekit-sip && docker compose --profile multinode down
# ── 4.2 部署栈 ─────────────────────────────────────────────────────
mkdir -p /opt/asterisk-ari && cd /opt/asterisk-ari # = 仓库 deploy/ 目录
docker compose up -d # ast1 + mock-provider
docker compose --profile scale up -d # (可选) ast2 第二节点
# ── 4.3 验证 ───────────────────────────────────────────────────────
./scripts/health.sh # 容器 + ARI REST + trunk qualify + 活动通道
./scripts/call.sh +15105550123 12 1 # 端到端外呼 (节点1)
./scripts/call.sh +15105550123 10 2 # 端到端外呼 (节点2, 验证扩展)
```
配置仅 4 个文件 (~30 行, conf/): `http.conf` (ARI HTTP 8088)、`ari.conf` (用户 outbound)、`pjsip.conf` (出站 trunk endpoint+aor, 无注册, qualify 10s 探活)、`rtp.conf` (RTP 段 10000-10800)。
⚠ 踩坑1 (**最关键**): ARI 要求 Stasis app 有**活跃的 websocket 事件订阅**, 否则通道一进 app 就被立即挂断 — 表现为外呼 1.5s 即 BYE、轮询通道 404。mock-agent 内 `AppKeeper` 用原生 socket 保持 `/ari/events` 连接 (只丢事件), 控制面仍纯 REST。
⚠ 踩坑2: `pjsip show contacts` 显示 `NonQual` 只是首个 qualify 周期前的初始态, 等一个周期 (已设 10s) 即 `Avail`, 不是故障。
⚠ 踩坑3: agent 回发 RTP 的地址不是 external_host 自己, 而是通道变量 `UNICASTRTP_LOCAL_ADDRESS/PORT` (Asterisk 为该 externalMedia 通道分配的本机 RTP 端口); 拿不到时可用收到的首包源地址兜底。
⚠ 踩坑4: andrius/asterisk 的 entrypoint 会对 `/etc/asterisk` 做 chown — 按文件 RO 挂载没问题 (chown 带 `|| true` 兜底); 不要挂空目录, 否则内置配置全丢导致 Stasis 初始化失败。
## 5. 外呼链路验证 (实测输出)
`./scripts/call.sh +15105550123 12 1`:
```
[agent] stasis app 'outbound' running (event ws connected)
[agent] externalMedia channel up: UnicastRTP/mock-agent-12981-...
[agent] asterisk RTP return addr: 172.18.0.2:10550
[agent] INVITE sent via mock-trunk -> +15105550123 (channel callee-1788153974-699)
[agent] event StasisStart PJSIP/mock-trunk-00000000
[agent] callee answered: state=Up
[agent] bridged [em-... + callee-...], streaming 12.0s
[agent] RX frames=500 avg_rms=8374
RESULT number=+15105550123 rx_frames=512 rx_avg_rms=8318 tx_frames=512 bidirectional=YES
```
mock-provider (被叫侧) 日志:
```
INCOMING_INVITE from=172.18.0.2:5060 uri="sip:+15105550123@mock-provider:5060"
ANSWERED media=172.18.0.3:40000 codec=PCMU peer=172.18.0.2:10528 ← SDP 协商 PCMU
ACK confirmed, RTP bridging
CALL_DONE reason=bye dur=13.7s sent_rtp=517 recv_rtp=512 recv_avg_amplitude=8167
```
**双向语音证明**: agent RX=512 帧 (被叫→ASR 方向, RMS 8318 = 440Hz 检测到) 与 provider recv_rtp=512 (LLM/TTS→被叫方向) **完全对称**; 挂断后 ARI channels=0, 无泄漏。换真实 LLM/ASR 只需替换 mock-agent 的 TX 音源与 RX 消费循环。
## 6. 运行状态检测
`./scripts/health.sh` (实测输出):
```
== 容器状态 ast1 / ast2 / ast-mock-provider Up (healthy)
== ARI REST PASS ast1 /asterisk/info (200) PASS ast2 (200)
== PJSIP trunk PASS ast1 mock-provider Avail PASS ast2 Avail (OPTIONS qualify)
== 活动呼叫 PASS ast1 active channels=0
result: ok=5 fail=0
```
持续监控: `watch -n5 ./scripts/health.sh`; 呼叫级状态: provider 日志 `CALL_DONE` 行 (时长/双向计数); 容器级: docker healthcheck (镜像自带)。层级与 livekit 版对齐: 容器 + HTTP/API + 业务功能三层。
## 7. 水平扩展实证 (双节点, 零配置变更)
**① 加节点即扩容**: `docker compose --profile scale up -d` 拉起 ast2, 共用同一套 conf, 无任何既有节点配置改动。
**② 3 路并发分摊双节点 (发起侧调度)**:
```
$ (./scripts/call.sh +15105550123 10 1 &) (./scripts/call.sh +15105550123 10 1 &) (./scripts/call.sh +15105550123 10 2 &)
RESULT ... rx_frames=401 rx_avg_rms=8438 tx_frames=401 bidirectional=YES
RESULT ... rx_frames=398 rx_avg_rms=7991 tx_frames=401 bidirectional=YES
RESULT ... rx_frames=404 rx_avg_rms=8077 tx_frames=405 bidirectional=YES
$ docker logs ast-mock-provider | grep INCOMING | grep -oE "from=[0-9.]+" | sort | uniq -c
2 from=172.18.0.2 ← ast1 处理 2 路 (源 IP = ast1)
1 from=172.18.0.4 ← ast2 处理 1 路 (源 IP = ast2)
```
→ 加副本即扩容、发起侧调度、双节点各自满血双向, 三项均实证。
资源基线 (3 并发): ast1 1.4%CPU/53M, ast2 1.1%CPU/52M — 2vCPU 单节点距容量上限还有两个数量级余量。
## 8. 系统调优
**内核** (沿用已应用的 `/etc/sysctl.d/99-livekit-sip.conf`, RTP/UDP 通用, 无需改动):
```conf
net.core.rmem_max = 16777216 # UDP 收缓冲上限 (16M, 默认 208K 高并发必丢包)
net.core.wmem_max = 16777216
net.core.rmem_default = 1048576
net.core.wmem_default = 1048576
net.core.netdev_max_backlog = 4096
net.ipv4.ip_local_port_range = 10000 65000 # 出向 SIP/RTP 源端口
net.ipv4.udp_mem = 8388608 12582912 16777216 # 全局 UDP 页缓存 (3.5G 内存安全值)
```
验证: `nstat -az | grep -i udp` (`UdpRcvbufErrors` 应为 0)。
**组件级**:
- rtp.conf: 每呼叫 2 腿 × 1 对端口 (RTP+RTCP); 800 端口 ≈ 200 并发/节点, 不够就扩段或加节点 (加节点更快, 见 §7)。
- pjsip: `direct_media=no` 必须保留 — 媒体必须过 Asterisk 进桥, 否则 externalMedia 拿不到音频; `qualify_frequency=10` 快速摘除故障 trunk。
- asterisk.conf (如需): `maxcalls`/`maxload` 做背压 (默认无限, 生产按容量设)。
- 容器: `restart: unless-stopped` 已配; 双节点各 ~52M 内存, 3.5G 机器无需 mem limit。
- 监控: ARI `GET /channels` 数 + docker healthcheck; 告警项: 活动通道数骤降、trunk `Unavailable`、容器重启、`UdpRcvbufErrors>0`。
## 9. 生产化清单 (Mock → 真实运营商)
1. trunk: pjsip.conf 加 `outbound_auth` (digest 用户名/密码) 与 `from_user/from_domain`, contact 换运营商 SIP 域名/IP; 我方 5060/udp、RTP 段公网放行 (阿里云安全组)。
2. ARI 安全: 8088/8089 仅对 agent 网段放行; 换强密码; 可开 http.conf TLS。
3. NAT: 阿里云 ECS 1:1 NAT — PJSIP transport 配 `external_media_address/external_signaling_address` (公网 IP); externalMedia 的 external_host 用 agent 可达地址。
4. AI 侧: mock-agent 替换为真实 STT/LLM/TTS (RX 帧喂识别、TX 换合成音频, RTP/ARI 接口不变)。
5. 高可用: 节点 ≥2 (已验证); 调度侧带健康检查摘除故障节点; 需要呼入/注册时前置 Kamailio dispatcher。
asterisk/deploy/conf/ari.conf
+8
View File
@@ -0,0 +1,8 @@
[general]
enabled = yes
pretty = no
[outbound]
type = user
read_only = no
password = ari_mock_7c3f9e1b5a2d4806
asterisk/deploy/conf/http.conf
+4
View File
@@ -0,0 +1,4 @@
[general]
enabled = yes
bindaddr = 0.0.0.0
bindport = 8088
asterisk/deploy/conf/pjsip.conf
+21
View File
@@ -0,0 +1,21 @@
; PJSIP — 出站 trunk (mock-provider = Mock SIP 运营商)
; 仅出站外呼: endpoint + aor(qualify 探活), 无注册/无鉴权; 换真实运营商改 contact 并加 outbound_auth
[transport-udp]
type = transport
protocol = udp
bind = 0.0.0.0:5060
[mock-trunk]
type = endpoint
transport = transport-udp
context = from-external
disallow = all
allow = ulaw
direct_media = no
aors = mock-trunk
[mock-trunk]
type = aor
contact = sip:mock-provider:5060
qualify_frequency = 10 ; 首次探测 ≤10s 内完成, 重启后健康检查快速变绿
asterisk/deploy/conf/rtp.conf
+6
View File
@@ -0,0 +1,6 @@
; RTP 端口段: 每呼叫占 2 个端口对 (被叫腿 + externalMedia 腿, 各含 RTCP)
; 10000-10800 = 800 端口 ≈ 200 并发呼叫/节点
[general]
rtpstart = 10000
rtpend = 10800
strictrtp = no
asterisk/deploy/docker-compose.yml
+47
View File
@@ -0,0 +1,47 @@
name: asterisk-ari
services:
asterisk1:
image: andrius/asterisk:latest
container_name: ast1
volumes:
- ./conf/http.conf:/etc/asterisk/http.conf:ro
- ./conf/ari.conf:/etc/asterisk/ari.conf:ro
- ./conf/pjsip.conf:/etc/asterisk/pjsip.conf:ro
- ./conf/rtp.conf:/etc/asterisk/rtp.conf:ro
ports: ["8088:8088"]
restart: unless-stopped
networks: [astari]
# 第二个 Asterisk 节点: 无共享状态, 出站外呼各自独立 (水平扩展验证)
asterisk2:
image: andrius/asterisk:latest
container_name: ast2
volumes:
- ./conf/http.conf:/etc/asterisk/http.conf:ro
- ./conf/ari.conf:/etc/asterisk/ari.conf:ro
- ./conf/pjsip.conf:/etc/asterisk/pjsip.conf:ro
- ./conf/rtp.conf:/etc/asterisk/rtp.conf:ro
ports: ["8089:8088"]
restart: unless-stopped
networks: [astari]
profiles: [scale]
# Mock SIP 运营商 (UAS): 应答 Asterisk 的出站 INVITE
mock-provider:
image: python:3.12-alpine
container_name: ast-mock-provider
working_dir: /app
command: python mock/mock-provider.py
volumes:
- ./mock:/app/mock:ro
environment:
SIP_PORT: "5060"
RTP_PORT: "40000"
RING_DELAY: "1.5"
restart: unless-stopped
networks: [astari]
networks:
astari:
driver: bridge
asterisk/deploy/mock/__pycache__/mock-agent.cpython-313.pyc
BIN
View File
Binary file not shown.
asterisk/deploy/mock/__pycache__/mock-provider.cpython-313.pyc
BIN
View File
Binary file not shown.
asterisk/deploy/mock/mock-agent.py
Executable
+295
View File
@@ -0,0 +1,295 @@
#!/usr/bin/env python3
"""Mock AI Agent — Asterisk/ARI 版.
外呼流程 (控制面纯 REST + RTP; 另加一个原生 socket websocket 仅用于保持 Stasis app 运行,
否则 ARI 会把进入 app 的通道直接挂断):
0. 连接 /ari/events?app=outbound —— 让 Stasis app 处于运行态 (只丢事件, 不解析业务)
1. POST /bridges 建立 mixing 桥
2. POST /channels/externalMedia 外部媒体通道 (ulaw), Asterisk 把桥内混音发到 external_host,
本侧回发地址 = 通道变量 UNICASTRTP_LOCAL_ADDRESS/PORT (兜底: 首包源地址)
3. POST /channels originate 经 PJSIP trunk 外呼被叫
4. 轮询被叫通道至 Up (应答), 双通道入桥
5. RTP 双向: RX=被叫音频(模拟 ASR 输入, RMS 统计); TX=440Hz 间歇音(模拟 LLM/TTS 输出)
6. DELETE 通道+桥 → BYE
纯 stdlib (urllib + asyncio + 原生 socket)。换真实 LLM/ASR: 替换 TX 音源与 RX 消费即可。
用法: mock-agent.py --ari-base http://asterisk1:8088 --external-host <本容器名>:40001 \
--number +15105550123 [--duration 15]
"""
import argparse, asyncio, base64, json, math, os, random, socket, struct, sys, threading, time
import urllib.error, urllib.parse, urllib.request
def log(*a): print(f"[agent {time.strftime('%H:%M:%S')}]", *a, flush=True)
# ---------- G.711 u-law (与 mock-provider 同源) ----------
def ulaw_encode(s: int) -> int:
s = max(-32635, min(32635, s)); sign = 0x80 if s < 0 else 0
if sign: s = -s
s += 0x84
e = 7
for i in range(7, -1, -1):
if s & (0x40 << i): e = i; break
return (~(sign | (e << 4) | ((s >> (e + 3)) & 0x0F))) & 0xFF
def ulaw_decode(u: int) -> int:
u = ~u & 0xFF
sign = u & 0x80; e = (u >> 4) & 0x07; m = u & 0x0F
s = ((m << (e + 3)) + 0x84) << 2
return -s if sign else s
def tone_payload(ts: float) -> bytes:
"""20ms / 160 samples @8kHz, 440Hz 500ms-on 500ms-off (模拟 TTS)"""
out = bytearray()
on = (ts % 1.0) < 0.5
for i in range(160):
t = ts + i / 8000.0
v = int(6000 * math.sin(2 * math.pi * 440 * t)) if on else 0
out.append(ulaw_encode(v))
return bytes(out)
# ---------- ARI REST ----------
class Ari:
def __init__(self, base, user, password):
self.base = base.rstrip("/")
self.auth = "Basic " + base64.b64encode(f"{user}:{password}".encode()).decode()
def req(self, method, path, params=None):
url = f"{self.base}/ari{path}"
if params:
url += "?" + urllib.parse.urlencode(params)
r = urllib.request.Request(url, method=method)
r.add_header("Authorization", self.auth)
try:
with urllib.request.urlopen(r, timeout=15) as resp:
raw = resp.read()
return resp.status, (json.loads(raw) if raw else None)
except urllib.error.HTTPError as e:
raw = e.read()
try:
return e.code, (json.loads(raw) if raw else None)
except Exception:
return e.code, {"error": raw.decode("utf-8", "replace")}
# ---------- ARI Stasis app 保活 (极简事件 websocket, 纯 stdlib) ----------
class AppKeeper:
"""ARI 要求 Stasis app 有活跃 websocket 订阅, 否则进入 app 的通道会被立即挂断.
本类只保持连接并丢弃事件, 控制面仍走 REST."""
def __init__(self, base, app, user, password):
u = urllib.parse.urlparse(base)
self.host, self.port = u.hostname, u.port or 80
self.app, self.user, self.password = app, user, password
self.sock = None
def start(self):
key = base64.b64encode(os.urandom(16)).decode()
req = ("GET /ari/events?app=" + urllib.parse.quote(self.app)
+ "&api_key=" + urllib.parse.quote(f"{self.user}:{self.password}")
+ f" HTTP/1.1\r\nHost: {self.host}:{self.port}\r\n"
+ "Upgrade: websocket\r\nConnection: Upgrade\r\n"
+ f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n")
self.sock = socket.create_connection((self.host, self.port), timeout=10)
self.sock.sendall(req.encode())
resp = b""
while b"\r\n\r\n" not in resp:
chunk = self.sock.recv(1) # 逐字节读响应头, 不吞后续 ws 帧
if not chunk:
raise ConnectionError("ws closed during handshake")
resp += chunk
if b"101" not in resp.split(b"\r\n", 1)[0]:
raise ConnectionError(f"ws handshake failed: {resp[:120]!r}")
self.sock.settimeout(None)
threading.Thread(target=self._drain, daemon=True).start()
def _recv_exact(self, n):
buf = b""
while len(buf) < n:
c = self.sock.recv(n - len(buf))
if not c:
raise ConnectionError("ws closed")
buf += c
return buf
def _drain(self):
try:
while True:
b0 = self._recv_exact(1)[0]
b1 = self._recv_exact(1)[0]
ln = b1 & 0x7F
if ln == 126:
ln = struct.unpack("!H", self._recv_exact(2))[0]
elif ln == 127:
ln = struct.unpack("!Q", self._recv_exact(8))[0]
payload = self._recv_exact(ln) if ln else b""
op = b0 & 0x0F
if op == 0x8: # close
break
if op == 0x9: # ping → pong
mask = os.urandom(4) # client 帧必须带 mask
masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
self.sock.sendall(bytes((0x8A, 0x80 | (len(payload) & 0x7F))) + mask + masked)
continue
if op == 0x1 and payload:
try:
ev = json.loads(payload)
except Exception:
continue
if ev.get("type") in ("StasisStart", "StasisEnd", "ChannelDestroyed"):
log(f"event {ev['type']} {(ev.get('channel') or {}).get('name', '')}")
except Exception:
pass
def stop(self):
try:
if self.sock:
self.sock.close()
except Exception:
pass
# ---------- RTP 媒体 ----------
class Rx(asyncio.DatagramProtocol):
def __init__(self, stats):
self.stats, self.transport = stats, None
def connection_made(self, transport):
self.transport = transport
def datagram_received(self, data, addr):
if len(data) < 12: return
self.stats["rx"] += 1
payload = data[12:]
dec = [ulaw_decode(b) for b in payload[:40]]
self.stats["amp_sum"] += sum(abs(x) for x in dec) / max(1, len(dec))
if self.stats["tx_to"] is None: # 兜底: 用首包源地址作回程
self.stats["tx_to"] = addr
if self.stats["rx"] % 250 == 0:
log(f"RX frames={self.stats['rx']} avg_rms={self.stats['amp_sum']/self.stats['rx']:.0f}")
async def media_loop(listen_port, tx_to, duration):
stats = {"rx": 0, "amp_sum": 0.0, "tx": 0, "tx_to": tx_to}
loop = asyncio.get_running_loop()
rx_t, _ = await loop.create_datagram_endpoint(
lambda: Rx(stats), local_addr=("0.0.0.0", listen_port))
async def tx():
seq = random.randint(0, 65535); ts = random.randint(0, 10**6)
ssrc = random.randint(1, 2**31)
t0 = time.time()
while time.time() - t0 < duration:
if stats["tx_to"]:
payload = tone_payload(time.time() - t0)
hdr = struct.pack("!BBHII", 0x80, 0, seq & 0xFFFF, ts & 0xFFFFFFFF, ssrc)
rx_t.sendto(hdr + payload, stats["tx_to"])
stats["tx"] += 1
seq += 1; ts += 160
await asyncio.sleep(0.02)
try:
await asyncio.wait_for(tx(), timeout=duration + 5)
finally:
rx_t.close()
return stats
# ---------- 主流程 ----------
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--ari-base", default="http://127.0.0.1:8088")
ap.add_argument("--ari-user", default="outbound")
ap.add_argument("--ari-pass", default="ari_mock_7c3f9e1b5a2d4806")
ap.add_argument("--app", default="outbound")
ap.add_argument("--external-host", required=True, help="本侧 RTP 地址 ip:port (供 Asterisk 回发混音)")
ap.add_argument("--number", default="+15105550123")
ap.add_argument("--caller-id", default="+15109990001")
ap.add_argument("--trunk", default="mock-trunk")
ap.add_argument("--duration", type=float, default=15)
ap.add_argument("--ring-timeout", type=float, default=30)
args = ap.parse_args()
listen_port = int(args.external_host.rsplit(":", 1)[1])
ari = Ari(args.ari_base, args.ari_user, args.ari_pass)
suffix = f"{int(time.time())}-{random.randint(100, 999)}"
bridge_id, em_id, callee_id = f"br-{suffix}", f"em-{suffix}", f"callee-{suffix}"
# 0. 保持 Stasis app 运行 (否则通道进 app 即被挂断)
keeper = AppKeeper(args.ari_base, args.app, args.ari_user, args.ari_pass)
keeper.start()
log(f"stasis app '{args.app}' running (event ws connected)")
def fail(msg, detail=None):
log(f"FAIL: {msg}", detail or "")
cleanup()
sys.exit(1)
def cleanup():
for m, p in (("DELETE", f"/channels/{callee_id}"), ("DELETE", f"/channels/{em_id}"),
("DELETE", f"/bridges/{bridge_id}")):
try: ari.req(m, p)
except Exception: pass
keeper.stop()
# 1. 混音桥
st, r = ari.req("POST", "/bridges", params={"type": "mixing", "bridgeId": bridge_id})
if st not in (200, 201): fail("create bridge", (st, r))
# 2. 外部媒体通道 (Asterisk → 本侧 RTP)
st, em = ari.req("POST", "/channels/externalMedia", params={
"app": args.app, "external_host": args.external_host,
"format": "ulaw", "channelId": em_id})
if st not in (200, 201): fail("create externalMedia", (st, r))
log(f"externalMedia channel up: {em.get('name')}")
# 3. Asterisk 侧 RTP 回送地址
tx_to = None
st, v = ari.req("GET", f"/channels/{em_id}/variable", {"variable": "UNICASTRTP_LOCAL_ADDRESS"})
ip = (v or {}).get("value") if st == 200 else None
st, v = ari.req("GET", f"/channels/{em_id}/variable", {"variable": "UNICASTRTP_LOCAL_PORT"})
port = (v or {}).get("value") if st == 200 else None
if ip and port:
tx_to = (ip, int(port))
log(f"asterisk RTP return addr: {ip}:{port}")
else:
log("UNICASTRTP vars not set; will reply-to-source on first RX")
# 4. 发起外呼
st, ch = ari.req("POST", "/channels", params={
"endpoint": f"PJSIP/{args.number}@{args.trunk}",
"app": args.app, "appArgs": "outbound",
"callerId": args.caller_id, "channelId": callee_id,
"timeout": int(args.ring_timeout)})
if st not in (200, 201): fail("originate", (st, ch))
log(f"INVITE sent via {args.trunk} -> {args.number} (channel {callee_id})")
# 5. 等待被叫应答
deadline = time.time() + args.ring_timeout
state = None
while time.time() < deadline:
st, ch = ari.req("GET", f"/channels/{callee_id}")
if st == 200:
state = ch.get("state")
elif st in (404, 410):
fail("callee channel gone before answer")
if state == "Up":
break
time.sleep(0.25)
if state != "Up":
fail(f"no answer (state={state})")
log(f"callee answered: state={state}")
# 6. 双通道入桥 → 双向 RTP
st, r = ari.req("POST", f"/bridges/{bridge_id}/addChannel",
params={"channel": f"{em_id},{callee_id}"})
if st not in (200, 201, 204): fail("addChannel", (st, r))
log(f"bridged [{em_id} + {callee_id}], streaming {args.duration}s "
f"(RX=ASR mock, TX=440Hz LLM/TTS mock)")
# 7. 媒体收发
stats = asyncio.run(media_loop(listen_port, tx_to, args.duration))
# 8. 挂断收尾
cleanup()
n = stats["rx"]
amp = stats["amp_sum"] / n if n else 0.0
print(f"RESULT number={args.number} rx_frames={n} rx_avg_rms={amp:.0f} "
f"tx_frames={stats['tx']} bidirectional={'YES' if n > 50 and amp > 100 else 'NO'}",
flush=True)
if __name__ == "__main__":
main()
asterisk/deploy/mock/mock-provider.py
Executable
+244
View File
@@ -0,0 +1,244 @@
#!/usr/bin/env python3
"""Mock SIP 运营商 (UAS): 接收 Asterisk(PJSIP) 出站 INVITE, 振铃->应答, 双向收发 PCMU/PCMA RTP.
仅 stdlib, 无依赖。一个 RTP 端口复用所有呼叫(按来源 IP 匹配呼叫)。
用法: python3 mock-provider.py (环境变量 SIP_PORT/RTP_PORT/RING_DELAY)
"""
import asyncio, math, os, random, socket, struct, time
SIP_PORT = int(os.environ.get("SIP_PORT", "5060"))
RTP_PORT = int(os.environ.get("RTP_PORT", "40000"))
RING_DELAY = float(os.environ.get("RING_DELAY", "1.5"))
MY_IP = socket.gethostbyname(socket.gethostname())
def log(*a): print(f"[provider {time.strftime('%H:%M:%S')}]", *a, flush=True)
# ---------- G.711 编码 ----------
def ulaw_encode(s: int) -> int:
s = max(-32635, min(32635, s)); sign = 0x80 if s < 0 else 0
if sign: s = -s
s += 0x84
e = 7
for i in range(7, -1, -1):
if s & (0x40 << i): e = i; break
return (~(sign | (e << 4) | ((s >> (e + 3)) & 0x0F))) & 0xFF
def alaw_encode(s: int) -> int:
s = max(-32635, min(32635, s)); sign = 0x80 if s < 0 else 0
if sign: s = -s
e = 7
for i in range(7, -1, -1):
if s & (0x40 << i): e = i; break
m = (s >> (e + 3)) & 0x0F
return ((sign | (e << 4) | m) ^ 0x55) & 0xFF
def ulaw_decode(u: int) -> int:
u = ~u & 0xFF
sign = u & 0x80; e = (u >> 4) & 0x07; m = u & 0x0F
s = ((m << (e + 3)) + 0x84) << 2
return -s if sign else s
def tone_payload(codec: str, ts: float) -> bytes:
"""20ms / 160 samples @8kHz, 440Hz 500ms-on 500ms-off"""
out = bytearray()
on = (ts % 1.0) < 0.5
for i in range(160):
t = ts + i / 8000.0
v = int(6000 * math.sin(2 * math.pi * 440 * t)) if on else 0
out.append(ulaw_encode(v) if codec == "PCMU" else alaw_encode(v))
return bytes(out)
# ---------- SIP 报文 ----------
def parse_msg(data: bytes):
head, _, body = data.decode("utf-8", "replace").partition("\r\n\r\n")
lines = head.split("\r\n")
headers = []
for ln in lines[1:]:
n, _, v = ln.partition(":")
headers.append((n.strip().lower(), v.strip()))
return lines[0], headers, body
def hget(headers, name):
for n, v in headers:
if n == name: return v
return None
def build_resp(first, headers, code, reason, tag=None, extra=None, body=""):
out = [f"SIP/2.0 {code} {reason}"]
for n, v in headers:
if n == "via": out.append(f"Via: {v}")
out.append(f"From: {hget(headers,'from')}")
to = hget(headers, "to")
if tag and "tag=" not in to: to = f"{to};tag={tag}"
out.append(f"To: {to}")
out.append(f"Call-ID: {hget(headers,'call-id')}")
out.append(f"CSeq: {hget(headers,'cseq')}")
if extra:
out += extra
if body:
out.append(f"Content-Type: application/sdp")
out.append(f"Content-Length: {len(body)}")
return ("\r\n".join(out) + "\r\n\r\n" + body).encode()
def parse_sdp_offer(body: str):
"""返回 (对端媒体IP, 对端媒体端口, payload类型列表)"""
ip, port, pts = None, None, []
for ln in body.splitlines():
if ln.startswith("c=IN IP4 "): ip = ln.split()[2]
elif ln.startswith("m=audio "):
parts = ln.split(); port = int(parts[1]); pts = [p for p in parts[3:] if p.isdigit()]
return ip, port, pts
def sdp_answer(codec_pt: str, codec_name: str) -> str:
sid = random.randint(1, 10**9)
return (f"v=0\r\no=mock {sid} {sid+1} IN IP4 {MY_IP}\r\ns=mock\r\n"
f"c=IN IP4 {MY_IP}\r\nt=0 0\r\n"
f"m=audio {RTP_PORT} RTP/AVP {codec_pt}\r\n"
f"a=rtpmap:{codec_pt} {codec_name}/8000\r\na=sendrecv\r\n")
# ---------- 呼叫对象 ----------
class Call:
def __init__(self, first, headers, addr):
self.id = hget(headers, "call-id")
self.tag = os.urandom(5).hex()
self.first, self.headers, self.addr = first, headers, addr
self.peer_ip = self.peer_port = None
self.codec = "PCMU"; self.pt = "0"
self.ok_raw = None; self.confirmed = False; self.closed = False
self.sent = self.recv = self.recv_bytes = self.recv_amp_sum = 0
self.t0 = time.time()
self.last_resp = None
calls: dict[str, Call] = {}
rtp_sock = None
def pick_codec(pts):
if "0" in pts: return "0", "PCMU"
if "8" in pts: return "8", "PCMA"
return (pts[0], "PCMU") if pts else ("0", "PCMU")
async def ringing_then_answer(call, sip_t):
await asyncio.sleep(RING_DELAY)
if call.closed: return
body = sdp_answer(call.pt, call.codec)
call.ok_raw = build_resp(call.first, call.headers, 200, "OK", tag=call.tag,
extra=[f"Contact: <sip:mock@{MY_IP}:{SIP_PORT}>",
"User-Agent: MockProvider/1.0"], body=body)
call.last_resp = call.ok_raw
sip_t.sendto(call.ok_raw, call.addr)
log(f"ANSWERED call_id={call.id} media={MY_IP}:{RTP_PORT} codec={call.codec} peer={call.peer_ip}:{call.peer_port}")
start_rtp(call)
for _ in range(5): # 重传 200 OK 直到 ACK (UDP 可靠性)
await asyncio.sleep(1.0)
if call.confirmed or call.closed: break
sip_t.sendto(call.ok_raw, call.addr)
def start_rtp(call):
async def sender():
seq = random.randint(0, 65535); ts = random.randint(0, 10**6)
ssrc = random.randint(1, 2**31)
pt = 0 if call.codec == "PCMU" else 8
t0 = time.time()
while not call.closed:
now = time.time() - t0
payload = tone_payload(call.codec, now)
hdr = struct.pack("!BBHII", 0x80, pt, seq & 0xFFFF, ts & 0xFFFFFFFF, ssrc)
try:
rtp_sock.sendto(hdr + payload, (call.peer_ip, call.peer_port))
except OSError:
pass
call.sent += 1; seq += 1; ts += 160
await asyncio.sleep(0.02)
asyncio.create_task(sender())
def close_call(call, reason):
if call.closed: return
call.closed = True
dur = time.time() - call.t0
avg = call.recv_amp_sum / call.recv if call.recv else 0
log(f"CALL_DONE call_id={call.id} reason={reason} dur={dur:.1f}s "
f"sent_rtp={call.sent} recv_rtp={call.recv} recv_bytes={call.recv_bytes} recv_avg_amplitude={avg:.0f}")
calls.pop(call.id, None)
def on_sip(data, addr, sip_t):
first, headers, body = parse_msg(data)
method = first.split()[0] if first else "?"
cid = hget(headers, "call-id")
if method == "INVITE":
call = calls.get(cid)
if call: # 重传: 回最后一条响应
if call.last_resp: sip_t.sendto(call.last_resp, addr)
return
call = Call(first, headers, addr)
call.peer_ip, call.peer_port, pts = parse_sdp_offer(body)
call.pt, call.codec = pick_codec(pts)
calls[cid] = call
log(f"INCOMING_INVITE from={addr[0]}:{addr[1]} uri=\"{first.split()[1]}\" call_id={cid}")
sip_t.sendto(build_resp(first, headers, 100, "Trying"), addr)
sip_t.sendto(build_resp(first, headers, 180, "Ringing"), addr)
call.last_resp = build_resp(first, headers, 180, "Ringing")
asyncio.create_task(ringing_then_answer(call, sip_t))
elif method == "ACK":
call = calls.get(cid)
if call and not call.confirmed:
call.confirmed = True
log(f"ACK call_id={cid} confirmed, RTP bridging")
elif method == "BYE":
call = calls.get(cid)
sip_t.sendto(build_resp(first, headers, 200, "OK", tag=call.tag if call else None), addr)
if call: close_call(call, "bye")
elif method == "CANCEL":
call = calls.get(cid)
sip_t.sendto(build_resp(first, headers, 487, "Request Terminated",
tag=call.tag if call else None), addr)
if call: close_call(call, "cancelled")
elif method == "OPTIONS":
sip_t.sendto(build_resp(first, headers, 200, "OK", tag="mock"), addr)
class RtpProto(asyncio.DatagramProtocol):
def datagram_received(self, data, addr):
if len(data) < 12: return
pt = data[1] & 0x7F
payload = data[12:]
# 优先 ip:port 精确匹配 (同 IP 多并发呼叫时按 RTP 端口区分), 兜底仅 IP
for call in list(calls.values()):
if not call.closed and call.peer_ip == addr[0] and call.peer_port == addr[1]:
self._tally(call, pt, payload); return
for call in list(calls.values()):
if not call.closed and call.peer_ip == addr[0]:
self._tally(call, pt, payload); return
@staticmethod
def _tally(call, pt, payload):
call.recv += 1; call.recv_bytes += len(payload) + 12
if payload and pt in (0, 8):
dec = [ulaw_decode(b) for b in payload[:40]]
call.recv_amp_sum += sum(abs(x) for x in dec) / len(dec)
class SipProto(asyncio.DatagramProtocol):
def __init__(self): self.t = None
def connection_made(self, transport): self.t = transport
def datagram_received(self, data, addr):
if self.t: on_sip(data, addr, self.t)
async def main():
global rtp_sock
loop = asyncio.get_running_loop()
sip_t, _ = await loop.create_datagram_endpoint(
lambda: SipProto(), local_addr=("0.0.0.0", SIP_PORT))
rtp_t, _ = await loop.create_datagram_endpoint(
RtpProto, local_addr=("0.0.0.0", RTP_PORT))
rtp_sock = rtp_t
log(f"mock provider up: sip=udp/{SIP_PORT} rtp=udp/{RTP_PORT} ip={MY_IP}")
log("waiting for outbound INVITE from asterisk ...")
await asyncio.Event().wait()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
asterisk/deploy/scripts/call.sh
Executable
+18
View File
@@ -0,0 +1,18 @@
#!/bin/bash
# 端到端外呼验证: mock-agent(ARI+RTP) → Asterisk → PJSIP trunk → mock-provider(应答) → 双向 RTP 统计
# 用法: ./call.sh [被叫号码] [时长秒] [asterisk节点 1|2]
set -euo pipefail
cd "$(dirname "$0")/.."
NUMBER="${1:-+15105550123}"
DURATION="${2:-15}"
NODE="${3:-1}"
NET="asterisk-ari_astari"
NAME="mock-agent-$$"
docker run --rm --name "$NAME" --network "$NET" \
-v "$PWD/mock:/app/mock:ro" -w /app \
python:3.12-alpine python mock/mock-agent.py \
--ari-base "http://asterisk${NODE}:8088" \
--external-host "${NAME}:40001" \
--number "$NUMBER" --duration "$DURATION"
asterisk/deploy/scripts/health.sh
Executable
+44
View File
@@ -0,0 +1,44 @@
#!/bin/bash
# 运行状态检测: 容器 + ARI REST + PJSIP trunk 探活(qualify) + 活动呼叫数
set -uo pipefail
USER="outbound"; SECRET="ari_mock_7c3f9e1b5a2d4806"
ok=0; fail=0
pass() { echo "PASS $1"; ok=$((ok+1)); }
faill() { echo "FAIL $1"; fail=$((fail+1)); }
running() { [ -n "$(docker ps -q -f name=$1)" ]; }
echo "== 容器状态"
docker ps --filter "name=ast" --format "{{.Names}}\t{{.Status}}"
echo "== ARI REST"
for n in "ast1 8088" "ast2 8089"; do
set -- $n
if running "$1"; then
c=$(curl -s -o /dev/null -w "%{http_code}" -u "$USER:$SECRET" --max-time 3 "http://127.0.0.1:$2/ari/asterisk/info")
[ "$c" = 200 ] && pass "$1 ARI /asterisk/info ($c)" || faill "$1 ARI /asterisk/info ($c)"
else
echo "SKIP $1 (未运行)"
fi
done
echo "== PJSIP trunk 探活 (OPTIONS qualify)"
for c in ast1 ast2; do
if running "$c"; then
out=$(docker exec "$c" asterisk -rx "pjsip show contacts" 2>/dev/null | grep mock-provider) || true
if echo "$out" | grep -q "Avail"; then
pass "$c trunk mock-provider Avail"
else
faill "$c trunk mock-provider (contact: ${out:-none})"
fi
fi
done
echo "== 活动呼叫 (ARI channels)"
if running ast1; then
n=$(curl -s -u "$USER:$SECRET" --max-time 3 http://127.0.0.1:8088/ari/channels | python3 -c 'import json,sys; print(len(json.load(sys.stdin)))' 2>/dev/null)
[ -n "$n" ] && pass "ast1 active channels=$n" || faill "ast1 active channels"
fi
echo
echo "result: ok=$ok fail=$fail"
[ $fail -eq 0 ]
asterisk/deployment.md
+23
View File
@@ -0,0 +1,23 @@
目标机器：39.106.106.246
登录用户: root
任务目标：在 livekit 运行时下线后，部署 Asterisk 并以 ARI + Mock SIP 实现 SIP 对接与运行状态检测。
项目地址： https://github.com/asterisk/asterisk (容器镜像 andrius/asterisk = Asterisk 22 LTS)
项目背景：使用 Asterisk 对接 MOCK SIP 服务商，作为 AI 外呼机器人基础能力底座，使用 MOCK 对接 LLM ASR 实现双向语音通信, 本项目仅实现外呼功能，不实现多人房间与呼入功能调研。给出可实现最快水平扩展的结论方案并赋实践证据。给出详细的部署过程文档与系统调优方案。
---
## 交付状态 (2026-08-31 全部完成)
| 目标 | 状态 | 证据 |
|---|---|---|
| livekit 运行时下线 | ✅ | lk-* 容器清零, 7880/7883/8081/8082/9091/9092 端口释放 |
| Asterisk 部署 (双节点) | ✅ | ast1/ast2 healthy, ARI /asterisk/info 200 ×2 (Asterisk 22.10.1) |
| Mock SIP 对接 (外呼) | ✅ | INVITE→180→200 OK(PCMU)→ACK, trunk qualify Avail ×2 |
| 双向语音 (Mock LLM/ASR) | ✅ | `RESULT ... bidirectional=YES` (agent RX 512帧 RMS 8318; provider recv_rtp=512, 双向计数完全对称) |
| 运行状态检测 | ✅ | `scripts/health.sh` 容器+ARI REST+trunk qualify+通道数, ok=5 fail=0 |
| 水平扩展结论+实证 | ✅ | 3 路并发分摊双节点 (INVITE 源 172.18.0.2×2 / 172.18.0.4×1), 零配置变更 |
| 部署文档+调优方案 | ✅ | 本仓库 `README.md` (§4 部署含踩坑, §8 sysctl 沿用已应用) |
**交付物**: 服务器 `/opt/asterisk-ari/` (compose+conf+mock+scripts+README), 本地 `deploy/` 同源; 原始任务与完成状态见本文件, 详细文档见 `README.md`。
livekit/README.md
+198
View File
@@ -0,0 +1,198 @@
# LiveKit + livekit-sip AI 外呼底座 — 部署与运维文档
> 目标机器: `39.106.106.246` (阿里云 2vCPU/3.5G, Debian 13) · 交付物: `/opt/livekit-sip/`
> 项目: https://github.com/livekit/livekit · 插件: https://github.com/livekit/sip
> 范围: **仅外呼** (CreateSIPParticipant 工作流), Mock SIP 运营商 + Mock LLM/ASR, 已端到端验证双向语音。
## 1. 水平扩展结论 (先行)
**结论: 共享 Redis 的无状态水平复制 —— livekit-server 与 livekit-sip 均不在本机存呼叫状态, 扩容 = 加副本, 无需协调器/迁移。**
| 层 | 状态存放 | 扩容方式 | 调度机制 |
|---|---|---|---|
| livekit-sip 网关 | Redis (trunk/呼叫状态) + 各自持有 RTP 流 | 复制 service 块 (独立端口段) | CreateSIPParticipant 经 Redis **随机派发**到任一网关节点 |
| livekit-server | Redis (rooms/room_node_map/nodes) | 加节点, 客户端连任一节点 | 房间归属节点经 Redis 自动路由, 跨节点媒体互通 |
| AI Agent (生产) | 无状态 worker | `--scale`/K8s HPA | AgentDispatch 按 agent_name 派发 |
**最快扩容路径**: `docker compose` 里复制一个 `sipN` service 块 (改宿主端口映射即可, 容器网内 5060/10000-10200 天然隔离); K8s 场景用 Deployment + NodePort/hostPort 固定 SIP+RTP 端口段。实测从 1 网关扩到 2 网关**零配置变更** (共用同一 `sip.yaml`), 呼叫自动分摊 (见 §7)。
瓶颈与边界: 单网关受 CPU (RTP 打包/转发) 与 RTP 端口段限制; 本文配置 200 端口/网关, 实际每呼叫占 1 个 UDP mux 端口, 单节点并发上限≈端口数。LiveKit 官方口径单节点(4C8G)数百路通话; 水平扩展后容量线性叠加, 上限在 Redis 与出口带宽。
## 2. 架构
```
┌────────────────────────── docker net: lksip ──────────────────────────┐
lk CLI / 业务后端 │ ┌──────────┐ ws ┌─────────────┐ ┌─────────────┐ │
(host, :7880 API) ───────┼─►│ livekit │◄───────►│ livekit2 │ │ redis │ │
CreateSIPParticipant │ │ :7880 │ Redis │ :7883(multinode)│ │ :6379 │ │
│ trunk/呼叫状态 │ └────┬─────┘ 集群 └─────────────┘ └──────┬──────┘ │
▼ 全部经 Redis │ │ RTC(ICE 50100-50200/udp) │ 状态存储 │
┌─────────────┐ │ ┌────▼──────────┐ ┌──────────────┐ │ │
│ redis │◄─────────┼──┤ sip1 (网关) │ │ sip2 (网关) │──► 随机被派发 CreateSIPParticipant│
└─────────────┘ │ │ 5060/udp+tcp │ │ 5060/udp+tcp │ │ │
│ │ RTP 10000- │ │ RTP 10000- │ │ │
mock-agent (host) │ │ 10200/udp │ │ 10200/udp │ │ │
订阅=Mock ASR │ └──────┬───────┘ └──────┬───────┘ │ │
发布=Mock LLM/TTS ───────┼─────────┘ SIP INVITE / RTP(PCMU) │ │
│ └──────────►┌────────────────┐ │ │
│ │ mock-provider │──────┘ (SIP UAS, 440Hz 音) │
│ │ 5060/udp 40000 │ 纯 stdlib Python │
│ └────────────────┘ │
└───────────────────────────────────────────────────────────────────────┘
```
外呼信令流 (仅出站):
```
业务后端 → lk sip participant create --trunk ST_xxx --call +1510...
→ LiveKit API → Redis 派发到某 sip 网关
→ INVITE sip:+1510...@mock-provider → 100/180 → 200 OK(SDP, PCMU) → ACK
→ 被叫以 SIP participant 身份进房, 与房间内 agent(=AI) 双向 RTP
```
## 3. 端口规划
| 端口 | 组件 | 说明 |
|---|---|---|
| 7880/tcp | livekit (node1) | HTTP/WS API, 宿主映射 7880 |
| 7883/tcp→7880 | livekit2 (node2) | 同一 Redis, 双节点集群 |
| 50100-50200/udp | livekit ×2 | RTC ICE (各容器独立 netns 不冲突) |
| 5060/udp+tcp | sip1/sip2 | SIP 信令 (容器网内, 未映射宿主) |
| 10000-10200/udp | sip1/sip2 | RTP 媒体 (同上) |
| 8081/8082 | sip1/sip2 health | `/healthz` + `/` (HTTP 200) |
| 9091/9092 | sip1/sip2 prometheus | `/metrics` |
| 5060/udp, 40000/udp | mock-provider | Mock 运营商 SIP+RTP |
生产外网仅需放行: SIP 5060/udp+tcp (+5061/tls) 与 RTP 段; LiveKit 客户端走 7880 + RTC 段。
## 4. 从零部署步骤 (全部实测)
```bash
# ── 4.1 Docker (Debian 13, 官方源; 国内需配 mirror) ─────────────────────
apt-get update && apt-get install -y ca-certificates curl gnupg
install -m 0755 -d /etc/apt/keyrings
curl -fsSL https://download.docker.com/linux/debian/gpg -o /etc/apt/keyrings/docker.asc
echo "deb [arch=amd64 signed-by=/etc/apt/keyrings/docker.asc] https://download.docker.com/linux/debian trixie stable" \
> /etc/apt/sources.list.d/docker.list
apt-get update && apt-get install -y docker-ce docker-ce-cli containerd.io docker-compose-plugin
# ⚠ 踩坑1(Debian 13): 安装后 docker group 缺失 → docker.socket 216/GROUP; 需 groupadd docker
# ⚠ 踩坑2(最小系统): 无 iptables → bridge 驱动失败 "iptables not found"; 需 apt install iptables nftables
cat > /etc/docker/daemon.json <<'EOF'
{ "registry-mirrors": ["https://docker.m.daocloud.io", "https://docker.1ms.run"] }
EOF
systemctl restart docker
# ── 4.2 内核调优 (见 §8, 已应用 /etc/sysctl.d/99-livekit-sip.conf) ─────
# ── 4.3 部署栈 ─────────────────────────────────────────────────────────
mkdir -p /opt/livekit-sip && cd /opt/livekit-sip # = 仓库 deploy/ 目录
docker compose up -d # redis + livekit + sip1 + sip2 + mock-provider
docker compose --profile multinode up -d livekit2 # (可选) LiveKit 第2节点
# ── 4.4 Mock agent 依赖 (host python) ──────────────────────────────────
apt-get install -y python3-venv
python3 -m venv /opt/lk-venv
/opt/lk-venv/bin/pip install -i https://mirrors.aliyun.com/pypi/simple/ livekit livekit-api
# ⚠ 踩坑3: SDK 1.x 拆包 — rtc 在 `livekit`, AccessToken 在 `livekit-api`;
# grants 必须传 VideoGrants() 对象; frame.data 是 buffer 非 bytes。
# ── 4.5 建 trunk + 发起外呼 ────────────────────────────────────────────
./scripts/call.sh <房间> <被叫E.164> [时长秒] # 首次自动创建 trunk 并缓存于 Redis
```
trunk 定义 (`mock/trunk.json`, 换真实运营商时改 address/auth):
```json
{ "trunk": { "name": "mock-provider", "address": "mock-provider", "numbers": ["+15105550123"] } }
```
## 5. 外呼链路验证 (实测输出)
`./scripts/call.sh test-room-5 +15105550123 12`:
```
== 2. Mock agent 加入房间 test-room-5 (发布TTS音, 订阅被叫音频)
[agent] joined room=test-room-5 peers=[]
[agent] publishing 440Hz tone (500ms on / 500ms off)
== 3. 发起 SIP 外呼 -> +15105550123
[agent] subscribed: callee-30395 track=TR_AMu2wATzwBLwqP
SIPCallID: SCL_LEGvhUjpz3N9 ← CreateSIPParticipant 成功, 被叫应答进房
== 4. 等待 agent 结束并输出双向统计
[agent] RX frames=1000 avg_rms=4667
RESULT room=test-room-5 rx_frames=1015 rx_avg_rms=4640 bidirectional=YES
```
mock-provider (被叫侧) 日志:
```
INCOMING_INVITE from=172.18.0.6:5060 uri="sip:+15105550123@mock-provider"
ANSWERED media=172.18.0.3:40000 codec=PCMU peer=172.18.0.6:10034 ← SDP 协商 PCMU
ACK confirmed, RTP bridging
CALL_DONE sent_rtp=… recv_rtp=… recv_avg_amplitude=… ← 双向 RTP 计数
```
**双向语音证明**: agent RX=被叫→ASR 方向 (1015 帧, RMS 4640 = 440Hz 检测到); provider recv = LLM/TTS→被叫方向 (RTP 计数>0)。换真实 LLM/ASR 只需替换 mock-agent 内的发布/消费循环 (livekit-agents 框架同接口)。
## 6. 运行状态检测
`./scripts/health.sh` (实测输出):
```
== 容器状态 lk-livekit2/lk-sip1/lk-sip2/lk-livekit/lk-redis/lk-mock-provider Up
== HTTP 健康 PASS livekit(200) PASS sip1(200) PASS sip2(200) # sip /healthz
== API 功能 PASS lk room list PASS lk sip outbound list
result: ok=5 fail=0
```
持续监控: `watch -n5 ./scripts/health.sh`; 指标: `curl :9091/metrics`, `curl :9092/metrics` (prometheus, 含活动呼叫/错误); 呼叫级状态: provider 容器日志 `CALL_DONE` 行; 房间级: `lk room list`。
## 7. 水平扩展实证 (同机双网关 + 双 LiveKit 节点)
**① 双 LiveKit 节点集群 (共 Redis, 自动发现)**
```
$ docker exec lk-redis redis-cli hlen nodes
2 ← node1 ND_BwShgqdiB8AQ + node2 ND_moj9RpSqq5Ap
```
并发验证中 agent 全部连 **node2(:7883)**, CreateSIPParticipant 经 **node1(:7880)**, 同房媒体互通 → 跨节点路由成立 (RESULT 见 ③)。
**② 双 SIP 网关随机分摊 (无状态验证)**
```
$ docker logs lk-sip1 | grep -oE '"room": "burst-[0-9]"' | sort | uniq -c
15 "room": "burst-1" ← sip1 处理 burst-1
$ docker logs lk-sip2 | ...
15 "room": "burst-2" ← sip2 处理 burst-2
15 "room": "burst-3" ← 及 burst-3
$ docker logs lk-mock-provider | grep INCOMING | tail -3
from=172.18.0.5:5060 call_id=lIIip… ← 两个来源 IP = sip1/sip2
from=172.18.0.6:5060 call_id=mw4Fq…
from=172.18.0.5:5060 call_id=BCVgl…
```
**③ 3 路并发外呼全部双向 (agent 连 node2)**
```
RESULT room=burst-1 rx_frames=810 rx_avg_rms=3162 bidirectional=YES
RESULT room=burst-2 rx_frames=866 rx_avg_rms=3066 bidirectional=YES
RESULT room=burst-3 rx_frames=778 rx_avg_rms=2228 bidirectional=YES
```
→ 加副本即扩容、Redis 自动调度、跨节点房间互通, 三项均实证。资源基线 (3 并发): livekit 2×6%CPU/36M, sip 2×20%CPU/30M, redis 1%/8M。
## 8. 系统调优
**已应用** (`/etc/sysctl.d/99-livekit-sip.conf`, `sysctl --system` 生效):
```conf
net.core.rmem_max = 16777216 # UDP 收缓冲上限 (pion/WebRTC 推荐 16M; 默认 208K 高并发必丢包)
net.core.wmem_max = 16777216
net.core.rmem_default = 1048576
net.core.wmem_default = 1048576
net.core.netdev_max_backlog = 4096 # 网卡积压队列 (默认 1000)
net.ipv4.ip_local_port_range = 10000 65000 # 出向 SIP/RTP 源端口 (含 RTP 段)
net.ipv4.udp_mem = 8388608 12582912 16777216 # 全局 UDP 页缓存 (3.5G 内存安全值)
net.ipv4.udp_rmem_min/wmem_min = 16384
```
调优前基线: `UdpRcvbufErrors=0` (3 并发, 默认缓冲已够); 上述参数为高并发 (>100 路) 预置。验证命令: `nstat -az | grep -i udp`。
**组件级**:
- sip 网关: `rtp_port: 10000-10200` (每节点 200 并发余量); `max_cpu_utilization: 0.9` 默认即可, 超 90% 自动拒绝新呼叫 (背压); `log_level: info` 生产勿用 debug。
- livekit: RTC 段 50100-50200/节点; 房间自动关闭 (empty timeout) 回收端口; `logging.json: true` 便于采集。
- redis: 纯缓存用法 (`--save ""`), trunk/房间状态可丢 (重建即恢复); 若把 Redis 当持久层需开 AOF。
- 容器: `restart: unless-stopped` 已配; 单机 3.5G 下建议给 livekit/sip 设 mem limit (如 512M) 防互挤。
- 监控: 9091/9092 /metrics 接 Prometheus; 告警项: sip 活动呼叫数骤降、livekit 5xx、UdpRcvbufErrors>0、容器重启。
## 9. 生产化清单 (Mock → 真实运营商)
1. trunk: `address` 换运营商 SIP 域名/IP, 加 `auth_username/auth_password`, 号码用 E.164; 我方 5060/udp+tcp、RTP 段需公网放行 (阿里云安全组)。
2. NAT: 阿里云 ECS 为 1:1 NAT, sip.yaml 改 `use_external_ip: true` (STUN 自动探测公网 IP 写入 SDP), livekit.yaml 同理; 或显式 `nat_1_to_1_ip`。
3. 安全: SIP over TLS (`tls.port: 5061`) + SRTP (`media_enc`), 或 IP 白名单 ACL; LiveKit API key 轮换。
4. AI 侧: mock-agent 替换为 livekit-agents 框架 (同房间/同 track 接口), STT/LLM/TTS 三段插拔。
5. 高可用: Redis 单点 → 云 Redis 主从; 网关 ≥2 跨宿主机; LiveKit 节点 ≥2 (本文已验证集群模式)。
livekit/deploy/conf/livekit.yaml
+15
View File
@@ -0,0 +1,15 @@
# LiveKit Server 配置 — 单机双节点共享 Redis 组成集群
port: 7880
bind_addresses: [""]
redis:
address: redis:6379
keys:
devkey: "f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04"
rtc:
tcp_port: 7881
port_range_start: 50100
port_range_end: 50200
use_external_ip: false
logging:
level: info
json: true
livekit/deploy/conf/sip.yaml
+12
View File
@@ -0,0 +1,12 @@
# livekit-sip 网关配置 — sip1/sip2 共用(各自独立 netns, 端口不冲突)
api_key: devkey
api_secret: "f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04"
ws_url: ws://livekit:7880
redis:
address: redis:6379
health_port: 8081
prometheus_port: 9091
log_level: info
sip_port: 5060
rtp_port: 10000-10200
use_external_ip: false
livekit/deploy/docker-compose.yml
+75
View File
@@ -0,0 +1,75 @@
name: livekit-sip
services:
redis:
image: redis:7-alpine
container_name: lk-redis
command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru --save "" --appendonly no
restart: unless-stopped
networks: [lksip]
livekit:
image: livekit/livekit-server:latest
container_name: lk-livekit
command: --config /etc/livekit.yaml
volumes:
- ./conf/livekit.yaml:/etc/livekit.yaml:ro
ports: ["7880:7880"]
restart: unless-stopped
networks: [lksip]
depends_on: [redis]
# 第二个 LiveKit 节点: 同一 Redis 即自动组成多节点集群(水平扩展验证)
livekit2:
image: livekit/livekit-server:latest
container_name: lk-livekit2
command: --config /etc/livekit.yaml
volumes:
- ./conf/livekit.yaml:/etc/livekit.yaml:ro
ports: ["7883:7880"]
restart: unless-stopped
networks: [lksip]
depends_on: [redis]
profiles: [multinode]
# SIP 网关 x2: 无状态, 经 Redis 被随机调度, 天然水平扩展
sip1:
image: livekit/sip:latest
container_name: lk-sip1
command: --config /etc/sip.yaml
volumes:
- ./conf/sip.yaml:/etc/sip.yaml:ro
ports: ["8081:8081", "9091:9091"]
restart: unless-stopped
networks: [lksip]
depends_on: [redis, livekit]
sip2:
image: livekit/sip:latest
container_name: lk-sip2
command: --config /etc/sip.yaml
volumes:
- ./conf/sip.yaml:/etc/sip.yaml:ro
ports: ["8082:8081", "9092:9091"]
restart: unless-stopped
networks: [lksip]
depends_on: [redis, livekit]
# Mock SIP 运营商 (UAS): 应答 livekit-sip 的出站 INVITE
mock-provider:
image: python:3.12-alpine
container_name: lk-mock-provider
working_dir: /app
command: python mock/mock-provider.py
volumes:
- ./mock:/app/mock:ro
environment:
SIP_PORT: "5060"
RTP_PORT: "40000"
RING_DELAY: "1.5"
restart: unless-stopped
networks: [lksip]
networks:
lksip:
driver: bridge
livekit/deploy/mock/mock-agent.py
+78
View File
@@ -0,0 +1,78 @@
#!/usr/bin/env python3
"""Mock AI Agent: 加入房间, 发布 440Hz 间歇音(模拟 TTS 输出),
订阅 SIP 参与者音频并统计 RMS(模拟 ASR 输入), 验证双向语音。
依赖: pip install livekit
"""
import argparse, array, asyncio, math, struct, time
from livekit import rtc
from livekit.api import AccessToken, VideoGrants
def rms(data) -> float:
buf = bytes(data)[:4096]
a = array.array("h"); a.frombytes(buf)
if not a: return 0.0
return (sum(x * x for x in a) / len(a)) ** 0.5
async def main():
ap = argparse.ArgumentParser()
ap.add_argument("--url", default="ws://127.0.0.1:7880")
ap.add_argument("--api-key", default="devkey")
ap.add_argument("--api-secret", default="f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04")
ap.add_argument("--room", required=True)
ap.add_argument("--identity", default="mock-agent")
ap.add_argument("--duration", type=float, default=20)
args = ap.parse_args()
token = (AccessToken(args.api_key, args.api_secret)
.with_identity(args.identity).with_name("MockAgent")
.with_grants(VideoGrants(room_join=True, room=args.room,
can_publish=True, can_subscribe=True)))
room = rtc.Room()
stats = {"frames": 0, "rms_sum": 0.0, "src": None}
done = asyncio.Event()
def on_sub(track, pub, participant):
if track.kind != rtc.TrackKind.KIND_AUDIO:
return
print(f"[agent] subscribed: {participant.identity} track={track.sid}", flush=True)
async def consume():
stream = rtc.AudioStream(track)
async for ev in stream:
f = ev.frame
stats["frames"] += 1
stats["rms_sum"] += rms(f.data)
if stats["frames"] % 250 == 0:
print(f"[agent] RX frames={stats['frames']} avg_rms={stats['rms_sum']/stats['frames']:.0f}", flush=True)
stats["src"] = asyncio.create_task(consume())
room.on("track_subscribed", on_sub)
await room.connect(args.url, token.to_jwt())
peers = [p.identity for p in room.remote_participants.values()]
print(f"[agent] joined room={args.room} peers={peers}", flush=True)
src = rtc.AudioSource(48000, 1)
track = rtc.LocalAudioTrack.create_audio_track("agent-tts", src)
await room.local_participant.publish_track(track)
print("[agent] publishing 440Hz tone (500ms on / 500ms off)", flush=True)
async def tone():
t, end = 0.0, time.time() + args.duration
while time.time() < end:
samples = []
on = (t % 1.0) < 0.5
for _ in range(960): # 20ms @ 48kHz
v = int(6000 * math.sin(2 * math.pi * 440 * t)) if on else 0
samples.append(v); t += 1 / 48000
buf = struct.pack("<960h", *samples)
await src.capture_frame(rtc.AudioFrame(buf, 48000, 1, 960))
await asyncio.sleep(0.02)
await tone()
n, r = stats["frames"], (stats["rms_sum"] / stats["frames"]) if stats["frames"] else 0.0
print(f"RESULT room={args.room} rx_frames={n} rx_avg_rms={r:.0f} "
f"bidirectional={'YES' if n > 50 and r > 100 else 'NO'}", flush=True)
await room.disconnect()
if __name__ == "__main__":
asyncio.run(main())
livekit/deploy/mock/mock-provider.py
+237
View File
@@ -0,0 +1,237 @@
#!/usr/bin/env python3
"""Mock SIP 运营商 (UAS): 接收 livekit-sip 出站 INVITE, 振铃->应答, 双向收发 PCMU/PCMA RTP.
仅 stdlib, 无依赖。一个 RTP 端口复用所有呼叫(按来源 IP 匹配呼叫)。
用法: python3 mock-provider.py (环境变量 SIP_PORT/RTP_PORT/RING_DELAY)
"""
import asyncio, math, os, random, socket, struct, time
SIP_PORT = int(os.environ.get("SIP_PORT", "5060"))
RTP_PORT = int(os.environ.get("RTP_PORT", "40000"))
RING_DELAY = float(os.environ.get("RING_DELAY", "1.5"))
MY_IP = socket.gethostbyname(socket.gethostname())
def log(*a): print(f"[provider {time.strftime('%H:%M:%S')}]", *a, flush=True)
# ---------- G.711 编码 ----------
def ulaw_encode(s: int) -> int:
s = max(-32635, min(32635, s)); sign = 0x80 if s < 0 else 0
if sign: s = -s
s += 0x84
e = 7
for i in range(7, -1, -1):
if s & (0x40 << i): e = i; break
return (~(sign | (e << 4) | ((s >> (e + 3)) & 0x0F))) & 0xFF
def alaw_encode(s: int) -> int:
s = max(-32635, min(32635, s)); sign = 0x80 if s < 0 else 0
if sign: s = -s
e = 7
for i in range(7, -1, -1):
if s & (0x40 << i): e = i; break
m = (s >> (e + 3)) & 0x0F
return ((sign | (e << 4) | m) ^ 0x55) & 0xFF
def ulaw_decode(u: int) -> int:
u = ~u & 0xFF
sign = u & 0x80; e = (u >> 4) & 0x07; m = u & 0x0F
s = ((m << (e + 3)) + 0x84) << 2
return -s if sign else s
def tone_payload(codec: str, ts: float) -> bytes:
"""20ms / 160 samples @8kHz, 440Hz 500ms-on 500ms-off"""
out = bytearray()
on = (ts % 1.0) < 0.5
for i in range(160):
t = ts + i / 8000.0
v = int(6000 * math.sin(2 * math.pi * 440 * t)) if on else 0
out.append(ulaw_encode(v) if codec == "PCMU" else alaw_encode(v))
return bytes(out)
# ---------- SIP 报文 ----------
def parse_msg(data: bytes):
head, _, body = data.decode("utf-8", "replace").partition("\r\n\r\n")
lines = head.split("\r\n")
headers = []
for ln in lines[1:]:
n, _, v = ln.partition(":")
headers.append((n.strip().lower(), v.strip()))
return lines[0], headers, body
def hget(headers, name):
for n, v in headers:
if n == name: return v
return None
def build_resp(first, headers, code, reason, tag=None, extra=None, body=""):
out = [f"SIP/2.0 {code} {reason}"]
for n, v in headers:
if n == "via": out.append(f"Via: {v}")
out.append(f"From: {hget(headers,'from')}")
to = hget(headers, "to")
if tag and "tag=" not in to: to = f"{to};tag={tag}"
out.append(f"To: {to}")
out.append(f"Call-ID: {hget(headers,'call-id')}")
out.append(f"CSeq: {hget(headers,'cseq')}")
if extra:
out += extra
if body:
out.append(f"Content-Type: application/sdp")
out.append(f"Content-Length: {len(body)}")
return ("\r\n".join(out) + "\r\n\r\n" + body).encode()
def parse_sdp_offer(body: str):
"""返回 (对端媒体IP, 对端媒体端口, payload类型列表)"""
ip, port, pts = None, None, []
for ln in body.splitlines():
if ln.startswith("c=IN IP4 "): ip = ln.split()[2]
elif ln.startswith("m=audio "):
parts = ln.split(); port = int(parts[1]); pts = [p for p in parts[3:] if p.isdigit()]
return ip, port, pts
def sdp_answer(codec_pt: str, codec_name: str) -> str:
sid = random.randint(1, 10**9)
return (f"v=0\r\no=mock {sid} {sid+1} IN IP4 {MY_IP}\r\ns=mock\r\n"
f"c=IN IP4 {MY_IP}\r\nt=0 0\r\n"
f"m=audio {RTP_PORT} RTP/AVP {codec_pt}\r\n"
f"a=rtpmap:{codec_pt} {codec_name}/8000\r\na=sendrecv\r\n")
# ---------- 呼叫对象 ----------
class Call:
def __init__(self, first, headers, addr):
self.id = hget(headers, "call-id")
self.tag = os.urandom(5).hex()
self.first, self.headers, self.addr = first, headers, addr
self.peer_ip = self.peer_port = None
self.codec = "PCMU"; self.pt = "0"
self.ok_raw = None; self.confirmed = False; self.closed = False
self.sent = self.recv = self.recv_bytes = self.recv_amp_sum = 0
self.t0 = time.time()
self.last_resp = None
calls: dict[str, Call] = {}
rtp_sock = None
def pick_codec(pts):
if "0" in pts: return "0", "PCMU"
if "8" in pts: return "8", "PCMA"
return (pts[0], "PCMU") if pts else ("0", "PCMU")
async def ringing_then_answer(call, sip_t):
await asyncio.sleep(RING_DELAY)
if call.closed: return
body = sdp_answer(call.pt, call.codec)
call.ok_raw = build_resp(call.first, call.headers, 200, "OK", tag=call.tag,
extra=[f"Contact: <sip:mock@{MY_IP}:{SIP_PORT}>",
"User-Agent: MockProvider/1.0"], body=body)
call.last_resp = call.ok_raw
sip_t.sendto(call.ok_raw, call.addr)
log(f"ANSWERED call_id={call.id} media={MY_IP}:{RTP_PORT} codec={call.codec} peer={call.peer_ip}:{call.peer_port}")
start_rtp(call)
for _ in range(5): # 重传 200 OK 直到 ACK (UDP 可靠性)
await asyncio.sleep(1.0)
if call.confirmed or call.closed: break
sip_t.sendto(call.ok_raw, call.addr)
def start_rtp(call):
async def sender():
seq = random.randint(0, 65535); ts = random.randint(0, 10**6)
ssrc = random.randint(1, 2**31)
pt = 0 if call.codec == "PCMU" else 8
t0 = time.time()
while not call.closed:
now = time.time() - t0
payload = tone_payload(call.codec, now)
hdr = struct.pack("!BBHII", 0x80, pt, seq & 0xFFFF, ts & 0xFFFFFFFF, ssrc)
try:
rtp_sock.sendto(hdr + payload, (call.peer_ip, call.peer_port))
except OSError:
pass
call.sent += 1; seq += 1; ts += 160
await asyncio.sleep(0.02)
asyncio.create_task(sender())
def close_call(call, reason):
if call.closed: return
call.closed = True
dur = time.time() - call.t0
avg = call.recv_amp_sum / call.recv if call.recv else 0
log(f"CALL_DONE call_id={call.id} reason={reason} dur={dur:.1f}s "
f"sent_rtp={call.sent} recv_rtp={call.recv} recv_bytes={call.recv_bytes} recv_avg_amplitude={avg:.0f}")
calls.pop(call.id, None)
def on_sip(data, addr, sip_t):
first, headers, body = parse_msg(data)
method = first.split()[0] if first else "?"
cid = hget(headers, "call-id")
if method == "INVITE":
call = calls.get(cid)
if call: # 重传: 回最后一条响应
if call.last_resp: sip_t.sendto(call.last_resp, addr)
return
call = Call(first, headers, addr)
call.peer_ip, call.peer_port, pts = parse_sdp_offer(body)
call.pt, call.codec = pick_codec(pts)
calls[cid] = call
log(f"INCOMING_INVITE from={addr[0]}:{addr[1]} uri=\"{first.split()[1]}\" call_id={cid}")
sip_t.sendto(build_resp(first, headers, 100, "Trying"), addr)
sip_t.sendto(build_resp(first, headers, 180, "Ringing"), addr)
call.last_resp = build_resp(first, headers, 180, "Ringing")
asyncio.create_task(ringing_then_answer(call, sip_t))
elif method == "ACK":
call = calls.get(cid)
if call and not call.confirmed:
call.confirmed = True
log(f"ACK call_id={cid} confirmed, RTP bridging")
elif method == "BYE":
call = calls.get(cid)
sip_t.sendto(build_resp(first, headers, 200, "OK", tag=call.tag if call else None), addr)
if call: close_call(call, "bye")
elif method == "CANCEL":
call = calls.get(cid)
sip_t.sendto(build_resp(first, headers, 487, "Request Terminated",
tag=call.tag if call else None), addr)
if call: close_call(call, "cancelled")
elif method == "OPTIONS":
sip_t.sendto(build_resp(first, headers, 200, "OK", tag="mock"), addr)
class RtpProto(asyncio.DatagramProtocol):
def datagram_received(self, data, addr):
if len(data) < 12: return
pt = data[1] & 0x7F
payload = data[12:]
for call in list(calls.values()):
if not call.closed and call.peer_ip == addr[0]:
call.recv += 1; call.recv_bytes += len(data)
if payload and pt in (0, 8):
dec = [ulaw_decode(b) for b in payload[:40]]
call.recv_amp_sum += sum(abs(x) for x in dec) / len(dec)
return
class SipProto(asyncio.DatagramProtocol):
def __init__(self): self.t = None
def connection_made(self, transport): self.t = transport
def datagram_received(self, data, addr):
if self.t: on_sip(data, addr, self.t)
async def main():
global rtp_sock
loop = asyncio.get_running_loop()
sip_t, _ = await loop.create_datagram_endpoint(
lambda: SipProto(), local_addr=("0.0.0.0", SIP_PORT))
rtp_t, _ = await loop.create_datagram_endpoint(
RtpProto, local_addr=("0.0.0.0", RTP_PORT))
rtp_sock = rtp_t
log(f"mock provider up: sip=udp/{SIP_PORT} rtp=udp/{RTP_PORT} ip={MY_IP}")
log("waiting for outbound INVITE from livekit-sip ...")
await asyncio.Event().wait()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
livekit/deploy/mock/trunk.json
+7
View File
@@ -0,0 +1,7 @@
{
"trunk": {
"name": "mock-provider",
"address": "mock-provider",
"numbers": ["+15105550123"]
}
}
livekit/deploy/scripts/call.sh
+38
View File
@@ -0,0 +1,38 @@
#!/bin/bash
# 端到端外呼验证: agent 进房 -> CreateSIPParticipant 呼出 -> mock 运营商应答 -> 双向 RTP 统计
# 用法: ./call.sh [房间名] [被叫号码] [时长秒]
set -euo pipefail
cd "$(dirname "$0")/.."
ROOM="${1:-call-$(date +%s)}"
NUMBER="${2:-+15105550123}"
DURATION="${3:-15}"
SECRET="f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04"
LK="docker run --rm --network host -e LIVEKIT_URL=http://127.0.0.1:7880 \
-e LIVEKIT_API_KEY=devkey -e LIVEKIT_API_SECRET=$SECRET livekit/livekit-cli:latest"
echo "== 1. 确保 outbound trunk 存在 (mock-provider)"
TRUNK_ID=$($LK sip outbound list --json 2>/dev/null | python3 -c '
import json,sys
for t in json.load(sys.stdin).get("items", []):
if t.get("name") == "mock-provider":
print(t.get("sipTrunkId","")); break')
if [ -z "$TRUNK_ID" ]; then
TRUNK_ID=$($LK sip outbound create mock/trunk.json 2>/dev/null | awk '/^SIPTrunkID:/ {print $2}')
echo " created trunk: $TRUNK_ID"
else
echo " reuse trunk: $TRUNK_ID"
fi
echo "== 2. Mock agent 加入房间 $ROOM (发布TTS音, 订阅被叫音频)"
/opt/lk-venv/bin/python mock/mock-agent.py --room "$ROOM" --duration "$DURATION" &
AGENT_PID=$!
sleep 2
echo "== 3. 发起 SIP 外呼 -> $NUMBER"
$LK sip participant create --room "$ROOM" --identity "callee-$RANDOM" \
--trunk "$TRUNK_ID" --number "$NUMBER" --call "$NUMBER" \
--name "MockCallee" --wait --timeout 60s || true
echo "== 4. 等待 agent 结束并输出双向统计"
wait $AGENT_PID
livekit/deploy/scripts/health.sh
+32
View File
@@ -0,0 +1,32 @@
#!/bin/bash
# 运行状态检测: 各组件健康 + 集群节点 + trunk 配置
set -uo pipefail
SECRET="f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04"
ok=0; fail=0
chk() { # name, command...
local name="$1"; shift
if out=$("$@" 2>&1); then echo "PASS $name"; ok=$((ok+1))
else echo "FAIL $name: $(echo "$out" | head -1)"; fail=$((fail+1)); fi
}
echo "== 容器状态"
docker ps --filter "name=lk-" --format "{{.Names}}\t{{.Status}}"
echo "== HTTP 健康"
code() { curl -s -o /dev/null -w "%{http_code}" --max-time 3 "$1"; }
for p in "livekit http://127.0.0.1:7880/" "sip1 http://127.0.0.1:8081/healthz" "sip2 http://127.0.0.1:8082/healthz"; do
set -- $p
c=$(code "$2")
[ "$c" = "200" ] || [ "$c" = "204" ] && { echo "PASS $1 ($c)"; ok=$((ok+1)); } || { echo "FAIL $1 ($c)"; fail=$((fail+1)); }
done
echo "== LiveKit API 功能检查"
LK=(docker run --rm --network host -e LIVEKIT_URL=http://127.0.0.1:7880
-e LIVEKIT_API_KEY=devkey -e LIVEKIT_API_SECRET=$SECRET livekit/livekit-cli:latest)
chk "lk room list" "${LK[@]}" room list
chk "lk sip outbound list" "${LK[@]}" sip outbound list
echo
echo "result: ok=$ok fail=$fail"
[ $fail -eq 0 ]
livekit/deployment.md
+24
View File
@@ -0,0 +1,24 @@
目标机器：39.106.106.246
登录用户: root
任务目标：部署 Livekit 并融合 livekit-sip 插件使用Mock实现SIP对接与运行状态检测。
项目地址： https://github.com/livekit/livekit
插件地址：
项目背景：使用livekit对接MOCK SIP服务商，作为AI外呼机器人基础能力底座，使用MOCK对接LLM ASR实现双向语音通信, 本项目仅实现外呼功能，不实现多人房间与呼入功能调研。给出可实现最快水平扩展的结论方案并赋实践证据。给出详细的部署过程文档与系统调优方案。
---
## 交付状态 (2026-08-31 全部完成)
| 目标 | 状态 | 证据 |
|---|---|---|
| LiveKit 部署 (双节点集群) | ✅ | Redis `hlen nodes` = 2; health.sh ok=5 fail=0 |
| livekit-sip 网关 ×2 融合 | ✅ | sip1/sip2 `/healthz` 200, 共 Redis 随机派发 |
| Mock SIP 对接 (外呼) | ✅ | INVITE→180→200 OK(PCMU)→ACK, mock-provider 全 stdlib |
| 双向语音 (Mock LLM/ASR) | ✅ | `RESULT ... bidirectional=YES` (agent RX 1015帧 RMS 4640; provider 双向 RTP 计数) |
| 运行状态检测 | ✅ | `scripts/health.sh` 容器+HTTP+API 三层检查 |
| 水平扩展结论+实证 | ✅ | 3 路并发分摊到双网关 (172.18.0.5/0.6 双源 INVITE), agent 连 node2 / 呼叫经 node1 跨节点互通 |
| 部署文档+调优方案 | ✅ | 本仓库 `README.md` (§4 部署含踩坑, §8 sysctl 已应用) |
**交付物**: 服务器 `/opt/livekit-sip/` (compose+conf+mock+scripts+README), 本地 `deploy/` 同源; 原始任务与完成状态见本文件, 详细文档见 `README.md`。