Langchain + chatglm-6b
文章目录
- Langchain + chatglm-6b
- 前言
- 一、实验记录
- 1.1 环境配置
- 1.2 代码理解
- 1.3 补充内容
- 二、总结
前言
介绍：一种利用 ChatGLM-6B + langchain 实现的基于本地知识的 ChatGLM 应用
Github: https://github.com/imClumsyPanda/langchain-ChatGLM
一、实验记录
1.1 环境配置
conda create -n lang python=3.8
conda activate langpip install -r requirement.txt
1.2 代码理解
1-将文章使用句号进行分句
2-将句子使用text2vec进行向量化存储
3-将query进行向量化后，和FAISS库里的正文文档进行topk相似度的召回
4-召回到的正文和query构造成prompt，输入到chatglm中
prompt模板的设置：
prefix_prompt = \
"""基于以下已知信息，简洁和专业的来回答用户的问题。如果无法从中得到答案，请说 "根据已知信息无法回答该问题" 或 "没有提供足够的相关信息"，不允许在答案中添加编造成分，答案请使用中文。已知内容:{context}问题:{question}"""
核心代码
def _call(self, inputs: Dict[str, str]) -> Dict[str, Any]:"""question = inputs[self.input_key] 获取querydocs = self._get_docs(question) 根据query获取上下文中最相似的正文answer = self.combine_documents_chain.run(input_documents=docs, question=question)if self.return_source_documents:return {self.output_key: answer, "source_documents": docs}else:return {self.output_key: answer}@abstractmethodasync def acombine_docs(self, docs: List[Document], **kwargs: Any) -> Tuple[str, dict]:"""Combine documents into a single string asynchronously."""def _call(self, inputs: Dict[str, Any]) -> Dict[str, str]:docs = inputs[self.input_key]# Other keys are assumed to be needed for LLM predictionother_keys = {k: v for k, v in inputs.items() if k != self.input_key}output, extra_return_dict = self.combine_docs(docs, **other_keys)extra_return_dict[self.output_key] = outputreturn extra_return_dict
然后将找回来的正文和query进行concate之后，统一输入到llm中，对历史信息还是没有处理
def _get_inputs(self, docs: List[Document], **kwargs: Any) -> dict:# Format each document according to the promptdoc_strings = [format_document(doc, self.document_prompt) for doc in docs]# Join the documents together to put them in the prompt.inputs = {k: vfor k, v in kwargs.items()if k in self.llm_chain.prompt.input_variables}inputs[self.document_variable_name] = "\n\n".join(doc_strings)return inputs
处理后的数据
data = \
{'question': '大气污染预警分为多少级', 'context': '建立健全空气质量预报预警应急 体系，加强大气污染防治工作，不断促进大气环境质量改善。\n\n 各责任部门和街道办应根据本预案的相关规定，在大气 污染预警与应急响应期间落实具体的污染防控措施。\n\n 1.5 大气污染预警与应急响应分级 1.5.1 预警分级 大气污染预警分为三级，依次用黄色、橙色和红色表示。\n\n同时，确保大气污染预警应急信息有 效传达，提醒市民采取适当的健康防护措施。\n\n 负责向区政府、市生态环境局报告污染信息；负责判断大 气污染级别，提出发布预警信息的建议；负责空气质量监 市生态环境局龙华 测、预测和信息发布；负责组织专家会商；倡导社会各界 管理局 自觉参与污染减排；确定重点减排企业名单，督促检查工 业企业落实污染减排措施。\n\n获取气象监测预报信息，与市 生态环境局共同会商空气污染预报。'
}
调用
self.llm_chain.predict(**inputs), {}
生成答案的过程：
def generate(self, input_list: List[Dict[str, Any]]) -> LLMResult:"""Generate LLM result from inputs."""prompts, stop = self.prep_prompts(input_list) input_list上文的datareturn self.llm.generate_prompt(prompts, stop)def prep_prompts(self, input_list: List[Dict[str, Any]]) -> Tuple[List[PromptValue], Optional[List[str]]]:"""Prepare prompts from inputs."""stop = Noneif "stop" in input_list[0]:stop = input_list[0]["stop"]prompts = []for inputs in input_list:selected_inputs = {k: inputs[k] for k in self.prompt.input_variables}prompt = self.prompt.format_prompt(**selected_inputs) _colored_text = get_colored_text(prompt.to_string(), "green")_text = "Prompt after formatting:\n" + _colored_textself.callback_manager.on_text(_text, end="\n", verbose=self.verbose)if "stop" in inputs and inputs["stop"] != stop:raise ValueError("If `stop` is present in any inputs, should be present in all.")prompts.append(prompt)return prompts, stop
self.prompt.format_prompt 即是之前设置的 prefix_prompt
构造完毕之后是：
基于以下已知信息，简洁和专业的来回答用户的问题。如果无法从中得到答案，请说 "根据已知信息无法回答该问题" 或 "没有提供足够的相关信息"，不允许在答案中添加编造成分，答案请使用中文。已知内容:2 组织机构和职责 2.1 领导机构 龙华区大气质量提升工作领导小组是龙华区突发事件 应急委员会下设的专项应急指挥部。深圳市龙华区大气质量提升工作 领导小组（以下简称“区大气领导小组”）负责组织、协调、 指挥大气污染的预警与应急响应工作。区大气领导小组成员单位包括：区委 宣传部、市生态环境局龙华管理局、区财政局、区城市管理 和综合执法局、区住房和建设局、区水务局、区教育局、区 4 卫生健康局、区建筑工务署、市交通运输局龙华管理局、龙 华交警大队和各街道办。深圳市龙华区大气质量提升工作领导小组办公室 年 月 日 252.2 工作机构 区大气领导小组办公室为工作机构，设在市生态环境局 龙华管理局，负责区大气领导小组的日常工作，包括确定预 警与响应等级，督查区有关部门落实预警与应急响应措施。经请示深圳市龙华区大气质量提升工作领导小组同意，现决定启 动《深圳市龙华区大气污染应急预案》，请你单位依据大气污染应急 预案的相关规定，立即组织力量参与应急处置工作。问题:龙华区大气质量提升工作领导小组是什么
1.3 补充内容
langchain的基于运行原理
def run(self) -> None:with claim_worker_thread("asyncio"):threadlocals.loop = self.loopwhile True:item = self.queue.get()if item is None:# Shutdown command receivedreturncontext, func, args, future = itemif not future.cancelled():result = Noneexception: Optional[BaseException] = Nonetry:result = context.run(func, *args)except BaseException as exc:exception = excif not self.loop.is_closed():self.loop.call_soon_threadsafe(self._report_result, future, result, exception)self.queue.task_done()
构造一个队列等待用户的信号，一旦页面中监测到信号，依次去判断哪个按钮触发的信号，再根据信号的钩函数处理对应的逻辑
二、总结
这里的整体逻辑还是相对比较简单的，每个环节都还可以继续优化，作者在群里说了，项目代码已经在持续优化中了。
