躬行笔记 » AI 响应卡顿像便秘？AWS API Gateway 流式传输救命指南，从架构到代码全解析

兄弟们，咱们今天不聊虚的，来聊点让每个人都头疼的事儿——等待。

你不管是做运维还是做开发，肯定都遇到过这种场景：老板兴冲冲地让你搞个 AI Agent，吹得天花乱坠，什么“智能客服”、“自动化分析”。你在本地环境（Localhost）一跑，嘿，行云流水，字符一个个蹦出来，跟黑客帝国似的，看着特爽。

结果呢？一部署到生产环境，完了。

用户问一句：“这月报表咋样？”
然后就是漫长的沉默。
屏幕上那个 Loading 圈转啊转，转得人心慌。5秒，10秒，20秒... 就在用户以为系统挂了准备关网页的时候，“啪”一下，几千字的一大坨回复突然全弹出来了。

这体验，说实话，跟便秘好不容易通了似的，虽然结果有了，但过程极其痛苦。

为啥？因为生产环境得套壳啊！你总不能把 Agent 的运行时直接裸奔在公网上吧？那安全团队不得拿着刀来找你？咱们得套上 API Gateway，得加上 WAF 防火墙，得搞身份认证（Auth）。

但问题就出在这儿。传统的 API Gateway 就像个强迫症，它非得把后端 Agent 吐出来的所有数据都接完了、打包好了，才肯一次性吐给前端用户。这就是所谓的“缓冲（Buffering）”。对于普通 API 没毛病，但对于生成式 AI 这种一边想一边吐字的场景，简直就是灾难。

要么裸奔换速度，要么安全换卡顿？

以前是这样，没得选。但最近 AWS 终于干了件人事儿，API Gateway 支持响应流（Response Streaming）了！这意味着咱们既能加上企业级的防护罩，又能让字符像流水一样实时推给用户。

今天这篇长文，我就带大家把这个坑彻底填平。咱们从架构设计、代码实现，一直聊到那些文档里不会告诉你的“深坑”。

准备好咖啡，咱们开整。

咱们到底要搭个什么架构？

咱们先别急着贴代码，先把脑子里的图画清楚。

你现在手头有一个 Agent，这玩意儿在本地跑得很欢。现在我们要把它扔到 AWS 上，并且要做到“既要又要”——既要安全（Authentication & WAF），又要快（Streaming）。

这套架构的核心其实就四个角儿：

用户（User）：就是那个等着看字儿的可怜人。
Cognito：这是咱们的门卫大爷，负责发证（Token）。
API Gateway：这是公司大门，负责查证，并且最关键的是，它得是个“也就是个门框”，不能挡着里面递出来的东西（开启流式传输）。
AgentCore Runtime：这是咱们干活的大脑，藏在最里面。

流程大概是这么个野路子：

用户先去找 Cognito 刷脸，拿个 ID Token（注意，这里有个大坑，后面细说）。然后用户拿着这个 Token 去敲 API Gateway 的门。API Gateway 瞅一眼 Token，行，是真的，放行。然后它把请求转给里面的 AgentCore Runtime。这个 Runtime 再查一遍 Token（双重保险，稳如老狗），然后开始干活，干出一个字就往外吐一个字。

这时候，最神奇的事情发生了：API Gateway 不再攒着了，它收到一个字就往外扔一个字。

这就是我们要达到的效果。看着简单？那是你还没开始配 CDK，等你配的时候就知道啥叫“细节是魔鬼”了。

这里的四个关键“机关”

要想这套流水线跑通，有四个地方必须拧紧了，松一个都不行。

1. 别拿 Access Token 忽悠人，要用 ID Token

这是很多兄弟容易搞混的地方。做 OAuth2 或者 OIDC 认证的时候，手里通常有一把 Token。

Cognito 的 API Gateway Authorizer（授权器），它认死理儿，它要的是 ID Token。为啥？因为 ID Token 里才带着用户的身份信息（比如 sub 字段，就是用户的唯一标识）。Access Token 是给资源服务器用的，但在咱们这个特定的网关认证场景下，API Gateway 更喜欢看 ID Token 来验明正身。

所以，让你前端的小伙伴注意了，请求头里得这么写：
Authorization: Bearer <你的_ID_Token>

如果你塞个 Access Token 进去，网关大概率会给你甩个 401 Unauthorized，到时候你查半天日志都不知道为啥，别问我怎么知道的，说多了都是泪。

而且，这还是个纵深防御（Defense in Depth）的设计。API Gateway 验一次，里面的 AgentCore Runtime 还要再验一次。这就像进小区保安拦一下，进单元门还得刷次卡，安全感拉满。

2. 找对门牌号：/invocations 端点

咱们用的这个 AgentCore Runtime，它不是随便哪个 URL 都能接流式请求的。它有一个专门的“接待室”，叫 /invocations。

