LangChain + ChromaDB 实战：从0搭建本地知识库

# ❌ 不使用 LangChain（硬编码方式）
# 问题：每个环节都要自己写，代码复杂，难以维护

def rag_without_langchain(question, vector_store, llm):
    # 1. 手动检索
    query_embedding = embedding_model.encode([question])
    results = vector_store.search(query_embedding, top_k=5)
    contexts = [r['content'] for r in results]
    
    # 2. 手动构建 Prompt
    prompt = f"基于以下上下文回答问题：\n\n{contexts}\n\n问题：{question}"
    
    # 3. 手动调用 LLM
    response = llm.generate(prompt)
    
    return response

# ✅ 使用 LangChain（声明式方式）
# 优点：代码简洁，逻辑清晰，易于扩展

def rag_with_langchain(question, retriever, llm):
    # 创建一个 RAG Chain，一行代码搞定
    rag_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever
    )
    
    # 调用 Chain
    response = rag_chain.invoke(question)
    return response

# 步骤1：创建虚拟环境（推荐，避免依赖冲突）
python3 --version  # 确认是 3.10+
python3 -m venv ~/rag-env
source ~/rag-env/bin/activate

# 步骤2：安装 LangChain 相关包
# 基础 LangChain
pip install langchain==0.2.0 langchain-core==0.2.0 langchain-community==0.2.0

# 向量数据库
pip install chromadb==0.5.0

# Embedding 模型
pip install sentence-transformers==2.7.0

# PDF 解析（可选）
pip install pypdf==4.2.0 unstructured==0.14.0

# LLM 接口（如果你用 OpenAI 或其他 API）
pip install openai==1.30.0

# 或者用本地模型（推荐 Ollama）
pip install ollama==0.1.0

# Web 服务（可选）
pip install fastapi==0.111.0 uvicorn==0.29.0 streamlit==1.34.0

# 向量化处理加速（可选，加速 Embedding 计算）
pip install faiss-cpu  # CPU 版本
# 或
pip install faiss-gpu  # GPU 版本

# 验证所有核心包是否安装成功
python -c "
import langchain
import chromadb
import sentence_transformers
print(f'LangChain: {langchain.__version__}')
print(f'ChromaDB: {chromadb.__version__}')
print(f'sentence-transformers: {sentence_transformers.__version__}')
print('✅ 所有依赖安装成功！')
"

from langchain_core.documents import Document

# 创建一个 Document
doc = Document(
    page_content="这是文档的正文内容",
    metadata={
        "source": "product_manual.pdf",  # 文档来源
        "page": 3,                       # 页码
        "category": "用户手册"           # 自定义分类
    }
)

print(f"内容: {doc.page_content}")
print(f"元数据: {doc.metadata}")

# 批量创建
docs = [
    Document(page_content="第一个文档", metadata={"id": 1}),
    Document(page_content="第二个文档", metadata={"id": 2}),
]

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    MarkdownTextSplitter,
    PythonCodeTextSplitter
)

# ========== RecursiveCharacterTextSplitter（最常用） ==========
# 按优先级分割：段落 → 换行 → 句号 → 单词 → 字符
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每个 chunk 的目标字符数
    chunk_overlap=50,     # 相邻 chunk 之间的重叠字符数
    length_function=len,  # 用 len() 计算长度
    separators=[
        "\n\n",   # 第一优先级：段落（两个换行）
        "\n",     # 第二优先级：单换行
        "。",     # 第三优先级：中文句号
        "！",
        "？",
        "；",
        "，",
        " ",      # 英文空格
        "",       # 最后按字符分割
    ]
)

# ========== MarkdownTextSplitter ==========
# 专为 Markdown 设计，保留标题、列表等结构
md_splitter = MarkdownTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)

# ========== PythonCodeTextSplitter ==========
# 专为代码设计，保持函数、类的完整性
code_splitter = PythonCodeTextSplitter(
    chunk_size=1000,
    chunk_overlap=100
)

# 使用示例
text = """
# 产品手册

这是一个非常重要的产品文档。

## 第一章：产品介绍

本公司成立于2020年，专注于AI技术研发。

## 第二章：使用方法

1. 打开产品
2. 输入账号密码
3. 开始使用

具体操作步骤：
- 第一步：访问网址
- 第二步：点击登录
- 第三步：输入信息
"""

# 分割文本
chunks = recursive_splitter.split_text(text)
print(f"分割成 {len(chunks)} 个块")
for i, chunk in enumerate(chunks):
    print(f"\n--- Chunk {i+1} ---")
    print(chunk[:100])

