房地产网站建设存在问题_环境设计排版素材网站_大数据精准营销_中国疫情最新数据

【RAG 项目实战 05】重构：封装代码

---

NLP Github 项目：

• NLP 项目实践：fasterai/nlp-project-practice

  介绍：该仓库围绕着 NLP 任务模型的设计、训练、优化、部署和应用，分享大模型算法工程师的日常工作和实战经验

• AI 藏经阁：https://gitee.com/fasterai/ai-e-book

  介绍：该仓库主要分享了数百本 AI 领域电子书

• AI 算法面经：fasterai/nlp-interview-handbook#面经

  介绍：该仓库一网打尽互联网大厂NLP算法面经，算法求职必备神器

• NLP 剑指Offer：https://gitee.com/fasterai/nlp-interview-handbook

  介绍：该仓库汇总了 NLP 算法工程师高频面题

---

• 新增 common.py
• 改造 llm_util.py
• 新增 chain_util.py
• 新增 msg_util.py
• 改造 app.py

新增 common.py

# @Author：青松
# 公众号：FasterAI
# Python, version 3.10.14
# Pytorch, version 2.3.0
# Chainlit, version 1.1.301class Constants:MODEL_NAME = {'QianFan': 'QianFan'}

改造 llm_util.py

# @Author：青松
# 公众号：FasterAI
# Python, version 3.10.14
# Pytorch, version 2.3.0
# Chainlit, version 1.1.301from common import Constants
from langchain_community.chat_models import QianfanChatEndpoint# 加载环境变量
from dotenv import load_dotenvload_dotenv()def get_llm(model_name):llm = Nonetry:if model_name == Constants.MODEL_NAME['QianFan']:llm = QianfanChatEndpoint(streaming=True,model="ERNIE-Speed-8K",)except:llm = get_default_llm()finally:if llm is None:llm = get_default_llm()return llmdef get_default_llm():default_llm = QianfanChatEndpoint(streaming=True,model="ERNIE-Speed-8K",)return default_llm

新增 chain_util.py

# @Author：青松
# 公众号：FasterAI
# Python, version 3.10.14
# Pytorch, version 2.3.0
# Chainlit, version 1.1.301from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.vectorstores import VectorStoredef get_chat_chain(llm):# 提示模板中添加 chat_historyprompt = ChatPromptTemplate.from_messages([("system","你是一个中国古诗词专家，能准确的一字不差的背诵很多古诗词，请用你最大的能力来回答用户的问题。",),MessagesPlaceholder("chat_history"),("human", "{question}"),])chat_chain = prompt | llm | StrOutputParser()return chat_chain

新增 msg_util.py

# @Author：青松
# 公众号：FasterAI
# Python, version 3.10.14
# Pytorch, version 2.3.0
# Chainlit, version 1.1.301import chainlit as cl
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory# 存储对话历史
store = {}def get_session_history(session_id: str) -> BaseChatMessageHistory:if session_id not in store:store[session_id] = ChatMessageHistory()return store[session_id]async def send_welcome_msg():# todo： 添加 FasterAI 知识星球图片以及 FastAI 知识库地址image = cl.Image(url="https://qingsong-1257401904.cos.ap-nanjing.myqcloud.com/wecaht.png")# 发送一个图片await cl.Message(content="**青松** 邀你关注 **FasterAI**， 让每个人的 AI 学习之路走的更容易些！立刻扫码开启 AI 学习、面试快车道 **(^_^)** ",elements=[image],).send()async def response_with_history_by_astream(message: cl.Message, chain, session_id):# 用 RunnableWithMessageHistory 包装 Chain 添加对话历史能力runnable_with_history = RunnableWithMessageHistory(chain,get_session_history,input_messages_key="question",history_messages_key="chat_history",)msg = cl.Message(content="")# 配置中使用 session_id 进行大模型交互async for chunk in runnable_with_history.astream({"question": message.content},config=RunnableConfig(configurable={"session_id": session_id},callbacks=[cl.LangchainCallbackHandler()])):await msg.stream_token(chunk)await msg.send()

改造 app.py

# @Author：青松
# 公众号：FasterAI
# Python, version 3.10.14
# Pytorch, version 2.3.0
# Chainlit, version 1.1.301import chainlit as clfrom common import Constants
import chain_util
import llm_util
import msg_util# 获取大模型实例
llm = llm_util.get_llm(Constants.MODEL_NAME['QianFan'])@cl.password_auth_callback
def auth_callback(username: str, password: str):""" 持久化客户端聊天历史代码，不需要请删除 """if (username, password) == ("admin", "admin"):return cl.User(identifier="admin", metadata={"role": "admin", "provider": "credentials"})else:return None@cl.on_chat_start
async def on_chat_start():""" 监听会话开始事件 """# 添加 session_idcl.user_session.set('session_id', 'abc2')# 发送欢迎信息await msg_util.send_welcome_msg()# 初始化链init_chains()@cl.on_message
async def on_message(message: cl.Message):""" 监听用户消息事件 """# 获得对话链chat_chain = cl.user_session.get("chat_chain")# 获取当前的 session_idsession_id = cl.user_session.get("session_id")# 使用对话历史通过 astream 的方式响应用户消息await msg_util.response_with_history_by_astream(message, chat_chain, session_id)def init_chains():""" 初始化系统中的链 """# 对话链chat_chain = chain_util.get_chat_chain(llm)cl.user_session.set("chat_chain", chat_chain)

---

【动手学 RAG】系列文章：

• 【RAG 项目实战 01】在 LangChain 中集成 Chainlit
• 【RAG 项目实战 02】Chainlit 持久化对话历史
• 【RAG 项目实战 03】优雅的管理环境变量
• 【RAG 项目实战 04】添加多轮对话能力
• 【RAG 项目实战 05】重构：封装代码
• 【RAG 项目实战 06】使用 LangChain 结合 Chainlit 实现文档问答
• 【RAG 项目实战 07】替换 ConversationalRetrievalChain（单轮问答）
• 【RAG 项目实战 08】为 RAG 添加历史对话能力