Agentic RAG工程实战

from typing import TypedDict, List, Optional, Annotated
from langchain_core.documents import Document
import operator


class GraphState(TypedDict):
    """
    Corrective RAG 的全局状态

    所有节点通过读写这个 State 来通信，
    不要用全局变量或者把数据存到节点外面。
    """
    # 输入
    question: str            # 用户原始问题（不变）

    # 查询改写
    query: str               # 改写后的查询（用于检索）

    # 检索结果
    documents: List[Document]     # 向量库检索到的文档
    web_results: List[Document]   # Web Search 的结果
    final_documents: List[Document]  # 合并后的最终文档集

    # 评分
    relevance_score: float   # 检索质量综合得分（0-1）
    graded_documents: List[dict]  # 每个文档的详细评分

    # 生成结果
    generation: str          # 当前生成的答案

    # 质量检测
    hallucination_check: str  # "supported" 或 "hallucination"
    answer_quality: str       # "useful" 或 "not_useful"

    # 控制变量
    iterations: int           # 当前循环次数
    web_search_triggered: bool  # 是否触发了 Web Search（用于日志）

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from config import LLM_MODEL, LLM_TEMPERATURE

llm = ChatOpenAI(model=LLM_MODEL, temperature=LLM_TEMPERATURE)

REWRITE_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是一个查询优化专家，负责将用户的自然语言问题改写为更适合向量检索的查询。

    改写规则：
    1. 保留问题的核心意图，不要改变语义
    2. 展开缩写和口语表达（"怎么搞" → "如何实现"）
    3. 添加相关的专业术语（如果能判断领域的话）
    4. 如果问题包含多个子问题，只保留最核心的那个
    5. 输出单一的改写后查询，不要解释

    直接输出改写后的查询，不要任何前缀。"""),
    ("human", "原始问题：{question}")
])

rewrite_chain = REWRITE_PROMPT | llm | StrOutputParser()


def rewrite_query(state: GraphState) -> dict:
    """查询改写节点"""
    question = state["question"]

    # 首次进入或迭代重试时都改写
    if state.get("iterations", 0) == 0:
        # 首次：基于原始问题改写
        rewritten = rewrite_chain.invoke({"question": question})
    else:
        # 重试：基于原始问题 + 上次答案失败的信息改写
        rewritten = rewrite_chain.invoke({
            "question": f"""原始问题：{question}
            上次生成的答案没有很好地回答这个问题，请换一个角度改写查询。"""
        })

    return {
        "query": rewritten.strip(),
        "iterations": state.get("iterations", 0) + 1
    }

import weaviate
from langchain_weaviate import WeaviateVectorStore
from langchain_openai import OpenAIEmbeddings
from config import WEAVIATE_URL, WEAVIATE_API_KEY, WEAVIATE_COLLECTION

# 初始化 Weaviate 客户端（复用连接，不要在每次请求时重新连接）
weaviate_client = weaviate.connect_to_local(
    host="localhost",
    port=8080,
    grpc_port=50051,
)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = WeaviateVectorStore(
    client=weaviate_client,
    index_name=WEAVIATE_COLLECTION,
    text_key="content",
    embedding=embeddings,
)

# 使用混合检索（向量 + BM25），比纯向量检索效果更好
retriever = vectorstore.as_retriever(
    search_type="hybrid",  # Weaviate 支持混合搜索
    search_kwargs={
        "k": 5,            # 返回 top-5 文档
        "alpha": 0.7,      # 0=纯BM25, 1=纯向量, 0.7偏向语义
    }
)


def retrieve(state: GraphState) -> dict:
    """向量检索节点"""
    query = state["query"]
    documents = retriever.invoke(query)

    return {"documents": documents}

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
from typing import List
from config import LLM_MODEL, RELEVANCE_THRESHOLD

llm = ChatOpenAI(model=LLM_MODEL, temperature=0)

GRADER_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是一个专业的检索质量评估员。

    评估检索到的文档片段是否能帮助回答用户问题。

    返回 JSON 格式，不要包含其他内容：
    {{
        "score": 0.0到1.0之间的数值,
        "reason": "评分理由，一句话"
    }}

    评分标准：
    - 1.0：文档直接包含问题的精确答案
    - 0.8：文档高度相关，包含回答问题所需的核心信息
    - 0.6：文档相关，包含部分有用信息
    - 0.4：文档略相关，有少量参考价值
    - 0.2以下：文档基本不相关，对回答没有帮助

    严格评分，不要过于宽松。"""),
    ("human", "用户问题：{question}\n\n文档内容：\n{document}")
])

grader_chain = GRADER_PROMPT | llm | JsonOutputParser()


