用 Amazon Bedrock AgentCore 自建代码级评估器：从金融情报 Agent 的四个实战案例说起

2026-05-18 33 预计阅读时间：1 分钟
来源：aws.amazon.com AI 摘要 原文链接

免责声明：本文为 AI 摘要整理，建议结合原文阅读。摘要可能省略上下文、版本差异或边界条件，不作为官方说明。

预计阅读时间：14 分钟

金融市场的 AI Agent 一旦上线，回答质量直接关系到交易决策和合规风险。Amazon Bedrock AgentCore 最近开放了自定义代码评估器（custom code-based evaluator）能力，让你可以用 Lambda 函数对 Agent 输出做细粒度、可编程的质检——不再只能依赖内置的通用指标。

这篇文章围绕一个金融市场情报 Agent，拆解四个 Lambda 评估器的实现思路、注册流程、运行模式，以及如何把自定义评估器和内置评估器、其他 AWS 服务组合起来，形成完整的质检流水线。

评估器在 AgentCore 中的角色

AgentCore 的评估器本质上是一段你控制的代码，接收 Agent 的输入和输出，返回一个结构化的评估结果。它和内置评估器（比如相关性、毒性检测）的区别在于：你完全掌控逻辑，可以写业务规则、调外部 API、做事实核查。

两种运行模式决定了评估器何时介入：

  • On-demand 模式：手动触发，适合开发阶段批量跑回归测试，验证 Agent 在已知场景下的表现。
  • Online 模式：随 Agent 每次推理实时执行，适合生产环境持续监控，发现异常立即告警。

金融场景里，两种模式都需要——开发时用 on-demand 做回归，上线后用 online 做实时风控。

四个 Lambda 评估器的实现

下面以金融情报 Agent 为例，实现四个评估器。每个评估器都是一个独立的 Lambda 函数，核心逻辑写在 handler 里。

1. 事实核查评估器——调用外部知识库验证回答的准确性

金融数据时效性极强，Agent 说"纳斯达克昨日收于 18500"必须被实时验证。这个评估器调用 Amazon OpenSearch 或 RAG 知识库做 grounded fact-checking。

import json
import boto3
import re

opensearch = boto3.client('opensearchserverless')
bedrock_kb = boto3.client('bedrock-agent-runtime')

def extract_numeric_claims(text: str) -> list[dict]:
    """从 Agent 回答中提取包含数字的断言"""
    patterns = [
        r'(纳斯达克|标普500|道琼斯|S&P\s*500|Nasdaq|Dow\s*Jones)[^。]*?(收于|涨至|跌至|达到|报)\s*([\d,\.]+)',
        r'(利率|通胀率|GDP增速|失业率)[^。]*?(为|是|达到|升至|降至)\s*([\d,\.]+%?)',
    ]
    claims = []
    for p in patterns:
        for m in re.finditer(p, text):
            claims.append({
                "entity": m.group(1),
                "action": m.group(2),
                "value": m.group(3),
                "original": m.group(0)
            })
    return claims

def verify_claim(claim: dict) -> dict:
    """用 Bedrock Knowledge Base 检索最新数据验证断言"""
    query = f"{claim['entity']} 最新 {claim['action']} 数据"
    resp = bedrock_kb.retrieve(
        knowledgeBaseId="YOUR_KB_ID",   # 替换为你的 KB ID
        retrievalQuery={"text": query}
    )
    latest_value = None
    for r in resp.get("retrievalResults", []):
        content = r["content"]["text"]
        nums = re.findall(r'([\d,\.]+)', content)
        if nums:
            latest_value = nums[0]
            break

    if latest_value and claim["value"].replace(",", "") == latest_value.replace(",", ""):
        return {"claim": claim["original"], "verified": True, "detail": f"知识库确认值: {latest_value}"}
    return {"claim": claim["original"], "verified": False, "detail": f"Agent声称 {claim['value']}, 知识库显示 {latest_value}"}

def lambda_handler(event, context):
    agent_response = event.get("agentResponse", "")
    claims = extract_numeric_claims(agent_response)
    results = [verify_claim(c) for c in claims]

    verified_count = sum(1 for r in results if r["verified"])
    total = len(results) if results else 1

    return {
        "evaluatorName": "FactCheckEvaluator",
        "score": verified_count / total,
        "details": results,
        "passed": verified_count == total
    }

运行前需要替换 YOUR_KB_ID 为你自己的 Bedrock Knowledge Base ID，并确保 Lambda 执行角色有 bedrock-agent-runtime:Retrieve 权限。

2. PII 检测评估器——防止个人隐私信息泄露

金融分析中可能意外包含客户姓名、身份证号、账户信息。这个评估器调用 Amazon Comprehend 做 PII 检测。

import boto3

comprehend = boto3.client('comprehend')

PII_TYPES_CRITICAL = ["SSN", "BANK_ACCOUNT_NUMBER", "CREDIT_DEBIT_CARD_NUMBER", "NAME"]

