两周把 AI 助手接入照护流程：Pelago 的无服务器实践与工程边界

2026-07-23 31 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：9 分钟

数字医疗公司面临一个很现实的扩展难题：会员数量增长后，如何继续提供有上下文、有温度的个性化互动，同时不让照护团队被重复沟通压垮。专注于物质使用障碍支持的 Pelago，使用 AWS 构建了 AI 助手，并在两周内将概念推进到实际照护场景。这个案例值得关注的地方，不只是模型能力，而是团队如何用无服务器架构缩短交付路径，并把风险控制嵌入工作流。

两周交付的关键不是“写一个聊天框”

医疗 AI 助手至少横跨四个环节：接收会员消息、获取经过授权的上下文、调用模型生成结果，以及决定结果应直接回复还是交给照护人员。真正困难的是最后一步。

一个通用的处理链路可以拆成：

  1. API 接收消息并完成身份认证。
  2. 工作流读取最小必要的会员上下文。
  3. 模型生成候选回复，并给出风险分类。
  4. 策略层检查危机信号、敏感主题和输出格式。
  5. 低风险内容进入回复流程，高风险内容转交照护团队。
  6. 审计系统记录决策元数据，但避免写入不必要的敏感正文。

无服务器架构适合这种早期产品：团队不必先维护常驻服务器，可以围绕请求、事件和队列逐步增加组件。按调用扩缩容也有助于应对消息量波动。不过，来源摘要只明确提到 Pelago 使用了 AWS，并未列出具体 AWS 服务。下面的 API Gateway、Lambda 和 DynamoDB 组合，是基于该场景给出的可改造实践，不代表 Pelago 的完整实现。

把模型放在受控工作流中

在医疗场景里，模型输出不应天然等同于医疗建议。更稳妥的设计是让模型承担有限任务，例如总结会员表达、起草支持性回复、识别需要人工关注的内容，并由确定性的策略代码决定后续动作。

可以把输出约束为结构化对象：

{
  "risk_level": "low",
  "reason": "No immediate safety signal detected",
  "draft_reply": "Thank you for sharing this. Would you like to tell us what made today especially difficult?",
  "requires_human_review": false
}

结构化输出带来三个直接收益：策略代码不必解析自然语言；每次路由决策可以审计；模型或提示词升级时，可以针对固定字段运行回归测试。

风险边界需要写进系统，而不只是提示词。涉及自伤、过量用药、急性戒断症状或其他紧急情况时，应触发明确的升级路径。提示词可以辅助识别，但不能代替经过验证的临床规则、人工值班机制和当地紧急援助流程。

一个可改造的 AWS Serverless 最小示例

下面假设使用 AWS SAM、API Gateway、Lambda 和 DynamoDB，模型调用位置暂时用确定性函数代替。这样可以先验证鉴权之外的 API 形状、风险路由和数据保留策略，再接入经过组织批准的模型服务。

创建 template.yaml：

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

Resources:
  AssistantFunction:
    Type: AWS::Serverless::Function
    Properties:
      Runtime: python3.12
      Handler: app.handler
      Timeout: 10
      MemorySize: 256
      Environment:
        Variables:
          AUDIT_TABLE: !Ref AuditTable
      Policies:
        - DynamoDBWritePolicy:
            TableName: !Ref AuditTable
      Events:
        MessageApi:
          Type: Api
          Properties:
            Path: /messages
            Method: post

  AuditTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: request_id
          AttributeType: S
      KeySchema:
        - AttributeName: request_id
          KeyType: HASH
      TimeToLiveSpecification:
        AttributeName: expires_at
        Enabled: true

Outputs:
  ApiUrl:
    Value: !Sub 'https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/messages'

再创建 app.py：

import json
import os
import time
import uuid

import boto3

TABLE = boto3.resource("dynamodb").Table(os.environ["AUDIT_TABLE"])
HIGH_RISK_TERMS = {"overdose", "suicide", "self-harm", "can't breathe"}


def classify(message: str) -> dict:
    normalized = message.lower()
    matched = sorted(term for term in HIGH_RISK_TERMS if term in normalized)

    if matched:
        return {
            "risk_level": "high",
            "reason": "Potential urgent safety signal",
            "draft_reply": None,
            "requires_human_review": True,
        }

    return {
        "risk_level": "low",
        "reason": "No configured urgent term detected",
        "draft_reply": (
            "Thank you for sharing this. A care team member can help you "
            "work through what happened."
        ),
        "requires_human_review": False,
    }


def handler(event, context):
    payload = json.loads(event.get("body") or "{}")
    message = str(payload.get("message", "")).strip()
    if not message:
        return {
            "statusCode": 400,
            "headers": {"content-type": "application/json"},
            "body": json.dumps({"error": "message is required"}),
        }

    request_id = str(uuid.uuid4())
    result = classify(message)

    # Only decision metadata is retained in this example, not message content.
    TABLE.put_item(Item={
        "request_id": request_id,
        "risk_level": result["risk_level"],
        "requires_human_review": result["requires_human_review"],
        "created_at": int(time.time()),
        "expires_at": int(time.time()) + 30 * 24 * 60 * 60,
    })

    return {
        "statusCode": 200,
        "headers": {"content-type": "application/json"},
        "body": json.dumps({"request_id": request_id, **result}),
    }

安装 AWS SAM CLI 并配置 AWS 凭据后，可以部署：

sam build
sam deploy --guided

也可以先运行本地 API：

sam build
sam local start-api
curl -s http://127.0.0.1:3000/messages \
  -H 'content-type: application/json' \
  -d '{"message":"I am having a difficult day"}'

这个示例故意不保存会员消息，也没有实现真实模型调用、用户认证和人工工单系统。生产环境必须补齐这些能力，并根据组织政策配置加密、网络隔离、访问控制、审计、数据驻留和保留周期。示例中的关键词匹配只能用于演示，不能作为临床风险分类器。

从原型进入照护流程前要检查什么

两周可以验证产品路径，却不足以证明系统在所有患者群体、表达方式和紧急场景下都可靠。上线策略应当把范围收窄：先处理低风险、可逆、容易人工复核的任务，再依据真实数据扩展自动化程度。

建议在正式采用前检查以下项目：

  • 明确助手能做什么、不能做什么，避免把支持性沟通包装成诊断或治疗。
  • 为高风险消息定义人工升级负责人、响应时限和失败兜底。
  • 使用去标识化测试集评估漏报、误报、语气、偏差和结构化输出稳定性。
  • 对提示词、模型版本、策略规则和知识内容分别进行版本管理。
  • 默认不记录完整提示词和回复，日志只保留排障与审计所需的最小字段。
  • 为超时、限流、模型不可用和队列积压设计降级模式。
  • 监控人工接管率、风险漏报、回复延迟和照护团队负担，而不只看调用量。

Pelago 的案例说明，无服务器基础设施可以显著压缩从概念到可用工作流的时间。但在医疗场景中，交付速度必须与清晰的责任边界同时出现：模型负责生成候选内容，策略层负责限制行为，照护团队保留对高风险决策的控制权。

# AI # AWS # Go # 云原生 # 云平台 # 全栈 # 后端 # 架构设计

相关推荐