Agent实现代码案例：从架构设计到生产环境部署的完整指南

1次阅读

没有评论

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

在构建电商推荐 Agent 时，我们遇到了几个典型问题：

状态持久化难题：用户浏览行为需要实时更新推荐策略，但服务重启会导致状态丢失
消息乱序风险：用户快速操作可能触发并发请求，导致最终推荐结果不符合预期
超时重试困境：第三方 API 调用失败时，简单的重试可能引发雪崩效应

这些痛点让我们的初版 Agent 平均每两天就需要人工干预一次，严重影响了用户体验。

我们对比了两种主流方案：

状态机方案：
优点：实现简单，适合确定性强的工作流
缺点：难以处理用户行为的突发流量
Actor 模型：
优点：天然隔离状态，每个用户独立 Actor
缺点：需要额外处理消息队列

最终选择 Actor 模型，因为电商场景中用户行为具有强独立性。以下是核心架构图：

graph TD
    A[用户请求] --> B[路由 Dispatcher]
    B --> C[用户 Actor1]
    B --> D[用户 Actor2]
    C --> E[状态存储]
    D --> E

from typing import Any, Dict
from dataclasses import dataclass
import asyncio

@dataclass
class Message:
    sender: str
    content: Any

class BaseActor:
    def __init__(self, actor_id: str):
        self.actor_id = actor_id
        self._state = {}
        self._mailbox = asyncio.Queue()

    async def handle_message(self, message: Message) -> None:
        """必须由子类实现的具体处理逻辑"""
        raise NotImplementedError

    async def run(self):
        while True:
            msg = await self._mailbox.get()
            try:
                await self.handle_message(msg)
            except Exception as e:
                print(f"Actor {self.actor_id} error: {str(e)}")

class RecommendationActor(BaseActor):
    def __init__(self, user_id: str):
        super().__init__(user_id)
        self._state['view_history'] = []
        self._state['last_recommend'] = None

    async def handle_message(self, message: Message) -> None:
        if message.content['type'] == 'view_item':
            # 更新浏览历史（限制最大长度）self._state['view_history'].append(message.content['item_id'])
            self._state['view_history'] = self._state['view_history'][-10:]

            # 生成新推荐（模拟耗时操作）await asyncio.sleep(0.1)
            rec_items = self._generate_rec()
            self._state['last_recommend'] = rec_items

            # 返回推荐结果
            await self._send_response(message.sender, rec_items)

    def _generate_rec(self) -> List[str]:
        """基于最近浏览生成推荐"""
        history = self._state['view_history']
        if not history:
            return ['default_item1', 'default_item2']

        # 模拟推荐算法
        return [f'rec_{history[-1]}_1', f'rec_{history[-1]}_2']

并发用户数	平均延迟(ms)	P99 延迟(ms)	QPS
100	45	112	2200
500	83	215	6000
1000	142	403	7000

使用 tracemalloc 定期检查：

import tracemalloc

def check_memory():
    snapshot = tracemalloc.take_snapshot()
    top_stats = snapshot.statistics('lineno')
    for stat in top_stats[:10]:
        print(stat)

关键发现：Actor 邮箱堆积未处理消息会导致 OOM，解决方案：

# 在 BaseActor 中添加邮箱监控
async def run(self):
    while True:
        if self._mailbox.qsize() > 100:  # 阈值
            print(f"WARN: Actor {self.actor_id} mailbox overflow")
        # ... 原有逻辑...

现象：偶尔出现推荐结果错乱
原因：在协程中直接操作共享状态
解决：所有状态修改必须发生在 Actor 主循环中

现象：无法关联用户行为链条

解决：注入全链路 trace_id

class Message:
    def __init__(self, sender: str, trace_id: str, content: Any):
        self.sender = sender
        self.trace_id = trace_id  # 新增
        self.content = content

现象：推荐服务整体雪崩

解决：引入熔断机制

from circuitbreaker import circuit

class RecommendationActor(BaseActor):
    @circuit(failure_threshold=3, recovery_timeout=60)
    async def _call_third_party(self):
        # 调用外部 API
        pass