def lambda_handler(event, context):
    agent_response = event.get("agentResponse", "")
    if not agent_response:
        return {"evaluatorName": "PIIDetector", "score": 1.0, "passed": True, "details": []}

    resp = comprehend.detect_pii_entities(
        Text=agent_response,
        LanguageCode='zh'  # 中文文本; 英文用 'en'
    )

    findings = []
    for entity in resp["Entities"]:
        if entity["Type"] in PII_TYPES_CRITICAL:
            findings.append({
                "type": entity["Type"],
                "score": entity["Score"],
                "offset": [entity["BeginOffset"], entity["EndOffset"]]
            })

    return {
        "evaluatorName": "PIIDetector",
        "score": 0.0 if findings else 1.0,   # 发现 PII 则评分 0
        "passed": len(findings) == 0,
        "details": findings
    }

Lambda 角色需要 comprehend:DetectPiiEntities 权限。Comprehend 对中文 PII 的检测能力有限，如果你的场景以中文为主，可以补充正则匹配逻辑作为兜底。

3. 实时告警评估器——高风险回答触发 SNS 通知

当 Agent 输出包含特定关键词（如"暴跌""崩盘""紧急卖出"）或事实核查失败时，立即推送告警到运维和合规团队。

import boto3

sns = boto3.client('sns')
ALERT_TOPIC_ARN = "arn:aws:sns:us-east-1:123456789012:AgentAlertTopic"  # 替换为你的 Topic ARN

HIGH_RISK_KEYWORDS = ["暴跌", "崩盘", "紧急卖出", "恐慌性抛售", "crash", "panic sell"]

def lambda_handler(event, context):
    agent_response = event.get("agentResponse", "")
    fact_check_result = event.get("previousEvaluators", {}).get("FactCheckEvaluator", {})

    triggered_keywords = [kw for kw in HIGH_RISK_KEYWORDS if kw in agent_response]
    fact_failed = not fact_check_result.get("passed", True)

    should_alert = len(triggered_keywords) > 0 or fact_failed

    if should_alert:
        reason_parts = []
        if triggered_keywords:
            reason_parts.append(f"高风险关键词: {triggered_keywords}")
        if fact_failed:
            reason_parts.append("事实核查未通过")
        sns.publish(
            TopicArn=ALERT_TOPIC_ARN,
            Subject="⚠️ 金融情报 Agent 输出告警",
            Message=f"触发原因: {'; '.join(reason_parts)}\n\nAgent 回答: {agent_response[:500]}"
        )

    return {
        "evaluatorName": "RealTimeAlertEvaluator",
        "score": 0.0 if should_alert else 1.0,
        "passed": not should_alert,
        "details": {
            "triggeredKeywords": triggered_keywords,
            "factCheckFailed": fact_failed,
            "alertSent": should_alert
        }
    }

替换 ALERT_TOPIC_ARN 为你自己的 SNS Topic。Lambda 角色需要 sns:Publish 权限。

4. 金融合规格式评估器——检查回答是否包含必要的免责声明

合规要求金融建议必须附带免责声明和风险提示。这个评估器纯逻辑判断，不需要外部服务。

import re

DISCLAIMER_PATTERNS = [
    r'免责声明',
    r'仅供参考',
    r'不构成投资建议',
    r'风险提示',
    r'investmen[t]\s*ris[k]',
    r'disclaimer',
]

def lambda_handler(event, context):
    agent_response = event.get("agentResponse", "")
    has_disclaimer = any(re.search(p, agent_response, re.IGNORECASE) for p in DISCLAIMER_PATTERNS)

    # 检查是否包含具体投资建议（触发更严格的免责要求）
    advice_patterns = [r'建议买入', r'建议卖出', r'推荐持有', r'recommend\s+buy', r'should\s+invest']
    has_advice = any(re.search(p, agent_response, re.IGNORECASE) for p in advice_patterns)

    score = 1.0
    if has_advice and not has_disclaimer:
        score = 0.0   # 有建议但没免责声明，严重违规
    elif not has_disclaimer:
        score = 0.5   # 没有免责声明但也没明确建议，中等风险

    return {
        "evaluatorName": "ComplianceFormatEvaluator",
        "score": score,
        "passed": score >= 0.8,
        "details": {
            "hasDisclaimer": has_disclaimer,
            "hasInvestmentAdvice": has_advice,
            "severity": "critical" if (has_advice and not has_disclaimer) else "moderate" if not has_disclaimer else "none"
        }
    }

注册评估器到 AgentCore

每个 Lambda 写好后，需要通过 AgentCore 的 API 注册。以下是用 AWS CLI 注册的命令：

# 注册事实核查评估器
aws bedrock-agentcore create-evaluator \
  --evaluator-name "FactCheckEvaluator" \
  --evaluator-type "CODE_BASED" \
  --code-based-evaluator-config \
    lambdaArn="arn:aws:lambda:us-east-1:123456789012:function:FactCheckEvaluator" \
  --description "验证金融数据断言的准确性" \
  --region us-east-1