from langchain_community.embeddings import HuggingFaceEmbeddings

# 创建 Embedding 模型（使用 BAAI/bge-large-zh-v1.5，中文最强）
embeddings = HuggingFaceEmbeddings(
    model_name='BAAI/bge-large-zh-v1.5',  # 模型名称
    model_kwargs={'device': 'cpu'},        # 使用 CPU（GPU 用 'cuda'）
    encode_kwargs={'normalize_embeddings': True}  # 归一化，便于计算余弦相似度
)

# 测试 Embedding
test_text = "如何安装 Python 环境？"
embedding_vector = embeddings.embed_query(test_text)
print(f"向量维度: {len(embedding_vector)}")  # 1024（BGE-large）
print(f"向量前5个值: {embedding_vector[:5]}")

# 对多个文本生成 Embedding
texts = [
    "Python 安装教程",
    "如何配置环境变量",
    "Java 编程基础"
]
vectors = embeddings.embed_documents(texts)
print(f"生成了 {len(vectors)} 个向量，每个维度 {len(vectors[0])}")

from langchain_community.vectorstores import Chroma

# ========== 创建向量数据库 ==========
# 方法1：直接从文档列表创建

# 准备文档
from langchain_core.documents import Document

docs = [
    Document(page_content="Python 环境安装教程", metadata={"source": "install.pdf"}),
    Document(page_content="Django 框架使用指南", metadata={"source": "django.pdf"}),
    Document(page_content="机器学习基础概念", metadata={"source": "ml.pdf"}),
]

# 创建向量数据库（会自动计算 Embedding）
vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=embeddings,           # 使用前面定义的 Embedding 模型
    persist_directory="./chroma_db",  # 持久化存储路径
    collection_name="knowledge_base"   # collection 名称
)

print(f"向量数据库创建成功，共 {vectorstore._collection.count()} 个文档")

# ========== 方法2：从已有的向量数据库加载 ==========
vectorstore_load = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="knowledge_base"
)

# ========== 相似度检索 ==========
# 检索最相关的文档
results = vectorstore_load.similarity_search(
    query="Python怎么安装",      # 查询文本
    k=3                          # 返回前3个最相关结果
)

print(f"\n检索到 {len(results)} 个相关文档:")
for i, doc in enumerate(results, 1):
    print(f"\n结果 {i}:")
    print(f"内容: {doc.page_content}")
    print(f"元数据: {doc.metadata}")

# ========== 带分数的检索（返回相似度得分）==========
results_with_scores = vectorstore_load.similarity_search_with_score(
    query="Python怎么安装",
    k=3
)

for doc, score in results_with_scores:
    print(f"内容: {doc.page_content[:50]}..., 相似度得分: {score:.4f}")

from langchain_core.retrievers import RetrieverOutputLike

# 将 VectorStore 转换为 Retriever
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 3}  # 默认返回3个结果
)

# 使用 retriever 检索
docs = retriever.invoke("Python怎么安装")
print(f"检索到 {len(docs)} 个文档")

# ========== 多查询检索（MultiQueryRetriever）==========
# 自动将用户查询改写成多个不同版本，增加召回率
from langchain.retrievers.multi_query import MultiQueryRetriever

llm = ...  # 需要一个 LLM 来改写查询

multi_retriever = MultiQueryRetriever(
    retriever=retriever,
    llm=llm,
    prompt="你是一个AI助手。请根据原始问题，生成3个不同的搜索查询，"
           "从不同角度表达同一个问题，以便检索到更多相关内容。"
)

# 使用多查询检索
docs = multi_retriever.invoke("Python怎么安装")
print(f"多查询检索到 {len(docs)} 个文档")

# ========== Ensemble Retriever（组合检索器）==========
# 结合多个检索器的结果，用 Reciprocal Rank Fusion 算法融合排序
from langchain.retrievers.ensemble import EnsembleRetriever

# 例如：结合稀疏检索（BM25）和密集检索（Embedding）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, dense_retriever],  # 多个检索器
    weights=[0.3, 0.7]                             # 权重分配
)

my-rag-project/
├── app.py                  # FastAPI 主应用
├── config.py               # 配置文件
├── requirements.txt         # 依赖列表
├── data/                   # 知识库文档目录
│   ├── docs/               # 原始文档（PDF/Word/MD）
│   └── chroma_db/          # ChromaDB 向量数据库
├── scripts/
│   ├── ingest.py           # 文档导入脚本
│   └── test.py             # 测试脚本
└── src/
    ├── __init__.py
    ├── document_loader.py   # 文档加载器
    ├── text_splitter.py    # 文本分割器
    ├── vectorstore.py      # 向量数据库管理
    ├── chain.py            # RAG Chain
    └── api.py              # API 接口

