Python+LangChain/LangGraph框架开发Ai智能体系列��21节 | LangGraph 生产级的Agentic RAG Agent示例

LLM 重写查询，更新 current_query

from pathlib import Pathimport sys as _sys_SCRIPT_DIR = Path(__file__).parent_DAY08_DIR = _SCRIPT_DIR.parent / "day08"_sys.path.insert(0, str(_SCRIPT_DIR))# ============================================================# Agent 状态定义（Agentic RAG 扩展版）# ============================================================from typing import Annotated, TypedDictfrom langchain_core.messages import BaseMessage, HumanMessage, AIMessagefrom operator import add as add_messagesclass AgentState(TypedDict):    """    LangGraph Agentic RAG Agent 的状态定义。    对比 Day8 的 AgentState：      Day8 只有 3 个字段：messages / rag_context / route      Day9 新增 4 个字段：retrieval_score / retry_count / original_query / current_query    新增字段说明：      retrieval_score：LLM 评估的检索质量评分（0-1）      retry_count：当前重试次数（用于防止无限循环）      original_query：用户原始查询（不会被修改，用于最终生成）      current_query：当前检索查询（可能被重写节点修改）    为什么需要 original_query 和 current_query 分离？      - 重写节点只修改 current_query      - 生成节点需要用 original_query 来回答用户的问题      - 如果只有一个 query，重写后用户原始问题就丢失了    """    messages: Annotated[list[BaseMessage], add_messages]  # 消息历史    rag_context: str | None           # RAG 检索到的上下文    route: str                        # 路由标签（rag / chat）    retrieval_score: float            # 检索质量评分（0-1）    retry_count: int                  # 当前重试次数    original_query: str               # 用户原始查询（不变）    current_query: str                # 当前检索查询（可能被重写）# ============================================================# 构建完整的 Agentic RAG Agent# ============================================================def _build_agentic_rag_agent(max_retries: int = 3, score_threshold: float = 0.6):    """    构建 LangGraph Agentic RAG Agent（含评估 + 重试循环边）。    架构：      ┌─ router ─→ retrieve ─→ evaluate ─→ generate ──→ END      │                              │      │                              ↓ 不够      └→ generate               rewrite ──→ retrieve（循环）                                 (retry_count++)    节点：      router：LLM 意图分类（rag / chat）      retrieve：向量检索（基于 current_query）      evaluate：LLM 评估检索质量（0-1 评分）      rewrite：LLM 重写查询（更新 current_query）      generate：LLM 基于上下文生成回答    条件边：      route_decision：router 的路由（rag → retrieve / chat → generate）      evaluate_decision：evaluate 后的走向（够 → generate / 不够 → rewrite）    循环边：      rewrite → retrieve（重试检索）      防护：retry_count >= max_retries 时强制走向 generate    参数：        max_retries：最大重试次数（默认 3）        score_threshold：评分阈值（默认 0.6）    返回：        compiled graph    """    from langgraph.graph import StateGraph, END    from langchain_ollama import ChatOllama, OllamaEmbeddings    import chromadb    import re    # ── 初始化 LLM ────────────────────────────────────────    llm = ChatOllama(model="qwen3:4b", temperature=0, num_ctx=8192)    persist_dir = str(_DAY08_DIR / "chroma_db")    # ── 节点函数 ────────────────────────────────────────────    def router_node(state: AgentState) -> dict:        """        路由节点：LLM 判断用户意图，决定走 rag 还是 chat。        对比 Day8 的 router：          Day8 有三个分支：rag / tool / chat          Day9 简化为两个：rag / chat（Agentic RAG 的核心是检索决策）        判断标准：          - rag：问题涉及技术概念，需要检索知识库          - chat：闲聊/问候/开放性问题，LLM 直接回答        """        msgs = state["messages"]        if msgs and isinstance(msgs[-1], HumanMessage):            query = msgs[-1].content        elif msgs and hasattr(msgs[-1], 'content'):            query = msgs[-1].content        else:            query = str(msgs[-1]) if msgs else ""        router_prompt = f"""你是一个意图分类器，根据用户问题判断下一步处理方式。用户问题：{query}请判断应该走哪个分支（只输出一个词：rag / chat）：- rag：问题涉及 LangGraph、MCP、RAG、LangSmith、嵌入模型、向量数据库、Agent、智能体等技术概念- chat：问题不需要检索知识库，LLM 可以直接回答（如问候、闲聊、开放性问题）只输出一个词：rag / chat"""        response = llm.invoke(router_prompt)        route = response.content.strip().lower()        if route not in ("rag", "chat"):            route = "chat"        # 初始化 original_query 和 current_query        return {            "route": route,            "original_query": query,            "current_query": query,            "retry_count": 0,        }    def retrieve_node(state: AgentState) -> dict:        """        检索节点：基于 current_query 执行向量检索。        注意：          - 使用 current_query（可能被重写节点修改过）而非 original_query          - 每次检索都重新实例化 OllamaEmbeddings（Day8 踩坑经验）          - 检索结果存入 rag_context 供后续节点使用        """        query = state.get("current_query", "")        try:            client = chromadb.PersistentClient(path=persist_dir)            collection = client.get_collection(name="day08_rag")            embeddings = OllamaEmbeddings(model="nomic-embed-text")            query_vector = embeddings.embed_query(query)            results = collection.query(                query_embeddings=[query_vector],                n_results=3,                include=["documents", "metadatas", "distances"],            )            parts = []            raw_docs = results.get("documents", [[]])[0]            raw_metas = (results.get("metadatas") or [[]])[0]            for i, content in enumerate(raw_docs):                src = raw_metas[i].get("source", "?") if i < len(raw_metas) else "?"                parts.append(f"[来源{i+1}（{src}）]\n{content}")            rag_context = "\n\n".join(parts)        except Exception as e:            rag_context = f"[检索失败：{e}]"        return {"rag_context": rag_context}    def evaluate_node(state: AgentState) -> dict:        """        评估节点：LLM 评估检索质量，返回 0-1 评分。        这是 Agentic RAG 的灵魂节点：          - 传统 RAG（Day8）没有这一步，检索完直接生成          - Agentic RAG 先评估"信息够不够"，再决定下一步        评分标准：          - 0.8-1.0：检索结果完全能回答问题          - 0.6-0.8：部分相关但不够充分          - 0.0-0.6：基本无关        决策：          - score >= threshold → 走 generate（信息充分）          - score < threshold → 走 rewrite（信息不足，需要重写查询重试）        """        rag_ctx = state.get("rag_context") or ""        original_query = state.get("original_query", "")        eval_prompt = f"""你是一个检索质量评估器。根据用户问题和检索到的上下文，评估检索结果的相关度。【用户问题】{original_query}【检索到的上下文】{rag_ctx[:500]}请评估检索结果对回答用户问题的帮助程度，只输出一个 0 到 1 之间的数字：- 0.8-1.0：检索结果完全能回答问题- 0.6-0.8：部分相关但不够充分- 0.4-0.6：有一定关联但缺少关键信息- 0.0-0.4：基本无关只输出一个数字，不要其他内容："""        try:            response = llm.invoke(eval_prompt)            score_text = response.content.strip()            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))        except Exception:            score = 0.5        retry_count = state.get("retry_count", 0)        print(f"    [评估] score={score:.2f} | retry={retry_count}/{max_retries}")        return {            "retrieval_score": score,            "retry_count": retry_count,  # 不在这里加，在 rewrite 节点加        }    def rewrite_node(state: AgentState) -> dict:        """        重写节点：LLM 重写查询，更新 current_query。        触发条件：evaluate_node 评分 < score_threshold        重写策略：          1. 添加更具体的关键词          2. 换一种表述方式          3. 拆分为更聚焦的子问题        注意：          - 只修改 current_query，不修改 original_query          - retry_count + 1（用于防止无限循环）          - 如果重写失败，保持原查询并增加重试计数        """        original_query = state.get("original_query", "")        current_query = state.get("current_query", "")        eval_score = state.get("retrieval_score", 0.0)        retry_count = state.get("retry_count", 0)        rewrite_prompt = f"""你是一个查询优化专家。用户提出了一个问题，但当前检索结果不够相关。【原始问题】{original_query}【当前检索查询】{current_query}【检索结果评分】{eval_score:.2f}（满分 1.0，低于 {score_threshold} 表示信息不充分）请重写检索查询，使其更精确、更有针对性地找到相关信息。重写策略：  1. 添加更具体的关键词  2. 换一种表述方式  3. 拆分为更聚焦的子问题只输出重写后的查询，不要其他内容："""        try:            response = llm.invoke(rewrite_prompt)            new_query = response.content.strip().strip('"\'""''')            if len(new_query) < 3:                new_query = current_query            print(f"    [重写] \"{current_query[:40]}\" -> \"{new_query[:40]}\"")        except Exception as e:            new_query = current_query            print(f"    [重写] 失败：{e}，保持原查询")        return {            "current_query": new_query,            "retry_count": retry_count + 1,  # 在这里增加重试计数        }    def generate_node(state: AgentState) -> dict:        """        生成节点：LLM 基于上下文生成最终回答。        这是终止节点：          - rag 路由：基于检索到的上下文回答          - chat 路由：直接回答        注意：          - 使用 original_query（用户原始问题）而非 current_query          - 如果有 rag_context，基于上下文回答          - 如果没有，说明是 chat 路由或检索失败        """        rag_ctx = state.get("rag_context") or ""        route = state.get("route", "chat")        original_query = state.get("original_query", "")        # 构造 prompt        if route == "rag" and rag_ctx:            prompt = f"""你是一个有帮助的 AI 助手，基于提供的上下文回答问题。【上下文】{rag_ctx}【问题】{original_query}请简洁、准确地回答。如果上下文没有相关信息，请说明"当前上下文没有相关信息"。回答时引用来源编号。"""        else:            prompt = f"""你是一个有帮助的 AI 助手。直接简洁地回答用户问题。【问题】{original_query}"""        response = llm.invoke(prompt)        return {"messages": [AIMessage(content=response.content)]}    # ── 构建图 ──────────────────────────────────────────────    graph = StateGraph(AgentState)    # 添加节点    graph.add_node("router", router_node)    graph.add_node("retrieve", retrieve_node)    graph.add_node("evaluate", evaluate_node)    graph.add_node("rewrite", rewrite_node)    graph.add_node("generate", generate_node)    # 入口：router    graph.set_entry_point("router")    # 条件边 1：router 的路由决策    def route_decision(state: AgentState) -> str:        return state.get("route", "chat")    graph.add_conditional_edges(        "router",        route_decision,        {            "rag": "retrieve",   # rag → 检索            "chat": "generate",  # chat → 直接生成        }    )    # 固定边：retrieve → evaluate    graph.add_edge("retrieve", "evaluate")    # 条件边 2：evaluate 后的走向（Agentic RAG 的核心！）    def evaluate_decision(state: AgentState) -> str:        """        评估后的条件决策：        判断逻辑：          1. score >= threshold → "generate"（信息充分，直接生成）          2. score < threshold 且 retry < max_retries → "rewrite"（重写查询重试）          3. score < threshold 且 retry >= max_retries → "generate"（强制生成，防止无限循环）        这是 Agentic RAG 区别于传统 RAG 的关键：          传统 RAG 没有这个决策点，检索完直接生成。          Agentic RAG 在这里决定是继续检索还是生成回答。        """        score = state.get("retrieval_score", 0.0)        retry = state.get("retry_count", 0)        if score >= score_threshold:            print(f"    [决策] 评分 {score:.2f} >= {score_threshold}，信息充分 -> generate")            return "generate"        elif retry < max_retries:            print(f"    [决策] 评分 {score:.2f} < {score_threshold}，重试 {retry}/{max_retries} -> rewrite")            return "rewrite"        else:            print(f"    [决策] 评分 {score:.2f} < {score_threshold}，已达最大重试 -> generate（强制）")            return "generate"    graph.add_conditional_edges(        "evaluate",        evaluate_decision,        {            "generate": "generate",  # 信息充分 → 生成            "rewrite": "rewrite",    # 信息不足 → 重写        }    )    # 循环边：rewrite → retrieve（重试检索）    graph.add_edge("rewrite", "retrieve")    # 终止边：generate → END    graph.add_edge("generate", END)    return graph.compile()# ============================================================# 演示函数# ============================================================def demo_agentic_rag():    """    完整演示 LangGraph Agentic RAG Agent。    测试用例：      1. 精确查询："LangGraph 的核心概念是什么？"         → rag 路由 → 检索 → 评估 → 信息充分 → 直接生成      2. 模糊查询："怎么让 AI 自己决定要不要查资料？"         → rag 路由 → 检索 → 评估 → 信息不足 → 重写 → 再检索 → 循环      3. 闲聊："你好，介绍一下你自己"         → chat 路由 → 直接生成（不检索）    """    import time    print("\n" + "=" * 60)    print("Day9-03 · LangGraph Agentic RAG Agent 演示")    print("=" * 60)    # 构建 Agent    print("\n[构建] LangGraph Agentic RAG Agent...")    try:        agent = _build_agentic_rag_agent(max_retries=3, score_threshold=0.6)        print("[OK] Agent 编译成功")    except Exception as e:        print(f"[错误] Agent 构建失败：{e}")        return    # 测试用例    test_cases = [        {            "query": "LangGraph 的核心概念是什么？",            "desc": "精确技术查询 -> rag路由 -> 评估通过 -> 直接生成",        },        {            "query": "怎么让 AI 自己决定要不要查资料？",            "desc": "模糊查询 -> rag路由 -> 评估不足 -> 重写查询 -> 重试",        },        {            "query": "你好，请介绍一下你自己",            "desc": "闲聊问候 -> chat路由 -> 直接生成",        },    ]    for i, case in enumerate(test_cases, 1):        print(f"\n{'=' * 60}")        print(f"测试 {i}/{len(test_cases)}：{case['desc']}")        print(f"用户：{case['query']}")        print("-" * 40)        start = time.time()        try:            result = agent.invoke(                {                    "messages": [HumanMessage(content=case["query"])],                    "rag_context": None,                    "route": "",                    "retrieval_score": 0.0,                    "retry_count": 0,                    "original_query": "",                    "current_query": "",                }            )            elapsed = time.time() - start            msgs = result["messages"]            route = result.get("route", "?")            score = result.get("retrieval_score", 0.0)            retries = result.get("retry_count", 0)            original = result.get("original_query", "")            current = result.get("current_query", "")            final_msg = msgs[-1] if msgs else None            answer = final_msg.content if hasattr(final_msg, 'content') else str(final_msg)            print(f"\n  [结果]")            print(f"  路由：{route}")            if route == "rag":                print(f"  评估评分：{score:.2f}")                print(f"  重试次数：{retries}")                if original != current:                    print(f"  查询重写：\"{original[:30]}\" -> \"{current[:30]}\"")            print(f"  AI：{answer.strip()[:200]}")            print(f"  耗时：{elapsed:.2f}s")        except Exception as e:            elapsed = time.time() - start            print(f"  [执行失败] {e}")            print(f"  如遇 Ollama 连接问题，请确认：ollama list")    print("\n" + "=" * 60)    print("Day9-03 演示完成！")    print("=" * 60)    print("\nDay9 关键收获：")    print("  1. Agentic RAG = 传统 RAG + 评估 + 重试循环")    print("  2. 条件循环边是 LangGraph 实现 Agentic RAG 的核心语法")    print("  3. max_retries 防止无限循环，score_threshold 控制质量门槛")    print("  4. original_query / current_query 分离，重写不丢失原始问题")    print("  5. 面试回答：Agentic RAG 把检索作为 Agent 的工具之一，")    print("     能自主决定何时检索、用什么策略检索、检索结果是否充分")if __name__ == "__main__":    demo_agentic_rag()