Python+LangChain/LangGraph框架开发Ai智能体系列��19节 | Agentic RAG 基础 | 从＂检索一次＂到＂决策循环＂

def _load_collection(persist_directory: str = None, collection_name: str = "day08_rag"):    """    从本地持久化目录加载 Chroma Collection。    复用之前文章已建好的索引数据，避免重复创建。    默认指向 day08/chroma_db 目录。    参数：        persist_directory：Chroma 数据库目录（默认 day08/chroma_db）        collection_name：Collection 名称（默认 day08_rag）    返回：        chromadb.Collection 对象（含 .query() 方法）    """    import chromadb    if persist_directory is None:        persist_directory = str(_DAY08_DIR / "chroma_db")    client = chromadb.PersistentClient(path=persist_directory)    collection = client.get_collection(name=collection_name)    print(f"  [OK] Collection '{collection_name}' 加载成功，共 {collection.count()} 条记录")    return collection

def _vector_search(collection, query: str, top_k: int = 3) -> list[dict]:    """    执行一次向量检索，返回原始结果。    封装 Chroma 的 query 调用，统一处理嵌入和查询逻辑。    参数：        collection：Chroma Collection 对象        query：查询字符串        top_k：返回的文档数量    返回：        list[dict]，每个 dict 包含 content / metadata / distance    """    from langchain_ollama import OllamaEmbeddings    embeddings = OllamaEmbeddings(model="nomic-embed-text")    query_vector = embeddings.embed_query(query)    results = collection.query(        query_embeddings=[query_vector],        n_results=top_k,        include=["documents", "metadatas", "distances"],    )    docs = []    raw_docs = results.get("documents", [[]])[0]    raw_metas = (results.get("metadatas") or [[]])[0]    raw_dists = (results.get("distances") or [[]])[0]    for i, content in enumerate(raw_docs):        meta = raw_metas[i] if i < len(raw_metas) else {}        distance = raw_dists[i] if i < len(raw_dists) else None        docs.append({            "content": content,            "metadata": meta,            "distance": distance,        })    return docs

def _call_llm(prompt: str, model: str = "qwen3:4b") -> str:    """    调用 Ollama LLM 生成回复。    封装 ChatOllama 调用，统一错误处理。    参数：        prompt：输入提示词        model：Ollama 模型名称    返回：        LLM 生成的文本字符串    """    from langchain_ollama import ChatOllama    llm = ChatOllama(model=model, temperature=0, num_ctx=4096)    response = llm.invoke(prompt)    return response.content.strip()

def _simple_rag_query(collection, query: str, top_k: int = 3) -> dict:    """    传统 RAG 实现：检索一次 → 直接生成回答。    这是 Day8 学过的模式：      1. 将 query 嵌入为向量      2. Chroma 检索 Top-K 文档      3. 拼接上下文 → LLM 生成    局限性：      - 如果查询模糊，检索结果可能不相关      - 没有评估机制，无法判断检索质量      - 只检索一次，无法自我修正    参数：        collection：Chroma Collection        query：用户查询        top_k：检索文档数量    返回：        dict：{"answer": str, "context": list[dict], "retrieval_count": int}    """    # 步骤 1：检索    docs = _vector_search(collection, query, top_k)    # 步骤 2：拼接上下文    context_parts = []    for i, doc in enumerate(docs):        src = doc["metadata"].get("source", "?")        context_parts.append(f"[来源{i+1}（{src}）]\n{doc['content']}")    context_str = "\n\n".join(context_parts)    # 步骤 3：LLM 生成    prompt = f"""你是一个有帮助的 AI 助手，基于提供的上下文回答问题。【上下文】{context_str}【问题】{query}请简洁、准确地回答。如果上下文无法回答，请说明"当前上下文没有相关信息"。"""    answer = _call_llm(prompt)    return {        "answer": answer,        "context": docs,        "retrieval_count": 1,  # 传统 RAG 只检索一次    }

