搭建环境

安装好的工具

✅ Python 3.11.9 

✅ pip 24.0 

✅ 所有库都能装

✅ qwen2:7b 模型正常

✅ ollama + langchain + sentence-transformers + faiss-cpu + python-docx + PyPDF2

✅ Ollama 安装，已运行，qwen2:7b 模型已下载

1.创建项目，添加依赖

# 批量安装所有需要的库
pip install langchain langchain-community chromadb gradio sentence-transformers python-docx pypdf tiktoken

# 如果速度慢，使用清华源
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple langchain langchain-community chromadb gradio sentence-transformers python-docx pypdf tiktoken

2. 创建必要的文件夹
New-Item -ItemType Directory -Force -Path src, web, data\raw_docs, data\vector_db, data\uploads

3.写好项目内容

"""
知识库核心类 - RAG检索增强生成（简化版，兼容新版本）
"""
import os
from pathlib import Path
from typing import List, Dict, Any

# LangChain组件
from langchain_community.document_loaders import (
    TextLoader,
    PyPDFLoader,
    UnstructuredWordDocumentLoader,
    UnstructuredMarkdownLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.llms import Ollama
from langchain_core.prompts import PromptTemplate

# 导入配置
from src.config import (
    EMBEDDING_MODEL, LLM_MODEL, VECTOR_DB_DIR,
    CHUNK_SIZE, CHUNK_OVERLAP, TOP_K_RESULTS,
    SUPPORTED_EXTENSIONS
)


class LocalKnowledgeBase:
    """本地知识库管理系统"""

    def __init__(self):
        """初始化知识库"""
        print("🚀 初始化知识库系统...")

        # 1. 加载嵌入模型
        print(f"📥 加载嵌入模型: {EMBEDDING_MODEL}")
        print("   (首次使用需要下载模型，约470MB，请耐心等待...)")
        self.embeddings = OllamaEmbeddings(
            model=EMBEDDING_MODEL,
            base_url="http://localhost:11434"
        )
        print("   ✅ 嵌入模型加载完成")

        # 2. 连接Ollama
        print(f"🤖 连接Ollama模型: {LLM_MODEL}")
        self.llm = Ollama(
            model=LLM_MODEL,
            temperature=0.1,
            base_url="http://localhost:11434"
        )
        print("   ✅ Ollama连接成功")

        # 3. 初始化向量数据库
        self.vector_store = None
        self.db_path = str(VECTOR_DB_DIR)

        # 4. 如果已有数据库，自动加载
        if os.path.exists(self.db_path) and os.listdir(self.db_path):
            try:
                print(f"📚 加载已有向量数据库...")
                self.vector_store = Chroma(
                    persist_directory=self.db_path,
                    embedding_function=self.embeddings
                )
                print("   ✅ 数据库加载成功")
            except Exception as e:
                print(f"   ⚠️ 数据库加载失败: {e}")

        print("✅ 知识库初始化完成！\n")

    def add_document(self, file_path: str) -> bool:
        """添加单个文档到知识库"""
        if not os.path.exists(file_path):
            print(f"❌ 文件不存在: {file_path}")
            return False

        file_ext = Path(file_path).suffix.lower()
        if file_ext not in SUPPORTED_EXTENSIONS:
            print(f"❌ 不支持的文件类型: {file_ext}")
            print(f"   支持的类型: {', '.join(SUPPORTED_EXTENSIONS)}")
            return False

        print(f"\n📄 添加文档: {Path(file_path).name}")

        try:
            if file_ext == '.txt':
                loader = TextLoader(file_path, encoding='utf-8')
            elif file_ext == '.md':
                loader = UnstructuredMarkdownLoader(file_path)
            elif file_ext == '.pdf':
                loader = PyPDFLoader(file_path)
            elif file_ext == '.docx':
                loader = UnstructuredWordDocumentLoader(file_path)
            else:
                return False

            # 加载文档
            documents = loader.load()
            print(f"   📖 加载了 {len(documents)} 个文档段落")

            # 文本分块
            text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=CHUNK_SIZE,
                chunk_overlap=CHUNK_OVERLAP,
                separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
            )
            splits = text_splitter.split_documents(documents)
            print(f"   ✂️  分为 {len(splits)} 个文本块")

            # 添加到向量数据库
            if self.vector_store:
                self.vector_store.add_documents(splits)
                self.vector_store.persist()
            else:
                self.vector_store = Chroma.from_documents(
                    documents=splits,
                    embedding=self.embeddings,
                    persist_directory=self.db_path
                )
                self.vector_store.persist()

            print(f"   ✅ 添加成功！")
            return True

        except Exception as e:
            print(f"   ❌ 添加失败: {e}")
            return False

    def add_folder(self, folder_path: str) -> Dict[str, int]:
        """批量添加文件夹中的所有文档"""
        if not os.path.exists(folder_path):
            print(f"❌ 文件夹不存在: {folder_path}")
            return {'success': 0, 'failed': 0}

        print(f"\n📁 批量导入文件夹: {folder_path}")

        stats = {'success': 0, 'failed': 0}

        for file in os.listdir(folder_path):
            file_path = os.path.join(folder_path, file)
            if os.path.isfile(file_path):
                file_ext = Path(file_path).suffix.lower()
                if file_ext in SUPPORTED_EXTENSIONS:
                    if self.add_document(file_path):
                        stats['success'] += 1
                    else:
                        stats['failed'] += 1

        print(f"\n📊 批量导入完成: 成功 {stats['success']} 个, 失败 {stats['failed']} 个")
        return stats

    def query(self, question: str, top_k: int = None) -> Dict[str, Any]:
        """向知识库提问 - 使用直接调用方式"""
        if not self.vector_store:
            return {
                "answer": "❌ 知识库为空，请先添加文档！\n\n提示：将文档放到 data/raw_docs/ 目录，然后运行 add_folder()",
                "sources": []
            }

        if not question:
            return {"answer": "❌ 请输入问题", "sources": []}

        k = top_k or TOP_K_RESULTS

        print(f"\n💬 处理问题: {question[:50]}...")

        # 1. 检索相关文档
        docs = self.vector_store.similarity_search(question, k=k)

        if not docs:
            return {
                "answer": "❌ 未找到相关内容，请尝试其他问题或添加更多文档。",
                "sources": []
            }

        print(f"   📚 检索到 {len(docs)} 个相关片段")

        # 2. 构建上下文
        context = "\n\n---\n\n".join([doc.page_content for doc in docs])

        # 3. 构建 Prompt
        prompt_template = """你是一个专业的知识库助手。请根据以下参考资料回答用户的问题。

重要规则：
1. 必须基于参考资料回答，不要编造信息
2. 如果参考资料中没有相关信息，请明确说"资料中未找到相关内容"
3. 回答要简洁、准确、有条理

参考资料：
{context}

用户问题：{question}

请回答："""

        prompt = PromptTemplate(
            template=prompt_template,
            input_variables=["context", "question"]
        )

        # 4. 直接调用 LLM
        formatted_prompt = prompt.format(context=context, question=question)
        answer = self.llm.invoke(formatted_prompt)

        # 5. 整理来源
        sources = []
        for i, doc in enumerate(docs[:3], 1):
            content = doc.page_content[:200].replace('\n', ' ')
            sources.append(f"来源{i}: {content}...")

        return {
            "answer": answer,
            "sources": sources
        }

    def get_stats(self) -> Dict[str, Any]:
        """获取知识库统计信息"""
        return {
            "has_database": self.vector_store is not None,
            "db_path": self.db_path,
            "embedding_model": EMBEDDING_MODEL,
            "llm_model": LLM_MODEL,
            "status": "ready" if self.vector_store else "empty"
        }

    def clear_database(self, confirm: bool = False) -> bool:
        """清空知识库"""
        if not confirm:
            print("⚠️ 请设置 confirm=True 以确认清空数据库")
            return False

        try:
            import shutil
            if os.path.exists(self.db_path):
                shutil.rmtree(self.db_path)
                os.makedirs(self.db_path, exist_ok=True)
                self.vector_store = None
                print("✅ 知识库已清空")
                return True
        except Exception as e:
            print(f"❌ 清空失败: {e}")
            return False

