自愈型RAG系統(tǒng)：從脆弱管道到閉環(huán)智能體的工程實(shí)踐

RAG系統(tǒng)在生產(chǎn)環(huán)境中有個(gè)老大難問題：脆弱。演示時(shí)用精心準(zhǔn)備的問題去問，效果看起來很驚艷。但真正上線后，用戶的問題五花八門，向量數(shù)據(jù)庫返回的文檔語義上相似但實(shí)際答非所問，LLM又特別喜歡討好，拿著一堆噪音數(shù)據(jù)照樣能編出一套看似合理的答案。

那么問題出在哪呢？標(biāo)準(zhǔn)RAG是典型的開環(huán)架構(gòu)：輸入 → 嵌入 → 檢索 → 生成，一條線走到底。每個(gè)環(huán)節(jié)都假設(shè)上游輸出是完美的，一旦某步出錯(cuò)，錯(cuò)誤就會(huì)一路傳導(dǎo)到最終結(jié)果。

要做企業(yè)級(jí)的RAG應(yīng)用，必須轉(zhuǎn)向閉環(huán)系統(tǒng)，也就是所謂的自愈RAG。這里的核心思路是讓系統(tǒng)具備自省能力：檢測(cè)到問題后能自主糾正，而不是把錯(cuò)誤直接甩給用戶。

第一部分：自動(dòng)檢索

RAG的第一個(gè)坑其實(shí)是用戶本身。沒人會(huì)按照向量搜索的最佳實(shí)踐來寫查詢，要么用行話縮寫，要么問題模糊不清，要么一個(gè)問題里塞了好幾件事。自愈系統(tǒng)需要在輸入端加一道"防護(hù)欄"，把這些原始查詢轉(zhuǎn)換成高質(zhì)量的檢索請(qǐng)求。

策略1：假設(shè)文檔嵌入（HyDE）

傳統(tǒng)檢索是拿短問題去匹配長文檔，比如用"crag架構(gòu)"這幾個(gè)字去搜整段技術(shù)文檔。這種模態(tài)不匹配會(huì)嚴(yán)重影響召回質(zhì)量。

HyDE的思路是這樣的，先讓LLM根據(jù)問題"編造"一個(gè)假設(shè)性的答案，然后用這個(gè)假設(shè)答案去做向量檢索。因?yàn)榧僭O(shè)答案和真實(shí)文檔在形態(tài)上更接近，匹配效果自然更好。

文檔片段展示了其工作方式，HyDE能處理各類查詢，且不需要修改底層的GPT-3和Contriever/mContriever模型。

比如說：

用戶查詢："CRAG評(píng)分器怎么工作的？"

HyDE生成："CRAG評(píng)分器通過評(píng)估檢索文檔的相關(guān)性來運(yùn)作，它會(huì)對(duì)每個(gè)文檔打分……"（虛構(gòu)內(nèi)容）

向量搜索：用生成的內(nèi)容去檢索，而不是用原始問題

代碼實(shí)現(xiàn)（hyde.py）：

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.indices.query.query_transform import HyDEQueryTransform
from llama_index.core.query_engine import TransformQueryEngine
from llama_index.llms.openai import OpenAI
# 1. 配置用于生成假設(shè)文檔的LLM
Settings.llm = OpenAI(model="gpt-4-turbo", temperature=0.7)
def build_hyde_engine(index):
# 初始化HyDE轉(zhuǎn)換
# include_original=True 確保同時(shí)搜索原始查詢和假設(shè)文檔
hyde = HyDEQueryTransform(include_original=True)
# 創(chuàng)建標(biāo)準(zhǔn)檢索引擎
base_query_engine = index.as_query_engine(similarity_top_k=5)
# 用TransformQueryEngine包裝
# 這個(gè)中間件會(huì)攔截查詢，生成假設(shè)文檔，然后執(zhí)行搜索
hyde_engine = TransformQueryEngine(base_query_engine, query_transform=hyde)
return hyde_engine
# 使用示例
# index = VectorStoreIndex.from_documents(docs)
# engine = build_hyde_engine(index)
# response = engine.query("Explain the self-correction mechanism in CRAG")

策略2：查詢分解

