Adaptive RAG 系统搭建：LangGraph、FastAPI 与 Streamlit 实战

# backend/rag_pipeline.py
from typing import List
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document

class AdaptiveRAG:
    """Adaptive Retrieval Pipeline"""
    def __init__(self, vector_db: FAISS):
        self.db = vector_db

    def retrieve(self, query: str) -> List[Document]:
        if not query.strip():
            return []
        # 启发式策略：根据 Token 数决定 k 值
        token_count = len(query.split())
        k = 3 if token_count < 6 else 8
        return self.db.similarity_search(query, k=k)

def build_vector_store(texts: List[str]) -> FAISS:
    """从原始文本构建 FAISS 索引（POC 阶段）"""
    embeddings = HuggingFaceEmbeddings(
        model_name="sentence-transformers/all-MiniLM-L6-v2"
    )
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=100
    )
    docs = []
    for text in texts:
        chunks = splitter.split_text(text)
        for chunk in chunks:
            docs.append(chunk)
    return FAISS.from_texts(docs, embeddings)

注意 build_vector_store 函数每次启动都重建索引，生产环境应该加载持久化的数据库。自适应的核心逻辑其实很简单：查询短于 6 个词就取 3 条结果，否则拉 8 条。这是一个粗粒度的启发式方法，在 POC 阶段够用，生产环境可以替换成更精细的分类器。

LangGraph 工作流 (graph_workflow.py)

整个工作流只有两个节点：retrieve 负责检索，reason 负责推理生成答案。GraphState 作为 TypedDict 在节点间传递状态。流程很线性——先检索再推理，然后结束。实际项目中可以在这个图上加验证节点、循环重试等分支，LangGraph 的图结构天然支持这种扩展。

# backend/graph_workflow.py
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain.schema import Document
from langchain_openai import ChatOpenAI

class GraphState(TypedDict):
    question: str
    docs: List[Document]
    answer: str

def create_workflow(rag):
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    workflow = StateGraph(GraphState)

    async def retrieve_node(state: GraphState):
        docs = rag.retrieve(state["question"])
        return {"docs": docs}

    async def reasoning_node(state: GraphState):
        question = state["question"]
        docs = state.get("docs", [])
        context = "\n\n".join([d.page_content for d in docs])
        prompt = f"""
        You are a technical assistant. Use ONLY the context below to answer the question.
        If the answer is not in the context, say you don't know.
        Context: {context}
        Question: {question}
        """
        response = await llm.ainvoke(prompt)
        return {"answer": response.content}

    workflow.add_node("retrieve", retrieve_node)
    workflow.add_node("reason", reasoning_node)
    workflow.set_entry_point("retrieve")
    workflow.add_edge("retrieve", "reason")
    workflow.add_edge("reason", END)
    return workflow.compile()

FastAPI 后端 (app.py)

后端在启动时完成向量库构建和工作流初始化，之后通过 /ask 端点接收查询请求。这里用了 global 变量来持有 workflow 实例——POC 阶段这样做没问题，上生产建议用依赖注入替代。

# backend/app.py
import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from dotenv import load_dotenv
from rag_pipeline import AdaptiveRAG, build_vector_store
from graph_workflow import create_workflow

load_dotenv()
app = FastAPI(title="Adaptive RAG API")

class AskRequest(BaseModel):
    query: str

@app.on_event("startup")
async def startup_event():
    global workflow
    # 示例知识库（生产环境请替换为真实文档）
    sample_docs = [
        "LangGraph supports stateful workflows and retry logic.",
        "Adaptive RAG dynamically changes retrieval depth based on query complexity.",
        "FastAPI is a high-performance async Python framework.",
    ]
    vector_db = build_vector_store(sample_docs)
    rag = AdaptiveRAG(vector_db)
    workflow = create_workflow(rag)

@app.post("/ask")
async def ask(payload: AskRequest):
    if not payload.query.strip():
        raise HTTPException(status_code=400, detail="Query cannot be empty")
    try:
        result = await workflow.ainvoke({"question": payload.query})
        return {"response": result["answer"]}
    except Exception as e:
        raise HTTPException(status_code=500, detail="Internal RAG processing error")

