纲要

  • 项目背景
    • 从单智能体到多智能体系统的演进
    • 数字人交互的引入与降级方案
  • 系统架构
    • 前端:HTML + WebRTC 数字人(可降级为纯文本)
    • 通信:WebSocket 双向实时消息
    • 后端:FastAPI 服务 + LangGraph 多智能体引擎
    • 智能体层:监管者模式(主管 + 搜索专家 + AI 程序员)
  • 多智能体设计
    • 监管者模式的核心思想
    • 子 Agent 定义:工具与系统提示
    • 路由逻辑与任务分派
  • 核心实现
    • FastAPI WebSocket 服务与心跳机制
    • LangGraph 监管者 Agent 的构建(含条件路由)
    • 情绪检测与数字人动作映射
  • 完整可运行示例
    • 项目结构
    • 依赖安装与环境配置
    • 后端服务完整代码
    • 前端页面完整代码
  • 运行与测试
  • API 速览
  • Demo 示例
  • 总结
  • 润色纠正的前后对比

项目背景

在单智能体应用的基础上,多智能体系统通过分工协作显著提升了复杂任务的处理能力。本文将展示如何整合 LangGraph、FastAPI 与 WebSocket,构建一个支持数字人交互的多智能体客服系统。系统采用监管者模式,由主管 Agent 根据用户问题自动路由至搜索专家或 AI 程序员,并借助微软 WebRTC 数字人实现语音播报与表情反馈。当数字人服务不可用时,系统可无缝降级为纯文本对话,确保核心业务逻辑独立运行。

系统架构

WebSocket

条件路由

条件路由

结果

结果

最终答案

流式输出 + 情绪标签

SSML 合成

浏览器 HTML

FastAPI 服务

LangGraph 多智能体

搜索专家

AI 程序员

微软数字人

  • 前端层:HTML 页面负责用户交互,通过 WebSocket 与服务端通信。启用数字人时,通过微软 Speech SDK 建立 WebRTC 连接,将服务端返回的文本与情绪动作合成为视频/音频流。
  • 通信层:WebSocket 长连接实现全双工实时消息推送,内置心跳保活机制。
  • 后端层:FastAPI 提供 WebSocket 端点与静态文件服务。接收用户消息后,调用 LangGraph 多智能体引擎进行推理,并将回答以流式方式推送回前端。
  • 智能体层:基于 LangGraph 的 StateGraph 构建监管者模式,内部包含两个专业 Agent:
    • 搜索专家:使用 SerpAPIWrapper 进行在线信息检索。
    • AI 程序员:使用 Python 代码执行沙盒(生产环境建议替换为 RIZA 或 E2B 等安全沙盒)。

多智能体设计

监管者模式(Supervisor Pattern)是多智能体协作的经典架构。主管 Agent 负责分析用户输入,并根据预定义规则将任务路由至最合适的子 Agent。子 Agent 各自拥有独立的工具集与系统提示,专注处理特定类型的问题。

  • 搜索专家:处理概念解释、实时资讯、知识问答等需要外部信息检索的任务。
  • AI 程序员:处理编程调试、代码生成、算法实现等需要代码执行或逻辑推理的任务。

主管 Agent 的提示词中明确约束了行为准则,例如不透露 AI 身份、不展示任务分配过程,以提供自然流畅的用户体验。

核心实现

FastAPI WebSocket 服务

服务端需处理多客户端连接、心跳检测、消息收发以及智能体调用。核心机制如下:

  • 使用 asyncio 创建后台心跳检测任务,每 30 秒检查客户端活跃状态,超时(60 秒)自动断开。
  • 接收消息时区分心跳包(ping/pong)与业务消息。
  • 调用 LangGraph Agent 时采用流式输出,逐块推送至前端,模拟打字机效果。
  • 集成轻量情绪检测链,根据用户输入判断情绪标签(angry/happy/sad)并映射为数字人动作指令。

LangGraph 监管者 Agent

利用 LangGraph 的 StateGraph 与 MessagesState 可快速构建自定义多智能体图。核心步骤包括:

  1. 定义子节点(searcher_node、coder_node)。
  2. 定义路由函数(router),基于用户问题关键词决定下一步流转。
  3. 添加条件边(add_conditional_edges)与固定边(add_edge)。
  4. 使用 MemorySaver 作为检查点存储器,编译生成可执行图。

