在 AWS 上构建多智能体客服：应对体育博彩的地域规则、实时风控与流量洪峰

from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class SupportRequest:
    message: str
    state: str
    customer_id: str


def responsible_gaming_agent(req: SupportRequest) -> str:
    return (
        "Your request has been routed to the responsible gaming workflow. "
        "Do not provide promotional or wagering guidance."
    )


def jurisdiction_agent(req: SupportRequest) -> str:
    return f"Look up the approved betting rules for state={req.state}."


def account_agent(req: SupportRequest) -> str:
    return f"Query the permitted account-support view for customer={req.customer_id}."


def general_agent(req: SupportRequest) -> str:
    return "Search the approved customer-support knowledge base."


AGENTS: dict[str, Callable[[SupportRequest], str]] = {
    "responsible_gaming": responsible_gaming_agent,
    "jurisdiction": jurisdiction_agent,
    "account": account_agent,
    "general": general_agent,
}

HIGH_RISK_TERMS = {
    "self exclude",
    "self-exclude",
    "stop gambling",
    "betting too much",
    "控制不住投注",
    "自我排除",
}


def route(req: SupportRequest) -> tuple[str, str]:
    text = req.message.lower()

    # 硬安全规则必须先于模型分类执行。
    if any(term in text for term in HIGH_RISK_TERMS):
        agent_name = "responsible_gaming"
    elif any(term in text for term in ("legal", "allowed", "state rule", "州规则")):
        agent_name = "jurisdiction"
    elif any(term in text for term in ("deposit", "withdraw", "account", "提款", "账户")):
        agent_name = "account"
    else:
        agent_name = "general"

    return agent_name, AGENTS[agent_name](req)


if __name__ == "__main__":
    request = SupportRequest(
        message="I am betting too much and want to self-exclude",
        state="MA",
        customer_id="cust_123",
    )
    selected_agent, action = route(request)
    print({"agent": selected_agent, "action": action})

python3 support_router.py

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31

Resources:
  SupportQueue:
    Type: AWS::SQS::Queue
    Properties:
      VisibilityTimeout: 90
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt SupportDeadLetterQueue.Arn
        maxReceiveCount: 3

  SupportDeadLetterQueue:
    Type: AWS::SQS::Queue

  SupportRouter:
    Type: AWS::Serverless::Function
    Properties:
      Runtime: python3.12
      Handler: app.handler
      Timeout: 30
      MemorySize: 1024
      ReservedConcurrentExecutions: 100
      Environment:
        Variables:
          SUPPORT_QUEUE_URL: !Ref SupportQueue
      Policies:
        - SQSSendMessagePolicy:
            QueueName: !GetAtt SupportQueue.QueueName
      Events:
        Api:
          Type: Api
          Properties:
            Path: /support/messages
            Method: post