实用指南：利用Self-RAG与LangGraph打造能够自我纠错的AI智能体。
实用指南操作流程：使用Self-RAG与LangGraph构建具备自我纠错功能的AI智能体——为采用该架构的团队提供合同模板、检查清单及可直接使用的代码片段。
来源: Build a Self-Correcting AI Agent with Self-RAG & LangGraph
本指南将逐步构建从原材料到可运行系统的完整流程，主题为：使用 Self-RAG 与 LangGraph 构建具备自我修正能力的 AI 智能体。重点在于可操作的步骤、明确的检查点，以及可直接放入代码仓库的代码，无需猜测其用途。
为何标准 RAG 流水线如此脆弱
为了解释标准 RAG 流水线为何脆弱，应在修改代码之前明确输入内容、各步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。 配置信息应与应用程序代码分开存放。环境文件、密钥存储和功能标志应集中于一个位置，以便操作人员无需查看整个流程即可进行审计。 需注明真正作为答案依据的段落。如果没有引用，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。
线性流水线：检索与假设
对于线性流程：在修改代码之前，需明确输入内容、各步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的功能。 必须引用那些真正作为答案依据的段落。如果没有引用，操作人员就无法区分是虚假信息还是索引缺失导致的问题。
[User Query] ──> [Vector Database] ──> [Top-K Chunks] ──> [LLM Prompt] ──> [Final Answer]
无声故障的成因分析
在修改代码之前，对于“无声故障的成因分析”，需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型、可测试的单元。当某个步骤出现故障时，故障原因应指向单一责任主体，而非复杂的流程链。需引用实际作为答案依据的段落；没有引用的话，操作人员就无法区分是虚假信息还是索引缺失所致。
根本原因：盲目信任
为了解决根本问题——盲目信任，应在修改代码之前明确输入内容、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的半完成状态。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分幻觉与索引缺失。
解决方案：引入自我反思循环
解决方案：引入自反射循环，即在修改代码之前明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外费用。需注明实际作为答案依据的段落；没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。解决方案：引入自反射循环，即在修改代码之前明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。需同时记录正常流程与异常恢复流程，重试机制、人工审核环节以及死信处理方式都是生产环境中的重要组成部分。
CT，之后再进行优化。从盲目检索到自我反思：Self-RAG模式
在研究“从盲目检索到自我反思：Self-RAG模式”时，首先需明确相关约定：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤出错时，错误应指向单一责任点，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来衡量召回率。频繁更换提示词往往无法解决检索效果不佳的问题。
自我修正的三大支柱
在研习《自我修正的三大支柱》时，首先写下相关契约：所需的输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，杜绝默许的部分完成情况。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难改善较差的检索效果。
[ Query ]
│
▼
1. Is Retrieval Needed? ──(No)──► Generate directly from parametric memory
│
(Yes)
▼
[ Retrieve ]
│
▼
2. Are Docs Relevant? ──(No)──► Reformulate search query & retry
│
(Yes)
▼
[ Generate ]
│
▼
3. Is Answer Faithful? ──(No)──► Regenerate or extract better context
│
(Yes)
▼
[ Final Output ]
反思令牌作为控制信号
将反射令牌用作控制信号时，首先需明确合约内容：所需的输入参数、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 缓存稳定的系统指令和工具架构。重复发送相同的起始数据是导致资源浪费的常见原因。 将反射令牌用作控制信号时，首先需明确合约内容：所需的输入参数、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。
自我反射的架构
“自我反思架构”在被视为可度量的对象时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
graph TD
Start[User Query] --> DecideNeed{Is Retrieval Needed?}
DecideNeed -- No --> GenerateDirect[Generate Output]
DecideNeed -- Yes --> Retrieve[Retrieve Documents]
Retrieve --> GradeDocs[Grade Document Relevance]
GradeDocs -- Irrelevant --> Fallback[Rewrite Query / Fallback]
GradeDocs -- Relevant --> GenerateRef[Generate Response with Context]
GenerateRef --> GradeFaith[Grade Faithfulness & Hallucinations]
GradeFaith -- Hallucinated --> GenerateRef
GradeFaith -- Faithful --> GradeUtility[Grade Utility]
GradeUtility -- Low Utility --> Rewrite[Refine Response]
GradeUtility -- High Utility --> End[Final Answer]
流程管道与工作流的比较
将管道与工作流进行比较时，最好将其视为可度量的对象。在扩大范围之前，先记录一份最佳案例、一个失败实例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。 应将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
利用 LangGraph 进行架构设计：从线性链到状态机
《用LangGraph进行架构设计：从线性链到状态机》这一指南在被视为可度量的对象时效果最佳。在扩大范围之前，先记录一个理想案例、一个失败案例以及回滚说明。 在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，就能避免在系统从演示环境过渡到共享环境时出现意外费用。 应将分块策略与检索策略分开。当质量指标发生变化时，调整其中一个不应迫使重新编写另一个。 《用LangGraph进行架构设计：从线性链到状态机》这一指南在被视为可度量的对象时效果最佳。在扩大范围之前，先记录一个理想案例、一个失败案例以及回滚说明。 需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的内容。
Linear RAG: [Query] ──> [Retrieve] ──> [Generate] ──> [Response] (No correction)
Self-RAG: [Query] ──> [Retrieve] ──> [Evaluate] ──┬─ (Valid) ──> [Response]
▲ │
└─────── (Invalid) ──────┘ (Loop & Refine)
线性链的局限性
在探讨线性链的局限性时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一责任主体，而非复杂的流程链。必须引用那些真正为答案提供依据的段落；没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失所致。
揭开AI状态机的面纱
为了解析人工智能状态机，在修改代码之前需明确输入参数、各步骤的负责人以及退出条件。操作人员应能够从已知的检查点重新执行相应步骤，而无需猜测隐藏状态。 将这一阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分幻觉内容与索引缺失问题。
┌──────────────────────┐
│ Start │
└──────────┬───────────┘
▼
┌──────────────┐ ┌──────────────────────┐
│ Formulate │◄─┤ Retrieve Docs ├─┐
│ New Query │ └──────────────────────┘ │
└──────▲───────┘ │
│ ▼
┌──────┴───────┐ ┌──────────────────────┐ │
│ Re-Retrieve │ │ Grade Relevance │─┘
└──────────────┘ └──────────┬───────────┘
▼
┌──────────────────────┐
│ Generate Answer │
└──────────┬───────────┘
▼
┌──────────────────────┐
│ Grade Hallucination │
└──────────────────────┘
LangGraph核心组件详解
在《深入了解LangGraph核心组件》中，修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果旁记录执行时间以及token或查询成本。提前显示成本可避免在流程从演示环境切换到共享环境时出现意外费用。 需注明实际用于得出答案的对应内容。没有引用依据的话，操作员就无法区分是幻觉还是索引缺失导致的错误。 在《深入了解LangGraph核心组件》中，修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理措施都是产品本身的组成部分，而非后续才添加的完善功能。
在实现中央图状态时，首先需明确相关规范：所需的输入参数、成功信号，以及部分失败时的处理方式。这样的清单能确保后续的代码修改始终符合预期。 建议采用小型、可测试的单元而非庞大的脚本。当某个步骤出现故障时，故障应指向单一的责任模块，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
from typing import List, TypedDict
class GraphState(TypedDict):
"""
Represents the central state of our Self-RAG agent.
This TypedDict serves as the single source of truth passed between
each node in the LangGraph execution flow.
"""
question: str # The original user query
generation: str # The most recent answer generated by the LLM
documents: List[str] # Relevant text passages retrieved from our database
steps: List[str] # An audit trail tracking the sequence of execution steps
构建图结构，第一部分：定义状态节点与动作节点
在学习《构建图结构，第一部分：定义状态节点与动作节点》时，首先需写出规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法改善较差的检索效果。
理解基于状态的编排机制
在学习基于状态的编排时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在记录功能结果的同时，还需标注执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定的问题集测试系统的召回率。仅仅更换提示词往往无法改善较差的检索效果。 在学习基于状态的编排时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。
+-------------------------------------------------------------+
| GRAPH STATE |
| - question: str |
| - documents: List[Document] |
| - generation: str |
| - relevance_grade: str |
+-------------------------------------------------------------+
|
+----------------------+----------------------+
| | |
v v v
+--------------+ +--------------+ +--------------+
| Retrieve | | Grade Docs | | Generate |
| Node | | Node | | Node |
+--------------+ +--------------+ +--------------+
设计共享状态账本
在设计共享状态账本时，将其视为可度量的对象会更为高效。在扩大范围之前，先记录一份标准操作流程、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应指向单一责任模块，而非复杂的流程链。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
from typing import List, Dict, Any
from typing_extensions import TypedDict
from langchain_core.documents import Document
class GraphState(TypedDict):
"""
Represents the state of our Self-RAG graph.
Attributes:
question: The user query string.
documents: A list of retrieved document chunks.
generation: The generated answer from the LLM.
relevance_grade: The evaluation score of our documents ('yes' or 'no').
"""
question: str
documents: List[Document]
generation: str
relevance_grade: str
检索节点：与知识库的交互
“检索节点：与知识库的交互”这一环节若被视为可度量的对象，效果会更好。在扩大范围之前，先记录一份最佳案例、一个失败实例以及回滚说明。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的、不完整的处理结果。 将分块策略与检索策略分开。当质量指标发生变化时，调整其中一项不应迫使重新编写另一项。
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.embeddings import FakeEmbeddings
# Initialize a mock vector store to simulate a production knowledge base
embeddings = FakeEmbeddings(size=1536)
vectorstore = InMemoryVectorStore(embeddings)
# Seed our vector store with sample documentation chunks
vectorstore.add_documents([
Document(page_content="LangGraph is a library for building stateful, multi-actor applications with LLMs.", metadata={"source": "docs"}),
Document(page_content="Self-RAG is an architecture that introduces self-reflection and grading into retrieval chains.", metadata={"source": "paper"}),
Document(page_content="Vector databases index text embeddings to enable fast semantic similarity search.", metadata={"source": "tech-blog"})
])
# Create a retriever interface
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
def retrieve(state: GraphState) -> Dict[str, Any]:
"""
Retrieve documents from the vector database based on the current question.
Args:
state (GraphState): The current state of the graph.
Returns:
dict: An update to the state containing the retrieved documents.
"""
print("--- NODE: RETRIEVING DOCUMENTS ---")
question = state["question"]
# Perform the semantic vector search
retrieved_docs = retriever.invoke(question)
# Return the updated document list to the graph state
return {"documents": retrieved_docs}
“生成节点：基于上下文的综合处理”
“生成节点：上下文感知合成”作为一种可度量的对象来处理时效果最佳。在扩大范围之前，先记录一个理想输出样本、一个失败案例以及回滚说明。 在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。 应将分块策略与检索策略分开。当质量指标发生变化时，调整其中一个不应强制要求重新编写另一个。 “生成节点：上下文感知合成”作为一种可度量的对象来处理时效果最佳。在扩大范围之前，先记录一个理想输出样本、一个失败案例以及回滚说明。 需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.language_models.fake import FakeListLLM
# Set up a mock LLM that simulates generating structured responses
# In production, replace this with: from langchain_openai import ChatOpenAI
llm = FakeListLLM(responses=[
"Self-RAG is an advanced architecture that uses self-reflection and document grading to improve retrieval accuracy."
])
def generate(state: GraphState) -> Dict[str, Any]:
"""
Generate an answer using the retrieved documents as grounding context.
Args:
state (GraphState): The current state of the graph.
Returns:
dict: An update to the state containing the generated answer.
"""
print("--- NODE: GENERATING ANSWER ---")
question = state["question"]
documents = state["documents"]
# Format the retrieved documents into a single context block
context = "\n\n".join([doc.page_content for doc in documents])
# Construct the instruction prompt
prompt = ChatPromptTemplate.from_template(
"You are an assistant for question-answering tasks.\n"
"Use the following pieces of retrieved context to answer the question.\n"
"If you do not know the answer, say that you do not know.\n"
"Question: {question}\n"
"Context: {context}\n"
"Answer:"
)
# Format the prompt with our state variables and run the model
chain = prompt | llm
generation = chain.invoke({"question": question, "context": context})
# Save the generated response to the state
return {"generation": generation}
“评分文档节点：质量检查员”
对于“等级文档节点”：质量检查员在修改代码之前应明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应指向单一责任主体，而非复杂的流程链。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分是虚假信息还是索引缺失所致。
+-------------------+
| Retrieved Docs |
+-------------------+
|
v
+-----------------------+
| Is Doc Relevant? |
+-----------------------+
/ \
Yes / \ No
/ \
v v
+-------------------+ +------------------+
| Keep in State | | Filter / Discard |
| (Pass to Gen) | | (Trigger Loop) |
+-------------------+ +------------------+
from pydantic import BaseModel, Field
from langchain_core.output_parsers import JsonOutputParser
class DocumentGrade(BaseModel):
"""Binary score for document relevance assessment."""
binary_score: str = Field(
description="Relevance score for the document. Must be 'yes' or 'no'."
)
def grade_documents(state: GraphState) -> Dict[str, Any]:
"""
Assess whether the retrieved documents are relevant to the user question.
Filters out any irrelevant documents from the state.
Args:
state (GraphState): The current state of the graph.
Returns:
dict: Updated documents list and a global relevance grade flag.
"""
print("--- NODE: GRADING RETRIEVED DOCUMENTS ---")
question = state["question"]
documents = state["documents"]
# Use a mock parser to simulate LLM grading
# In production, use: llm.with_structured_output(DocumentGrade)
parser = JsonOutputParser(pydantic_object=DocumentGrade)
filtered_docs = []
has_relevant_doc = "no"
# Evaluate each document individually
for doc in documents:
# Construct a strict grading prompt
grade_prompt = (
f"Question: {question}\n"
f"Document: {doc.page_content}\n"
"Assess if this document is relevant to the question. Output JSON with key 'binary_score' as 'yes' or 'no'."
)
# Simulating LLM evaluation logic (in production, run the LLM chain here)
# For our mock run, if the query terms overlap with the document, mark 'yes'
is_relevant = any(term.lower() in doc.page_content.lower() for term in question.split())
score = "yes" if is_relevant else "no"
if score == "yes":
print(f" -> DOCUMENT GRADING: [RELEVANT] - Source: {doc.metadata.get('source')}")
filtered_docs.append(doc)
has_relevant_doc = "yes"
else:
print(f" -> DOCUMENT GRADING: [NOT RELEVANT] - Source: {doc.metadata.get('source')}")
# Update state with filtered documents and set the control flag
return {
"documents": filtered_docs,
"relevance_grade": has_relevant_doc
}
构建图结构，第二部分：用条件边连接决策
在“构建图结构，第二部分：用条件边连接决策”中，应在修改代码之前明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功标准，并拒绝默许的半完成状态。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分幻觉内容与索引缺失问题。
+------------------+
| route_question |
+--------+---------+
|
+--------------+--------------+
| |
[web_search] [retrieve]
| |
+--------------> +------------+-------------+
| grade_documents |
+------------+--------------+
|
+-----+-----+
| Decision |
+-----+-----+
|
+-----------------+-----------------+
| |
(transform_query) (generate)
^ |
| +------+------+
| | Decision |
| +------+------+
| |
+--- [hallucination / unsat] -------+------> [__end__]
入口点路由器：先检查查询
对于入口点路由器：在修改代码之前，应先检查查询内容，明确输入参数、该步骤的负责人以及退出标准。操作人员应当能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本信息可以避免在流程从演示环境转向共享环境时出现意外费用。 需注明实际作为答案依据的段落。如果没有引用，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。 对于入口点路由器：在修改代码之前，应先检查查询内容，明确输入参数、该步骤的负责人以及退出标准。操作人员应当能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及死信处理都属于其中的一部分。
产品，而非后续的优化。# router.py
from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from state import AgentState
# Define the structured output schema for our router
class RouteQuery(BaseModel):
"""Route a user query to the most authoritative datasource."""
datasource: Literal["vectorstore", "web_search"] = Field(
...,
description="Given a user question choose to route it to web search or a vectorstore.",
)
def route_question(state: AgentState) -> str:
"""
Routes the initial question to either the vectorstore retrieval node or web search.
Args:
state (AgentState): The current graph state containing the user query.
Returns:
str: The next node to execute ("vectorstore" or "web_search").
"""
print("--- ROUTING QUESTION ---")
question = state["question"]
# Initialize our LLM with structured output mapping to our Pydantic schema
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_router = llm.with_structured_output(RouteQuery)
# Create the routing system prompt
system_prompt = """You are an expert router directing user questions to a vectorstore or web search.
The vectorstore contains highly technical documentation regarding LangGraph, LangChain, and state machines.
Use the vectorstore for these specialized topics. Otherwise, use web_search for general knowledge or current events."""
route_prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{question}"),
])
# Run the router chain
router_chain = route_prompt | structured_router
decision = router_chain.invoke({"question": question})
if decision.datasource == "web_search":
print("--- ROUTING TO WEB SEARCH ---")
return "web_search"
print("--- ROUTING TO VECTORSTORE ---")
return "retrieve"
上下文门：评估文档相关性
在研究“上下文门：评估文档相关性”时，首先列出相关内容：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词很难解决检索效果不佳的问题。
# context_gate.py
from state import AgentState
def decide_to_generate(state: AgentState) -> str:
"""
Evaluates the document grading state to choose the next step.
If any documents were graded as relevant, we proceed to answer generation.
If all documents were deemed irrelevant, we rewrite the query to try again.
Args:
state (AgentState): The current state of the agent.
Returns:
str: Next node to execute ("generate" or "transform_query").
"""
print("--- ASSESSING DOCUMENT RELEVANCE ---")
# Retrieve the decision variable set by the grading node
run_web_search = state.get("run_web_search", False)
if run_web_search:
# No documents were relevant, we must rewrite and search the web
print("--- DECISION: ALL DOCUMENTS IRRELEVANT, ROUTING TO QUERY TRANSFORMATION ---")
return "transform_query"
# We found relevant documents, we can proceed safely to generation
print("--- DECISION: RELEVANT DOCUMENTS FOUND, ROUTING TO GENERATE ---")
return "generate"
自我修正循环：验证准确性与实用性
在研读《自我修正循环：验证忠实性与实用性》时，首先需列出相关契约：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
+-------------------------+
| Answer Generated |
+------------+------------+
|
v
+--------------------------------+
| Is the answer grounded in |
| retrieved context? |
+----------------+---------------+
|
+---------------+---------------+
| |
[No] [Yes]
| |
v v
+-------------------+ +---------------------+
| Re-route to: | | Is the answer |
| "generate" | | actually useful? |
+-------------------+ +----------+----------+
|
+--------------+--------------+
| |
[No] [Yes]
| |
v v
+-------------------+ +------------------+
| Re-route to: | | Re-route to: |
| "transform_query" | | "__end__" |
+-------------------+ +------------------+
# self_correction.py
from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from state import AgentState
class GradeHallucinations(BaseModel):
"""Binary score for hallucination present in generation."""
binary_score: Literal["yes", "no"] = Field(
...,
description="Answer is grounded in the facts, 'yes' or 'no'",
)
class GradeAnswer(BaseModel):
"""Binary score to assess if answer addresses question."""
binary_score: Literal["yes", "no"] = Field(
...,
description="Answer addresses the question, 'yes' or 'no'",
)
def grade_generation_v_documents_and_question(state: AgentState) -> str:
"""
Grades the generation for hallucinations and utility.
Args:
state (AgentState): The current graph state.
Returns:
str: Next step path ("supported" -> check utility, "not supported" -> regenerate, "not useful" -> rewrite query)
"""
print("--- CHECKING FOR HALLUCINATIONS ---")
question = state["question"]
documents = state["documents"]
generation = state["generation"]
# Initialize evaluator LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ----------------------------------------
# STEP 1: Groundedness (Hallucination) Check
# ----------------------------------------
hallucination_grader = llm.with_structured_output(GradeHallucinations)
hallucination_system_prompt = """You are an objective grader assessing whether an LLM generation is grounded in / supported by a set of retrieved facts.
Give a binary score of 'yes' if the generation is strictly grounded in and supported by the provided facts. Otherwise, give 'no'."""
hallucination_prompt = ChatPromptTemplate.from_messages([
("system", hallucination_system_prompt),
("human", "Set of facts: \n\n {documents} \n\n LLM generation: {generation}"),
])
# Format document contents into a single string context
doc_context = "\n\n".join([doc.page_content for doc in documents])
hallucination_chain = hallucination_prompt | hallucination_grader
res = hallucination_chain.invoke({"documents": doc_context, "generation": generation})
if res.binary_score == "no":
print("--- DECISION: HALLUCINATION DETECTED! RE-TRYING GENERATION ---")
return "generate"
print("--- DECISION: GENERATION IS GROUNDED IN DOCUMENTS ---")
# ----------------------------------------
# STEP 2: Utility (Answer Relevance) Check
# ----------------------------------------
answer_grader = llm.with_structured_output(GradeAnswer)
answer_system_prompt = """You are an objective grader assessing whether an LLM generation addresses and resolves the user's question.
Give a binary score of 'yes' if the answer directly addresses and completely answers the user's question. Otherwise, give 'no'."""
answer_prompt = ChatPromptTemplate.from_messages([
("system", answer_system_prompt),
("human", "User question: \n\n {question} \n\n LLM generation: {generation}"),
])
answer_chain = answer_prompt | answer_grader
res_answer = answer_chain.invoke({"question": question, "generation": generation})
if res_answer.binary_score == "yes":
print("--- DECISION: ANSWER IS USEFUL AND ACCURATE. ENDING GRAPH ---")
return "useful"
print("--- DECISION: ANSWER DOES NOT FULLY ADDRESS QUESTION. REWRITING QUERY ---")
return "not useful"
组装与编译StateGraph
在构建和编译StateGraph时，首先需明确相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改始终符合预期。 在记录功能结果的同时，还需标注执行时间以及Token或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，应先使用固定的问题集测试系统的召回率。仅仅更换提示词往往无法改善较差的检索效果。 在构建和编译StateGraph时，首先需明确相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改始终符合预期。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误消息处理都是产品不可或缺的部分，而非后续才需要补充的功能。
# graph_builder.py
from langgraph.graph import END, StateGraph
from state import AgentState
# Import node functions from Part 1
from nodes import retrieve, generate, transform_query, web_search, grade_documents
# Import conditional routing edges
from router import route_question
from context_gate import decide_to_generate
from self_correction import grade_generation_v_documents_and_question
# Initialize our state-based graph
workflow = StateGraph(AgentState)
# ----------------------------------------
# 1. Register All Computational Nodes
# ----------------------------------------
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)
workflow.add_node("web_search", web_search)
# ----------------------------------------
# 2. Configure Node Routing & Transitions
# ----------------------------------------
# The entry point of our execution is dynamic, decided by the routing edge
workflow.set_conditional_entry_point(
route_question,
{
"web_search": "web_search",
"retrieve": "retrieve",
},
)
# After retrieval, we must immediately grade our retrieved documents
workflow.add_edge("retrieve", "grade_documents")
# Once documents are graded, the Context Gate decides where to go
workflow.add_conditional_edges(
"grade_documents",
decide_to_generate,
{
"transform_query": "transform_query",
"generate": "generate",
},
)
# If query transformation occurs, we execute a web search to gather new data
workflow.add_edge("transform_query", "web_search")
# After web searching, we feed those newly acquired documents back to the grade step
workflow.add_edge("web_search", "grade_documents")
# Once generation occurs, our Self-Correction edge decides the final quality check
workflow.add_conditional_edges(
"generate",
grade_generation_v_documents_and_question,
{
"generate": "generate", # Loop back to fix hallucination
"not useful": "transform_query", # Loop back to rewrite query and search
"useful": END, # Exit cleanly
},
)
# ----------------------------------------
# 3. Compile and Run the Graph
# ----------------------------------------
app = workflow.compile()
# visualize.py
from graph_builder import app
try:
# Print the plain text visualization of the state machine architecture
app.get_graph().print_ascii()
except Exception as e:
print(f"ASCII generation failed: {e}")
+----------------+
| __start__ |
+-------+--------+
|
+---------+---------+
v v
+------------+ +------------+
| web_search | | retrieve |
+-----+------+ +-----+------+
| |
| v
| +-----+------+
| | grade_docs |
| +-----+------+
| |
+---------+---------+
|
v
+---------+---------+
| decide_to_gen |
+----+---------+----+
| |
+----------+ +----------+
v v
+-----+------+ +-----+------+
| transform_q| | generate |
+-----+------+ +-----+------+
| |
| v
| +---------+---------+
| | self_correct |
| +----+----+----+----+
| | | |
+--------------------------+ | +---> [__end__]
v
(regenerate)
工程综合方法
将工程综合方法视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份优秀的实现方案、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应指向单一责任主体，而非复杂的流程链。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
实际应用场景
将“实际应用场景”视为可度量的对象来处理，效果最佳。在扩大范围之前，先收集一份优秀的案例记录、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地只完成部分工作。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
高级客户支持机器人
将高级客户支持机器人视为可度量的对象使用效果最佳。在扩大应用范围之前，先记录一份理想的对话文本、一个故障案例以及回滚说明。在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，就能避免在系统从演示环境过渡到共享环境时出现意外账单。应将分块策略与检索策略分开处理，当质量指标发生变化时，修改其中一项无需强制重写另一项。将高级客户支持机器人视为可度量的对象使用效果最佳。在扩大应用范围之前，先记录一份理想的对话文本、一个故障案例以及回滚说明。需同时记录正常流程与恢复流程的文档，重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。
自动化事实核查系统
对于自动化事实核查系统，在修改代码之前需明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比庞大的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一责任主体，而非复杂的流程链。必须引用实际作为答案依据的段落；没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失所致。
动态推荐引擎
对于动态推荐引擎，在修改代码之前需明确输入参数、各处理步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，设定成功标准，杜绝无声的半完成状态。 必须引用实际作为答案依据的段落。若没有引用，操作人员就无法区分幻觉内容与索引缺失问题。
科学研究助手
对于科学研究助手而言，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本可避免在流程从演示环境转向共享环境时出现意外费用。 需注明实际作为答案依据的段落。没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。 对于科学研究助手而言，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续需要补充的内容。
何时该使用哪种方法？
在思考“何时该使用哪种方法？”时，首先列出相关约定：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出错时，错误应指向单一责任点，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词很难解决检索效果不佳的问题。
生产环境约束：成本、延迟与可观测性
在研究“生产环境防护机制：成本、延迟与可观测性”时，首先需明确契约内容：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合要求。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，定义成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
通过非对称路由平衡成本与延迟
在处理通过非对称路由平衡成本与延迟的问题时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在记录功能结果的同时，还需标注处理时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境切换到共享环境时出现意外费用。 在调整提示词之前，应先使用固定的问题集测试系统的召回率。仅仅更换提示词往往无法改善较差的检索效果。 在处理通过非对称路由平衡成本与延迟的问题时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都属于产品功能的一部分，而非后续需要补充的内容。
LangGraph 状态中的强制停止断路器
将 LangGraph 状态中的强制停止断路器视为可测量的对象使用效果最佳。在扩大范围之前，先记录一份理想的输出样本、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应指向单一责任模块，而非复杂的流程链。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
from typing import Dict, TypedDict, Literal
# Define graph state with a loop counter to act as a circuit breaker
class GraphState(TypedDict):
question: str
generation: str
documents: list[str]
loop_step: int # Tracks iterations to prevent runaway execution costs
MAX_RETRIES = 3
def grade_generation_route(state: GraphState) -> Literal["web_search", "finalize", "max_retries_exceeded"]:
"""
Evaluates the quality of generation and routes the agent accordingly.
Why: Acts as a deterministic circuit breaker when LLMs get stuck in correction loops.
"""
current_step = state.get("loop_step", 0)
# 1. Evaluate circuit breaker before making any expensive LLM calls
if current_step >= MAX_RETRIES:
return "max_retries_exceeded"
# 2. Simulated grader check (In production, this queries your grading model)
is_accurate = False # Assume we failed the grade in this step
if is_accurate:
return "finalize"
else:
# Route back for retrieval/generation rewrite
return "web_search"
用于确定性评分的结构化提示词
“结构化提示以实现确定性评分”这一方法在被视为可度量的模型表面时效果最佳。在扩大应用范围之前，先记录一个理想输出样本、一个失败案例以及回滚说明。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。 需为每轮对话及每次会话设定token预算。智能工具往往会过度扩展上下文，设置上限可避免演示过程变成意外的费用账单。
from langchain_core.pydantic_v1 import BaseModel, Field
class GradeDocument(BaseModel):
"""Binary score for document relevance assessment."""
binary_score: str = Field(
description="Relevance score for the document. Must be 'yes' or 'no'."
)
explanation: str = Field(
description="A brief, single-sentence justification for the given score."
)
利用可观测性引擎追踪图执行过程
将可观测性引擎用于追踪图执行时，若将其视为可度量的对象会更为有效。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。 在功能结果之外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在系统从演示环境过渡到共享环境时出现意外费用。 应将分块策略与检索策略分开。当质量指标发生变化时，调整其中一个不应迫使重新编写另一个。 将可观测性引擎用于追踪图执行时，若将其视为可度量的对象会更为有效。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。 需同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。
[User Query] ──> [retrieve] ──> [grade_documents] (Score: No)
│
▼
[finalize] <── [generate] <── [web_search] (Updated State: Loop Count = 1)
状态驱动工作流能揭示什么
为了解析状态驱动工作流能揭示的信息，应在修改代码之前明确输入参数、各步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏的状态。 相较于庞大的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一责任模块，而非复杂的流程链。 需引用实际支撑答案的段落。没有引用的话，操作人员就无法区分是虚假信息还是索引缺失所致。
运营检查清单
将运营检查清单视为可量化的基准，其效果最佳。在扩大范围之前，需记录一份标准操作范本、一个故障案例以及回滚说明。 配置信息应与应用程序代码分开存放。环境文件、密钥存储和功能开关都应集中于一个位置，以便操作人员无需查看整个系统结构即可进行审核。
应将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制重新编写另一项。
对于涉及资金支出或更改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务上的完整性。
编写简短的操作手册：说明如何轮换密钥、如何清空队列以及如何回滚上一次的导入操作。
将此阶段视为输入数据与经过验证的输出结果之间的契约。为相关产物命名，明确成功标准，杜绝默许的半完成状态。
在升级整个系统之前，先冻结版本，为关键流程保存标准模板，并确认回滚步骤。共享环境需要设置速率限制、进行租户检查，同时明确密钥轮换的负责人。与其追求花哨的一次性演示，不如注重扎实的可靠性。
关于 eeb69aedbcbc 的批处理说明：不要将提供者密钥放入仓库中，设定单会话令牌上限，并将转录内容存储在评估测试用例的旁边，以便后续更换模型时仍能保持可比性。