这个端点是专门为 OAuth2 设计的，而且它天生就支持“异步生成器模式（Async Generator Pattern）”。

你的 URL 拼出来应该长这样：
https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{runtime_id}/invocations?qualifier=DEFAULT&accountId={account}

别自作聪明去调别的接口，别的接口要么不支持流式，要么认证方式不对。就认准这个 /invocations。它能处理那种跑得死慢死慢的长连接，这对 AI 来说太重要了。

3. 开启“直通模式”：Response Transfer Mode

这是本文最最最核心的一点。

API Gateway 默认是个“囤积癖”，它喜欢把后端返回的数据存满了再发。我们要做的，就是强行把它的这个毛病改过来。

但是！AWS CDK 目前的高级封装（L2 Construct）里，居然还没有直接暴露这个参数！ 这种事在 AWS 也是见怪不怪了，新功能出来，CDK 总是慢半拍。

这时候，咱们就得用点“黑魔法”了——Escape Hatch（逃生舱口）。我们要直接操作底层的 CloudFormation 资源。

代码得这么写（Python CDK 也就是这味儿）：

# 咱们得先拿到底层的 CfnMethod 对象
cfn_method = post_method.node.default_child

# 然后，强行给它打个补丁
cfn_method.add_property_override("Integration.ResponseTransferMode", "STREAM")

没有这两行代码，你的 API Gateway 就还是那个囤积癖。你会发现后端明明已经在 yield 数据了，前端还是一片空白，直到最后那一瞬间。加上这两行，世界瞬间通透了。

4. 代码得会“喘气”：返回异步生成器

基础设施配好了，咱们的业务代码（Agent 代码）也得配合。

如果你在代码里直接 return result，那前面的努力全白费。return 意味着“我话说完了”，那肯定是一次性的。

我们要用 yield。在 Python 里，配合 async，这就是异步生成器。

看看这个对比：

错误的写法（憋气版）：

@app.entrypoint
async def invoke(payload, context):
    # 这就像是一口气把话全憋肚子里
    result = await agent.run(prompt)
    return result

正确的写法（喘气版）：

@app.entrypoint
async def invoke(payload, context):
    # 定义一个内部函数，专门负责一点一点往外吐
    async def generate_stream():
        # 假设 agent.stream_async 是个能吐字的方法
        async for chunk in agent.stream_async(prompt):
            yield chunk
            
    # 注意，这里返回的是函数本身，不是执行结果！
    return generate_stream()

Runtime 看到你返回的是个生成器，它就懂了：“噢，这兄弟要开始表演了。”然后它就会自动建立流式连接，把你 yield 出来的每一个 chunk 实时传输出去。

为什么这套架构能成？

咱们来盘道盘道，这套方案到底香在哪。

其实做运维架构，最难的永远是 Trade-off（权衡）。安全团队想要 WAF，想要严防死守；产品经理想要用户体验丝般顺滑；开发团队想要少写点鉴权代码。

这套架构把这三方的嘴都堵上了：

安全没丢：WAF 依然在 API Gateway 上挡着，SQL 注入、XSS 攻击根本进不来。身份验证用了 Cognito，还是双重校验，稳得一批。
体验拉满：流式传输让用户感觉 AI 反应极快（虽然其实也是在逐字生成，但看着动就不心慌）。
突破极限：这还有个隐藏福利——超时时间。
普通的 API Gateway 请求，29秒不返回就超时断开了。对于稍微复杂点的 AI 任务（比如又要联网搜索，又要读 PDF，还要写总结），29秒根本不够用。
但是！开启流式传输后，这个硬限制放宽到了 15分钟（只要连接保持活跃）。这就给 Agent 留出了大把的时间去思考人生、去处理复杂逻辑，完全不用担心被网关掐断。

除此之外，API Gateway 那些自带的监控、日志、流控（Throttling），全都白送。咱们不用自己去造轮子搞什么 Nginx 配置，省心省力。

踩坑实录：我是怎么浪费了两个晚上的

虽然理论很完美，但落地的时候，现实总是会给我几个大嘴巴子。为了不让兄弟们重蹈覆辙，我把自己踩过的坑列出来。

坑一：忘了那个该死的 Property Override

我第一次部署的时候，自以为 CDK 写得很溜，用 HttpIntegration 一把梭。结果部署完，前端还是转圈圈。我查了后端日志，明明看到数据在生成啊？

我抓包，抓 TCP 包，查 CloudWatch，折腾到凌晨两点。最后才发现，API Gateway 控制台里，那个 Integration 的设置里，"Content Handling" 还是默认的。

就是因为 CDK 没原生支持那个参数，我漏写了 add_property_override("Integration.ResponseTransferMode", "STREAM")。加上之后，重部署，秒出。那一刻，我想砸键盘。

坑二：URL 瞎拼

