tesseract --list-langs

pip --version

pip install langchain==0.1.17
pip install langchain-community
pip install chromadb==0.4.24
pip install pypdf
pip install python-docx
pip install pillow pytesseract
pip install gradio==4.20.0
pip install websocket-client
pip install pandas openpyxl xlrd

# xinghuo_llm.py
import base64
import datetime
import hashlib
import hmac
import json
import threading
from urllib.parse import urlencode
from langchain_core.language_models.llms import LLM
from websocket import create_connection

# ====== 訊飛星火認證配置 ======
APPID = "你自己的"
APIKey = "你自己的"
APISecret = "你自己的"
SPARK_URL = "wss://你自己的"

def create_url():
    """生成帶鑑權信息的WebSocket URL"""
    host = "spark-api.xf-yun.com"
    path = "/v1/x1"
    now = datetime.datetime.now(datetime.timezone.utc)
    date = now.strftime("%a, %d %b %Y %H:%M:%S GMT")
    
    # 構造簽名原文
    signature_origin = f"host: {host}\ndate: {date}\nx-date: {date}\nGET {path} HTTP/1.1"
    
    # 生成HMAC-SHA256簽名
    signature_sha = hmac.new(
        APISecret.encode("utf-8"),
        signature_origin.encode("utf-8"),
        digestmod=hashlib.sha256,
    ).digest()
    signature_sha_base64 = base64.b64encode(signature_sha).decode("utf-8")
    
    # 生成authorization頭
    authorization_origin = (
        f'api_key="{APIKey}", algorithm="hmac-sha256", '
        f'headers="host date x-date request-line", signature="{signature_sha_base64}"'
    )
    authorization = base64.b64encode(authorization_origin.encode("utf-8")).decode("utf-8")
    
    # 拼接請求參數
    params = {
        "authorization": authorization,
        "date": date,
        "x-date": date,
        "host": host
    }
    return SPARK_URL + "?" + urlencode(params)

class SparkX1LLM(LLM):
    """訊飛星火X1-32K模型的LangChain封裝"""
    
    def _call(self, prompt: str, stop=None, **kwargs) -> str:
        """同步調用模型"""
        url = create_url()
        result = []
        
        try:
            # 創建WebSocket連接
            ws = create_connection(url)
            
            # 構造請求payload
            payload = {
                "header": {"app_id": APPID},
                "parameter": {
                    "chat": {
                        "domain": "x1",  # 必須是x1（對應/v1/x1接口）
                        "temperature": 0.5,
                        "max_tokens": 2048
                    }
                },
                "payload": {
                    "message": {
                        "text": [{"role": "user", "content": prompt}]
                    }
                }
            }
            
            # 發送請求
            ws.send(json.dumps(payload))
            
            # 接收響應
            while True:
                message = ws.recv()
                data = json.loads(message)
                
                # 檢查錯誤
                header = data.get("header", {})
                code = header.get("code")
                if code != 0:
                    error_msg = data.get("message", "未知錯誤")
                    raise RuntimeError(f"訊飛API錯誤[{code}]: {error_msg}")
                
                # 提取文本內容
                choices = data.get("payload", {}).get("choices", {})
                text_content = choices.get("text", [{}])[0].get("content", "")
                status = choices.get("status")
                
                result.append(text_content)
                
                # 檢查是否結束
                if status == 2:
                    break
            
            # 關閉連接
            ws.close()
            return "".join(result)
        
        except Exception as e:
            raise RuntimeError(f"WebSocket調用失敗: {str(e)}")

    @property
    def _llm_type(self) -> str:
        return "spark-x1"

# document_loader.py
import os
import pandas as pd
from langchain_core.documents import Document
from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader
from PIL import Image
import pytesseract

def excel_to_text(filepath: str) -> str:
    """將Excel文件轉換為結構化文本"""
    try:
        xls = pd.ExcelFile(filepath)
        full_text = f"文件: {os.path.basename(filepath)}\n\n"
        
        for sheet_name in xls.sheet_names:
            # 讀取工作表，空值轉為空字符串
            df = pd.read_excel(filepath, sheet_name=sheet_name, dtype=str).fillna("")
            if df.empty:
                continue
                
            full_text += f"工作表: {sheet_name}\n"
            
            # 獲取列名（假設第一行為標題）
            headers = df.columns.tolist()
            
            # 遍歷每一行
            for idx, row in df.iterrows():
                row_items = []
                for col in headers:
                    val = str(row[col]).strip()
                    if val:  # 只添加非空值
                        row_items.append(f"{col}={val}")
                if row_items:  # 只添加非空行
                    full_text += f"第{idx+1}行: " + "; ".join(row_items) + "\n"
            full_text += "\n"
            
        return full_text.strip()
    except Exception as e:
        return f"Excel文件解析失敗 ({os.path.basename(filepath)}): {str(e)}"

