强强联合！LangChain与CrewAI构建基于RAG的智能查询解答系统

在当今数字化时代，企业和教育机构每天都会收到海量的咨询问题。无论是客户支持、销售团队的提问，还是内部员工的咨询，手动回复这些问题不仅耗时费力，还容易出现回答不一致的情况。而基于人工智能的查询解答系统，能够快速、准确且高效地提供答案，极大地提升了工作效率和用户体验。

今天，我们就来聊聊如何利用LangChain、ChromaDB和CrewAI构建一个基于检索增强生成（RAG）的智能查询解答系统。这个系统不仅能自动处理各种问题，还能生成精准的回答，帮助企业和教育机构更好地服务用户。

为什么我们需要AI驱动的查询解答系统？

在传统的业务流程中，手动回复客户或学员的咨询问题是一个极其低效的过程。客户希望得到即时的回复，而企业则需要快速获取准确的信息来做出决策。AI驱动的查询解答系统通过自动化处理这些问题，不仅减轻了人工负担，还能提供一致且高质量的回答。

在客户服务领域，AI系统可以自动回复常见问题，提升客户满意度；在销售和市场营销中，它可以实时提供产品细节和客户洞察；在金融、医疗、教育和电商等行业，AI驱动的查询处理能够确保操作顺畅，提升用户体验。

深入了解RAG工作流程

在动手构建系统之前，我们先来了解一下检索增强生成（RAG）系统是如何工作的。RAG架构主要分为三个关键阶段：索引、检索和生成。

1. 构建向量存储（文档处理与存储）

系统首先需要处理和存储相关文档，以便能够快速检索。具体步骤如下：

• 文档切分：将大型文档切分为更小的文本块，以便高效检索。
• 嵌入模型：利用基于AI的嵌入模型将这些文本块转换为向量表示。
• 向量存储：将向量化的数据索引并存储在数据库（如ChromaDB）中，以便快速查找。

2. 查询处理与检索

当用户提交问题时，系统会先检索相关数据，然后再生成回答。具体步骤如下：

• 用户查询输入：用户提交问题或请求。
• 向量化：利用嵌入模型将查询转换为数值向量。
• 搜索与检索：系统在向量存储中搜索最相关的文本块并检索出来。

3. 增强与回答生成

为了生成准确的回答，系统会将检索到的数据与原始查询结合。具体步骤如下：

• 增强查询：将检索到的文档块与原始查询结合。
• LLM处理：利用大型语言模型（LLM）根据查询和检索到的上下文生成最终回答。
• 最终回答：系统向用户提供一个准确且富有上下文的回答。

构建基于RAG的查询解答系统

接下来，我们将通过一个实际案例，展示如何构建一个基于RAG的查询解答系统。这个系统将高效地回答学员的问题，帮助他们更好地学习。

选择合适的数据用于查询解答

在构建RAG系统之前，最重要的就是数据。一个结构良好的知识库是关键，因为回答的准确性和相关性完全依赖于数据的质量。以下是一些适合不同类型用途的数据：

• 客户支持数据：常见问题解答（FAQ）、故障排除指南、产品手册和过去的客户互动记录。
• 销售与市场数据：产品目录、价格详情、竞争对手分析和客户咨询记录。
• 内部知识库：公司政策、培训文档和标准操作流程（SOP）。
• 财务与法律文件：合规指南、财务报告和监管政策。
• 用户生成内容：论坛讨论、聊天记录和反馈表单，这些都能提供真实的用户问题。

在我们的学员查询解答系统中，我们尝试了多种数据类型，最终发现使用课程视频的字幕是最有效的方法。字幕提供了与学员问题直接相关的结构化和详细内容，能够快速生成相关答案。

构建查询解答系统的架构

在动手编写代码之前，我们需要先规划系统的架构。系统需要完成以下三个主要任务：

1. 从字幕文件（SRT）中提取并存储课程内容。
2. 根据学员的查询检索相关的课程材料。
3. 利用AI驱动的代理生成结构化的回答。

为了实现这些功能，我们将系统分为三个组件：

• 字幕处理：从SRT文件中提取文本，处理并将其嵌入存储到ChromaDB中。
• 检索：根据学员的查询搜索并检索相关的课程材料。
• 查询回答代理：利用CrewAI生成结构化且准确的回答。

实现步骤

现在，我们已经规划好了系统的架构，接下来就是动手实现。

1. 导入必要的库

