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 lang
pip 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] 获取query
docs = 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}
@abstractmethod
async 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 prediction
other_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] = output
return extra_return_dict
然后将找回来的正文和query进行concate之后，统一输入到llm中，对历史信息还是没有处理
def _get_inputs(self, docs: List[Document], **kwargs: Any) -> dict:
# Format each document according to the prompt
doc_strings = [format_document(doc, self.document_prompt) for doc in docs]
# Join the documents together to put them in the prompt.
inputs = {
k: v
for 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上文的data
return 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 = None
if "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_text
self.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 卫生健康局、区建筑工务署、市交通运输局龙华管理局、龙 华交警大队和各街道办。
深圳市龙华区大气质量提升工作领导小组办公室 年 月 日 25
2.2 工作机构 区大气领导小组办公室为工作机构，设在市生态环境局 龙华管理局，负责区大气领导小组的日常工作，包括确定预 警与响应等级，督查区有关部门落实预警与应急响应措施。
经请示深圳市龙华区大气质量提升工作领导小组同意，现决定启 动《深圳市龙华区大气污染应急预案》，请你单位依据大气污染应急 预案的相关规定，立即组织力量参与应急处置工作。
问题:
龙华区大气质量提升工作领导小组是什么
1.3 补充内容
langchain的基于运行原理
def run(self) -> None:
with claim_worker_thread("asyncio"):
threadlocals.loop = self.loop
while True:
item = self.queue.get()
if item is None:
# Shutdown command received
return
context, func, args, future = item
if not future.cancelled():
result = None
exception: Optional[BaseException] = None
try:
result = context.run(func, *args)
except BaseException as exc:
exception = exc
if not self.loop.is_closed():
self.loop.call_soon_threadsafe(
self._report_result, future, result, exception
)
self.queue.task_done()
构造一个队列等待用户的信号，一旦页面中监测到信号，依次去判断哪个按钮触发的信号，再根据信号的钩函数处理对应的逻辑
二、总结
这里的整体逻辑还是相对比较简单的，每个环节都还可以继续优化，作者在群里说了，项目代码已经在持续优化中了。
