在之前的教程中，我們主要探討了個人級 RAG（檢索增強生成）應用的實現方式和優化技巧。

但在企業級應用中，知識管理和智能檢索的需求更為複雜，涉及多個部門，各自具備獨立的業務領域、數據存儲方式和算法需求。因此，高效管理和檢索知識，確保不同部門靈活訪問知識庫，同時滿足數據隔離、安全性和共享機制，是企業級知識管理系統的核心挑戰。

本章將介紹如何使用 LazyLLM 快速構建面向企業級的數據庫管理和檢索召回服務，滿足上述複雜需求。

一、企業級知識庫的多樣性需求

在企業實際應用中，知識庫不再是簡單的信息堆疊，而需要面對來自權限、共享方式與安全保障等多個維度的複雜訴求。下面我們從典型場景出發，系統闡述這些多樣性需求及其應對策略。

場景一：按部門隔離的知識訪問（權限管理多樣性）

客户背景：大型製造企業

企業下設多個職能部門（如研發、採購、銷售），各自負責不同領域的信息收集與管理。文檔內容涵蓋供應鏈合作、成本核算、產品規劃等，信息敏感度高，內部訪問需嚴格隔離。

• 部門專屬知識庫：各業務單元可自主維護產品文檔、市場分析、財務報告等專業內容，系統自動隔離非授權訪問。
• 智能標籤體系：支持"研發-技術白皮書"、"市場-競品分析"等專業標籤體系，實現精準檢索與受控共享。
• 管理駕駛艙：高管層可通過"戰略視圖"標籤獲取跨部門知識摘要，確保決策支持的同時維護數據安全。

場景二：多種共享方式的協同需求（共享方式多樣性）

客户背景：諮詢服務公司

公司經常與不同客户進行聯合項目，涉及文檔共享、階段報告、項目材料和算法資源等。客户使用的工具和偏好多樣，需要靈活的共享機制以滿足業務合作與文檔保密的平衡。

• 算法資源共享：支持llm模型、embedding模型、檢索算法等核心組件在企業內部分享複用。
• 差異化調用的知識複用：一個項目知識庫可同時服務內部顧問、客户技術團隊及第三方分析機構，能夠配置不同的召回規則實現召回解耦，滿足多角色精準訪問。

場景三：多重安全策略保障內容安全（安全保障多樣性）

客户背景：金融科技公司

知識庫包含大量敏感內容，如用户金融行為分析、監管合規方案、審計材料等，對信息安全的要求極高。

安全需求：

• 敏感詞智能過濾：內置多級敏感詞識別策略，結合上下文進行動態判斷，在問答與檢索過程中自動提示、替換或阻斷輸出，防止企業內部黑名單、客户機密、涉密術語等信息泄露。
• 全鏈路知識加密：知識文檔在上傳、解析、入庫、傳輸及生成階段均可啓用對稱或非對稱加密機制，確保知識在整個生命週期中不被竊取或篡改。
• 私有化部署方案：平台可在企業內網私有服務器或專屬雲環境中完成全棧部署，包括知識庫、向量引擎、檢索模塊與模型推理服務，確保知識數據不經公網傳輸。系統可無縫集成企業認證、權限與日誌體系，形成閉環安全防護結構。

企業級應用場景對知識庫提出了更多維度的要求，LazyLLM 針對這些需求，在權限管理、共享模式以及安全保障三方面提供瞭解決方案。

二、權限多樣性以及解決方案

（一）權限隔離：支持多部門獨立知識運營

在大型企業中，各部門通常擁有獨立的文檔體系，這些文檔可能包含敏感的業務信息、內部操作手冊或關鍵流程文件。為了確保信息安全與使用合規，企業對文檔的管理能力和隔離機制提出了更高要求。常見的管理難題包括：

• 如何支持知識庫的高頻更新與維護？
• 如果同一篇文檔被多個部門使用，需分別入庫多次？導致數據冗餘和管理困難？

為此，lazyllm內置靈活的文檔管理服務，提供了一套完整的文檔增刪改查功能，用户可以方便地添加新文檔、修改已有內容、刪除過期文檔，並在需要時進行檢索，確保知識庫始終保持最新狀態。不同的知識庫相互隔離，可幫助企業按部門、崗位、項目等維度靈活劃分知識庫訪問邊界，實現“誰能看、看什麼、看多少”的可控策略。

例如，在同一知識庫存儲內，支持利用文檔管理組功能進行分組管理，同一文檔只需解析一次。

1. 文檔管理服務

文檔管理服務的啓用非常簡單，只需要在創建 document 對象時，將 manager 參數設為 ui ，即可開啓文檔管理功能。例如：

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/main/rag/codes/chapter17/document_manager.py）

from lazyllm.tools import Document
import time

path = "path/to/docs"
docs = Document(path, manager='ui')
# 註冊分組
Document(path, name='法務文檔管理組', manager=docs.manager)
Document(path, name='產品文檔管理組', manager=docs.manager)
# 啓動服務
docs.start()
time.sleep(3600)

啓動後頁面如下：

文檔管理服務開啓後，可在 Web 頁面中便捷地查看不同分組內的文檔，並支持對文檔進行快速增刪操作。

2.文檔管理後端 API 服務

該Web服務內置了基於Gradio的默認前端界面。若企業需要定製更專業的前端界面，可通過manager=True參數僅啓動後端API服務，隨後基於接口自由開發個性化前端。

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/main/rag/codes/chapter17/document_manager.py）

from lazyllm.tools import Document
import time

path = "path/to/docs"
docs = Document(path, manager=True)
# 註冊分組
Document(path, name='法務文檔管理組', manager=docs.manager)
Document(path, name='產品文檔管理組', manager=docs.manager)
# 啓動服務
docs.start()
time.sleep(3600)

啓動後，Redoc 頁面如下，展示了可用的後端接口。

（二）權限多樣性：支持更細粒度的訪問控制

在企業實際運營中，權限需求遠不止“哪個部門訪問哪個文檔”這麼簡單。不同小組、崗位甚至項目成員，常常需要在共享部分知識的同時保護敏感內容不被誤讀或泄露。

為滿足這種複雜且多變的需求，LazyLLM 提供了基於標籤檢索的權限控制能力。企業可為文檔打上如部門、崗位、時間、文件類型等多個標籤，併為不同角色配置相應的訪問權限，實現精細化管理。例如，市場部門可能希望檢索與“產品推廣”相關的文檔，而研發部門可能更關注“技術規格”類的文檔。

除了基於標籤的篩選需求，用户還希望在檢索時能夠指定特定文檔集進行查詢，而不是搜索整個知識庫。例如，法務部門可能只希望檢索最近一年內的合同文件，而不是所有歷史合同。

因此，系統需要支持針對知識庫中的部分文檔集合進行精準查詢，提升搜索的精準度和效率。

企業需要通過標準化鑑權機制（如基於角色RBAC、基於屬性ABAC、基於策略PBAC）精細控制文檔訪問。

• 如何依據標準化鑑權機制組織內容和設置訪問權限，確保信息合規使用？
• 如何根據權限等級細化訪問控制，如同一部門內不同人員擁有不同等級的訪問權限？

LazyLLM解決方案 ——— 基於標籤（Tag）的權限控制機制：

• 每個文檔可在上傳時綁定預定義標籤（如部門、項目、安全等級）
• 檢索時支持基於標籤的過濾，僅返回符合條件的內容

🔍 示例：模擬基於角色的權限控制（RBAC）

目標：讓“法務一部”員工僅能檢索本部門的文檔

