LangGraph中的智能RAG：评分器、条件路由与网页搜索备用方案

该语义RAG流程由LangChain的核心组件、FastAPI后端以及Streamlit聊天界面前端构成，能够处理PDF上传，将文档转换为嵌入向量，并通过聊天界面回答问题。但它也存在一定的局限性：LangChain的原始组件仅能实现线性流程，即从提示词经LLM直接输出到解析器。任何需要条件分支、循环、动态路由或由LLM自主决策的场景都超出了这种结构，而简单的if/else语句也只能实现浅层、预定义的流程。

LangGraph解决了这一问题。由此形成的代理式工作流具备以下特点：

• 流程中直接内置条件分支和循环结构
• LLM可自主选择网页搜索或RAG检索方式，无需硬编码规则
• 流程转换通过LangGraph的节点和边来管理

基于LLM的评分器，用于评估检索到的上下文及生成的答案

根据这些评分结果进行路由处理

通过StateGraph在状态转换之间保持状态一致性

将最终构建的图结构接入FastAPI后端，以支撑生产环境级别的聊天机器人

最终得到的就是一个能够从向量存储或网络中获取信息并作出响应的代理式RAG系统。

LangGraph术语

该系统的词汇概念主要由四部分构成：StateGraph负责协调执行流程；State是一种带类型的字典，用于存储和更新处理流程中的数据；Nodes则是能够改变状态并作为图结构中计算单元的Python函数；Edges则表示节点之间的连接，而条件边则支持分支和循环结构。

配置

请将API密钥放入项目根目录下的.env文件中：

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
Hugging_Face_TOKEN=<your huggingface token>
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>

然后在 Pydantic 的 BaseSettings 类中读取它们：

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

    """Loads the dotenv file. Including this is necessary to get
    pydantic to load a .env file."""
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

项目结构

文件夹结构如下：

project/
├── rag_process/
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── graphs
│   │── chains.py       # llm chains
│   ├── nodes.py        # graph nodes. Defined as Python functions
│   ├── graph.py        # graph workflow
│   ├── consts.py       # define node names as constants
│   └── state.py        # typed dictionary to store graph states
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── dependencies.py     # dependency inject to fastAPI endpoint function
├── test_chains.py      # pytests
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

检索链、FastAPI服务以及Streamlit客户端均来自之前的流程。新增的部分包括图结构、其状态类，以及用于检查检索到的文档相关性、识别幻觉内容并判断答案是否与问题相关的评估器。

LLM链

这些链位于graphs/chains.py中，使用LangChain表达语言（LCEL）实现。共定义了七个链：

• retrieve_documents从Pinecone中获取Top-K个相似文档
• generate_answer根据问题及上下文生成答案
• retrival_grader判断检索到的文档是否具有相关性
• web_search执行网络搜索并返回相关上下文
• answer_grader检查答案是否回应了问题

hallucination_grader 用于检查答案是否基于上下文。

select_routes 会将关于智能体、提示工程及对抗性攻击的问题发送给 RAG，其他问题则通过网络搜索处理。

链式操作的单元测试

在智能体系统中，测试比平常更为重要，因为大语言模型的输出具有非确定性，且会随着模型更新、供应商调整参数或嵌入向量变化而出现偏差。test_chains.py 中的测试使用 pytest 并结合 pytest.mark.asyncio 来处理异步链式操作。RAG 检索的通过测试示例如下：

@pytest.mark.asyncio
async def test_rag_retrieval_yes() -> None:
    prompt = "what is prompt engineer ?"
    searched_documents = await retrieve_documents(prompt)
    searched_document = searched_documents[0]
    # logger.info(f"searched_content: {searched_document.page_content}")
    retrieval_grade = await retrieval_grader(searched_document, prompt)
    assert retrieval_grade

它涉及提示工程相关内容，并将检索到的第一份文档传递给retrieval_grader。由于该主题已有文档在Pinecone中覆盖，评分器应返回True。该测试能够防止嵌入向量或模型行为发生变化时出现故障，及早发现模型更新带来的退化问题，并验证RAG与网络搜索之间的路由逻辑。相应的负向测试可确认评分器返回False，从而让流程转向网络搜索。

状态与节点

状态是一个存储在graphs/state.py中的带类型定义的字典：

class GraphState(TypedDict):

    question: str
    generation: str
    web_search: bool
    documents: list[str]

节点负责读取和更新此状态。这些节点定义在 graphs/nodes.py 中，是普通的 Python 函数，用于读取状态、调用大语言模型链并返回更新后的字段。在条件边中使用的节点会返回下一个要执行的节点名称，总共有七个此类节点。

