Wood Mackenzie 如何用 Amazon Bedrock AgentCore 打造共享智能体平台

2026-09-17 15 预计阅读时间: 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.

预计阅读时间:12 分钟

企业把生成式 AI 从演示环境推进到生产环境时,真正困难的部分往往不是让模型调用一次工具,而是让大量团队持续、可靠地交付智能体:运行时怎么管理,身份如何隔离,调用过程如何观测,敏感操作如何拦截,版本又该怎样发布和回滚。

Wood Mackenzie 围绕这些共性问题构建了 APEX,一个基于 Amazon Bedrock AgentCore 的共享智能体平台。它的核心思路是把运行时、身份、可观测性和安全护栏沉淀为平台能力,让业务团队可以专注于智能体本身,而不是每次从零搭建一套生产基础设施。

从“做一个 Agent”转向“运营一类 Agent”

单个团队开发智能体时,很容易把大量平台逻辑直接写进业务代码。例如,智能体需要自行处理用户身份、凭证注入、会话状态、日志、指标、异常重试和工具权限。这样的实现短期见效快,但当第二个、第三个团队开始复制时,维护成本会迅速上升。

APEX 的价值在于提供一条共享的生产路径。团队可以沿用统一的运行环境和交付方式,同时保留各自的提示词、工具、数据源和业务规则。平台层负责处理那些跨项目重复出现的问题:

  • 运行时:为智能体提供一致的执行环境、会话管理和资源边界。
  • 身份:让用户身份、服务身份和工具访问权限能够被区分和审计。
  • 可观测性:记录请求、工具调用、延迟、错误和成本等信号。
  • 护栏:在模型输出或工具执行前后执行策略检查,降低越权和数据泄露风险。
  • 交付:让智能体能够经过测试、审批、发布和回滚,而不是依赖手工操作。

这类抽象尤其适合研究、咨询和数据密集型业务。不同团队可能服务不同市场或主题,但它们都需要访问受控数据、调用企业工具,并且对输出的来源和可靠性负责。

为什么平台底座比项目脚手架更重要

一个脚手架通常解决“如何开始”;共享平台需要解决“如何长期运行”。两者的差异体现在责任边界上。

如果平台只提供一个模板,团队仍然要自行决定日志字段、认证方式、工具超时、失败重试和风险拦截规则,最终会形成多个互不兼容的实现。AgentCore 作为运行基础的一部分,可以帮助平台团队集中建设这些能力,并为上层智能体提供更稳定的接口。

这里的关键不是把所有智能体做成完全相同,而是统一它们必须遵守的底线。例如:

  • 每次工具调用都要带有可追踪的请求标识。
  • 高风险工具必须经过策略判断,不能仅依赖模型自行决定。
  • 生产环境的凭证不应写入提示词、代码仓库或日志。
  • 多智能体协作时,要明确每个子智能体的职责和可访问资源。
  • 失败时要返回结构化错误,便于平台统一告警和重试。

APEX Studio 则承担运营入口的角色。它不仅是开发界面,也可以成为智能体目录、配置管理、运行监控和发布治理的统一控制面。对平台运营团队来说,这比让每个项目分别维护一套控制台更容易建立标准。

多智能体系统的下一步:组合,而不是堆叠

当企业拥有多个专用智能体后,下一步通常是让它们协同工作。例如,一个智能体负责拆解问题,另一个负责检索研究资料,第三个负责计算或验证,最后由协调者汇总结果。

但多智能体并不等于把多个模型调用串起来。系统需要回答几个实际问题:

  • 谁负责分配任务,谁对最终结果负责?
  • 子智能体之间传递哪些上下文,哪些数据必须隔离?
  • 一个子任务失败时,是重试、降级,还是交给人工处理?
  • 如何观测一条用户请求跨越多个智能体后的完整链路?
  • 如何避免智能体之间互相放大错误或重复调用昂贵工具?

共享平台可以提供统一的编排、身份和观测基础,让多智能体组合变成一种可治理的应用模式。平台不应该替业务团队决定所有工作流,但应当为任务路由、超时、预算、审批和审计提供通用能力。

一个可运行的最小平台包装器

