Strand Agents Harness SDK：把Agent循环做成生产级运行时

上周四深夜，我在给一套客服自动化系统做线上故障复盘。用户反馈说 Agent 答非所问，我打开服务端日志，发现问题出在一个已经运行了三周的工具调用上——外部天气接口返回格式变了，Agent 的循环在这个工具上报错后没有按预期恢复，而是在同一个死循环里重试了 11 次，最后输出了一段"我暂时无法回答"的兜底文案。

这事最讽刺的地方在于：我们当时已经用了某个主流 Agent 框架，但它只保证了"状态编排"这一层。工具调用的超时、异常恢复、并发隔离、幂等重试，全都是我在框架外面手写补丁补上去的。这个感受，就是我这篇开源项目周更系列第 227 篇要聊 Strands Agents Harness SDK 的第一个原因——它把 Agent 循环外面那层没人管的运行时，做成了可以直接拿进生产的 SDK，而且入口简单到接近一行代码。

下面我从设计动机、核心抽象、生产化实践到迁移建议，完整拆一遍这个项目。

1. 手写 Agent 循环的代价：那层没人在意的运行时

1.1 一个 600 行 agent_loop.py 的教训

在我完全不借助任何框架的早期版本，核心循环长这样：

async def agent_loop(query: str): messages = [{"role": "system", "content": SYSTEM_PROMPT}] messages.append({"role": "user", "content": query}) for step in range(MAX_STEPS): resp = await llm.chat(messages, tools=TOOLS) if resp.stop_reason == "tool_calls": for call in resp.tool_calls: try: result = await execute_tool(call) except TimeoutError: result = default_error_result(call, "timeout") messages.append(format_tool_results(resp)) continue return resp.content

看起来短短二十来行，对吧？真正上线之后，我在外面套了一层又一层"生产补丁"：

• 并发补丁：三个用户同时发起对话时，同一个messages变量会被多处协程直接复用，互相污染，最后答非所问。我得给每个会话单独建一份上下文实例，还要在入口加信号量限制同一时刻最多处理多少个任务。
• 超时补丁：某个第三方工具接口默认不设超时，一次不返回就占住一个 worker，直到网关把它杀掉。我需要对每一种工具设置不同的超时阈值。
• 幂等补丁：向工单系统创建工单这类操作如果因网络抖动重试，会产生重复工单。我必须给所有"会产生副作用"的工具手动加幂等键。
• 死循环补丁：LLM 反复调用同一个失败的搜索工具时，如果没有最大步数限制，单次会话可以空转十几分钟，白白烧掉 token。
• 审计补丁：用户投诉"你怎么把我的信息发错了"，我连"当时模型到底调用了哪些工具、参数是什么"都拿不到完整记录，因为日志只打了 info 级别。

这些补丁加起来，围绕核心循环我写了六百多行辅助代码。诚实地说，那六百行里真正跟业务相关的不到三分之一，剩下的全是通用运行时的东西——重试、隔离、限流、恢复、审计。这部分恰好就是 Strands Agents Harness SDK 想替你解决的事情。

1.2 Harness 与 Agent 框架的真实分工

先澄清概念。很多人把 Agent 框架和 Agent Harness 混着用，但它们解决的不是同一层问题。

我习惯用这个类比：Agent 框架是汽车的动力和传动系统，决定发动机怎么把力气传到轮子上，对应 LangGraph、Agent SDK 这类工具负责的状态机、节点流转；而 Harness 是车身、仪表盘、安全带和故障诊断系统，它不改变 Agent 内部逻辑，但保证这辆车在路上安全稳定地跑，出了故障能告警、能记录、能兜底。

在更早的工程语境里，test harness 指的是"给被测对象套一个稳定运行环境的那层夹具"。Agent Harness 借用的是同一个意思：给 Agent 套一层可靠的运行环境。

所以你会看到，Harness SDK 的核心 API 往往不是"怎么定义节点、怎么连边"，而是"怎么声明运行策略、怎么注入隔离、怎么暴露事件流、怎么持久化状态"。它对 LLM 本身没有强绑定，对工具也无感知，它关心的是 Agent 执行环境。

我整理了一个对比表，顺便说明为什么很多团队用框架打了原型之后，还得再上 Harness：

当然这里不是踩一捧一。Strands Harness SDK 自己也依赖了不少成熟框架的解码逻辑，它更多是"补上运行时那一层"，而不是要替代所有编排工具。想清楚了这一点，你就知道什么时候该选它，什么时候其实用不到。

