在语音识别（ASR）接入开发中，一个“足够真实”的 Mock Server 可以极大提升开发效率。
尤其当你需要：

• 调试 streaming 行为

• 验证 client timeout / retry

• 对比不同 ASR 引擎（Google / Verbio / AWS）

• 模拟延迟、卡顿、partial / final 结果

这时候，一个可配置 + 多引擎 + 可控时序的 Mock Server 就非常有价值。

本文将带你用 Cursor 从 0 到 1 搭建这样一个系统。

---

一、设计目标

我们要实现的不是一个简单 mock，而是一个：

👉 多引擎 ASR 仿真框架

核心能力包括：

• 支持多个 ASR 引擎（Google / Verbio / AWS）

• gRPC Streaming 模拟（贴近真实 ASR）

• 支持 partial / final 结果

• 支持延迟控制（固定 + 随机波动）

• 完全由 YAML 配置驱动

---

二、整体架构

系统分为三层：

Client
   ↓
Mock Server（gRPC 层）
   ↓
Engine 层（Google / Verbio / AWS）
   ↓
Delay & Behavior（统一控制）

关键点：

• 协议层与引擎逻辑分离

• 所有行为由 config.yaml 控制

• 延迟系统统一抽象

---

三、项目初始化

创建基础目录：

mkdir asr-mock-server
cd asr-mock-server
touch spec.md config.yaml speech.proto

---

四、定义 config.yaml（核心）

配置文件决定一切行为：

engine:
  default: google

engines:
  google:
    behavior:
      delay:
        initial_silence:
          type: uniform
          min: 0
          max: 2

      scripted_responses:
        - transcript: "hel"
          is_final: false
          delay:
            type: uniform
            min: 0.2
            max: 0.5

        - transcript: "hello"
          is_final: false
          delay:
            type: uniform
            min: 0.3
            max: 0.6

        - transcript: "hello world"
          is_final: true
          delay:
            type: normal
            mean: 1
            std: 0.2

  verbio:
    behavior:
      fixed_response:
        transcript: "hola mundo"
        is_final: true

---

五、定义 gRPC 接口（speech.proto）

最小 Streaming 接口：

syntax = "proto3";

service Speech {
  rpc StreamingRecognize (stream StreamingRecognizeRequest)
      returns (stream StreamingRecognizeResponse);
}

message StreamingRecognizeRequest {
  oneof streaming_request {
    StreamingRecognitionConfig streaming_config = 1;
    bytes audio_content = 2;
  }
}

message StreamingRecognitionConfig {
  RecognitionConfig config = 1;
}

message RecognitionConfig {
  string model = 1;
  string language_code = 2;
}

message StreamingRecognizeResponse {
  repeated StreamingRecognizeResult results = 1;
}

message StreamingRecognizeResult {
  repeated SpeechRecognitionAlternative alternatives = 1;
  bool is_final = 2;
}

message SpeechRecognitionAlternative {
  string transcript = 1;
}

---

六、编写 Cursor Prompt（spec.md）

这是最关键的一步：

You are building a multi-engine ASR mock framework.

Goal:
Create a Python gRPC mock server that simulates multiple ASR providers with realistic streaming behavior.

Requirements:
- Use grpc.aio (async streaming)
- First request is config, then audio
- Responses must be streamed over time (yield)

Architecture:
- Plugin-based engines (google / verbio)
- Shared delay resolver
- Config-driven (config.yaml only)

Delay system:
- fixed / uniform / normal
- no negative delay
- used for initial silence and per response

Strict rules:
- DO NOT convert streaming to unary
- DO NOT hardcode behavior
- MUST follow config.yaml

---

七、用 Cursor 生成代码

在 Cursor 中执行：

Generate the full project based on this spec, config.yaml, and proto file.

生成后，再执行一次优化：

Refactor to ensure:
- strict layering (protocol / engine / delay)
- no hardcoded logic
- full config-driven behavior

---

八、实现延迟系统（核心）

统一延迟解析函数：

import random

def resolve_delay(cfg):
    if not isinstance(cfg, dict):
        return float(cfg)

    t = cfg.get("type", "fixed")

    if t == "fixed":
        return float(cfg.get("value", 0))

    elif t == "uniform":
        return random.uniform(cfg["min"], cfg["max"])

    elif t == "normal":
        return max(0, random.gauss(cfg["mean"], cfg["std"]))

---

九、模拟 Streaming 行为

核心逻辑：

await asyncio.sleep(initial_silence)

for item in scripted_responses:
    delay = resolve_delay(item["delay"])
    await asyncio.sleep(delay)

    yield response(item)

---

十、关键实现细节

1️⃣ initial silence（前置静默）

用于模拟：

• ASR 冷启动

• 网络延迟

• 服务端处理时间

---

2️⃣ partial → final 演进

推荐这样设计：

hel → hello → hello wor → hello world

更贴近真实 ASR。

---

3️⃣ 多引擎支持

通过配置切换：

engine = config["engine"]["default"]
behavior = config["engines"][engine]

---

十一、运行与测试

安装依赖：

pip install grpcio grpcio-tools pyyaml

运行服务：

python server.py

---

十二、进阶能力（推荐）

✅ 1. 随机种子（可复现）

random_seed: 12345

---

✅ 2. 最大延迟限制

max_delay_cap: 5

---

✅ 3. 错误注入

error_injection:
  type: timeout
  after_sec: 3

---

十三、总结

通过 Cursor + 合理的 Spec 设计，我们可以快速构建一个：

👉 可扩展、可配置、接近真实行为的 ASR Mock 平台

关键在于三点：

1. 配置驱动（config.yaml）

2. 流式行为模拟（async streaming）

3. 延迟建模（随机分布）

---