文档智能问答系统 RAG 接入中转 API 实战：从 Embedding 到智能问答的全链路实现

我叫李明，是一名独立开发者。最近在做一个企业内部知识库查询系统，客户希望员工可以通过自然语言直接查询海量产品文档、操作手册和技术规范。传统的关键词搜索无法理解语义，每次搜索都要翻好几页才能找到答案。用户抱怨连连，老板催得紧，我必须尽快上线一个"能理解人话"的智能问答功能。

调研了一圈，RAG（Retrieval-Augmented Generation，检索增强生成）是目前最成熟的企业级方案。但调用 OpenAI API 成本太高，高并发时费用吓人。国内直连速度也不稳定，经常超时。最终我选择了 HolySheep AI 作为中转 API——汇率对标美元官方价，人民币充值直接折算，还支持微信/支付宝，最重要的是国内延迟<50ms，高并发下完全不卡。

这篇文章记录我从零搭建文档智能问答系统的完整过程，包括文档切分、向量嵌入、语义检索和大模型生成的完整链路，所有代码基于 HolySheep API，可直接跑通。

技术架构概览

整个 RAG 系统分为四个核心模块：文档处理、向量存储、语义检索、大模型生成。数据流向如下：

文档解析：PDF、Word、TXT 等格式 → 结构化文本块
文本切分：长文档 → 语义完整的 Chunk（通常 500-1000 字）
向量嵌入：调用 Embedding API，将每个 Chunk 转换为高维向量
向量存储：存入向量数据库（ChromaDB / Milvus），支持 ANN 相似度检索
语义检索：用户问题 → 向量 → Top-K 相关文档块
答案生成：上下文 + 问题 → 大模型 API → 生成答案

HolySheep API 同时支持 Embedding 和 Chat Completion，我用同一个平台搞定了两个环节，代码管理更简洁，账单也一目了然。

环境准备与依赖安装

首先安装必要的 Python 包，建议使用虚拟环境隔离依赖：

# 创建并激活虚拟环境
python -m venv rag-env
source rag-env/bin/activate  # Linux/Mac
rag-env\Scripts\activate  # Windows

安装核心依赖
pip install langchain langchain-community
pip install chromadb  # 向量数据库
pip install openai   # HolySheep 兼容 OpenAI SDK
pip install tiktoken  # 文本分词
pip install pypdf     # PDF 解析
pip install python-dotenv  # 环境变量管理

HolySheep API 密钥配置

在项目根目录创建 .env 文件，填入你的 HolySheep API 密钥。注册后即可获得免费试用额度，充值支持微信和支付宝，非常方便：

import os
from dotenv import load_dotenv

load_dotenv()

HolySheep API 配置
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

使用 LangChain 接入 HolySheep
from langchain_openai import OpenAIEmbeddings
from langchain_community.chat_models import ChatOpenAI

初始化 Embedding 模型（用于文档向量化）
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key=HOLYSHEEP_API_KEY,
    openai_api_base=HOLYSHEEP_BASE_URL,
)

初始化大语言模型（用于答案生成）
llm = ChatOpenAI(
    model="gpt-4o",
    openai_api_key=HOLYSHEEP_API_KEY,
    openai_api_base=HOLYSHEEP_BASE_URL,
    temperature=0.3,  # 低温度保证答案准确性
)

这里有两个关键点：一是 base_url 必须设为 https://api.holysheep.ai/v1，这是 HolySheep 的标准接口地址；二是 API Key 格式为 YOUR_HOLYSHEEP_API_KEY，在控制台获取后填入即可。

文档解析与文本切分

RAG 效果的核心在于文档切分策略。切得太碎，语义丢失；切得太长，上下文窗口浪费。我的经验是按段落 + 固定长度重叠来切，确保每个 Chunk 语义完整：

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

def load_and_split_documents(pdf_path: str, chunk_size: int = 800, chunk_overlap: int = 100):
    """
    加载 PDF 文档并进行语义切分
    
    Args:
        pdf_path: PDF 文件路径
        chunk_size: 每个文本块的目标字符数
        chunk_overlap: 相邻文本块的重叠字符数
    """
    # 加载 PDF
    loader = PyPDFLoader(pdf_path)
    documents = loader.load()
    
    # 使用递归字符切分器，优先按段落切分
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "！", "？", " ", ""],
        length_function=len,
    )
    
    # 切分文档
    chunks = text_splitter.split_documents(documents)
    
    print(f"原始文档数: {len(documents)}")
    print(f"切分后文本块数: {len(chunks)}")
    
    return chunks

示例：处理产品手册
pdf_path = "./docs/product_manual.pdf"
document_chunks = load_and_split_documents(pdf_path)

