心理访谈转录平台构建：流式VAD采集、声纹聚类、WebSocket实时推送全链路项目解析27.0

# 会话主表
class DBSession(Base):
    __tablename__ = "sessions"
    id = Column(String, primary_key=True, index=True)
    risk_level = Column(String, default="none")
    mode = Column(String, default="chat")
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    handled = Column(Boolean, default=False)
    summary_text = Column(Text, default="")
    emotion_tags = Column(Text, default="")
    transcribe_state = Column(String, default="idle")
    full_audio_path = Column(String, nullable=True)

# 对话消息转录表
class DBSessionMessage(Base):
    __tablename__ = "session_messages"
    id = Column(String, primary_key=True)
    session_id = Column(String, ForeignKey("sessions.id"))
    role = Column(String)
    text = Column(Text)
    audio_path = Column(String, nullable=True)
    speaker = Column(String, nullable=True)
    start_ms = Column(Integer, nullable=True)
    end_ms = Column(Integer, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)

pip install fastapi uvicorn sqlalchemy pydantic funasr numpy scikit-learn
# ffmpeg单独安装，用于音频转码

class WsHub:
    def __init__(self):
        self.clients: List[WebSocket] = []
    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.clients.append(ws)
    def disconnect(self, ws: WebSocket):
        if ws in self.clients:
            self.clients.remove(ws)
    async def broadcast(self, payload: dict):
        dead = []
        for c in self.clients:
            try:
                await c.send_json(payload)
            except Exception:
                dead.append(c)
        for d in dead:
            self.disconnect(d)
hub = WsHub()

@app.websocket("/ws/transcript")
async def ws_transcript(ws: WebSocket):
    await hub.connect(ws)
    try:
        while True:
            await ws.receive_text()
    except WebSocketDisconnect:
        hub.disconnect(ws)
    except Exception:
        hub.disconnect(ws)

class UpdateMsgTextReq(BaseModel):
    msg_id: str
    text: Optional[str] = None
    speaker: Optional[str] = None
    teacher_account: Optional[str] = "teacher01"

@app.post("/api/teacher/message/update")
async def update_message_text(req: UpdateMsgTextReq, db: Session = Depends(get_db)):
    m = db.query(DBSessionMessage).filter(DBSessionMessage.id == req.msg_id).first()
    if not m:
        raise HTTPException(status_code=404, detail="消息不存在")
    old_text, old_spk = m.text, m.speaker
    if req.text is not None:
        m.text = req.text
    if req.speaker is not None:
        m.speaker = req.speaker
    db.add(DBAuditLog(id=str(uuid.uuid4()), operator=req.teacher_account,
                      action="edit_message", session_id=m.session_id))
    db.commit()
    print("[edit] %s: text %r -> %r | speaker %r -> %r" % (req.msg_id, old_text, m.text, old_spk, m.speaker))
    return {"ok": True, "msg_id": m.id, "text": m.text, "speaker": m.speaker}

def wav_duration_ms(path):
    try:
        with open(path, "rb") as f:
            head = f.read(12)
            if len(head) < 12 or head[0:4] != b"RIFF" or head[8:12] != b"WAVE":
                return None
            f.seek(12)
            channels = rate = bits = data_size = 0
            while True:
                hdr = f.read(8)
                if len(hdr) < 8:
                    break
                cid, csize = hdr[0:4], int.from_bytes(hdr[4:8], "little")
                if cid == b"fmt ":
                    body = f.read(csize)
                    if len(body) >= 16:
                        channels = int.from_bytes(body[2:4], "little")
                        rate = int.from_bytes(body[4:8], "little")
                        bits = int.from_bytes(body[14:16], "little")
                elif cid == b"data":
                    data_size = csize
                    break
                else:
                    f.seek(csize + (csize % 2), 1)
            if not (channels and rate and bits and data_size):
                return None
            return int(data_size / (channels * bits / 8) / rate * 1000)
    except Exception:
        return None