Streamlit UI (ui.py)

前端就这么几行：输入框接收问题，按钮触发请求，拿到结果直接渲染。Streamlit 的好处就是不用折腾前端那套东西，做 POC 验证概念足够。

# frontend/ui.py
import streamlit as st
import requests

API_URL = "http://localhost:8000/ask"
st.set_page_config(page_title="Adaptive RAG Assistant")
st.title("Adaptive RAG Support Assistant")

query = st.text_input("Enter your question")
if st.button("Ask"):
    if not query.strip():
        st.warning("Please enter a question.")
    else:
        try:
            with st.spinner("Thinking..."):
                response = requests.post(API_URL, json={"query": query}, timeout=60)
                response.raise_for_status()
                answer = response.json()["response"]
                st.markdown("### Answer:")
                st.write(answer)
        except Exception as e:
            st.error(f"Error: {e}")

运行项目

安装依赖并配置环境变量后，即可启动服务。

pip install -r requirements.txt
export OPENAI_API_KEY="your_key_here"
uvicorn backend.app:app --reload
streamlit run frontend/ui.py

内部执行流程

在 UI 中输入一条查询，例如 How does retry logic work in LangGraph workflows?。请求先到达 FastAPI 后端。LangGraph 工作流从 retrieve 节点启动，Adaptive RAG 根据查询长度动态选定检索深度——短查询取 k=3，长查询取 k=8。从向量数据库拉到相关文档块后，reasoning 节点把这些上下文拼装成 prompt，交给 LLM 生成答案。LLM 的回答被限定在检索到的上下文范围内，最终结果沿原路返回到 UI。

一切正常的话，你现在手上就有了一条完整的端到端 RAG 管道：UI → API → Graph → Retriever → LLM → Response。

下一步：生产部署

POC 跑通了但离生产还有距离。下面按模块列一下需要补强的方向。

检索层 向量相似度搜索可以跟 BM25 关键词搜索做混合，在一些 edge case 上召回率会好很多。检索完 top-k 文档后，再套一层 Cross-Encoder 做重排序，排序精度能上一个台阶。如果系统要支持多团队或多租户，还得引入基于命名空间的文档隔离，防止跨域信息泄漏。

工作流 当前工作流里没有验证环节。生产环境建议加一个验证节点，检查生成的答案是否真的有检索上下文支撑——这对控制幻觉至关重要。另外，如果要做多轮对话，就需要往图里加记忆节点来持久化对话状态。

重试与回退 LLM 调用失败后要有重试机制。主模型不可用时能自动降级到更小或更便宜的备选模型。超时控制和优雅降级也不能少。成本控制也值得考虑：简单查询走轻量模型，只在必要时才升级到大模型。

可观测性与评估 日志要记全——检索分数、命中的文档、响应延迟、Token 消耗，这些都得有。定期做离线评估，准备好测试数据集，跑检索质量和回答质量的指标。幻觉监控单独拎出来盯，追踪那些答案脱离检索上下文的 case。

UI 改进 聊天界面得支持历史记录和多轮对话。回答来源要做高亮——用户应该能看到答案是从哪几段文档生成的。再加上反馈按钮，让用户对回答打分，收集回来的数据可以用于后续评估和微调。

部署与基础设施 前后端都做 Docker 容器化，保证部署的可复现性。上云的话 AWS、GCP、Azure 都行，记得配自动扩缩容。端点要上 HTTPS 和 Token 认证（JWT 或 OAuth）。生产环境的 API Key 不要再靠环境变量了，换托管的密钥存储服务。

总结

一个模块化、可扩展的 RAG 架构就搭建完成了，它完全可以从当前的 POC 状态逐步演化成生产级系统。自适应检索、有状态编排、可扩展 API、简洁的交互界面——这几个构建块拼在一起，基本覆盖了现代 LLM 应用的核心架构需求。