应用于电网事故查询的知识检索系统

一、研究目标

知识库的构建：根据用户给定的Excel数据文件，通过Embedding模型构建知识库，同时执行向量检索和关键词检索，实现混合检索，返回与用户查询相关联的数据候选集。
工作流的开发：通过LangGraph构建一套完整的工作流程，实现从用户输入到知识检索，最后完成结果输出的工作流。

二、系统结构

2.1工作流架构

2.2项目流程

1、加载电网事故Excel数据，通过文本分割，提取Excel文件内的事故信息（事故情况、处理过程）并用Document对象存储。创建一个事故情况到处理过程的映射词典，用于后续的精确匹配。

2、通过Embedding模型构建知识库，分别生成语义检索器和关键词检索器，按照设定的关键词与语义比例进行混合检索，返回与用户输入（事故情况）最相关的k个结果，构成候选结果集。

3、进入工作流查询生成节点，判断用户输入是否需要进行检索，如果需要检索，则进入生成回复节点，否则直接响应。

4、进入生成回复节点后，调用大模型根据提示词从混合检索得到的结果集中寻找与用户输入相匹配的事故情况。

4.1、如果匹配成功，则直接原文输出该事故情况对应的处理情况。

4.2、如果匹配失败，则转交用户选择，进入重试检索节点，再次从检索结果中选择最匹配的项，或者进入精确匹配节点，跳过大模型，直接从映射词典中寻找事故匹配项。

5、输出用户输入对应的事故处理过程。

三、关键实现细节

3.1工作流程图

3.2实现细节

3.2.1配置环境

1、编程语言：python3.12.9

2、开发环境：PyCharm 2025.1.1.1

3、大模型：qwen2.5:32b

4、向量模型：bge-m3

5、导入依赖：

pandas

langchain

langgraph

langchain_core

langchain_ollama

langchain_openai

langchain_community

3.2.2 用户数据处理

通过pandas库的read_excel方法读取excel数据，以行为单位格式化excel中的两列数据（事故情况和处理过程），创建Document文档对象，逐行数据添加到文档对象中。

def read_excel_to_documents(file_path):
    """读取Excel文件并将每行转换为独立文档"""
    df = pd.read_excel(file_path)
    documents = []
    for index, row in df.iterrows():
        # 构建文档内容，将事故情况和处理过程格式化
        content = f"事故情况：{row[0]}\n处理过程：{row[1]}"
        # 创建文档对象，包含内容和元数据
        doc = Document(
            page_content=content,
            metadata={
                "source": file_path,
                "row_index": index,
                "事故情况": row[0],
                "处理过程": row[1]
            }
        )
        documents.append(doc)
    return documents

3.2.3构建映射词典

在构建Document文档时，创建事故情况到处理过程的映射词典，用于后续无法通过大模型找到匹配项时，使用精确匹配直接从词典中找到对应的事故处理过程，确保用户能够得到想要的查询结果。

def create_accident_mapping(documents):
    """创建事故情况到处理过程的映射字典"""
    accident_to_process = {}
    for doc in documents:
        accident = doc.metadata["事故情况"]
        process = doc.metadata["处理过程"]
        accident_to_process[accident] = process
    return accident_to_process

# 创建映射字典
accident_mapping = create_accident_mapping(docs)

3.2.4构建混合检索

通过Embedding模型（bge-m3）构建向量存储，利用embedding的语义特征，计算向量之间的相似度来评估语句之间的关系。

# 创建向量存储
vectorstore = InMemoryVectorStore.from_documents(
    documents=docs,
    embedding=embeddings
)

vector_retriever = vectorstore.as_retriever(
    search_kwargs={"k": 2},
    lambda_mult=1
)

使用BM25Retriever构建关键词查询器，根据词频、逆文档频率以及文档长度归一化三个维度，计算查询与Document的相关性分数，最终根据分数排序返回最相关的结果。

同时执行语义查询和关键词查询，按照[关键词：语义] = [0.8 : 0.2]比例计算，从两类查询结果中选择匹配用户问题的最佳结果。

bm25_retriever = BM25Retriever.from_documents(docs)
bm25_retriever.k = 2
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.8, 0.2]
)

retriever_tool = create_retriever_tool(
    ensemble_retriever,
    "混合检索事故",
    "输入事故情况，根据这段原文结合关键词和语义搜索知识库并返回事故情况的对应处理过程"
)

3.2.5查询生成节点

根据用户输入，判断是否需要进行知识库检索，如果是电网相关的用户问题，则会调用3.2.4步骤的混合检索工具，进入后续工作流，如果是无关输入，则不会调用混合检索工具并直接返回输出。

