Agent人机交互图解：从架构设计到生产环境实战

1次阅读

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

传统人机交互系统常面临以下核心问题：

状态同步困难：用户会话状态分散在内存、数据库或不同服务中，难以保证强一致性
并发能力弱：同步阻塞式处理导致高延迟，无法有效利用系统资源
异常恢复复杂：中断的交互流程缺乏可追溯性，重启后状态重建成本高
扩展性差：新增交互逻辑需修改核心代码，违反开闭原则

stateDiagram-v2
    [*] --> Idle
    Idle --> Processing: 接收用户输入
    Processing --> Waiting: 发起异步操作
    Waiting --> Processing: 收到外部事件
    Processing --> Idle: 返回最终响应
    Processing --> Error: 发生异常
    Error --> Processing: 重试成功
    Error --> Idle: 放弃任务

关键组件说明：

事件总线：采用发布 / 订阅模式解耦各处理模块
上下文存储：使用带版本号的快照记录会话状态
超时控制器：自动清理长时间无响应的会话

from dataclasses import dataclass
from typing import Optional
import time

@dataclass
class SessionContext:
    session_id: str
    current_state: str  
    last_active: float = time.time()
    retry_count: int = 0
    # 使用版本号解决并发更新冲突
    version: int = 0

import asyncio
from redis.asyncio import Redis

class EventHandler:
    def __init__(self, redis: Redis):
        self.redis = redis
        # 指数退避参数（单位：秒）self.BACKOFF_FACTOR = [1, 2, 4, 8, 16]  

    async def handle_event(self, event: dict):
        """
        处理事件的核心方法
        :param event: 包含 type/data 等字段的字典
        :return: 处理结果状态码
        """
        try:
            ctx = await self._load_context(event['session_id'])
            if not ctx:
                return 404

            # 状态机路由逻辑
            handler = getattr(self, f"on_{event['type']}", None)
            if not handler:
                return 400

            return await handler(ctx, event['data'])
        except Exception as e:
            await self._schedule_retry(event, str(e))
            return 500

    async def _schedule_retry(self, event: dict, error: str):
        """实现指数退避的重试机制"""
        retry_count = event.get('retry_count', 0)
        if retry_count >= len(self.BACKOFF_FACTOR):
            await self._dead_letter_queue(event)
            return

        delay = self.BACKOFF_FACTOR[retry_count]
        await self.redis.xadd(
            'retry_events',
            {'payload': json.dumps(event), 'error': error},
            id=f"{time.time()}-{retry_count}"
        )

测试环境：4 核 8G 云服务器，Redis 6.2

模式	并发用户数	平均响应时间(ms)	最大 QPS
同步阻塞	100	320	312
异步非阻塞	100	85	1,178
异步 + 批量处理	100	62	1,614

关键优化手段：

使用 uvloop 替代默认事件循环
批量合并 Redis 写入操作
采用连接池管理数据库连接

监控指标 ：设置xlen(event_stream) 告警阈值
横向扩展：动态增加消费者组实例数量
降级策略：非关键消息转存到 S3 异步处理

async def clean_stale_sessions(redis: Redis, timeout: int = 3600):
    """清理超过 1 小时未活动的会话"""
    now = time.time()
    async for key in redis.scan_iter("session:*"):
        last_active = await redis.hget(key, "last_active")
        if last_active and now - float(last_active) > timeout:
            await redis.delete(key)