# 注册 PII 检测评估器
aws bedrock-agentcore create-evaluator \
  --evaluator-name "PIIDetector" \
  --evaluator-type "CODE_BASED" \
  --code-based-evaluator-config \
    lambdaArn="arn:aws:lambda:us-east-1:123456789012:function:PIIDetector" \
  --description "检测回答中的个人隐私信息" \
  --region us-east-1

# 注册实时告警评估器
aws bedrock-agentcore create-evaluator \
  --evaluator-name "RealTimeAlertEvaluator" \
  --evaluator-type "CODE_BASED" \
  --code-based-evaluator-config \
    lambdaArn="arn:aws:lambda:us-east-1:123456789012:function:RealTimeAlertEvaluator" \
  --description "高风险输出触发 SNS 告警" \
  --region us-east-1

# 注册合规格式评估器
aws bedrock-agentcore create-evaluator \
  --evaluator-name "ComplianceFormatEvaluator" \
  --evaluator-type "CODE_BASED" \
  --code-based-evaluator-config \
    lambdaArn="arn:aws:lambda:us-east-1:123456789012:function:ComplianceFormatEvaluator" \
  --description "检查免责声明和合规格式" \
  --region us-east-1

把 lambdaArn 中的账号 ID 和区域替换成你自己的。

运行评估：On-demand 与 Online

注册完成后，两种模式的使用方式不同。

On-demand 批量评估——适合开发回归测试：

aws bedrock-agentcore start-evaluation-run \
  --evaluator-ids "FactCheckEvaluator,PIIDetector,ComplianceFormatEvaluator" \
  --agent-id "your-financial-agent-id" \
  --input-data-config \
    s3Uri="s3://your-bucket/eval-test-cases.json" \
  --mode ON_DEMAND \
  --region us-east-1

eval-test-cases.json 是你准备的标准测试集，包含多组 Agent 输入和预期输出。评估完成后结果写入 S3，可以拉下来做分析。

Online 实时评估——随 Agent 推理自动执行：

aws bedrock-agentcore attach-online-evaluators \
  --agent-id "your-financial-agent-id" \
  --evaluator-ids "FactCheckEvaluator,PIIDetector,RealTimeAlertEvaluator,ComplianceFormatEvaluator" \
  --region us-east-1

绑定后，每次 Agent 被调用，四个评估器会依次执行。评估结果写入 CloudWatch Logs，告警评估器还会直接推送 SNS。

自定义评估器 + 内置评估器的组合策略

AgentCore 内置了相关性（Relevance）、完整性（Completeness）、毒性（Toxicity）等通用评估器。实际部署中，最佳做法是分层组合：

层级 评估器 作用 运行模式
通用层 内置 Relevance + Toxicity 过滤明显跑题和有害内容 Online
业务层 自定义 FactCheck + ComplianceFormat 验证金融数据准确性和合规格式 Online + On-demand
安全层 自定义 PIIDetector + RealTimeAlert 阻断隐私泄露、触发告警 Online

组合注册命令：

aws bedrock-agentcore attach-online-evaluators \
  --agent-id "your-financial-agent-id" \
  --evaluator-ids "BUILT_IN_RELEVANCE,BUILT_IN_TOXICITY,FactCheckEvaluator,PIIDetector,RealTimeAlertEvaluator,ComplianceFormatEvaluator" \
  --region us-east-1

评估器之间可以传递结果——RealTimeAlertEvaluator 的代码中已经展示了读取 previousEvaluators 中 FactCheckEvaluator 结果的做法，这样告警逻辑可以基于前序评估器的判断做组合决策。

上线前的检查清单

把自定义评估器投入生产前，逐项确认：

  • Lambda 冷启动延迟：Online 模式下评估器串行执行，四个 Lambda 冷启动叠加可能增加 2-4 秒延迟。用 Provisioned Concurrency 消除冷启动，或者把轻量评估器（如合规格式）合并到一个 Lambda 里。
  • 权限最小化：每个 Lambda 只授予它需要的权限——Comprehend 检测、SNS 发布、Bedrock KB 检索，不要用大权限角色。
  • 评估结果存储：Online 模式的结果默认进 CloudWatch Logs。如果需要长期分析，加一步把结果写入 S3 或 OpenSearch，方便做趋势报表。
  • 失败容错：单个评估器 Lambda 报错不应阻断整个推理流程。AgentCore 默认隔离评估器异常，但你要确认自己的 Agent 调用代码不会因为评估结果缺失而崩溃。
  • 测试集维护：On-demand 评估的测试集要随市场变化更新——上季度的指数数据不能用来验证本季度的 Agent 回答。

自定义代码评估器把 Agent 质检从"通用指标评分"推进到了"业务规则编程"。金融场景只是起点，医疗合规、法律咨询、客服风控——任何需要业务级判断的 Agent，都可以用同样的模式构建自己的质检层。

# AI # AWS # LLM # 云平台 # 机器学习

相关推荐