config

"""
项目配置文件
"""
import os
from pathlib import Path

# 项目根目录（自动检测）
BASE_DIR = Path(__file__).parent.parent

# 数据目录
DATA_DIR = BASE_DIR / "data"
RAW_DOCS_DIR = DATA_DIR / "raw_docs"      # 存放原始文档
VECTOR_DB_DIR = DATA_DIR / "vector_db"    # 向量数据库
UPLOADS_DIR = DATA_DIR / "uploads"        # 上传文件临时目录

# 模型配置
EMBEDDING_MODEL = "nomic-embed-text"
LLM_MODEL = "qwen2:7b"
OLLAMA_BASE_URL = "http://localhost:11434"

# 向量数据库配置
CHUNK_SIZE = 500        # 文本分块大小（字符数）
CHUNK_OVERLAP = 100     # 分块重叠大小（保持语义连贯）
TOP_K_RESULTS = 3       # 检索相关文档数量

# Web界面配置
WEB_HOST = "127.0.0.1"
WEB_PORT = 7860

# 支持的文件类型
SUPPORTED_EXTENSIONS = ['.txt', '.md', '.pdf', '.docx']

# 自动创建必要的目录
for dir_path in [DATA_DIR, RAW_DOCS_DIR, VECTOR_DB_DIR, UPLOADS_DIR]:
    dir_path.mkdir(parents=True, exist_ok=True)

