TASK04【Datawhale 组队学习】构建RAG应用

langchain可以便捷地调用大模型，并将其结合在以langchain为基础框架搭建的个人应用中。

将LLM接入LangChain

1. from langchain_openai import ChatOpenAI
2. 实例化一个 ChatOpenAI 类,实例化时传入超参数来控制回答，配置API密钥
3. llm.invoke(“prompt”)
4. 提示模板（PromptTemplates）：该文本提供有关当前特定任务的附加上下文。prompt=“模板”，text:用户输入 prompt.format(text=text)
   聊天模型的接口是基于消息（message），而不是原始的文本，PromptTemplates 也可以用于产生消息列表，在这种样例中prompt不仅包含了输入内容信息，也包含了每条message的信息(角色、在列表中的位置等)。
   一个ChatPromptTemplate是一个ChatMessageTemplate 的列表。每个 ChatMessageTemplate 包含格式化该聊天消息的说明（其角色以及内容）。

• 定义设定system_template，human_template是用户的输入
• ChatPromptTemplate构造参数包含list下的两个元组参数[
  (“system”, template),
  (“human”, human_template),
  ]，这里可以添加更多的角色的消息

from langchain_core.prompts import ChatPromptTemplatetemplate = "你是一个翻译助手，可以帮助我将 {input_language} 翻译成 {output_language}."
human_template = "{text}"chat_prompt = ChatPromptTemplate([("system", template),("human", human_template),
])text = "我带着比身体重的行李，\
游入尼罗河底，\
经过几道闪电 看到一堆光圈，\
不确定是不是这里。\
"
messages  = chat_prompt.invoke({"input_language": "中文", "output_language": "英文", "text": text})

import os
# 新的导入语句os.environ['ZHIPUAI_API_KEY']="这里写上自己的api"
api_key = os.environ["ZHIPUAI_API_KEY"] #
from zhipuai_llm import ZhipuaiLLM
from dotenv import find_dotenv, load_dotenv# 读取本地/项目的环境变量。# find_dotenv()寻找并定位.env文件的路径
# load_dotenv()读取该.env文件，并将其中的环境变量加载到当前的运行环境中
# 如果你设置的是全局的环境变量，这行代码则没有任何作用。
_ = load_dotenv(find_dotenv())

构建检索问答链

我们可以使用LangChain的LCEL(LangChain Expression Language, LangChain表达式语言)来构建workflow，LCEL可以支持异步（ainvoke）、流式(stream)、批次处理(batch)等多种运行方式，同时还可以使用LangSmith无缝跟踪。

from langchain_core.runnables import RunnableLambda
def combine_docs(docs):return "\n\n".join(doc.page_content for doc in docs)combiner = RunnableLambda(combine_docs)
retrieval_chain = retriever | combinerretrieval_chain.invoke("南瓜书是什么？")

LCEL中要求所有的组成元素都是Runnable类型，前面我们见过的ChatModel、PromptTemplate等都是继承自Runnable类。上方的retrieval_chain是由检索器retriever及组合器combiner组成的，由|符号串连，数据从左向右传递，即问题先被retriever检索得到检索结果，再被combiner进一步处理并输出。

llm=zhipuai_model
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.output_parsers import StrOutputParsertemplate = """使用以下上下文来回答最后的问题。如果你不知道答案，就说你不知道，不要试图编造答
案。最多使用三句话。尽量使答案简明扼要。请你在回答的最后说“谢谢你的提问！”。
{context}
问题: {input}
"""
# 将template通过 PromptTemplate 转为可以在LCEL中使用的类型
prompt = PromptTemplate(template=template)qa_chain = (RunnableParallel({"context": retrieval_chain, "input": RunnablePassthrough()})| prompt| llm| StrOutputParser()
)
question_1 = "什么是南瓜书？"
question_2 = "Prompt Engineering for Developer是谁写的？"
result = qa_chain.invoke(question_1)
print("大模型+知识库后回答 question_1 的结果：")
print(result)
result = qa_chain.invoke(question_2)
print("大模型+知识库后回答 question_2 的结果：")
print(result)
llm.invoke(question_1).content

# 无历史记录
messages = qa_prompt.invoke({"input": "南瓜书是什么？","chat_history": [],"context": ""}
)
for message in messages.messages:print(message.content)

# 有历史记录
messages = qa_prompt.invoke({"input": "你可以介绍一下他吗？","chat_history": [("human", "西瓜书是什么？"),("ai", "西瓜书是指周志华老师的《机器学习》一书，是机器学习领域的经典入门教材之一。"),],"context": ""}
)
for message in messages.messages:print(message.content)

