Autogen多Agent群组聊天对接FastAPI+Websocket问题求助

方案可行性确认 #

完全可行。Autogen原生支持多Agent群组对话的轮询回复，也允许每个Agent触发用户提问，结合FastAPI Websocket的实时通信能力，完全能实现你要的效果。

Websocket断开的核心原因及修复步骤 #

1. 连接超时或心跳缺失 #

多Agent对话时，Agent内部交互耗时比单Agent长很多，若Websocket没有心跳机制或超时设置过短，前端会判定连接失效而断开。

• 修复：给FastAPI Websocket加心跳，延长超时时间：

  from fastapi import WebSocket, WebSocketDisconnect
  import asyncio
  
  @app.websocket("/ws/chat")
  async def websocket_chat(websocket: WebSocket):
      await websocket.accept()
      try:
          while True:
              # 并行处理心跳和消息接收，避免阻塞
              tasks = [
                  asyncio.create_task(websocket.receive_text()),
                  asyncio.create_task(asyncio.sleep(30))
              ]
              done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
              
              for task in pending:
                  task.cancel()
              
              if done:
                  data = done.pop().result()
                  # 处理多Agent对话逻辑
                  await run_multi_agent_chat(data, websocket)
              else:
                  # 发送心跳包保持连接
                  await websocket.send_text("__PING__")
      except WebSocketDisconnect:
          print("用户断开连接")
  

2. 同步处理阻塞Websocket线程 #

单Agent对话逻辑简单，同步处理不会有问题，但多Agent对话用同步代码会卡住Websocket的消息循环，导致连接中断。

• 修复：全程用Autogen的异步模式，消息推送异步执行：

  from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
  
  async def run_multi_agent_chat(user_input, websocket):
      # 初始化多个Agent
      main_assist = AssistantAgent(
          name="主助手",
          llm_config={"config_list": [{"model": "gpt-3.5-turbo", "api_key": "你的密钥"}]}
      )
      specialist = AssistantAgent(
          name="领域专家",
          llm_config={"config_list": [{"model": "gpt-3.5-turbo", "api_key": "你的密钥"}]}
      )
      user_proxy = UserProxyAgent(
          name="用户代理",
          code_execution_config=False,
          human_input_mode="NEVER"  # 由前端处理用户输入
      )
  
      # 创建群组聊天
      group_chat = GroupChat(
          agents=[user_proxy, main_assist, specialist],
          messages=[],
          max_round=10
      )
      manager = GroupChatManager(groupchat=group_chat, llm_config={"config_list": [...]})
  
      # 自定义消息推送函数，实时发往前端
      async def forward_message(sender, recipient, message):
          await websocket.send_text(f"[{sender.name}]: {message['content']}")
      
      # 替换所有Agent的send方法为异步推送
      for agent in [main_assist, specialist, user_proxy]:
          agent.send = forward_message
  
      # 启动异步对话
      await user_proxy.initiate_chat(manager, message=user_input)
  

3. 前端未适配多消息流 #

多Agent依次发消息时，前端如果没有正确处理高频消息或重复监听，可能导致连接异常。

• 修复：React端简化Websocket逻辑，过滤心跳包，稳定处理消息：

  import { useEffect, useState } from 'react';
  
  function ChatComponent() {
      const [messages, setMessages] = useState([]);
  
      useEffect(() => {
          const ws = new WebSocket('ws://localhost:8000/ws/chat');
          
          ws.onmessage = (e) => {
              const msg = e.data;
              // 过滤心跳包，只处理有效消息
              if (msg !== '__PING__') {
                  setMessages(prev => [...prev, msg]);
              }
          };
  
          ws.onerror = (err) => console.error('连接出错:', err);
          ws.onclose = () => console.log('连接关闭');
  
          return () => ws.close();
      }, []);
  
      return (
          <div>
              {messages.map((msg, idx) => <div key={idx}>{msg}</div>)}
          </div>
      );
  }
  
  export default ChatComponent;
  

实现所有Agent提问能力的关键配置 #

要让每个Agent都能向用户提问，需要做两点：

• 给每个Agent设置human_input_mode为"ALWAYS"，这样当Agent需要用户反馈时，会触发输入请求：

  specialist = AssistantAgent(
      name="领域专家",
      llm_config={"config_list": [...]},
      human_input_mode="ALWAYS",
      system_message="你是XX领域专家，遇到需要用户补充信息的问题直接提问"
  )
  

• 在后端的消息推送逻辑中，识别Agent的提问请求，向前端发送特殊标识，让前端弹出输入框收集用户回复，再把回复发回后端继续对话：

  async def forward_message(sender, recipient, message):
      # 判断是否是Agent向用户提问
      if sender.human_input_mode == "ALWAYS" and "请你" in message['content']:
          await websocket.send_text(f"__NEED_REPLY__[{sender.name}]: {message['content']}")
      else:
          await websocket.send_text(f"[{sender.name}]: {message['content']}")
  

前端收到__NEED_REPLY__开头的消息时，显示输入框，用户输入后把内容发回Websocket，后端将回复作为用户输入传给Agent，继续对话流程。

内容的提问来源于stack exchange，提问作者SSwift