LLM Agent 稳定性实战：错误处理与重试设计

一个达到工业级水准的 LLM agent 不仅仅是一个返回响应的 API 调用。它需要能够应对瞬时故障、畸形输出和 Schema 校验失败等问题。为了提升 agent 系统的稳定性，我在 LLM 调用外层引入了两个装饰器，分别处理三种常见的失败模式：

• JSON 解码错误 — 将模型之前格式错误的响应提供给它，并要求其重新生成有效的 JSON
• Pydantic BaseModel 校验错误 — 将具体的 Schema 校验错误信息提供给模型，并要求其修正响应
• 429 速率限制错误 — 以适当的退避策略进行重试

这种设计将可靠性逻辑与各 agent 函数分离，使每个 LLM 调用只需专注于自身任务，而装饰器则提供统一的恢复机制。

JSON 和 Pydantic BaseModel

为了处理有问题的模型输出，我首先使用了 json-repair 包：

json_repair

PS：如果觉得这个包有用，请考虑给作者一个 ⭐ 以示支持。

这使得系统能够从模型响应中常见的 JSON 格式化问题中恢复。如果修复后的结果仍为空，我会手动抛出一个 JSONDecodeError。

这很有用，因为重试装饰器可以拦截该异常，并向模型传递一条有意义的重试消息。与其盲目地让模型再试一次，不如告诉它：

Your previous response was not valid JSON.

并通过 e.doc 包含之前的响应。这样模型就能将失败的输出作为上下文，从而正确地重新生成响应。

下一个场景是 Pydantic ValidationError。响应可能是有效的 JSON，但仍无法匹配预期的 Schema。这种情况下，简单地让模型重试也没有特别大的用处。重试消息应该包含具体哪些字段校验失败以及预期的约束条件。

这些信息可以通过 e.errors() 获取。我可以提取字段位置和校验消息，然后将详细信息反馈给模型。这将重试从一次盲目的重新生成变成了一次有针对性的修正。

def llm_retry(max_retries=3):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            retry_message = None

            for attempt in range(max_retries):
                try:
                    return await func(*args, retry_message=retry_message, **kwargs)

                except ValidationError as e:
                    errors = [
                        {"field": ".".join(map(str, x["loc"])), "message": x["msg"]}
                        for x in e.errors()
                    ]
                    retry_message = (
                        "Your previous response failed schema validation.\n\n"
                        f"Validation errors:\n{errors}\n\n"
                        "Correct these errors and return the complete response as JSON only."
                    )

                except json.JSONDecodeError as e:
                    retry_message = (
                        "Your previous response was not valid JSON.\n\n"
                        f"Previous response:\n{e.doc}\n\n"
                        "Regenerate the complete response as valid JSON only. "
                        "Do not include Markdown or explanations."
                    )

            raise RuntimeError(f"LLM failed after {max_retries} attempts")

        return wrapper
    return decorator

429 Rate-Limit 装饰器

这部分相对直接。我使用一个简单的函数工厂来创建一个装饰器，包裹在 LLM 调用外层，自动重试收到 429 Too Many Requests 响应的请求。

装饰器在每次重试之间应用指数退避策略，使系统能够从临时的速率限制中恢复，而无需每个 LLM 调用都实现自己的重试逻辑。

def Retry429(trials=3, delay=1.0, backoff=2.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            current_delay = delay

            for attempt in range(trials):
                try:
                    return await func(*args, **kwargs)
                except RateLimitError:
                    pass
                except httpx.HTTPStatusError as e:
                    if e.response.status_code != 429:
                        raise
                except httpx.HTTPError:
                    raise

                if attempt == trials - 1:
                    raise

                await asyncio.sleep(current_delay)
                current_delay *= backoff

        return wrapper

    return decorator

在有了这些装饰器之后，我就可以将 LLM 调用定义为一个简单的函数。每个函数只负责构造输入参数和返回输出参数，这些参数分别由 LLMRequest 和 LLMResponse 表示。

将这些接口定义为 Pydantic BaseModel，为系统不同部分之间提供了一致的契约，使整个 agent 架构更加系统化和可维护。

更重要的是，重试和恢复机制不再与具体的 LLM 实现耦合。这使我能够支持不同类型的 LLM 调用，例如图像生成和 VLM 推理，而无需在每个函数中重复粘贴相同的重试和校验逻辑。各函数保持专注于自身的特定任务，共享的装饰器则处理通用的可靠性问题。

    @Retry429(trials=10, delay=10)
    @llm_retry()
    async def generate(self, llm_request: LLMRequest, retry_message=None)-> LLMResponse:
        args = {}
        # Just some args parsing
        if retry_message is not None:
            args["messages"].append({"role": "system", "content": retry_message})

        response = await self.client.chat.completions.create(**args)
        self.add_token_usage(response)
        if not response:
            raise RuntimeError("LLM returned empty response")

        if not response.choices:
            logger.error(
                "LLM returned no choices. Response=%s",
                response.model_dump()
            )
            raise RuntimeError("LLM returned no choices")
        content = response.choices[0].message.content

        json_schema = None
        if llm_request.json_schema is not None:
            content_json = json_repair.loads(content, return_objects=True) 
            if len(content_json) == 0:
                raise json.JSONDecodeError("Invalid json", content, 0)
            if isinstance(content_json, list):
                json_schema = []
                for content in content_json:
                    json_schema.append(llm_request.json_schema(**content))
            elif llm_request.bulk:
                json_schema = [llm_request.json_schema(**content_json)]
            else:
                json_schema = llm_request.json_schema(**content_json)

我从未想到《Fluent Python》中的概念——尤其是数据模型、函数工厂和装饰器——在设计这样一个优雅的系统时会如此有用。

看到这些概念如何组合起来构建一个系统化、可维护的 LLM 基础设施，这提醒我们：扎实的软件工程基础在构建现代 AI 系统时仍然具有极高的价值。

这也鼓励我通过更深入地阅读、理解基础原理，并通过实际项目中学以致用来不断提升自己的工程技能。