一、项目背景与工业级挑战

在工业设备售后服务领域，传统的人工客服模式面临着诸多严峻挑战。工业设备涵盖电机、变频器、PLC、传感器等大量专业设备，其售后问题往往涉及复杂的技术参数、故障代码和维修流程。客服人员需要经过长期培训才能胜任，而面对种类繁多的问题，人工客服的响应时效也难以满足工业场景对效率的高要求。随着大语言模型（LLM）技术的成熟与落地，构建一个能够精准理解工业设备领域专业问题、自动检索知识库、并给出权威回答的智能客服系统成为行业共识。

本文将深度介绍一个基于 LangGraph 实现的工业设备售后智能客服 RAG Agent 项目。该系统采用工作流编排的方式，将查询增强、话题分类、文档检索、相关性评估、响应生成等环节有机整合，实现了从用户问题到专业回答的端到端处理。更重要的是，这套系统具备"自愈"能力——当检索结果不理想时，它能够自动优化查询策略重新检索，而不是简单地返回不准确的信息。

1.1 系统界面展示

等待回答动画界面

流式消息输出界面

1.2 工业级场景的独特挑战

在深入技术实现之前，我们需要理解工业售后场景与普通客服场景的本质差异。工业领域的知识具有高度专业性和安全敏感性，简单的"检索-生成"线性流程往往难以胜任，主要体现在以下几个关键方面。

• 指代消解问题是工业对话中最常见的挑战之一。用户在描述问题时往往会省略主语或使用代词，例如"它为什么不动了"、“那个故障怎么处理”。在缺乏上下文的情况下，检索系统根本无法定位问题设备是变频器还是电机，是哪一型号的产品。这种指代不明会导致检索结果与用户实际需求严重偏离，从而生成毫无参考价值的回答。

• 知识幻觉风险是工业场景中不可忽视的安全隐患。如果向量检索找到了不相关的文档，大模型可能会基于错误的信息"一本正经地胡说八道"。在工业安全领域，一条错误的维修建议可能导致设备损坏甚至人身安全事故。因此，系统必须具备识别和过滤低质量检索结果的能力，避免将不可靠的信息传递给用户。

• 资源效率问题同样值得关注。工业客服场景中，用户的问题并非都与设备相关——可能涉及闲聊、竞争对手产品对比或个人咨询。如果对所有问题都进行知识库检索，不仅浪费计算资源，还会增加不必要的响应延迟。通过话题预过滤，系统可以快速识别并处理无关问题，将宝贵的计算资源集中在真正需要专业知识的问题上。

1.3 技术栈概览

本项目采用的技术栈经过深思熟虑，既满足了工业场景对数据安全的要求，又保证了系统的性能和可维护性。

技术层级	技术选型	说明
前端	React 18 + TypeScript + Vite	用户界面交互，提供流式响应的实时渲染
后端	FastAPI + Python 3.10+	高性能 RESTful API 服务，异步支持优秀
LLM	Ollama qwen3:8B	本地大语言模型部署，数据不出域
Embedding	bge-m3:latest	中文向量嵌入模型，专为中文场景优化
向量数据库	ChromaDB	轻量级向量存储，支持持久化
工作流编排	LangGraph + LangChain	多阶段工作流管理，支持条件路由和循环
Markdown渲染	markdown-it	前端富文本展示，支持结构化内容呈现

选择 Ollama + qwen3:8B 的组合是出于工业客户对数据隐私的极高要求。通过本地部署方式，企业的敏感技术资料和客户数据完全不需要外传，有效规避了数据泄露风险。同时，Qwen 3 系列模型在中文专业术语理解方面表现出色，能够准确理解工业领域的各种技术表述。

1.4 项目整体架构

从宏观视角来看，整个系统可以分为四个层次：前端交互层负责用户界面的呈现和 SSE 流式响应的实时渲染；后端服务层提供 API 接口并运行 LangGraph 工作流；知识库层存储经过向量化的专业文档，支持语义检索；LLM 层提供对话生成和文本理解能力。

二、LangGraph 工作流架构设计

2.1 从"链"到"图"的思维跃迁

在传统的 LangChain 应用中，开发者通常使用 Chain（链） 来构建处理流程。链是线性的，像传送带一样从 A 点到 B 点，依次执行预定义的操作。这种设计在处理简单任务时足够有效，但在复杂的工业售后场景中往往捉襟见肘。

考虑这样一个典型的工业客服场景：用户询问"电机启动不了怎么办" 。系统首先需要检索相关文档，但如果检索结果不相关（可能是检索到了变频器的内容而非电机的），系统应该能够识别这个问题并重新优化查询策略。这种"检索-评估-不达标-优化查询-再检索"的闭环流程，用传统的线性链式结构几乎无法实现。

LangGraph 的引入正是为了解决这一根本性限制。LangGraph 的本质是一个带有状态的有限状态机，它将处理流程从"传送带"升级为"决策网络"。

LangGraph 的核心组件包括三个要素：

• 节点（Nodes） 负责执行具体的动作，如检索文档、生成回答或评估相关性；
• 边（Edges） 定义动作之间的基本流向，决定处理步骤的先后顺序；
• 条件边（Conditional Edges） 是整个系统的"大脑"，它根据当前状态（如相关性评估结果）动态决定下一步应该走向哪个节点。

深度分析：这种设计模拟了人类专家的思维方式。当专家面对一个问题时，如果第一次检索没有找到相关信息，他会换个关键词再搜索一遍，而不是直接告诉用户"我不知道"。LangGraph 的循环结构赋予了 Agent 这种"自我修正"和"反思"的能力，这也是本方案被称为"自愈式"RAG 的根本原因。

