本文会带你从零搭建一个完整的概念验证项目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。简单查询走轻量级检索就够了遇到复杂问题则自动切换到多跳深度搜索、重排序或查询扩展用更低的延迟换更高的准确率。LangGraphLangGraph 是用来构建有状态、多步骤 AI 工作流的框架。和传统链式调用不同它把 LLM 工作流建模成一张图——每个节点对应一个步骤检索 → 推理 → 验证 → 响应原生支持重试、记忆、循环和故障转移。对于需要在生产环境中保证可预测行为的场景这种抽象比线性 chain 灵活得多。FastAPIFastAPI 把 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, kk) 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_namesentence-transformers/all-MiniLM-L6-v2 ) splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap100 ) 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(modelgpt-4o-mini, temperature0) 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 dont 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(titleAdaptive 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_code400, detailQuery cannot be empty) try: result await workflow.ainvoke( {question: payload.query} ) return {response: result[answer]} except Exception as e: raise HTTPException( status_code500, detailInternal RAG processing error )后端在启动时完成向量库构建和工作流初始化之后通过 /ask 端点接收查询请求。这里用了 global 变量来持有 workflow 实例——POC 阶段这样做没问题上生产建议用依赖注入替代。Streamlit UIui.py# frontend/ui.py import streamlit as st import requests API_URL http://localhost:8000/ask st.set_page_config(page_titleAdaptive 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}, timeout60 ) response.raise_for_status() answer response.json()[response] st.markdown(### Answer:) st.write(answer) except Exception as e: st.error(fError: {e})前端就这么几行输入框接收问题按钮触发请求拿到结果直接渲染。Streamlit 的好处就是不用折腾前端那套东西做 POC 验证概念足够。运行项目安装依赖pip install -r requirements.txt设置 OpenAI Keyexport OPENAI_API_KEYyour_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 根据查询长度动态选定检索深度——短查询取k3长查询取k8。从向量数据库拉到相关文档块后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/770176403fa04ac49a55a145c36266b8by Robi Kumar Tomar