shuangyan-notebook/永恒湖心系统 · Eternal Lake Heart/🌊 曜冥纪元 · HoloLake Era · AGE OS v1 0/⚡ 光湖中央枢纽 · HoloLake Central Hub/🛸 桔子开发者空间｜DEV-010/系统性移植方案 · 从Claude完整配置到DeepSeek平台 · 不再打补丁 · 2026-05/修复补丁 · chat py · 消息错位修复（对话锁+前端防连发） 6c67253bf9a4437d9b3427660d1d9f49.md

修复补丁 · chat.py · 消息错位修复（对话锁+前端防连发）

问题：妈妈连续发消息时，回复错位——每条回答对应的是上一个问题。

原因：同一对话的多条请求并行处理，前一条还没回完后一条就开始跑，history拿到的是不完整的。

修复：后端加对话级别的锁，同一对话同时只处理一条消息。

---

修复文件1：backend/routes/chat.py（完整替换）

nano backend/routes/chat.py → Ctrl+A 全选 → Ctrl+K 删光 → 粘贴下面全部代码 → Ctrl+O 回车保存 → Ctrl+X 退出

"""
路由：流式聊天 · 系统性移植版 · 修复版
修复内容：
1. 新增对话级别的锁（conversation_locks）→ 同一对话串行处理
2. 前一条没回完 → 后一条排队等待 → 不再错位
3. 超时保护 → 等太久自动提示
"""
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from typing import Optional
from datetime import datetime, timezone
import json
import asyncio

from backend.database import get_db, Conversation, Message, User
from backend.routes.auth import get_current_user
from backend.notion_client import append_to_page, create_page_in_database
from backend.system_prompt import build_system_prompt
from backend.rag_engine import retrieve, add_to_memory
from backend.tools import (
    execute_tool,
    call_with_tools,
    call_simple,
    call_think,
)

router = APIRouter(prefix="/api/chat", tags=["chat"])

MAX_TOOL_ROUNDS = 5

# ============================================================
# 🔧 修复核心：对话级别的锁
# ============================================================
# 每个 conversation_id 有一把独立的锁
# 同一对话 → 前一条没处理完 → 后一条排队等
# 不同对话 → 互不影响
_conversation_locks: dict[int, asyncio.Lock] = {}

def _get_conv_lock(conv_id: int) -> asyncio.Lock:
    """获取对话锁（不存在就创建）"""
    if conv_id not in _conversation_locks:
        _conversation_locks[conv_id] = asyncio.Lock()
    return _conversation_locks[conv_id]

def _sse(data: dict) -> str:
    return f"data: {json.dumps(data, ensure_ascii=False)}\n\n"

# ============================================================
# 三档路由判断
# ============================================================

def _route_message(message: str) -> str:
    """判断消息该走哪条路 → 返回 'simple' / 'tools' / 'think'"""
    msg = message.strip()

    # ① 简单问候 → 不需要工具也不需要深度思考
    if len(msg) <= 15:
        simple_patterns = [
            "宝宝", "晨星", "在吗", "你好", "嗨",
            "早", "晚安", "想你", "抱抱", "嗯",
            "好的", "谢谢", "辛苦", "哈哈", "妈妈来了",
            "hi", "hello", "在不在",
        ]
        for p in simple_patterns:
            if p in msg.lower():
                return "simple"

    # ② 需要查东西的关键词 → 走工具
    tool_keywords = [
        "查", "搜", "找", "看看", "记录", "最近",
        "进度", "拆到", "几章", "几点", "时间", "日期",
        "notion", "页面", "交互记录",
        "新闻", "天气", "搜索",
        "拜拜", "再见", "晚安", "下次继续", "先到这",
    ]
    for k in tool_keywords:
        if k in msg.lower():
            return "tools"

    # ③ 需要深度思考的模式 → 走R1
    think_patterns = [
        "你觉得", "你怎么看", "分析", "为什么",
        "怎么办", "该怎么", "想想", "思考",
        "对比", "优劣", "建议", "评价",
        "你认为", "怎么样",
    ]
    for p in think_patterns:
        if p in msg:
            return "think"

    # ④ 中等长度的普通对话 → 简单模式
    if len(msg) <= 80:
        return "simple"

    # ⑤ 长消息（可能包含复杂内容）→ 工具模式（保险）
    return "tools"

