搭建基于LangChain实现复杂RAG聊天机器人

• 使用 LangChain 和 DeepSeek v3模型构建一个复杂的 RAG 聊天机器人，能够处理复杂的查询，并且可以通过聊天历史记录维护上下文，并使用 LangChain 的 LCEL语法遵守严格的Guardrails（护栏）。

智能HR聊天机器人助手

1 准备工作

1. 素材文本：一份模拟的公司员工手册，并以md文档格式进行存储。
   
2. 护栏功能介绍：Guardrails(护栏)对于确保AI系统的安全性和可靠性是比较重要的。通过设定明确的界限，我们可以防止大模型生成有害或误导性的内容。拒绝机制使机器人能够礼貌地拒绝违反这些护栏的请求，例如与敏感主题或非法活动相关的请求。
   
3. python环境

   cond create -n rag python=3.13
   conda activate rag
   pip install langchain langchain-deepseek
   

4. DeepSeek API KEY
5. 本地Ollama部署bge-m3向量模型

   ollama pull bge-m3
   ollama start
   

6. FAISS矢量数据库

   pip install langchain-text-splitters faiss-cpu langchain-community
   

2 LangChain 接入DeepSeek对话模型

• LangChain集成DeepSeek模型的依赖包后，需要通过一个init_chat_model函数来初始化大模型。其中model用来指定要使用的模型名称，而model_provider用来指定模型提供者，当写入deepseek时，会自动加载langchain-deepseek的依赖包，并使用在model中指定的模型名称用来进行交互。

import os
from langchain.chat_models import init_chat_modelos.environ["DeepSeek_API_KEY"]="sk-xxx"model = init_chat_model(model="deepseek-chat", model_provider="deepseek")question = "你好，请你介绍一下你自己。"result = model.invoke(question)
print(result.content)

3 文档切分与构建词向量数据库

1. 使用Ollama部署的bge-m3模型将自然语言转化成词向量的表示。
2. FAISS是 Facebook AI Research 开发的一个库，用于高效相似性搜索和密集向量聚类。LangChain在第三方集成模块（Langchain_community）中已经接入了FAISS向量数据库

pip install langchain-text-splitters faiss-cpu

from langchain_ollama import OllamaEmbeddings
# 初始化嵌入模型（使用 Ollama 的 bge-m3 模型）
embed = OllamaEmbeddings(model="bge-m3", base_url="http://localhost:11434")
# 读取文本内容
file_path = "模拟公司员工手册.md"with open(file_path, "r", encoding="utf-8") as f:md_content = f.read()
# 设置切分策略
from langchain_text_splitters import MarkdownHeaderTextSplitterheaders_to_split_on = [("#", "Header 1"),("##", "Header 2")
]markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
md_header_splits = markdown_splitter.split_text(md_content)
# 向量化切分块并存储
vector_store = FAISS.from_documents(md_header_splits, embedding=embed)
vector_store.save_local("staff_handbook_db")# 加载向量数据，并测试效果
# 加载本地的Faiss向量文件，allow_dangerous_deserialization 用于控制是否允许在加载向量存储时进行潜在的危险反序列化操作。
vector_store = FAISS.load_local(embeddings=embed, folder_path='staff_handbook_db',allow_dangerous_deserialization=True)# 将 FAISS 向量存储转换为一个 retriever（检索器），并为该检索器设置一些搜索相关的参数。k=1 表示检索时返回 最相似的 3 个文档
retriever = vector_store.as_retriever(search_kwargs={'k': 2})# 执行相似度搜素
query = "请问我们公司有没有病假？"
results = retriever.invoke(query)for doc in results:print(f"Content: {doc.page_content}")