2. Strands Harness SDK 的设计骨架：把循环协议化

2.1 核心抽象：Spec、Runtime、ToolExecutor、MemoryShard

Strands Harness SDK 通过"声明式 spec + 运行时执行"来工作。这个设计最大的好处是：业务代码和运行策略解耦。业务方只需要把 Agent 的输入输出描述清楚，运行时的并发、重试、沙箱、持久化都可以通过配置调整，不需要改核心代码。

下面这几个抽象是理解它行为的关键，接口命名以我接触的 v0.6.x 版本为例，不同版本可能调整，但思路一致。

• HarnessSpec（运行规格）：一份声明式配置，描述 Agent 的模型来源、工具清单、记忆策略、并发上限、最大步数、超时策略。它是运行时一切默认值的来源。
• HarnessRuntime（运行时执行器）：根据 Spec 构建的执行主体，内部维护任务队列、循环调度、事件发布、检查和持久化。对外暴露run()和stream()。
• ToolExecutor（工具执行沙箱）：每个工具在注册时声明自己的权限、超时、重试次数、是否幂等等属性。真正调用外部工具时，由它统一包一层执行逻辑。
• MemoryShard（记忆分片）：把系统提示词、用户消息、工具调用结果分别存放在不同的记忆切片中，避免互相污染；支持滑动窗口、摘要压缩等淘汰策略。

我在本地跑了一个最小配置，看起来就是这样：

# agent.yaml name: research-agent model: provider: openai-compatible model: deepseek-v3 timeout: 60s tools: - name: web_search description: 搜索引擎检索，用于获取最新信息 timeout: 30s retries: 2 idempotent: true permissions: [network-read] - name: send_email description: 发送邮件，必须有明确收件人 timeout: 10s retries: 1 idempotent: false permissions: [email-send] memory: strategy: sliding-window max_turns: 20 runtime: max_steps: 15 concurrency: 8

这段 yaml 没有任何业务逻辑，但它已经决定了这个 Agent 的绝大多数运行行为。业务代码里只需要写工具函数本体，其余交给 Runtime。

2.2 从手写循环到一行代码：build() 之后发生了什么

当你完成了 spec，真正启动 Agent 的入口非常短：

import asyncio from strands_agents.harness import Harness async def main(): app = await Harness.from_spec("agent.yaml").build() response = await app.run("查一下最近三天竞品发布了哪些新版本，整理成清单") print(response.output) asyncio.run(main())

看到没有，手写版里最痛苦的for step in range(MAX_STEPS)、超时包裹、失败恢复、记忆注入，全部被Harness.from_spec().build()封装掉了。

但"一行代码"不是黑魔法。它背后的机制，按顺序大概是：

1. 启动时解析 spec，构建工具注册表，把所有工具的 json_schema 和权限信息加载进 ToolExecutor；
2. 初始化 MemoryShard，预置 system prompt 分片与用户分片；
3. run() 之后进入循环：取一条用户消息、放入 user 分片，然后派发 LLM 调用；如果模型返回需要调用工具，就到 ToolExecutor 里执行，执行结果写回 tool_result 分片；判断循环是否应该结束（达到 max_steps、模型输出非工具意图、或者出现不可恢复错误）；
4. 整个过程向外发布事件流，方便业务方做追踪和审计。

这段流程我在源码里逐行看过。相比手写循环，它的关键区别不是"用魔法替代逻辑"，而是把所有通用逻辑抽象成了可配置、可观测、可替换的组件。对一个要长期维护的系统来说，这比在业务代码里堆 if/else 干净得多。

我还发现它支持 stream 模式，可以实时观察 Agent 在干什么：

async for event in app.stream("帮我把下周的计划同步到日历里"): if event.type == "tool_call": print(f"[工具] {event.tool_name}，参数: {event.arguments}") elif event.type == "llm_output": print(f"[模型] {event.text[:80]}...")

线上排查时这种事件流很有用。你不一定需要理解 Runtime 内部每个细节，只要订阅它的事件，就能知道 Agent 走到哪一步了。

3. 真正的生产级：五道框架不帮你过的坎

这一章重点讲讲，为什么很多 Agent 项目"能跑"和"能上线"的距离那么远。以下五个问题，几乎每一个都是手写循环时最容易忽略、又最容易在线上引爆的点。

3.1 并发：AI Agent 怎么扛住真实流量