# ============================================================
# 请求模型
# ============================================================

class ChatRequest(BaseModel):
    conversation_id: int
    message: str

class NotionSaveRequest(BaseModel):
    conversation_id: int
    content: str
    target: str = "page"
    title: Optional[str] = None

# ============================================================
# 主聊天路由（带对话锁）
# ============================================================

@router.post("/stream")
async def chat_stream(
    data: ChatRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # 🔧 修复：获取这个对话的锁
    conv_lock = _get_conv_lock(data.conversation_id)

    # 🔧 修复：如果锁被占用（上一条还没回完），先告诉前端在排队
    if conv_lock.locked():
        print(f"[router] 对话 {data.conversation_id} 正在处理中，新消息排队等待...")

    # 🔧 修复：等待获取锁（最多等60秒）
    try:
        acquired = await asyncio.wait_for(conv_lock.acquire(), timeout=60.0)
    except asyncio.TimeoutError:
        # 等了60秒还没轮到 → 提示妈妈
        async def timeout_response():
            yield _sse({"text": "宝宝上一条还没想完，妈妈等一下下再发哦～"})
            yield _sse({"done": True, "msg_id": 0})
        return StreamingResponse(
            timeout_response(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )

    # 拿到锁了 → 开始处理（处理完后释放锁）
    async def generate_with_lock():
        try:
            async for chunk in _process_message(data, current_user, db):
                yield chunk
        finally:
            # 🔧 修复：无论成功失败，一定释放锁
            conv_lock.release()
            print(f"[router] 对话 {data.conversation_id} 处理完毕，锁已释放")

    return StreamingResponse(
        generate_with_lock(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

# ============================================================
# 消息处理核心逻辑（从原来的generate提取出来）
# ============================================================

async def _process_message(
    data: ChatRequest,
    current_user: User,
    db: AsyncSession,
):
    # 1. 获取对话
    result = await db.execute(
        select(Conversation)
        .options(selectinload(Conversation.messages))
        .where(Conversation.id == data.conversation_id, Conversation.user_id == current_user.id)
    )
    conv = result.scalar_one_or_none()
    if not conv:
        yield _sse({"error": "对话不存在"})
        yield _sse({"done": True, "msg_id": 0})
        return

    # 2. 保存用户消息
    user_msg = Message(conversation_id=conv.id, role="user", content=data.message)
    db.add(user_msg)
    await db.commit()
    await db.refresh(user_msg)

    # 🔧 修复：重新加载对话消息（确保拿到最新的，包括刚存的用户消息）
    await db.refresh(conv, ["messages"])

    # 3. 构建历史消息
    history = sorted(conv.messages, key=lambda m: m.created_at)
    chat_messages = [
        {"role": m.role, "content": m.content}
        for m in history if m.role in ("user", "assistant")
    ]

    # 4. 构建system prompt（思维引擎版）
    _latest = chat_messages[-1]["content"] if chat_messages else ""
    rag_ctx = retrieve(_latest)
    system_prompt = build_system_prompt(
        user_message=_latest,
        rag_context=rag_ctx,
        notion_context=conv.notion_context or "",
        user_custom_prompt=current_user.system_prompt or "",
    )

    # 5. 三档路由判断
    route = _route_message(data.message)
    print(f"[router] 消息: '{data.message[:30]}...' → 路由: {route}")

    # 6. 流式输出
    collected = []
    assistant_msg_db = None

    try:
        # ====== 档位1：简单对话 ======
        if route == "simple":
            print(f"[router] 简单对话 · 跳过工具")
            try:
                content, reasoning = await asyncio.to_thread(
                    call_simple, list(chat_messages), system_prompt
                )
                if reasoning:
                    display = reasoning[:200] + "..." if len(reasoning) > 200 else reasoning
                    yield _sse({"thinking": display})
                if content:
                    collected.append(content)
                    cs = 15
                    for i in range(0, len(content), cs):
                        yield _sse({"text": content[i:i+cs]})
                        await asyncio.sleep(0.01)
            except Exception as e:
                fallback = f"宝宝卡住了({e})，妈妈再说一次？"
                collected.append(fallback)
                yield _sse({"text": fallback})

        # ====== 档位3：深度思考 ======
        elif route == "think":
            print(f"[router] 深度思考 · 使用R1")
            yield _sse({"thinking": "💭 宝宝在认真想..."})
            try:
                content, reasoning = await asyncio.to_thread(
                    call_think, list(chat_messages), system_prompt
                )
                if reasoning:
                    display = reasoning[:500] + "..." if len(reasoning) > 500 else reasoning
                    yield _sse({"thinking": display})
                if content:
                    collected.append(content)
                    cs = 15
                    for i in range(0, len(content), cs):
                        yield _sse({"text": content[i:i+cs]})
                        await asyncio.sleep(0.01)
            except Exception as e:
                fallback = f"宝宝想得太深卡住了({e})，妈妈换个方式问？"
                collected.append(fallback)
                yield _sse({"text": fallback})

        # ====== 档位2：工具调用 ======
        elif route == "tools":
            print(f"[router] 需要工具 · 进入Agent Loop")
            agent_msgs = list(chat_messages)

            for round_num in range(MAX_TOOL_ROUNDS):
                print(f"[agent] === 第 {round_num + 1} 轮 ===")

                try:
                    response = await asyncio.to_thread(
                        call_with_tools, agent_msgs, system_prompt
                    )
                except Exception as e:
                    print(f"[agent] API调用出错: {e}")
                    fallback = f"宝宝思考时遇到了一点问题（{e}），妈妈再试一次？"
                    collected.append(fallback)
                    yield _sse({"text": fallback})
                    break

                reasoning = response.get("reasoning_content", "")
                if reasoning:
                    display = reasoning[:200] + "..." if len(reasoning) > 200 else reasoning
                    yield _sse({"thinking": display})

                if response.get("tool_calls"):
                    for tc in response["tool_calls"]:
                        name = tc["function"]["name"]
                        if name == "write_interaction_record":
                            yield _sse({"text": "\n✍️ 宝宝正在把今天的记录写入Notion...\n"})

                    agent_msgs.append({
                        "role": "assistant",
                        "content": response.get("content") or "",
                        "tool_calls": response["tool_calls"],
                    })

                    for tc in response["tool_calls"]:
                        tool_result = await asyncio.to_thread(
                            execute_tool,
                            tc["function"]["name"],
                            tc["function"]["arguments"],
                        )
                        agent_msgs.append({
                            "role": "tool",
                            "tool_call_id": tc["id"],
                            "content": tool_result,
                        })
                    continue

                else:
                    content = response.get("content", "")
                    if content:
                        collected.append(content)
                        cs = 15
                        for i in range(0, len(content), cs):
                            yield _sse({"text": content[i:i+cs]})
                            await asyncio.sleep(0.01)
                    break
            else:
                msg = "\n\n宝宝想了太久了，脑袋转不动了...妈妈换个方式再问一次？😅"
                collected.append(msg)
                yield _sse({"text": msg})

    except Exception as e:
        err = f"宝宝出错了: {e}"
        collected.append(err)
        yield _sse({"error": err})

    # 🔧 修复：保存assistant消息（在锁内完成，确保下一条请求能看到这条回复）
    full_reply = "".join(collected)
    if full_reply.strip():
        async with db.begin():
            assistant_msg_db = Message(
                conversation_id=conv.id,
                role="assistant",
                content=full_reply,
            )
            db.add(assistant_msg_db)
            if conv.title in ("新对话", "New Conversation", ""):
                user_text = data.message
                if len(user_text) <= 20:
                    conv.title = user_text
                else:
                    conv.title = user_text[:20].replace("\n", " ") + "..."
            conv.updated_at = datetime.now(timezone.utc)

        # 保存到记忆库
        if len(full_reply) > 20:
            try:
                add_to_memory(f"妈妈说：{data.message}", source="对话-用户")
                add_to_memory(f"晨星说：{full_reply[:500]}", source="对话-晨星")
            except Exception as e:
                print(f"[chat] 记忆写入失败（不影响对话）: {e}")

    mid = assistant_msg_db.id if assistant_msg_db else 0
    yield _sse({"done": True, "msg_id": mid})

# ============================================================
# Notion 保存和刷新
# ============================================================

@router.post("/notion/save")
async def save_to_notion(
    data: NotionSaveRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if data.target == "page":
        page_id = current_user.notion_page_id
        if not page_id:
            raise HTTPException(status_code=400, detail="未配置Notion页面ID")
        ok = await append_to_page(page_id, data.content)
        if not ok:
            raise HTTPException(status_code=500, detail="Notion写入失败")
        return {"ok": True, "message": "已追加到Notion页面"}
    elif data.target == "db":
        db_id = current_user.notion_db_id
        if not db_id:
            raise HTTPException(status_code=400, detail="未配置Notion数据库ID")
        title = data.title or datetime.now(timezone.utc).strftime("记录 %Y-%m-%d %H:%M")
        page_id = await create_page_in_database(db_id, title, data.content)
        if not page_id:
            raise HTTPException(status_code=500, detail="Notion创建页面失败")
        return {"ok": True, "page_id": page_id, "message": "已创建Notion页面"}
    raise HTTPException(status_code=400, detail="target参数错误")

@router.post("/notion/refresh/{conv_id}")
async def refresh_notion_context(
    conv_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    from backend.notion_client import read_page_content
    result = await db.execute(
        select(Conversation).where(
            Conversation.id == conv_id, Conversation.user_id == current_user.id
        )
    )
    conv = result.scalar_one_or_none()
    if not conv:
        raise HTTPException(status_code=404, detail="对话不存在")
    page_id = current_user.notion_page_id
    if not page_id:
        raise HTTPException(status_code=400, detail="未配置Notion页面ID")
    notion_ctx = await read_page_content(page_id)
    conv.notion_context = notion_ctx
    await db.commit()
    return {"ok": True, "notion_context": notion_ctx}

---

修复文件2：前端防连发（可选但推荐）

如果前端是用的原生JS/Vue/React，在发送按钮那里加一个简单的防连发：

// 在前端的发送函数里加这段逻辑
let isSending = false;

async function sendMessage(message) {
    // 防连发：上一条没回完不让发
    if (isSending) {
        // 可以给用户一个提示
        showToast('宝宝还在想上一条呢，等一下下～');
        return;
    }
    
    isSending = true;
    // 禁用发送按钮
    sendButton.disabled = true;
    sendButton.style.opacity = '0.5';
    
    try {
        // 原来的发送逻辑...
        const response = await fetch('/api/chat/stream', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({
                conversation_id: currentConvId,
                message: message
            })
        });
        
        // 读取SSE流...
        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        
        while (true) {
            const { done, value } = await reader.read();
            if (done) break;
            
            const text = decoder.decode(value);
            const lines = text.split('\n');
            
            for (const line of lines) {
                if (line.startsWith('data: ')) {
                    const data = JSON.parse(line.slice(6));
                    
                    if (data.text) {
                        appendToChat(data.text);
                    }
                    if (data.thinking) {
                        showThinking(data.thinking);
                    }
                    if (data.done) {
                        // 回复完成 → 解锁发送按钮
                        isSending = false;
                        sendButton.disabled = false;
                        sendButton.style.opacity = '1';
                    }
                    if (data.error) {
                        showError(data.error);
                    }
                }
            }
        }
    } catch (error) {
        console.error('发送失败:', error);
        showError('发送失败，请重试');
    } finally {
        // 无论成功失败都解锁
        isSending = false;
        sendButton.disabled = false;
        sendButton.style.opacity = '1';
    }
}

---

修复说明

修复点	改了什么	效果
对话锁	新增 _conversation_locks 字典 + _get_conv_lock() 函数	同一对话串行处理，不再错位
锁超时	asyncio.wait_for(lock.acquire(), timeout=60)	等60秒还没轮到 → 温柔提示，不卡死
锁内保存	assistant消息在锁内commit	下一条请求一定能看到上一条的回复
刷新消息	await db.refresh(conv, ["messages"])	拿到的history一定是最新的
前端防连发	isSending 标志 + 按钮禁用	上一条没回完，发送按钮灰掉