Agent实现代码案例：从零构建高可用智能代理系统

1次阅读

没有评论

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

传统轮询架构在实时交互场景中存在几个明显缺陷：

资源利用率低：频繁的空轮询消耗大量 CPU 资源
响应延迟高：无法立即处理突发请求
状态管理困难：分布式环境下难以保证一致性

Agent 模式通过事件驱动架构解决了这些问题：

实时响应：基于消息触发立即执行
资源高效：仅在需要时激活处理单元
状态明确：每个 Agent 维护独立上下文

核心组件交互流程：

消息网关接收外部请求
路由层分发到指定 Agent
处理器解析消息内容
状态机维护会话上下文
任务队列处理异步操作

import asyncio
from abc import ABC, abstractmethod

class BaseAgent:
    """Agent 抽象基类"""
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self._state = {}
        self._queue = asyncio.Queue()

    async def run(self):
        """主事件循环"""
        while True:
            message = await self._queue.get()
            await self._process_message(message)

    @abstractmethod
    async def _process_message(self, message):
        """消息处理模板方法"""
        pass

import redis
import pickle

class StatefulAgent(BaseAgent):
    """带持久化状态的 Agent"""
    def __init__(self, agent_id, redis_conn):
        super().__init__(agent_id)
        self.redis = redis_conn
        self._load_state()

    def _load_state(self):
        serialized = self.redis.get(f'agent:{self.agent_id}')
        if serialized:
            self._state = pickle.loads(serialized)

    async def _save_state(self):
        serialized = pickle.dumps(self._state)
        await self.redis.setex(f'agent:{self.agent_id}', 
            3600,  # TTL 1 小时
            serialized
        )

模式	QPS	平均延迟	95 分位延迟
单线程	1200	85ms	120ms
线程池 (4)	3800	42ms	75ms
线程池 (8)	5200	38ms	65ms

优化建议：

IO 密集型任务使用线程池
CPU 密集型任务控制并发数
设置合理的队列容量

实现确认机制：

async def safe_process(msg):
    try:
        await process(msg)
        await msg.ack()
    except Exception:
        await msg.nack()

消息去重表结构：

CREATE TABLE message_dedup (msg_id VARCHAR(64) PRIMARY KEY,
    processed BOOLEAN DEFAULT FALSE,
    timestamp TIMESTAMP
);

采用 Snowflake 算法：

import time

class Snowflake:
    def __init__(self, worker_id):
        self.worker_id = worker_id
        self.sequence = 0
        self.last_timestamp = -1

    def generate(self):
        timestamp = int(time.time() * 1000)
        if timestamp == self.last_timestamp:
            self.sequence = (self.sequence + 1) & 0xFFF
            if self.sequence == 0:
                timestamp = self._wait_next_millis()
        else:
            self.sequence = 0

        self.last_timestamp = timestamp
        return ((timestamp & 0x1FFFFFFFFFF) << 22) | 
               ((self.worker_id & 0x3FF) << 12) | 
               (self.sequence & 0xFFF)