用戶問"Llama-3和GPT-4在代碼任務(wù)上誰表現(xiàn)更好"，簡(jiǎn)單檢索很難找到一篇文檔同時(shí)包含兩個(gè)模型的對(duì)比數(shù)據(jù)。查詢分解就是把這種復(fù)合問題拆成原子級(jí)子查詢："Llama-3代碼能力"和"GPT-4代碼能力"，分別檢索后再合并結(jié)果。

代碼實(shí)現(xiàn)（query_decomposition.py）：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List
# 定義輸出結(jié)構(gòu)
class SubQueries(BaseModel):
"""待檢索的子問題集合"""
questions: List[str] = Field(description="List of atomic sub-questions.")
# 配置規(guī)劃用的LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
system_prompt = """You are an expert researcher. Break down the user's complex query.
into simple, atomic sub-queries that a search engine can answer."""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{query}")
])
# 構(gòu)建處理鏈
planner = prompt | llm.with_structured_output(SubQueries)
def plan_query(query: str):
result = planner.invoke({"query": query})
return result.questions
# 使用示例
# sub_qs = plan_query("Compare Llama-3 and GPT-4 on coding benchmarks")
# print(sub_qs)
# 輸出:

第二部分：控制層

文檔檢索回來了如何判斷它們靠不靠譜？CRAG的做法是在流程里加一個(gè)"評(píng)分員"角色，對(duì)每個(gè)檢索到的文檔進(jìn)行相關(guān)性評(píng)估。如果發(fā)現(xiàn)數(shù)據(jù)質(zhì)量不行，系統(tǒng)不會(huì)硬著頭皮生成答案，而是觸發(fā)備用方案（比如去搜網(wǎng)頁）。

檢索評(píng)估器的工作原理：評(píng)估檢索文檔與輸入的相關(guān)性，估算置信度，然后根據(jù)結(jié)果觸發(fā)不同的后續(xù)動(dòng)作——{正確、錯(cuò)誤、模糊}三種狀態(tài)對(duì)應(yīng)不同處理路徑。

這種分支決策邏輯用圖結(jié)構(gòu)來實(shí)現(xiàn)最合適，LangGraph正好派上用場(chǎng)。

CRAG工作流程如下：

1. 檢索：拿到候選文檔
2. 評(píng)分：LLM判斷每個(gè)文檔"相關(guān)"還是"不相關(guān)"
3. 決策：相關(guān)就直接生成答案；不相關(guān)則改寫查詢后去搜網(wǎng)頁

代碼實(shí)現(xiàn)（corrective_rag.py）：

from typing import List, TypedDict
from langchain_core.prompts import PromptTemplate
from langchain_core.documents import Document
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph, START
# --- 1. 狀態(tài)定義 ---
class GraphState(TypedDict):
question: str
generation: str
web_search: str # 'Yes'或'No'標(biāo)記
documents: List
# --- 2. 組件初始化 ---
grader_llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
generator_llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
web_tool = TavilySearchResults(k=3)
# --- 3. 節(jié)點(diǎn)定義 ---
def grade_documents(state):
"""
自愈核心節(jié)點(diǎn)：過濾低質(zhì)量文檔
"""
print("---CHECK RELEVANCE---")
question = state["question"]
documents = state["documents"]
# 二分類結(jié)構(gòu)化輸出
structured_llm = grader_llm.with_structured_output(dict)
prompt = PromptTemplate(
template="""You are a grader assessing relevance.
Doc: {context}
Question: {question}
Return JSON with key 'score' as 'yes' or 'no'.""",
input_variables=["context", "question"],
)
chain = prompt | structured_llm
filtered_docs = []
web_search = "No"
for d in documents:
grade = chain.invoke({"question": question, "context": d.page_content})
if grade.get('score') == 'yes':
filtered_docs.append(d)
else:
# 丟失上下文時(shí)觸發(fā)回退
web_search = "Yes"
return {"documents": filtered_docs, "question": question, "web_search": web_search}
def transform_query(state):
"""
自我糾正：重寫查詢以提升網(wǎng)頁搜索效果
"""
print("---TRANSFORM QUERY---")
question = state["question"]
# 簡(jiǎn)易重寫鏈
prompt = PromptTemplate(template="Rewrite this for web search: {question}", input_variables=["question"])
chain = prompt | generator_llm
better_q = chain.invoke({"question": question}).content
return {"question": better_q}
def web_search_node(state):
print("---WEB SEARCH---")
docs = web_tool.invoke({"query": state["question"]})
# 網(wǎng)頁結(jié)果追加到已有文檔
web_results = [Document(page_content=d["content"]) for d in docs]
return {"documents": state["documents"] + web_results}
def generate(state):
print("---GENERATE---")
# 這里接標(biāo)準(zhǔn)RAG生成鏈
# generation = rag_chain.invoke(...)
return {"generation": "Final Answer Placeholder"}
# --- 4. 圖構(gòu)建 ---
workflow = StateGraph(GraphState)
# 添加節(jié)點(diǎn)
workflow.add_node("retrieve", lambda x: {"documents": []}) # 檢索占位
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("transform_query", transform_query)
workflow.add_node("web_search_node", web_search_node)
workflow.add_node("generate", generate)
# 添加邊
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
def decide_to_generate(state):
if state["web_search"] == "Yes":
return "transform_query"
return "generate"
workflow.add_conditional_edges(
"grade_documents",
decide_to_generate,
{"transform_query": "transform_query", "generate": "generate"}
)
workflow.add_edge("transform_query", "web_search_node")
workflow.add_edge("web_search_node", "generate")
workflow.add_edge("generate", END)
app = workflow.compile()

