企业级RAG系统2.0架构设计：从向量检索到Agent决策（0909版）

企业级RAG系统2.0架构设计：从向量检索到Agent决策

为什么需要RAG 2.0

传统RAG（1.0）的典型流程是：用户提问 → 文本切分 → 向量检索 → LLM生成回答。这个流程存在三个明显缺陷：

1. 单一检索路径：无论什么问题都用向量相似度，无法区分"查文档"和"做计算"

2. 无记忆能力：每轮对话独立处理，多轮上下文丢失

3. 无自我优化：检索质量不好就接受，不会主动调整策略

RAG 2.0的核心改进：把检索变成一个Agent可以动态决策的过程，而非固定的流水线。

整体架构

用户输入
    │
    ▼
┌─────────────┐
│  Query Router│ ← 判断问题类型：文档查询/代码搜索/计算/闲聊
└──────┬──────┘
       │
   ┌───┴───────────────────┐
   ▼                       ▼
┌─────────┐          ┌──────────┐
│DocSearch│          │ CodeSearch│
│(向量库)  │          │(代码索引) │
└────┬────┘          └────┬─────┘
     │                    │
     └────────┬───────────┘
              ▼
     ┌────────────────┐
     │  Result Aggregator │ ← 合并多源结果，去重+排序
     └────────┬─────────┘
              │
              ▼
     ┌────────────────┐
     │  Reflection     │ ← 评估答案质量，不够好则重新检索
     └────────┬────────┘
              │
              ▼
     ┌────────────────┐
     │  LLM Generator  │ ← 最终答案生成
     └────────────────┘

LangGraph编排实现

使用LangGraph实现Agent循环：

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional
import ChromaDB

class RAGState(TypedDict):
    question: str
    query_type: str          # doc/code/compute/chitchat
    search_results: List[dict]
    reflection_score: float  # 0-1，答案质量评分
    final_answer: str
    retry_count: int

# 节点1：查询路由
def route_query(state: RAGState) -> RAGState:
    prompt = f"""分类以下问题类型：{state['question']}
    输出：doc(文档查询) / code(代码搜索) / compute(计算) / chitchat(闲聊)"""
    # 调用LLM进行分类
    state['query_type'] = call_llm(prompt)
    return state

# 节点2：多源检索
def search(state: RAGState) -> RAGState:
    results = []
    if state['query_type'] in ('doc', 'code', 'compute'):
        # 向量检索
        results += chroma_db.similarity_search(state['question'], top_k=5)
    if state['query_type'] == 'code':
        # 代码索引检索
        results += code_index.search(state['question'], top_k=3)
    state['search_results'] = deduplicate(results)
    return state

# 节点3：结果聚合
def aggregate(state: RAGState) -> RAGState:
    # 合并去重，按相关性排序
    state['search_results'] = rank_and_dedup(state['search_results'])
    return state

# 节点4：反思评估
def reflect(state: RAGState) -> RAGState:
    # 用LLM自评检索结果是否充分
    prompt = f"""评估以下检索结果是否足以回答：{state['question']}
    结果：{state['search_results']}
    输出0-1的充分度评分"""
    state['reflection_score'] = float(call_llm(prompt))
    return state

# 节点5：生成答案
def generate(state: RAGState) -> RAGState:
    context = "\n".join([r['content'] for r in state['search_results']])
    prompt = f"""基于以下资料回答问题：{state['question']}
    资料：{context}"""
    state['final_answer'] = call_llm(prompt)
    return state

# 构建图
graph = StateGraph(RAGState)
graph.add_node("route", route_query)
graph.add_node("search", search)
graph.add_node("aggregate", aggregate)
graph.add_node("reflect", reflect)
graph.add_node("generate", generate)

graph.set_entry_point("route")
graph.add_conditional_edges(
    "route",
    lambda s: s['query_type'],
    {"doc": "search", "code": "search", "compute": "search", "chitchat": "generate"}
)
graph.add_edge("search", "aggregate")
graph.add_edge("aggregate", "reflect")
graph.add_conditional_edges(
    "reflect",
    lambda s: "generate" if s['reflection_score'] >= 0.7 else "search",
    {"generate": "generate", "search": "search"}
)
graph.add_edge("generate", END)

关键设计要点

1. 查询路由的准确率至关重要——路由错误会导致检索方向完全错误。建议用few-shot prompt + 置信度阈值，低置信度时走"混合检索"兜底。

2. 反思节点的性价比最高——加入一个自检步骤，让LLM判断"当前检索结果够不够回答这个问题"，不够就自动再搜一轮，比单纯提升top_k更智能。

3. 去重策略影响最终质量——多源检索容易出现重复内容，需按embedding相似度聚类去重，保留每个簇的代表性片段。

部署方案

本地部署（开发环境）：

docker-compose up -d
# 包含：ChromaDB + LangGraph服务 + API Gateway

生产部署（K8s）：
- 向量库：Milvus或Qdrant（支持分布式集群）
- Agent运行时：自托管LangGraph服务
- 缓存层：Redis存储热门查询结果（TTL=30min）
- 监控：Prometheus + Grafana追踪检索延迟和准确率

与传统RAG对比