先说结论：Agent 的并发难点不在"能同时跑多少个任务"，而在"怎么在 LLM 调用和工具调用各自不同的耗时下，仍然保持稳定的吞吐和错误隔离"。

假设一个用户请求平均需要 5 次 LLM 调用和 3 次工具调用。LLM 调用耗时可能 2~10 秒，工具调用可能 0.1~30 秒不等。如果你用最朴素的"来一个请求开一个协程"的方式，系统会在某些慢工具上堆积大量协程，最后把上游 API 限流打爆。

我实际生产中的做法是分桶限流：

import asyncio from strands_agents.harness import Harness llm_sem = asyncio.Semaphore(4) # 同一时刻最多 4 个并发 LLM 调用 tool_sem = asyncio.Semaphore(16) # 工具调用可以更宽一些 async def guarded_run(app, query: str): async with llm_sem: return await app.run(query) async def main(): app = await Harness.from_spec("agent.yaml").build() queries = [...] # 线上待处理的请求 results = await asyncio.gather(*(guarded_run(app, q) for q in queries))

为什么给 LLM 调用单独一个小的信号量？因为同一个模型供应商 API 在并发超过一定数量后，基本都会返回 429 限流或触发内部排队，这时再盲目加大并发只会把错误率拉高。而工具调用大多是内部系统或普通 HTTP 服务，相对更耐受并发，所以给它单独的、更宽的额度。

如果你不用 SDK 而是自己手写循环，记住这个原则同样有用：不要用一个统一的并发阈值管所有调用类型。并发控制必须按下游服务的实际承受能力来划分，而不是按我们希望接收多少用户请求来划分。

3.2 工具调用的权限边界：别让 LLM 拿着万能钥匙

Agent 接入的工具越多，权限失控的风险越大。一个很常见但很危险的写法是：在工具函数里直接读取环境变量里的 API Key，或者给所有工具函数挂一个requests.Session，让 Agent 能访问任意 URL。开发环境里跑通没问题，一旦上线，LLM 的输出是不可完全预测的，你没法保证模型不会根据某种提示词注入，要求工具去访问一个内网地址。

Harness 的处理方式是：工具权限在注册时就声明，运行时在沙箱里帮你拦截。我通常会给工具声明这几类权限：

在配置里，每个工具只能拿到它声明的最小权限，而不是整个 Agent 的大盘权限。比如web_search只给network-read，它就拿不到你的数据库连接串；send_email只给email-send，它也无法读取服务器上的本地文件。

这一点看似简单，真到生产环境就非常关键。之前我见过一个团队，Agent 在跟用户闲聊时，因为系统提示词里混入了一段被污染的文本，模型竟然主动调用了"读取服务器环境变量"的工具，打算把配置发到外部。如果当时没有权限沙箱，这就是一次严重的数据泄露；有了权限声明之后，ToolExecutor 会在执行前就拒绝并且记入审计日志。

注意，secrets-read这类权限不是说完全不能用，而是必须做到最小化、可审计。一个负责运维的 Agent 可能需要读取部署机器的关键信息，但绝不应该在同一个对话中同时拥有"读取 secret"和"网络发送"两个权限，因为你无法预测模型会不会把它们组合起来造成数据外发。这种权限组合拳在安全设计里一定要一并考虑。

3.3 状态持久化与记忆边界：别把所有东西都塞给模型

记忆是 Agent 最容易做乱的地方。很多手写循环会直接把工具调用结果 append 到 messages 里，再连带 system prompt 一起发回模型。短期看没问题，但会产生两个隐患：

1. 上下文窗口被工具输出填满，真实用户指令的权重被稀释；
2. 工具输出（比如网页全文甚至有恶意的页面文本）直接进入了 system prompt 所在的位置，等于让不可信数据影响模型对系统角色的理解。

MemoryShard 的核心思路是分片。它把 system_prompt、user_conversation、tool_result 放在不同的 shard 里，并且支持不同的淘汰和压缩策略。举个典型配置：

memory: shards: - name: system_prompt strategy: fixed max_tokens: 1000 - name: tool_result strategy: summarize max_tokens: 2000 - name: conversation strategy: sliding-window max_turns: 20

这样一来，工具调用结果即使非常大，也只会在注入模型前被压缩到 2000 token 以内；system_prompt 永远保持固定内容；会话历史不会被无限拉长，窗口永远只保留最近 20 轮。

