RAG + Agent 架构模式：检索、推理与执行如何分工

# Classic RAG：一次检索，一次生成
def classic_rag(query: str) -> str:
    chunks = retriever.search(query, top_k=5)
    context = "\n".join(c.text for c in chunks)
    return llm.generate(
        system="根据以下上下文回答用户问题，如果上下文无法回答请说明。",
        user=f"上下文：\n{context}\n\n问题：{query}"
    )

# Agentic RAG：多轮检索 + 自主判断
def agentic_rag(query: str, max_hops: int = 3) -> str:
    state = {"question": query, "evidence": [], "hops": 0}
 
    while state["hops"] < max_hops:
        # Agent 决策：继续检索还是直接回答
        decision = agent.decide(state)
 
        if decision.action == "answer":
            return agent.answer(state)
 
        if decision.action == "retrieve":
            chunks = retriever.search(decision.query, top_k=5)
            state["evidence"].extend(chunks)
            state["hops"] += 1
 
        elif decision.action == "call_tool":
            result = tools.execute(decision.tool, decision.params)
            state["evidence"].append(result)
            state["hops"] += 1
 
    # 超过最大轮次，用已有信息尽力回答
    return agent.answer(state)

# Multi-Agent RAG：分工检索 + 汇总
def multi_agent_rag(query: str) -> str:
    # 编排 Agent 拆解任务
    sub_tasks = orchestrator.decompose(query)
    results = {}
 
    for task in sub_tasks:
        # 根据子任务类型分配给专业 Agent
        if task.domain == "internal_docs":
            results[task.id] = internal_agent.execute(task)
        elif task.domain == "public_web":
            results[task.id] = web_agent.execute(task)
        elif task.domain == "database":
            results[task.id] = sql_agent.execute(task)
 
    # 编排 Agent 汇总所有子结果
    return orchestrator.synthesize(query, results)

# 查询预处理：意图识别 + query 改写
def preprocess_query(query: str) -> dict:
    analysis = llm.generate(
        system="分析用户查询的意图、领域、复杂度。",
        user=query
    )
 
    result = {
        "original": query,
        "intent": analysis.intent,       # "search" | "execute" | "mixed"
        "domain": analysis.domain,        # "internal" | "public" | "mixed"
        "complexity": analysis.complexity # "simple" | "multi_hop" | "agentic"
    }
 
    # 如果 query 模糊，做 step-back 改写
    if analysis.complexity == "ambiguous":
        result["rewritten"] = step_back_rewrite(query)
 
    # 如果是组合问题，拆分子查询
    if analysis.complexity == "multi_hop":
        result["sub_queries"] = decompose_query(query)
 
    return result

# 混合检索：BM25 + Dense + RRF 融合 + Reranker
def hybrid_search(query: str, top_k: int = 20) -> list[Chunk]:
    # 两路并行检索
    dense_results = vector_db.search(query, top_k=top_k)
    bm25_results = bm25_index.search(query, top_k=top_k)
 
    # Reciprocal Rank Fusion 融合
    fused = reciprocal_rank_fusion(
        [dense_results, bm25_results],
        k=60
    )
 
    # 取 top-10 送 reranker 精排
    candidates = sorted(fused.items(), key=lambda x: -x[1])[:10]
    reranked = reranker.rerank(query, candidates)
 
    return reranked[:5]
 
 
def reciprocal_rank_fusion(
    rankings: list[list[str]],
    k: int = 60
) -> dict[str, float]:
    """融合 N 个排序列表为统一分数。"""
    scores: dict[str, float] = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank)
    return scores

# 带终止条件的 Agent 检索循环
class AgenticRetriever:
    def __init__(self, max_hops=3, min_confidence=0.8):
        self.max_hops = max_hops
        self.min_confidence = min_confidence
 
    def search(self, query: str) -> SearchResult:
        state = SearchState(query=query)
 
        for hop in range(self.max_hops):
            # 检索
            chunks = hybrid_search(state.current_query)
            state.add_evidence(chunks)
 
            # Agent 评估：信息够不够？
            assessment = self.agent.assess(state)
 
            if assessment.confidence >= self.min_confidence:
                return SearchResult(
                    answer=assessment.answer,
                    evidence=state.all_evidence,
                    hops=hop + 1,
                    confidence=assessment.confidence
                )
 
            if assessment.action == "rewrite_query":
                state.current_query = assessment.new_query
            elif assessment.action == "call_tool":
                tool_result = execute_tool(assessment.tool_call)
                state.add_evidence([tool_result])
 
        # 超过最大轮次，用已有信息尽力回答
        return SearchResult(
            answer=self.agent.force_answer(state),
            evidence=state.all_evidence,
            hops=self.max_hops,
            confidence=state.best_confidence,
            truncated=True  # 标记为未完全解决
        )

# 生成时强制引用，验证时检查引用合法性
def generate_with_citation(query: str, evidence: list[Chunk]) -> Answer:
    # 给每个 chunk 编号，要求模型在回答中标注引用
    numbered = "\n".join(
        f"[{i+1}] {c.text} (来源: {c.source}, 页码: {c.page})"
        for i, c in enumerate(evidence)
    )
 
    response = llm.generate(
        system=(
            "根据上下文回答。每个事实陈述后标注引用编号，"
            "格式：[1][2]。如果上下文无法支持你的回答，明确说明。"
        ),
        user=f"上下文：\n{numbered}\n\n问题：{query}"
    )
 
    # 验证引用：模型引用的编号必须真实存在于证据中
    cited_ids = extract_citations(response.text)
    valid_ids = set(range(1, len(evidence) + 1))
    invalid = [cid for cid in cited_ids if cid not in valid_ids]
 
    if invalid:
        # 引用有幻觉，触发重新生成或标记警告
        response.warning = f"无效引用编号: {invalid}"
 
    return response