• generate 函数会调用 generate_answer 并返回答案。
• retrieve 函数会调用 retrieve_documents 并返回文档内容。
• search_web 函数会调用 web_search，并将搜索结果添加到文档中。
• grade_documents 函数会筛选出相关文档，并在需要时将 web_search 设置为 True。
• grade_generation 函数会运行幻觉检测和答案质量评估功能，然后返回“有用”、“无用”或“不支持”等结果。

decide_to_generate会根据web_search的值选择WEBSEARCH或GENERATE。

route_question会将与智能体相关的问题发送到RAG，其他问题则发送到网络搜索。

构建StateGraph

该图结构是在graphs/graph.py中生成的，最终的工作流程表现为检索、评分、生成和搜索之间的分支结构。构建它需要几个步骤。

首先，为每个节点分配一个字符串标识符。将这些名称作为常量保存在graphs/consts.py中，既能提高图的可读性，又能避免拼写错误：

RETRIEVE = "retrieve"
GRADE_DOCUMENTS = "grade_documents"
GENERATE = "generate"
WEBSEARCH = "websearch"

接下来，创建 StateGraph 并为每个节点注册其对应的函数：

workflow = StateGraph(GraphState)

workflow.add_node(RETRIEVE, retrieve)
workflow.add_node(GRADE_DOCUMENTS, grade_documents)
workflow.add_node(GENERATE, generate)
workflow.add_node(WEBSEARCH, search_web)

每次调用 add_node 都会将一个名称与某个函数关联起来，这种名称与功能的分离使得工作流程更易于维护、可视化及扩展。

第三，设置条件式的入口点。LangGraph 不要求硬编码起始节点，而是允许函数对问题进行判断并选择第一条处理路径。此处 route_question 负责决定是从网络搜索还是 RAG 检索开始处理：

workflow.set_conditional_entry_point(
    route_question,
    {
        WEBSEARCH: WEBSEARCH,
        RETRIEVE: RETRIEVE,
    }
)

该函数会首先被执行，根据主题返回 "WEBSEARCH" 或 "RETRIEVE"，LangGraph 会利用其返回值来选择起始节点。与智能体相关的问题会交由 RAG 处理，而一般性知识则从网络搜索开始。

第四，用边连接节点。静态边始终遵循固定路径，而条件边则根据大语言模型的决策来选择下一个节点：

workflow.add_edge(RETRIEVE, GRADE_DOCUMENTS)
workflow.add_conditional_edges(
    GRADE_DOCUMENTS,
    decide_to_generate,
    {
        WEBSEARCH: WEBSEARCH,
        GENERATE: GENERATE,
    },
)

workflow.add_conditional_edges(
    GENERATE,
    grade_generation,
    {
        "not supported": GENERATE,
        "useful": END,
        "not useful": WEBSEARCH,
    }
)

workflow.add_edge(WEBSEARCH, GENERATE)
workflow.add_edge(GENERATE, END)

这些环节共同构成了一种动态工作流程：大语言模型负责选择初始路径，文档相关性决定是采用RAG还是网络搜索，而答案质量则决定是进一步优化、回退还是结束流程。这样的结构将普通的RAG流程转变为更为强大的智能系统。

最后，编译该图结构，并可选择将其可视化：

app = workflow.compile()

app.get_graph().draw_mermaid_png(output_file_path="graphs.png")

通过 FastAPI 提供服务

在 dependencies.py 中，一个函数导入已编译的图结构并异步调用它：

from graphs.graph import app


async def get_generation(body: RAGRequest=Body(...)) -> dict:
    try:
        generation = await app.ainvoke(body.dict())

        return {
            "answer": generation.get("generation", ""),
            "web_search": generation.get("web_search", False)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

FastAPI会将其作为依赖项注入，然后返回生成的答案以及一个标志，指示该答案来自RAG还是网络搜索。

实际测试

之前流程中的Streamlit前端可以直接复用。询问如何成为百万富翁会自动路由到网络搜索，而询问什么是提示工程则会路由到基于语义的RAG路径。这体现了动态路由机制，同时让每个答案的来源一目了然。

总结一下：LangGraph StateGraph现在负责管理状态，工作流通过条件边实现带有循环和分支的动态转换，LLM评估器则负责判断文档相关性、检测幻觉以及评估答案质量，而API则会告知用户答案来自RAG还是网络。这一系列工作的下一步是使用工具和ReAct框架，以实现更强的智能行为。