Agent的设计与实现：从零构建高可用的智能代理系统

1次阅读

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

构建生产级智能代理系统时，开发者往往会遇到以下几个关键挑战：

状态一致性维护：Agent 需要在分布式环境下保持状态同步，避免数据不一致
消息异步处理：高并发场景下消息的顺序性和可靠性难以保证
故障恢复机制：系统崩溃后如何快速恢复服务并保证数据完整性

特性	Actor 模型	状态机
并发模型	天然并发（每个 Actor 独立运行）	需要外部调度
状态管理	内部封装	显式状态转换
扩展性	易于水平扩展	需额外设计分布式协调
调试难度	较难（异步交互复杂）	相对简单（状态可追踪）

import asyncio
from collections import deque

class AsyncTaskQueue:
    def __init__(self, max_size=1000):
        self._queue = deque(maxlen=max_size)
        self._event = asyncio.Event()

    async def put(self, item):
        self._queue.append(item)
        self._event.set()

    async def get(self):
        while not self._queue:
            await self._event.wait()
            self._event.clear()
        return self._queue.popleft()

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), 
       wait=wait_exponential(multiplier=1, min=4, max=10))
async def rpc_call(endpoint, payload):
    async with aiohttp.ClientSession() as session:
        async with session.post(endpoint, json=payload) as resp:
            if resp.status >= 500:
                raise RuntimeError(f"Server error: {resp.status}")
            return await resp.json()

import pickle
from datetime import datetime

class StateSnapshot:
    @staticmethod
    async def save(agent_state, path):
        snapshot = {'timestamp': datetime.utcnow().isoformat(),
            'state': agent_state
        }
        with open(path, 'wb') as f:
            pickle.dump(snapshot, f)

    @staticmethod
    async def load(path):
        with open(path, 'rb') as f:
            return pickle.load(f)

import cProfile

async def main():
    profiler = cProfile.Profile()
    profiler.enable()

    # 你的 Agent 主逻辑
    await agent.run()

    profiler.disable()
    profiler.print_stats(sort='cumtime')

对象池应用：对频繁创建销毁的对象使用对象池

from object_pool import ObjectPool

class ConnectionPool:
    def __init__(self):
        self.pool = ObjectPool(
            create_func=self._create_conn,
            max_size=100
        )

    async def _create_conn(self):
        return await create_database_connection()

避免循环引用：使用 weakref 处理对象间引用

import jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"

def create_token(payload):
    return jwt.encode({
        **payload,
        'exp': datetime.utcnow() + timedelta(minutes=30)
    }, SECRET_KEY, algorithm="HS256")

def verify_token(token):
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except jwt.PyJWTError:
        return None

from collections import defaultdict
import time

class RateLimiter:
    def __init__(self, max_calls, period):
        self.max_calls = max_calls
        self.period = period
        self.calls = defaultdict(list)

    async def check(self, client_id):
        now = time.time()
        calls = self.calls[client_id]

        # 移除过期记录
        calls = [t for t in calls if t > now - self.period]
        self.calls[client_id] = calls

        if len(calls) >= self.max_calls:
            raise RuntimeError("Rate limit exceeded")

        calls.append(now)
        return True