FastAPI + WebSocket 实时事件流架构

FastAPI + WebSocket 实时事件流架构

学习目标

• 理解 ChatDev 的 FastAPI 服务器架构与路由分层
• 掌握 WebSocket 连接管理与实时事件推送机制
• 了解同步/异步执行路由的设计差异
• 能够理解 WebSocket 日志器如何将工作流事件推送到前端

---

项目实践

服务器架构

ChatDev 使用 server_main.py 作为入口，启动 Uvicorn 服务器：

def main():
    import uvicorn
    args = build_parser().parse_args()
    uvicorn.run(
        "server.app:app",
        host=args.host,
        port=args.port,
        reload=args.reload,
        ws="wsproto",            # 使用 wsproto 作为 WebSocket 协议
        **build_reload_kwargs(args),
    )

路由分层

server/app.py
    ↓
注册路由模块:
├── /api/execute          → execute.py (工作流执行)
├── /api/execute-sync     → execute_sync.py (同步执行)
├── /api/sessions         → sessions.py (会话管理)
├── /api/workflows        → workflows.py (工作流 CRUD)
├── /api/tools            → tools.py (工具列表)
├── /api/artifacts        → artifacts.py (产物管理)
├── /api/batch            → batch.py (批量执行)
├── /api/health           → health.py (健康检查)
├── /api/vuegraphs        → vuegraphs.py (图数据)
├── /api/uploads          → uploads.py (文件上传)
└── /ws                   → websocket.py (WebSocket)

服务层架构

server/services/
├── workflow_run_service.py    # 工作流执行生命周期
├── session_execution.py       # 会话执行管理
├── session_store.py           # 会话持久化
├── websocket_manager.py       # WebSocket 连接池
├── websocket_executor.py      # WebSocket 事件推送
├── websocket_logger.py        # WebSocket 日志器
├── attachment_service.py      # 附件服务
├── batch_run_service.py       # 批量执行服务
├── message_handler.py         # 消息处理
├── prompt_channel.py          # 人机交互通道
├── artifact_dispatcher.py     # 产物分发
├── artifact_events.py         # 产物事件
└── vuegraphs_storage.py       # 图数据存储

WebSocket 连接管理

websocket_manager.py 维护连接池：

WebSocketManager
    ↓
connection_pool: Dict[session_id, List[WebSocket]]
    ↓
广播事件到 session 的所有连接

工作流程：

同步 vs 异步执行

ChatDev 提供两种执行路由：

路由	行为	适用场景
/api/execute	异步执行，立即返回 session_id	长时间运行的工作流
/api/execute_sync	同步执行，等待完成后返回结果	短工作流 / 测试

异步执行的完整流程：

POST /api/execute
    ↓
创建 session → 启动后台线程执行 → 立即返回 {session_id}
    ↓
前端通过 WebSocket 监听 session_id 的事件流
    ↓
执行完成 → 通过 WebSocket 推送完成事件

热重载安全

ChatDev 的 --reload 机制经过精心设计：

RELOAD_SOURCE_DIRS = [
    "check", "entity", "functions", "mcp_example",
    "runtime", "schema_registry", "server",
    "tools", "utils", "workflow",
]

_RELOAD_EXCLUDE_DIRS = ("WareHouse", "logs", "data", "temp", "node_modules")

• 白名单：只监听源码目录（RELOAD_SOURCE_DIRS）
• 黑名单：排除输出目录（RELOAD_EXCLUDE_DIRS）
• 多深度排除：WareHouse/*、WareHouse/*/* … 最多 10 层

这防止 Agent 生成的文件（如 WareHouse/demo/foo.py）触发服务器重启。

问题与规避

问题	表现	规避策略
WebSocket 断开	事件丢失	会话存储持久化 + 重连后回放
长连接超时	Nginx/代理超时断开	心跳机制或增大代理超时
异步执行异常	后台线程静默失败	通过 WebSocket 推送错误事件
热重载误触发	Agent 生成文件触发重启	确保 RELOAD_EXCLUDES 完整
watchfiles 未安装	StatReload 忽略 exclude	日志警告提示安装 watchfiles

设计取舍

WebSocket vs SSE

维度	WebSocket	SSE
通信方向	双向	单向（服务器→客户端）
取消任务	支持（客户端发消息）	不支持
协议复杂度	高（需 wsproto/uWebSockets）	低（HTTP EventSource）
ChatDev 的选择	✅ 使用	❌ 未采用

ChatDev 选择 WebSocket 的核心原因是双向通信——前端需要能够取消正在执行的工作流。

同步 vs 异步路由

维度	同步路由	异步路由
阻塞	阻塞请求直到完成	立即返回
超时	受 HTTP 超时限制	不受限
实时反馈	无	WebSocket 推送
适用场景	短工作流 / CLI 调用	长工作流 / Web UI

参考来源

• ChatDev Server 源码
• FastAPI 官方文档
• Uvicorn 官方文档