第三部分：自動(dòng)排序

向量檢索用的雙編碼器（Bi-Encoder）速度快但精度有限。文檔被壓縮成單個(gè)向量后，很多語義細(xì)節(jié)都丟了。解決辦法是引入交叉編碼器（Cross-Encoder）做二次排序。

交叉編碼器把查詢和文檔作為一個(gè)整體輸入，直接輸出相關(guān)性分?jǐn)?shù)的計(jì)算開銷比較大，所以一般采用兩階段策略：

1. 粗篩：向量庫快速召回Top 50
2. 精排：交叉編碼器對(duì)這50個(gè)文檔重新打分，保留Top 5

代碼實(shí)現(xiàn)（reranker.py）：

from sentence_transformers import CrossEncoder
class Reranker:
def __init__(self):
# 加載MS MARCO優(yōu)化過的模型
self.model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
def rerank(self, query, documents, top_k=5):
# 構(gòu)造配對(duì)：[[query, doc1], [query, doc2]...]
pairs = [[query, doc] for doc in documents]
# 批量打分
scores = self.model.predict(pairs)
# 排序截取
results = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, score in results[:top_k]]

第四部分：自動(dòng)學(xué)習(xí)

高級(jí)的自愈系統(tǒng)不只是即時(shí)修復(fù)問題，還會(huì)從歷史錯(cuò)誤中學(xué)習(xí)，避免同樣的坑反復(fù)踩。實(shí)現(xiàn)方式是動(dòng)態(tài)少樣本學(xué)習(xí)（Dynamic Few-Shot Learning）。

當(dāng)系統(tǒng)生成了一個(gè)好答案（用戶點(diǎn)了贊），就把這對(duì)查詢-答案存到一個(gè)專門的"黃金樣本庫"里。后續(xù)遇到相似問題時(shí)，檢索這些成功案例注入到prompt中，相當(dāng)于用系統(tǒng)自己的成功經(jīng)驗(yàn)來指導(dǎo)新的回答。

代碼實(shí)現(xiàn)（dynamic_prompting.py）：

from llama_index.core import VectorStoreIndex, Document
from llama_index.core.prompts import PromptTemplate
class LearningManager:
def __init__(self):
self.good_examples = []
self.index = None
def add_good_example(self, query, answer):
"""用戶點(diǎn)贊時(shí)調(diào)用"""
doc = Document(text=f"Q: {query}\nA: {answer}")
self.good_examples.append(doc)
# 重建索引（生產(chǎn)環(huán)境建議用支持增量更新的向量庫）
self.index = VectorStoreIndex.from_documents(self.good_examples)
def get_dynamic_prompt(self, current_query):
if not self.index:
return ""
# 檢索相似的歷史成功案例
retriever = self.index.as_retriever(similarity_top_k=2)
nodes = retriever.retrieve(current_query)
examples_text = "\n\n".join([n.text for n in nodes])
return f"Here are examples of how to answer correctly:\n{examples_text}"
# 在管道中使用
# manager = LearningManager()
# few_shot_context = manager.get_dynamic_prompt(user_query)
# final_prompt = f"{few_shot_context}\n\nQuestion: {user_query}..."

