从零实现RAG问答应用：AI应用开发实践

来源：互联网 2026-07-29 20:24:04

基于LangChain等工具实现RAG问答应用，通过文件解析、文档切片、向量化存储、检索相关片段及大模型生成回答的流程，解决大模型知识截止与私有数据访问问题。实践涵盖技术选型、核心代码及界面设计，并指出链式调用、向量库只读错误、检索混淆等常见问题。

上周把一本三国历史小说扔进了一个 RAG 系统，发现它居然能直接回答"诸葛亮第几次北伐在哪年"这类问题——这篇文章就记录了这个过程是怎么实现的。

一、RAG是什么

大模型的知识有个硬伤——它有截止日期，训练完就固定了，而且完全没法访问私有数据。RAG 就是来补这个短板的：把私有文档喂给模型，让它结合这些内容来生成回答。

长期稳定更新的攒劲资源： >>>点此立即查看<<<

提供额外知识的过程叫检索增强（RA），模型生成回答的过程是 G，合起来就是检索增强生成——RAG（Retrieval Augmented Generation）。直白点说，就是给大模型开了一扇"外设知识库"的门。

二、技术选型和准备工作

工具清单如下：

工具	用途
LangChain	大模型应用开发框架
ChromaDB	向量数据库
BAAI/bge-m3	Embedding 模型
DeepSeek	大语言模型
硅基流动API	模型 API 平台
Streamlit	界面设计

三、核心代码实现

整体流程其实并不复杂：

加载文件——上传并读取文件内容
文档切片——将文档切分成小片段，片段大小可以指定
向量化——将文档内容转换为多维向量数据
存数据库——将向量化的数据存入数据库
检索相关片段——用问题从向量数据库中查找 Top-k 个文档片段
生成回答——问题 + 片段一起送给大模型生成回答

这里需要先了解一下 LangChain 的几个核心对象：

对象	作用
Document	文档片段，包含 page_content、metadata 等
Splitter	将长文档切分成小片段
Embeddings	把文字变成向量（封装了 API 调用）
VectorStore	向量数据库封装（ChromaDB 等）
Retriever	检索器，从 VectorStore 里检索相关文档
PromtTemplate	格式化创建提示词
LLM / ChatModel	调用大模型 API

先安装依赖：

pip install langchain langchain-community langchain-openai chromadb

1. 文件解析、切片、向量化、存库

第一步是解析上传的文件，返回 Document 对象：

def parse_uploaded_file(uploaded_file):
    """解析上传的文件，返回 Document 对象，失败返回 None。"""
    filename = uploaded_file.name
    data = uploaded_file.read()
    if filename.endswith('.txt'):
        try:
            text = data.decode('utf-8')
        except:
            text = data.decode('gbk', errors='ignore')
    elif filename.endswith('.pdf'):
        import pypdf
        reader = pypdf.PdfReader(io.BytesIO(data))
        text = '\n\n'.join([p.extract_text() or '' for p in reader.pages])
    else:
        return None
    text = text.strip()
    if not text:
        return None
    return Document(page_content=text, metadata={'source': filename})

然后从 Document 列表构建内存向量库：

def build_from_documents(docs_list):
    """从 Document 列表构建内存向量库。"""
    # 文档切片
    splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50, length_function=len)
    chunks = splitter.split_documents(docs_list)
    emb, _ = get_models()  # 向量化工具
    vs = Chroma.from_documents(
        documents=chunks,
        embedding=emb,
        persist_directory=None  # 内存模式
    )  # 存入数据库
    return vs, chunks, len(chunks)

2. 检索、生成

检索和生成这一步，核心是用一个提示词模板把上下文、历史对话和问题组装起来：

RAG_PROMPT = """
你是文档问答助手。请严格基于以下文档内容回答问题。
如文档中没有相关信息，请说"文档中未提及此内容"，不要编造。
文档内容：{context}
对话历史：{history}
当前问题：{question}
回答：
"""

def rag_answer(question, vectorstore, all_chunks, chat_history, k=3):
    # 向量检索
    relevant_docs = vectorstore.as_retriever(search_kwargs={"k": k}).invoke(question)
    context = "\n\n---\n\n".join([f"[{d.metadata.get('source','')}]\n{d.page_content}" for d in relevant_docs])
    recent = chat_history[-6:]
    history_text = "\n".join([f"{m['role'].upper()}: {m['content']}" for m in recent]) if recent else "（无历史）"
    prompt = ChatPromptTemplate.from_template(RAG_PROMPT)
    _, llm = get_models()
    response = llm.invoke(prompt.format_messages(context=context, history=history_text, question=question))
    return response.content, relevant_docs

3. 界面设计

界面用 Streamlit 来实现，先安装：

pip install streamlit
# 验证安装成功
streamlit hello

主页面 UI 代码如下：

