LangGraph范式-retries

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

LangGraph范式-retries

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 继续验证。
  • 作用:实现重试循环,确保在验证失败时动态调整流程。

Paragoger衍生者AI训练营。发布者:稻草人,转载请注明出处:https://www.shxcj.com/archives/9626

(0)
上一篇 2025-05-09 5:25 下午
下一篇 2025-05-13 10:34 上午

相关推荐

发表回复

登录后才能评论
本文授权以下站点有原版访问授权 https://www.shxcj.com https://www.2img.ai https://www.2video.cn