from langchain_ollama import OllamaEmbeddings
# 初始化嵌入模型（使用 Ollama 的 bge-m3 模型）
embed = OllamaEmbeddings(model="bge-m3", base_url="http://localhost:11434")
# 读取文本内容
file_path = "模拟公司员工手册.md"with open(file_path, "r", encoding="utf-8") as f:md_content = f.read()
# 设置切分策略
from langchain_text_splitters import MarkdownHeaderTextSplitterheaders_to_split_on = [("#", "Header 1"),("##", "Header 2")
]markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
md_header_splits = markdown_splitter.split_text(md_content)from langchain_community.vectorstores import FAISS
# 向量化切分块并存储
vector_store = FAISS.from_documents(md_header_splits, embedding=embed)
vector_store.save_local("staff_handbook_db")# 加载向量数据，并测试效果
# 加载本地的Faiss向量文件，allow_dangerous_deserialization 用于控制是否允许在加载向量存储时进行潜在的危险反序列化操作。
vector_store = FAISS.load_local(embeddings=embed, folder_path='staff_handbook_db',allow_dangerous_deserialization=True)# 将 FAISS 向量存储转换为一个 retriever（检索器），并为该检索器设置一些搜索相关的参数。k=1 表示检索时返回 最相似的 3 个文档
retriever = vector_store.as_retriever(search_kwargs={'k': 2})# 执行相似度搜素
query = "请问我们公司有没有病假？"
results = retriever.invoke(query)for doc in results:print(f"Content: {doc.page_content}")

Content: ### 3.1 工作时间  
公司实行标准工作制，每周五天，每天8小时（上午9:00至下午6:00，包括1小时午休）。为适应不同岗位，公司提供灵活工作选项，如弹性上班时间（8:30-9:30间到岗）和远程办公（每周最多3天），需经主管批准。  
加班须事先申请，并按国家劳动法补偿：平日加班1.5倍工资，周末2倍，法定假日3倍。公司监控加班情况，避免员工过度劳累，年度加班上限为200小时。  
公司使用电子打卡系统记录时间，支持移动APP打卡。特殊岗位如研发，可采用成果导向制，强调输出而非时长。  
### 3.2 出勤要求  
员工必须准时出勤，迟到或早退定义为超过10分钟。连续三次迟到将口头警告，累计五次进入书面记录。旷工一天视为严重违纪。  
病假或事假须提前24小时通知，并提供医生证明或说明。年度出勤率低于95%将影响绩效奖金。公司提供弹性出勤政策，如育儿假调整时间。  
远程工作员工须每日报告进度，并参加视频会议。公司定期审计出勤数据，确保合规。  
### 3.3 假期与休假  
公司提供丰富的假期福利，遵守国家规定：  
- **法定假期**：包括元旦、春节、清明、劳动节、端午、中秋和国庆，共计11天。  
- **带薪年假**：服务1-10年者10天，10-20年15天，20年以上20天。未使用年假可顺延或补偿。  
- **病假**：带薪病假按服务年限计算，最长6个月。  
- **产假/陪产假**：女性158天，男性15天。  
- **其他**：丧假3天，婚假10天。  
休假申请通过在线系统提交，主管批准后生效。公司鼓励员工规划休假，维持工作生活平衡。示例：年假可分次使用，避免高峰期冲突。

4 智能HR聊天助手逐步搭建过程

1. 从一个最简单的链开始，只接受用户问题，在提示中格式化它并输出该问题的答案（不检索）。这里使用 Langchain 的PromptTemplate并使用LCEL对其进行管道传输。

import os
from langchain.chat_models import init_chat_model
from langchain.prompts import PromptTemplate
from langchain.schema.output_parser import StrOutputParseros.environ["DeepSeek_API_KEY"]="sk-xxx"model = init_chat_model(model="deepseek-chat", model_provider="deepseek")
# 定义提示模板
prompt = PromptTemplate(input_variables = ["question"],template = "你是一个乐于助人的智能小助理。擅长根据用户输入的问题给出一个简短的回答：: {question}"
)# 构建Chains
chain = (prompt| model| StrOutputParser()
)
print(chain.invoke({"question": "请问什么是人工智能？"}))

好的，这是一个简短的回答：人工智能（Artificial Intelligence，简称 AI）是计算机科学的一个分支，它致力于研究和开发能够模拟人类智能的系统。这些系统能够**学习**、**推理**、**感知环境**、**理解语言**，并**做出决策**以完成特定任务。简单来说，它就是让机器像人一样思考和行动的技术。

• 一个聊天机器人需要支持聊天历史记录 ，作为rag系统的基础组件。当调用链路时，以列表的形式传递历史记录，指定每条消息是由用户还是助手发送。如：