def load_documents_from_folder(folder_path: str):
    """從文件夾加載所有支持類型的文檔"""
    documents = []
    
    # 遍歷文件夾中所有文件
    for filename in os.listdir(folder_path):
        filepath = os.path.join(folder_path, filename)
        try:
            # PDF文件處理
            if filename.lower().endswith(".pdf"):
                loader = PyPDFLoader(filepath)
                docs = loader.load()
                for doc in docs:
                    doc.metadata["source"] = filename
                documents.extend(docs)
            
            # Word文件處理
            elif filename.lower().endswith((".docx", ".doc")):
                loader = Docx2txtLoader(filepath)
                docs = loader.load()
                for doc in docs:
                    doc.metadata["source"] = filename
                documents.extend(docs)
            
            # 圖片文件處理
            elif filename.lower().endswith((".png", ".jpg", ".jpeg", ".bmp")):
                text = pytesseract.image_to_string(
                    Image.open(filepath), 
                    lang="chi_sim+eng"
                )
                doc = Document(
                    page_content=text,
                    metadata={"source": filename}
                )
                documents.append(doc)
            
            # Excel文件處理
            elif filename.lower().endswith((".xlsx", ".xls")):
                text = excel_to_text(filepath)
                doc = Document(
                    page_content=text,
                    metadata={"source": filename}
                )
                documents.append(doc)
                
        except Exception as e:
            print(f"⚠️ 加載 {filename} 失敗: {str(e)}")
    
    print(f"✅ 共加載 {len(documents)} 個文檔片段")
    return documents

# rag_pipeline.py
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from .xinghuo_llm import SparkX1LLM
from .document_loader import load_documents_from_folder
import os

class LocalRAG:
    def __init__(self, knowledge_dir="knowledge_base", persist_dir="vector_db"):
        self.knowledge_dir = knowledge_dir
        self.persist_dir = persist_dir
        self.vector_db = None
        self.qa_chain = None
        
        # 創建知識庫目錄（如果不存在）
        os.makedirs(knowledge_dir, exist_ok=True)
        
        # 初始化嵌入模型（完全離線）
        self.embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/all-MiniLM-L6-v2",
            model_kwargs={'device': 'cpu'}
        )
    
    def rebuild_knowledge_base(self):
        """重建知識庫（加載文檔 + 創建向量庫）"""
        print("🔄 正在加載文檔...")
        documents = load_documents_from_folder(self.knowledge_dir)
        
        if not documents:
            raise ValueError(f"知識庫目錄 {self.knowledge_dir} 中沒有找到有效文檔!")
        
        print("✂️ 正在分割文本...")
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
        )
        splits = text_splitter.split_documents(documents)
        
        print("🧠 正在創建向量數據庫...")
        self.vector_db = Chroma.from_documents(
            documents=splits,
            embedding=self.embeddings,
            persist_directory=self.persist_dir
        )
        self.vector_db.persist()  # 保存到磁盤
        
        print("🔗 正在初始化問答鏈...")
        llm = SparkX1LLM()  # 使用訊飛星火
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=llm,
            retriever=self.vector_db.as_retriever(
                search_type="similarity",
                search_kwargs={"k": 3}
            ),
            return_source_documents=True
        )
        print("✅ 知識庫重建完成!")
    
    def query(self, question: str):
        """執行問答"""
        if not self.qa_chain:
            raise RuntimeError("請先重建知識庫！")
        
        print(f"\n❓ 用户問題: {question}")
        response = self.qa_chain.invoke({"query": question})
        
        # 提取引用來源
        sources = []
        for doc in response["source_documents"]:
            source = doc.metadata["source"]
            if source not in sources:
                sources.append(source)
        
        return {
            "answer": response["result"],
            "sources": sources
        }

# main.py
import gradio as gr
from rag_pipeline import LocalRAG
import os
import shutil

# 初始化RAG系統
rag_system = LocalRAG()

def upload_files(files):
    """處理上傳的文件"""
    saved_files = []
    for file in files:
        dest_path = os.path.join(rag_system.knowledge_dir, os.path.basename(file.name))
        shutil.copy(file.name, dest_path)
        saved_files.append(os.path.basename(file.name))
    return f"✅ 已上傳 {len(saved_files)} 個文件: {', '.join(saved_files)}"

def rebuild_knowledge_base():
    """重建知識庫"""
    try:
        rag_system.rebuild_knowledge_base()
        return "✅ 知識庫重建成功！現在可以提問了。"
    except Exception as e:
        return f"❌ 重建失敗: {str(e)}"

def answer_question(question):
    """回答問題"""
    try:
        if not rag_system.qa_chain:
            return "請先重建知識庫！", []
        
        result = rag_system.query(question)
        sources = "\n".join([f"• {source}" for source in result["sources"]])
        return result["answer"], sources
    except Exception as e:
        return f"生成回答時出錯: {str(e)}", ""

# 創建Gradio界面
with gr.Blocks(title="本地知識助手 - 訊飛星火版") as demo:
    gr.Markdown("# 🧠 本地知識助手 (訊飛星火版)")
    gr.Markdown("### 支持 PDF / Word / Excel / 圖片文件 · 完全離線運行")
    
    with gr.Tab("上傳文件"):
        file_upload = gr.Files(label="選擇文件（支持 PDF/Word/Excel/圖片）")
        upload_btn = gr.Button("上傳文件")
        upload_status = gr.Textbox(label="上傳狀態", interactive=False)
        upload_btn.click(upload_files, inputs=file_upload, outputs=upload_status)
        
        rebuild_btn = gr.Button("重建知識庫", variant="primary")
        rebuild_status = gr.Textbox(label="重建狀態", interactive=False)
        rebuild_btn.click(rebuild_knowledge_base, outputs=rebuild_status)
    
    with gr.Tab("問答"):
        question = gr.Textbox(label="輸入你的問題", placeholder="例如：文檔中提到的關鍵技術是什麼？")
        ask_btn = gr.Button("獲取答案", variant="primary")
        
        with gr.Row():
            answer = gr.Textbox(label="回答", lines=10)
            sources = gr.Textbox(label="參考來源", lines=5)
        
        ask_btn.click(answer_question, inputs=question, outputs=[answer, sources])

# 啓動應用
if __name__ == "__main__":
    demo.launch(server_name="127.0.0.1", server_port=7860)