构建AI驱动的学习支持系统，首先需要导入一些关键的库：

importpysrt
fromlangchain.text_splitterimportRecursiveCharacterTextSplitter
fromlangchain.schemaimportDocument
fromlangchain.embeddingsimportOpenAIEmbeddings
fromlangchain.vectorstoresimportChroma
fromcrewaiimportAgent, Task, Crew
importpandasaspd
importast
importos
importtime
fromtqdmimporttqdm

这些库的作用如下：

• pysrt：用于从SRT字幕文件中提取文本。
• RecursiveCharacterTextSplitter：将大段文本切分为更小的块，以便更好地检索。
• Document：表示结构化的文本文档。
• OpenAIEmbeddings：将文本转换为数值向量，用于相似性搜索。
• Chroma：将嵌入存储在向量数据库中，便于高效检索。
• CrewAI（Agent、Task、Crew）：定义处理学员查询的AI代理。
• pandas：以DataFrame形式处理结构化数据。
• ast：将基于字符串的数据结构解析为Python对象。
• os：提供系统级操作，如读取环境变量。
• tqdm：在长时间运行的任务中显示进度条。

2. 设置环境

为了使用OpenAI的API进行嵌入，我们需要加载API密钥并配置模型设置。

步骤1：从本地文件中读取API密钥

withopen('/home/janvi/Downloads/openai.txt','r')asfile:
  openai_api_key = file.read()

步骤2：将API密钥存储为环境变量

os.environ['OPENAI_API_KEY'] = openai_api_key

步骤3：指定OpenAI模型

os.environ["OPENAI_MODEL_NAME"] ='gpt-4o-mini'

通过这些配置，我们可以确保系统能够高效地处理和存储嵌入。

3. 提取并存储字幕数据

字幕文件中包含了视频讲座的宝贵信息，是AI检索系统中结构化内容的丰富来源。有效地提取和处理字幕数据，能够让我们在回答学员问题时快速检索到相关信息。

步骤1：从SRT文件中提取文本

我们使用pysrt库从SRT文件中提取文本，并将其组织成结构化的形式，以便进一步处理和存储。

defextract_text_from_srt(srt_path):
 """从SRT字幕文件中提取文本"""
  subs = pysrt.open(srt_path)
  text =" ".join(sub.textforsubinsubs)
 returntext

由于课程可能包含多个字幕文件，我们需要系统地组织和迭代这些文件，以便无缝提取文本。

course_folders = {
 "深度学习入门（使用PyTorch）":"C:\M\Code\GAI\Learn_queries\Subtitle_Introduction_to_Deep_Learning_Using_Pytorch",
 "构建生产级RAG系统（使用LlamaIndex）":"C:\M\Code\GAI\Learn_queries\Subtitle of Building Production-Ready RAG systems using LlamaIndex",
 "LangChain入门（构建生成式AI应用与代理）":"C:\M\Code\GAI\Learn_queries\Subtitle_introduction_to_langchain_using_agentic_ai"
}

course_srt_files = {}

forcourse, folder_pathincourse_folders.items():
  srt_files = []
 forroot, _, filesinos.walk(folder_path):
    srt_files.extend(os.path.join(root, file)forfileinfilesiffile.endswith(".srt"))
 ifsrt_files:
    course_srt_files[course] = srt_files

这些提取的文本将成为我们AI驱动学习支持系统的基础，使我们能够进行高级检索和查询解答。

步骤2：将字幕存储到ChromaDB

接下来，我们将课程字幕存储到ChromaDB中，包括文本切分、嵌入生成、持久化存储和成本估算。

（1）为ChromaDB设置持久化目录

persist_directory是一个文件夹路径，用于保存存储的数据。这样即使程序重新启动，嵌入数据也能保留下来。

persist_directory ="./subtitles_db"

（2）将文本切分为更小的块

大型文档（如整个课程字幕）可能会超出嵌入的标记限制。为了处理这种情况，我们使用RecursiveCharacterTextSplitter将文本切分为更小的、有重叠的块，以提高搜索精度。

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)

每个块的长度为1000个字符，为了在块之间保留上下文，我们将前一个块的200个字符包含在下一个块中。这种重叠有助于保留重要细节，提高检索精度。

（3）初始化OpenAI嵌入和ChromaDB向量存储

我们需要将文本转换为数值向量表示，以便进行相似性搜索。OpenAI的嵌入功能允许我们将课程内容编码为可以高效搜索的格式。

