使用讯飞星火 Spark X1-32K 打造本地知识助手

（Python 3.10 + PyCharm 环境 · 无需显卡 · 支持 PDF/Word/Excel/图片 · 含图形界面、windows10 TLSC）

第一章：环境准备（30分钟）

步骤 1：安装 Tesseract OCR（10分钟）

•  下载安装包

• 打开浏览器访问：https://github.com/UB-Mannheim/tesseract/wiki

• 点击 tesseract-ocr-w64-setup-5.3.0.20221222.exe 下载（64位系统）  

•  运行安装

• 双击下载的安装包 •

在 "Select Additional Language Data" 界面：

• 勾选 Chinese (Simplified)

• 勾选 English  

• 其他选项保持默认，点击 "Next" 直到完成  

•  配置环境变量

• 按 Win + R 输入 sysdm.cpl 回车

• 切换到 "高级" 选项卡 → 点击 "环境变量"

• 在 "系统变量" 区域找到 Path → 点击 "编辑"

• 点击 "新建" → 输入：
C:\Program Files\Tesseract-OCR（按照实际路径修改）

• 点击 "确定" 保存所有窗口

验证安装

• 按 Win + R 输入 cmd 回车

• 在命令行输入：

tesseract --list-langs

·确认输出中包含 chi_sim 和 eng

步骤 2：创建项目虚拟环境（5分钟）

•  打开 PyCharm

• 启动 PyCharm

• 顶部菜单：File → New Project

• 位置（Location）：G:\PythonProject\spark_local_rag（建议修改为你自己的路径）

• Python 解释器：选择已安装的 Python 3.10

• 勾选 ✔️ Create a virtual environment

• 虚拟环境名称：spark_env • 点击 "Create"  

•  激活虚拟环境

• 在 PyCharm 底部找到 Terminal 标签页（或按 Alt + F12）

• 输入命令验证：

pip --version

确认输出路径包含 spark_env（例如：G:\PythonProject\spark_local_rag\spark_env\Scripts\pip.exe）

步骤 3：安装依赖包（15分钟）

在 Terminal 中逐行执行（复制粘贴每行后按回车）：

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

第二章：创建项目结构（5分钟）

在 PyCharm 左侧 Project 面板中：

• 右键点击项目根目录 → New → Directory

• 名称：knowledge_base（存放知识文件）  

• 右键点击项目根目录 → New → Python File
创建以下 4 个文件：

• xinghuo_llm.py

• document_loader.py

• rag_pipeline.py

• main.py    

第三章：编写核心代码（逐行操作）

文件 1：xinghuo_llm.py（讯飞星火集成）

1. 在 PyCharm 中打开 xinghuo_llm.py
2. 复制以下完整代码到文件中：

# 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)# 构造请求payloadpayload = {"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)}")@propertydef _llm_type(self) -> str:return "spark-x1"

文件 2：document_loader.py（多格式文件加载）

• 打开 document_loader.py

• 复制以下完整代码：

# 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 pytesseractdef 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:continuefull_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"] = filenamedocuments.extend(docs)# Word文件处理elif filename.lower().endswith((".docx", ".doc")):loader = Docx2txtLoader(filepath)docs = loader.load()for doc in docs:doc.metadata["source"] = filenamedocuments.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

文件 3：rag_pipeline.py（RAG核心逻辑）

1. 打开 rag_pipeline.py
2. 复制以下完整代码：

# 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 osclass LocalRAG:def __init__(self, knowledge_dir="knowledge_base", persist_dir="vector_db"):self.knowledge_dir = knowledge_dirself.persist_dir = persist_dirself.vector_db = Noneself.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}

文件 4：main.py（图形界面）

1. 打开 main.py
2. 复制以下完整代码：

# 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"], sourcesexcept 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)

第四章：测试运行（10分钟）

步骤 1：准备测试文件

1. 在项目目录中找到 knowledge_base 文件夹
2. 放入以下测试文件（可自行创建）：
   • 测试文档.pdf（含一段中文文字）
   • 员工信息.xlsx（内容如下）：

     姓名	部门	工龄
     张三	技术部	3年
     李四	市场部	5年

步骤 2：启动应用

• 在 PyCharm 中右键点击 main.py

• 选择 Run 'main'

• 等待控制台出现：

Running on local URL: http://127.0.0.1:7860

步骤 3：使用图形界面

1. 自动打开浏览器（或手动访问 http://127.0.0.1:7860）
2. 切换到 "上传文件" 标签页：
   • 点击 "选择文件" 按钮
   • 选择 测试文档.pdf 和 员工信息.xlsx
   • 点击 "上传文件" → 显示上传成功
   • 点击 "重建知识库" → 等待 1-2 分钟（首次较慢）

1. 切换到 "问答" 标签页：
   • 输入问题：张三在哪个部门？
   • 点击 "获取答案"
   • 预期结果：
     • 回答区域：张三在技术部工作。
     • 参考来源：• 员工信息.xlsx