def get_asr_model():
    global _asr_model, _asr_broken
    if _asr_model is None and not _asr_broken:
        try:
            from funasr import AutoModel
            _asr_model = AutoModel(
                model="iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch",
                vad_model="iic/speech_fsmn_vad_zh-cn-16k-common-pytorch",
                punc_model="iic/punc_ct-transformer_zh-cn-common-vocab272727-pytorch",
                spk_model="iic/speech_campplus_sv_zh-cn_16k-common",
                device="cpu", disable_update=True)
            print("[init] 语音转录模型已加载：paraformer-zh + cam++ 说话人分离")
        except Exception as e:
            _asr_broken = True
            print("[warn] 语音转录模型不可用（仅保存音频原件）：", type(e).__name__, e)
    return _asr_model

def assign_speaker(sid, emb):
    st = live_state(sid)
    if emb is None:
        return "spk0"
    max_spk = 0 if st.multi_party else TWO_PARTY_MAX
    with st.lock:
        best, best_sim = None, -1.0
        for spk, c in st.centers.items():
            sim = float(np.dot(emb, c))
            if sim > best_sim:
                best, best_sim = spk, sim
        if best is not None and best_sim >= SPK_THR:
            n = st.counts.get(best, 1)
            merged = st.centers[best] * n + emb
            norm = float(np.linalg.norm(merged))
            st.centers[best] = merged / norm if norm > 0 else emb
            st.counts[best] = n + 1
            return best
        if best is not None and max_spk and len(st.centers) >= max_spk:
            st.counts[best] = st.counts.get(best, 1) + 1
            return best
        spk = "spk%d" % len(st.centers)
        st.centers[spk] = emb
        st.counts[spk] = 1
        return spk

def transcribe_diarized(wav_path):
    """整段音频 → 带说话人的句子列表 [{speaker, text, start_ms, end_ms}]（按时间顺序）"""
    model = get_asr_model()
    if model is None:
        return []
    res = model.generate(input=wav_path, batch_size_s=300, return_spk_res=True)
    if not res:
        return []
    item = res[0]
    segs = item.get("sentence_info") or []
    out = []
    for s in segs:
        txt = (s.get("text") or "").strip()
        if not txt:
            continue
        out.append({
            "speaker": "spk%d" % int(s.get("spk", 0)),
            "text": txt,
            "start_ms": int(s.get("start", 0)),
            "end_ms": int(s.get("end", 0)),
        })
    if not out:   # 没有句子级信息时退化为整段文本
        txt = clean_asr_text(item.get("text", ""))
        if txt:
            out.append({"speaker": "spk0", "text": txt, "start_ms": 0, "end_ms": 0})

    # 同一说话人、间隔很短的相邻句子合并，避免碎句过多
    merged = []
    for seg in out:
        if merged and merged[-1]["speaker"] == seg["speaker"] and seg["start_ms"] - merged[-1]["end_ms"] < 1200:
            merged[-1]["text"] += seg["text"]
            merged[-1]["end_ms"] = seg["end_ms"]
        else:
            merged.append(dict(seg))
    return merged

# ===================== 腾讯混元大模型接入（TokenHub平台） =====================
from openai import OpenAI

api_key = os.environ.get('TENCENT_API_KEY') 
llm_client = OpenAI(
    api_key=api_key,
    base_url="https://tokenhub.tencentmaas.com/v1",
)
model_name = "hy3-preview"


def llm_invoke(prompt: str, system_prompt: str = "你是一个专业的校园对话分析助手。请严格按照要求输出JSON格式结果，不要输出任何多余内容。") -> str:
    """调用腾讯混元大模型（同步阻塞，异步路由里请配合 asyncio.to_thread 使用）"""
    response = llm_client.chat.completions.create(
        model=model_name,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,
    )
    return response.choices[0].message.content


def _extract_json(text: str) -> Optional[dict]:
    """从容错解析 LLM 返回的 JSON（容忍 ```json 包裹 / 前后多余文字）"""
    if not text:
        return None
    t = text.strip()
    m = re.search(r"```(?:json)?\s*(\{.*\})\s*```", t, re.S)
    if m:
        t = m.group(1)
    else:
        s, e = t.find("{"), t.rfind("}")
        if s >= 0 and e > s:
            t = t[s:e + 1]
    try:
        return json.loads(t)
    except Exception:
        return None