# 打印配置信息
print("="*50)
print("📁 项目配置信息")
print("="*50)
print(f"项目目录: {BASE_DIR}")
print(f"文档目录: {RAW_DOCS_DIR}")
print(f"数据库目录: {VECTOR_DB_DIR}")
print(f"上传目录: {UPLOADS_DIR}")
print(f"嵌入模型: {EMBEDDING_MODEL}")
print(f"LLM模型: {LLM_MODEL}")
print("="*50 + "\n")

app.py

"""
Gradio Web界面 - 知识库问答系统（兼容 Gradio 6.x）
"""
import sys
from pathlib import Path

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

import gradio as gr
from src.knowledge_base import LocalKnowledgeBase
from src.config import RAW_DOCS_DIR, SUPPORTED_EXTENSIONS

# 初始化知识库
kb = LocalKnowledgeBase()


def ask_question(message: str, history: list) -> str:
    """
    处理用户提问
    """
    if not message or not message.strip():
        return "请输入问题"

    # 调用知识库
    result = kb.query(message.strip())

    # 格式化回答
    response = f"🤖 **回答：**\n{result['answer']}\n\n"

    if result['sources']:
        response += f"📚 **参考来源：**\n"
        for source in result['sources']:
            response += f"• {source}\n\n"

    return response


def batch_import() -> str:
    """
    批量导入文档
    """
    if not RAW_DOCS_DIR.exists():
        return f"❌ 文档目录不存在: {RAW_DOCS_DIR}"

    # 检查是否有文档
    files = [f for f in RAW_DOCS_DIR.iterdir() if f.suffix.lower() in SUPPORTED_EXTENSIONS]

    if not files:
        return f"❌ 文档目录为空\n请将文档放到: {RAW_DOCS_DIR}\n\n支持的格式: {', '.join(SUPPORTED_EXTENSIONS)}"

    # 批量导入
    stats = kb.add_folder(str(RAW_DOCS_DIR))

    return f"""✅ **批量导入完成！**

📊 统计信息：
- 成功导入: {stats['success']} 个文档
- 失败: {stats['failed']} 个文档
- 文档目录: {RAW_DOCS_DIR}

💡 现在可以在「问答」标签页提问了！"""


def get_status() -> str:
    """
    获取系统状态
    """
    stats = kb.get_stats()

    status_text = f"""
### 📊 系统状态

| 项目 | 状态 |
|------|------|
| 知识库状态 | {'✅ 已就绪' if stats['has_database'] else '⚠️ 未初始化'} |
| 嵌入模型 | {stats['embedding_model']} |
| LLM模型 | {stats['llm_model']} |
| 数据库路径 | {stats['db_path']} |
| 文档目录 | {RAW_DOCS_DIR} |
| 支持格式 | {', '.join(SUPPORTED_EXTENSIONS)} |
"""

    return status_text