st.title("历史知识助手")
if st.session_state.vs and st.session_state.doc_count > 0:
    st.caption(f"知识库已创建，文档{st.session_state.doc_count}个，切片{st.session_state.chunk_count}个")
else:
    st.caption("知识库未创建,请点左侧按钮创建/重建")

# 对话历史
for item in st.session_state.chat_history:
    with st.chat_message(item["role"]):
        st.write(item["content"])
        # 显示来源，可折叠
        if item["role"] == "assistant" and item.get("sources"):
            sources = item.get("sources")
            with st.expander(" 参考来源"):
                for i, source in enumerate(sources):
                    src = source.metadata.get("source", "未知")
                    st.write(f"**{i+1}. {src}**")
                    st.caption(f"{source.page_content[:30]}...")
st.markdown("---")

# 输入框
if question := st.chat_input("请根据知识库提问："):
    if not st.session_state.vs:
        st.error("知识库未创建,请点左侧按钮创建/重建")
        st.stop()
    # 展示用户消息
    st.session_state.chat_history.append({"role": "user", "content": question})
    with st.chat_message("user"):
        st.write(question)
    # 展示助手消息
    with st.chat_message("assistant"):
        with st.spinner("正在检索文献并生成回答..."):
            answer, sources = rag_answer(
                question,
                st.session_state.vs,
                [m for m in st.session_state.chat_history if m['role'] != 'system'],
                k=k_value
            )
            st.write(answer)
            with st.expander(" 参考来源"):
                for i, source in enumerate(sources):
                    src = source.metadata.get("source", "未知")
                    st.write(f"**{i+1}. {src}**")
                    st.caption(f"{source.page_content[:30]}...")
    st.markdown("---")
    st.session_state.chat_history.append(
        {"role": "assistant", "content": answer, "sources": sources}
    )

四、界面展示

五、踩坑记录、未来改进

坑一：看不懂链式调用

第一次看到这段代码的时候，整个人都是懵的：

prompt = ChatPromptTemplate.from_template(RAG_PROMPT)
chain = prompt | llm | StrOutputParser()
answer = chain.invoke({"context": context, "history": history_text, "question": question})

这里的 | 不是"或"，是 LangChain 定义的连接符，作用是把左边的输出传给右边作为输入。拆开来写就一目了然了：

formatted_prompt = prompt.invoke({"context": context, "history": history_text, "question": question})
llm_response = llm.invoke(formatted_prompt)
answer = StrOutputParser().invoke(llm_response)

链式写法的好处是，如果后续想换个模型或换个输出解析器，只改一行就行，其他代码完全不受影响。

坑二：重建向量库报 readonly 错误

用持久化模式（persist_directory 指定路径）时，第二次点击"重建知识库"按钮就报错了：

chromadb.errors.InternalError: Query error: Database error: error returned from database: (code: 1032) attempt to write a readonly database

先后折腾了三种方案，都没成功：

shutil.rmtree 删目录——失败，删完重建还是报错
gc.collect() 强制回收旧对象——失败，对象还在 session_state 里被引用着
把新库建到临时目录再 shutil.move——失败，move 操作本身也触发了同样的错误

最后才搞明白根本原因：错误码 1032 是 SQLITE_READONLY_DBMOVED，SQLite 通过 inode 追踪文件，只要对 chroma.sqlite3 执行了删除或移动，inode 就变了，旧连接检测到变化就会拒绝写入。正确解法其实很简单——根本不用碰文件系统，直接调用 ChromaDB 自己的 API：

existing_vs.delete_collection()
# 在 SQLite 内部删数据，文件 inode 不变

坑三：向量检索"认错人"

上传了三篇文档——曹操介绍、诸葛亮介绍、赤壁之战介绍。问曹操能答，但问诸葛亮却回答"文档中未提及"。打印检索日志后发现，问"诸葛亮是哪里人"时，检索到的 3 个片段全部来自 caocao.txt。

原因在于这三篇文档内容高度相关，人名互相穿插——诸葛亮文章里多次提到曹操，曹操文章里也提到了赤壁和诸葛亮。向量语义检索计算的是整体语义相似度，在这种互相交织的文档里，就很容易"认错人"。这个问题单靠调参是解决不了的，得从检索策略上改进。

六、下一步改进方向

混合检索：向量语义检索 + BM25 关键词检索，两路结果用 RRF 算法融合。BM25 基于词频，"诸葛亮"这个词在哪篇文档出现多，得分就高，不受语义干扰
Reranking 重排序：检索召回 20 个片段后，用专门的重排序模型精选最相关的 3 个，比直接截断效果更好
Query 改写：用户提问"说说你知道的诸葛亮"这类口语化问题，直接用来检索效果差，先用大模型改写成"诸葛亮生平历史成就"再检索

侠游戏发布此文仅为了传递信息，不代表侠游戏网站认同其观点或证实其描述