• 定義標籤字段：department
• 上傳文檔時指定：department = 法務一部
• 檢索時自動注入過濾條件：filter={"department": "法務一部"}

通過這一機制，實現了基於角色的隔離訪問，在保障數據安全的同時，也簡化了權限策略的實施。

在實際應用中，鑑權邏輯應由後端統一管理，算法側不直接處理鑑權。這樣可以確保權限控制的集中化和安全性，避免因算法側繞過權限而引發的安全漏洞。

1. 基於標籤的訪問控制

我們可以通過 元數據(metadata)管理 和 檢索過濾(filter) 來實現靈活的分類和查詢功能，僅需以下兩步：

第一步：Metadata 添加

使用元數據過濾需指定milvus數據庫，並且聲明要指定的字段，以部門department為例，示例如下：

CUSTOM_DOC_FIELDS = {"department": DocField(data_type=DataType.VARCHAR, max_size=65535, default_value=' ')}
milvus_store_conf = {
    'type': 'milvus',
    'kwargs': {
        'uri': os.path.join(db_path, "milvus.db"),
        'index_kwargs': [
            {
                'embed_key': 'bge_m3_dense',
                'index_type': 'IVF_FLAT',
                'metric_type': 'COSINE',
            },
            {
                'embed_key': 'bge_m3_sparse',
                'index_type': 'SPARSE_INVERTED_INDEX',
                'metric_type': 'IP',
            }
        ]

    },
}


law_knowledge_base = Document(
    data_path, 
    name='法務知識庫', 
    manager="ui", 
    doc_fields=CUSTOM_DOC_FIELDS,  # 指定要過濾的字段
    store_conf=milvus_store_conf,  # 開啓milvus數據庫
    embed=OnlineEmbeddingModule(source="glm", embed_model_name="embedding-2"))

在通過文檔管理服務上傳文件時，用户可為文件指定需要設定的元數據（metadata）分類信息。例如：

第二步：Metadata 查詢

在查詢時，用户可以通過 filter 機制指定需要過濾的分類信息。可以通過以下方式進行篩選，來僅檢索來自法務一部和法務二部的文檔。

retriever_support = Retriever(
    [law_knowledge_base, support_knowledge_base],
    group_name=...,
    similarity=...,
    topk=2
)

support_question = "客户關於合同投訴的處理方式"
support_res_nodes = retriever_support(
    support_question， 
    filters={'department':['法務一部']}   # 指定已定義的過濾條件
)

通過添加Metadata和定義filter機制，最終實現了對文檔的多權限管理。

以下示例同時展示了兩種檢索方式：

• 使用 filter（僅檢索“法務一部”文檔）
• 不使用 filter（檢索所有文檔）

該對比僅用於功能展示目的，以便理解系統的過濾機制。

在實際應用中，系統可實現強制綁定過濾條件，確保用户只能檢索其所屬部門的文檔，從而實現文檔隔離與權限控制的統一。

📌 進階：細化權限等級的權限控制

等級 1：普通員工，僅能查看基礎財務報表。

等級 2：主管，能查看部門預算和項目支出。

等級 3：經理及以上，能夠訪問財務決策和敏感報表。

1.註冊 權限等級- permission_level 字段：

CUSTOM_DOC_FIELDS = {"department": DocField(data_type=DataType.VARCHAR, max_size=32, default_value=''), 
                     "permission_level": DocField(data_type=DataType.INT32, default_value=1)}

2.上傳文檔同時標記權限等級（如permission_level = 1）

files = [('files', ('普通文檔.pdf', io.BytesIO(...)),
          ('files', ('敏感文檔.pdf', io.BytesIO(...))]
metadatas=[{"department": "法務一部", "permisssion_level": 1},
           {"department": "法務一部", "permisssion_level": 2}]))

3.檢索時指定權限等級

nodes = retriever(query, filters={'department': ['法務一部'], "permission_level": [1,2]} )

三、共享方式多樣性以及解決方案

除了權限控制，企業在知識共享方面也面臨多樣化需求：

一方面，不同團隊間常需共享算法資源以提升複用效率；另一方面，多個部門之間也存在知識庫交叉使用的需求，支持多對多的知識複用關係。

這些場景對靈活的共享機制提出了更高要求。

（一）共享靈活性：支持多源知識與算法自由適配

在企業中，多個部門可能共享相同的算法進行數據處理、推理和決策，但由於各自的業務領域不同，每個部門通常擁有獨立的知識庫，存儲各自領域的專屬信息。

因此，系統需要支持同一算法可作用於多個不同的知識庫，確保算法在不同部門的適用性。

比如：在一家金融公司中，風險控制部門和市場分析部門可能都使用相同的文本解析和嵌入算法來預處理數據，但風險控制部門的知識庫主要包含歷史交易和客户信用記錄，而市場分析部門的知識庫則包含市場動態和競爭對手情報。系統需要支持在不同的知識庫中複用相同的數據處理算法。

另一方面，也有部分企業場景中，不同的部門可能使用各自定製的算法進行數據分析和決策，但某些部門之間可能需要共享同一知識庫，以便在統一的信息源基礎上進行差異化計算。

系統需要支持不同算法可作用於同一知識庫，以滿足這種業務需求。

比如：在一家電商企業中，推薦系統部門可能使用基於協同過濾的嵌入算法對用户行為進行建模，而搜索優化部門可能使用基於詞向量的相似度排序算法來提升搜索結果的相關性。這兩個部門可能都基於同一個用户行為數據集，系統需要支持在同一知識庫中獨立運行不同的嵌入和排序算法，生成針對性優化的結果。

接下來，我們將介紹如何利用 lazyllm 實現 RAG 流程中算法模塊的共享與靈活配置。

LazyLLM支持靈活的算法插件機制，為構建可插拔的智能系統提供了基礎保障。我們可以預先定義一個 全局算法註冊器，將常用的解析器、嵌入模型和相似度計算方法統一管理，從而在構建知識庫和檢索器（Retriever）時，實現算法的靈活組合與複用。

注：以下代碼demo僅展示框架，實際算法需根據需求自行實現。

class AlgorithmRegistry:
    """全局算法註冊器，實現算法與知識庫的解耦複用"""

    # --------------------------
    # 文檔解析算法池
    # --------------------------
    DOC_PARSERS = {
        "magic_pdf": MagicPDFReader(),  # 帶OCR的高級PDF解析
        "basic_pdf": SimplePDFParser(),   # 輕量PDF解析
        "docx": OfficeParser(),           # 嚴格模式Word解析
        "html": BeautifulSoupParser()
    }

    # --------------------------
    # 節點解析算法池
    # --------------------------
    NODE_PARSERS = {
        "semantic_chunk": SemanticNodeSplitter(),  # 語義分塊
        "graph_based": KnowledgeGraphParser()  # 金融實體識別
    }

    # --------------------------
    # 嵌入模型池
    # --------------------------
    EMBEDDINGS = {
        "general": SentenceTransformer(),    # 通用語義
        "finance": FinBERTEmbedding(),# 金融領域專用
        "bio": BioClinicalBERT.from_pretrained() 
    }

    # --------------------------
    # 相似度計算策略
    # --------------------------
    @staticmethod
    @fc_register(name="euclidean")                            
    def euclidean_sim(query, nodes):
        pass

    @staticmethod
    @fc_register(name="ifdif")                                 
    def ifdif_sim(query, node):
        pass

前文代碼示例展示瞭如何構建一個“算法池”，企業可根據自身需求進行擴展和維護。完成註冊後，系統可根據具體業務場景，按需選擇合適的算法組件，實現“算法即服務”的設計理念。

以一個對準確性要求極高的金融風控場景為例：

