构建高可用Agent Skills Marketplace的技术架构与实战

1次阅读

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

近年来，随着 AI Agent 的普及，技能市场（Skills Marketplace）成为连接开发者与用户的重要平台。但在实际运营中，这类平台常面临三个核心问题：

技能发现效率低 ：传统分类检索无法满足长尾需求，用户常需要翻页多次才能找到合适技能
匹配精准度不足 ：简单的标签匹配导致大量无关结果，影响用户体验
高并发场景崩溃 ：热门技能上线时突发流量经常导致服务不可用

graph TD
    A[客户端] --> B[API Gateway]
    B --> C[技能发现服务]
    B --> D[交易服务]
    B --> E[用户服务]
    C --> F[推荐引擎]
    D --> G[支付网关]
    E --> H[认证服务]
    F --> I[技能图谱]

服务拆分原则
按业务能力划分：发现、交易、用户三个核心域
每个服务独立数据库，通过事件总线同步关键数据

事件驱动实现

# 技能发布事件示例
class SkillPublishedEvent:
    def __init__(self, skill_id, categories):
        self.timestamp = datetime.utcnow()
        self.skill_id = skill_id
        self.categories = categories  # 用于推荐系统实时更新

# 事件处理器
@event_listener('SkillPublished')
def update_recommendations(event):
    vector = generate_skill_vector(event.categories)
    redis.zadd('skill:trending', {event.skill_id: vector.score})

采用改进的 TF-IDF+ 余弦相似度计算：

def calculate_similarity(query, skill):
    """
    query: 用户搜索词分词后的列表
    skill: 技能对象的特征字典
    返回 0 - 1 之间的匹配分数
    """
    # 特征加权（标题权重最高）weights = {'title': 0.6, 'description': 0.3, 'tags': 0.1}

    # 多字段联合计算
    total_score = 0
    for field, weight in weights.items():
        field_text = ' '.join(skill[field]) if isinstance(skill[field], list) else skill[field]
        tfidf = TfidfVectorizer().fit_transform([field_text, query])
        total_score += cosine_similarity(tfidf[0:1], tfidf[1:2])[0][0] * weight

    # 热度衰减因子（防止老技能长期霸榜）age_penalty = 1 / (1 + log(1 + (now() - skill['publish_time']).days))
    return total_score * age_penalty

关键状态机设计：

type OrderState string

const (
    Created    OrderState = "CREATED"
    Paid       OrderState = "PAID"
    Delivered  OrderState = "DELIVERED"
    Completed  OrderState = "COMPLETED"
    Cancelled  OrderState = "CANCELLED"
)

func (s *OrderService) HandlePayment(ctx context.Context, orderID string) error {
    // 使用分布式锁防止重复处理
    lock := redis.NewLock(fmt.Sprintf("lock:order:%s", orderID))
    if !lock.Acquire(5 * time.Second) {return errors.New("operation in progress")
    }
    defer lock.Release()

    // 状态机校验
    currentState := s.repo.GetState(orderID)
    if currentState != Created {return fmt.Errorf("invalid state transition from %s", currentState)
    }

    // 更新状态并发出领域事件
    if err := s.repo.UpdateState(orderID, Paid); err != nil {return err}
    eventbus.Publish(OrderPaidEvent{OrderID: orderID})
    return nil
}

客户端缓存 ：静态技能数据通过 ETag 实现 304 响应
边缘缓存 ：使用 Cloudflare Workers 缓存热门搜索结果

服务端缓存 ：

# 带击穿保护的 Redis 缓存装饰器
def cache_protected(key_fn, ttl=300):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            cache_key = key_fn(*args, **kwargs)

            # 先尝试获取缓存
            cached = redis.get(cache_key)
            if cached is not None:
                return json.loads(cached)

            # 防止缓存击穿
            lock_key = f"lock:{cache_key}"
            if not redis.setnx(lock_key, 1, ex=5):
                time.sleep(0.1)
                return wrapper(*args, **kwargs)

            try:
                result = fn(*args, **kwargs)
                redis.setex(cache_key, ttl, json.dumps(result))
                return result
            finally:
                redis.delete(lock_key)
        return wrapper
    return decorator

user_db_0     user_db_1    # 按 user_id 哈希分片
│            │
└─ order_0   └─ order_1   # 子表按订单创建时间范围分区

// 关键校验逻辑
func ValidateToken(token string) (*Claims, error) {
    // 使用 JWKS 轮转机制
    keyFunc := func(token *jwt.Token) (interface{}, error) {kid := token.Header["kid"].(string)
        return jwks.GetKey(kid)
    }

    claims := &Claims{}
    _, err := jwt.ParseWithClaims(token, claims, keyFunc)
    return claims, err
}

行为指纹分析（设备 +IP+ 行为模式）

滑动窗口限流：

# 基于 Redis 的滑动窗口计数器
def check_rate_limit(user_id, action, limit=10, window=60):
    now = int(time.time())
    key = f"rate:{user_id}:{action}"

    # 使用管道保证原子性
    with redis.pipeline() as pipe:
        pipe.zadd(key, {now: now})
        pipe.zremrangebyscore(key, 0, now - window)
        pipe.zcard(key)
        _, _, count = pipe.execute()

    return count <= limit

事件顺序问题 ：
解决方案：在每个事件中添加单调递增的 sequence_id
使用 Kafka 分区键保证同技能事件的顺序性

分布式事务 ：

# 最终一致性模式示例
def complete_order(order_id):
    try:
        # 1. 本地事务更新订单状态
        with transaction.atomic():
            order = Order.objects.select_for_update().get(id=order_id)
            order.status = 'COMPLETED'
            order.save()

        # 2. 异步触发结算
        celery.send_task('settle_commission', args=[order_id])
    except Exception as e:
        # 3. 补偿机制
        dlq.send({
            'order_id': order_id,
            'error': str(e)
        })