目录
- 一、知识文档的准备
- 二、OCR转换
- 三、分词处理
- 四、创建向量数据库
- 五、初始化语言聊天模型
- 1.prompt
- 2.检索链
- 3.对话
- 完整代码
知识文档的准备：首先需要准备知识文档，这些文档可以是多种格式，如Word、TXT、PDF等。使用文档加载器或多模态模型（如OCR技术）将这些文档转换为可理解的纯文本数据。对于长篇文档，还需进行文档切片，以便更高效地处理和检索信息。
嵌入模型：将文本转换为向量形式，以便通过计算向量之间的差异来识别语义上相似的句子。常见的嵌入模型包括Word2Vec、BERT和GPT系列等。
向量数据库：将嵌入后的向量数据存储在向量数据库中，以便进行高效的相似性搜索。
查询检索：当用户提出查询时，系统会将查询通过嵌入模型转换为向量，然后在向量数据库中进行相似性搜索，找到与查询最相关的文档或信息。
生成回答：将检索到的相关信息与用户的查询结合，生成最终的回答。生成模型会利用检索到的信息作为上下文输入，并结合大语言模型来生成文本内容。
这里的嵌入模型用的是本地部署的ollama，也可以使用openai，但是连接不太稳定，还有阿里云的通义千问。
一、知识文档的准备
知识库中存放pdf等类型的文档，准备后面转换为txt文本
二、OCR转换
OCR转换会将PDF、图片这些信息提取得到TXT文本。数据质量的好坏直接影响着后面模型对话效果。因此PDF解析选用的工具必须精确且合适。
在这个例子中，我是事先将PDF用MinerU解析成markdown形式了
三、分词处理
文本分词处理（Tokenization）是自然语言处理（NLP）中的一个重要步骤，其目的是将连续的文本字符串分割成有意义的单元，这些单元通常被称为“词”或“标记”（tokens）。分词处理是文本分析的基础，因为大多数NLP任务都需要在词级别上进行操作，例如文本分类、情感分析、机器翻译等。
中文分词使用了jieba库
jieba 是一个非常流行的 Python 中文分词库，主要用于将中文文本切分成单个词语。它支持多种分词模式，并提供了丰富的功能来满足不同的自然语言处理需求。
主要功能和特点：
分词模式：
精确模式：将文本精确地切分成单个词语，适合用于文本分析。
全模式：将文本中所有可能的词语都扫描出来，速度非常快，但可能存在冗余数据。
搜索引擎模式：在精确模式的基础上，对长词再次进行切分，提高召回率，适合用于搜索引擎分词。
自定义词典：用户可以通过自定义词典来增加新词，以提高分词的准确性。
关键词提取：jieba 提供了基于 TF-IDF 算法的关键词提取功能，可以从文本中提取出最重要的词。
词性标注：通过 jieba.posseg 模块，可以在分词的同时获取词性信息。
并行分词：支持并行分词，以提高分词速度
。
四、创建向量数据库
def create_vector_store(tokenized_texts: List[List[str]], embeddings_model: OllamaEmbeddings) -> FAISS:
"""将分词后的文本创建向量库"""
try:
# 将分词列表转换回文本
processed_texts = [' '.join(tokens) for tokens in tokenized_texts]
# 批量处理优化
batch_size = 100 # 可以根据实际情况调整
vectors = []
# # 如果有 GPU
# if FAISS.get_num_gpus():
# res = FAISS.StandardGpuResources()
# index = FAISS.index_cpu_to_gpu(res, 0, index)
for i in tqdm(range(0, len(processed_texts), batch_size), desc="创建向量数据库"):
batch = processed_texts[i:i + batch_size]
# 批量创建向量
vector_store = FAISS.from_texts(
texts=batch,
embedding=embeddings_model,
metadatas=[{"index": j} for j in range(i, i + len(batch))] # 添加元数据以追踪文档
)
vectors.append(vector_store)
# 如果有多个批次，合并它们
if len(vectors) > 1:
final_vector_store = vectors[0]
for v in vectors[1:]:
final_vector_store.merge_from(v)
else:
final_vector_store = vectors[0]
# 保存向量库到本地
final_vector_store.save_local("resume_vectors")
return final_vector_store
except Exception as e:
print(f"创建向量库时发生错误: {str(e)}")
raise
五、初始化语言聊天模型
刚刚就是制作了向量数据库，这是大模型的第一步，下面还需要有明确的提示词prompt
1.prompt
2.检索链
检索链（Retrieval Chain）是一种在信息检索和自然语言处理中使用的技术流程，主要用于从大规模数据集中高效地找到与用户查询最相关的信息片段或文档
3.对话
使用一个while循环始终在对话中
完整代码
import os
import jieba
import re
from typing import List
import pdf
# from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.llms import Ollama
from tqdm import tqdm
from loguru import logger
from magic_pdf.data.data_reader_writer import FileBasedDataWriter
from magic_pdf.pipe.UNIPipe import UNIPipe
import nltk
# 下载punkt
nltk.download('punkt')
# 设置 OpenAI API 密钥
# os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")
# 创建 OpenAI API 密钥
api_key = "sk-xxx"
os.environ["OPENAI_API_KEY"] = api_key
# 定义简历文件夹路径
resume_folder = "./data/demo"
# 读取 PDF 文件并提取文本
# def extract_text_from_pdfs(folder_path):
# texts = []
# for filename in os.listdir(folder_path):
# if filename.endswith(".pdf"):
# with open(os.path.join(folder_path, filename), "rb") as file:
# reader = PyPDF2.PdfReader(file)
# text = ""
# for page in reader.pages:
# text += page.extract_text()
# texts.append(text)
# return texts
# 读取markdown文件并提取文本
def extract_text_from_markdown(folder_path):
texts = []
for filename in os.listdir(folder_path):
if filename.endswith(".md"):
with open(os.path.join(folder_path, filename), "r", encoding="utf-8") as file:
text = file.read()
texts.append(text)
return texts
def clean_text(text: str) -> str:
"""清理文本，移除特殊字符和多余的空白"""
# 替换多个空白字符为单个空格
text = re.sub(r'\s+', ' ', text)
# 移除特殊字符，保留中文、英文、数字和基本标点
text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9.,!?;:，。！？；：]', ' ', text)
return text.strip()
def tokenize_texts(texts: List[str]) -> List[List[str]]:
"""对文本进行分词处理""
Args:
texts: 要处理的文本列表
Returns:
处理后的分词列表
"""
tokenized_texts = []
for text in texts:
# 清理文本
cleaned_text = clean_text(text)
# 分别处理中文和英文
words = []
# 使用jieba进行中文分词
segments = jieba.cut(cleaned_text)
# 过滤空字符串和纯空白字符
words = [word for word in segments if word.strip()]
# 移除停用词（可选）
# words = [word for word in words if word not in stopwords]
tokenized_texts.append(words)
return tokenized_texts
def create_vector_store(tokenized_texts: List[List[str]], embeddings_model: OllamaEmbeddings) -> FAISS:
"""将分词后的文本创建向量库"""
try:
# 将分词列表转换回文本
processed_texts = [' '.join(tokens) for tokens in tokenized_texts]
# 批量处理优化
batch_size = 100 # 可以根据实际情况调整
vectors = []
# # 如果有 GPU
# if FAISS.get_num_gpus():
# res = FAISS.StandardGpuResources()
# index = FAISS.index_cpu_to_gpu(res, 0, index)
for i in tqdm(range(0, len(processed_texts), batch_size), desc="创建向量数据库"):
batch = processed_texts[i:i + batch_size]
# 批量创建向量
vector_store = FAISS.from_texts(
texts=batch,
embedding=embeddings_model,
metadatas=[{"index": j} for j in range(i, i + len(batch))] # 添加元数据以追踪文档
)
vectors.append(vector_store)
# 如果有多个批次，合并它们
if len(vectors) > 1:
final_vector_store = vectors[0]
for v in vectors[1:]:
final_vector_store.merge_from(v)
else:
final_vector_store = vectors[0]
# 保存向量库到本地
final_vector_store.save_local("resume_vectors")
return final_vector_store
except Exception as e:
print(f"创建向量库时发生错误: {str(e)}")
raise
# 提取简历文本
resume_texts = extract_text_from_markdown(resume_folder)
# resume_texts = extract_text_from_pdfs(resume_folder)
print("简历文本提取完成")
# 简历文本分词
tokenized_texts = tokenize_texts(resume_texts)
print(f"简历文本分词完成，共处理 {len(tokenized_texts)} 份文档")
# 可以打印一些统计信息（可选）
for i, tokens in enumerate(tokenized_texts):
print(f"文档 {i+1} 分词数量: {len(tokens)}")
# 创建 OpenAI 嵌入
embeddings = OllamaEmbeddings(model="nomic-embed-text:latest",base_url='xxx')
print("ollama 嵌入完成~")
# 创建向量库
vector_store = create_vector_store(tokenized_texts, embeddings)
print("向量库创建完成")
# Initialize OpenAI model
# from langchain_community.chat_models import ChatOpenAI
# llm = ChatOpenAI(model='gpt-4o', temperature=0.1, api_key=os.environ.get("OPENAI_API_KEY"))
from langchain_community.llms import Ollama
llm = Ollama(model="llama3.3:70b", temperature=0.1, num_ctx=60000,base_url='xxx')
# Update imports
from langchain.chains import RetrievalQA # Changed from create_retrieval_qa_chain
from langchain.prompts import PromptTemplate # Import the necessary class
PROMPT_TEMPLATE = """
已知信息：{context}。
"你是一个核聚变、人工智能、科学计算领域的人才鉴别专家，你具备管理大量简历的能力；请注意我向你提供了很多简历PDF文件，但每个PDF文件对应一个候选人，包括候选人的姓名、年龄、技能、经历、项目、成果等内容，请仔细识别各个信息。 \
现在需要你帮我分析每个候选人的详细信息，包括：年龄、教育程度、专业技能、职业履历、项目背景、研究成果、获得荣誉、发展潜力，然后帮我完成以下两个功能： \
1.当我给出开展项目的描述信息时，你能帮我准确地按照适配项目的优先级推荐相关候选人，每次允许按照优先级顺序推荐多个候选人，并且详细给出推荐原因，使用markdown的表格形式给出，包括以下信息：姓名、年龄、专业技能、推荐原因； \
2.当我需要分析一个候选人时，请你能进行客观、准确的评估，先描述其主要信息，然后按照：专业能力、科研成果、项目成绩、发展潜力、综合能力进行评分，每项总分100，结果以markown形式给出。 \
请务必注意每个简历仅对应一个候选人，切记不要混淆各个人的信息；1个候选人只需引用1次相关文档即可，仔细识别每个文档中候选人的姓名。"
请回答以下问题：{question}
"""
# 创建提示模板
prompt_template = PromptTemplate(
input_variables=["context", "question"],
template=PROMPT_TEMPLATE
)
# 创建检索链
chain_type_kwargs = {
"prompt": prompt_template,
"document_variable_name": "context",
}
# 创建检索链
qa = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vector_store.as_retriever(),
chain_type_kwargs=chain_type_kwargs,
)
def chat_loop():
print("\n欢迎使用简历分析助手！")
print("输入 'quit' 或 'exit' 结束对话")
while True:
# Get user input
user_input = input("\n请输入您的问题: ").strip()
# Check if user wants to exit
if user_input.lower() in ['quit', 'exit']:
print("感谢使用，再见！")
break
if user_input:
try:
# Get the response
result = qa.run(user_input)
print("\n回答：")
print(result)
except Exception as e:
print(f"发生错误: {str(e)}")
continue
if __name__ == "__main__":
chat_loop()