2.2 为什么选择 LangGraph 而非 LCEL

LangChain Expressive Language（LCEL）是 LangChain 提供的声明式链式调用语法，它能够简洁地描述"数据从 A 流到 B 再流到 C"这样的线性流程。然而，LCEL 在处理循环（Loops） 方面的能力非常有限。

在本项目的需求中，我们需要在特定条件下触发"重新检索"的循环，这需要：

1. 保存循环次数的状态（防止无限循环）
2. 根据中间结果决定是否继续循环
3. 在循环中累积和传递上下文信息

这些需求超出了 LCEL 的能力范围，而 LangGraph 的图结构天然支持条件分支和循环，非常适合实现这种复杂的控制流逻辑。

2.3 工作流整体流程设计

整个工作流采用分阶段处理的设计理念，每个阶段都有明确的职责边界和输入输出规范。

从流程图中可以清晰地看出系统的两个主要分支：话题相关时会进入知识库检索和响应生成的完整流程；话题无关时则直接使用兜底策略快速响应。当检索到文档后，系统会进行相关性评估，如果评估结果不理想，会触发查询优化循环重新检索，而不是简单地返回低质量的回答。

2.4 核心状态管理：ConversationState

在 LangGraph 中，状态（State）是连接所有节点的"血液"。每个节点接收上一个节点传递的状态，进行处理后更新状态并传递给下一个节点。良好的状态设计是系统稳定运行的基础。

class ConversationState(TypedDict):
    """
    对话状态 - 工作流中传递的状态数据结构
    
    这个 TypedDict 定义了系统运行所需的所有状态字段。
    使用 TypedDict 的好处是：
    1. 类型安全：确保节点间传递的数据类型正确
    2. 代码可读性：明确每个字段的含义和用途
    3. IDE 支持：获得更好的代码补全和错误检查
    """
    conversation_history: Optional[List[BaseMessage]]  # 对话历史，维持多轮对话上下文
    retrieved_documents: Optional[List[Document]]      # 检索到的知识片段，作为生成的"弹药库"
    topic_relevance: Optional[str]                     # 话题相关性判定，"门卫"角色
    enhanced_query: Optional[str]                      # 改写后的搜索关键词，"引擎"角色
    should_generate: Optional[bool]                    # 是否应该生成响应的标志位
    optimization_attempts: Optional[int]               # 查询优化计数器，用于防止死循环
    current_query: Optional[HumanMessage]              # 当前用户查询，原始输入

技术要点解析：

• 类型安全：使用 TypedDict 严格限制数据流向，确保节点间传递的数据不会因为 key 值错误而崩溃。在 Python 的动态类型系统中，这种显式类型定义能够提前发现许多潜在的 bug。

• 防御式编程：optimization_attempts 字段记录了循环次数，这是生产环境中的必备设计。如果没有这个计数器，当检索结果始终不理想时，系统可能陷入无限循环，不断消耗 API 调用次数和用户等待时间。通过限制优化次数（通常设置为 2-3 次），系统能够在多次尝试失败后优雅地退出循环并给用户适当的反馈。

• 职责分离：每个状态字段都有明确的职责。"门卫"负责过滤无关话题，"引擎"负责优化查询，"弹药库"存储检索结果——这种设计让系统的数据流清晰可追溯。

2.5 工作流定义实现

工作流的构建采用声明式配置方式，通过 LangGraph 提供的 API 将各个处理节点串联成完整的流程。

"""工作流构建器 - LangGraph流程编排核心实现"""
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from backend.models.state import ConversationState
from backend.nodes.enhancer import enhance_user_query
from backend.nodes.validator import validate_topic_relevance
from backend.nodes.retriever import fetch_relevant_content
from backend.nodes.assessor import assess_document_relevance
from backend.nodes.generator import generate_contextual_response
from backend.nodes.optimizer import optimize_search_query

def build_workflow():
    """
    构建完整的RAG Agent工作流
    
    工作流构建遵循以下步骤：
    1. 创建状态图 - 定义状态的数据结构
    2. 添加处理节点 - 每个节点是一个独立的处理函数
    3. 定义边连接 - 确定节点之间的执行顺序
    4. 配置条件路由 - 根据中间结果动态决定下一步走向
    5. 设置入口和出口 - 明确工作流的起点和终点
    """
    # 1. 创建状态图，指定状态类型
    workflow = StateGraph(ConversationState)

    # 2. 添加处理节点，每个节点负责特定的业务逻辑
    workflow.add_node("enhance_query", enhance_user_query)
    workflow.add_node("validate_topic", validate_topic_relevance)
    workflow.add_node("fetch_content", fetch_relevant_content)
    workflow.add_node("assess_relevance", assess_document_relevance)
    workflow.add_node("generate_response", generate_contextual_response)
    workflow.add_node("optimize_query", optimize_search_query)

    # 3. 定义边连接 - 顺序执行的基础流程
    workflow.add_edge("enhance_query", "validate_topic")

    # 4. 条件路由 - 话题验证后的分支决策
    workflow.add_conditional_edges(
        "validate_topic",
        route_by_topic,  # 路由函数，根据状态决定下一步
        {
            "fetch_content": "fetch_content",       # 话题相关，进入文档检索
            "handle_off_topic": "handle_off_topic", # 话题无关，进入兜底处理
        }
    )

    # 5. 条件路由 - 相关性评估后的分支决策
    workflow.add_conditional_edges(
        "assess_relevance",
        route_by_document_quality,
        {
            "generate_response": "generate_response",  # 有相关文档，生成回答
            "optimize_query": "optimize_query",       # 无相关文档，优化查询后重试
            "handle_no_results": "handle_no_results", # 多次优化后仍无结果，处理失败情况
        }
    )

    # 6. 查询优化后重新检索 - 形成闭环
    workflow.add_edge("optimize_query", "fetch_content")

    # 7. 设置终止节点和入口
    workflow.add_edge("generate_response", END)
    workflow.set_entry_point("enhance_query")

    # 8. 添加内存检查点，实现对话历史的持久化
    memory = MemorySaver()
    compiled_workflow = workflow.compile(checkpointer=memory)

    return compiled_workflow