import os
import sys
from zhipuai_llm import ZhipuaiLLM
os.environ['ZHIPUAI_API_KEY']=""
api_key = os.environ["ZHIPUAI_API_KEY"] #填写控制台中获取的 APIKey 信息sys.path.append(os.getcwd())# 将父目录放入系统路径中import streamlit as st
# from langchain_openai import ChatOpenAIfrom langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch, RunnablePassthrough
from zhipuai_embedding import ZhipuAIEmbeddings
from langchain_community.vectorstores import Chromadef get_retriever():# 定义 Embeddingsembedding = ZhipuAIEmbeddings()# 向量数据库持久化路径persist_directory = 'D:\\code\\llm_universe\\chroma'# 加载数据库vectordb = Chroma(persist_directory=persist_directory,embedding_function=embedding)return vectordb.as_retriever()
def combine_docs(docs):return "\n\n".join(doc.page_content for doc in docs["context"])def get_qa_history_chain():retriever = get_retriever()zhipuai_model = ZhipuaiLLM(model_name="glm-4-plus", temperature=0.1, api_key=api_key)llm = zhipuai_modelcondense_question_system_template = ("请根据聊天记录总结用户最近的问题，""如果没有多余的聊天记录则返回用户的问题。")condense_question_prompt = ChatPromptTemplate([("system", condense_question_system_template),("placeholder", "{chat_history}"),("human", "{input}"),])retrieve_docs = RunnableBranch((lambda x: not x.get("chat_history", False), (lambda x: x["input"]) | retriever, ),condense_question_prompt | llm | StrOutputParser() | retriever,)system_prompt = ("你是一个问答任务的助手。 ""请使用检索到的上下文片段回答这个问题。 ""如果你不知道答案就说不知道。 ""请使用简洁的话语回答用户。""\n\n""{context}")qa_prompt = ChatPromptTemplate.from_messages([("system", system_prompt),("placeholder", "{chat_history}"),("human", "{input}"),])qa_chain = (RunnablePassthrough().assign(context=combine_docs)| qa_prompt| llm| StrOutputParser())qa_history_chain = RunnablePassthrough().assign(context = retrieve_docs, ).assign(answer=qa_chain)return qa_history_chaindef gen_response(chain, input, chat_history):response = chain.stream({"input": input,"chat_history": chat_history})for res in response:if "answer" in res.keys():yield res["answer"]
def main():st.markdown('### 🦜🔗 动手学大模型应用开发')# st.session_state可以存储用户与应用交互期间的状态与数据# 存储对话历史if "messages" not in st.session_state:st.session_state.messages = []# 存储检索问答链if "qa_history_chain" not in st.session_state:st.session_state.qa_history_chain = get_qa_history_chain()# 建立容器 高度为500 pxmessages = st.container(height=550)# 显示整个对话历史for message in st.session_state.messages: # 遍历对话历史with messages.chat_message(message[0]): # messages指在容器下显示，chat_message显示用户及ai头像st.write(message[1]) # 打印内容if prompt := st.chat_input("Say something"):# 将用户输入添加到对话历史中st.session_state.messages.append(("human", prompt))# 显示当前用户输入with messages.chat_message("human"):st.write(prompt)# 生成回复answer = gen_response(chain=st.session_state.qa_history_chain,input=prompt,chat_history=st.session_state.messages)# 流式输出with messages.chat_message("ai"):output = st.write_stream(answer)# 将输出存入st.session_state.messagesst.session_state.messages.append(("ai", output))if __name__ == "__main__":main()

运行成功图

遇到的问题

解决了一个很久以来都觉得奇怪的问题，pip的下载的包是依据与conda虚拟环境的，jupyter内核也有运行的conda虚拟环境，这些都是相互依赖的，我出现的问题是jupyternotebook运行在环境2的，而直接pip下载的包是在环境1里面的，所以需要conda activate 环境2之后再进行包的下载。jupyter notebook的python核版本可以自己进行设置，但是如果在环境中运行了!python --version可能显示的是环境1的version。使用import sys print(sys.executable)可以查看当前jupyternotebook运行的python的环境。

“D:\anaconda\set\share\jupyter\kernels\python3_10_13”这里可以修改运行的python核的路径，直接进行复制


再修改argv的路径就行。display_name和文件夹的名字需要相同

元组tuple()
列表list[]
字典dictionary{“key”:“value”,“key”:“value”}
集合set{a,b,c}
数组array

import array
arr = array.array("i",[1,2,3])

队列：queue

from collections import deque#双端
q = deque([1,2,3])
q.append(4)
q.popleft()#队头删除，返回1

堆：heap

import heapq
heap = [3,1,2]
heapq.heapify(heap)#转换为最小堆
heapq.heappush(heap,0)# 插入 元素0
heapq.heappop(heap)#弹出最小元素