ANALYZE_SYSTEM_PROMPT = """你是校园心理辅导对话分析助手。请根据带说话人标注的对话记录，输出严格JSON，字段如下：
{
  "summary": "80字以内的对话摘要，概括主要话题与结论",
  "risk_level": "danger|warning|low|none 四选一（danger=涉及自伤/伤人/极端情绪；warning=明显情绪困扰或冲突；low=轻微负面情绪；none=正常交流）",
  "emotion_tags": ["情绪标签数组，2~4个，如：焦虑、抵触、平静、积极"]
}
只输出JSON，不要任何解释。"""


def llm_analyze_session(transcript_text: str) -> Optional[dict]:
    """整段对话 → LLM 分析 {summary, risk_level, emotion_tags}；失败返回 None（不影响主流程）"""
    if not (transcript_text or "").strip():
        return None
    try:
        raw = llm_invoke(
            "以下是老师与学生的对话记录，请分析：\n" + transcript_text[:6000],
            system_prompt=ANALYZE_SYSTEM_PROMPT,
        )
        data = _extract_json(raw)
        if not data:
            return None
        risk = data.get("risk_level")
        if risk not in ("danger", "warning", "low", "none"):
            risk = "none"
        tags = data.get("emotion_tags") or []
        if not isinstance(tags, list):
            tags = []
        tags = [str(t).strip()[:12] for t in tags if str(t).strip()][:6]
        return {
            "summary": str(data.get("summary", "")).strip()[:200],
            "risk_level": risk,
            "emotion_tags": tags,
        }
    except Exception as e:
        print("[warn] 混元分析失败：", type(e).__name__, e)
        return None

function handleFrame(buf, off, sampleIdx){
    const tMs = sampleIdx / SR * 1000;
    const rms = frameRms(buf, off);
    const thr = Math.max(0.012, noiseFloor * 3.0);
    const frame = buf.subarray(off, off + FRAME);
    if(!inSpeech){
        noiseFloor = noiseFloor * 0.98 + rms * 0.02;
        if(rms > thr){
            voiceMs += 20;
            segChunks.push(frame.slice());
            if(voiceMs >= START_VOICE_MS){
                inSpeech = true; silMs = 0;
                segStartMs = Math.max(0, tMs - voiceMs + 20);
            }
        }else{
            voiceMs = 0;
            if(segChunks.length) segChunks = [];
        }
        return;
    }
    segChunks.push(frame.slice());
    silMs = rms > thr ? 0 : silMs + 20;
    const segMs = segChunks.length * 20;
    if(silMs >= END_SIL_MS){
        const keep = Math.max(1, Math.floor((segMs - silMs) / 20));
        finishSegment(segChunks.slice(0, keep), segStartMs, segStartMs + keep * 20);
        inSpeech = false; silMs = 0; voiceMs = 0; segChunks = [];
    }else if(segMs >= MAX_SEG_MS){
        finishSegment(segChunks.slice(), segStartMs, segStartMs + segMs);
        inSpeech = false; silMs = 0; voiceMs = 0; segChunks = [];
    }
}

# ===================== 接口：接收消息，存入转录记录（给智能体服务调用） =====================
@app.post("/api/archive/save_msg")
async def save_chat_message(req: SaveMessageReq, db: Session = Depends(get_db)):
    """智能体主服务调用：保存每一轮对话转录文本"""
    sid = req.session_id
    sess = db.query(DBSession).filter(DBSession.id == sid).first()
    if not sess:
        sess = DBSession(id=sid, risk_level=req.risk_level or "none", mode=req.mode)
        db.add(sess)
    else:
        # 高危标记一旦打上就不会降级
        if req.risk_level and sess.risk_level != "danger":
            sess.risk_level = req.risk_level
    db.commit()

    msg = DBSessionMessage(
        id=str(uuid.uuid4()),
        session_id=sid,
        role=req.role,
        text=req.text,
        audio_path=req.audio_path
    )
    db.add(msg)
    db.commit()

    # 实时推送给已连接的老师端
    await hub.broadcast({
        "type": "new_message",
        "msg_id": msg.id,
        "session_id": sid,
        "role": msg.role,
        "text": msg.text,
        "time": local_str(msg.created_at)
    })
    return {"ok": True, "msg_id": msg.id}

# 安装依赖
pip install fastapi uvicorn sqlalchemy pydantic funasr numpy scikit-learn
# 安装ffmpeg，ubuntu: apt install ffmpeg windows下载二进制加入环境变量
python transcript_archive.py