设计要点说明：

工作流的核心优势在于其条件路由能力。通过 add_conditional_edges 方法，系统能够根据中间状态动态决定下一步走向。当话题验证节点返回 IRRELEVANT 时，工作流直接跳转到兜底回答节点，完全绕过向量检索和生成阶段，这种设计极大地节省了计算资源。当相关性评估发现没有相关文档时，工作流会回到查询优化节点，尝试用不同的检索策略重新搜索，给系统提供了"自我修正"的机会。

三、核心模块深度解析

3.1 查询增强器：破解多轮对话的"谜语"

在多轮对话中，用户的问题往往是简短的、依赖上下文的。例如，用户可能先问"15kW永磁同步电机的参数"，然后接着问"怎么安装"。如果没有上下文，第二句的检索将无法定位到正确的产品文档。查询增强器的核心职责就是将这种上下文依赖的问题转化为独立完整的检索查询。

"""查询增强器 - 智能问题重写与上下文融合"""
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_ollama import ChatOllama

def enhance_user_query(state: ConversationState) -> ConversationState:
    """
    增强用户查询，将上下文相关的查询转化为自包含的优化查询。
    
    策略设计：
    1. 首个问题直接使用原始查询，避免不必要的改写引入噪声
    2. 有对话历史时，生成简洁的增强查询（不超过50字）
    3. 融入上下文关键信息，保持原始意图，使用专业术语
    
    这种设计平衡了"信息完整度"和"检索精度"的需求。
    """
    original_query = state["current_query"].content
    history = state.get("conversation_history", [])

    # 策略1：首次对话直接使用原始查询
    # 第一轮对话时，上下文信息有限，强行改写反而可能引入噪声
    if len(history) <= 1:
        state["enhanced_query"] = original_query
    else:
        # 策略2：多轮对话时进行查询增强
        # 构建增强提示词
        enhancement_prompt = SystemMessage(
            content="""你是工业设备售后客服的查询优化专家。
你的任务是将用户问题改写为独立、完整的搜索查询（不超过50字）。

改写规则：
1. 融入上下文中的关键信息（产品型号、故障现象等）
2. 保持原始问题的核心意图
3. 使用工业领域的专业术语
4. 输出必须简洁，直接是优化后的查询，不要任何解释"""
        )

        # 提取最近的历史消息用于上下文理解
        context_messages = [enhancement_prompt]
        # 将最近3轮对话历史纳入上下文
        for msg in history[-3:]:
            context_messages.append(msg)
        # 添加当前问题
        context_messages.append(HumanMessage(content=f"请将这个问题改写为独立查询：{original_query}"))

        # 调用 LLM 生成增强查询
        llm = ChatOllama(
            model="qwen3:8B", 
            temperature=0.1,  # 低温度确保输出一致性
            reasoning=False   # 关闭思考过程，直接输出结果
        )
        response = llm.invoke(context_messages)
        state["enhanced_query"] = response.content.strip()

    return state

深度技术分析：

• 这个实现体现了几个重要的设计考量。首轮豁免机制是第一个关键点——如果用户刚开始对话就进行查询增强，往往会因为缺乏上下文而导致改写效果不佳，甚至可能偏离用户原始意图。因此，直接使用原始查询是更稳妥的选择。

• 字数限制策略是第二个关键点。增强后的查询被限制在 50 字以内，这是基于向量检索原理的考量。Embedding 模型（如本项目使用的 bge-m3）在处理极长文本时，核心语义会被稀释到大量的token中，导致向量相似度计算的"信噪比"下降。将查询控制在合理长度内，能够让检索结果更加聚焦和准确。

• Few-shot Prompting 技巧是第三个关键点。通过在系统提示词中明确列出改写规则，系统能够生成更加规范和高质量的输出。这种提示词工程技巧在工业场景中尤为重要，因为用户的原始问题可能表述模糊，而系统生成的检索查询必须足够精确。

用户原始输入	对话上下文	增强后的查询
“怎么安装？”	之前在讨论"15kW永磁同步电机"	“15kW永磁同步电机安装调试步骤说明”
“报错E-004是什么？”	之前在聊"某型号变频器"	“某型号变频器E-004故障代码含义及处理方案”
“那个故障怎么处理？”	之前在讨论"电机过热问题"	“电机过热的故障排查流程和处理方法”

3.2 话题分类器：构建系统的"护城河"

工业客服系统不应该也不需要回答所有问题。无效问题（如闲聊、竞争对手产品对比、个人咨询等）如果进入知识库检索流程，不仅浪费计算资源，还可能返回误导性的结果。话题分类器的作用就是作为"门卫"，在检索流程之前就对问题进行预过滤。

"""话题验证器 - 智能领域分类与无关问题过滤"""
import re

