CRAG架构：解决大模型幻觉问题的关键技术

1. 大模型幻觉问题的本质与CRAG的诞生

在大模型技术爆发的今天，GPT-4等通用大模型已经展现出惊人的文本理解和生成能力。然而，这些模型在严肃商业场景中的应用却面临一个根本性挑战——幻觉问题（Hallucination）。这种现象表现为模型会自信地生成看似合理但实际错误或虚构的内容，即使面对完全超出其知识范围的问题。

1.1 幻觉问题的技术根源

从技术架构来看，大模型的幻觉问题源于其概率生成机制的本质：

1. 自回归生成特性：大模型通过预测下一个token的概率分布逐步生成文本，这种机制本质上是一种"有根据的猜测"
2. 训练目标偏差：模型被优化为生成流畅、连贯的文本，而非绝对准确的事实
3. 知识边界模糊：模型缺乏明确的"不知道"机制，无法区分已知和未知领域

在金融风控等场景中，这种特性可能导致严重后果。例如，当询问"某上市公司2024年Q2的财务数据"时，模型可能基于过时或无关的上下文生成看似专业实则错误的回答。

1.2 传统RAG方案的局限性

检索增强生成（RAG）技术通过引入外部知识库部分缓解了这一问题，但标准实现存在明显缺陷：

• 垃圾进，垃圾出(Garbage In, Garbage Out)：如果检索到的文档质量差或相关性低，模型仍会基于错误上下文生成回答
• 相关性判断缺失：传统流程缺乏对检索结果的严格评估环节
• 错误传播风险：低质量文档中的错误信息可能被模型放大和传播

code复制传统RAG流程：
用户问题 -> 向量检索 -> 直接拼接上下文 -> LLM生成

2. CRAG架构设计与核心思想

2.1 CRAG的技术突破点

Corrective RAG(CRAG)通过引入"文档评估层"实现了质的飞跃。其核心创新在于：

1. 双阶段验证机制：
   • 第一阶段：文档相关性评估
   • 第二阶段：生成内容质量评估
2. 动态路由决策：
   • 相关文档：进入生成流程
   • 不相关文档：触发补救措施或拒绝回答
3. 可扩展的补救策略：
   • 查询重写
   • 备用检索源
   • 人工审核介入

2.2 LangGraph的技术优势

相比传统链式架构，LangGraph为CRAG提供了理想的实现框架：

3. 生产级CRAG系统实现

3.1 环境配置与依赖管理

建议使用Poetry进行依赖管理，核心依赖包括：

python复制[tool.poetry.dependencies]
python = "^3.9"
langgraph = "^0.0.30"
langchain-openai = "^0.1.0"
pydantic = "^2.5.0"

注意：生产环境应严格固定版本号以避免兼容性问题

3.2 状态机设计精要

CRAGState定义了系统运行时的完整上下文：

python复制class CRAGState(TypedDict):
    question: str               # 原始问题(immutable)
    generation: Optional[str]   # 生成结果
    documents: List[Document]   # 当前文档集合
    search_rewrite: Optional[str] # 查询重写结果
    evaluation: Dict[str, Any]  # 评估元数据
    fallback_triggered: bool    # 是否触发备用方案
    execution_path: List[str]   # 执行路径追踪

关键设计考虑：

• 不可变字段与可变字段分离
• 执行路径记录用于审计追踪
• 评估元数据存储完整决策依据

3.3 文档评估节点实现细节

评估器是CRAG系统的核心组件，需要平衡准确性与效率：

python复制def build_document_grader(llm: BaseChatModel):
    class GradingCriteria(BaseModel):
        relevance: Literal["yes", "no"] = Field(
            description="文档是否包含问题答案的关键要素")
        confidence: float = Field(
            ge=0, le=1, description="判断的置信度分数") 
        reasoning: str = Field(
            description="简要评估逻辑")
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """您是一个专业的文档评估专家。请基于以下标准判断文档与问题的相关性：
1. 文档是否包含回答问题所需的关键实体、数据或概念
2. 文档内容是否在问题的时间范围内
3. 文档来源是否具有足够权威性

只需评估文档内容本身，不考虑知识库外的信息。"""),
        ("human", "问题：{question}\n\n文档内容：{document}")
    ])
    
    return prompt | llm.with_structured_output(GradingCriteria)