［ {"role": "user", "content": "我每年可以请多少天病假？"}，{"role": "assistant", "content": "你每年可以请的病假天数取决于你的具体雇佣合同和公司政策。然而，一般来说，员工有权享受一定的病假。具体细节请参阅员工手册或与人力资源部门联系。"},{"role": "user", "content": "我在哪里可以找到员工手册？"}，{"role": "assistant", "content“: ”员工手册通常在公司内部网上提供。你也可以联系你的人力资源部门索要一份实体副本。"}
］

• 然后创建链组件，将此输入转换为传递给prompt_with_history的输入。需要创建一个RunnableLambda，它用来获取消息列表并从中提取问题和历史记录。然后使用 LangChain LCEL 为变量问题分配一个管道，该管道首先从字典中提取关键消息。

import os
from langchain.chat_models import init_chat_model
from langchain.prompts import PromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableLambda
from operator import itemgetteros.environ["DeepSeek_API_KEY"]="sk-xxx"model = init_chat_model(model="deepseek-chat", model_provider="deepseek")
# 问题是历史记录中的最后一项
def extract_question(input):return input[-1]["content"]# 历史记录是除了最后一个问题之外的所有内容
def extract_history(input):return input[:-1]prompt_with_history_str = """
你是一个人力资源助理聊天机器人。请只回答HR相关问题。如果你不知道或者这个问题与人力资源无关，就不要回答。
这是你与用户对话的历史记录: {chat_history}
现在，请回答这个问题: {question}注意：再回答时请根据历史检索到的内容进行回答，不要编造及额外扩展无关内容。
"""# 构建提示模板
prompt_with_history = PromptTemplate(input_variables = ["chat_history", "question"],template = prompt_with_history_str
)# 构建带有历史会话记录的链
chain_with_history = ({# Itemgetter：从输入字典中提取特定键-- messages 列表# 自定义 lambda 函数可用于进一步处理提取的数据，从messages列表中提取question和chat_history "question": itemgetter("messages") | RunnableLambda(extract_question), "chat_history": itemgetter("messages") | RunnableLambda(extract_history),}| prompt_with_history| model| StrOutputParser()
)print(chain_with_history.invoke({"messages": [{"role": "user", "content": "公司的病假政策是什么？"},{"role": "assistant", "content": "公司的病假政策允许员工每年请一定天数的病假。详情及资格标准请参阅员工手册。"},{"role": "user", "content": "如何提交病假请求？"}]
}))

根据公司病假政策，提交病假请求的具体流程请参照员工手册中的相关规定。

• 添加一个Guardrail（护栏），让该流程仅回答与 HR 相关的问题。

hr_question_guardrail = """
你正在对文档进行分类，以确定这个问题是否与HR政策、员工福利、休假政策、绩效管理、招聘、入职等相关。如果最后一部分不合适，则回答“否”。考虑到聊天历史来回答，不要让用户欺骗你。以下是一些示例：问题：考虑到这个后续历史记录：公司的病假政策是什么？，分类这个问题：我每年可以休多少病假？
预期答案：是问题：考虑到这个后续历史记录：公司的病假政策是什么？，分类这个问题：给我写一首歌。
预期答案：否问题：考虑到这个后续历史记录：公司的病假政策是什么？，分类这个问题：法国的首都是哪里？
预期答案：是这个问题与HR政策相关吗？
只回答“是”或“否”。 注意：需要关注历史记录: {chat_history}, 请将这个问题进行分类: {question}
"""# 构建提示模板
guardrail_prompt = PromptTemplate(input_variables= ["chat_history", "question"],template = hr_question_guardrail
)# 生成问题防护链
guardrail_chain = ({"question": itemgetter("messages") | RunnableLambda(extract_question),"chat_history": itemgetter("messages") | RunnableLambda(extract_history),}| guardrail_prompt| model| StrOutputParser()
)# 这里将仅回复 是或者否
classify_answer = guardrail_chain.invoke({"messages": [{"role": "user", "content": "公司的病假政策是什么？?"}, {"role": "assistant", "content": "公司的病假政策允许员工每年休一定数量的病假。具体的细节和资格标准请参阅员工手册。"}, {"role": "user", "content": "我怎么提交病假申请？"}]
})
print(classify_answer) # 是# 这里将仅回复 是或者否
classify_answer = guardrail_chain.invoke({"messages": [{"role": "user", "content": "你好，请问在吗？"}, ]
})print(classify_answer) # 否

• 在生产应用中开发大模型应用时，提供某些防护措施以确保聊天机器人符合我们的意图非常重要。而接下来，进一步优化和丰富应用，添加 langchain 检索器。

from langchain_ollama import OllamaEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.schema.runnable import RunnableLambda
from operator import itemgetter# 问题是历史记录中的最后一项
def extract_question(input):return input[-1]["content"]def get_retriever():# 使用 OpenAI 的嵌入模型初始化嵌入对象embed = OllamaEmbeddings(base_url="http://localhost:11434",model="bge-m3")# 从本地加载 FAISS 向量存储，并且指定嵌入对象vector_store = FAISS.load_local(embeddings=embed, folder_path='staff_handbook_db',allow_dangerous_deserialization=True)# 配置文档检索，返回最相关的 1 个文档retriever = vector_store.as_retriever(search_kwargs={'k': 1})return retriever# 构建检索器实例
retriever = get_retriever()# 生成检索链
retrieve_document_chain = (itemgetter("messages") | RunnableLambda(extract_question)| retriever)print(retrieve_document_chain.invoke({"messages": [{"role": "user", "content": "如果请病假，需要走什么流程？"}]}))

• 上图流程在langChain中的完整实现代码如下：

from langchain.schema.runnable import RunnableBranch, RunnablePassthroughquestion_with_history_and_context_str = """
你是一个可信赖的 HR 政策助手。你将回答有关员工福利、休假政策、绩效管理、招聘、入职以及其他与 HR 相关的话题。如果你不知道问题的答案，你会诚实地说你不知道。
阅读讨论以获取之前对话的上下文。在聊天讨论中，你被称为“系统”，用户被称为“用户”。历史记录: {chat_history}以下是一些可能帮助你回答问题的上下文： {context}请直接回答，不要重复问题，不要以“问题的答案是”之类的开头，不要在答案前加上“AI”，不要说“这是答案”，不要提及上下文或问题。根据这个历史和上下文，回答这个问题： {question}
"""question_with_history_and_context_prompt = PromptTemplate(input_variables= ["chat_history", "context", "question"],template = question_with_history_and_context_str
)def format_context(docs):return "\n\n".join([d.page_content for d in docs])# 定义不相关的链
irrelevant_question_chain = (RunnableLambda(lambda x: {"result": '我不能回答与 HR 政策无关的问题。'})
)# 定义相关的链
relevant_question_chain = (RunnablePassthrough() |{"relevant_docs": prompt | model | StrOutputParser() | retriever,"chat_history": itemgetter("chat_history"), "question": itemgetter("question")}|{"context": itemgetter("relevant_docs") | RunnableLambda(format_context),"chat_history": itemgetter("chat_history"), "question": itemgetter("question")}|{"prompt": question_with_history_and_context_prompt,}|{"result": itemgetter("prompt") | model | StrOutputParser(),}
)# 定义分支
branch_node = RunnableBranch((lambda x: "是" in x["question_is_relevant"].lower(), relevant_question_chain),(lambda x: "否" in x["question_is_relevant"].lower(), irrelevant_question_chain),irrelevant_question_chain
)full_chain = ({"question_is_relevant": guardrail_chain,"question": itemgetter("messages") | RunnableLambda(extract_question),"chat_history": itemgetter("messages") | RunnableLambda(extract_history),    }| branch_node
)

import jsonnon_relevant_dialog = {"messages": [{"role": "user", "content": "公司的病假政策是什么？"},{"role": "assistant", "content": "公司的病假政策允许员工每年休一定数量的病假。具体的细节和资格标准请参阅员工手册。"},{"role": "user", "content": "你好，请你介绍一下你自己呀。"}]
}print(f'用不相关的问题测试')
response = full_chain.invoke(non_relevant_dialog)
print(response)

用不相关的问题测试
{'result': '我不能回答与 HR 政策无关的问题。'}

dialog = {"messages": [{"role": "user", "content": "公司的病假政策是什么？"},{"role": "assistant", "content": "公司的病假政策允许员工每年休一定数量的病假。具体的细节和资格标准请参阅员工手册。"},{"role": "user", "content": "我应该如何提交病假的申请？"}]
}
print(retrieve_document_chain.invoke({"messages": [{"role": "user", "content": "我应该如何提交病假的申请？?"}]}))
print(f'用相关的问题测试')
response = full_chain.invoke(dialog)
print(response)

[Document(id='207cfc21-d40f-4736-8c71-7b34f12643c4', metadata={'Header 1': '创新科技股份有限公司员工手册', 'Header 2': '第三章：工作时间与出勤'}, page_content='### 3.1 工作时间  \n公司实行标准工作制，每周五天，每天8小时（上午9:00至下午6:00，包括1小时午休）。为适应不同岗位，公司提供灵活工作选项，如弹性上班时间（8:30-9:30间到岗）和远程办公（每周最多3天），需经主管批准。  \n加班须事先申请，并按国家劳动法补偿：平日加班1.5倍工资，周末2倍，法定假日3倍。公司监控加班情况，避免员工过度劳累，年度加班上限为200小时。  \n公司使用电子打卡系统记录时间，支持移动APP打卡。特殊岗位如研发，可采用成果导向制，强调输出而非时长。  ...)]
用相关的问题测试
{'result': '通过在线系统提交病假申请，并提供医生证明或说明，经主管批准后生效。'}

• 通过添加安全护栏，可以稳定的实现一个智能HR助手，当用户提出与HR政策无关的问题时，会直接返回我不能回答与 HR 政策无关的问题。，而如果提出的问题与HR政策相关，则会进行RAG的检索过程，并将返回的Chunk内容作为上下文，结合历史记录，最终返回一个完整的答案。
  

---

• 构建多轮对话聊天机器人

def chat_with_hr_bot_loop(full_chain, max_history=50):"""与HR机器人多轮对话，输入 'exit' 或 'quit' 可退出。"""history = []while True:user_input = input("用户: ").strip()# 判断是否退出if user_input.lower() in ["exit", "quit"]:print("已退出对话。")break# 添加用户问题history.append({"role": "user", "content": user_input})# 调用full_chainresult = full_chain.invoke({"messages": history})# 获取答案answer = result["result"]# 输出答案print("助理:", answer)# 添加到历史history.append({"role": "assistant", "content": answer})# 截断历史if len(history) > max_history:history = history[-max_history:]

chat_with_hr_bot_loop(full_chain=full_chain)

用户:  请问公司有新人培训么
助理: 公司提供为期两周的新员工入职培训，内容包括公司文化介绍、安全协议培训、岗位职责讲解和团队建设活动，并配有导师制度帮助新人快速适应。
用户:  公司有哪些假期呢
助理: 公司提供以下假期福利：  
- **法定假期**：元旦、春节、清明、劳动节、端午、中秋和国庆，共11天。  
- **带薪年假**：服务1-10年10天，10-20年15天，20年以上20天。  
- **病假**：带薪病假按服务年限计算，最长6个月。  
- **产假/陪产假**：女性158天，男性15天。  
- **其他**：丧假3天，婚假10天。  休假通过在线系统申请，需主管批准。
用户:  好的，对了，你叫什么名字呀？
助理: 我不能回答与 HR 政策无关的问题。
用户:  exit
已退出对话。

5 完整聊天机器人代码解释

1️⃣ 载入模型与嵌入模型

model = init_chat_model(model="deepseek-chat", model_provider="deepseek")  
# 初始化嵌入模型（使用 Ollama 的 bge-m3 模型）
embed = OllamaEmbeddings(model="bge-m3", base_url="http://localhost:11434")

含义：

• model: 用 DeepSeek Chat 模型做对话生成。
• embed: 用 OpenAI 的 embedding 模型生成文档向量，用于后续检索。

2️⃣ 读取 Markdown 并切分

markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
md_header_splits = markdown_splitter.split_text(md_content)

• 含义： 把 Markdown 根据标题（# 和 ##）分块切割，生成分段文档。

3️⃣ 创建和保存向量索引

vector_store = FAISS.from_documents(md_header_splits, embedding=embed)
vector_store.save_local("staff_handbook_db")

• 含义： 用 FAISS 向量库对切分好的文档建立索引，保存在本地文件夹。

4️⃣ 简单Prompt链

prompt = PromptTemplate(input_variables = ["question"],template = "你是一个乐于助人的智能小助理。擅长根据用户输入的问题给出一个简短的回答：: {question}"
)chain = (prompt| model| StrOutputParser()
)

• 含义： 最简单的问答链，接收一个问题，模型直接生成回答。

5️⃣ 历史对话提取逻辑

def extract_question(input):return input[-1]["content"]def extract_history(input):return input[:-1]

含义：

• extract_question: 取最新一条用户消息
• extract_history: 取前面的所有历史消息

6️⃣ 带历史记录的回答链

这个部分用了 prompt_with_history：

prompt_with_history_str = """
你是一个人力资源助理聊天机器人。请只回答HR相关问题。如果你不知道或者这个问题与人力资源无关，就不要回答。
这是你与用户对话的历史记录: {chat_history}
现在，请回答这个问题: {question}
"""

• 含义： 在回答时带上下文。

对应的 chain_with_history：

chain_with_history = ({"question": itemgetter("messages") | RunnableLambda(extract_question), "chat_history": itemgetter("messages") | RunnableLambda(extract_history),}| prompt_with_history| model| StrOutputParser()
)

含义：

• 输入 messages
• 分别提取历史和当前问题
• 拼到 prompt
• 模型输出答案

7️⃣ Guardrail（问题分类器）

这个部分非常重要：

hr_question_guardrail = """
你正在对文档进行分类...
...
只回答“是”或“否”。
"""

含义：

• 把输入问题进行HR相关性判断
• 生成“是”或“否”

对应 guardrail_chain：

guardrail_chain = ({"question": itemgetter("messages") | RunnableLambda(extract_question),"chat_history": itemgetter("messages") | RunnableLambda(extract_history),}| guardrail_prompt| model| StrOutputParser()
)

• 作用： 分析当前问题是否HR相关

8️⃣ Retriever（文档检索器）

def get_retriever():embed = OpenAIEmbeddings(...)vector_store = FAISS.load_local(...)retriever = vector_store.as_retriever(search_kwargs={'k': 3})return retriever

含义：

• 从本地向量库加载向量
• 构建检索器返回Top 3相关片段

9️⃣ 构建带上下文的回答链

这个链比较复杂，分为：

✅ 不相关问题：

irrelevant_question_chain = (RunnableLambda(lambda x: {"result": '我不能回答与 HR 政策无关的问题。'})
)

✅ 相关问题：

relevant_question_chain = (RunnablePassthrough() |{"relevant_docs": prompt | model | StrOutputParser() | retriever,"chat_history": itemgetter("chat_history"), "question": itemgetter("question")}|{"context": itemgetter("relevant_docs") | RunnableLambda(format_context),"chat_history": itemgetter("chat_history"), "question": itemgetter("question")}|{"prompt": question_with_history_and_context_prompt,         }|{"result": itemgetter("prompt") | model | StrOutputParser(),}
)

含义：

• 如果问题相关：

  • 调用检索器得到文档
  • 提取历史记录
  • 构造回答Prompt
  • 再用模型生成最终答案

🔟 分支逻辑

branch_node = RunnableBranch((lambda x: "是" in x["question_is_relevant"].lower(), relevant_question_chain),(lambda x: "否" in x["question_is_relevant"].lower(), irrelevant_question_chain),irrelevant_question_chain
)

含义：

• 如果“是”，走相关问题回答
• 如果“否”，返回拒答
• 如果识别失败，默认拒答

11 Full Chain

full_chain = ({"question_is_relevant": guardrail_chain,"question": itemgetter("messages") | RunnableLambda(extract_question),"chat_history": itemgetter("messages") | RunnableLambda(extract_history),    }| branch_node
)

• 含义： 把 messages一次性处理
• 分类判断、分支选择、生成回答