def validate_topic_relevance(state: ConversationState) -> ConversationState:
    """
    验证用户查询是否属于电机售后知识领域。
    
    这是一个典型的 Guardrail（护城河）设计。
    通过 LLM 的 Zero-shot 分类能力，快速判断问题是否在服务范围内。
    
    分类策略：
    - RELEVANT：进入完整 RAG 流程
    - IRRELEVANT：直接使用兜底回答，绕过向量库检索
    """
    enhanced_query = state.get("enhanced_query", "")

    # 构建分类提示词，明确列出相关和无关话题的边界
    classification_prompt = SystemMessage(
        content="""你是工业设备售后客服系统的话题分类专家。

相关话题（RELEVANT）包括：
- 电机、变频器、伺服驱动器等工业设备的产品规格和参数
- 电机安装、调试、维护、保养方法
- 电机无法启动、过热、振动、异响等故障排查
- 变频器故障代码解读（如 E-001, E-004, F-023 等）
- 轴承更换、皮带调整等配件更换和维修指南
- 产品保修政策、售后服务流程、技术支持联系方式

无关话题（IRRELEVANT）包括：
- 与工业设备无关的一般性问题（如天气、交通、新闻等）
- 其他品牌产品的咨询和对比
- 个人问题或无关闲聊
- 涉及敏感政治或法律问题的咨询

请直接回答：RELEVANT 或 IRRELEVANT，不要有其他内容。"""
    )

    # 调用 LLM 获取分类结果
    llm = ChatOllama(
        model="qwen3:8B", 
        temperature=0,  # 温度设为0，确保分类结果稳定一致
        reasoning=False
    )
    raw_response = llm.invoke([classification_prompt, HumanMessage(content=enhanced_query)])
    
    # 使用正则解析响应，兼容不同的大写/小写格式
    match = re.search(r'\b(RELEVANT|IRRELEVANT)\b', raw_response.content, re.IGNORECASE)
    classification = match.group(1).upper() if match else "IRRELEVANT"

    state["topic_relevance"] = classification
    return state

技术要点分析：

• 明确的话题边界定义是这个分类器的核心。与其给出一个模糊的定义（如"与工业设备相关的问题"），不如直接列出相关话题和无关话题的具体例子。这种 Few-shot 风格的定义方式让模型更容易理解分类标准，也减少了边界情况的误判。

• Zero-shot 分类策略意味着模型不需要任何标注样本就能完成分类任务。通过在提示词中提供分类类别和判断标准，模型能够直接根据这些规则进行判断。这种方式在工业场景中非常实用，因为我们可以随时更新话题边界而无需重新训练模型。

• 温度参数设为0是一个重要的工程实践。话题分类的输出应该是确定性的，不应该因为随机性而忽左忽右。将 temperature 设为 0 可以确保相同的输入总是得到相同的分类结果，保证系统的稳定性和可预测性。

• 跳过检索的效率收益：当问题被判定为 IRRELEVANT 时，工作流直接跳转到兜底回答节点，完全不经过向量库检索和生成逻辑。这意味着系统节省了至少 2-3 次 LLM 调用和 1 次向量检索的开销，对于高频的无效问题场景，这个优化带来的收益非常可观。

3.3 向量检索器：语义搜索的核心实现

向量检索是 RAG 系统的基础组件。本项目使用 ChromaDB 存储知识库文档，通过 bge-m3 嵌入模型将文本转换为向量，支持高效的语义相似度检索。

"""知识库模块 - 向量数据库初始化与语义检索"""
from langchain_ollama import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document

def initialize_vector_store():
    """
    初始化向量数据库，加载预置的工业设备知识库文档。
    
    实现步骤：
    1. 初始化 bge-m3 嵌入模型
    2. 加载 10 个专业文档并转换为 Document 对象
    3. 创建 Chroma 向量库，设定持久化存储路径
    4. 配置检索器，返回 top-K 相关文档
    """
    # 1. 初始化嵌入模型
    embeddings = OllamaEmbeddings(
        model="bge-m3:latest",
        base_url="http://localhost:11434",
    )

    # 2. 创建知识库文档（10个专业文档）
    documents = create_motor_knowledge_documents()

    # 3. 创建向量存储
    vector_store = Chroma.from_documents(
        documents=documents,
        embedding=embeddings,
        persist_directory="./backend/knowledge/chroma_db",
    )

    # 4. 创建检索器，设定返回K=3篇文档
    # search_kwargs 控制检索行为：k=3 表示返回相似度最高的 3 篇文档
    retriever = vector_store.as_retriever(search_kwargs={"k": 3})

    return retriever

def fetch_relevant_content(state: ConversationState) -> ConversationState:
    """
    从向量数据库中检索与增强查询相关的文档。
    
    检索策略说明：
    - 使用增强后的查询（包含上下文信息）进行检索
    - 检索结果直接存入状态，供后续评估节点使用
    """
    retriever = get_retriever()  # 获取全局检索器实例
    
    # 执行语义检索
    retrieved_docs = retriever.invoke(state["enhanced_query"])
    
    # 更新状态
    state["retrieved_documents"] = retrieved_docs

    return state

知识库文档结构：

本项目模拟构建了 10 个专业文档覆盖工业设备售后的主要知识领域。

文档名称	类别	内容概要
product_specs.pdf	产品规格	电机功率、电压、极数、转速等参数
installation_guide.pdf	安装调试	安装要求、调试步骤、接线规范
troubleshooting.pdf	故障排查	无法启动故障分析与解决流程
overheat_analysis.pdf	故障排查	电机过热原因识别与处理方法
vibration_analysis.pdf	故障排查	振动异常诊断与平衡调整
maintenance.pdf	维护保养	日常维护规范、润滑周期检查
vfd_codes.pdf	故障代码	变频器故障代码解读与对应措施
bearing_replacement.pdf	配件更换	轴承更换指南、工具与步骤
warranty_policy.pdf	保修政策	保修条款、期限、流程说明
service_process.pdf	售后服务	服务流程、联系方式、技术支持

检索策略优化：

