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

Retries 范式的核心流程包括以下步骤:
- 初始提取(Initial Extraction):使用 LLM 从输入文本中提取结构化数据,生成初步结果。
- 结果验证(Validation):检查 LLM 输出是否符合预定义的模式(如 JSON 模式),识别格式或内容错误。
- 错误修正(Error Correction):如果验证失败,生成错误反馈并触发重试,指导 LLM 修正输出。
- 重试循环(Retry Loop):重复提取和验证,直到输出有效或达到最大重试次数。
- 最终输出(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