def _agentic_rag_query(    collection,    query: str,    top_k: int = 3,    max_retries: int = 3,    score_threshold: float = 0.6,) -> dict:    """    Agentic RAG 实现：检索 → 评估 → 重写 → 再检索的决策循环。    与传统 RAG 的核心区别：      1. 检索后，LLM 评估结果质量（0-1 评分）      2. 评分不足时，LLM 重写查询，重新检索      3. 最多重试 max_retries 轮，防止无限循环      4. 记录每次检索的评分和查询，便于调试    参数：        collection：Chroma Collection        query：用户查询        top_k：每次检索的文档数量        max_retries：最大重试次数        score_threshold：评分阈值（低于此值触发重试）    返回：        dict：{            "answer": str,            "context": list[dict],            "retrieval_count": int,            "history": list[dict],  # 每轮检索的评分和查询记录        }    """    current_query = query    history = []  # 记录每轮检索的评分和查询    best_docs = []  # 保留最好的检索结果    for attempt in range(max_retries + 1):  # 初始 1 次 + 最多 max_retries 次重试        # ── 步骤 1：检索 ──────────────────────────────────        docs = _vector_search(collection, current_query, top_k)        # ── 步骤 2：评估检索质量 ────────────────────────────        context_str = "\n".join(            f"- {doc['content'][:100]}" for doc in docs        )        eval_prompt = f"""你是一个检索质量评估器。根据用户问题和检索到的上下文，评估检索结果的相关度。【用户问题】{query}【检索到的上下文】{context_str}请评估检索结果对回答用户问题的帮助程度，只输出一个 0 到 1 之间的数字：- 0.8-1.0：检索结果完全能回答问题- 0.6-0.8：检索结果部分相关，但信息不够充分- 0.4-0.6：检索结果有一定关联，但缺少关键信息- 0.0-0.4：检索结果与问题基本无关只输出一个数字，不要其他内容："""        try:            score_text = _call_llm(eval_prompt)            # 提取数字：取字符串中第一个浮点数            import re            match = re.search(r'(\d+\.?\d*)', score_text)            score = float(match.group(1)) if match else 0.5            score = max(0.0, min(1.0, score))  # 钳位到 [0, 1]        except Exception:            score = 0.5  # 评估失败时默认中等评分        history.append({            "attempt": attempt + 1,            "query": current_query,            "score": score,            "doc_count": len(docs),        })        # 保留评分最高的结果        if not best_docs or score > max(h["score"] for h in history[:-1]):            best_docs = docs        # ── 步骤 3：判断是否需要重试 ────────────────────────        if score >= score_threshold:            print(f"    [评估] 第 {attempt+1} 轮评分 {score:.2f} >= {score_threshold}，信息充分")            break        print(f"    [评估] 第 {attempt+1} 轮评分 {score:.2f} < {score_threshold}，信息不充分")        # 已达最大重试次数，强制退出        if attempt >= max_retries - 1:            print(f"    [评估] 已达最大重试次数 {max_retries}，强制生成")            break        # ── 步骤 4：重写查询 ────────────────────────────────        rewrite_prompt = f"""你是一个查询优化专家。用户提出了一个问题，但当前检索结果不够相关。【原始问题】{query}【当前检索查询】{current_query}【检索结果评分】{score:.2f}（满分 1.0）请重写检索查询，使其更精确、更有针对性地找到相关信息。重写策略：  1. 添加更具体的关键词  2. 换一种表述方式  3. 拆分为更聚焦的子问题只输出重写后的查询，不要其他内容："""        try:            new_query = _call_llm(rewrite_prompt)            if new_query and len(new_query) > 2:                current_query = new_query                print(f"    [重写] 新查询：{current_query[:60]}...")            else:                print(f"    [重写] 重写失败，保持原查询")                break  # 重写失败，不再重试        except Exception:            print(f"    [重写] 重写异常，保持原查询")            break    # ── 步骤 5：生成最终回答 ──────────────────────────────────    context_parts = []    for i, doc in enumerate(best_docs):        src = doc["metadata"].get("source", "?")        context_parts.append(f"[来源{i+1}（{src}）]\n{doc['content']}")    context_str = "\n\n".join(context_parts)    generate_prompt = f"""你是一个有帮助的 AI 助手，基于提供的上下文回答问题。【上下文】{context_str}【问题】{query}请简洁、准确地回答。如果上下文无法回答，请说明"当前上下文没有相关信息"。"""    answer = _call_llm(generate_prompt)    return {        "answer": answer,        "context": best_docs,        "retrieval_count": len(history),        "history": history,    }

def_load_collection(persist_directory: str = None, collection_name: str = "day08_rag"):    """    从本地持久化目录加载 Chroma Collection。    复用之前文章已建好的索引数据，避免重复创建。    默认指向 day08/chroma_db 目录。    参数：        persist_directory：Chroma 数据库目录（默认 day08/chroma_db）        collection_name：Collection 名称（默认 day08_rag）    返回：        chromadb.Collection 对象（含 .query() 方法）    """    import chromadb    if persist_directory isNone:        persist_directory = str(_DAY08_DIR / "chroma_db")    client = chromadb.PersistentClient(path=persist_directory)    collection = client.get_collection(name=collection_name)    print(f"  [OK] Collection '{collection_name}' 加载成功，共 {collection.count()} 条记录")    return collection