#!/usr/bin/env python3
"""
配置文件
"""
import os

# ============== 路径配置 ==============
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, "data")
CHROMA_DB_DIR = os.path.join(DATA_DIR, "chroma_db")

# ============== Embedding 配置 ==============
EMBEDDING_MODEL = "BAAI/bge-large-zh-v1.5"  # 中文最强 Embedding
EMBEDDING_DEVICE = "cpu"  # 或 "cuda"（GPU）
EMBEDDING_DIM = 1024      # BGE-large 的向量维度

# ============== 文本分割配置 ==============
CHUNK_SIZE = 500          # 每块目标字符数
CHUNK_OVERLAP = 50         # 重叠字符数

# ============== LLM 配置 ==============
# 方式1：使用 Ollama 本地模型（推荐，无需 API Key）
OLLAMA_BASE_URL = "http://localhost:11434"
OLLAMA_MODEL = "qwen2:7b"  # 或 "llama3" 等

# 方式2：使用 OpenAI API
# OPENAI_API_KEY = "sk-xxxx"  # 你的 API Key
# OPENAI_MODEL = "gpt-3.5-turbo"

# ============== ChromaDB 配置 ==============
COLLECTION_NAME = "knowledge_base"
PERSIST_DIRECTORY = CHROMA_DB_DIR

# ============== 检索配置 ==============
DEFAULT_TOP_K = 5          # 默认召回数量
RERANK_TOP_K = 3           # 重排后返回数量

#!/usr/bin/env python3
"""
文档导入脚本：将本地文档导入向量数据库
"""

import sys
import os
from pathlib import Path
from loguru import logger

# 添加项目根目录到 path
sys.path.insert(0, str(Path(__file__).parent.parent))

from langchain_community.document_loaders import (
    PyPDFLoader,
    UnstructuredMarkdownLoader,
    TextLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma

import config


def load_documents(directory: str):
    """
    遍历目录，加载所有支持的文档
    
    支持格式：PDF、Markdown、TXT
    """
    loader_map = {
        '.pdf': PyPDFLoader,
        '.md': UnstructuredMarkdownLoader,
        '.txt': TextLoader
    }
    
    documents = []
    path = Path(directory)
    
    for file_path in path.rglob('*'):
        if not file_path.is_file():
            continue
        
        suffix = file_path.suffix.lower()
        if suffix not in loader_map:
            continue
        
        try:
            loader_class = loader_map[suffix]
            loader = loader_class(str(file_path))
            
            # load() 返回 List[Document]
            docs = loader.load()
            
            # 为每个文档添加源文件路径元数据
            for doc in docs:
                doc.metadata['source_file'] = str(file_path)
                doc.metadata['file_type'] = suffix
            
            documents.extend(docs)
            logger.info(f"✅ 加载: {file_path.name} ({len(docs)} 页/块)")
            
        except Exception as e:
            logger.error(f"❌ 加载失败: {file_path.name}, 错误: {e}")
    
    return documents


def split_documents(documents, chunk_size=500, chunk_overlap=50):
    """
    将长文档切分成小块
    """
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
    )
    
    chunks = text_splitter.split_documents(documents)
    logger.info(f"文档切分完成：{len(documents)} 文档 → {len(chunks)} 块")
    
    return chunks


def create_vectorstore(chunks, persist_directory, embedding_model_name):
    """
    创建向量数据库
    """
    logger.info(f"正在加载 Embedding 模型: {embedding_model_name}")
    
    embeddings = HuggingFaceEmbeddings(
        model_name=embedding_model_name,
        model_kwargs={'device': config.EMBEDDING_DEVICE},
        encode_kwargs={'normalize_embeddings': True}
    )
    
    logger.info("正在计算 Embedding 并存入向量数据库...")
    
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_directory,
        collection_name=config.COLLECTION_NAME
    )
    
    logger.info(f"✅ 向量数据库创建完成，共 {vectorstore._collection.count()} 个块")
    
    return vectorstore