# 创建Gradio界面（兼容 Gradio 6.x）
with gr.Blocks(title="个人知识库AI助手") as demo:
    gr.Markdown("""
    # 📚 个人知识库AI助手
    
    基于 **RAG（检索增强生成）** 技术的本地知识库系统，使用 **Qwen2:7B** 大模型。
    
    ### ✨ 特性
    - 🔒 完全本地运行，数据安全
    - 📄 支持 TXT、PDF、Word、Markdown 文档
    - 💡 智能检索 + 精准回答
    - 🎯 提供答案来源引用
    """)

    with gr.Tab("💬 问答"):
        chatbot = gr.Chatbot(
            height=500,
            label="对话历史"
        )

        with gr.Row():
            msg = gr.Textbox(
                label="输入问题",
                placeholder="例如：这篇文章主要讲了什么？",
                scale=4
            )
            clear = gr.Button("🗑️ 清空对话", scale=1)

        gr.Examples(
            examples=[
                "请总结一下文档的主要内容",
                "文档中提到了哪些重要观点？",
                "根据文档，请解释一下核心概念",
                "文档的作者或来源是什么？"
            ],
            inputs=msg,
            label="💡 示例问题"
        )

    with gr.Tab("📁 文档管理"):
        gr.Markdown(f"""
        ### 📂 文档目录
        文档存放路径: `{RAW_DOCS_DIR}`
        
        #### 支持的文档格式
        - 📝 TXT 文本文件
        - 📘 Markdown 文档
        - 📕 PDF 文件
        - 📗 Word 文档 (.docx)
        
        #### 使用步骤
        1. 将文档复制到上述目录
        2. 点击下方「批量导入」按钮
        3. 等待处理完成
        4. 切换到「问答」标签页提问
        """)

        import_btn = gr.Button("🔄 批量导入文档", variant="primary")
        import_output = gr.Textbox(label="导入结果", lines=10)

        import_btn.click(batch_import, outputs=import_output)

    with gr.Tab("📊 状态"):
        status_btn = gr.Button("🔄 刷新状态")
        status_output = gr.Markdown(get_status())
        status_btn.click(get_status, outputs=status_output)

    # 问答功能
    def respond(message, chat_history):
        bot_message = ask_question(message, chat_history)
        chat_history.append((message, bot_message))
        return "", chat_history

    msg.submit(respond, [msg, chatbot], [msg, chatbot])
    clear.click(lambda: None, None, chatbot, queue=False)


# 启动配置
if __name__ == "__main__":
    print("\n" + "="*60)
    print("🚀 启动知识库Web界面")
    print("="*60)
    print(f"📍 本地访问: http://127.0.0.1:7860")
    print("📋 使用步骤:")
    print("   1. 先到「文档管理」标签页导入文档")
    print("   2. 再到「问答」标签页提问")
    print("   3. 按 Ctrl+C 停止服务")
    print("="*60 + "\n")

    demo.launch(
        server_name="127.0.0.1",
        server_port=7860,
        share=False
    )

run.py

#!/usr/bin/env python
"""
知识库系统统一启动入口
"""
import sys
import subprocess
from pathlib import Path

# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent))


def check_ollama():
    """检查Ollama服务是否运行"""
    import requests
    try:
        response = requests.get("http://localhost:11434/api/tags", timeout=2)
        if response.status_code == 200:
            return True
    except:
        pass
    return False


def check_model():
    """检查qwen2模型是否已下载"""
    import requests
    try:
        response = requests.get("http://localhost:11434/api/tags", timeout=2)
        if response.status_code == 200:
            models = response.json()
            for model in models.get('models', []):
                if 'qwen2:7b' in model.get('name', ''):
                    return True
    except:
        pass
    return False


def main():
    print("=" * 60)
    print("📚 个人知识库系统")
    print("=" * 60)

    # 1. 检查Ollama服务
    print("\n🔍 检查Ollama服务...")
    if not check_ollama():
        print("❌ Ollama服务未运行！")
        print("\n请按以下步骤操作：")
        print("1. 打开新的终端窗口")
        print("2. 运行命令: ollama serve")
        print("3. 等待Ollama启动完成")
        print("4. 回到本窗口按任意键继续...")
        input()

        # 重新检查
        if not check_ollama():
            print("❌ Ollama仍未启动，请手动启动后再试")
            return

    print("✅ Ollama服务运行中")

    # 2. 检查模型
    print("\n🔍 检查qwen2:7b模型...")
    if not check_model():
        print("⚠️ qwen2:7b模型未下载")
        print("\n正在下载模型（首次下载约4GB，请耐心等待）...")
        try:
            subprocess.run(["ollama", "pull", "qwen2:7b"], check=True)
            print("✅ 模型下载完成")
        except:
            print("❌ 模型下载失败，请手动运行: ollama pull qwen2:7b")
            return
    else:
        print("✅ qwen2:7b模型已就绪")

    # 3. 启动Web界面
    print("\n" + "=" * 60)
    print("🚀 启动Web界面...")
    print("=" * 60)

    try:
        from web.app import demo
        demo.launch(server_name="127.0.0.1", server_port=7860, share=False)
    except KeyboardInterrupt:
        print("\n\n👋 已停止服务")
    except Exception as e:
        print(f"\n❌ 启动失败: {e}")
        print("\n请检查:")
        print("1. 是否已安装所有依赖: pip install -r requirements.txt")
        print("2. 虚拟环境是否已激活")
        print("3. Ollama是否正常运行")


if __name__ == "__main__":
    main()

4.搭建成功