预览前两个文本块
for i, chunk in enumerate(document_chunks[:2]):
    print(f"\n--- Chunk {i+1} ---")
    print(chunk.page_content[:200] + "...")

这里用了 LangChain 的 RecursiveCharacterTextSplitter，它会优先按段落切分，遇到超长段落再按句子、单词递减切分，确保每个 Chunk 的语义完整性。

向量存储与语义检索

切分好的文本块需要转成向量存入向量数据库。我用 ChromaDB 作为轻量级解决方案，适合中小规模知识库（百万级向量以内完全够用）：

import chromadb
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OpenAIEmbeddings

def create_vector_store(chunks, persist_directory: str = "./chroma_db"):
    """
    创建向量数据库并存储文档嵌入
    
    Args:
        chunks: 切分后的文档块列表
        persist_directory: 向量数据库持久化路径
    """
    # 初始化 Chroma 向量库
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,  # 使用 HolySheep 的 Embedding 模型
        persist_directory=persist_directory,
    )
    
    # 显式持久化
    vectorstore.persist()
    
    print(f"向量数据库创建成功，共存储 {vectorstore._collection.count()} 条向量")
    return vectorstore

def semantic_search(query: str, vectorstore, top_k: int = 5):
    """
    语义检索：查询与文档块的相关性
    
    Args:
        query: 用户问题
        vectorstore: 向量数据库实例
        top_k: 返回最相关的 K 条结果
    """
    docs_with_scores = vectorstore.similarity_search_with_score(query, k=top_k)
    
    print(f"\n检索到 {len(docs_with_scores)} 条相关文档：")
    for i, (doc, score) in enumerate(docs_with_scores):
        print(f"\n[结果 {i+1}] 相似度分数: {score:.4f}")
        print(f"内容预览: {doc.page_content[:150]}...")
    
    return docs_with_scores

创建向量库
vectorstore = create_vector_store(document_chunks)

测试语义检索
query = "产品保修期是多久？如何申请售后？"
results = semantic_search(query, vectorstore, top_k=5)

检索结果会返回文档内容和相似度分数。分数越低（接近 0），表示相关性越强。在实际应用中，我会设置一个阈值（如 0.7），过滤掉相关性太低的检索结果，避免大模型被错误上下文误导。

RAG 问答核心：检索 + 生成

检索到相关文档后，需要组装 Prompt 让大模型基于上下文回答问题。这里用 LangChain 的 LCEL（LangChain Expression Language）构建 RAG Chain：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

def create_rag_chain(vectorstore):
    """
    构建 RAG 问答链：检索 + 生成
    """
    # 定义系统提示词
    system_prompt = """你是一个专业的技术支持助手。请根据以下参考文档回答用户问题。

重要规则：
1. 只使用参考文档中的信息回答，不要编造内容
2. 如果参考文档中没有相关信息，明确告知用户"根据提供的文档，无法回答这个问题"
3. 回答要条理清晰，使用适当的格式
4. 如果涉及步骤或流程，请使用编号列表

参考文档：
{context}
"""
    
    # 构建 Prompt 模板
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        ("human", "{question}"),
    ])
    
    # 定义上下文格式化函数
    def format_context(docs_with_scores):
        context_parts = []
        for i, (doc, score) in enumerate(docs_with_scores, 1):
            context_parts.append(f"[文档 {i}]\n{doc.page_content}")
        return "\n\n".join(context_parts)
    
    # 组装 RAG Chain
    rag_chain = (
        {
            "context": lambda x: format_context(x["docs"]),
            "question": RunnablePassthrough()
        }
        | prompt
        | llm
        | StrOutputParser()
    )
    
    return rag_chain

def answer_question(question: str, vectorstore, top_k: int = 5):
    """
    问答入口函数
    """
    # 1. 语义检索
    docs = vectorstore.similarity_search_with_score(question, k=top_k)
    
    # 2. RAG 生成
    rag_chain = create_rag_chain(vectorstore)
    answer = rag_chain.invoke({"docs": docs})
    
    return answer, docs

完整问答演示
if __name__ == "__main__":
    # 示例问题
    question = "产品保修期是多久？如何申请售后？"
    
    print(f"问题: {question}\n")
    print("-" * 50)
    
    # 获取答案
    answer, retrieved_docs = answer_question(question, vectorstore)
    
    print(f"答案:\n{answer}")

这个 RAG Chain 的核心思想是"检索增强生成"：先用 Embedding 模型找到与问题语义最相关的文档片段，再把这些片段作为上下文喂给大模型。大模型只基于提供的上下文回答，避免了幻觉问题，同时节省了 token 消耗（比直接用长上下文窗口经济得多）。