def generate_query_or_respond(state: MessagesState):
    response = (
        response_model
        .bind_tools([retriever_tool]).invoke(state["messages"])
    )
    return {"messages": [response]}

3.2.6生成回复节点

按照设计好的提示词规则（仅当事故情况完全匹配知识库内容时才输出、直接复制处理过程原文、无匹配项时输出“检索失败”），使用qwen2.5:32b模型，从调用知识库检索工具检索得到的结果中试图找到与用户输入问题相匹配的事故情况及对应处理过程。如果匹配成功，则直接原文输出处理过程，如果匹配失败，则输出“检索失败”。

def generate_answer(state: MessagesState):
    """生成回复"""
    question = state["messages"][0].content
    context = state["messages"][-1].content
    prompt = PROMPT.format(question=question, context=context)
    response = response_model.invoke([{"role": "user", "content": prompt}])

    return {"messages": [response]}

3.2.7人机交互选择

如果匹配成功，则继续等待用户的下一个输入问题，匹配失败则需要进入人机交互环节（human_choice），由用户选择进行重试检索或者精确匹配（终端输入1或者2进行选择），当用户输入1、2以外的内容时，提醒用户输入正确选项。

# 人机交互选择
def human_choice(state: MessagesState):
    """判断是否需要使用精确匹配或重新检索"""
    last_mess = state["messages"][-1].content
    if "检索失败" in last_mess:
        print("\n知识检索失败，你可以选择：")
        print("1. 精确匹配（exact_match）")
        print("2. 重新检索（retry_retrieve）")
        while True:
            sel = input("请输入你的选择（1或2）：")
            if sel == "1":
                return "exact_match"
            elif sel == "2":
                return "retry_retrieve"
            else:
                print("输入无效，请重新输入。")
    return END

3.2.8精确匹配节点

由于知识检索工具可能无法返回目标匹配项，或者调用LLM时不能匹配正确的事故情况，为了确保能够百分百对用户的输入做出正确的回答，用户可以选择使用精确匹配，在RAG搜索失败时，直接根据用户的输入从映射词典中寻找对应的匹配项，搜索到匹配项后返回对应的事故处理过程。

def exact_match(state: MessagesState):
    """精确匹配节点，当RAG失败时直接查找处理过程"""
    user_query = state["messages"][0].content.strip()
    # 尝试精确匹配
    if user_query in accident_mapping:
        return {"messages": [AIMessage(content=accident_mapping[user_query])]}
    for accident, process in accident_mapping.items():
        # 去除空格和标点后比较
        clean_query = re.sub(r'[^\w]', '', user_query.lower())
        clean_accident = re.sub(r'[^\w]', '', accident.lower())
        if clean_query == clean_accident:
            return {"messages": [AIMessage(content=process)]}

3.2.9重试检索节点

此前匹配失败，可能是生成回复节点中提示词表述不清晰，导致LLM未能从候选集中找到匹配项，本节点中修改提示词内容，不要求完全匹配，而是从知识检索工具中得到的候选集中找到匹配程度最大的事故项，并输出对应事故的处理过程。

def retry_retrieve(state: MessagesState):
    """重试检索节点，再次从检索结果中选择最匹配的项"""
    user_query = state["messages"][0].content.strip()
    context = state["messages"][-2].content
    prompt = RETRY_PROMPT.format(user_query=user_query, context=context)
    response = response_model.invoke([{"role": "user", "content": prompt}])
    return {"messages": [response]}

四、总结

4.1系统总结

在本项目中，完成了Excel文件的数据读取、知识库和混合检索的构建及LangGraph工作流等系统的设计。通过实际测试，设计了一套切实可行的应用于电网事故查询的知识检索系统。

通过用户给定的Excel数据文件，将该文件转换成Document形式，实现了知识库的混合检索，根据用户的输入返回最相关的候选集合，通过LangGraph构建的工作流能够从候选集中找到匹配事故对应的处理过程。

系统同样考虑了错误处理，当知识库检索失败或者LLM调用匹配失败时，添加了人机交互功能，让用户选择进行重试检索或者精确匹配，精确匹配保证了能够让用户得到输入的准确匹配结果。

4.2未来工作

1、关键词检索工具选择

本系统目前使用的关键词检索算法为BM25，在特定情况下仍需优化。建议将该算法与其他关键词检索算法结合，以提高知识库检索的准确率。

2、匹配失败时错误处理

当工作流调用LLM进行匹配出现检索失败时，用户可以选择精确匹配操作从映射字典中寻找匹配项，面对大数据量时，该方法的时间代价较大，需要考虑其他方法如数据库存储替换字典来进行精确匹配。