选择 k=3 返回 3 篇文档是一个平衡后的决定。如果返回太少（如 1 篇），可能遗漏关键信息；如果返回太多（如 10 篇），会稀释 LLM 的注意力，增加上下文长度和推理成本。3 篇文档通常包含了回答用户问题所需的主要信息，同时不会让 LLM 处理过长的上下文。

3.4 相关性评估器：对抗"知识幻觉"的关键防线

这是本方案中最具工业工程价值的部分。向量检索是基于数学相似度的计算，它可能会找回一些"看起来语义接近但实际毫不相关"的文档。例如，检索"电机发热"可能返回一篇讨论"夏天机房温度管理"的文章，因为两者都涉及"热"这个概念。相关性评估器的作用就是作为"质检员"，用 LLM 判断检索结果是否真正包含回答问题所需的信息。

"""相关性评估器 - 文档质量控制与幻觉防范"""
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_ollama import ChatOllama
import re

def assess_document_relevance(state: ConversationState) -> ConversationState:
    """
    评估每个检索到的文档是否与用户问题真正相关。
    
    这是 Corrective RAG (CRAG) 架构的核心实现。
    
    评估流程：
    1. 遍历检索返回的每篇文档
    2. 让 LLM 判断文档是否包含回答问题所需的信息
    3. 只保留被判定为 RELEVANT 的文档
    4. 如果没有相关文档，触发查询优化循环
    """
    enhanced_query = state.get("enhanced_query", "")
    llm = ChatOllama(
        model="qwen3:8B", 
        temperature=0,  # 温度0确保评估结果一致
        reasoning=False
    )

    # 评估提示词
    assessment_prompt = SystemMessage(
        content="""你是文档相关性评估专家。
你的任务是判断文档内容是否包含回答用户问题所需的关键信息。

评估标准：
- RELEVANT：文档包含用户问题的答案或相关信息
- IRRELEVANT：文档与用户问题无关，或只包含极少量的边缘信息

请直接回答：RELEVANT 或 IRRELEVANT，不要有其他内容。"""
    )

    relevant_documents = []

    # 逐篇评估文档
    for doc in state["retrieved_documents"]:
        user_question = HumanMessage(
            content=f"用户问题：{enhanced_query}\n\n待评估文档内容：\n{doc.page_content}"
        )

        response = llm.invoke([assessment_prompt, user_question])
        
        # 解析评估结果
        match = re.search(r'\b(RELEVANT|IRRELEVANT)\b', response.content, re.IGNORECASE)
        if match and match.group(1).upper() == "RELEVANT":
            relevant_documents.append(doc)

    # 更新状态
    state["retrieved_documents"] = relevant_documents
    state["should_generate"] = len(relevant_documents) > 0

    return state

为什么需要二次评估：

向量检索的原理是计算查询和文档的向量余弦相似度，这种数学方法有其固有的局限性。相似度只能代表"语义接近"，不能保证"逻辑正确"。一篇讨论"电机发热"的文档和一篇讨论"机房温度过高"的文档在向量空间中可能非常接近，但对用户问题的价值却天差地别。

通过 LLM 进行二次评估，相当于请了一位"领域专家"来审核检索结果。这位专家能够理解文档的真正含义，判断它是否真正有助于回答用户的问题。这种人机协作的方式极大地提高了系统的可靠性。

触发自我修正机制：

当所有文档都被评估为 IRRELEVANT 时，系统不会尝试生成回答（因为那必将导致幻觉），而是自动进入 optimize_query 节点。这意味着系统具备了"知道自己不知道"的能力——当没有足够的信息时，它选择承认不足并尝试更好的检索策略，而不是给出一个不可靠的答案。

3.5 响应生成器：专业与可读性的平衡

当所有前置检查通过后，响应生成器负责综合对话历史、检索到的相关文档和当前问题，生成最终的回答。这个阶段需要平衡专业性和可读性——回答既要准确反映技术内容，又要让用户容易理解。

"""响应生成器 - 上下文感知与专业回答"""
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate

def generate_contextual_response(state: ConversationState) -> ConversationState:
    """
    基于检索到的相关文档和对话历史生成最终响应。
    
    生成策略：
    1. 构建包含对话历史和文档上下文的完整提示词
    2. 要求模型使用 Markdown 格式输出，便于阅读
    3. 在回答末尾标注信息来源
    4. 将生成的响应追加到对话历史
    """
    conversation_context = state["conversation_history"]
    relevant_docs = state["retrieved_documents"]
    enhanced_question = state["enhanced_query"]

    # 1. 构建文档上下文
    doc_context = ""
    for i, doc in enumerate(relevant_docs):
        source = doc.metadata.get("source", "未知来源")
        category = doc.metadata.get("category", "未分类")
        doc_context += f"\n【参考资料 {i+1}】\n"
        doc_context += f"来源：{source}（{category}）\n"
        doc_context += f"内容：{doc.page_content}\n"

    # 2. 构建对话历史文本
    history_text = ""
    for msg in conversation_context:
        role = "用户" if isinstance(msg, HumanMessage) else "客服"
        history_text += f"{role}：{msg.content}\n"

    # 3. 构建响应提示模板
    response_template = """你是专业的工业设备售后客服工程师。

你的任务是根据知识库信息，用专业的 Markdown 格式回答用户的问题。

格式要求：
1. 使用 Markdown 格式组织内容，层次分明
2. 一级标题用 "## " 开头，二级标题用 "### " 开头
3. 步骤性的内容使用编号列表（1. 2. 3.）
4. 重点内容用 "**...**" 加粗强调
5. 专业术语或代码用 "`...`" 标注
6. 在回答末尾用 "> 来源：xxx" 标注信息来源

请根据以下信息回答用户的问题：

【对话历史】
{conversation_history}

【相关知识库内容】
{document_context}

【当前问题】
{current_question}

请给出专业、准确、结构清晰的回答："""

    # 4. 创建提示模板并填充数据
    prompt = ChatPromptTemplate.from_template(response_template)
    messages = prompt.format_messages(
        conversation_history=history_text,
        document_context=doc_context,
        current_question=enhanced_question
    )

    # 5. 调用 LLM 生成响应
    llm = ChatOllama(
        model="qwen3:8B",
        temperature=0.3,  # 适度温度，允许一定的表达灵活性
        reasoning=False
    )
    response = llm.invoke(messages)

    # 6. 处理并存储响应
    generated_response = response.content.strip()

    # 将 AI 响应追加到对话历史
    state["conversation_history"].append(AIMessage(content=generated_response))

    return state