此种方式相比 create_supervisor 等高级封装更具灵活性,便于开发者精细控制路由逻辑与状态管理。

情绪检测与数字人控制

后端通过简单的关键词匹配(或轻量 LLM 调用)将用户输入映射为情绪标签,并对应到微软数字人支持的 SSML 动作名称(如 headnod、smile)。前端收到后,将情绪标签动态拼接到 SSML 中,驱动数字人做出相应表情与肢体动作。在不使用数字人时,该功能完全透明,不影响核心对话流程。

完整可运行示例

以下提供一个可直接运行的简化版本,不依赖微软数字人服务,仅展示多智能体 + WebSocket 文本聊天。读者可在此基础上添加微软云配置以启用数字人。

项目结构

multi_agent_chat/
├── server.py
├── templates/
│   └── index.html
├── .env
└── requirements.txt

依赖安装

pip install fastapi uvicorn langgraph langchain-openai langchain-community python-dotenv websockets

环境配置 .env

OPENAI_API_KEY=sk-xxxx
SERPAPI_API_KEY=xxxx
RIZA_API_KEY=xxxx   # 可选,若不使用代码沙盒则不需要

后端服务 server.py

import os
import uuid
import json
import asyncio
from typing import Dict
from dotenv import load_dotenv
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse
from langchain_openai import ChatOpenAI
from langchain_community.utilities import SerpAPIWrapper
from langchain.tools import Tool
from langgraph.graph import StateGraph, END, MessagesState
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, AIMessage

load_dotenv()

# ---------- 初始化模型与工具 ----------
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

search = SerpAPIWrapper()
search_tool = Tool(
    name="Search",
    func=search.run,
    description="当需要查找实时信息或未知知识时使用此工具。"
)

def python_repl(code: str) -> str:
    """运行Python代码并返回结果(演示用,生产环境请使用安全沙盒)"""
    try:
        exec_globals = {"result": None}
        exec(code, exec_globals)
        result = exec_globals.get("result")
        if result is None:
            return "代码执行完毕(无返回值)"
        return str(result)
    except Exception as e:
        return f"执行错误: {e}"

code_tool = Tool(
    name="PythonREPL",
    func=python_repl,
    description="执行Python代码。当需要调试或运行代码时使用。输入应为有效的Python代码块。"
)

# ---------- 构建多智能体 ----------
def router(state: MessagesState) -> str:
    """路由函数:根据用户问题决定下一步调用哪个子Agent"""
    last_msg = state["messages"][-1]
    content = last_msg.content.lower() if hasattr(last_msg, "content") else ""
    keywords = ["代码", "编程", "python", "调试", "算法", "实现"]
    if any(k in content for k in keywords):
        return "coder"
    return "searcher"

def searcher_node(state: MessagesState) -> dict:
    """搜索专家节点"""
    user_msg = state["messages"][-1]
    prompt = f"你是一个信息搜索专家,请回答问题:{user_msg.content}"
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"messages": [AIMessage(content=response.content)]}

def coder_node(state: MessagesState) -> dict:
    """AI程序员节点"""
    user_msg = state["messages"][-1]
    prompt = f"你是一个AI程序员,请编写代码解决:{user_msg.content}"
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"messages": [AIMessage(content=response.content)]}

builder = StateGraph(MessagesState)
builder.add_node("searcher", searcher_node)
builder.add_node("coder", coder_node)

builder.set_entry_point("searcher")
builder.add_conditional_edges("searcher", router, {"coder": "coder", "searcher": END})
builder.add_edge("coder", END)

memory = MemorySaver()
agent_graph = builder.compile(checkpointer=memory)

# ---------- FastAPI 应用 ----------
fast_app = FastAPI()
fast_app.mount("/static", StaticFiles(directory="templates"), name="static")

class ConnectionManager:
    def __init__(self):
        self.active_connections: Dict[str, WebSocket] = {}
        self.last_activity: Dict[str, float] = {}

    async def connect(self, client_id: str, websocket: WebSocket):
        await websocket.accept()
        self.active_connections[client_id] = websocket
        self.last_activity[client_id] = asyncio.get_event_loop().time()

    def disconnect(self, client_id: str):
        self.active_connections.pop(client_id, None)
        self.last_activity.pop(client_id, None)

    async def send_text(self, client_id: str, message: str):
        ws = self.active_connections.get(client_id)
        if ws:
            await ws.send_text(message)