embeddings = OpenAIEmbeddings(openai_api_key=openai_api_key)

这里，OpenAIEmbeddings()使用我们的OpenAI API密钥初始化嵌入模型，确保每段文本都能转换为高维向量表示。

（4）初始化ChromaDB

现在，我们将这些向量嵌入存储到ChromaDB中。

vectorstore = Chroma(
  collection_name="course_materials",
  embedding_function=embeddings,
  persist_directory=persist_directory
)

collection_name="course_materials"在ChromaDB中创建了一个专门的集合，用于组织所有与课程相关的嵌入。embedding_function=embeddings指定了OpenAI嵌入用于将文本转换为数值向量。persist_directory=persist_directory确保所有存储的嵌入在程序重新启动后仍然可用。

（5）估算存储课程数据的成本

在将文档添加到向量数据库之前，估算标记使用成本是非常重要的。由于OpenAI按每1000个标记收费，我们需要提前估算成本，以便高效管理开支。

COST_PER_1K_TOKENS =0.0001# 每1000个标记的成本（使用'text-embedding-ada-002'模型）
TOKENS_PER_CHUNK_ESTIMATE =750# 每1000字符块的估计标记数

total_tokens =0
total_cost =0

start_time = time.time()

COST_PER_1K_TOKENS=0.0001定义了使用OpenAI嵌入时每1000个标记的成本。TOKENS_PER_CHUNK_ESTIMATE=750估计每个1000字符块包含约750个标记。total_tokens和total_cost变量用于跟踪整个执行过程中处理的数据量和产生的成本。start_time变量记录了开始时间，用于测量整个过程的耗时。

（6）检查并添加课程到ChromaDB

我们希望避免重新处理已经存储在向量数据库中的课程。因此，我们先查询ChromaDB，检查课程是否已经存在。如果不存在，我们则提取并存储其字幕数据。

forcourse, srt_listincourse_srt_files.items():
  existing_docs = vectorstore._collection.get(where={"course": course})
 ifnotexisting_docs['ids']:
    srt_texts = [extract_text_from_srt(srt)forsrtinsrt_list]
    course_text ="\n\n\n\n".join(srt_texts)
    doc = Document(page_content=course_text, metadata={"course": course})
    chunks = text_splitter.split_documents([doc])

字幕通过extract_text_from_srt()函数提取，多个字幕文件通过\n\n\n\n连接，以提高可读性。创建了一个Document对象，存储完整的字幕文本及其元数据。最后，使用text_splitter.split_documents()将文本切分为更小的块，以便高效处理和检索。

（7）估算标记使用量和成本

在将块添加到ChromaDB之前，我们估算成本。

chunk_count = len(chunks)
batch_tokens = chunk_count * TOKENS_PER_CHUNK_ESTIMATE
batch_cost = (batch_tokens /1000) * COST_PER_1K_TOKENS
total_tokens += batch_tokens
total_cost += batch_cost

chunk_count表示切分后的块数量。batch_tokens根据块数量估算总标记数。batch_cost计算当前课程的处理成本。total_tokens和total_cost累加每次处理的值，以跟踪整体处理量和开支。

（8）将块添加到ChromaDB

vectorstore.add_documents(chunks)
print(f"已添加课程：{course}(块数：{chunk_count}, 成本：${batch_cost:.4f})")

处理后的块被存储到ChromaDB中，以便高效检索。程序会显示添加的块数和估算的处理成本。

如果课程已经存在，则会显示以下信息：

print(f"课程已存在：{course}")

一旦所有课程处理完成，我们计算并显示最终结果。

end_time = time.time()

print(f"\n课程嵌入更新完成！?")
print(f"总处理块数：{total_tokens // TOKENS_PER_CHUNK_ESTIMATE}")
print(f"估算总标记数：{total_tokens}")
print(f"估算总成本：${total_cost:.4f}")
print(f"总耗时：{end_time - start_time:.2f}秒")

end_time - start_time计算总处理时间。系统会显示处理的块数、估算的标记使用量、总成本以及整个嵌入过程的总结。

4. 查询并回答学员问题

一旦字幕存储到ChromaDB中，系统需要一种方式来检索相关内容，以便在学员提交问题时提供答案。这个检索过程通过相似性搜索实现，它能够识别与输入问题最相关的存储文本段。

工作原理