進(jìn)階方向：DSPy自動(dòng)優(yōu)化

如果想要更程序化的優(yōu)化方式，DSPy是個(gè)值得關(guān)注的框架。它把prompt當(dāng)成可優(yōu)化的程序來處理，他會(huì)跑一遍驗(yàn)證集并根據(jù)準(zhǔn)確率等指標(biāo)自動(dòng)重寫prompt和更新少樣本示例。

import dspy
# 1. 定義RAG簽名
class GenerateAnswer(dspy.Signature):
"""用簡(jiǎn)短事實(shí)性答案回答問題"""
context = dspy.InputField()
question = dspy.InputField()
answer = dspy.OutputField()
# 2. 定義模塊
class RAG(dspy.Module):
def __init__(self):
super().__init__()
self.retrieve = dspy.Retrieve(k=3)
self.generate = dspy.ChainOfThought(GenerateAnswer)
def forward(self, question):
context = self.retrieve(question).passages
return self.generate(context=context, question=question)
# 3. 優(yōu)化
# MIPROv2會(huì)運(yùn)行管道，遇到失敗就重試并重寫指令
# 目標(biāo)是最大化指定metric（精確匹配、語義相似度等）
optimizer = dspy.MIPROv2(metric=dspy.evaluate.SemanticF1)
optimized_rag = optimizer.compile(RAG(), trainset=training_data)

完整系統(tǒng)集成

各個(gè)組件都準(zhǔn)備好了：HyDE、查詢分解、CRAG、交叉編碼器重排序、動(dòng)態(tài)提示?，F(xiàn)在把它們串成一個(gè)完整的自愈RAG系統(tǒng)。這個(gè)編排層負(fù)責(zé)協(xié)調(diào)整個(gè)流程：解析查詢、增強(qiáng)檢索、校驗(yàn)上下文、優(yōu)化相關(guān)性、收集反饋學(xué)習(xí)、最終生成穩(wěn)定可靠的答案。

