在这个故事中, 我们将探索基于一组文档的LangChain的问答功能. 我们将描述一个简单的HR应用程序示例，该应用程序扫描文件夹中包含cv的一组文件，并询问ChatGPT 3.十大网博靠谱平台候选人的5个问题.
我们使用了LangChain v0.0.190年在这里.
这里问的问题/说明是:
求职者的名字是什么?
这个候选人的专业是什么?
请提取所有超链接.
应聘者有多少年的手机开发经验?
简历中提到了哪些大学?
这个简单的人力资源应用程序嵌入在Web服务器中(使用FastAPI和uvicorn)，它将这些问题的答案显示为一个简单的网页.
LangChain文档处理能力
LangChain 具有多个文档处理链:
的 的东西 链将文档包含到它的问题(提示)上下文中，并将问题(提示)公开给LLM.
这张图描绘了 的东西 来自Lang链文档的文档链:
应用程序流
在我们的HR应用程序中有两个工作流:
- 第一个工作流(QA创建过程)为每个特定的候选人创建问题和答案列表
- 第二个是十大网博靠谱平台获取web请求，将其转换为HTML并发送响应
QA创建过程执行以下步骤:
- 它循环预定义文件夹中的所有文档(仅pdf和docx文件)
- 它检查特定候选人的问答是否已经被缓存过.
- 如果先前缓存了答案，那么它们将被加载到内存中
- 如果QA没有被缓存，文本将被提取并分割.
- 工作流检查是否有任何缓存的嵌入.
- 如果存在缓存嵌入，则将它们加载到内存中.
- 如果没有，则查询ChatGPT，并检索提取文本的一组嵌入
- 嵌入集保存在本地目录中，因此被缓存.
- 使用嵌入集查询ChatGPT的所有答案.
- QA存储在内存中并缓存在文件系统中.
要了解什么是嵌入，请访问 OpenAI十大网博靠谱平台嵌入的页面.
用户界面
这个应用程序的用户界面是一个单一的页面，显示与用户相关的QA的可扩展卡:
设置和执行
上面描述的流程的实现可以在这个GitHub仓库中找到:
GitHub - gilfernandes/document_stuff_playground: Playground项目，用于探索Lang链的…
Playground项目用于探索Lang链的文档填充问答- GitHub…
实现
我们提到web应用程序有两个流. 实现有两个部分: 问题/答案创建过程 web服务器:
问题/答案创建过程
document_stuff_playground / document_extract.main·gilfernandes/document_stuff_playground
Playground项目用于探索Lang链的Document 的东西ing与问答…
我们使用了“gpt-3”.5-turbo-16k”模型以及OpenAI嵌入.
类配置():
模型= 'gpt-3.5-turbo-16k”
llm = ChatOpenAI(model=model, temperature=0)
embeddings = OpenAIEmbeddings()
Chunk_size = 2000
Chroma_persist_directory = 'chroma_store'
candidate_infos_cache = Path('candidate_infos_cache')
如果不是，candidate_infos_cache.存在():
candidate_infos_cache.mkdir ()我们定义了一个数据类来包含文件名和问题/答案对:
@dataclass
类CandidateInfo ():
candidate_file: str
问题:list[(str, str)]该函数完成大部分工作并循环遍历文件, 提取它们的内容, 创建嵌入, 创建答案并收集结果如下:
def extract_candidate_infos(doc_folder: Path) -> List[CandidateInfo]:
"""
从“doc_folder”中的每个pdf或docx文件中提取问题和答案
并将它们保存在一个列表中. 首先，它循环遍历文件，提取其内容
作为嵌入并缓存这些，然后与ChatGPT交互. 答案是
保存在数据结构中并缓存. 如果答案已经提供给候选人
它们是从泡菜文件中读取的.
:param doc_folder存放候选文档的文件夹.
:返回候选问题和答案的列表.
"""
如果不是doc_folder.存在():
“候选文件夹{doc_folder}不存在。!")
返回[]
candidate_list: list[CandidateInfo] = []
扩展:list[str] = ['**/*.pdf”、“* * / *.多克斯']
对于extensions中的extension:
对于doc_folder中的doc.rglob(扩展):
File_key = doc.阀杆
Cached_candidate_info = read_saved_candidate_info (file_key)
如果cached_candidate_info为None:
Docsearch = process_document(doc)
打印(f“加工{医生}”)
如果docsearch不是None:
qa = RetrievalQA.from_链_type (llm = cfg.Llm, 链_type=“stuff”，retriver =docsearch.as_retriever ())
Question_list = []
对于问题中的问题:
#在这里问问题
question_list.追加(问题,qa.运行(问题)))
(candidate_file=file_key, questions=question_list)
write_candidate_infos (file_key candidate_info)
candidate_list.追加(candidate_info)
其他:
print(f"无法从{doc}检索内容")
其他:
candidate_list.追加(cached_candidate_info)
返回candidate_list我们还缓存了结果，以防止太多的API调用.
我们有一个函数来泡菜(序列化)收集到的结果:
Def write_candidate_info (file_key, candidate_info):
Cached_file = CFG.candidate_infos_cache / file_key
使用open(cached_file， "wb")作为f:
泡菜.转储(candidate_info f)另一个检查CV文件是否已经在本地文件系统上有相应的QA:
def read_saved_candidate_infos(file_key: str) -> Union[None, CandidateInfo]:
Cached_file = CFG.candidate_infos_cache / file_key
试一试:
如果cached_file.存在():
使用open(cached_file， "rb")作为f:
返回泡菜.负载(f)
例外情况如下:
print(f"无法处理{file_key}")
回来没有下面的函数在本地文件系统中提取并持久化嵌入:
def extract_embeddings(texts: List[Document], doc_path: Path) -> Chroma:
"""
要么将Chroma嵌入保存在本地，要么从磁盘(如果存在的话)读取它们.
:返回嵌入周围的色度包装器.
"""
Embedding_dir = f"{cfg.chroma_persist_directory} / {doc_path.茎}”
如果路径(embedding_dir).存在():
返回色度(persist_directory=embedding_dir, embedding_function=cfg).嵌入的)
试一试:
docsearch =色度.from_documents(文本,cfg.嵌入,persist_directory = embedding_dir)
docsearch.persist ()
例外情况如下:
print(f"Failed to process {doc_path}: {str(e)}")
回来没有
返回docsearch我们正在使用 Chroma，一个开源嵌入 这里的数据库. Chroma允许保存嵌入(表示标记的数字向量)及其元数据.
这个函数读取文档并提取文本:
def process_document(doc_path) -> Chroma:
"""
通过从文档中加载文本来处理文档.
它支持两种格式:pdf和docx. 然后分裂
大块的文本，然后从中提取嵌入.
:param doc_path包含文档的路径或表示该路径的字符串.
:返回嵌入周围的色度包装器.
"""
如果不是isinstance(doc_path, Path):
doc_path =路径(doc_path)
如果不是doc_path.存在():
print(f"文档({doc_path})不存在. 请检查”)
其他:
打印(f”处理{doc_path}”)
(PDFPlumberLoader(str(doc_path)).后缀== ".pdf”
其他Docx2txtLoader (str (doc_path)))
doc_list: List[Document] = loader.load ()
print(f"已提取的文档:{len(doc_list)}")
对于i, doc在enumerate(doc_list)中:
i += 1
如果len(医生.Page_content) == 0
print(f"文档有空页:{i}")
其他:
打印(f"页{i}的长度:{len(doc.page_content)}”)
text_splitter = CharacterTextSplitter(chunk_size=cfg).chunk_size chunk_overlap = 0)
文本= text_splitter.split_documents (doc_list)
返回extract_embeddings(文本，doc_path)Web服务器
实现的这一部分为存储在内存中的结果提供服务 FastAPI 框架. 您也可以只创建一个REST服务，并使用现代web框架开发一个应用程序.
我们的实现使用一个调度的后台线程来更新十大网博靠谱平台候选人的qa列表，以防cv保存在输入目录中:
类CandidateCache ():
candidate_info_html = "Processing, please wait ...
"
candidate_cache = CandidateCache()
Sleep_time = 60 * 10
类BackgroundTasks(线程.线程):
"""
在后台处理文档和查询聊天GPT.
"""
def __init__(自我, candidate_cache: CandidateCache, sleep_time = sleep_time):
super ().__init__ ()
自我.candidate_cache,自我.Sleep_time = candidate_cache, Sleep_time
自我.data_cache = sleep_time的n 然后，你可以找到一个html生成端点来呈现一个简单的页面:
@app.get(" /候选人.html”,response_class = HTMLResponse)
Async def hello_html():
def generate_timestamp ():
#获取当前日期和时间
Now =日期时间.现在()
#获取英文的周、日、月、年和时间
工作日=现在.strftime(“%”)
日=现在.strftime(“% d”)
月=现在.strftime(“% B”)
年=现在.Y strftime(“%”)
时间=现在.strftime(“% H: % M: % S”)
#创建时间戳字符串
时间戳= f"{星期}，{日}{月}{年}{时间}"
返回时间戳
返回f”“”
服务器通过 uvicorn:
如果__name__ == '__main__':
print(“快速API设置”)
uvicorn.运行(应用程序、主机= " 0.0.0.0”,端口= 8000)结论
构建基于文档的基本人力资源应用程序相对容易 LangChain 使用法学硕士来回答有关这些文档的问题. 大部分繁重的工作(与LLM通信，嵌入生成)由 LangChain，它也为不同的文件类型提供广泛的支持.
对于人力资源，您还可以创建一个应用程序，该应用程序将候选人与 地图排名 链. LangChain 也支持这样的功能，但我们将在后续的博客中对此进行探讨.