def main():
    """主函数"""
    # 配置日志
    logger.add(sys.stdout, colorize=True, format="{time} {message}")
    
    docs_dir = os.path.join(config.BASE_DIR, "data", "docs")
    
    # 步骤1：加载文档
    logger.info(f"开始加载文档目录: {docs_dir}")
    documents = load_documents(docs_dir)
    
    if not documents:
        logger.error("没有找到任何文档，请先将文档放入 data/docs/ 目录")
        return
    
    logger.info(f"共加载 {len(documents)} 个文档")
    
    # 步骤2：切分文档
    chunks = split_documents(
        documents,
        chunk_size=config.CHUNK_SIZE,
        chunk_overlap=config.CHUNK_OVERLAP
    )
    
    # 步骤3：创建向量数据库
    vectorstore = create_vectorstore(
        chunks,
        persist_directory=config.PERSIST_DIRECTORY,
        embedding_model_name=config.EMBEDDING_MODEL
    )
    
    logger.info("🎉 文档导入完成！")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
RAG Chain：检索 + 生成的核心逻辑
"""

from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain.chains import RetrievalQA
from langchain_community.chat_models import ChatOllama
import config


def create_retriever(vectorstore, llm=None):
    """
    创建检索器
    
    这里使用 MultiQueryRetriever 来增加召回率
    它会利用 LLM 将用户查询改写成多个版本
    """
    if llm is None:
        # 创建 Ollama LLM（用于查询改写）
        llm = ChatOllama(
            base_url=config.OLLAMA_BASE_URL,
            model=config.OLLAMA_MODEL,
            temperature=0.3
        )
    
    # 多查询检索器
    retriever = MultiQueryRetriever(
        retriever=vectorstore.as_retriever(search_kwargs={"k": config.DEFAULT_TOP_K}),
        llm=llm,
        prompt="""
你是一个AI助手。请根据原始问题，生成3个不同的搜索查询。
这些查询应该从不同角度表达同一个问题，以便检索到更多相关内容。
每个查询占一行，只输出查询，不要其他内容。

原始问题：{question}
"""
    )
    
    return retriever


def create_rag_chain(vectorstore, llm=None):
    """
    创建完整的 RAG Chain
    
    Chain 流程：
    用户问题 → 检索相关文档 → 构建 Prompt → 调用 LLM → 返回答案
    """
    
    # 1. 创建 LLM（如果没有提供）
    if llm is None:
        llm = ChatOllama(
            base_url=config.OLLAMA_BASE_URL,
            model=config.OLLAMA_MODEL,
            temperature=0.2,  # 较低温度，保证准确性
            timeout=120       # 超时时间（秒）
        )
    
    # 2. 创建检索器
    retriever = create_retriever(vectorstore, llm)
    
    # 3. 定义 Prompt 模板
    # 注意：{context} 来自检索结果，{question} 来自用户输入
    template = """基于以下上下文信息回答用户的问题。

上下文信息：
{context}

用户问题：{question}

回答要求：
1. 只使用上下文中的信息进行回答，不要编造不在上下文中的内容
2. 如果上下文信息不足以回答问题，明确说明"根据现有资料无法回答"
3. 回答要准确、简洁、有条理
4. 在回答结尾标注信息来源，格式：[来源: 文件名]

