Agent项目实战：从零构建高可用智能代理系统

1次阅读

没有评论

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

在开发智能代理系统时，我们常常遇到几个典型问题：

架构耦合 ：业务逻辑、网络通信、数据处理代码混杂在一起，导致系统难以维护和扩展
响应延迟 ：同步阻塞式处理无法应对高并发请求，用户等待时间过长
状态管理混乱 ：多任务并行时状态同步困难，容易出现脏读 / 脏写问题
容错性差 ：缺乏完善的异常处理机制，单个任务失败可能导致整个系统崩溃

Python 生态中有多种并发处理方案可选，我们对比了三种主流方案：

Celery：适合分布式任务队列场景，但需要额外中间件（RabbitMQ/Redis）
Ray：专为 AI 负载设计，集群部署较复杂
Asyncio：Python 原生异步 IO 库，轻量级且学习曲线平缓

考虑到我们的 Agent 系统需要快速迭代和简单部署，最终选择基于 Asyncio 构建核心异步处理引擎。

我们采用经典的三层架构：

接口层 ：处理 HTTP/gRPC 请求，参数校验和协议转换
逻辑层 ：核心业务逻辑实现，包含任务编排和状态管理
数据层 ：对接数据库和缓存系统，处理数据持久化

下面是一个完整的异步任务处理器实现：

import asyncio
from typing import Dict, Any
import aiohttp

class AsyncTaskHandler:
    def __init__(self, redis_conn):
        self.redis = redis_conn
        self.session = aiohttp.ClientSession()

    async def process_task(self, task_id: str, params: Dict[str, Any]):
        """
        处理单个异步任务
        :param task_id: 任务唯一标识
        :param params: 任务参数
        """
        try:
            # 1. 更新任务状态为运行中
            await self.redis.set(f"task:{task_id}:status", "running")

            # 2. 执行实际业务逻辑
            result = await self._execute_business_logic(params)

            # 3. 保存结果并更新状态
            await self.redis.set(f"task:{task_id}:result", result)
            await self.redis.set(f"task:{task_id}:status", "completed")

            return result
        except Exception as e:
            # 异常处理
            await self.redis.set(f"task:{task_id}:status", "failed")
            await self.redis.set(f"task:{task_id}:error", str(e))
            raise

    async def _execute_business_logic(self, params):
        """示例业务逻辑：调用外部 API 并处理结果"""
        async with self.session.get(
            "https://api.example.com/data",
            params=params,
            timeout=10
        ) as resp:
            if resp.status == 200:
                return await resp.json()
            raise ValueError(f"API 请求失败: {resp.status}")

    async def close(self):
        """清理资源"""
        await self.session.close()

使用 Redis 作为中央状态存储，确保多进程间状态一致：

import aioredis

class StateManager:
    def __init__(self, redis_url):
        self.redis = aioredis.from_url(redis_url)

    async def get_task_state(self, task_id):
        """获取任务状态"""
        return await self.redis.get(f"task:{task_id}:status")

    async def set_task_state(self, task_id, state):
        """设置任务状态（带过期时间）"""
        pipeline = self.redis.pipeline()
        pipeline.set(f"task:{task_id}:status", state)
        pipeline.expire(f"task:{task_id}:status", 86400)  # 24 小时过期
        await pipeline.execute()

我们对比了同步和异步模式处理 1000 个网络请求的耗时：

模式	并发数	总耗时 (s)	平均延迟 (ms)
同步	10	45.2	452
异步	100	5.7	57

异步模式性能提升近 8 倍。

HTTP 连接池 ：复用 aiohttp.ClientSession，避免重复创建连接
Redis 连接池 ：使用 aioredis 连接池，推荐配置：
最大连接数 = 预计并发数 × 1.2
空闲连接超时 = 300 秒

分布式环境下必须保证任务只执行一次：

async def safe_process_task(self, task_id):
    """确保任务幂等执行"""
    # 使用 Redis 的 SETNX 原子操作
    locked = await self.redis.setnx(f"task:{task_id}:lock", "1")
    if not locked:
        raise ValueError("任务正在执行中")

    try:
        return await self.process_task(task_id)
    finally:
        await self.redis.delete(f"task:{task_id}:lock")

任务重试 ：对可重试异常自动重试 3 次
死信队列 ：将失败任务存入特殊队列供人工处理
健康检查 ：定期检测并重启异常工作进程

使用 pytest 测试异步代码：

@pytest.mark.asyncio
async def test_task_processing():
    """测试任务处理流程"""
    redis = aioredis.from_url("redis://localhost")
    handler = AsyncTaskHandler(redis)

    try:
        result = await handler.process_task("test123", {"param": "value"})
        assert isinstance(result, dict)

        status = await redis.get("task:test123:status")
        assert status == b"completed"
    finally:
        await handler.close()
        await redis.close()

当前系统已支持单 Agent 高效运行，要扩展到多 Agent 协作场景，我们需要考虑：