• 查询输入：学员提交与课程相关的问题。
• 按课程过滤：系统确保检索仅限于相关课程材料。
• ChromaDB中的相似性搜索：将查询转换为嵌入，ChromaDB检索最相似的存储文本块。
• 返回顶部结果：系统选择最相关的三个文本段。
• 格式化输出：检索到的文本被格式化并呈现为进一步处理的上下文。

defretrieve_course_materials(query: str, course):
 """按课程名称检索课程材料"""
  filter_dict = {"course": course}
  results = vectorstore.similarity_search(query, k=3, filter=filter_dict)
 return"\n\n".join([doc.page_contentfordocinresults])

例如：

course_name ="深度学习入门（使用PyTorch）"
question ="什么是梯度下降？"
context = retrieve_course_materials(query=question, course=course_name)
print(context)

从输出中可以看到，ChromaDB通过相似性搜索，根据课程名称和问题检索到最相关的信息。

为什么使用相似性搜索？

• 语义理解：与关键词搜索不同，相似性搜索能够找到与查询语义相关的文本。
• 高效检索：系统无需扫描整个文档，只需检索最相关的部分。
• 提升答案质量：通过按课程过滤并按相关性排序，学员能够获得高度针对性的内容。

这种机制确保学员提交问题时，能够从存储的课程材料中获得相关且上下文准确的信息。

5. 实现AI查询回答代理

检索到相关课程材料后，下一步是利用AI驱动的代理生成有意义的回答。我们使用CrewAI定义一个智能代理，负责分析查询并生成结构化的回答。

步骤1：定义代理

查询回答代理通过清晰的角色和背景故事来指导其行为，以便更好地回答学员的问题。

query_answer_agent = Agent(
  role="学习支持专家",
  goal="您需要为学员提供最准确的回答",
  backstory="""
  您是一家专注于数据科学、机器学习和生成式AI的在线教育公司学员查询解答部门的负责人。您负责回答学员关于课程内容、作业、技术问题和行政问题的咨询。您礼貌、圆滑，并且对可以改进的地方负有责任感。
  """,
  verbose=False
)

在代码块中，我们首先定义了代理的角色为“学习支持专家”，因为它充当虚拟助教的角色，回答学员的问题。然后，我们定义了目标，确保代理在回答时优先考虑准确性和清晰性。最后，我们将verbose设置为False，这样在不需要调试时，执行过程将保持安静。这种清晰定义的代理角色确保回答既有帮助性，又结构化，且符合教育平台的语气。

步骤2：定义任务

定义了代理之后，我们需要为其分配任务。

query_answering_task = Task(
  description="""
  尽您所能回答学员的问题。尽量保持回答简洁，不超过100个单词。
  这是问题：{query}

  这是从课程字幕中提取的相关内容，仅在需要时使用：{relevant_content}。
  由于这些内容是从课程字幕中提取的，可能存在拼写错误，请在回答中纠正这些错误。

  这是与学员之前的对话记录：{thread}。
  在对话中，以“学员”开头的是学员的问题，以“支持”开头的是您的回答。请根据之前的对话适当调整您的回答。

  这是学员的全名：{learner_name}。
  如果不确定学员的名字，直接用“嗨”开头。
  在回答的结尾添加一些适当的、鼓励性的安慰语句，例如“希望您觉得有帮助”、“希望这些信息有用。继续努力！”、“很高兴能帮到您！随时联系我。”等。

  如果您不确定答案，请注明：“抱歉，我不确定这个问题的答案，我会稍后回复您。”
  """,
  expected_output="简洁准确的回答",
  agent=query_answer_agent
)

接下来，我们来分解分配给AI的任务。处理学员的查询时，{query}代表学员的问题。回答应简洁（不超过100个单词）且准确。如果需要使用课程内容，{relevant_content}是从存储在ChromaDB中的字幕中提取的，AI必须在回答中纠正任何拼写错误。

如果存在之前的对话，{thread}有助于保持连贯性。学员的问题以“学员”开头，而之前的回答以“支持”开头，这使得代理能够提供与上下文相关的回答。通过{learner_name}实现个性化——代理会用学员的名字称呼他们，如果不确定名字，就简单地用“嗨”开头。

为了使回答更具吸引力，AI会在结尾添加一句积极的结束语，比如“希望您觉得有帮助！”或者“随时联系我。”如果AI不确定答案，它会明确说明：“抱歉，我不确定这个问题的答案，我会稍后回复您。”这种方法确保了回答的礼貌性、清晰性和结构化，提升了学员的参与度和信任感。