使用 HolySheep API 的另一个优势是价格。GPT-4o 的 output 价格相比官方有大幅优惠，而 DeepSeek V3.2 的价格更是低至 $0.42/MTok，非常适合高并发的文档问答场景。

部署与 API 服务封装

为了让前端或其他服务调用，我用 FastAPI 封装成 REST API：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(title="RAG 文档问答系统", version="1.0.0")

全局变量存储向量库
vectorstore = None

class QuestionRequest(BaseModel):
    question: str
    top_k: Optional[int] = 5

class AnswerResponse(BaseModel):
    answer: str
    sources: List[dict]

@app.on_event("startup")
async def load_vectorstore():
    """启动时加载向量数据库"""
    global vectorstore
    vectorstore = Chroma(
        persist_directory="./chroma_db",
        embedding_function=embeddings,
    )
    print("向量数据库加载完成")

@app.post("/api/ask", response_model=AnswerResponse)
async def ask_question(request: QuestionRequest):
    """问答接口"""
    if not vectorstore:
        raise HTTPException(status_code=500, detail="向量数据库未初始化")
    
    try:
        answer, docs = answer_question(request.question, vectorstore, request.top_k)
        
        # 提取来源信息
        sources = []
        for i, (doc, score) in enumerate(docs):
            sources.append({
                "content": doc.page_content[:200] + "...",
                "score": round(score, 4),
                "source": doc.metadata.get("source", "unknown")
            })
        
        return AnswerResponse(answer=answer, sources=sources)
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """健康检查接口"""
    return {"status": "healthy", "vector_count": vectorstore._collection.count()}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

部署后，只需 POST 请求即可获取答案：

curl -X POST http://localhost:8000/api/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "产品保修期是多久？", "top_k": 3}'

常见报错排查

1. API 认证失败（401 Unauthorized）

错误信息：AuthenticationError: Incorrect API key provided

原因：API Key 填写错误或未正确加载环境变量

解决步骤：

确认 .env 文件中 HOLYSHEEP_API_KEY 格式正确，无多余空格
检查是否调用了 load_dotenv() 加载环境变量
确认使用的是 HolySheep 的 Key，不是其他平台的 Key
登录 HolySheep 控制台查看 Key 是否有效

2. 向量检索结果为空

错误信息：检索返回 0 条结果

原因：向量数据库未正确初始化或文档加载失败

解决步骤：

检查 PDF 文件路径是否正确，文件是否存在
确认 vectorstore.persist() 已调用，数据已持久化
验证 Embedding 模型能正常工作：单独测试 embeddings.embed_query("测试文本")
检查向量数据库目录权限，确保可读写

3. 大模型生成内容不相关

表现：检索到的文档片段正确，但生成的答案偏离主题

原因：检索相关性阈值过低，或者 Prompt 引导不足

解决步骤：

提高相似度阈值：只保留分数 < 0.6 的结果
优化 Prompt：明确要求"仅基于提供上下文回答"
尝试增加 top_k 值，提供更多候选上下文
降低 temperature 参数（建议 0.2-0.3），减少随机性

4. 请求超时或响应慢

错误信息：RequestTimeoutError

文档智能问答系统 RAG 接入中转 API 实战：从 Embedding 到智能问答的全链路实现

技术架构概览

环境准备与依赖安装

rag-env\Scripts\activate # Windows

安装核心依赖

HolySheep API 密钥配置

HolySheep API 配置

使用 LangChain 接入 HolySheep

初始化 Embedding 模型（用于文档向量化）

初始化大语言模型（用于答案生成）

文档解析与文本切分

示例：处理产品手册

预览前两个文本块

向量存储与语义检索

创建向量库

测试语义检索

RAG 问答核心：检索 + 生成

完整问答演示

部署与 API 服务封装

全局变量存储向量库

常见报错排查

1. API 认证失败（401 Unauthorized）

2. 向量检索结果为空

3. 大模型生成内容不相关

4. 请求超时或响应慢

相关资源

相关文章

技术架构概览

环境准备与依赖安装

rag-env\Scripts\activate # Windows

安装核心依赖

HolySheep API 密钥配置

HolySheep API 配置

使用 LangChain 接入 HolySheep

初始化 Embedding 模型（用于文档向量化）

初始化大语言模型（用于答案生成）

文档解析与文本切分

示例：处理产品手册

预览前两个文本块

向量存储与语义检索

创建向量库

测试语义检索

RAG 问答核心：检索 + 生成

完整问答演示

部署与 API 服务封装

全局变量存储向量库

常见报错排查

1. API 认证失败（401 Unauthorized）

2. 向量检索结果为空

3. 大模型生成内容不相关

4. 请求超时或响应慢

相关资源

相关文章

🔥 推荐使用 HolySheep AI