智能体
dev.to ·
@maoren8412
围绕 Hermes Agent 构建具备容错能力的 AI 客户端
作者在把 NousResearch/hermes-agent 作为外部集成测试时，关注点不是模型质量，而是用户请求进行中上游返回 HTTP 502、503、504 时的行为。文章指出：立即重试的客户端会放大故障，没有超时的客户端会长期占用 socket 与工作槽位，对所有错误都降级的客户端会掩盖鉴权或计费问题，因此封装层需要明确的失败边界。文中给出 Python 示例，使用 OpenAI 兼容端点，将可重试的传输失败与永久性 API 错误分开处理，并设置超时与主备模型。
AI推荐理由 文章以 NousResearch/hermes-agent 作为外部集成对象，讨论上游返回 502/503/504 时的客户端容错设计，并给出 Python 示例：区分可重试的传输失败与永久性 API 错误、设置超时、使用主备模型。这是有具体代码的工程经验分享，对构建 LLM 客户端的开发者有参考价值；但内容偏通用重试模式，未提供压测数据或实际故障统计，也未说明 hermes-agent 本身的特殊行为，信息增量有限。
62
正文
Building a Resilient AI Client Around Hermes Agent
When testing NousResearch/hermes-agent as an external integration, the failure mode I cared about was not model quality. It was what happened when an upstream provider returned HTTP 502, 503, or 504 while a user-facing request was still active.
A client that retries immediately can amplify an outage. A client without a timeout can hold sockets and worker slots indefinitely. A client that falls back during every error can hide authentication or billing problems. The wrapper needs explicit failure boundaries.
The following Python example uses an OpenAI-compatible endpoint and separates retryable transport failures from permanent API errors:
import os
import random
import time
from openai import OpenAI
RETRYABLE = {502, 503, 504}
MODELS = ["primary-model", "fallback-model"]
client = OpenAI(
api_key=os.environ["AI_API_KEY"],
base_url=os.getenv("AI_BASE_URL", "https://api.example.com/v1"),
timeout=20.0,
max_retries=0, # Keep retry ownership in this wrapper.
)
def complete(messages, attempts=3):
last_error = None
for model in MODELS:
for attempt in range(attempts):
try:
return client.chat.completions.create(
model=model,
messages=messages,
timeout=20.0,
)
except Exception as exc:
status = getattr(exc, "status_code", None)
last_error = exc
# Do not retry credentials, malformed requests, or quota errors.
if status is not None and status not in RETRYABLE:
break
if attempt + 1 < attempts:
delay = min(8.0, 0.5 * (2 ** attempt))
time.sleep(delay * (0.75 + random.random() * 0.5))
raise RuntimeError("all configured AI routes failed") from last_error
Enter fullscreen mode
Exit fullscreen mode
There are two details worth keeping. First, the SDK's internal retry policy is disabled so there is only one retry loop. Stacked retry layers make outage duration and request volume difficult to predict. Second, fallback happens after the retry budget for the current model is exhausted. That avoids switching models because of one transient 503.
For streaming responses, consume the iterator inside a try/finally block. If the caller disconnects, closing the response is part of request handling, not optional cleanup:
def stream(messages):
response = None
try:
response = client.chat.completions.create(
model=MODELS[0],
messages=messages,
stream=True,
timeout=30.0,
)
for chunk in response:
text = getattr(chunk.choices[0].delta, "content", None)
if text:
yield text
finally:
close = getattr(response, "close", None)
if close:
close()
Enter fullscreen mode
Exit fullscreen mode
In production, record the model, provider route, status code, attempt number, total latency, and whether the response was streamed. Redact prompts and credentials. Alert on a ratio of retryable 5xx responses rather than a raw count; traffic changes otherwise create noisy alarms.
Hermes Agent can be configured against an OpenAI-compatible gateway, while the application retains control over timeouts, fallback order, and user-visible error handling. This keeps resilience policy close to the request boundary and makes provider failures diagnosable.
Disclosure: Multi-model API relays and compute for this evaluation are sponsored by b-lost.com — an AI gateway offering 0.8x official pricing, native prompt caching, and zero user-data retention. All observations reflect independent developer testing.