刚开始我以为只要是 Runtime 的 URL 就能流式。结果我把 URL 写成了 .../runtimes/{id}/chat（这是我自己瞎编的一个路径），结果报 404 或者直接被拒绝。

一定要敬畏官方文档（虽然 AWS 文档有时候也挺烂），/invocations 才是唯一的真神。而且后面那个 qualifier=DEFAULT 最好也带上，省得它有时候抽风找不到版本。

坑三：Python 的异步陷阱

在写 Agent 代码时，如果你在 async for 循环里调用了一个同步的阻塞操作（比如用 requests 库发请求，而不是 aiohttp），那整个流式传输会被卡住！

因为 Python 的 asyncio 是单线程的 Event Loop。你一个同步操作卡住，整个 Loop 就停了，yield 根本发不出去。一定要确保你的 Agent 内部调用的所有 I/O 操作都是 awaitable 的。

深度复盘：数据到底是怎么流过去的？

为了让大家理解得更透彻，咱们把显微镜拿出来，看看当用户发个“你好”的时候，底层到底发生了啥。

握手阶段：
用户前端拿到 Cognito 的 Token，向 API Gateway 发起 HTTP POST 请求。这时候，HTTP 头里不仅仅有 Authorization，还会协商传输编码。
第一道关卡：
API Gateway 的 Authorizer 拦截请求，解析 JWT Token。签名对不对？过期没？aud（受众）是不是我？一切 OK，放行。
连接建立：
API Gateway 与后端的 AgentCore Runtime 建立连接。注意，因为开启了 STREAM 模式，API Gateway 会告诉客户端：“老铁，我要用 Transfer-Encoding: chunked 了啊，我不告诉你内容多长，反正我有就发。”
Runtime 接棒：
Runtime 收到请求，再次校验 Token。然后调用你的 Python invoke 函数。
数据蹦迪：
你的代码开始 yield "你"。
Runtime 捕获到这个字符，把它封装成一个 chunk，推给 API Gateway。
API Gateway 根本不存，转手就推给用户的浏览器。
浏览器收到 chunk，JavaScript 处理一下，屏幕上蹦出个“你”字。
紧接着 yield "好"... 如此循环。

这就是一个完整的链条。你会发现，身份验证在两头，流式传输贯穿中间。

哪怕是神方案，也有约束

咱们搞运维的，不能只报喜不报忧。这方案虽然好，但也有几个硬指标你得心里有数：

1. 空闲超时（Idle Timeout）
这就是我前面说的“你要时不时喘口气”。
如果是区域性（Regional）或者私有端点，5分钟内没有任何数据传输，连接就会断。
如果是边缘优化（Edge-optimized）端点，更短，只有 30秒。
所以，如果你的 Agent 思考时间特别长（比如在跑一个复杂的 SQL 分析），记得中间发点“心跳包”或者“正在思考中...”的占位符，别让连接凉了。

2. 带宽限制
AWS 也是要过日子的。
前 10MB 的数据，随便流，不限速。
超过 10MB 之后，速度会被限制在 2MB/s。
不过讲道理，对于文本聊天机器人，你要是能聊出 10MB 的纯文本，那用户估计也没耐心看了。除非你在流式传输图片或者音频，否则这个限制基本可以忽略。

3. 这一套搞不了的事儿
因为开启了流式，有些 API Gateway 的传统艺能就失效了：

VTL 模板转换：你想在网关层修改响应体？不行，流式数据改不了，因为网关根本不知道完整的响应长啥样。
响应缓存：也没法缓存了，毕竟是实时的。
自动 Gzip 压缩：流式数据的压缩机制不一样，普通的压缩配置可能不生效。

总结一下

这一顿操作下来，我们实际上是完成了一次架构的进化。

我们不再把 Agent 当作一个传统的 REST API 来对待，而是把它看作一个实时的事件流生成器。通过配置 API Gateway 的 STREAM 模式，配合 Cognito 的安全机制，我们在 AWS 上成功搭建了一个既符合企业安全合规，又具备互联网极致体验的 AI 服务。

别小看这几秒钟的优化。在 AI 时代，用户的耐心是以毫秒计算的。能让字符在屏幕上跳动起来，你的服务就已经赢了一半。

这就是运维的价值，兄弟们。不是搬服务器，而是用架构的力量，解决最实际的痛点。

如果你对文中提到的 CDK 代码细节，或者 Cognito 的配置还有疑问，或者你在部署过程中遇到了什么奇葩的报错，欢迎在评论区留言，咱们一起避坑。

最后，搞技术的都懂，收藏夹里吃灰不如转发出去装杯。
如果你觉得这篇文章帮你省了几个小时的调试时间，麻烦点个赞、点个在看。

更硬核的云原生、SRE、DevOps 实战经验，请关注 @运维躬行录，咱们下期见！

AI 响应卡顿像便秘？AWS API Gateway 流式传输救命指南，从架构到代码全解析