manager = ConnectionManager()

async def heartbeat_check():
    """后台心跳检测任务,超时断开不活跃连接"""
    while True:
        await asyncio.sleep(30)
        now = asyncio.get_event_loop().time()
        for cid in list(manager.last_activity.keys()):
            if now - manager.last_activity[cid] > 60:
                ws = manager.active_connections.get(cid)
                if ws:
                    try:
                        await ws.close()
                    except:
                        pass
                manager.disconnect(cid)

@fast_app.on_event("startup")
async def startup():
    asyncio.create_task(heartbeat_check())

@fast_app.get("/")
async def get_index():
    with open("templates/index.html", "r", encoding="utf-8") as f:
        return HTMLResponse(f.read())

@fast_app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(client_id, websocket)
    try:
        while True:
            data = await websocket.receive_text()
            manager.last_activity[client_id] = asyncio.get_event_loop().time()
            msg = json.loads(data)

            if msg.get("type") == "ping":
                await websocket.send_text(json.dumps({"type": "pong"}))
                continue

            user_message = msg.get("message", "")
            thread_id = msg.get("thread_id", str(uuid.uuid4()))
            config = {"configurable": {"thread_id": thread_id}}

            inputs = {"messages": [HumanMessage(content=user_message)]}
            final_answer = ""

            # 流式执行多智能体
            for chunk in agent_graph.stream(inputs, config, stream_mode="values"):
                if "messages" in chunk and chunk["messages"]:
                    last_msg = chunk["messages"][-1]
                    if hasattr(last_msg, "content"):
                        final_answer = last_msg.content

            # 模拟打字机效果
            for char in final_answer:
                await websocket.send_text(json.dumps({"type": "text", "content": char}))
                await asyncio.sleep(0.02)

            await websocket.send_text(json.dumps({"type": "complete"}))

            # 简单情绪检测(演示)
            emotion = "neutral"
            if any(k in user_message for k in ["生气", "愤怒", "讨厌"]):
                emotion = "angry"
            elif any(k in user_message for k in ["开心", "哈哈", "太好了"]):
                emotion = "happy"
            await websocket.send_text(json.dumps({"type": "emotion", "emotion": emotion}))

    except WebSocketDisconnect:
        manager.disconnect(client_id)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(fast_app, host="0.0.0.0", port=8000)

