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

1次阅读

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

传统脚本与 Agent 系统的核心差异在于 持续交互能力。脚本通常一次性执行完成就退出，而 Agent 需要长期运行并处理动态请求。新手常遇到这些挑战：

并发控制：同时处理多个用户请求时，容易出现资源竞争或响应延迟
状态管理：对话上下文在多轮交互中如何持久化（比如购物车状态保持）
容错处理：第三方 API 失败时如何优雅降级而不导致服务雪崩

适用场景：固定流程的业务（如银行 IVR 系统）
性能指标：
QPS：500-1000（纯内存操作）
延迟：<50ms
内存占用：约 100MB/ 万规则

适用场景：开放域对话（如客服机器人）
性能指标：
QPS：5-20（依赖模型大小）
延迟：200-2000ms
GPU 内存：6GB 起（7B 参数模型）

# 路由决策示例
def route_request(user_input):
    if "账户余额" in user_input:  # 规则优先
        return rule_engine.query(user_input)
    else:  # 其他走 LLM
        return llm.generate(user_input)

import asyncio
from circuitbreaker import circuit

class BaseAgent:
    def __init__(self):
        self.task_queue = asyncio.Queue()
        self.context_cache = {}  # 用户 ID -> 对话上下文

    @circuit(failure_threshold=3, recovery_timeout=60)  # 熔断机制
    async def handle_request(self, user_id, message):
        """处理用户请求的入口方法"""
        try:
            context = self._get_context(user_id)
            task = asyncio.create_task(self._process(message, context))
            await self.task_queue.put(task)
            return await task
        except Exception as e:
            self._fallback_response(e)  # 优雅降级

    async def _process(self, message, context):
        # 实际业务逻辑
        pass

import random
from functools import wraps

def retry_with_backoff(retries=3, max_delay=4):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(retries):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == retries - 1:
                        raise
                    delay = min((2 ** attempt) + random.random(), max_delay)
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

# locustfile.py
from locust import HttpUser, task

class AgentUser(HttpUser):
    @task
    def test_chat(self):
        self.client.post("/chat", 
            json={"user_id": "test", "message": "hello"})

运行命令：locust -f locustfile.py --headless -u 1000 -r 100

监控 asyncio 任务的正确方式：

import tracemalloc

tracemalloc.start()

# 定期执行
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
for stat in top_stats[:10]:
    print(stat)  # 打印内存占用 Top10

对话状态丢失
现象：用户多轮交互后历史记录消失
解决：使用 Redis 持久化上下文，设置 TTL 过期时间
死锁
现象：Agent 完全停止响应

解决：给所有 async 操作添加超时装饰器

async with async_timeout.timeout(5):  # 最多等待 5 秒
    await some_io_operation()

API 限流
现象：第三方接口返回 429 错误

解决：实现令牌桶算法限流

from pyrate_limiter import RateLimiter
limiter = RateLimiter(100, 60)  # 每分钟 100 次

扩展功能：为 Agent 增加 OAuth2.0 鉴权模块

# TODO 实现步骤
1. 安装 authlib 库：pip install authlib
2. 创建路由 /auth/token 获取 access_token
3. 在 BaseAgent 中添加 @requires_auth 装饰器
4. 验证示例：curl -H "Authorization: Bearer {token}" http://agent/chat

通过本文的实践方案，我们构建了一个具备熔断保护、自动重试、状态管理的 Agent 系统核心框架。建议在实际部署时：