import os
import json
import asyncio
from typing import List, Dict, Any, Optional
from datetime import datetime
# 導(dǎo)入各組件
from hyde import build_hyde_engine, Settings
from query_decomposition import plan_query, SubQueries
from corrective_rag import app as crag_app, GraphState
from reranker import Reranker
from dynamic_prompting import LearningManager
# 核心依賴
from llama_index.core import VectorStoreIndex, Document, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from sentence_transformers import CrossEncoder
class SelfHealingRAGSystem:
"""
完整自愈RAG系統(tǒng)，整合全部組件
"""
def __init__(self, openai_api_key: str = None):
"""初始化RAG系統(tǒng)"""
# API密鑰配置
if openai_api_key:
os.environ["OPENAI_API_KEY"] = openai_api_key
# 組件初始化
print(" Initializing Self-Healing RAG System...")
# 核心LLM
self.llm = OpenAI(model="gpt-4-turbo", temperature=0.3)
Settings.llm = self.llm
# 初始化各組件
self.reranker = Reranker()
self.learning_manager = LearningManager()
self.vector_index = None
self.hyde_engine = None
# 演示數(shù)據(jù)
self.sample_documents = self._create_sample_documents()
self._setup_vector_index()
# 統(tǒng)計(jì)
self.query_stats = {
"total_queries": 0,
"hyde_used": 0,
"decomposed_queries": 0,
"crag_activated": 0,
"reranked": 0,
"learning_applied": 0
}
print("? System initialized successfully!")
def _create_sample_documents(self) -> List[Document]:
"""創(chuàng)建演示用的示例文檔"""
sample_texts = [
"""Retrieval-Augmented Generation (RAG) is a technique that combines
pre-trained language models with external knowledge retrieval. RAG systems
retrieve relevant documents from a knowledge base and use them to generate
more accurate and factual responses.""",
"""Corrective RAG (CRAG) introduces a self-correction mechanism that grades
retrieved documents for relevance. If documents are deemed irrelevant, the
system triggers alternative retrieval strategies like web search.""",
"""HyDE (Hypothetical Document Embeddings) improves retrieval by generating
hypothetical documents that answer the query, then searching for real documents
similar to these hypothetical ones.""",
"""Cross-encoder reranking provides more accurate document scoring compared
to bi-encoder similarity search. It processes query-document pairs together
to produce refined relevance scores.""",
"""DSPy enables automatic prompt optimization by treating prompts as programs
that can be compiled and optimized against specific metrics like accuracy
or semantic similarity.""",
"""Self-healing RAG systems implement feedback loops that learn from successful
query-answer pairs, storing them as examples for future similar queries to
improve performance over time.""",
"""Query decomposition breaks complex multi-part questions into atomic
sub-queries that can be individually processed and then combined for
comprehensive answers.""",
"""Vector databases enable semantic search by converting documents into
high-dimensional embeddings that capture semantic meaning rather than
just keyword matches."""
]
return [Document(text=text, metadata={"id": i}) for i, text in enumerate(sample_texts)]
def _setup_vector_index(self):
"""用示例文檔構(gòu)建向量索引"""
print(" Setting up vector index...")
self.vector_index = VectorStoreIndex.from_documents(self.sample_documents)
self.hyde_engine = build_hyde_engine(self.vector_index)
print("? Vector index ready!")
def enhanced_retrieve(self, query: str, use_hyde: bool = True, top_k: int = 5) -> List[Document]:
"""支持HyDE的增強(qiáng)檢索"""
print(f" Retrieving documents for: '{query}'")
if use_hyde:
print(" Using HyDE for enhanced retrieval...")
response = self.hyde_engine.query(query)
# 從HyDE響應(yīng)提取文檔
documents = response.source_nodes
self.query_stats["hyde_used"] += 1
else:
print(" Using standard retrieval...")
retriever = self.vector_index.as_retriever(similarity_top_k=top_k)
nodes = retriever.retrieve(query)
documents = nodes
# 轉(zhuǎn)換為Document對(duì)象
docs = []
for node in documents:
doc = Document(
page_content=node.text if hasattr(node, 'text') else str(node),
metadata=node.metadata if hasattr(node, 'metadata') else {}
)
docs.append(doc)
print(f" ? Retrieved {len(docs)} documents")
return docs
def decompose_and_retrieve(self, query: str) -> tuple[List[str], List[Document]]:
"""分解復(fù)雜查詢并分別檢索"""
print(f" Decomposing query: '{query}'")
try:
sub_queries = plan_query(query)
if len(sub_queries) > 1:
print(f" Decomposed into {len(sub_queries)} sub-queries:")
for i, sq in enumerate(sub_queries, 1):
print(f" {i}. {sq}")
# 對(duì)每個(gè)子查詢檢索
all_docs = []
for sq in sub_queries:
docs = self.enhanced_retrieve(sq, use_hyde=False, top_k=3)
all_docs.extend(docs)
self.query_stats["decomposed_queries"] += 1
return sub_queries, all_docs
else:
print(" ?? Query doesn't need decomposition")
docs = self.enhanced_retrieve(query)
return [query], docs
except Exception as e:
print(f" ?? Error in decomposition: {e}")
docs = self.enhanced_retrieve(query)
return [query], docs
def apply_crag(self, query: str, documents: List[Document]) -> tuple[List[Document], str]:
"""應(yīng)用CRAG過濾文檔"""
print(" Applying CRAG (Corrective RAG)...")
try:
# 準(zhǔn)備CRAG狀態(tài)
state = GraphState(
question=query,
generation="",
web_search="No",
documents=documents
)
# 正常情況下會(huì)跑完整CRAG流程
# 這里為演示做簡(jiǎn)化處理
filtered_docs = []
for doc in documents[:3]: # 演示限制
# 簡(jiǎn)單相關(guān)性檢查（實(shí)際應(yīng)該用LLM）
if any(keyword in doc.page_content.lower() for keyword in query.lower().split()):
filtered_docs.append(doc)
if len(filtered_docs) < len(documents):
self.query_stats["crag_activated"] += 1
print(f" CRAG filtered {len(documents) - len(filtered_docs)} irrelevant documents")
return filtered_docs, "Documents filtered by CRAG"
except Exception as e:
print(f" ?? Error in CRAG: {e}")
return documents, "CRAG not applied due to error"
def apply_reranking(self, query: str, documents: List[Document], top_k: int = 3) -> List[Document]:
"""交叉編碼器重排序"""
print(" Applying cross-encoder reranking...")
try:
# 提取文本用于重排序
doc_texts = [doc.page_content for doc in documents]
if len(doc_texts) > 1:
reranked_texts = self.reranker.rerank(query, doc_texts, top_k)
# 映射回Document對(duì)象
reranked_docs = []
for text in reranked_texts:
for doc in documents:
if doc.page_content == text:
reranked_docs.append(doc)
break
self.query_stats["reranked"] += 1
print(f" ? Reranked to top {len(reranked_docs)} documents")
return reranked_docs
else:
print(" ?? Not enough documents for reranking")
return documents
except Exception as e:
print(f" ?? Error in reranking: {e}")
return documents
def apply_dynamic_prompting(self, query: str) -> str:
"""動(dòng)態(tài)少樣本學(xué)習(xí)"""
print(" Applying dynamic prompting...")
try:
few_shot_context = self.learning_manager.get_dynamic_prompt(query)
if few_shot_context:
self.query_stats["learning_applied"] += 1
print(" ? Applied learned examples from previous successes")
else:
print(" ?? No relevant past examples found")
return few_shot_context
except Exception as e:
print(f" ?? Error in dynamic prompting: {e}")
return ""
def generate_answer(self, query: str, documents: List[Document], few_shot_context: str = "") -> str:
"""基于檢索文檔生成答案"""
print("?? Generating final answer...")
# 合并文檔內(nèi)容
context = "\n\n".join([doc.page_content for doc in documents[:3]])
# 構(gòu)建prompt，可選包含少樣本示例
prompt_parts = []
if few_shot_context:
prompt_parts.append(few_shot_context)
prompt_parts.extend([
"Context:",
context,
f"\nQuestion: {query}",
"\nAnswer based on the provided context:"
])
prompt = "\n".join(prompt_parts)
try:
response = self.llm.complete(prompt)
answer = response.text.strip()
print(" ? Answer generated successfully")
return answer
except Exception as e:
print(f" ?? Error generating answer: {e}")
return f"I apologize, but I encountered an error generating an answer: {e}"
def full_pipeline(self, query: str, user_feedback: bool = None, previous_answer: str = None) -> Dict[str, Any]:
"""
運(yùn)行完整自愈RAG管道
"""
start_time = datetime.now()
print(f"\n Starting Self-Healing RAG Pipeline")
print(f"Query: '{query}'")
print("=" * 60)
self.query_stats["total_queries"] += 1
# 步驟1：查詢?cè)鰪?qiáng)
sub_queries, documents = self.decompose_and_retrieve(query)
# 步驟2：文檔校驗(yàn)（CRAG）
filtered_docs, crag_status = self.apply_crag(query, documents)
# 步驟3：文檔重排序
reranked_docs = self.apply_reranking(query, filtered_docs)
# 步驟4：動(dòng)態(tài)提示
few_shot_context = self.apply_dynamic_prompting(query)
# 步驟5：答案生成
answer = self.generate_answer(query, reranked_docs, few_shot_context)
# 步驟6：學(xué)習(xí)（如有反饋）
if user_feedback is True and previous_answer:
try:
self.learning_manager.add_good_example(query, previous_answer)
print(" Added successful example to learning system")
except Exception as e:
print(f"?? Error adding to learning system: {e}")
end_time = datetime.now()
processing_time = (end_time - start_time).total_seconds()
result = {
"query": query,
"sub_queries": sub_queries,
"documents_found": len(documents),
"documents_filtered": len(filtered_docs),
"final_documents": len(reranked_docs),
"answer": answer,
"crag_status": crag_status,
"processing_time": processing_time,
"components_used": self._get_components_used()
}
print("\n" + "=" * 60)
print(f"? Pipeline completed in {processing_time:.2f} seconds")
print(f" Documents: {len(documents)} → {len(filtered_docs)} → {len(reranked_docs)}")
return result
def _get_components_used(self) -> List[str]:
"""獲取本次查詢用到的組件"""
components = ["Vector Retrieval"]
if self.query_stats["hyde_used"] > 0:
components.append("HyDE")
if self.query_stats["decomposed_queries"] > 0:
components.append("Query Decomposition")
if self.query_stats["crag_activated"] > 0:
components.append("CRAG")
if self.query_stats["reranked"] > 0:
components.append("Cross-Encoder Reranking")
if self.query_stats["learning_applied"] > 0:
components.append("Dynamic Prompting")
return components
def get_system_stats(self) -> Dict[str, Any]:
"""獲取系統(tǒng)統(tǒng)計(jì)信息"""
return {
"total_queries": self.query_stats["total_queries"],
"hyde_usage_rate": f"{(self.query_stats['hyde_used'] / max(1, self.query_stats['total_queries']) * 100):.1f}%",
"decomposition_rate": f"{(self.query_stats['decomposed_queries'] / max(1, self.query_stats['total_queries']) * 100):.1f}%",
"crag_activation_rate": f"{(self.query_stats['crag_activated'] / max(1, self.query_stats['total_queries']) * 100):.1f}%",
"reranking_rate": f"{(self.query_stats['reranked'] / max(1, self.query_stats['total_queries']) * 100):.1f}%",
"learning_rate": f"{(self.query_stats['learning_applied'] / max(1, self.query_stats['total_queries']) * 100):.1f}%",
"learned_examples": len(self.learning_manager.good_examples)
}
def demo_interactive_session():
"""交互式演示"""
print("""
Self-Healing RAG System Demo
================================
This system demonstrates:
? HyDE: Hypothetical Document Embeddings
? Query Decomposition: Breaking complex queries
? CRAG: Corrective RAG with document grading
? Cross-Encoder Reranking: Precision ranking
? Dynamic Learning: Few-shot from success examples
""")
# 初始化系統(tǒng)
system = SelfHealingRAGSystem()
# 演示用查詢
demo_queries = [
"What is RAG and how does it work?",
"Compare HyDE and standard retrieval methods",
"How does CRAG improve retrieval quality and what are the benefits of cross-encoder reranking?",
"Explain the self-correction mechanisms in modern RAG systems",
"What are the advantages of DSPy optimization for prompts?"
]
print(" Running Demo Queries...")
print("=" * 50)
results = []
for i, query in enumerate(demo_queries, 1):
print(f"\n Demo Query {i}/{len(demo_queries)}")
result = system.full_pipeline(query)
results.append(result)
print(f"\n Answer:")
print(f"{result['answer']}")
print(f"\n Components Used: {', '.join(result['components_used'])}")
# 模擬正反饋用于學(xué)習(xí)
if i > 1: # 第二個(gè)查詢開始加反饋
system.full_pipeline(query, user_feedback=True, previous_answer=result['answer'])
# 最終統(tǒng)計(jì)
print("\n" + "=" * 60)
print(" SYSTEM PERFORMANCE STATISTICS")
print("=" * 60)
stats = system.get_system_stats()
for key, value in stats.items():
print(f"{key.replace('_', ' ').title()}: {value}")
return system, results
if __name__ == "__main__":
# 設(shè)置OpenAI API密鑰
# os.environ["OPENAI_API_KEY"] = "your-key-here"
demo_interactive_session()

總結(jié)

經(jīng)典的RAG到自愈RAG，本質(zhì)上是從"檢索"到"推理"的升級(jí)。HyDE和查詢分解確保問對(duì)問題；CRAG和交叉編碼器確保讀對(duì)文檔；自動(dòng)學(xué)習(xí)機(jī)制則讓系統(tǒng)不再反復(fù)犯同樣的錯(cuò)。這套組合下來，RAG系統(tǒng)的泛化性會(huì)有質(zhì)的提升。

https://avoid.overfit.cn/post/d95478d7799646acbed0e0d2dc2c480d

作者：Subrata Samanta