步骤3：初始化CrewAI实例

现在我们已经定义了代理和任务，接下来初始化CrewAI，它能够动态处理学员的查询。

response_crew = Crew(
  agents=[query_answer_agent],
  tasks=[query_answering_task],
  verbose=False
)

agents=[query_answer_agent]将“学习支持专家”代理添加到团队中。tasks=[query_answering_task]将查询回答任务分配给这个代理。设置verbose=False可以保持输出简洁，除非需要调试。CrewAI能够动态处理多个学员的查询，使系统具有可扩展性和高效性，能够动态处理查询。

步骤4：为多个学员的查询生成回答

设置好AI代理后，我们需要动态处理存储在结构化数据集中的学员查询。

以下代码处理存储在CSV文件中的学员查询，并使用AI代理生成回答。它首先加载包含学员查询、课程详情和对话线程的数据集。reply_to_query函数提取相关细节，如学员姓名、课程名称和当前查询。如果存在之前的对话，它会提取出来以提供上下文。如果查询包含图片，则会跳过。然后，它从ChromaDB中检索相关的课程材料，并将查询、相关内容和之前的对话发送给AI代理，以生成结构化的回答。

df = pd.read_csv('C:\M\Code\GAI\Learn_queries\filtered_data_top3_courses.csv')

defreply_to_query(df, index=1):
  learner_name = df.iloc[index]["thread_starter"]
  course_name = df.iloc[index]["course"]
 ifdf.iloc[index]['number_of_replies'] >1:
    thread = ast.literal_eval(df.iloc[index]["modified_thread"])
 else:
    thread = []
  question = df.iloc[index]["current_query"]
 ifdf.iloc[index]['has_image'] ==True:
   return" "
  context = retrieve_course_materials(query=question, course=course_name)
  response_result = response_crew.kickoff(inputs={"query": question,"relevant_content": context,"thread": thread,"learner_name": learner_name})
  print('Q: ', question)
  print('\n')
  print('A: ', response_result)
  print('\n\n')

测试该函数时，我们为一个查询（index=1）执行它：

reply_to_query(df, index=1)

从输出中可以看到，它能够正常工作，仅针对一个索引生成回答。

现在，我们通过所有查询进行迭代，处理每一个查询，同时处理可能出现的错误。这确保了查询解答过程的高效自动化，能够动态处理多个学员的查询。

foriinrange(len(df)):
 try:
    reply_to_query(df, index=i)
 except:
    print("索引号出错：", i)
   continue

为什么这一步很重要？

• 自动化查询处理：系统能够高效处理多个学员的查询。
• 确保上下文相关性：回答基于检索到的课程材料和之前的对话生成。
• 可扩展性：该方法允许AI代理动态处理并回答数千个查询。
• 提升学习支持体验：学员能够收到个性化且数据驱动的回答。

这一步确保每个学员的查询都能被分析、结合上下文并有效回答，从而提升整体学习体验。

输出示例

从输出中可以看到，回答查询的过程已经实现自动化，首先是问题，然后是回答。

未来改进方向

为了进一步提升基于RAG的查询解答系统，我们可以考虑以下改进方向：

1. 常见问题及其解答：在查询解答框架内实现一个结构化的FAQ系统，能够即时回答常见问题，减少对人工支持的依赖。
2. 图像处理能力：增加分析和提取图像（如截图、图表或扫描文档）相关信息的能力，将使系统在教育和客户支持领域更具 versatility。
3. 改进图像列布尔值：完善图像列检测的逻辑，更准确地识别和处理基于图像的查询。
4. 语义切块和不同的切块技术：尝试不同的切块策略，如语义切块、固定长度分割和混合方法，可以提高检索精度和回答的上下文理解能力。

总结

这个基于RAG的查询解答系统利用LangChain、ChromaDB和CrewAI，高效地实现了学员支持的自动化。它从课程字幕中提取文本，将其嵌入存储到ChromaDB中，并通过相似性搜索检索相关内容。CrewAI代理处理查询，参考之前的对话，并生成结构化的回答，确保回答的准确性和个性化。

该系统提升了可扩展性、检索效率和回答质量，使自主学习更加互动。未来的改进方向包括多模态支持、更好的检索优化和增强的回答生成。通过自动化查询解答，这个系统简化了学习支持流程，为学员提供了更快、更具上下文意识的回答，提升了整体参与度。