同时，生产级还要求崩溃可恢复。Agent 在工具执行过程中可能因为进程重启、网络断连而中断，如果记忆和未完成的工具调用没有落盘，你就只能让用户重新发一遍。Strands 的 Runtime 支持 checkpoint 机制，我会配置定时持久化，运行中每隔一段时间把当前状态存下来。这样即使进程被 kill，也可以从最近一次 checkpoint 恢复，而不是从头开始。

单纯"能记住"不算本事，"记住之后不会串台、不会被撑爆、丢了还能恢复"才是生产级记忆的价值。

3.4 可观测性与审计追踪：没有 trace 的 Agent 没法排查

Agent 应用的排查难度，比传统后端高一个数量级。因为一次用户请求，可能经历多次 LLM 调用和多次工具调用，而每次 LLM 调用都是概率性的，同样的输入不一定有同样的输出。这意味着你无法通过重放请求来复现问题，必须依赖当时的完整上下文快照。

所以我特别看重 Harness 的事件流设计。每一个run()周期都会生成一个 run_id，工具调用、LLM 调用、记忆更新、异常恢复都会带上这个 run_id 发布事件。我可以把这些事件接入标准日志或 OTLP，形成完整 trace。

我的建议是至少记录以下几类事件：

• 每次 LLM 调用：请求的 prompt 摘要、模型返回的完整文本、token 数、耗时；
• 每次工具调用：工具名称、参数（敏感字段脱敏）、返回结果摘要、耗时、重试次数；
• 循环决策：为什么这个循环结束（是模型认为可以结束，还是到达 max_steps，还是异常终止）。

有一次排查线上问题，用户说 Agent 给我的报价是错的。我用 run_id 查 trace，很快定位到：在第二轮工具调用中，get_inventory返回了一个价格字段为 0 的异常记录，而模型没有校验这个字段，直接把 0 报给了用户。如果没有事件 trace，这种问题只能靠反复复现去碰运气。对合规要求更高的团队来说，审计日志也是硬性需求——每个工具在什么时间、被谁、以什么参数调用过，都必须留痕。

3.5 重试、超时与失败恢复：别一股脑全丢给模型重来

最后这道坎是关于失败的。手写循环最常见的失败处理是"catch 到异常之后，把错误文本重新发给模型，让它再试一次"。这个做法在演示时很聪明，但上线后经常变成灾难：

• 非幂等操作被重试，产生重复订单；
• 工具一直失败，模型一直重试，token 迅速烧完；
• 单个任务失败没有隔离，把整个服务拖垮。

Harness 的做法是把失败策略拆成可以精细控制的配置项。我在这类配置上踩过不少坑，推荐这样设置：

runtime: max_steps: 15 tool_timeout: 20s llm_timeout: 60s retry_policy: max_retries: 2 backoff: exponential jitter: true failure_handling: on_tool_error: report_to_model # 或 abort max_error_reports: 3 # 最多向模型报告几次工具错误

我特别强调max_steps和max_error_reports这两个值。社区里最常见的 agent execution terminated due to error 这种报错，往往不是模型崩溃，而是循环在某个坏工具上反复重试，直到超过最大步数。限制错误报告次数，可以让 Agent 在工具连续失败几次后主动放弃，而不是继续空转。

超时也要分桶：LLM 请求我一般给 60~120 秒，普通工具给 10~30 秒。如果你用一个统一超时，会遇到工具已经把错误恢复的重任转移到模型层，导致单次请求总时长被拉长的情况。

4. 从自研循环迁移到 Harness：三个最常见的摩擦点

迁移到 Harness 会让生产运行省心很多，但也不是改一行 import 就能完成的。以下三个摩擦点，是我在迁移自己项目、也给一些团队做咨询时最常遇到的。

4.1 工具参数 Schema 的摩擦：docstring 没法直接当 schema 用

如果你是从 LangChain 风格迁移过来的，最坑的往往是工具定义。LangChain 喜欢靠函数签名和 docstring 推导工具参数，而 Strands Harness 这类偏向运行时控制的 SDK，通常希望你显式声明参数模型，因为只有精确的 json_schema 才能让 LLM 稳定生成正确参数。

我的建议是统一用 Pydantic 声明：

from pydantic import BaseModel, Field from strands_agents.harness import tool class SearchArgs(BaseModel): query: str = Field(..., description="检索关键词，尽量精确简短") max_results: int = Field(5, description="返回结果条数，1-10 之间") @tool(schema=SearchArgs, permissions=["network-read"]) async def web_search(query: str, max_results: int = 5): """搜索引擎封装。""" ...