• 嵌入模型 選擇了針對金融文本微調的 FinBERT，以獲得更精確的語義編碼；
• 文檔解析器 採用具備 OCR 能力的高精度 PDF 解析工具，保障合同等複雜文件的結構完整性；
• 節點解析採用語義切分策略，以保留關鍵信息的語義上下文，提升召回質量。

代碼示例如下：

# 金融風控（高精度導向）-- 處理複雜PDF合同，需識別法律實體與財務條款

law_kb = Document("path/to/kb", name='金融風控知識庫', embed=AlgorithmRegistry.EMBEDDINGS['finance'])   # 選擇金融領域BERT微調模型

law_kb.add_reader(AlgorithmRegistry.DOC_PARSERS['magic_pdf'])   # 選擇處理複雜PDF合同，需識別精細法律實體與財務條款

law_kb.create_node_group(name='semantic_nodes', transform=AlgorithmRegistry.NODE_PARSERS['semantic_chunk'])  # 選擇語義分塊算法

# 定義 retriever 並選擇 節點組 和 相似度計算方式
retriever = Retriever(
    group_name="semantic_nodes",   
    similarity="cosine",      
    topk=1                
)

同一套算法在多個知識庫中的應用場景已在前面權限的部分討論過。

接下來，我們實現在同一知識庫中，通過不同文檔分組實現算法多樣化的場景。