生成策略深度解析：

• Markdown 格式要求是工业场景的特殊需求。在故障排查场景中，步骤必须清晰编号；在参数说明场景中，需要使用加粗和代码格式来突出关键信息。强制要求模型使用 Markdown 输出，确保了回答的可读性和专业感。

• 来源标注机制是保证回答可追溯性的关键。每条建议都标注了来自哪篇知识库文档，用户如果需要更详细的信息可以直接查阅原文。这种透明性在工业场景中尤为重要，因为维修人员需要确认信息的权威性后才能执行操作。

• 温度参数 0.3 的选择经过了权衡。温度 0 会让输出过于机械和重复，温度太高又可能产生不准确的表述。0.3 是一个"偏保守但不失灵活"的设置，能够在保证准确性的同时提供自然流畅的表达。

四、流式输出与用户体验优化

4.1 Server-Sent Events 协议选择

在智能客服场景中，用户期望能够实时看到回答的生成过程，而不是等待完整答案一次性呈现。流式输出通过边生成边展示的方式，显著提升了用户体验。本项目选择 Server-Sent Events（SSE） 协议来实现这一功能。

"""API路由 - 流式对话接口实现"""
from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
import json

router = APIRouter(prefix="/api", tags=["chat"])

async def generate_streaming_response(conversation_history, doc_context, current_question):
    """
    生成流式响应的事件生成器
    
    SSE 协议格式说明：
    - 每个数据块以 "data: " 开头
    - 以双换行符 "\n\n" 结尾
    - 结束时发送 "data: [DONE]\n\n"
    """
    # 构建提示词
    response_template = """基于以下信息回答用户问题：
    
    对话历史：{conversation_history}
    相关知识：{document_context}
    当前问题：{current_question}
    
    请用Markdown格式回答："""
    
    prompt = ChatPromptTemplate.from_template(response_template)
    messages = prompt.format_messages(
        conversation_history=conversation_history,
        document_context=doc_context,
        current_question=current_question
    )

    # 初始化 LLM
    llm = ChatOllama(
        model="qwen3:8B",
        temperature=0.3,
        reasoning=False,
    )

    # 使用 astream 方法流式输出
    async for chunk in llm.astream(messages):
        if chunk.content:
            # 构造 SSE 格式数据
            yield f"data: {json.dumps({'content': chunk.content})}\n\n"

    # 发送结束标记
    yield f"data: {json.dumps({'content': '[DONE]'})}\n\n"

@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    """
    流式对话接口
    
    处理流程：
    1. 接收用户问题和会话ID
    2. 执行 LangGraph 工作流获取相关文档
    3. 构建文档上下文
    4. 返回 SSE 流式响应
    """
    # 1. 构建工作流配置（包含会话ID用于状态持久化）
    config = {"configurable": {"thread_id": request.thread_id}}
    
    # 2. 执行工作流获取相关文档
    input_state = {"current_query": HumanMessage(content=request.message)}
    result = workflow.invoke(input_state, config=config)

    # 3. 构建文档上下文
    doc_context = ""
    for doc in result["retrieved_documents"]:
        doc_context += f"来源：{doc.metadata.get('source')}\n"
        doc_context += f"内容：{doc.page_content}\n"

    # 4. 获取对话历史文本
    history_text = "\n".join([msg.content for msg in result["conversation_history"]])

    # 5. 返回流式响应
    return StreamingResponse(
        generate_streaming_response(history_text, doc_context, request.message),
        media_type="text/event-stream",
    )

为什么选择 SSE 而非 WebSocket：

在客服回复场景中，数据流是典型的"服务器单向推送给客户端"模式——用户发送问题后，只需要接收服务器的响应，不需要向服务器发送数据。SSE 相比 WebSocket 有以下优势：

• 轻量级：SSE 基于 HTTP 协议，不需要建立 WebSocket 连接那样复杂的握手过程。
• 自动重连：SSE 内置了浏览器原生的重连机制，在网络不稳定的车间环境下表现更稳健。如果网络中断，浏览器会自动尝试重新连接。
• 实现简单：后端只需要不断 yield 数据块，不需要维护双向通信的状态。

4.2 前端流式解析与 Markdown 容错

前端需要解析 SSE 流并实时渲染内容。由于内容是流式传输的，Markdown 标签可能不完整（例如代码块只传了一半），这会导致渲染错误。前端必须实现容错处理。

// 前端流式解析与 Markdown 渲染

interface ChatMessage {
    id: string;
    role: 'user' | 'assistant';
    content: string;
}