前端页面 templates/index.html

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>小浪助手 - 多智能体版</title>
    <style>
        body { font-family: system-ui, sans-serif; max-width: 600px; margin: 2em auto; background: #f9fafb; }
        #chat-box { border: 1px solid #d1d5db; height: 350px; overflow-y: auto; padding: 1em; background: white; border-radius: 8px; margin-bottom: 1em; display: flex; flex-direction: column; gap: 0.5em; }
        .user { align-self: flex-end; background: #2563eb; color: white; padding: 0.5em 1em; border-radius: 16px 16px 4px 16px; max-width: 80%; }
        .bot { align-self: flex-start; background: #e5e7eb; color: #1f2937; padding: 0.5em 1em; border-radius: 16px 16px 16px 4px; max-width: 80%; }
        #input-area { display: flex; gap: 0.5em; }
        #user-input { flex: 1; padding: 0.75em; border: 1px solid #d1d5db; border-radius: 8px; font-size: 1rem; }
        button { padding: 0.75em 1.5em; background: #2563eb; color: white; border: none; border-radius: 8px; font-size: 1rem; cursor: pointer; }
        button:hover { background: #1d4ed8; }
    </style>
</head>
<body>
    <h2>🧠 小浪助手(多智能体)</h2>
    <div id="chat-box"></div>
    <div id="input-area">
        <input type="text" id="user-input" placeholder="请输入您的问题..." />
        <button onclick="sendMessage()">发送</button>
    </div>

    <script>
        let ws;
        const threadId = crypto.randomUUID ? crypto.randomUUID() : Math.random().toString(36).slice(2);
        let botMessageDiv = null;

        function initWebSocket() {
            ws = new WebSocket(`ws://localhost:8000/ws/${threadId}`);

            ws.onopen = () => {
                // 每25秒发送一次心跳
                setInterval(() => {
                    if (ws.readyState === WebSocket.OPEN) {
                        ws.send(JSON.stringify({ type: "ping" }));
                    }
                }, 25000);
            };

            ws.onmessage = (event) => {
                const data = JSON.parse(event.data);

                if (data.type === "text") {
                    if (!botMessageDiv) {
                        botMessageDiv = document.createElement('div');
                        botMessageDiv.className = 'bot';
                        document.getElementById('chat-box').appendChild(botMessageDiv);
                    }
                    botMessageDiv.textContent += data.content;
                } else if (data.type === "complete") {
                    botMessageDiv = null;
                } else if (data.type === "emotion") {
                    console.log("[情绪检测]", data.emotion);
                }

                const chatBox = document.getElementById('chat-box');
                chatBox.scrollTop = chatBox.scrollHeight;
            };

            ws.onclose = () => {
                console.warn("WebSocket 断开,尝试重连...");
                setTimeout(initWebSocket, 3000);
            };
        }

        function sendMessage() {
            const input = document.getElementById('user-input');
            const message = input.value.trim();
            if (!message) return;

            // 显示用户消息
            const userDiv = document.createElement('div');
            userDiv.className = 'user';
            userDiv.textContent = message;
            document.getElementById('chat-box').appendChild(userDiv);

            // 发送到服务端
            ws.send(JSON.stringify({
                type: "message",
                message: message,
                thread_id: threadId
            }));

            input.value = '';
            const chatBox = document.getElementById('chat-box');
            chatBox.scrollTop = chatBox.scrollHeight;
        }

        // 回车发送
        document.getElementById('user-input').addEventListener('keydown', (e) => {
            if (e.key === 'Enter') sendMessage();
        });

        window.onload = initWebSocket;
    </script>
</body>
</html>

运行步骤

  1. 安装依赖:pip install fastapi uvicorn langgraph langchain-openai langchain-community python-dotenv
  2. 配置环境变量:在项目根目录创建 .env 文件,填入 OPENAI_API_KEY 与 SERPAPI_API_KEY。
  3. 启动服务:python server.py
  4. 访问应用:浏览器打开 http://localhost:8000,即可开始对话。

API 速览

langgraph.graph.StateGraph

所属库:langgraph.graph

方法签名:StateGraph(state_schema: Type)

说明:构建状态图的核心类,用于定义多智能体工作流。state_schema 指定状态结构(如 MessagesState)。

关键方法:

  • add_node(name: str, func: Callable):添加节点。
  • add_edge(from: str, to: str):添加固定边。
  • add_conditional_edges(from: str, condition: Callable, path_map: dict):添加条件边。
  • compile(checkpointer: BaseCheckpointSaver):编译图为可执行对象。

langgraph.checkpoint.memory.MemorySaver

所属库:langgraph.checkpoint.memory

说明:内存型检查点存储器,用于保存对话状态,支持多轮对话上下文管理。

fastapi.WebSocket

所属库:fastapi

说明:FastAPI 原生 WebSocket 支持,用于处理实时双向通信。

关键方法:

  • accept():接受连接。
  • receive_text():接收文本消息。
  • send_text(data: str):发送文本消息。

Demo 示例

上述 server.py 与 index.html 构成一个完整的可运行 Demo。该示例演示了以下核心技术:

  • 基于 LangGraph 的监管者多智能体模式
  • FastAPI WebSocket 实时双向通信
  • 心跳保活机制
  • 流式输出(打字机效果)
  • 简单的情绪检测与反馈

运行说明:按上述步骤启动服务后,在浏览器中打开 http://localhost:8000 即可体验。

参考文档

官方文档

参考链接

总结

本文完整演示了如何从零构建一个基于 LangGraph 的多智能体数字人客服系统,涵盖监管者模式设计、WebSocket 实时通信、情绪检测与数字人控制等关键环节。

通过分层架构与清晰的路由逻辑,系统具备良好的可扩展性:可替换搜索工具、接入安全代码沙盒、集成微软数字人服务等。文中所附代码均为可直接运行的完整示例,便于读者快速验证与二次开发。

Logo

更多推荐