回答："""
    
    prompt = ChatPromptTemplate.from_template(template)
    
    # 4. 组装 Chain
    # 方式A：用 Runnable 方式组装（推荐，更灵活）
    chain = (
        {"context": retriever, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )
    
    return chain


def ask(chain, question):
    """
    向 RAG 系统提问
    
    参数:
        chain: RAG Chain 实例
        question: 用户问题
    
    返回:
        回答文本
    """
    response = chain.invoke(question)
    return response


# 使用示例
if __name__ == "__main__":
    # 加载向量数据库
    embeddings = HuggingFaceEmbeddings(
        model_name=config.EMBEDDING_MODEL,
        model_kwargs={'device': config.EMBEDDING_DEVICE},
        encode_kwargs={'normalize_embeddings': True}
    )
    
    vectorstore = Chroma(
        persist_directory=config.PERSIST_DIRECTORY,
        embedding_function=embeddings,
        collection_name=config.COLLECTION_NAME
    )
    
    # 创建 RAG Chain
    chain = create_rag_chain(vectorstore)
    
    # 提问测试
    question = "Python 如何安装第三方库？"
    print(f"问题: {question}")
    print(f"回答: {ask(chain, question)}")

#!/usr/bin/env python3
"""
RAG 知识库 Web 服务
"""

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Optional
import uvicorn
from loguru import logger

import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))

import config
from src.chain import create_rag_chain, create_retriever
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma


# ============== FastAPI 应用 ==============
app = FastAPI(
    title="RAG 知识库 API",
    description="基于 LangChain + ChromaDB 的本地知识库问答系统",
    version="1.0.0"
)

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境建议限制具体的域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# ============== 数据模型 ==============
class QuestionRequest(BaseModel):
    question: str = Field(..., description="用户问题", min_length=1, max_length=500)
    use_multi_query: bool = Field(default=True, description="是否使用多查询检索")
    top_k: int = Field(default=5, ge=1, le=20, description="召回数量")


class SourceDocument(BaseModel):
    content: str
    metadata: dict
    score: Optional[float] = None


class AnswerResponse(BaseModel):
    answer: str
    sources: List[SourceDocument]
    latency_ms: float


# ============== 全局变量 ==============
vectorstore = None
rag_chain = None


# ============== 启动事件 ==============
@app.on_event("startup")
async def startup_event():
    """应用启动时初始化"""
    global vectorstore, rag_chain
    
    logger.info("🚀 正在初始化 RAG 系统...")
    
    # 加载 Embedding 模型
    logger.info(f"加载 Embedding 模型: {config.EMBEDDING_MODEL}")
    embeddings = HuggingFaceEmbeddings(
        model_name=config.EMBEDDING_MODEL,
        model_kwargs={'device': config.EMBEDDING_DEVICE},
        encode_kwargs={'normalize_embeddings': True}
    )
    
    # 加载向量数据库
    logger.info(f"加载向量数据库: {config.PERSIST_DIRECTORY}")
    vectorstore = Chroma(
        persist_directory=config.PERSIST_DIRECTORY,
        embedding_function=embeddings,
        collection_name=config.COLLECTION_NAME
    )
    
    doc_count = vectorstore._collection.count()
    logger.info(f"📚 向量数据库加载完成，共 {doc_count} 个文档块")
    
    if doc_count == 0:
        logger.warning("⚠️ 向量数据库为空，请先运行 scripts/ingest.py 导入文档")
    
    # 创建 RAG Chain
    logger.info("创建 RAG Chain...")
    rag_chain = create_rag_chain(vectorstore)
    
    logger.info("✅ RAG 系统初始化完成！")


# ============== API 路由 ==============
@app.post("/api/ask", response_model=AnswerResponse)
async def ask_question(request: QuestionRequest):
    """处理问答请求"""
    import time
    start_time = time.time()
    
    try:
        logger.info(f"收到问题: {request.question}")
        
        # 检索相关文档
        retriever = vectorstore.as_retriever(
            search_kwargs={"k": request.top_k}
        )
        docs = retriever.get_relevant_documents(request.question)
        
        # 调用 RAG Chain
        answer = rag_chain.invoke(request.question)
        
        latency_ms = (time.time() - start_time) * 1000
        
        # 构建响应
        sources = [
            SourceDocument(
                content=doc.page_content,
                metadata=doc.metadata,
                score=None
            )
            for doc in docs
        ]
        
        return AnswerResponse(
            answer=answer,
            sources=sources,
            latency_ms=round(latency_ms, 2)
        )
        
    except Exception as e:
        logger.error(f"处理问题失败: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/health")
async def health_check():
    """健康检查"""
    return {
        "status": "healthy",
        "vectorstore_docs": vectorstore._collection.count() if vectorstore else 0,
        "embedding_model": config.EMBEDDING_MODEL,
        "llm_model": config.OLLAMA_MODEL
    }


@app.get("/api/stats")
async def get_stats():
    """获取系统统计"""
    if not vectorstore:
        raise HTTPException(status_code=503, detail="系统未初始化")
    
    return {
        "total_documents": vectorstore._collection.count(),
        "embedding_dimension": config.EMBEDDING_DIM,
        "collection_name": config.COLLECTION_NAME,
        "chunk_size": config.CHUNK_SIZE,
        "chunk_overlap": config.CHUNK_OVERLAP
    }


# ============== 启动服务 ==============
if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        log_level="info"
    )

# 错误信息：
# ImportError: cannot import name 'Matching' from 'chromadb.api.models'

# 原因：ChromaDB 版本不兼容

# 解决方法：指定兼容版本
pip install chromadb==0.4.22
# 或者
pip install langchain-chroma  # 专门的 LangChain + ChromaDB 集成包

# 错误信息：
# OSError: Could not find model BAAI/bge-large-zh-v1.5

# 解决方法1：使用镜像站
export HF_ENDPOINT=https://hf-mirror.com
huggingface-cli download BAAI/bge-large-zh-v1.5 --local-dir /tmp/models/bge-large-zh

# 解决方法2：手动下载后指定本地路径
embeddings = HuggingFaceEmbeddings(
    model_name="/tmp/models/bge-large-zh",
    model_kwargs={'device': 'cpu'}
)

# 解决方法3：如果网络很差，换用轻量级模型
embeddings = HuggingFaceEmbeddings(
    model_name="shibing624/text2vec-base-chinese"  # 体积小，效果还行
)

# 错误信息：
# ConnectionError: HTTPConnectionPool(host='localhost', port=11434)

# 解决方法1：确保 Ollama 已启动
ollama serve
# 验证
curl http://localhost:11434/api/tags

# 解决方法2：如果用 Docker 跑 Ollama
docker run -d -p 11434:11434 ollama/ollama:latest

# 解决方法3：先拉取模型（如果没有）
ollama pull qwen2:7b

# 错误信息：
# RuntimeError: CUDA out of memory

# 解决方法1：减小模型量化级别
# 7B 模型建议用 qwen2:7b（Int4 量化），而不是 qwen2:7b-instruct（未量化）

# 解决方法2：减少 ChromaDB 的 batch_size
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    batch_size=50  # 默认是 32，减小可以降低内存峰值
)

# 解决方法3：使用 CPU 推理
embeddings = HuggingFaceEmbeddings(
    model_kwargs={'device': 'cpu'}  # 不用 GPU
)

from langchain_community.vectorstores import FAISS

# 创建 FAISS 索引（比 ChromaDB 在大数据量下更快）
vectorstore = FAISS.from_documents(
    documents=chunks,
    embedding=embeddings
)

# 保存到本地
vectorstore.save_local("faiss_index")

# 加载
vectorstore_load = FAISS.load_local("faiss_index", embeddings)

# 当文档量很大时，使用批处理避免内存溢出

batch_size = 100
for i in range(0, len(chunks), batch_size):
    batch = chunks[i:i+batch_size]
    
    if i == 0:
        # 第一次创建 vectorstore
        vectorstore = Chroma.from_documents(
            documents=batch,
            embedding=embeddings,
            persist_directory=persist_dir
        )
    else:
        # 后续批次 add
        vectorstore.add_documents(batch)
    
    print(f"已处理 {min(i+batch_size, len(chunks))}/{len(chunks)}")

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 对于 I/O 密集型操作（如 API 调用），用异步方式加速
async def async_ask(chain, questions):
    loop = asyncio.get_event_loop()
    
    # 使用线程池并行执行
    with ThreadPoolExecutor(max_workers=5) as executor:
        tasks = [
            loop.run_in_executor(executor, lambda q: chain.invoke(q), q)
            for q in questions
        ]
        
        results = await asyncio.gather(*tasks)
    
    return results

# 使用示例
questions = ["问题1", "问题2", "问题3"]
answers = asyncio.run(async_ask(chain, questions))

# 1. 安装 Ollama（如果没有）
curl -fsSL https://ollama.com/install.sh | sh
ollama pull qwen2:7b

# 2. 准备测试文档（在 data/docs/ 目录下放几个 txt/md/pdf 文件）

# 3. 运行文档导入
cd my-rag-project
python scripts/ingest.py

# 4. 启动 Web 服务
python app.py

# 5. 测试 API
curl -X POST http://localhost:8000/api/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "你的文档相关内容"}'

# streamlit_app.py

import streamlit as st
import requests

st.title("📚 RAG 知识库问答")

# 输入框
question = st.text_input("请输入您的问题：", placeholder="例如：如何安装 Python？")

# 提交按钮
if st.button("提问"):
    if question:
        with st.spinner("思考中..."):
            try:
                response = requests.post(
                    "http://localhost:8000/api/ask",
                    json={"question": question},
                    timeout=60
                )
                
                if response.status_code == 200:
                    result = response.json()
                    
                    st.success("回答：")
                    st.write(result["answer"])
                    
                    st.info(f"⏱️ 响应时间: {result['latency_ms']:.0f}ms")
                    
                    with st.expander("📄 查看参考文档"):
                        for i, doc in enumerate(result["sources"], 1):
                            st.markdown(f"**文档 {i}**")
                            st.write(doc["content"][:200] + "...")
                            st.caption(f"来源: {doc['metadata']}")
                else:
                    st.error(f"请求失败: {response.text}")
                    
            except Exception as e:
                st.error(f"发生错误: {e}")

# 运行方式：
# streamlit run streamlit_app.py --server.port 8501