评估策略优化技巧：

• 使用较小模型(GPT-3.5-turbo)降低成本
• 设置temperature=0确保一致性
• 批量处理文档提高吞吐量

3.4 条件路由的工程实践

决策节点需要处理多种边界情况：

python复制def route_decision(state: CRAGState) -> str:
    # 检查执行步数防止无限循环
    if len(state["execution_path"]) >= MAX_STEPS:
        return "force_terminate"
    
    # 无文档情况
    if not state["documents"]:
        return "trigger_fallback"
    
    # 评估结果分析
    evaluations = state["evaluation"].get("document_grades", [])
    if not evaluations:
        return "trigger_fallback"
    
    # 计算平均置信度
    avg_conf = sum(e.confidence for e in evaluations)/len(evaluations)
    
    if avg_conf < CONFIDENCE_THRESHOLD:
        if state["fallback_triggered"]:
            return "human_intervention"  # 避免无限重试
        return "trigger_fallback"
    
    return "generate_answer"

4. 生产环境部署考量

4.1 性能优化策略

1. 异步处理：

   python复制async def async_retrieve(state: CRAGState):
       loop = asyncio.get_event_loop()
       with ThreadPoolExecutor() as executor:
           docs = await loop.run_in_executor(
               executor, retriever.invoke, state["question"])
       return {"documents": docs}
   

2. 缓存机制：

   • 对频繁查询的问题缓存评估结果
   • 使用Redis存储文档评估分数

3. 批处理优化：

   • 同时评估多个文档
   • 实现评估请求的批量发送

4.2 监控与可观测性

关键监控指标：

• 文档评估准确率
• 平均响应延迟
• 补救策略触发频率
• 生成内容质量评分

推荐使用Prometheus + Grafana构建监控看板，重点关注：

code复制rate(crag_fallback_triggered_total[5m]) / rate(crag_requests_total[5m])

5. 典型问题排查指南

5.1 评估不一致问题

症状：相同文档在不同时间获得不同评估结果

排查步骤：

1. 检查模型temperature设置
2. 验证prompt是否包含随机元素
3. 确认文档预处理一致性(如截断长度)

解决方案：

python复制# 在prompt中增加确定性要求
system_prompt += "\n重要：您的评估必须严格基于文档客观内容，避免主观推测。"

5.2 性能瓶颈分析

常见瓶颈点：

1. 文档评估延迟
2. 向量检索耗时
3. 网络请求开销

优化方案：

python复制# 并行评估文档
from concurrent.futures import ThreadPoolExecutor

def batch_grade(docs: List[Document], question: str):
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(grader.invoke, 
                          {"document": doc.page_content, "question": question})
            for doc in docs
        ]
        return [f.result() for f in futures]

6. 进阶扩展方向

6.1 多模态CRAG

扩展架构支持图像、表格等非文本内容：

python复制class MultiModalState(CRAGState):
    image_documents: List[ImageDocument]
    table_documents: List[TableDocument]

def multi_modal_grader(document: Union[TextDocument, ImageDocument]):
    # 实现跨模态评估逻辑
    ...

6.2 动态阈值调整

根据问题类型自动调整置信度阈值：

python复制def dynamic_threshold(question: str) -> float:
    q_type = classify_question_type(question)
    return {
        "factual": 0.8,
        "analytical": 0.7,
        "creative": 0.5
    }.get(q_type, 0.7)

在实际部署CRAG系统时，建议从简单配置开始，逐步增加复杂度。我们团队的实施经验表明，合理的文档评估节点可以降低40%以上的幻觉错误，但同时会增加约30%的响应延迟。关键是要在准确性和性能之间找到适合业务场景的平衡点。