class StreamingParser {
    /**
     * 处理 SSE 流式数据
     * 
     * 核心挑战：LLM 输出的内容可能是不完整的 Markdown
     * 例如：代码块可能只收到 ```python 而没有闭合
     */
    async parseStream(
        response: Response,
        onChunk: (content: string) => void
    ): Promise<string> {
        const reader = response.body?.getReader();
        const decoder = new TextDecoder();
        let accumulatedContent = '';
        let leftover = '';  // 处理跨块数据

        if (!reader) throw new Error('No response body');

        while (true) {
            const { done, value } = await reader.read();
            if (done) break;

            // 解码当前数据块
            const chunk = decoder.decode(value, { stream: true });
            
            // 处理跨块数据：可能一行数据被分到两个 chunk
            const lines = (leftover + chunk).split('\n');
            leftover = lines.pop() || '';  // 最后一行可能不完整

            for (const line of lines) {
                if (line.startsWith('data: ')) {
                    const data = line.slice(6).trim();
                    
                    if (data === '[DONE]') {
                        // 流结束
                        return accumulatedContent;
                    }

                    try {
                        const parsed = JSON.parse(data);
                        if (parsed.content) {
                            // 应用 Markdown 容错处理
                            const processed = this.processMarkdown(parsed.content);
                            accumulatedContent += processed;
                            onChunk(processed);
                        }
                    } catch (e) {
                        // 非 JSON 数据，直接使用
                        accumulatedContent += line;
                        onChunk(line);
                    }
                }
            }
        }

        return accumulatedContent;
    }