注意description一定要写清楚。模型生成 tool call 的时候，唯一可靠的信息来源就是字段描述。如果你只写query: str，模型经常会把一段完整问题原封不动塞进去，而不是提取关键词。把字段描述当 prompt 的一部分来写，工具调用的成功率会立竿见影。

4.2 同步阻塞工具：会在事件循环里埋炸弹

第二个高频坑是同步阻塞。很多内部工具是直接用requests.get()写的同步调用，在普通脚本里没问题，但在 Harness 的异步运行时里，一个同步调用会阻塞整个事件循环。你配置了concurrency: 8，但事件循环被一个耗时的同步请求卡住后，所有其他任务的调度都会失真。

最简单的迁移方式是包一层asyncio.to_thread：

import asyncio @tool async def get_inventory(product_id: str): return await asyncio.to_thread(sync_inventory_client.query, product_id)

to_thread会把同步函数放到线程池里执行，避免阻塞事件循环。如果你的工具本身很耗 CPU 而不是 I/O，那就建议不要放进 Agent 的异步路径，而是做成预计算结果。

4.3 旧会话历史直接灌入新系统：上下文格式不匹配

第三个摩擦点出现在有状态 Agent 的迁移中。很多自研系统把工具输出直接拼成一条 user 消息存在数据库里。迁移到 Harness 后，如果直接把这种脏历史塞进新的 MemoryShard，你会发现工具输出被当成用户指令被模型读取，行为立刻变怪。

我建议迁移时做一次记忆清洗：按 system、user、tool 三类重新归类历史消息，把工具输出单独抽出来，必要时压缩成摘要再回填。这个动作虽然是迁移成本的一部分，但能避免上线后出现很多莫名其妙的幻觉式回答。

4.4 渐进式迁移：从无状态 Agent 开始

最后给个战略建议：不要把所有 Agent 一次性迁移到 Harness，尤其是那些在线上跑了很久、积累了大量对话历史的有状态 Agent。先拿新场景、无状态、少工具的 Agent 试水，跑上一两周，把 trace 和审计日志接入现有监控，确认没问题后再逐步替换旧系统。我见过最顺利的迁移团队，基本上是"新 Agent 从第一天就用 Harness，旧 Agent 并行维护三个月，等所有流量切换完再下线旧服务"。

5. 边界与取舍：什么场景其实不该用 Harness

讲了这么多 Harness 的好处，我也要说清楚它不适用的情况。选型最怕的不是选错，而是什么都想用，最后搞出一套过度设计的系统。

5.1 单轮问答、工具数量极少时：重框架是负担

如果你只是写一个"给用户解释某个 API 用法"的简单脚本，整条链路只有一次 LLM 调用、一个知识库检索工具，那手写循环 50 行就能搞定。这时候上 Harness，光是把 yaml 配置、权限声明、沙箱规则写好，就会花掉比手写更长的时间，而且这种场景根本用不到并发隔离、审计追踪、checkpoint。项目复杂度没到那个份上，就别急着上重武器。

我判断是否用 Harness 的标准很简单：看这个 Agent 会不会同时被多个用户使用、会不会调用有副作用的工具、会不会需要恢复现场。三个条件里满足两个以上，Harness 就值得用；都不满足，手写循环更快。

5.2 需要深度掌控循环内部时：封装反而是墙

如果你的研究方向是 Agent 本身的循环机制，比如要实验自研的规划器、要控制每一步向量中间变量的上下文注入顺序，那 Harness 的封装会变成负担。它不是为可插拔的循环算法设计的，它面向的是"用既定的循环跑完生产任务"。对这种需求，你可能需要先 fork 源码，或者干脆使用更底层的编排框架。

我在选型时经常提醒自己一句话：一个好工具不是帮你解决所有问题的银弹，而是帮你把非核心问题外包掉，让你把精力集中在最重要的事情上。Harness 的价值在于把运行时细节从业务逻辑里剥离出来，让我们聚焦在工具质量和提示词设计上。

这次分析 Strands Agents Harness SDK，我最深的体会其实是：开源社区在处理 Agent 生产化这个问题上，已经慢慢从"教你写循环"走向"替你跑循环"了。如果你正在手写第三个agent_loop.py，或者正在为线上 Agent 的重试、并发、审计问题打补丁，可以认真看一眼这类方案，也许会让你的系统提早一步进入稳定运行的状态。