想知道如何搭建 Adaptive RAG 系統？這篇文章告訴你一切！

本文會帶你從零搭建一個完整的概念驗證項目（POC），技術棧涵蓋 Adaptive RAG、LangGraph、FastAPI 和 Streamlit 四個核心組件。Adaptive RAG 負責根據查詢復雜度自動調整檢索策略；LangGraph 把多步 LLM 推理組織成有狀態的可靠工作流；FastAPI 作為高性能后端暴露整條 AI 管道；Streamlit 則提供一個可以直接交互的前端界面。

讀完這篇文章，你拿到的不只是理論——而是一個跑得起來的端到端 AI 系統。

要構建的是一個技術支持智能助手。它能理解用戶查詢，根據問題復雜度動態選擇檢索深度（Adaptive RAG），通過 LangGraph 執行推理工作流，經由 FastAPI 返回結果，最后在 Streamlit UI 上呈現響應。

這個場景針對的是一個真實痛點：團隊面對大規模文檔集時，傳統 RAG 在模糊查詢或多步驟問題上經常答非所問。

技術概覽

Adaptive RAG

可以把 Adaptive RAG 理解為"搜索之前先思考"的 RAG。簡單查詢走輕量級檢索就夠了，遇到復雜問題則自動切換到多跳深度搜索、重排序或查詢擴展，用更低的延遲換更高的準確率。

LangGraph

LangGraph 是用來構建有狀態、多步驟 AI 工作流的框架。和傳統鏈式調用不同它把 LLM 工作流建模成一張圖——每個節點對應一個步驟（檢索 → 推理 → 驗證 → 響應），原生支持重試、記憶、循環和故障轉移。對于需要在生產環境中保證可預測行為的場景，這種抽象比線性 chain 靈活得多。

FastAPI

FastAPI 把 Adaptive RAG + LangGraph 包裝成 API 接口對外暴露，處理請求分發，天然適配異步 I/O。

Streamlit

前端用 Streamlit 搭建，聊天風格的界面，不需要寫 HTML/CSS做 POC 演示足夠了。

系統架構

數據流走向：

User → Query → Streamlit UI
Streamlit → Sends request → FastAPI
FastAPI → Passes query → LangGraph
LangGraph → Runs Adaptive RAG → Retriever
Retriever → Gets chunks → Vector DB
Vector DB → Returns results → LangGraph
LangGraph → Generates final response
FastAPI → Sends to UI → User

文件夾結構

項目結構盡量精簡：

ai-poc/
│
├── backend/ # 后端邏輯
│ ├── app.py # FastAPI API 服務器
│ ├── rag_pipeline.py # Adaptive RAG 檢索
│ ├── graph_workflow.py # LangGraph 工作流
│ ├── config.py # 配置和環境設置
│ ├── data/ # 源文檔
│ └── __init__.py # 包初始化器
│
├── frontend/ # UI 層
│ ├── ui.py # Streamlit 界面
│ └── __init__.py # 包初始化器
│
├── .env # API 密鑰和機密信息
├── requirements.txt # 項目依賴
└── README.md # 設置說明

requirements.txt 文件

fastapi
uvicorn[standard]
streamlit
requests
pydantic
langchain
langchain-community
langgraph
faiss-cpu
sentence-transformers
openai
python-dotenv

代碼實現（關鍵代碼片段）

Adaptive RAG 管道（rag_pipeline.py）

# 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 []
# Adaptive heuristic
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:
"""
Build FAISS index from raw texts (POC only).
In production load persisted DB instead.
"""
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)

自適應的核心邏輯其實很簡單：根據查詢的 Token 數決定檢索深度。查詢短于 6 個詞就取 3 條結果，否則拉 8 條。這是一個粗粒度的啟發式方法，在 POC 階段夠用，生產環境可以替換成更精細的分類器。

build_vector_store 函數從原始文本構建 FAISS 索引。注意這里每次啟動都重建索引，生產上應該加載持久化的數據庫。

LangGraph 工作流（graph_workflow.py）

# 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)
# Retrieval Node
async def retrieve_node(state: GraphState):
docs = rag.retrieve(state["question"])
return {"docs": docs}
# Reasoning Node
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}
# Add nodes
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("reason", reasoning_node)
# Connect nodes
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "reason")
workflow.add_edge("reason", END)
return workflow.compile()

整個工作流只有兩個節點：retrieve 負責檢索，reason 負責推理生成答案。GraphState 作為 TypedDict 在節點間傳遞狀態。流程很線性——先檢索再推理，然后結束。實際項目中可以在這個圖上加驗證節點、循環重試等分支，LangGraph 的圖結構天然支持這種擴展。

FastAPI 后端（app.py）

# 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")
# ---------------------------
# Startup Initialization
# ---------------------------
class AskRequest(BaseModel):
query: str
@app.on_event("startup")
async def startup_event():
global workflow
# Sample knowledge base (replace with real docs)
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)
# ---------------------------
# API Endpoint
# ---------------------------
@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"
)

后端在啟動時完成向量庫構建和工作流初始化，之后通過 /ask 端點接收查詢請求。這里用了 global 變量來持有 workflow 實例——POC 階段這樣做沒問題，上生產建議用依賴注入替代。

Streamlit UI（ui.py）

# 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}")

前端就這么幾行：輸入框接收問題，按鈕觸發請求，拿到結果直接渲染。Streamlit 的好處就是不用折騰前端那套東西，做 POC 驗證概念足夠。

運行項目

安裝依賴：

pip install -r requirements.txt

設置 OpenAI Key：

export OPENAI_API_KEY="your_key_here"
Or(For Windows)
setx 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 應用的核心架構需求。

https://avoid.overfit.cn/post/770176403fa04ac49a55a145c36266b8

by Robi Kumar Tomar