    /**
     * Markdown 预处理 - 修复流式输出中的格式问题
     * 
     * 这些问题在流式传输中非常常见：
     * 1. 代码块不配对（只有开头没有结尾）
     * 2. 标题没有空格（#标题 应该是 # 标题）
     * 3. 列表项缺少换行
     */
    processMarkdown(content: string): string {
        let processed = content;

        // 1. 补全未闭合的代码块
        const codeBlockCount = (processed.match(/```/g) || []).length;
        if (codeBlockCount % 2 !== 0) {
            processed += '\n\n```\n';
        }

        // 2. 修正标题：确保 # 后面有空格
        // "#电机故障" -> "# 电机故障"
        processed = processed.replace(/([^\n])(#{1,6}\s*)/g, '$1\n$2');

        // 3. 修正无序列表：句号后需要换行
        processed = processed.replace(/([。！？：；])([-*]\s+)/g, '$1\n$2');

        // 4. 处理 Markdown 语法元素
        // 确保粗体语法完整
        processed = processed.replace(/\*\*([^*]*)(\*|$)/g, (match, content, end) => {
            return end === '*' ? match : `**${content}**`;
        });

        return processed;
    }
}

容错策略深度解析：

跨块数据处理是流式解析的第一个挑战。由于网络传输的特性，一个逻辑行可能被分割到两个不同的数据块中。代码中使用 leftover 变量来暂存不完整的数据，等待下一个数据块的到来后再合并处理。

代码块补全是第二个关键问题。在 Markdown 中，代码块使用 标记开头和结尾。如果 LLM 正在生成一个代码示例，只传了一半就因网络或生成完成而中断，渲染器会收到一个不完整的代码块标记。代码通过检查 出现次数，如果是奇数则自动补全结尾。

标题空格修复解决了 LLM 可能输出的不规范 Markdown 格式。正确的 Markdown 标题应该是 “# 标题”（# 后有空格），但 LLM 有时可能输出 “#标题”。这种格式在部分渲染器中无法正确识别，需要前端自动修复。

列表换行处理确保了工业标准中的列表项能够正确换行显示。中文句号后直接跟列表项符号（如 "- "）会导致显示在同一行，影响可读性。代码自动在句末标点后添加换行。

五、前端界面实现

5.1 组件架构设计

前端采用 React + TypeScript 构建，组件设计遵循单一职责原则，每个组件只负责一个特定的功能。

5.2 Markdown 渲染与样式设计

工业场景需要结构化的回答界面，Markdown 渲染器和样式设计直接影响用户体验。

// Markdown 渲染器配置
import MarkdownIt from 'markdown-it';
import hljs from 'highlight.js';

// 初始化 markdown-it
const md = new MarkdownIt({
    html: false,              // 禁用 HTML 解析，防止 XSS
    linkify: true,            // 自动识别链接
    typographer: true,        // 智能标点转换
    breaks: true,             // 单换行转 <br>
    highlight: function(str, lang) {
        // 代码高亮
        if (lang && hljs.getLanguage(lang)) {
            try {
                return `<pre class="hljs"><code>${
                    hljs.highlight(str, { language: lang, ignoreIllegals: true }).value
                }</code></pre>`;
            } catch (__) {}
        }
        return `<pre class="hljs"><code>${md.utils.escapeHtml(str)}</code></pre>`;
    }
});

// 配置代码高亮主题
import 'highlight.js/styles/github-dark.css';

/* 浅色主题样式定义 */
:root {
    /* 基础颜色变量 */
    --bg-primary: #f8fafc;        /* 淡灰白背景 */
    --bg-secondary: #ffffff;      /* 白色卡片背景 */
    --bg-tertiary: #f1f5f9;       /* 次级背景 */
    
    /* 品牌色 - 天蓝色系 */
    --accent-primary: #0ea5e9;    /* 主色调 */
    --accent-secondary: #38bdf8;  /* 浅蓝色 */
    --accent-gradient: linear-gradient(135deg, #0ea5e9 0%, #38bdf8 100%);
    
    /* 文字颜色 */
    --text-primary: #1e293b;      /* 主要文字 */
    --text-secondary: #64748b;    /* 次要文字 */
    --text-inverse: #ffffff;      /* 反色文字 */
    
    /* 消息气泡 */
    --bubble-user: var(--accent-gradient);
    --bubble-assistant: var(--bg-secondary);
    
    /* 边框和阴影 */
    --border-light: #e2e8f0;
    --shadow-sm: 0 1px 2px rgba(0, 0, 0, 0.04);
    --shadow-md: 0 4px 6px -1px rgba(0, 0, 0, 0.05);
    
    /* 圆角 */
    --radius-sm: 8px;
    --radius-md: 12px;
    --radius-lg: 16px;
    --radius-xl: 24px;
}

/* 消息气泡样式 */
.message-bubble {
    border-radius: var(--radius-lg);
    padding: 12px 16px;
    box-shadow: var(--shadow-sm);
    transition: box-shadow 0.2s ease;
    line-height: 1.6;
    max-width: 85%;
}

.message-bubble:hover {
    box-shadow: var(--shadow-md);
}

/* 用户消息样式 */
.message-wrapper.user .message-bubble {
    background: var(--bubble-user);
    color: var(--text-inverse);
    border-bottom-right-radius: 4px;
}

/* 助手消息样式 */
.message-wrapper.assistant .message-bubble {
    background: var(--bubble-assistant);
    color: var(--text-primary);
    border-bottom-left-radius: 4px;
    border: 1px solid var(--border-light);
}

/* Markdown 内容样式 */
.markdown-content h1,
.markdown-content h2,
.markdown-content h3 {
    color: var(--text-primary);
    margin-top: 1.5em;
    margin-bottom: 0.5em;
}

.markdown-content code {
    background: var(--bg-tertiary);
    padding: 2px 6px;
    border-radius: 4px;
    font-family: 'Fira Code', monospace;
    font-size: 0.9em;
}

.markdown-content pre {
    background: #1e293b;
    border-radius: var(--radius-md);
    padding: 16px;
    overflow-x: auto;
    margin: 1em 0;
}

.markdown-content pre code {
    background: transparent;
    color: #e2e8f0;
    padding: 0;
}

/* 来源标注样式 */
.source-citation {
    margin-top: 12px;
    padding-top: 8px;
    border-top: 1px solid var(--border-light);
    font-size: 0.85em;
    color: var(--text-secondary);
}

六、技术实践总结与工程建议

6.1 核心设计决策回顾

在项目开发过程中，我们面临众多技术选型的决策。以下是主要决策点及其背后的考量：

决策点	最终选择	选择理由
工作流编排	LangGraph	支持条件路由、循环、状态管理等复杂控制流
LLM	qwen3:8B（Ollama）	本地部署保证数据安全，中文理解和专业术语能力强
Embedding	bge-m3	中文效果好，与 Ollama 生态兼容良好
向量数据库	ChromaDB	轻量级、易部署、支持持久化
流式协议	SSE	简单可靠、浏览器原生支持、内置自动重连
前端渲染	markdown-it	轻量、配置灵活、插件生态丰富
状态持久化	MemorySaver	内存存储适合单服务器部署，响应快速

6.2 关键优化策略

查询增强策略是系统智能化的核心体现。首次对话不进行增强，避免引入不必要的改写噪声；有历史对话时，将查询限制在 50 字以内，平衡信息完整度和检索精度。这种策略让系统既能处理简单的单轮问答，又能理解复杂的多轮对话。

话题预过滤机制显著提升了系统效率。通过明确的边界定义，系统能够快速识别并处理无关问题，将宝贵的计算资源集中在真正需要专业知识的问题上。这种设计体现了"有所不为"的工程哲学。

质量控制闭环是系统可靠性的保障。检索后的二次评估确保只有高质量文档进入生成阶段；无相关文档时触发查询优化循环，给系统"自我修正"的机会。这种设计让系统具备了"知道自己不知道"的能力。

流式输出优化平衡了用户体验和系统性能。工作流执行阶段采用同步方式确保数据完整性，响应生成阶段采用流式方式提供即时反馈。SSE 协议的选择兼顾了实现简单性和网络稳定性。

6.3 未来演进方向

本项目的架构具有良好的可扩展性，以下是几个值得探索的演进方向。

• 知识库动态更新是提升系统长期价值的关键。当前知识库是静态的 10 个文档，未来可以支持动态上传和更新文档，让系统能够及时纳入新的产品信息和故障案例。这需要实现文档解析、向量索引更新和质量审核等配套功能。

• **工具调用集成（Tool Use）**将大幅扩展系统的能力边界。下一步可以集成工业 IoT 平台的 API，让 Agent 能够查询设备的实时运行状态。当用户询问"3号电机的运行状态"时，Agent 可以调用工具获取实时电压、频率、温度等数据，给出更精准的诊断建议。

• 监控与可观测性是生产环境必备的能力。建议添加检索质量监控（哪些查询没有找到相关文档）、响应延迟监控（各阶段耗时分布）、用户满意度反馈等指标，为持续优化提供数据支撑。

• A/B 测试框架支持对比不同提示词和模型的效果。通过对不同版本的系统进行对照实验，可以科学地评估优化措施的效果，避免盲目调整参数。

结语

通过 LangGraph，我们将一个简单的 RAG 应用升级为了具有判断力和反思能力的 Agent。这套系统具备三大核心竞争力：可靠性——通过循环验证和自我修正，它比简单的 GPT 回答更严谨；经济性——本地部署的模型在保证效果的同时，无需支付昂贵的 API 费用，且支持离线运行；可维护性——由于每个环节都是独立的 Python 函数，可以轻松针对"检索不准"或"分类不清"进行单点优化。

在工业4.0和智能制造的大背景下，智能客服只是 AI 在工业领域应用的冰山一角。随着大模型技术的持续进步和行业理解的不断深化，AI 将在设备诊断、预测性维护、工艺优化等更多场景发挥价值，为工业数字化转型注入新的动力。

项目源代码

完整的项目代码和更详细的实现，请访问我的知识星球（ https://t.zsxq.com/CCi0k ），获取完整系统项目源代码。