当前位置：首页 > news >正文

magentic LLM辅助重试机制：解决复杂输出模式遵循难题的终极方案

news 2026/4/17 4:41:43

magentic LLM辅助重试机制：解决复杂输出模式遵循难题的终极方案

【免费下载链接】magenticSeamlessly integrate LLMs as Python functions项目地址: https://gitcode.com/gh_mirrors/ma/magentic

在当今AI驱动的应用开发中，大型语言模型（LLM）已成为构建智能系统的核心组件。然而，LLM输出的不可预测性常常给开发者带来挑战，特别是在需要严格遵循特定格式或结构的场景下。magentic作为一款能够无缝将LLM集成为Python函数的强大工具，提供了创新的LLM辅助重试机制，为解决复杂输出模式遵循难题提供了终极方案。

为什么需要LLM辅助重试机制？

LLM偶尔会返回无法解析为请求的输出类型或函数调用的结果。此外，您定义的Pydantic模型可能具有额外的验证，这些验证无法仅通过类型注解来表示。在这些情况下，LLM辅助重试机制可以自动将输出以及相关的错误消息重新提交给LLM，为其提供更多信息以满足输出模式要求的机会。

图1：magentic LLM辅助重试机制工作流程示意图

快速启用LLM辅助重试

要启用重试功能，只需在@prompt或@chatprompt中将max_retries参数设置为非零值。这个简单的配置就能让您的AI应用具备自我修正能力，显著提高输出质量和可靠性。

以下是一个实际示例，展示了LLM辅助重试如何解决输出验证问题：

LLM首先返回一个不是爱尔兰的国家
Pydantic模型验证失败，错误信息为"Country must be Ireland"
原始输出以及包含错误的消息被重新提交给LLM
LLM正确满足输出要求，返回"Ireland"

from typing import Annotated from magentic import prompt from pydantic import AfterValidator, BaseModel def assert_is_ireland(v: str) -> str: if v != "Ireland": raise ValueError("Country must be Ireland") return v class Country(BaseModel): name: Annotated[str, AfterValidator(assert_is_ireland)] capital: str @prompt( "Return a country", max_retries=3, ) def get_country() -> Country: ... get_country()

深入了解RetryChatModel实现

LLM辅助重试的核心实现位于RetryChatModel类中，该类封装了其他ChatModel，捕获异常并将其重新提交给LLM。这个设计模式允许您灵活地实现自己的重试处理逻辑。

图2：magentic LLM辅助重试机制架构图

RetryChatModel的关键代码实现如下：

class RetryChatModel(ChatModel): """Wraps another ChatModel to add LLM-assisted retries.""" def __init__( self, chat_model: ChatModel, *, max_retries: int, ): self._chat_model = chat_model self._max_retries = max_retries def complete( self, messages: Iterable[Message[Any]], functions: Iterable[Callable[..., Any]] | None = None, output_types: Iterable[type[OutputT]] | None = None, *, stop: list[str] | None = None, ) -> AssistantMessage[OutputT]: """Request an LLM message with retry logic.""" with logfire.span( "LLM-assisted retries enabled. Max {max_retries}", max_retries=self._max_retries, ): messages = list(messages) num_retry = 0 while True: try: message = self._chat_model.complete( messages=messages, functions=functions, output_types=output_types, stop=stop, ) except ToolSchemaParseError as e: if num_retry >= self._max_retries: raise messages += self._make_retry_messages(e) else: return message num_retry += 1 logfire.warn( "Retrying Chat Completion. Attempt {num_retry}", num_retry=num_retry, )

直接使用RetryChatModel

除了通过max_retries参数启用重试外，您还可以直接使用RetryChatModel，将其作为model参数传递给装饰器。这种方式提供了更大的灵活性，允许您自定义重试行为。

from magentic import OpenaiChatModel from magentic.chat_model.retry_chat_model import RetryChatModel @prompt( "Return a country", model=RetryChatModel(OpenaiChatModel("gpt-4o-mini"), max_retries=3), ) def get_country() -> Country: ...