做 Agent 这一年多,我最深的体会是:模型能力从来不是瓶颈,工程层才是。很多项目能跑通 Demo,却在真正上线后一败涂地——不是模型不够聪明,而是连接、插件、并发、记忆这些工程细节没有处理好。最近 Anthropic 社区和国内很多 Agent 团队都在讨论一个叫 Harness 的设计思路,也有人叫它 Harness Engineering。它要解决的恰恰就是前面说的这些工程层问题。这篇文章我会结合我自己踩过的坑,把 Harness 这类工程方法的骨架、关键参数、实操细节和排查经验都拆开讲一遍。适合正在做 Agent 项目、尤其是准备把项目推向生产环境的开发者参考。
1. 先说结论:Agent 项目的瓶颈从来不在模型层,而在工程层
1.1 为什么 Demo 能跑、上线就挂:我亲眼见过的三起翻车事故
第一起事故。某团队做了一个内部知识库问答 Agent,Demo 演示的时候非常惊艳,模型回答准确,工具调用行云流水。结果上线第二天,接口就开始周期性返回错误,日志里全是连接失败类的报错,Agent 直接瘫痪。根因是什么?代码里把 HTTP 连接数设成了无上限,服务端一限流,客户端这边立刻雪崩。
第二起事故。一个自动化运维 Agent,插件系统做得非常重,主程序启动时要加载十几个插件。换了一台机器部署之后,启动日志里出现插件加载失败、条目未激活的报错,两个插件加载失败。查下来一个原因是依赖库版本不兼容,另一个原因是插件的 manifest 里写的路径不适用跨平台场景。注意这台机器是 Windows,开发环境是 Linux,路径分隔符的差异而已,却让整个服务起不来。
第三起事故。一个给销售团队用的 Agent 工具,单用户测试时响应速度很好,但一放到 30 人小团队里,延迟直接从 2 秒飙到 15 秒,然后大量超时。原因也很简单:没有做并发控制,也没有做请求排队,所有任务一股脑地打给模型 API,既不限制并发数,也不会按优先级调度。
这三起事故的共同点是什么?都不是模型能力的问题。模型还是同一个模型,API 还是同一个 API,问题全出在工程层。
1.2 工程层到底指什么:一个能被拆解的边界定义
如果要给工程层下一个可操作的定义,我会把它拆成五块:
- 连接层:API 通信的超时、重试、熔断、流控,解决网络抖动会不会把 Agent 打死的问题。
- 执行层:工具调用的加载、注册、调度、超时控制、资源隔离,解决插件加载失败或工具卡死的问题。
- 编排层:任务分解、多步执行调度、上下文传递、失败恢复,解决多步骤任务走着走着就断的问题。
- 记忆层:短期记忆窗口、长期记忆存储、记忆提取与清除,解决 Agent 记不住事情或记错事情的问题。
- 安全层:提示注入防护、工具权限控制、敏感信息隔离,解决 Agent 被人利用或被数据污染的问题。
大多数项目死在工程层的根本原因,是团队把精力全放在了前面 10% 的 Demo 效果上,把后面 90% 的系统性问题交给了运气。模型选型可以抄别人的作业,Prompt 可以反复调,但工程层的每一块短板都会在真实流量下被无限放大。
2. 从 Anthropic 实践里长出来的 Harness 工程方法:它凭什么能救活走向死亡的 Agent 项目
2.1 Harness 在解决什么问题:把 Agent 从玩具变成系统
Harness 这个词,原意是马的挽具。给马套上挽具、装上缰绳,马才能拉车干活。在 Agent 工程里,Harness 的隐喻一模一样:给模型套上约束、接好动力系统,它才不是书房里的一匹野马,而是能拉货上路的牲口。
我理解 Harness 工程方法的核心思路,是把 Agent 当作一个需要被严格约束和编排的系统来设计,而不是当作一个对话模型来用。它的重点不在模型推理本身,而在模型外面的那层壳,也就是前面说的五层工程结构。
为什么社区会从 Anthropic 实践和 Claude Code 使用经验里提炼出这个思路?因为大家发现,凡是稳定落地的 Agent,几乎都长得很像:外部有一个稳定的执行引擎,模型只负责决策,工具调用、记忆读写、错误恢复全部由引擎接管。这个引擎就是 Harness。
2.2 核心组件拆解:Runner、Adapter、Tool Registry、Memory Layer
如果用代码的视角看,一套 Harness 设计通常包含下面几个核心组件:
- Runner(执行引擎):Agent 的主循环调度器,负责驱动模型推理、工具调用、结果回填这个循环。它决定了一次 Agent 任务怎么拆、怎么走、怎么停。
- Adapter(模型适配器):屏蔽不同模型接口差异的适配层。Anthropic 的 API、OpenAI 兼容接口、本地部署模型,只要实现同一个 Adapter 接口,Runner 就不用关心底层是谁。
- Tool Registry(工具注册中心):管理所有工具的定义、校验、加载的注册表。工具的 schema 是什么、参数怎么校验、执行超时上限是多少,都在这里统一登记。
- Memory Layer(记忆层):负责会话窗口管理、长期记忆读写和记忆压缩的抽象层,决定 Agent 记住什么、忘记什么。
- Guardrails(护栏):安全与权限控制模块,负责在模型输入输出上做过滤,在工具调用上做权限校验。
这些组件单独看都不复杂,但组合起来解决的问题是本质性的:模型只负责想,系统负责做,而且做的时候要有纪律。这也是 Harness 工程方法区别于普通 Agent 脚本的地方。
2.3 Harness 和主流 Agent 框架的差异
很多人会问:这不就是 LangChain 或者 LangGraph 做的事吗?我的看法是,方向类似,但重心完全不同。
主流框架的重心是编排自由度,给你一大堆节点、边、状态机,让你把复杂的任务流搭出来。而 Harness 工程方法的重心是生产可靠性,更关心模型超时了怎么办、插件加载失败了怎么办、记忆怎么压缩、连接怎么重试。具体差异我列一个表:
| 维度 | 主流 Agent 框架(如 LangChain) | Harness 工程方法 |
|---|---|---|
| 核心关注点 | 任务编排与扩展性 | 可靠性、可观测性与资源治理 |
| 模型接入 | 适配器内置,但配置较重 | 轻量 Adapter,强调可插拔 |
| 工具管理 | 提供工具抽象,但加载和隔离较弱 | 强调注册、校验、超时与资源隔离 |
| 记忆 | 提供基础 memory 组件 | 把记忆压缩、存取策略作为一等公民 |
| 安全 | 有少量内置,需自行扩展 | 作为必填模块,强制护栏 |
不是说框架不好,而是如果你只用框架的编排能力,却把连接、插件、并发这些工程问题留到上线再说,项目大概率会死在工程层。Harness 思路本质上是在补这些课。
3. 工程层第一道生死坎:连接与重试,别让连不上 API 杀死你的 Agent
3.1 一线故障实录:unable to connect 到底意味着什么
先说说那条最常见也最扎心的报错:unable to connect to anthropic services failed to connect to api.anthropic.com。
这条报错的字面意思是:Agent 进程尝试和 API 服务建立 TCP 连接时失败了。注意,这不代表服务真的挂了。我在生产环境里遇到过的真实原因有这么几类:
- 目标域名解析异常,DNS 记录不稳定或本地 DNS 缓存污染。
- 网络出口不稳定,丢包率升高导致 TCP 握手反复失败。
- 服务端对单 IP 的并发连接数做了限制,客户端连接风暴触发了拒绝。
- 网关层配置了传输层握手超时,连接迟迟建立不起来。
很多团队看到这个报错的第一反应是服务端出问题了,然后干等。实际上,大多数时候问题出在客户端——没有超时控制、没有重试、没有熔断,一个短暂的网络抖动就能让整个 Agent 崩掉。
3.2 连接层设计:超时、重试退避、熔断器的参数怎么给
在 Harness 设计里,连接层必须做硬约束。我给出一个经过生产验证的参数配置方案,可以直接参考:
- 建立连接超时:建议设在 5 到 10 秒。超过这个时间,与其继续等,不如直接失败并进入重试逻辑。
- 整体请求超时:模型推理请求建议 60 到 120 秒;工具类请求建议 30 秒以内。
- 重试策略:采用指数退避加抖动。第一次等 1 秒,第二次 2 秒,第三次 4 秒,依此类推,最多重试 3 到 4 次。
- 熔断器:连续失败达到 5 次后,进入熔断状态,在 30 秒内直接拒绝新请求,不再打给 API。
关于重试有个细节必须提醒:不是所有请求都适合重试。如果模型已经开始生成内容,在生成中途断连,重试之前最好确认是否涉及有副作用的工具调用。纯查询类任务重试是安全的,但如果 Agent 已经执行了工具调用(比如发了一封邮件),重试前一定要先恢复上下文,避免重复执行副作用。我见过团队因为盲目重试,同一个删除命令被连续执行了两次,这种事故比连接失败本身严重得多。
提示:重试要区分幂等和非幂等。读操作可以放心重试,写操作必须先查状态再决定是否重试。
3.3 可插拔 Provider 适配:让换模型成本接近为零
连接层还有一个容易被忽视的设计目标:可插拔。做 Agent 项目时不要把自己绑死在单一模型供应商上。
做法很简单,抽象一个 Provider 接口,接口里主要包含两个方法:
- complete(messages, tools):发送对话和工具描述,返回模型回复。
- stream(messages, tools):流式版本,用于流式输出场景。
Anthropic 的 Client 实现一个 Provider,OpenAI 兼容接口实现一个 Provider,本地部署的模型(比如团队自己用 vLLM 或 Ollama 搭的开源模型服务)同样实现一个 Provider。Runner 只依赖接口,不依赖具体实现。
这样做的好处是什么?一是当某个 API 连续故障时,可以快速切换到备用通道或备用模型;二是模型升级、替换时不需要改动主流程代码。我见过太多项目,模型调用代码散落在几十个文件里,想换个模型等于重写一遍。Harness 思路从一开始就要求你把这层隔离好。
4. 工程层第二道坎:插件机制与执行可靠性
4.1 插件加载失败的真实原因与排查路径
harness failed to load plugins 这类错误,我在多个项目里都碰到过。很多开发者一开始会觉得莫名其妙:我在本地明明跑得好好的,为什么部署到服务器就加载失败?
插件加载失败的原因其实非常固定,基本逃不出这几类:
- 路径问题:开发环境是 Windows,部署环境是 Linux,插件 manifest 里写死了反斜杠路径,导致找不到文件。
- 依赖冲突:插件 A 依赖库 X 的 1.x 版本,插件 B 依赖库 X 的 2.x 版本,依赖解析结果导致其中一个插件运行时报错。
- 缺失依赖:插件声明的依赖没有安装,或者安装顺序不对。
- 权限问题:插件目录的读写权限不足,导致初始化失败。
排查路径也有固定套路。第一步看启动日志,确认是哪个插件加载失败、失败在哪个生命周期阶段(读取配置、导入模块、初始化资源)。第二步看 manifest 声明和实际环境是否一致,路径、依赖版本逐项核对。第三步把插件逐个禁用、逐个启用,做二分定位。
4.2 插件注册、验证与依赖注入的正确姿势
在 Harness 设计里,我不推荐一个插件目录里放几十个脚本,启动时全量 import。更好的做法是引入一个轻量的注册表机制,让每个插件声明自己,由运行时统一加载:
- 每个插件提供一个 manifest 文件,声明插件名、版本、入口文件、依赖列表、权限需求。
- 启动时,Runner 先读取 manifest,做依赖检查、版本检查,再按依赖顺序加载插件。
- 加载成功后,插件向 Tool Registry 注册自己提供的工具列表。
- 加载失败时,插件不影响主程序启动,而是在日志中的失败列表里被单独标记。
拿前文说的跨平台路径问题为例,正确的做法是 manifest 里不要写绝对路径,也不要写平台相关的分隔符,而是约定一个相对路径加运行时解析的规则。踩过那次坑之后,我所有插件入口的统一写法是相对于插件目录的相对路径,而插件目录本身由 Runner 在系统临时目录或配置目录里动态创建。从那以后再没出过路径类加载事故。
加载顺序也很重要。插件 A 依赖插件 B 的工具,必须先加载 B 再加载 A,否则 A 在注册工具时找不到依赖。这个依赖关系必须在 manifest 里显式声明,不能靠 import 顺序碰运气。
4.3 工具执行的超时控制与资源隔离
插件加载只是第一步,更关键的是工具执行阶段的可靠性。工具是 Agent 的手,但这个手可能伸到任何地方——读文件、写数据库、调外部 API、执行命令。任何一次工具调用卡住,都会拖死整个 Agent 主循环。
所以 Harness 设计对工具执行有两条硬约束:
- 每个工具必须有超时上限。超时时间可以由工具声明,也可以由运行时统一设置,但默认值必须有。没有设置超时的工具,在注册阶段就被拒绝。
- 重工具要做资源隔离。执行 Shell 命令、跑 Python 脚本这类高风险工具,建议丢到子进程或隔离环境里执行,主进程通过管道读回结果。这样即便工具内部崩溃,也不会拖垮整个 Agent。
超时上限的设置是个取舍。设得太短,复杂的工具经常被误杀;设得太长,一次卡住的工具会让整个流程长时间阻塞。我实践下来的经验是:文件读写类工具设置 10 秒,网络请求类工具设置 30 秒,模型推理类工具设置 120 秒,Shell 命令类工具按需设置,但不要超过 120 秒。
5. 工程层第三道坎:并发、记忆与安全,生产级 Agent 的硬要求
5.1 AI Agent 怎么扛并发:限流、排队与池化
AI Agent 怎么扛并发?这是所有 Agent 项目从 Demo 走向生产的必经之路。
Agent 的并发和普通 Web 服务的并发不一样。普通接口处理一个请求可能只需要几十毫秒,Agent 处理一个任务可能要几秒甚至几十秒,期间还要反复调用模型 API、执行工具。所以 Agent 的并发不仅是请求进来要能接住,更是任务进来要能调度好。
我实践下来有三板斧。
第一,模型 API 调用必须限制并发数。模型服务商对并发都有配额限制,超过限制就会触发限流,导致大量请求失败。做法是用信号量或连接池控制同时进行的模型请求数量。比如把最大并发设为 8,当并发任务超过这个值时,多余的任务排队等待。
第二,任务队列要做优先级。不能所有任务完全公平排队。一个内部测试任务和一个用户实时对话任务不应该互相拖累。做法是给任务配置优先级分级,高优先级任务可以插队或被优先调度。
第三,对 Agent 的并发任务要做箱式隔离。每个任务的上下文、记忆、工具调用状态是独立的,不能在多个任务之间共享可变状态。这里的坑在于:有些人为了省内存,让多个任务共用一个会话上下文,结果任务 A 改了下文,任务 B 的下一步推理就错了。这种 bug 极其隐蔽,排查起来非常痛苦。
5.2 Agent 记忆的工程实现
记忆是 Agent 项目的另一个大坑。Demo 阶段你只需要把整个对话历史全部塞给模型,因为上下文只有十几轮;到了生产环境,对话历史可能跨越几十天,全部塞进去既不现实又浪费成本。所以 Harness 设计里,记忆层必须主动管理。
我把记忆分成三层:
- 短期记忆:当前任务进程内的上下文窗口,通常限制在模型上下文长度的八成以内,超出部分触发裁剪。
- 工作记忆:跨任务但短期内需要的状态,比如用户偏好、当前项目配置,存放在 Redis 之类的快速存储里。
- 长期记忆:跨会话的知识沉淀,比如用户的历史问题、常用工具参数、业务规则,存放在向量数据库中,按相关性检索后注入上下文。
记忆工程最关键的不是存,而是提炼。很多团队把长期记忆做成简单的所有历史记录全存,结果每次检索返回一堆冗余信息,反而污染了模型的推理。正确的做法是:提取关键事实,压缩归纳,再做检索。我见过做得好的项目,用模型对每一轮对话做一次摘要生成,把摘要存入向量库;检索的时候只把最相关的三五条摘要注入上下文。效果比粗暴的全文存储好得多。
关于记忆还有一个安全相关的衍生话题:记忆防御。社区里已经出现了针对 LLM Agent 记忆的攻击框架,比如 a-memguard 这类针对记忆的防御方法,专门检测和防御恶意指令通过记忆注入污染 Agent 行为的攻击方式。Agent 的记忆一旦被污染,后续所有会话都会带着被篡改的事实做推理,危害比单次提示注入更大。所以记忆层在写入之前要做校验,读到可疑记忆时要隔离并提示用户。
5.3 安全防线:从提示注入到工具权限
工程层的大坑最后落到安全。Agent 的杀伤力越大,安全出问题的后果越严重。
第一道防线是输入过滤。对用户输入做启发式检测,识别明显的提示注入特征,比如忽略之前的指令、扮演开发者模式这类模式。虽然这不能完全防御语义攻击,但能挡住大多数脚本化的尝试。
第二道防线是工具权限控制。工具注册表里必须声明每个工具需要的权限级别。只读工具、写工具、危险工具要分级。危险工具(比如删除文件、执行任意命令)必须二次确认,或者需要额外密钥才能启用。不要把危险工具和普通工具混在一起,由模型自由选择。
第三道防线是输出过滤和敏感信息保护。模型返回内容里可能携带不该外泄的数据,尤其是当工具调用了内部 API 时。在输出到用户之前,做一次脱敏过滤,把邮箱、电话、密钥等敏感字段打码。
可能在 Demo 阶段这三道防线都显得多余。但一旦 Agent 接入真实业务,任何一道防线的缺失都可能是事故级别的。
6. 实操实录:用 Harness 思路从零搭一个能上生产环境的 Agent 骨架
6.1 工程骨架与配置
说了这么多理念,接下来给一个可以直接抄作业的骨架实现。我用 Python 写一个极简但结构完整的 Harness 式 Agent 骨架,包含 Adapter、Runner、Tool Registry、Memory Layer 四个核心模块。
项目结构如下:
agent_harness/ ├── adapter/ │ ├── base.py # Provider 抽象接口 │ ├── anthropic_adapter.py │ └── local_adapter.py # 本地模型适配 ├── core/ │ ├── runner.py # 主循环执行引擎 │ ├── registry.py # 工具注册中心 │ └── memory.py # 记忆层 ├── tools/ │ ├── manifest.json # 插件声明 │ └── search_tool.py # 示例工具 ├── config.yaml # 运行时配置 └── main.py # 入口核心配置参数:
# config.yaml provider: name: anthropic base_url: https://api.anthropic.com max_concurrency: 8 # 最大并发模型请求数 timeout: connect: 10 # 连接超时(秒) read: 120 # 读取超时(秒) retry: max_attempts: 4 # 最大重试次数 base_delay: 1.0 # 指数退避基础等待(秒) max_delay: 8.0 # 最大退避等待(秒) circuit_breaker: failure_threshold: 5 # 触发熔断的连续失败次数 cooldown: 30 # 熔断冷却时间(秒) tools: auto_load: true enable_manifest_check: true default_timeout: 30 # 工具默认超时(秒)这里给的参数不是拍脑袋。connect 10 秒覆盖大部分网络环境的握手耗时;read 120 秒给长推理留足余量;retry 最大 4 次配合指数退避,能让一个 10 秒的临时故障在 15 秒内自愈,又不会在长期故障时无限轰炸 API;熔断 30 秒冷却,让系统在严重故障时能快速降级而不是持续雪崩。
6.2 核心代码实现
Provider 抽象接口:
# adapter/base.py from abc import ABC, abstractmethod from typing import Any class Provider(ABC): @abstractmethod def complete( self, messages: list[dict], tools: list[dict] | None = None, ) -> dict[str, Any]: """发送对话并返回模型回复""" @abstractmethod def stream( self, messages: list[dict], tools: list[dict] | None = None, ): """流式发送对话"""Anthropic 适配器:
# adapter/anthropic_adapter.py import anthropic from adapter.base import Provider class AnthropicAdapter(Provider): def __init__(self, api_key: str, base_url: str): self.client = anthropic.Anthropic( api_key=api_key, base_url=base_url, max_retries=4, # 客户端自带重试 timeout=120.0, ) def complete(self, messages, tools=None): response = self.client.messages.create( model="claude-sonnet-4", max_tokens=4096, messages=messages, tools=tools, ) # 为清晰起见,此处简化了响应解析逻辑 return response.model_dump() def stream(self, messages, tools=None): with self.client.messages.stream( model="claude-sonnet-4", max_tokens=4096, messages=messages, tools=tools, ) as stream: for text in stream.text_stream: yield text工具注册中心:
# core/registry.py from concurrent.futures import ThreadPoolExecutor class ToolRegistry: def __init__(self): self._tools: dict[str, dict] = {} def register( self, name: str, description: str, func: callable, timeout: int = 30, dangerous: bool = False, ): self._tools[name] = { "name": name, "description": description, "func": func, "timeout": timeout, "dangerous": dangerous, } def execute(self, name: str, args: dict): tool = self._tools.get(name) if not tool: raise KeyError(f"tool not found: {name}") # 超时保护:把工具调用放进线程池,限制执行时间 with ThreadPoolExecutor(max_workers=1) as pool: future = pool.submit(tool["func"], **args) return future.result(timeout=tool["timeout"])主循环 Runner:
# core/runner.py import asyncio from adapter.base import Provider from core.registry import ToolRegistry from core.memory import MemoryLayer class Runner: def __init__( self, provider: Provider, registry: ToolRegistry, memory: MemoryLayer, max_steps: int = 10, ): self.provider = provider self.registry = registry self.memory = memory self.max_steps = max_steps async def run(self, user_input: str) -> str: # 1. 从记忆层恢复上下文 messages = await self.memory.build_context(user_input) for step in range(self.max_steps): # 2. 模型推理,带上工具清单 tool_schemas = [ {"name": t["name"], "description": t["description"]} for t in self.registry._tools.values() ] response = await asyncio.to_thread( self.provider.complete, messages, tool_schemas ) # 3. 判断模型是否要求调用工具 tool_calls = response.get("tool_calls", []) if not tool_calls: # 没有工具调用,直接返回最终文本 return response["content"][0]["text"] # 4. 依次执行工具调用 for call in tool_calls: tool_name = call["name"] tool_args = call["arguments"] try: result = self.registry.execute(tool_name, tool_args) messages.append( { "role": "tool", "tool_call_id": call["id"], "content": str(result), } ) except Exception as exc: messages.append( { "role": "tool", "tool_call_id": call["id"], "content": f"ERROR: {exc}", } ) return "REACHED_MAX_STEPS"记忆层最小实现:
# core/memory.py class MemoryLayer: """极简记忆层:生产环境可替换为向量库 + 摘要存储""" def __init__(self, max_context_tokens: int = 8000): self._short_term: list[dict] = [] self.max_context_tokens = max_context_tokens async def build_context(self, user_input: str) -> list[dict]: # 这里读取短期记忆,并追加用户输入 messages = list(self._short_term) messages.append({"role": "user", "content": user_input}) return messages def save_turn(self, user_msg: dict, assistant_msg: dict): self._short_term.append(user_msg) self._short_term.append(assistant_msg) self._compact_if_needed() def _compact_if_needed(self): # 当记忆超长时,丢弃最早的对话 while len(self._short_term) > 20: self._short_term.pop(0)入口文件:
# main.py import asyncio from adapter.anthropic_adapter import AnthropicAdapter from core.memory import MemoryLayer from core.registry import ToolRegistry from core.runner import Runner async def main(): provider = AnthropicAdapter( api_key="your-api-key", base_url="https://api.anthropic.com", ) registry = ToolRegistry() # 注册 demo 工具 def get_current_time(): import datetime return datetime.datetime.now().isoformat() registry.register( "get_current_time", "获取当前时间", get_current_time, timeout=10, ) memory = MemoryLayer() runner = Runner(provider, registry, memory) result = await runner.run("现在几点?") print(result) if __name__ == "__main__": asyncio.run(main())这段代码虽然极简,但已经把 Harness 的核心骨架搭建出来了:Provider 隔离模型差异,Registry 管理工具并强制超时,Runner 控制主循环和步骤上限,Memory 负责上下文管理。
6.3 本地部署验证与踩坑记录
搭好骨架之后,我建议按下面几步做本地验证:
- 先用一个不需要工具的 Prompt 测试连通性,确认 Adapter 能正常收到模型回复。
- 再注册一个 2 秒内能完成的工具,测试工具调用链路。
- 故意把工具 timeout 设成 1 秒,触发超时异常,确认错误能回填给模型而不是整个程序崩溃。
- 最后用并发脚本做一次 20 个请求的压测,观察限流和超时情况。
这一步实际跑的时候,我踩过几个坑,写出来供参考。
第一个坑是同步阻塞卡死主循环。工具函数如果是同步的,但用了 requests 这类阻塞库,在 asyncio 事件循环里直接调用会卡住整个 Runner。解决办法是用 asyncio.to_thread 把工具调用丢到线程池里执行。这个坑非常隐蔽,因为 Demo 阶段只有一两个并发任务,卡顿不明显;压力一大,整个进程直接没有任何响应。
第二个坑是模型返回的 tool_calls 空转。有的模型在不确定时,会返回一个调用了工具但又不需要执行结果的工具调用,Agent 就会陷入调用工具、拿到结果、再调用的死循环。所以 Runner 里必须设置 max_steps 上限,超过步骤直接返回错误提示,绝对不能无限循环。
第三个坑是工具结果过大会撑爆上下文。一个数据库查询工具可能返回几千行记录,全部塞进 messages 里,下一轮模型推理就会超出上下文窗口。处理方式是对工具结果做截断,比如只保留前 100 个字符,或者先在工具层做摘要。
7. 常见问题排查速查表
把前面几章踩过的坑汇总成一张速查表,遇到问题时直接对着查。
7.1 连接类问题:报错与对策
| 报错或现象 | 可能原因 | 优先排查项 | 对策 |
|---|---|---|---|
| unable to connect / failed to connect | TCP 握手失败 | DNS 解析、网络出口、服务端限流 | 设置连接超时,启用重试与熔断 |
| connection timed out | 网络不稳定或出口受限 | 检查网络连通性 | 缩短连接超时,快速失败并重试 |
| rate limit exceeded | 超出并发配额 | 日志中的限流标记 | 降低并发数,设置重试退避 |
| 请求偶尔成功偶尔失败 | 客户端连接数过高 | 连接池配置 | 控制并发数,复用连接 |
7.2 插件与工具调用类问题
| 报错或现象 | 可能原因 | 优先排查项 | 对策 |
|---|---|---|---|
| failed to load plugins / entries did not activate | 插件加载失败 | manifest、依赖、路径 | 检查 manifest,按依赖序加载,失败标记隔离 |
| tool not found | 工具未注册 | 注册顺序、命名 | 检查 Tool Registry,确认工具已注册 |
| 工具执行卡死 | 无超时控制 | 工具 timeout 配置 | 为每个工具设置超时上限 |
| 工具结果太大 | 无截断策略 | 工具返回长度 | 截断或摘要工具结果 |
7.3 并发与稳定性问题
| 现象 | 可能原因 | 优先排查项 | 对策 |
|---|---|---|---|
| 延迟从 2 秒飙到 15 秒 | 并发不受控 | 并发数、请求排队 | 设置 max_concurrency,增加任务队列 |
| 进程无响应 | 同步阻塞卡住事件循环 | 工具调用方式 | 用 asyncio.to_thread 包装阻塞调用 |
| 上下文越界 | 记忆无裁剪 | 记忆层长度 | 设置记忆窗口上限,触发裁剪 |
| 任务上下文串扰 | 共享可变状态 | 会话隔离 | 每个任务独立上下文,禁止共享 |
最后再分享一点我做 Agent 工程的实际体会。很多团队在做 Agent 时,容易把模型能力当作项目的护城河,整天纠结要不要换一个更强的模型、Prompt 怎么写更花哨。但我的经验是,真正决定项目生死的,往往是模型外面那层不起眼的工程壳。连接层没有重试,一个网络抖动就能让服务瘫痪;插件系统没有超时和隔离,一个卡死的工具就能拖垮整个主循环;记忆层没有提炼和防御,检索出来的全是噪音,甚至会被恶意指令污染。Harness 工程方法最大的价值,是逼你把这些问题在架构阶段就想清楚,而不是等上线了再去填坑。
如果你最近也在做 Agent 项目,我建议你先停下来盘点一下:你的代码里有没有连接重试?工具调用有没有超时上限?并发有没有限制?记忆有没有做提炼?如果没有,那你的项目大概率还在工程层裸奔。趁这些问题还没变成事故,尽快把这一层补上。踩过坑之后你会发现,模型是子弹,工程才是枪。子弹质量固然重要,但没有一把靠谱的枪,什么子弹都打不出去。