def grade_documents(state: GraphState) -> dict:
    """文档评分节点"""
    question = state["question"]
    documents = state["documents"]

    graded_docs = []
    scores = []

    for doc in documents:
        try:
            result = grader_chain.invoke({
                "question": question,
                "document": doc.page_content[:2000]  # 限制长度，避免超出上下文
            })
            score = float(result.get("score", 0))
            graded_docs.append({
                "document": doc,
                "score": score,
                "reason": result.get("reason", "")
            })
            scores.append(score)
        except Exception:
            # 评分失败的文档给低分，不要崩溃整个流程
            graded_docs.append({
                "document": doc,
                "score": 0.3,
                "reason": "评分失败，给予默认低分"
            })
            scores.append(0.3)

    # 综合分：取评分最高的几个文档的平均值
    top_scores = sorted(scores, reverse=True)[:3]
    avg_score = sum(top_scores) / len(top_scores) if top_scores else 0

    # 过滤：只保留分数高于 0.5 的文档
    filtered_docs = [
        item["document"]
        for item in graded_docs
        if item["score"] >= 0.5
    ]

    return {
        "graded_documents": graded_docs,
        "relevance_score": avg_score,
        "final_documents": filtered_docs  # 初始设为过滤后的文档
    }

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.documents import Document
from config import TAVILY_API_KEY
import os

os.environ["TAVILY_API_KEY"] = TAVILY_API_KEY

web_search_tool = TavilySearchResults(
    max_results=3,
    search_depth="advanced",
    include_answer=True,
    include_raw_content=False,
    include_images=False,
)


def web_search(state: GraphState) -> dict:
    """Web Search 节点：当本地检索质量不足时触发"""
    query = state["query"]

    try:
        results = web_search_tool.invoke({"query": query})

        web_docs = []
        for r in results:
            if isinstance(r, dict) and "content" in r:
                web_docs.append(Document(
                    page_content=r["content"],
                    metadata={
                        "source": r.get("url", "web_search"),
                        "type": "web_search",
                        "title": r.get("title", ""),
                    }
                ))

        return {
            "web_results": web_docs,
            "web_search_triggered": True,
        }
    except Exception as e:
        # Web Search 失败不应该让整个流程崩溃
        return {
            "web_results": [],
            "web_search_triggered": True,
        }


def merge_docs(state: GraphState) -> dict:
 
 
  
PRO 会员专属
 
 本文为 PRO 会员专属内容，成为会员即可阅读全文。 
 
 登录 了解会员权益 
 
 PRO ¥199/年 · Pro 专属文章 + 2300+ 知识文档 + 会员社群 
 
 
 
 
 
本文作者：Elazer (石头)
 
原文链接：https://ss-data.cc/posts/kb-agentic-rag
 
版权声明：本文采用 CC BY-NC-SA 4.0 许可协议，转载请注明出处。
 
 
  
 Elazer (石头) 
11 年数据老兵，从分析师到架构专家。用真实经历帮数据人少走弯路。
 
 关于作者 加入社区 了解会员 
 
 
 
 
 
加入免费社群
 
和数据从业者一起交流成长
 了解详情 → 
 
 
成为会员
 
解锁全部内容 + 知识库
 查看权益 → 
 
 
  ← 上一篇 LLM评估体系   下一篇 → 数据分析师用实践驱动学习法，3周内独立完成销售分析报告  
 
 
相关文章
 
  
    
 
 
 AI · 方法 
 
LLM评估体系
 
传统 NLP 评估指标与人类判断之间的相关性，在 LLM 时代几乎崩塌了。BLEU 高不代表答案好，ROUGE 低不代表答案差。LLM 评估需要一套完全不同的体系。。#为什么 LLM 评估很难。#第一层：自动化评估（RAGAS）。#第二层：LLM-as-Judge。#评估驱动的开发工作流。1. 开放域输出，没有唯...
 
  
  PRO  
 
 
 AI · 方法 
 
向量数据库与RAG工程实践 - 让AI真正懂你的数据
 
传统搜索用关键词匹配——"苹果手机"搜不到"iPhone"的结果，因为字面上不相似。Embedding 技术的突破在于：把语义相似的内容映射到向量空间中相近的位置。。假设知识库有 100 万条文档，每条 Embedding 是 768 维的 float32：。存储：768 × 4 bytes × 1,000,00...
 
  
  PRO  
 
 
 AI · 方法 
 
AI多Agent协作系统
 
Agent概述 - 单Agent基础知识。Agent框架 - 框架详细使用。提示工程 - Agent Prompt设计。大语言模型 - Agent的核心引擎