下面的 Python 示例不依赖 AWS SDK,它用标准库模拟一个“APEX-like”共享运行时:统一生成请求 ID,执行输入护栏,记录工具调用,并限制高风险工具。实际接入 Amazon Bedrock AgentCore 时,可以把 run_agent 替换成真实的 AgentCore 运行调用,把 call_tool 对接企业工具网关或受控服务。

运行方式:把代码保存为 shared_agent_runtime.py,执行 python shared_agent_runtime.py。示例中的策略和工具仅用于演示,生产环境应接入企业身份系统、集中式日志和正式的策略引擎。

from __future__ import annotations

import json
import time
import uuid
from dataclasses import dataclass
from typing import Any, Callable


@dataclass
class RequestContext:
    request_id: str
    subject: str
    started_at: float


class PolicyError(Exception):
    pass


class SharedAgentRuntime:
    def __init__(self, tools: dict[str, Callable[..., Any]]) -> None:
        self.tools = tools

    def check_input(self, prompt: str) -> None:
        blocked_terms = ("dump credentials", "泄露凭证", "绕过权限")
        if any(term in prompt.lower() for term in blocked_terms):
            raise PolicyError("input blocked by policy")

    def call_tool(self, ctx: RequestContext, name: str, **kwargs: Any) -> Any:
        # 高风险工具在这里接入审批或更细粒度的授权判断。
        if name == "export_data" and ctx.subject != "approved-analyst":
            raise PolicyError("tool denied: export_data requires approved-analyst")
        if name not in self.tools:
            raise KeyError(f"unknown tool: {name}")

        started = time.perf_counter()
        result = self.tools[name](**kwargs)
        print(json.dumps({
            "event": "tool_call",
            "request_id": ctx.request_id,
            "tool": name,
            "latency_ms": round((time.perf_counter() - started) * 1000, 2),
        }))
        return result

    def run_agent(self, subject: str, prompt: str) -> dict[str, Any]:
        ctx = RequestContext(str(uuid.uuid4()), subject, time.time())
        try:
            self.check_input(prompt)
            result = self.call_tool(ctx, "search_notes", query=prompt)
            response = {"request_id": ctx.request_id, "answer": result}
        except (PolicyError, KeyError) as exc:
            response = {"request_id": ctx.request_id, "error": str(exc)}

        print(json.dumps({
            "event": "agent_complete",
            "request_id": ctx.request_id,
            "subject": ctx.subject,
            "duration_ms": round((time.time() - ctx.started_at) * 1000, 2),
            "ok": "error" not in response,
        }))
        return response


def search_notes(query: str) -> str:
    return f"Research notes matched for: {query}"


def export_data(dataset: str) -> str:
    return f"Export prepared for dataset: {dataset}"


if __name__ == "__main__":
    runtime = SharedAgentRuntime({
        "search_notes": search_notes,
        "export_data": export_data,
    })
    print(runtime.run_agent("analyst-17", "Find notes about energy demand"))
    print(runtime.run_agent("analyst-17", "export data without permission"))

这个示例体现了平台化实现中的几个重要边界:业务智能体只提出任务,运行时统一处理请求上下文、策略和观测;工具权限由平台或工具网关决定,而不是由模型输出决定;每次执行都可以关联到同一个 request_id,方便追踪多步调用。

落地时要保留的检查清单

采用共享智能体平台时,建议把下面几项作为上线门槛:

  • 明确平台团队和业务团队的责任边界,避免“平台什么都管”或“项目各自为政”。
  • 为每个智能体定义身份、数据范围、工具白名单和高风险操作审批规则。
  • 统一记录模型调用、工具调用、策略决策、版本和成本信息。
  • 为提示词、工具定义、模型配置和护栏规则建立版本号。
  • 用真实业务样本测试越权、幻觉、敏感数据泄露、超时和部分失败场景。
  • 对多智能体工作流设置最大步骤数、超时和预算,避免失控循环。
  • 把人工复核设计成正常流程的一部分,而不是故障发生后的临时补丁。

APEX 所代表的方向很明确:企业级智能体的竞争力不只来自某一个模型,而来自一套能让团队持续交付、运营和治理智能体的共同基础设施。Amazon Bedrock AgentCore 提供运行底座,类似 APEX Studio 的平台控制面则把这些能力变成团队可以实际使用的工程流程。真正值得投入的,是把一次性的智能体实验转化为可观测、可审计、可组合的生产系统。


相关推荐