LangGraph范式-retries

Retries（重试机制） 在 LangGraph 中是指当节点执行失败（如 LLM 调用超时、API 错误或逻辑异常）时，自动或有条件地重新尝试执行该节点，以提高工作流的可靠性和鲁棒性。重试机制是构建生产级 AI Agent 的关键，尤其在处理不可靠的外部依赖（如 LLM、API、工具）或动态用户交互时。

Retries 范式的核心流程包括以下步骤：

1. 初始提取（Initial Extraction）：使用 LLM 从输入文本中提取结构化数据，生成初步结果。
2. 结果验证（Validation）：检查 LLM 输出是否符合预定义的模式（如 JSON 模式），识别格式或内容错误。
3. 错误修正（Error Correction）：如果验证失败，生成错误反馈并触发重试，指导 LLM 修正输出。
4. 重试循环（Retry Loop）：重复提取和验证，直到输出有效或达到最大重试次数。
5. 最终输出（Final Output）：返回验证通过的结构化数据，或在失败时返回默认值或错误信息。

Retries 范式通过迭代优化，显著提高了 LLM 在结构化数据提取任务中的可靠性，特别适合处理复杂或不一致的输入文本。

重试策略

LangGraph 的重试机制支持多种策略(每条都挺关键的）

• 固定重试次数：如最多重试 3 次。
• 指数退避（Exponential Backoff）：在每次重试之间增加延迟（如 1s、2s、4s），避免对服务造成过载。
• 错误类型过滤：仅对特定错误（如 RateLimitError）重试，跳过不可恢复的错误（如 ValidationError）。
• 动态调整：根据状态或上下文动态修改重试次数或策略。

数据结构定义

Retries 范式需要维护一个状态对象，用于在节点间传递数据。代码定义了 ExtractionState 作为状态类型：

from typing import TypedDict, Optional, List, Dict, Any
from pydantic import BaseModel, Field
class ExtractionState(TypedDict):
    text: str 输入的非结构化文本，需从中提取数据。
    schema: Dict 预定义的 JSON 模式，描述期望的输出结构（如字段名和类型）
    max_retries: int 最大重试次数（如 3）
    attempts: int 当前尝试次数
    extracted_data: Optional[Dict] LLM 提取的初步数据（可能包含错误
    errors: List[str] 验证过程中发现的错误信息
    final_output: Optional[Dict] 最终验证通过的结构化数据

此外，代码使用 Pydantic 定义了提取目标的结构化模型，例如：

class Person(BaseModel):
    name: str = Field(description="The person's name")
    age: int = Field(description="The person's age")

• Person：一个 Pydantic 模型，定义了提取目标的字段（name 和 age）及其类型。
• 作用：Pydantic 模型用于生成 JSON 模式并验证 LLM 输出。

2.3 核心组件与工作流

Retries 范式的实现分为多个节点，每个节点负责工作流的一个阶段。以下是主要节点的分析。

2.3.1 初始提取（Extract Data）

第一步是使用 LLM 从输入文本中提取结构化数据：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
llm = ChatOpenAI(model="gpt-4o", temperature=0)

extract_prompt = ChatPromptTemplate.from_template(
    """Extract structured data from the following text according to the provided schema.
    Text: {text}
    Schema: {schema}
    Output the result as JSON:
    ```json
    {extracted_data}
If you cannot extract some fields, return null for those fields.
"""
)
extract_chain = extract_prompt | llm | JsonOutputParser()
def extract_data(state: ExtractionState):
text = state["text"]
schema = state["schema"]
result = extract_chain.invoke({"text": text, "schema": schema})
return {
"extracted_data": result,
"attempts": state["attempts"] + 1
}

2.3.2 结果验证（Validate Data）

第二步是验证 LLM 提取的数据是否符合模式：

from pydantic import ValidationError
def validate_data(state: ExtractionState):
    extracted_data = state["extracted_data"]
    schema = state["schema"]
    errors = []
    try:
        # Convert schema to Pydantic model for validation
        model = Person(**extracted_data)
        return {"errors": [], "final_output": extracted_data}
    except ValidationError as e:
        for error in e.errors():
            errors.append(f"Field {error['loc'][0]}: {error['msg']}")
        return {"errors": errors}

• 逻辑：
  • 使用 Pydantic 模型（Person）验证 extracted_data 是否符合模式。
  • 如果验证通过，返回空错误列表和最终输出。
  • 如果验证失败，收集错误信息（如字段类型错误或缺失）。
• 节点：validate_data 更新状态中的 errors 和 final_output（如果验证通过）。
• 特点：Pydantic 的验证功能确保输出的结构和类型正确。

2.3.3 错误修正（Correct Errors）

第三步是基于验证错误生成修正提示，触发重试：

correct_prompt = ChatPromptTemplate.from_template(
    """The previous attempt to extract data failed with the following errors:
    {errors}
    Original text: {text}
    Schema: {schema}
    Previous output: {previous_output}
    Please correct the errors and provide a new JSON output:
    ```json
    {corrected_data}
"""
)
correct_chain = correct_prompt | llm | JsonOutputParser()
def correct_errors(state: ExtractionState):
if not state["errors"]:
return state
text = state["text"]
schema = state["schema"]
errors = state["errors"]
previous_output = state["extracted_data"]
result = correct_chain.invoke({
"text": text,
"schema": schema,
"errors": errors,
"previous_output": previous_output
})
return {
"extracted_data": result,
"attempts": state["attempts"] + 1,
"errors": []
}

2.3.4 路由逻辑（Router）

代码定义了一个 `router` 函数，用于动态路由到下一个节点：

def router(state: ExtractionState):
    if state["final_output"]:
        return "end"
    if state["attempts"] >= state["max_retries"]:
        return "end"
    if state["errors"]:
        return "correct_errors"
    return "validate_data"

• 逻辑：
  • 如果 final_output 已生成，路由到 END。
  • 如果达到 max_retries，路由到 END。
  • 如果有错误，路由到 correct_errors。
  • 否则，路由到 validate_data 继续验证。
• 作用：实现重试循环，确保在验证失败时动态调整流程。