Agent人机交互前后端设计思路：从零构建高可用对话系统

1次阅读

没有评论

共计 2322 个字符，预计需要花费 6 分钟才能阅读完成。

刚接触 Agent 系统开发时，最容易遇到这两个问题：

场景一：多轮对话状态丢失
用户问 ” 查航班 ” 后接着说 ” 明天北京到上海的 ”，传统服务端会把两次请求当作独立事件处理，无法理解上下文关联
场景二：高并发时消息错乱
当 1000 个用户同时提问，服务端可能将用户 A 的应答错误返回给用户 B，这种错误在客服场景会造成严重事故

开发实时交互系统时，通信协议直接影响体验。我们对比两种主流方案：

RESTful API
优点：实现简单，兼容性好
缺点：需要客户端轮询，无法服务端主动推送
典型延迟：500ms 以上
WebSocket
优点：全双工通信，服务端可主动推送
缺点：需要维护长连接
典型延迟：50ms 以内

选型建议 ：对实时性要求高的对话系统（如在线客服、语音助手）首选 WebSocket，对兼容性要求高的简单查询可用 REST

使用 JSON Schema 定义消息格式，这个例子定义客户端请求格式：

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "properties": {"session_id": {"type": "string"},
    "user_input": {"type": "string"},
    "timestamp": {"type": "number"}
  },
  "required": ["session_id", "user_input"]
}

用 Redis 存储对话状态，Python 实现示例：

import redis

class DialogStateMachine:
    def __init__(self):
        self.redis = redis.StrictRedis(host='localhost', port=6379, db=0)

    def get_state(self, session_id):
        """获取当前对话状态"""
        return self.redis.hgetall(f"dialog:{session_id}")

    def update_state(self, session_id, new_state):
        """更新状态并设置过期时间"""
        pipe = self.redis.pipeline()
        pipe.hmset(f"dialog:{session_id}", new_state)
        pipe.expire(f"dialog:{session_id}", 3600)  # 1 小时过期
        pipe.execute()

以下代码展示如何处理 WebSocket 消息：

const WebSocket = require('ws');

const wss = new WebSocket.Server({port: 8080});

wss.on('connection', (ws) => {ws.on('message', (message) => {
    try {const msg = JSON.parse(message);
      // 业务处理逻辑
      processMessage(msg).then(response => {ws.send(JSON.stringify(response));
      });
    } catch (err) {console.error(` 消息解析失败: ${err}`);
      ws.send(JSON.stringify({error: 'INVALID_MESSAGE_FORMAT'}));
    }
  });

  // 心跳检测
  const heartbeat = setInterval(() => {if (ws.isAlive === false) return ws.terminate();
    ws.isAlive = false;
    ws.ping(null, false, true);
  }, 30000);

  ws.on('pong', () => {ws.isAlive = true;});
  ws.on('close', () => clearInterval(heartbeat));
});

我们测试了三种常见方案（测试数据基于 1000 条对话记录）：

JSON 原始数据
大小：1.2MB
压缩耗时：0ms
MessagePack
大小：860KB
压缩耗时：15ms
Protobuf
大小：780KB
压缩耗时：25ms

建议：对延迟敏感选 MessagePack，对带宽敏感选 Protobuf

使用 Locust 模拟不同并发量下的表现（单台 4 核 8G 服务器）：

并发用户数	平均响应时间	错误率
500	62ms	0%
1000	89ms	0.2%
2000	210ms	1.5%

服务端发送间隔建议 25-30 秒
客户端超时设为服务端间隔的 3 倍（如 90 秒）
需要处理网络抖动导致的误判

关键操作需要实现幂等，例如：

def handle_payment(user_id, order_id, amount):
    # 先查询是否已处理过
    if redis.get(f"paid:{order_id}"):
        return True

    # 业务处理...
    redis.setex(f"paid:{order_id}", 86400, "1")  # 24 小时缓存