构建响应迅速且支持多人的 AI 应用

LLM 应用开发系列

响应速度：LLM 应用的生命线

在 LLM 应用中，首字延迟 (TTFT, Time To First Token) 比总生成时间更重要。流式交互（Streaming）不仅是技术优化，更是心理学上的“进度暗示”。而当应用进入多人协作场景（如 AI 辅助文档、共享白板）时，挑战则升级为如何在流式输出的同时保持多端状态的一致性。

深度解析：流式传输协议选型

1. SSE (Server-Sent Events) —— 文本生成的首选

  • 原理：基于标准 HTTP 协议，服务端保持连接并持续推送数据块。
  • 优势：原生支持断线重连，对代理服务器友好，实现极其简单。
  • 局限：仅支持单向（Server -> Client）传输。

2. WebSockets —— 复杂交互的基石

  • 原理：全双工通信协议。
  • 优势：支持双向实时交互（如用户在 AI 生成时进行实时打断或修改）。
  • 局限：连接管理成本高，需要处理心跳检测和复杂的负载均衡。

3. gRPC-Web —— 企业级高性能选择

  • 优势：强类型约束（Protobuf），适合微服务架构。

工业级流式后端实现 (Python Asyncio)

在处理高并发流式请求时，必须使用异步框架以避免阻塞线程池。

import asyncio
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
from langchain.callbacks import AsyncIteratorCallbackHandler

app = FastAPI()

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    data = await request.json()
    handler = AsyncIteratorCallbackHandler()
    llm = ChatOpenAI(streaming=True, callbacks=[handler])
    
    # 在后台任务中运行生成逻辑
    asyncio.create_task(llm.agenerate(messages=[[HumanMessage(content=data["prompt"])]]))

    async def event_generator():
        async for token in handler.aiter():
            # 封装为标准 SSE 格式
            yield f"data: {json.dumps({'choices': [{'delta': {'content': token}}]})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

实时协作：当 AI 遇上 CRDT

在多人协作编辑器中，AI 生成的内容必须像人类输入一样，能够被其他协作成员实时看到，且不会覆盖他人的编辑。

1. Yjs 与分布式状态

Yjs 是一种高性能的 CRDT（无冲突复制数据类型）实现。

  • 逻辑：AI 作为一个“虚拟用户”加入 Yjs 共享文档。
  • 同步：AI 生成的每一个 Token 都通过 Yjs 的 Text.insert() 方法插入，自动同步到所有在线用户的浏览器。

2. 架构设计

[用户 A] <───> [WebSocket Server] <───> [用户 B]
                     │
                     ▼
              [AI Agent 节点]
          (监听 Yjs 变更，生成内容并回写)

极致体验：前端交互设计模式

1. 乐观 UI (Optimistic UI) 与打字机效果

不要等后端返回第一个 Token 才显示气泡。

  • 策略：用户点击发送后，立即在 UI 上渲染用户的消息，并显示一个“AI 正在思考…”的骨架屏或闪烁光标。

2. 实时打断 (Interruption Handling)

在流式输出过程中，用户可能发现 AI 跑题了，需要立即停止。

  • 实现：前端调用 AbortController.abort()，后端捕获 asyncio.CancelledError 并立即停止 LLM 调用，节省 Token 成本。

3. 自动滚动与用户锁定

  • 挑战：当 AI 快速输出时，页面会自动向下滚动，但如果用户正在向上翻看历史记录，自动滚动会造成极差的体验。
  • 解决：监听滚动事件，如果用户手动向上滚动，则暂停“自动跟随”，并显示“有新消息”的浮窗。

总结

实时协作与流式交互是衡量一个 LLM 应用“工业感”的重要指标。通过 SSE 降低首字延迟、CRDT 解决协作冲突、以及精细的前端交互控制，我们可以构建出像原生应用一样流畅、自然的 AI 协作体验。

参考资源

最佳实践

  1. 压缩数据：流式传输时尽量减少冗余元数据。
  2. 错误处理：处理网络中断时的自动重连和状态恢复。
  3. 性能优化：避免频繁的 DOM 更新，使用虚拟列表或防抖技术。
  4. 安全性：确保实时通道的身份验证和权限控制。

总结

实时性是现代 AI 应用的灵魂。通过流式输出降低等待感，通过 WebSockets/CRDT 实现多人协作，我们可以构建出像原生应用一样流畅、高效的智能交互体验。

参考资源

版权声明： 如无特别声明，本文版权归 sshipanoo 所有，转载请注明本文链接。

（采用 CC BY-NC-SA 4.0 许可协议进行授权）

本文标题：《 LLM应用开发——实时协作与流式交互 》

本文链接：http://localhost:3015/ai/%E5%AE%9E%E6%97%B6%E5%8D%8F%E4%BD%9C.html

本文最后一次更新为 天前，文章中的某些内容可能已过时！

目录