docs = Document(path, manager=True, embed=OnlineEmbeddingModule())
# 註冊分組
Document(path, name='法務文檔管理組', manager=docs.manager)
Document(path, name='產品文檔管理組', manager=docs.manager)
#  模擬文檔上傳
docs.start()
files = [('files', ('產品文檔.txt', io.BytesIO("這是關於產品的信息。該文檔由產品部編寫。\n來自產品文檔管理組".encode("utf-8")), 'text/plain'))]
files = [('files', ('法務文檔.txt', io.BytesIO("這是關於法律事務的説明。該文檔由法務部整理。\n來自法務文檔管理組".encode("utf-8")), 'text/plain'))]
…
# 為 產品文檔管理組 設置切分方式為按 段落 切分
doc1 = Document(path, name=‘產品文檔管理組', manager=docs.manager)
doc1.create_node_group(name="block", transform=lambda s: s.split("\n") if s else '')
retriever1 = Retriever([doc1], group_name="block", similarity="cosine", topk=3)

# 為 法務文檔管理組 設置切分方式為按 句子 切分
doc2 = Document(path, name=‘法務文檔管理組’, manager=docs.manager)
doc2.create_node_group(name=“line”, transform=lambda s: s.split(“。") if s else ‘’)
retriever2 = Retriever([doc2], group_name="line", similarity="cosine", topk=3)

（二）召回解耦：支持知識庫與召回服務靈活協同

為應對複雜的知識共享與複用需求，企業越來越需要靈活而高效的知識組織結構與管理能力：

1️⃣需要多對多的知識組織結構

• 企業往往希望通過一個統一的文檔管理服務，集中管理多個知識庫，既支持各業務部門對知識內容的獨立維護，又保障在需要時的受控共享。
• 同一知識庫還能被多個 RAG 召回系統調用，實現跨業務系統的知識複用，提升模型服務的覆蓋面與智能化能力。

2️⃣需要多業務場景的知識複用能力

• 面對客服、合規、風控、市場等多樣化業務需求，企業必須確保知識能夠高效複用，同時又能按場景獨立更新、靈活適配。

為滿足上述需求，LazyLLM不僅提供靈活的文檔管理模塊，還將文檔管理與 RAG 召回服務進行完全解耦，來滿足企業知識管理和召回需求的多樣性，這樣做的好處具體體現在：

• 多對多管理模式：一個文檔管理服務可以同時管理多個知識庫，支持不同業務部門的知識存儲需求。
• 多 RAG 適配：同一個知識庫可以適用於多個 RAG 召回服務，一個 RAG 召回服務可以從多個知識庫中檢索數據。

得益於這種解耦設計，確保了企業能夠在不同業務場景下，動態調整知識庫和 RAG 召回服務的綁定關係，滿足個性化的知識管理需求。

而具體實現起來，僅需以下兩步驟，可搭建多知識庫管理和召回流程。

1.初始化知識庫

初始化知識庫有兩種方式：

• 路徑定義方式：如果已有整理好的知識庫文檔，可直接通過指定文件路徑來定義知識庫。
• 動態管理方式：如果知識庫需要動態調整，可通過啓動知識庫服務後，進行上傳、刪除和管理操作。

説明：通過啓動服務來上傳文檔的方式僅能綁定一個路徑，若存在多個不同路徑的數據庫，建議使用定義路徑的方式；為不同知識庫靈活地註冊不同的算法，但需要保持node group的name一致，以便後續進行聯合召回。

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/7abc91dbb82a007a78731845dd8c360ac0cc1e75/rag/codes/chapter17/multi_retriever.py#L7）

from lazyllm.tools import Document
from lazyllm import OnlineEmbeddingModule
import time

# =============================
# 方法1.通過定義路徑的方式
# =============================

law_data_path = "path/to/docs/law"
product_data_path = "path/to/docs/product"
support_data_path = "path/to/docs/support"

law_knowledge_base = Document(law_data_path, name='法務知識庫'，embed=OnlineEmbeddingModule(source="glm", embed_model_name="embedding-2"))
product_knowledge_base = Document(product_data_path,name='產品知識庫'，embed=OnlineEmbeddingModule(source="glm", embed_model_name="embedding-2"))
support_knowledge_base = Document(support_data_path,name='客户服務知識庫'，embed=OnlineEmbeddingModule(source="glm", embed_model_name="embedding-2"))


# =============================
# 方法2.通過文檔上傳方式
# =============================

data_path = "path/to/docs"

law_knowledge_base = Document(data_path, name='法務知識庫', manager="ui"，embed=OnlineEmbeddingModule(source="glm", embed_model_name="embedding-2"))
# 通過法務知識庫的 manager 共享管理器
product_knowledge_base = Document(
    data_path,
    name='產品知識庫',
    manager=law_knowledge_base.manager,
)

law_knowledge_base.start()
# ... 服務啓動後手動增刪文件 ... #

2. 啓用 RAG 召回服務

定義好知識庫後，可根據需要為不同知識庫靈活配置召回服務。將文檔管理對象傳入定義的 Retriever 即可，Retriever 的具體使用方法詳見之前的教程 [ 第8講：不止是cosine！匹配策略決定你召回的質量 ]。

企業可通過業務需求配置數據處理算法註冊為node_group， 並在進行召回時使用。

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/7abc91dbb82a007a78731845dd8c360ac0cc1e75/rag/codes/chapter17/multi_retriever.py#L29）

from lazyllm import Retriever, SentenceSplitter

# 配置和定義數據處理算法, 可根據業務需要自定義
Document.create_node_group(name="sentences", transform=SentenceSplitter, chunk_size=1024, chunk_overlap=100)

# 組合法務 + 產品知識庫，處理與產品相關的法律問題
retriever_product = Retriever(
    [law_knowledge_base, product_knowledge_base],
    group_name="sentences",       # 分組名（根據業務需求選擇）
    similarity="cosine",       # 相似度參數（根據模型配置）
    topk=2                # 召回前2個最相關的結果
)

product_question = "A產品功能參數和產品合規性聲明"
product_res_nodes = retriever_product(product_question)

# 組合法務 + 客户知識庫，處理客户支持相關問題
retriever_support = Retriever(
    [law_knowledge_base, support_knowledge_base],
    group_name="sentences",
    similarity="cosine",
    topk=2
)

support_question = "客户投訴的處理方式以及會導致的法律問題"
support_res_nodes = retriever_support(support_question)
...

檢索結果示例如下，當檢索問題“A產品功能參數和產品合規性聲明”，可同時檢索到產品知識庫中的產品參數內容和法規知識庫中的相關條款：

print(f"query: {product_question }")
print("retrieve nodes:")
for node in product_res_nodes :
    print(node.text)
    print()
"""
query: A產品功能參數和產品合規性聲明
retrieve nodes:
A智能管理系統的功能參數
3.1 系統性能
併發處理能力：支持高併發訪問，滿足企業級應用需求。
響應時間：在高負載條件下，保持低延遲響應。
數據吞吐量：支持大規模數據的快速讀寫和處理。
3.2 兼容性
操作系統支持：兼容Linux和Windows操作系統。
數據庫兼容性：支持MySQL、PostgreSQL等主流數據庫。
3.3 安全性
數據加密：支持靜態數據和傳輸數據的加密。
訪問控制：基於角色的權限管理，確保數據安全。

1. 產品合規性聲明
公司在產品的設計、開發、生產和銷售過程中，嚴格遵守以下法律法規：
《產品質量法》——確保產品符合國家質量標準，不存在虛假宣傳。
《消費者權益保護法》——保障用户在使用產品過程中的合法權益。
《數據安全法》——嚴格保護用户個人信息，未經授權不得向第三方泄露。
2. 知識產權與法律責任
本公司產品中涉及的所有代碼、算法和技術均受《著作權法》和《專利法》保護。
用户不得擅自修改、複製或分發公司產品中的任何組件。
如因產品缺陷導致用户損失，公司承擔修復責任，但因用户使用不當或未遵守產品使用説明導致的損失，公司不承擔責任。
3. 合同履行責任
公司在合同中明確約定產品功能、交付標準和服務期限。
若因公司原因未履行合同約定內容，用户有權根據合同條款要求公司承擔違約責任。
若用户在產品中嵌入或調用外掛、腳本工具或未經授權的API，公司有權終止服務並保留追究法律責任的權利
"""

通過靈活的知識庫配置和可控的 RAG 召回服務，LazyLLM 能實現共享方式的多樣性。

四、對話管理以及解決方案

在企業實際業務中，對話系統需要能記住用户之前的聊天內容，並根據歷史對話更好地理解用户當前的需求。比如，客服機器人需要知道用户之前問過什麼問題，避免重複回答；或者AI助手能結合之前的聊天內容，優化當前的問題，讓回答更精準。

同時，系統要支持多用户同時使用，確保不同用户的對話互不干擾。比如，A用户在和機器人聊訂單問題，B用户在諮詢產品信息，系統要能區分他們的對話記錄，不會混淆。

此外，還要支持實時流式輸出，讓用户能像正常聊天一樣逐步看到回覆，而不是等待全部生成完才顯示。

為了實現這些功能，lazyllm提供了一個統一的配置中心globals，管理不同用户的對話歷史、上下文信息等，並在對話結束後自動清理不必要的數據，避免資源浪費。

通過 globals，系統能夠隔離不同會話的對話歷史與上下文，避免數據干擾，同時集中保存用户對話參數、傳入的文件路徑以及對話過程中產生的中間結果，供後續處理環節使用。這種設計保證了跨服務之間的數據一致性與高效流轉。

接下來介紹如何通過globals 實現歷史對話管理和多用户併發的對話管理。

（一）歷史對話管理

首先我們通過結合 globals 配置中心，實現一個支持歷史上下文、流式輸出的對話流程。lazyllm.OnlineChatModule(stream=True)初始化一個支持流式輸出的大語言模型，並使用 ThreadPoolExecutor 建立線程池，支持最多 50 路併發請求。每個請求在 slots 中分配一個空閒位置，確保同一時刻不會超過設定的最大連接數。

• 每次新會話啓動時，init\_session\_config 會結合默認 few-shot 提示與用户自定義的歷史對話，統一初始化到對應的 session_id 下的 globals 空間中，保證每個會話擁有獨立的上下文。
• 為了安全地在多線程環境中管理模型推理過程，respond_stream 方法在提交推理任務時使用了 contextvars 拷貝當前上下文，避免不同會話間上下文變量串聯。整個響應過程中，通過 FileSystemQueue 實現邊生成邊輸出的流式效果，並在推理結束後，把最新的用户輸入和助手回覆追加進會話歷史。

為了便捷地管理 session 生命週期，系統提供了with\_session 裝飾器自動完成上下文切換，而handle\_request 函數作為統一入口，能夠根據傳入的session_id 和歷史記錄，發起一輪新的流式對話。

這種機制不僅實現了多用户隔離、歷史記憶管理，還確保了高併發場景下的穩定性和一致性，為後續接入更多複雜交互提供了基礎支撐。

（代碼GitHub鏈接：https://github.com/FFFFFancy/Tutorial/blob/a09a84cdf0585a5c9d52af6db0e965be95d03123/rag/codes/chapter17/chat\_with\_history.py#L10）

from lazyllm import globals


llm = lazyllm.OnlineChatModule(stream=True)
threadPool = ThreadPoolExecutor(max_workers=50)
slots = [0] * 50

# 公共 few shot 歷史
DEFAULT_FEW_SHOTS = [
    {"role": "user", "content": "你是誰？"},
    {"role": "assistant", "content": "我是你的智能助手。"}
]

class ChatHistory(BaseModel):
    user: str
    assistant: str

class ChatRequest(BaseModel):
    user_input: str
    history: Optional[List[ChatHistory]] = None

def allocate_slot():
    for idx, val in enumerate(slots):
        if val == 0:
            slots[idx] = 1
            return idx
    return -1

def release_slot(session_id):
    if 0 <= session_id < len(slots):
        slots[session_id] = 0
        globals.pop(session_id)

def init_session_config(session_id, user_history=None):
    globals._init_sid(session_id)

    if user_history is not None:
        history = []
        # 合併 few-shot + 用户歷史
        history.extend(DEFAULT_FEW_SHOTS)
        for h in user_history:
            history.append({"role": "user", "content": h.user})
            history.append({"role": "assistant", "content": h.assistant})
        globals["global_parameters"]["history"] = history
    else:
        if "history" not in globals["global_parameters"]:
            globals["global_parameters"]["history"] = copy.deepcopy(DEFAULT_FEW_SHOTS)

def with_session(func):
    def wrapper(session_id, *args, **kwargs):
        globals._init_sid(session_id)
        return func(session_id, *args, **kwargs)
    return wrapper

class SessionResponder:
    def __init__(self):
        pass

    def respond_stream(self, session_id, model_in, user_history=None):
        init_session_config(session_id, user_history)

        print("[Respond Stream] Current SID:", globals._sid)
        history = globals["global_parameters"]["history"]
        print("history", history)

        ctx = contextvars.copy_context()
        func_future = threadPool.submit(lambda: ctx.run(llm, model_in, llm_chat_history=history))

        response = ''

        while True:
            assert session_id == globals._sid, f"\nSession ID mismatch: expected {session_id}, got {globals._sid}"

            if message := FileSystemQueue().dequeue():
                msg = "".join(message)
                response += msg
                yield msg
            elif func_future.done():
                break

        model_out = func_future.result()

        assert session_id == globals._sid, f"Session ID mismatch after LLM: expected {session_id}, got {globals._sid}"

        # 更新歷史
        globals["global_parameters"]["history"].append({
            "role": "user",
            "content": model_in
        })
        globals["global_parameters"]["history"].append({
            "role": "assistant",
            "content": model_out
        })

        return model_out

@with_session
def handle_request(session_id: str, user_input: str, user_history: Optional[List[ChatHistory]] = None):
    chat = SessionResponder()
    for chunk in chat.respond_stream(session_id, model_in=user_input, user_history=user_history):
        print(chunk, end='', flush=True)

整體來看，這段代碼依託 globals 可實現：

• 靈活加載和隔離管理不同用户的歷史對話；
• 支持系統內部預置 few-shot 示例，引導模型更好地理解任務；
• 在結合歷史上下文的基礎上，總結、改寫並生成新的對話內容或問題。

讓我們來看一下執行效果：

###################################
# 使用歷史對話
###################################
history = [
    ChatHistory(
        user="香蕉的英文是什麼？",
        assistant="香蕉的英文是banana"
    ),
    ChatHistory(
        user="那蘋果呢？",
        assistant="蘋果的英文是apple"
    )
]
user_input = "那橘子呢？"

response = ""
for chunk in respond_stream(session_id, user_input, history):
    response += chunk
# >>> 橘子的英文是orange。

#####################################
# 內部預置 few-shot 示例
#####################################
DEFAULT_FEW_SHOTS = [
    {"user": "請幫我改寫：'這個報告寫得還可以。'", 
     "assistant": "這份報告整體表現良好，但仍有提升空間。"},
    {"user": "請幫我優化：'我們的銷售業績不錯。'", 
     "assistant": "我們的銷售業績表現出色，達到了預期目標。"},
]

user_input = "請幫我改寫：'客户反饋我們服務態度很好。'"
# >>> 客户對我們服務態度給予了積極評價。


#####################################
# 總結歷史示例
#####################################
history = [
    ChatHistory(
        user="機器學習是什麼？",
        assistant="機器學習是一門開發算法和統計模型的科學，計算機系統使用這些算法和模型，在沒有明確指令的情況下，依靠既有模式和推理來執行任務"
    ),
    ChatHistory(
        user="機器學習的應用場景？",
        assistant="機器學習被廣泛應用於推薦系統中，如電商網站的商品推薦、社交媒體的內容推薦等。通過分析用户的歷史行為和偏好，機器學習算法可以預測用户可能感興趣的內容，並提供個性化的推薦。自然語言處理：自然語言處理是機器學習的另一個重要應用領域，包括語音識別、機器翻譯、情感分析、垃圾郵件過濾等。機器學習算法可以幫助計算機理解和生成人類語言，實現人機交互的智能化。圖像識別和處理：機器學習在圖像識別和處理方面也發揮着重要作用，如人臉識別、車牌識別、圖像檢索、物體識別等。通過訓練大量的圖像數據，機器學習算法可以學習並識別出圖像中的特徵，從而實現對圖像的智能處理。"
    )
]
user_input = "總結下這段對話"

# >>> 這段對話主要圍繞機器學習展開，首先解釋了機器學習的定義，即通過算法和統計模型，讓計算機系統在沒有明確指令的情況下，基於既有模式和推理完成任務。接着討論了機器學習的應用場景，包括推薦系統（如電商和社交媒體個性化推薦）、自然語言處理（如語音識別、機器翻譯、情感分析等）以及圖像識別和處理（如人臉識別、物體識別等）。最後總結了機器學習在智能化任務中的重要作用。

• 第一次請求只輸入 “橘子”，系統按歷史上下文正常生成與橘子相關的回答“orange”。
• 第三次請求輸入 "總結這段對話"，系統基於完整歷史成功輸出對話總結。
• 並且預先指定的對話也在ChatHistory中。

（二）多用户併發對話管理

同樣基於上述設計，以下代碼進一步完善了基於 globals 的多用户會話管理與歷史對話追蹤機制。系統初始化了一個流式推理模塊 lazyllm.OnlineChatModule(stream=True)，並通過 ThreadPoolExecutor 支持最多 50 個併發請求，同時通過 slots 數組管理連接資源，保證不同用户會話互不干擾。

• 每當新的請求到來時，init\_session\_config 方法根據傳入的用户歷史，初始化對應 session\_id 下的上下文環境，若無歷史則默認分配一個空白對話歷史，確保每條會話軌跡獨立。為了簡化多次會話調用的上下文切換，with\_session 裝飾器自動在執行函數前綁定正確的 session_id。
• 實際推理時，SessionResponder 類負責發起流式對話，內部使用 contextvars 捕獲當前上下文，保證即使在線程池中執行推理任務，也能維持正確的會話隔離。系統通過 FileSystemQueue 實現流式輸出，在推理過程中實時返回生成的內容，推理完成後，再將本輪對話完整地追加到歷史記錄中，以便後續連續對話使用。

最後，通過外部示例展示瞭如何用 handle_request 函數發起多輪對話，系統能夠正確地維護多個用户之間獨立且連續的對話流，保證不同用户的歷史上下文不會混淆，為支持高併發、強上下文連貫性的應用場景打下了良好的基礎。

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/a09a84cdf0585a5c9d52af6db0e965be95d03123/rag/codes/chapter17/chat\_with\_multi_user.py#L11）

llm = lazyllm.OnlineChatModule(stream=True)
threadPool = ThreadPoolExecutor(max_workers=50)
slots = [0] * 50

DEFAULT_FEW_SHOTS = []

class ChatHistory(BaseModel):
    user: str
    assistant: str

class ChatRequest(BaseModel):
    user_input: str
    history: Optional[List[ChatHistory]] = None

def allocate_slot():
    for idx, val in enumerate(slots):
        if val == 0:
            slots[idx] = 1
            return idx
    return -1

def release_slot(session_id):
    if 0 <= session_id < len(slots):
        slots[session_id] = 0
        globals.pop(session_id)

llm = lazyllm.OnlineChatModule(stream=True)

def init_session_config(session_id, user_history=None):
    globals._init_sid(session_id)
    # if globals._sid not in globals._Globals__data:
    #     globals._Globals__data[globals._sid] = copy.deepcopy(globals.__global_attrs__)

    if user_history is not None:
        globals["global_parameters"]["history"] = user_history
    else:
        if "history" not in globals["global_parameters"]:
            globals["global_parameters"]["history"] = []

def with_session(func):
    """自動綁定 session_id 的裝飾器"""
    def wrapper(session_id, *args, **kwargs):
        globals._init_sid(session_id)
        return func(session_id, *args, **kwargs)
    return wrapper

class SessionResponder:
    def __init__(self):
        pass

    def respond_stream(self, session_id, model_in, user_history=None):
        init_session_config(session_id, user_history)

        print("[Respond Stream] Current SID:", globals._sid)
        history = globals["global_parameters"]["history"]
        print("history", history)

        # 捕獲當前上下文（確保線程池提交的任務也帶上下文）
        ctx = contextvars.copy_context()
        func_future = threadPool.submit(lambda: ctx.run(llm, model_in, llm_chat_history=history))

        response = ''

        while True:
            assert session_id == globals._sid, f"\nSession ID mismatch: expected {session_id}, got {globals._sid}"

            if message := FileSystemQueue().dequeue():
                msg = "".join(message)
                response += msg
                yield msg
            elif func_future.done():
                break

        model_out = func_future.result()

        assert session_id == globals._sid, f"Session ID mismatch after LLM: expected {session_id}, got {globals._sid}"

        # globals["global_parameters"]["history"].append(model_out)
        globals["global_parameters"]["history"].append({
            "role": "user",
            "content": model_in
        })
        globals["global_parameters"]["history"].append({
            "role": "assistant",
            "content": model_out
        })

        return model_out

# 外部使用示例
@with_session
def handle_request(session_id: str, user_input: str):
    chat = SessionResponder()
    for chunk in chat.respond_stream(session_id, model_in=user_input):
        print(chunk, end='', flush=True)

if __name__ == "__main__":

    handle_request("user321", "蘋果的英文是什麼！")
    print("\n\n")
    handle_request("user123", "機器學習是什麼")
    print("\n\n")
    handle_request("user321", "香蕉呢")
    print("\n\n")
    handle_request("user123", "它有什麼用？")

效果示例：

• 用户 id1 問“蘋果的英文”，再問“香蕉”時，模型能記住當前會話是翻譯任務。
• 用户 id2 問“機器學習是什麼？”後，追問“它有什麼作用？”時，模型能關聯上下文解釋應用場景。

二者維度獨立的歷史對話內容，相互不影響。

🚨注意，實現上述功能需要用redis數據庫實現文件系統輸出管理，設置方法為：

export LAZYLLM\_DEFAULT\_FSQUEUE=SQLITE

export LAZYLLM\_FSQREDIS\_URL=redis://[user name]:[password]@[host]/[port]

五、安全需求以及解決方案

最後，企業在構建知識庫的過程中，不僅要保障內部的信息安全，還需關注面向公眾時的外部安全，防止數據泄露、信息濫用以及潛在的合規風險。

（一）企業安全

在企業知識庫建設中，安全始終是首要考慮因素。尤其是當知識庫中包含公司政策、財務報表、客户合同等高度敏感或私有數據時，任何信息泄露都可能帶來嚴重的法律責任和商業損失。為此，系統需具備完善的保護機制。

1. 加密

• 私有數據保護：通過數據隔離機制，確保不同業務或部門間的數據隔離，防止數據泄露。
• 知識加密：對文檔在存儲與傳輸過程中的全鏈路加密，確保數據機密性與完整性。
• 模型加密：支持模型調用過程中的數據加密，避免敏感信息泄露。

2. 私有化部署

• 本地化模型推理引擎：核心組件部署於內網環境，保障數據安全。
• 數據本地處理：確保知識數據在企業內部完成，避免外泄。
• 強化權限控制：結合網絡隔離和多因子認證，實現安全訪問。

3. 信創

為保障核心技術自主可控，系統全面兼容國家信創名錄中的軟硬件產品。

• 國產CPU：鯤鵬、龍芯等，提供高性能計算支持。

• 國產操作系統：麒麟、統信UOS等，確保系統底層安全。

• 國產數據庫：達夢、人大金倉等，敏感數據存得更放心。

• 全鏈路合規：從芯片（如鯤鵬/飛騰）到軟件均符合信創標準，通過國家信息安全認證。

（二）公眾安全

在企業級RAG系統中，公共安全不僅關乎企業自身的聲譽與合規風險，更關聯到模型輸出對社會輿論、信息安全乃至國家安全的影響。系統應具備以下能力，確保模型生成內容不突破公共安全底線：

• 涉暴涉恐內容識別：自動檢測與暴力、恐怖主義、極端言論相關的內容，防止模型成為非法信息的傳播通道。
• 涉政敏感內容攔截：對分裂言論、非法組織宣傳、造謠煽動等進行精準識別與阻斷，避免引發政治敏感風險。
• 羣體性事件識別：識別模型輸出中可能煽動公眾聚集、對抗或恐慌的內容，防範輿情發酵升級。
• 突發事件內容引導：在重大災害、公共衞生或社會事件發生時，提供經過驗證的信息輸出機制，減少虛假內容傳播風險。
• 跨境內容安全保障：在涉外輸出場景中，支持配置涉敏國家/地區、組織和外交事件的識別策略，確保對外表態合法合規。
• 安全聯動機制：支持與企業安全系統、監管平台或應急響應機制打通，實現內容風險的實時發現與快速響應。

通過公共安全模塊的建設，企業可有效防控大模型在生成內容過程中可能引發的社會層面風險，提升企業數字治理能力，踐行平台責任。

1. 如何維護公共安全 ？

在知識庫管理和檢索過程中，可能涉及一些敏感信息或違規內容，如個人隱私、法律合規性條款等。因此，系統需要具備全面的過濾機制，在數據上傳、存儲和檢索階段，自動識別和屏蔽敏感詞彙，防止敏感信息的誤用或泄露。

LazyLLM支持靈活的自定義規則配置，管理員可以根據企業實際需求，動態維護敏感詞列表，結合分詞、正則表達、DFA（Deterministic Finite Automaton）等算法實現精準過濾。

接下來我們來看如何將敏感詞過濾的典型算法——DFA算法，接入lazyllm實現敏感詞過濾。

第一步：

用定義lazyllm模塊的方式實現DFA算法的定義 ，將其包裝為可接入lazyllm的組件，詳見 [ 第8講：不止是cosine！匹配策略決定你召回的質量 ] - 基於 class 的自定義 Transform 算法。

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/7abc91dbb82a007a78731845dd8c360ac0cc1e75/rag/codes/chapter17/retriever\_with\_DFA.py#L10）

from lazyllm.tools.rag import DocNode, NodeTransform
from typing import List

# 定義DFA算法
class DFAFilter:
    def __init__(self, sensitive_words):
        self.root = {}
        self.end_flag = "is_end"
        for word in sensitive_words:
            self.add_word(word)

    def add_word(self, word):
        node = self.root
        for char in word:
            if char not in node:
                node[char] = {}
            node = node[char]
        node[self.end_flag] = True

    def filter(self, text, replace_char="*"):
        result = []
        start = 0
        length = len(text)

        while start < length:
            node = self.root
            i = start
            while i < length and text[i] in node:
                node = node[text[i]]
                if self.end_flag in node:
                    # 匹配到敏感詞，替換為指定字符
                    result.append(replace_char * (i - start + 1))
                    start = i + 1
                    break
                i += 1
            else:
                # 未匹配到敏感詞，保留原字符
                result.append(text[start])
                start += 1

        return ''.join(result)


# 註冊為transform
class DFATranform(NodeTransform):
    def __init__(self, sensitive_words: List[str]):
        super(__class__, self).__init__(num_workers=num_workers)
        self.dfafilter = DFAFilter(sensitive_words)

    def transform(self, node: DocNode, **kwargs) -> List[str]:
        return self.dfafilter.filter(node.get_text())

    def split_text(self, text: str) -> List[str]:
        if text == '':
            return ['']
        paragraphs = text.split(self.splitter)
        return [para for para in paragraphs]

第二步：

將定義的DFAFilter 註冊為文檔服務的node group。同樣使用上小節中"A產品功能參數和產品合規性聲明"的檢索問題，假設屏蔽【合同】這個詞，只需DFAFilter註冊為新的節點，並通過parent="sentences"繼承上一步的處理方式。

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/7abc91dbb82a007a78731845dd8c360ac0cc1e75/rag/codes/chapter17/retriever\_with\_DFA.py#L65）

from lazyllm import Retriever，SentenceSplitter
# 定義業務敏感詞
sensitive_words = ['合同']
# 將敏感詞過濾算法嵌入到業務邏輯中
Document.create_node_group(name="sentences", transform=SentenceSplitter, chunk_size=128, chunk_overlap=10)
Document.create_node_group(name="dfa_filter", parent="sentences"，transform=DFATranform(sensitive_words))

# 組合法務 + 產品知識庫，處理與產品相關的法律問題
retriever_product = Retriever(
    [law_knowledge_base, product_knowledge_base],
    group_name="dfa_filter",       # 指定 dfa_filter
    similarity="cosine",       
    topk=2                
)

product_question = "A產品功能參數和產品合規性聲明"
product_res_nodes = retriever_product(product_question)

可以看到，輸出結果已將“敏感詞”和“合同”替換為星號。在企業應用場景中，可以根據業務需求自定義敏感詞庫，以增強數據安全性。

屏蔽前：

屏蔽後：

全流程敏感詞過濾 在實際應用中，除了原文檔內容進行敏感詞過濾外，我們還需對用户輸入和大模型輸出進行同樣的處理。

with pipeline() as ppl:
    ppl.query_filter = lambda x: DFAFilter(sensitive_words).filter(x)
    ppl.retriever = Retriever(...)
    ppl.reranker = ...
    ppl.llm = ...
    ppl.output_filter = lambda x: DFAFilter(sensitive_words).filter(x)

針對企業級需求，lazyllm提供了靈活的文檔管理和召回服務。通過配置不同的算法和知識庫，系統能夠在不同業務場景下，滿足跨部門的數據處理和精準召回需求。

藉助數據庫管理功能，系統實現了數據隔離和權限控制，有效保障私有數據的安全性。

同時，系統支持標籤檢索和敏感詞過濾，進一步提升檢索的精準度和合規性，幫助企業在複雜的數據環境中高效管理和利用知識庫。

六、企業級RAG的總體實現思路

在前文中，我們從權限控制、共享方式、安全保障等多個維度詳細解析了企業級RAG系統在真實落地過程中面臨的核心需求與挑戰。

接下來，我們將整合上述要素，提出一個功能完善、可落地的企業級RAG搭建思路。

（一）架構圖

該企業級RAG系統主要由用户接入、意圖識別、檢索器、知識庫、算法庫、重排序器以及大模型模塊組成。

用户通過接入模塊提交查詢，請求首先經過意圖識別模塊，判斷查詢意圖並動態決定後續檢索策略。檢索器模塊由多個Retriever組成，能夠靈活調用不同的知識庫和算法庫，完成多策略、多數據源的檢索任務。知識庫用於存儲各類結構化或非結構化文檔，算法庫則包含多種向量化工具和解析器，支持知識數據的編碼和處理，二者均可按需靈活組合。檢索完成後，候選結果經過重排序模塊優化相關性，再由大語言模型(LLM) 基於優化後的內容進行生成，最終輸出符合用户需求的答案。

整個系統設計強調模塊解耦、策略靈活和生成增強，適配多用户、高併發和多場景的企業應用需求。

（二）代碼實現

接下來，以一個電商場景為例，我們將構建一個具備上述功能的 RAG 問答系統。本次示例中共使用三個知識庫，數據準備如圖所示，數據庫的構建方法已在第二講中詳細介紹，此處不再贅述。

我們將產品知識庫與法務知識庫聯合檢索，用於處理產品及法務相關問題；同時將法務知識庫與用户支持知識庫組合，以應對用户支持類問題。為此，首先構建兩條獨立的 RAG pipeline。

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/7abc91dbb82a007a78731845dd8c360ac0cc1e75/rag/codes/chapter17/ecommerce_rag.py#L180）

with pipeline() as product_law_ppl:
    product_law_ppl.retriever = retriever = Retriever(
            [doc_law, doc_product],
            group_name="dfa_filter",   
            topk=5, 
            embed_keys=['dense'],
        )
    product_law_ppl.reranker = Reranker(name="ModuleReranker",
                            model=OnlineEmbeddingModule(type='rerank'),
                            topk=2, output_format="content", join=True) | bind(query=product_law_ppl.input)
    product_law_ppl.formatter = (lambda nodes, query: dict(context_str=nodes, query=query)) | bind(query=product_law_ppl.input)
    product_law_ppl.llm = OnlineChatModule().prompt(lazyllm.ChatPrompter(prompt, extra_keys=["context_str"]))

with pipeline() as support_law_ppl:
    support_law_ppl.retriever = retriever = Retriever(
            [doc_law, doc_support],
            group_name="dfa_filter",   
            topk=5, 
            embed_keys=['dense'],
        )
    support_law_ppl.reranker = Reranker(name="ModuleReranker",
                            model=OnlineEmbeddingModule(type='rerank'),
                            topk=2, output_format="content", join=True) | bind(query=support_law_ppl.input)
    support_law_ppl.formatter = (lambda nodes, query: dict(context_str=nodes, query=query)) | bind(query=support_law_ppl.input)
    support_law_ppl.llm = OnlineChatModule().prompt(lazyllm.ChatPrompter(prompt, extra_keys=["context_str"]))

為用户提供統一的問答入口，並實現不同知識庫間的無縫切換，我們引入了用户意圖識別模塊，能夠根據查詢內容自動選擇合適的 RAG pipeline 進行處理。

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/7abc91dbb82a007a78731845dd8c360ac0cc1e75/rag/codes/chapter17/ecommerce_rag.py#L195）

def build_ecommerce_assistant():
    llm = OnlineChatModule(source='qwen', stream=False)
    intent_list = [
        "產品法務問題",
        "用户支持問題",
    ]

    with pipeline() as ppl:
        ppl.classifier = IntentClassifier(llm, intent_list=intent_list)
        with lazyllm.switch(judge_on_full_input=False).bind(_0, ppl.input) as ppl.sw:
            ppl.sw.case[intent_list[0], product_law_ppl]
            ppl.sw.case[intent_list[1], support_law_ppl]
    return ppl

為了實現多用户併發會話請求並維護獨立的上下文，我們通過 globals 管理器封裝了 EcommerceAssistant，確保用户問答的隔離性。

（代碼GitHub鏈接：https://github.com/LazyAGI/Tutorial/blob/7abc91dbb82a007a78731845dd8c360ac0cc1e75/rag/codes/chapter17/ecommerce_rag.py#L223）

def init_session(session_id, user_history: Optional[List[ChatHistory]] = None):
    globals._init_sid(session_id)

    if "global_parameters" not in globals or "history" not in globals["global_parameters"]:
        globals["global_parameters"]["history"] = []

    if not globals["global_parameters"]["history"]:
        # 初始化為 default few-shot
        globals["global_parameters"]["history"].extend(DEFAULT_FEW_SHOTS)

    if user_history:
        for h in user_history:
            globals["global_parameters"]["history"].append({"role": "user", "content": h.user})
            globals["global_parameters"]["history"].append({"role": "assistant", "content": h.assistant})

def build_full_query(user_input: str):
    """根據 globals 裏的歷史，生成帶歷史的 full query文本"""
    history = globals["global_parameters"]["history"]
    history_text = ""
    for turn in history:
        role = "用户" if turn["role"] == "user" else "助手"
        history_text += f"{role}: {turn['content']}\n"

    full_query = f"{history_text}用户: {user_input}\n助手:"
    return full_query

class EcommerceAssistant:
    def __init__(self):
        self.main_pipeline = build_ecommerce_assistant()

    def __call__(self, session_id: str, user_input: str, user_history: Optional[List[ChatHistory]] = None):
        init_session(session_id, user_history)

        full_query = build_full_query(user_input)

        # 把帶歷史的 query 輸入主 pipeline
        response = self.main_pipeline(full_query)

        # 更新歷史到 globals
        globals["global_parameters"]["history"].append({"role": "user", "content": user_input})
        globals["global_parameters"]["history"].append({"role": "assistant", "content": response})

        return response

（三）結果分析

運行日誌：

==================== user1：用户支持問題 ====================
用户 user1 提問：
「用户投訴某智能手錶的續航沒有達到宣傳效果，該怎麼處理」

助手回覆：
當用户投訴智能手錶續航未達到宣傳效果時，我們需要按照三級響應機制進行處理：

1. **技術驗證**：請用户提供設備序列號及續航測試視頻（需顯示完整充放電週期）。我們內部將通過工具[PERF_CHECK_V3]對比宣傳參數。

2. **法律評估**：觸發法務系統自動生成《參數差異分析報告》，參考案例FC-2024-021進行詳細對比。

3. **解決方案**：
   - 如果差異≤15%：我們將贈送用户1年延保（服務代碼SV-228）。
   - 如果差異＞15%：根據《消費者權益保護法》第23條，我們將提供換新或差價三倍賠償。

溝通話術："我們高度重視產品參數的準確性，將委託第三方機構（如SGS）進行復測，並在3個工作日內給您正式答覆。"

依據《消費者權益保護法》第23條，商品實際性能與宣傳參數差異超過行業標準允許誤差範圍（電子設備續航誤差±15%）即構成虛假宣傳。參考案例FC-2024-021：某品牌因智能手錶續航虛標28%被判定三倍賠償，並受到市場監督管理局50萬元行政處罰。

建議技術部門建立續航測試標準流程（需符合GB/T 35143-2023標準），所有宣傳數據必須附帶測試環境説明（如：實驗室環境25℃下連續使用）。

============================================================

====================user2：產品法務問題（帶歷史對話） ====================
用户 user2 的對話歷史：
1. 用户: 「你好」
   助手: 「你好呀！」
2. 用户: 「我想諮詢耳機宣傳內容是否合規」
   助手: 「當然，請詳細描述你的宣傳文案。」

用户 user2 新提問：
「骨傳導耳機不展示專利號」

助手回覆：
根據提供的信息，如果骨傳導耳機的宣傳中涉及已申請或已授權的專利技術（如骨傳導振子技術），但未明確標註專利號和專利類型（發明、實用新型或外觀設計），這可能違反《廣告法》第十二條以及相關實施細則的要求。

以下是具體分析：

1. **專利號標註問題**  
   - 根據法律規定，在廣告宣傳中提及專利技術時，必須完整標註專利號及專利類型。例如：“本產品採用骨傳導振子技術，專利號：ZL2024XXXXXXX，實用新型專利。”
   - 如果僅提到“已申請專利”而未標明具體專利號，則屬於不合規行為。參考AD-2024-017號處罰案例，某科技公司因未標註專利號被罰款87.6萬元。

2. **跨境廣告要求**  
   - 若該耳機宣傳涉及跨境市場（如出口到其他國家），還需同步標註專利授權國家代碼（如CN/EP/US）。例如：“專利號：ZL2024XXXXXXX，CN實用新型專利。”

3. **技術獨佔性聲明合規性**  
   - 宣傳文案需遵循“三不原則”：
     - 不暗示技術壟斷（避免使用“唯一”“獨家”等絕對化用語）。
     - 不擴大保護範圍（確保與專利權利要求書嚴格對應）。
     - 不隱瞞專利狀態變更（如專利被宣告無效，需在3個工作日內更新宣傳物料）。

### 建議
- **立即補充專利信息**：在宣傳中明確標註專利號及類型。如果專利正在申請中，應註明“專利申請中”而非直接宣稱“已申請專利”。
- **避免絕對化用語**：不要使用“全球領先”“唯一”等表述，以免引發虛假宣傳風險。
- **定期檢查專利狀態**：確保宣傳內容與實際專利狀態一致，及時更新。

如果你無法提供具體的專利號或專利類型，建議先調整宣傳文案，避免因違規而遭受處罰。例如，可以改為描述功能性特點，如“採用先進的骨傳導技術，實現IPX7級防水性能”。

============================================================

====================user1 ：用户支持問題跟進 ====================
用户 user1 繼續提問：
「這種投訴處理方式有什麼風險嗎？」

助手回覆：
這種處理方式雖然遵循了法律法規和行業標準，但仍可能存在以下潛在風險：

1. **用户信任受損**：如果用户的設備續航差異較大但未超過15%，僅提供延保服務可能無法完全修復用户體驗，進而影響品牌忠誠度。

2. **測試環境爭議**：宣傳數據通常基於實驗室條件（如25℃下連續使用），而實際使用場景複雜多變（高温、低温、高負載等）。若用户對測試環境説明不瞭解或不認可，可能引發進一步投訴。

3. **法務風險擴大**：如果部分產品續航虛標超過15%，且涉及較大用户羣體，可能會被認定為系統性虛假宣傳，從而面臨集體訴訟或更嚴重的行政處罰。

4. **技術驗證效率問題**：通過工具[PERF_CHECK_V3]進行對比需要一定時間，若響應速度過慢，可能導致用户不滿升級為輿情事件。

### 風險規避建議：
- **主動優化宣傳策略**：在產品頁面及包裝中明確標註續航測試環境（如“實驗室環境下25℃連續使用”），並補充説明實際使用可能存在的差異。
- **建立快速響應機制**：針對續航類投訴，設立專項客服團隊，確保在48小時內完成初步評估並向用户提供解決方案。
- **加強內部流程管控**：技術部門需定期更新續航測試標準流程，確保符合GB/T 35143-2023要求，並將測試結果與市場宣傳同步校準。
- **提供額外補償措施**：對於續航差異接近臨界值的用户，可考慮贈送配件（如充電器）或延長保修期，以提升用户滿意度。

最終目標是通過透明化溝通和積極應對，將潛在風險降至最低，同時維護品牌形象和用户信任。

根據以上日誌，我們實現了：

1. 自動意圖識別

RAG系統通過意圖識別功能，能夠自動分析用户的提問並選擇合適的處理流程（Pipeline）。

例如，當收到用户支持性問題時，其結合了用户支持庫的“三級響應機制”提出溝通話術，並結合和法規庫的《消費者權益保護法》第23條提出針對性的法律規定；當接收到產品法規問題時，其從產品庫中檢索到骨傳導振子技術相關內容，並結合法規庫的技術獨佔性聲明合規性相關內容。

這種意圖識別功能能夠使系統更高效地響應不同類型的問題，並且通過自動選擇合適的Pipeline，使得用户的需求得到快速而準確的滿足，減少了人工干預的需求，提升了整體響應速度和服務質量。

2. 多知識庫聯合檢索

在助手的回答中，我們可以看到不同知識庫內容的聯合應用。

首先，關於智能手錶續航的投訴問題，助手不僅依據技術文檔提供了具體的測試方法，還引用了法務知識庫中的條款和相關案例，全面涵蓋了產品質量、消費者權益等多維度的信息。

同樣，針對耳機宣傳合規性問題，助手結合了廣告法和專利法知識庫中的內容，提供了詳細的法律分析和合規建議。

這展示了RAG能力在處理複雜問題時的靈活性和高效性，通過跨多個領域的知識庫聯合提供精準答案。

3. 用户歷史對話分離

在回答中引入用户的歷史對話並加以分離。

對於用户user2的追加提問（投訴處理方式），通過引用歷史對話，保證了對用户之前提問的背景理解，確保了回答的準確性和連貫性。

4. 指定歷史對話

用户的歷史對話可以被明確指定，以便提供更加個性化和細化的回答。

例如，user2在諮詢耳機專利問題時，助手依據其前面的諮詢內容“耳機宣傳內容是否合規”進行相關的法律合規回答，並且通過引入歷史對話數據，確保了回答的針對性和層次性，避免了重複性回答，並通過“補充專利信息”等建議，強化了問題的解決路徑。

更多技術內容，歡迎移步 “LazyLLM” 討論！