使用 Chainlit 调用 TextGen API + 本地 LLM 实现 RAG Chatbot 应用

1. 引言

随着大语言模型（LLM）的快速发展，检索增强生成（RAG）已成为构建知识密集型应用的主流架构。然而，在实际部署中，我们常常面临两个核心挑战：如何快速构建一个交互式的演示界面，以及如何将 RAG 系统与本地部署的 LLM 高效集成。

本文将介绍一个完整的解决方案：利用Chainlit框架快速搭建聊天式前端界面，通过TextGen API统一接口调用本地部署的 LLM（如 Llama、Qwen、ChatGLM 等），并结合向量数据库构建一个端到端的 RAG 应用。这套方案兼顾了开发效率、部署灵活性和成本控制。

2. 技术栈与工具准备

2.1 在开始之前，请确保你的开发环境已安装以下工具和库：

• TextGen (oobabooga/textgen)或兼容的 API 服务：提供统一的 embedding、LLM API 端点。
• Local LLM：本地运行的大型语言模型，用于 TextGen，例如：all-mpnet-base-v2 、Qwen3-4B-Q4_K_M.gguf 和 Qwen3.5-4B-Q4_K_M.gguf 。
• Sentence Transformers：用于生成文本嵌入向量，结合 embedding LLM 使用。
• PostgreSQL：数据安裝PGVector扩展，支援向量。
• uv：An extremely fast Python package and project manager, written in Rust. uv repo
• Chainlit：用于构建 LLM 应用的 UI 框架。
• LangChain：用于编排 RAG 流程（文档加载、切分、向量化、检索）。

2.2 本地 Windows 项目环境，你可以通过以下命令安装核心依赖：

2.2.1 安装 uv

# 安装 uv: uv 并不需要 Python，透过 PowerShell 安装，避免与 Python 环境绑定。 powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

2.2.2 创建 Chainlit 项目

Get started with Overview - Chainlit

2.2.3 创建 uv 虚拟环境

进行项目文件夹，执行以下命令指定 Python 版本

uv venv --python 3.12

2.2.4 启动 uv 虚拟环境

.\.venv\Scripts\activate

如果启动出错，查看执行权限

Get-ExecutionPolicy

如果返回 Restricted，设定 ExecutionPolicy

Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned

2.2.5 安装依赖包 (uv sync)

成功启动虚拟环境后，使用uv sync自动读取pyproject.toml中的依赖并安装。

uv sync

pyproject.toml

[project] name = "ai-chatbot" version = "0.1.0" description = "This is a Python project for AI chatbot." readme = "README.md" requires-python = ">=3.12" dependencies = [ "chainlit==2.6.2", "fastapi", "langchain==0.3.27", "langchain-openai==0.3.28", "azure-identity", "pandas", "langchain-community", "langchain-postgres", "langchain-core", "psycopg[binary]", "openpyxl", "pypdf", "langfuse==2.60.5", "ipykernel>=7.1.0", "grandalf>=0.8", "python-dotenv", "uvicorn", ]

2.2.6 使用 uv pip 查看已安装的相关 Python 包

uv pip list

3. 搭建本地 LLM 服务（TextGen）

首先，我们需要一个本地运行的 LLM 服务。这里以TextGen为例，它支持多种模型格式并提供了兼容 OpenAI 的 API 接口。

3.1 启动 TextGen

1. 下载发布版本并安装依赖（TextGen 安装教程：TextGen for Local LLM-CSDN博客）。
2. 下载你需要的 LLM 模型文件（如 Qwen3.5-4B-Q4_K_M.gguf）。
3. 下载你需要的 Embedding 模型文件 （ Embedding model 安装教程：TextGen 启用 embedding 并安装 all-mpnet-base-v2 模型教程-CSDN博客）。
4. 启动 WebUI 服务，并启用 API 模式：

export OPENEDAI_EMBEDDING_MODEL="/home/ai/textgen-4.9/user_data/models/embedding/all-mpnet-base-v2" export OPENEDAI_EMBEDDING_DEVICE="auto" # "cpu" or "cuda" or auto echo "EMBEDDING_MODEL: $OPENEDAI_EMBEDDING_MODEL" echo "EMBEDDING_DEVICE: $OPENEDAI_EMBEDDING_DEVICE" ./textgen --no-electron --listen --listen-host 0.0.0.0 --model Qwen3.5-4B-Q4_K_M.gguf

服务启动后，默认会在http://localhost:5000提供 API 服务。关键的端点包括：

• /v1/embeddings：用于产生向量。
• /v1/chat/completions：用于对话（推荐）。
• /v1/completions：用于文本补全。

3.2 验证 API 连通性

使用简单的 Python 脚本测试 API 是否正常工作：

import requests url = "http://localhost:5000/v1/chat/completions" headers = {"Content-Type": "application/json"} data = { "messages": [ { "role": "user", "content": "Hello!" } ], "temperature": 0.6, "top_p": 0.95, "top_k": 20, "stream": true } response = requests.post(url, json=data, headers=headers) print(response.json())

4. 构建 RAG 核心流程

接下来，我们使用 LangChain 构建一个标准的 RAG 流程，包括文档加载、文本切分、向量化存储和检索。

范例目标：将整理好的 QA 数据索引到 PostgreSQL 数据库，使用 PGVector 进行向量储存
功能概述：
1. 从 Excel 档案读取 QA 数据
2. 将数据转为 LangChain 的 Document 格式（包含问题和元数据）
3. 使用 TextGen Embeddings 将问题转为向量
4. 将向量和元数据储存到 PostgreSQL 的 PGVector

4.1 文档加载与处理

import pandas as pd from env_config import load_env_config from langchain_community.document_loaders.dataframe import DataFrameLoader from langchain_postgres import PGVector from local_llm import TextGenEmbeddings # ---------------------------------------- # Step 1: 加载 Excel 原始数据 # ---------------------------------------- def read_qa_excel(filepath: str, preview_rows: int = 1): df = pd.read_excel(filepath) # 格式化输出：显示数据预览 print("=" * 80) print(f"📊 Step 1: 加载 QA 数据 - 预览前 {preview_rows} 笔") print("-" * 80) print(df.head(preview_rows).to_string(index=False)) # 使用 to_string 会对齐显示字段 print("=" * 80 + "\n") return df # ---------------------------------------- # Step 2: 将 DataFrame 转为 Document（含 metadata: Answer, Category, URL) # ---------------------------------------- def convert_df_to_docs(df: pd.DataFrame, preview_rows: int = 1): loader = DataFrameLoader(df, page_content_column="question") docs = loader.load() # 格式化输出：显示 Document 预览 print("=" * 80) print(f"📝 Step 2: 将 QA 转换为 Document - 预览前 {preview_rows} 笔") print("-" * 80) for i, doc in enumerate(docs[:preview_rows]): print(f"📄 Document {i+1}") print(f"🔹 Content : {doc.page_content}") print("🔸 Metadata:") for key, value in doc.metadata.items(): print(f" - {key}: {value}") print("-" * 80) print("=" * 80 + "\n") return docs

4.2 向量化与存储

# ---------------------------------------- # Step 3: 建立 PGVector 数据库联机字符串 # ---------------------------------------- def build_pgvector_connection_string(host, port, database, user, password): connection_string = PGVector.connection_string_from_db_params( driver = "psycopg", host = host, port = port, database = database, user = user, password = password, ) print("=" * 80) print("🔗 Step 3: 建立 PGVector 数据库联机字符串") print("-" * 80) print(f"Connection String: {connection_string}") print("=" * 80 + "\n") return connection_string # ---------------------------------------- # Step 4: 将 Document 索引到 PGVector # ---------------------------------------- def index_docs_to_pgvector(docs, collection_name, connection_string, embeddings): print("=" * 80) print(f"🚀 Step 4: 开始索引数据到 PGVector（集合名称: {collection_name}）") print("-" * 80) vectorstore = PGVector.from_documents( embedding=embeddings, documents=docs, collection_name=collection_name, connection=connection_string, pre_delete_collection=True ) print(f"✅ 成功索引 {len(docs)} 笔数据！") print("=" * 80 + "\n") return vectorstore

5. 集成 Chainlit 与 TextGen API

Chainlit 的核心是定义一个@cl.on_message异步函数来处理用户消息。我们将在此函数中串联 RAG 检索和 LLM 生成。

5.1 创建 Chainlit 应用

首先，创建一个chainlit.md文件作为应用欢迎页：

# Welcome to Local RAG Assistant This app demonstrates a Retrieval-Augmented Generation (RAG) system powered by a local LLM via TextGen API.

5.2 编写配制文件 env_config.py：

PGVector / DB 连接资讯 和 TextGen local LLM API URL、API KEY

env_config.py

# env_config.py import os from dotenv import load_dotenv def load_env_config(): load_dotenv() config = { # PGVector / DB "PGVECTOR_HOST" : os.getenv('PGVECTOR_HOST'), "PGVECTOR_PORT" : os.getenv('PGVECTOR_PORT'), "PGVECTOR_DATABASE" : os.getenv('PGVECTOR_DATABASE'), "PGVECTOR_USER" : os.getenv('PGVECTOR_USER'), "PGVECTOR_PWD" : os.getenv('PGVECTOR_PWD'), # TextGen local LLM "TEXTGEN_BASE_URL": os.getenv('TEXTGEN_BASE_URL'), "TEXTGEN_API_KEY": os.getenv('TEXTGEN_API_KEY'), } return config

.env

# PGVector PGVECTOR_HOST = "127.0.0.1" PGVECTOR_PORT = "5432" PGVECTOR_DATABASE = "rag" PGVECTOR_USER = "rag_user" PGVECTOR_PWD = "rag_pwd" # TextGen local LLM TEXTGEN_BASE_URL = "http://{your TextGen host IP}:5000/v1" TEXTGEN_API_KEY = "{fake_textgen_api_key}"

5.3 编写 Local LLM 文件 local_llm.py

用于请求 TextGen API 实现 Embedding

local_llm.py

import requests from typing import List, Any from langchain_core.embeddings import Embeddings class TextGenEmbeddings(Embeddings): def __init__(self, base_url: str, api_key: str, encoding_format: str = "float", timeout: int = 60): self.base_url = base_url.rstrip("/") self.api_key = api_key self.encoding_format = encoding_format self.timeout = timeout def _call_api(self, inputs: List[str]) -> List[List[float]]: payload = {"input": inputs if isinstance(inputs, list) else [inputs], "encoding_format": self.encoding_format} resp = requests.post( f"{self.base_url}/embeddings", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", }, json=payload, timeout=self.timeout, ) resp.raise_for_status() data = resp.json() return [item["embedding"] for item in data["data"]] def embed_documents(self, texts: List[str]) -> List[List[float]]: return self._call_api(texts) def embed_query(self, text: str) -> List[float]: return self._call_api([text])[0]

5.4 准备 RAG 知识库

将文件存放在 rag_knowledge 下，例如：labor_qa_dataset.xlsx 和 labor_leave_policy.pdf

labor_qa_dataset.xlsx 包含的栏位如下：

labor_qa_dataset

question	category	url	content_type	answer
问题，例如：劳工工作年资满一年有5天特别休假，这5天是包含满半年的2天吗？	类别：工时(休息、休假、请假)	规章制度的出度URL	qa	回答

labor_leave_policy.pdf 规章制度

5.5 RAG 知识库转换为向量并存储

此章节内容在第4单元已涉及，但它不完整，此处提供完整代码：

5.5.1 Excel QA 转换为向量并存储

0_indexing_qapair.py

# ----- 0.Indexing QA Pair ----- # 范例目标：将整理好的 QA Pair 数据索引到 PostgreSQL 数据库，使用 PGVector 进行向量储存 # 功能概述： # 1. 从 Excel 档案读取 QA 数据 # 2. 将数据转为 LangChain 的 Document 格式（包含问题和元数据） # 3. 使用 TextGen Embeddings 将问题转为向量 # 4. 将向量和元数据储存到 PostgreSQL 的 PGVector import pandas as pd from env_config import load_env_config from langchain_community.document_loaders.dataframe import DataFrameLoader from langchain_postgres import PGVector from local_llm import TextGenEmbeddings # ---------------------------------------- # Step 1: 加载 Excel 原始数据 # ---------------------------------------- def read_qa_excel(filepath: str, preview_rows: int = 1): df = pd.read_excel(filepath) # 格式化输出：显示数据预览 print("=" * 80) print(f"📊 Step 1: 加载 QA 数据 - 预览前 {preview_rows} 笔") print("-" * 80) print(df.head(preview_rows).to_string(index=False)) # 使用 to_string 会对齐显示字段 print("=" * 80 + "\n") return df # ---------------------------------------- # Step 2: 将 DataFrame 转为 Document（含 metadata: Answer, Category, URL) # ---------------------------------------- def convert_df_to_docs(df: pd.DataFrame, preview_rows: int = 1): loader = DataFrameLoader(df, page_content_column="question") docs = loader.load() # 格式化输出：显示 Document 预览 print("=" * 80) print(f"📝 Step 2: 将 QA 转换为 Document - 预览前 {preview_rows} 笔") print("-" * 80) for i, doc in enumerate(docs[:preview_rows]): print(f"📄 Document {i+1}") print(f"🔹 Content : {doc.page_content}") print("🔸 Metadata:") for key, value in doc.metadata.items(): print(f" - {key}: {value}") print("-" * 80) print("=" * 80 + "\n") return docs # ---------------------------------------- # Step 3: 建立 PGVector 数据库联机字符串 # ---------------------------------------- def build_pgvector_connection_string(host, port, database, user, password): connection_string = PGVector.connection_string_from_db_params( driver = "psycopg", host = host, port = port, database = database, user = user, password = password, ) print("=" * 80) print("🔗 Step 3: 建立 PGVector 数据库联机字符串") print("-" * 80) print(f"Connection String: {connection_string}") print("=" * 80 + "\n") return connection_string # ---------------------------------------- # Step 4: 将 Document 索引到 PGVector # ---------------------------------------- def index_docs_to_pgvector(docs, collection_name, connection_string, embeddings): print("=" * 80) print(f"🚀 Step 4: 开始索引数据到 PGVector（集合名称: {collection_name}）") print("-" * 80) vectorstore = PGVector.from_documents( embedding=embeddings, documents=docs, collection_name=collection_name, connection=connection_string, pre_delete_collection=True ) print(f"✅ 成功索引 {len(docs)} 笔数据！") print("=" * 80 + "\n") return vectorstore # ---------------------------------------- # 主程序 # ---------------------------------------- def main(): labor_qa_dataset_path = "./rag_knowledge/labor_qa_dataset.xlsx" print("=" * 80) print("🌟 QA Pair 索引流程开始") print("=" * 80 + "\n") # 加载环境变量 config = load_env_config() # 加载 QA 数据 df = read_qa_excel(filepath = labor_qa_dataset_path) # 转换为 Document docs = convert_df_to_docs(df) # 初始化嵌入模型 embeddings = TextGenEmbeddings( base_url=config["TEXTGEN_BASE_URL"], api_key=config["TEXTGEN_API_KEY"], ) # 建立数据库联机 connection_string = build_pgvector_connection_string(config['PGVECTOR_HOST'], int(config['PGVECTOR_PORT']), config['PGVECTOR_DATABASE'], config['PGVECTOR_USER'], config['PGVECTOR_PWD'] ) # 索引到 PGVector collection_name = "labor_qa_dataset@20260717" index_docs_to_pgvector(docs, collection_name, connection_string, embeddings) if __name__ == "__main__": main()

将 QA Pair 向量化并储存至 PGVector：

python 0_indexing_qapair.py

5.5.1 PDF 文档内容转换为向量并存储

1_indexing_longtxt.py

# ----- 1.Indexing Long Text ----- # 范例目标： # 将 PDF 文件的长文本转为嵌入向量并索引至 PGVector # 功能概述： # 1. 从 PDF 加载内容为 LangChain Document 格式 # 2. 将长内容切割为多段 chunk，提升检索效果 # 3. 使用 TextGen Embedding 将文本转为向量 # 4. 储存至 PGVector（PostgreSQL 向量数据库） from env_config import load_env_config from langchain_community.document_loaders import PyPDFLoader from langchain_postgres import PGVector from local_llm import TextGenEmbeddings # ---------------------------------------- # Step 1: 载入 PDF 并转为 Document # ---------------------------------------- def load_pdf(filepath: str): loader = PyPDFLoader(filepath) docs = loader.load() return docs # ---------------------------------------- # Step 2: 将长文本切割为多段 Chunk Document # ---------------------------------------- def chunk_pdf_docs(docs, chunk_size=200, chunk_overlap=50): from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap) chunks = splitter.split_documents(docs) for chunk in chunks: chunk.metadata["content_type"] = "pdf" return chunks # ---------------------------------------- # Step 3: 建立 PGVector 数据库联机字符串 # ---------------------------------------- def build_pgvector_connection_string(host, port, database, user, password): connection_string = PGVector.connection_string_from_db_params( driver = "psycopg", host = host, port = port, database = database, user = user, password = password, ) return connection_string # ---------------------------------------- # Step 4: 将 Document 索引到 PGVector # ---------------------------------------- def index_docs_to_pgvector(docs, collection_name, connection_string, embeddings): vectorstore = PGVector.from_documents( embedding=embeddings, documents=docs, collection_name=collection_name, connection=connection_string, pre_delete_collection=True # 每次执行会先清空 collection，确保一致性 ) print(f"✅ 成功索引 {len(docs)} 笔数据！") return vectorstore # ---------------------------------------- # 主程序 # ---------------------------------------- def main(): labor_leave_policy_pdf = "./rag_knowledge/labor_leave_policy.pdf" # 数据源 # 加载环境变量 config = load_env_config() # 加载并处理 PDF 文件 docs = load_pdf(labor_leave_policy_pdf) chunks = chunk_pdf_docs(docs, 200, 50) # 初始化嵌入模型 embeddings = TextGenEmbeddings( base_url=config['TEXTGEN_BASE_URL'], api_key=config['TEXTGEN_API_KEY'] ) # 建立数据库联机 connection_string = build_pgvector_connection_string(config['PGVECTOR_HOST'], int(config['PGVECTOR_PORT']), config['PGVECTOR_DATABASE'], config['PGVECTOR_USER'], config['PGVECTOR_PWD'] ) # 将 chunks 索引至指定 collection collection_name = "labor_leave_policy_pdf@20260717" index_docs_to_pgvector(chunks, collection_name, connection_string, embeddings) if __name__ == "__main__": main()

将长文本切割成 chunks 并向量化并储存至 PGVector：

python 1_indexing_longtxt.py

5.6 编写 RAG 问答应用

通过检索相关段落并结合 LLM 进行自然语言回答。使用 Chainlit 作为应用界面：

2_app.py

# ----- Chatbot 使用向量数据库进行 RAG 问答 ----- # 本范例示范如何使用已完成的向量索引，建构具备文件检索能力的 AI 聊天机器人。 # 向量数据源：0_indexing_qapair.py、1_indexing_longtxt.py import sys import asyncio import chainlit as cl from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough from langchain_openai.chat_models import ChatOpenAI from local_llm import TextGenEmbeddings from langchain_postgres import PGVector from langchain.retrievers import EnsembleRetriever from env_config import load_env_config if sys.platform == "win32": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) # ----------------------------- 模型建构函数 ----------------------------- def build_embedding(config): return TextGenEmbeddings( base_url=config['TEXTGEN_BASE_URL'], api_key=config['TEXTGEN_API_KEY'], ) def build_pg_connection_string(config): return PGVector.connection_string_from_db_params( driver = "psycopg", host = config['PGVECTOR_HOST'], port = int(config['PGVECTOR_PORT']), database = config['PGVECTOR_DATABASE'], user = config['PGVECTOR_USER'], password = config['PGVECTOR_PWD'], ) def build_pgvector_retriever(collection_name, connection_string, embeddings, score_threshold=0.5, k=2): from sqlalchemy.ext.asyncio import create_async_engine vectorstore = PGVector( connection=create_async_engine(connection_string), collection_name=collection_name, embeddings=embeddings, ) retriever = vectorstore.as_retriever( search_type="similarity_score_threshold", search_kwargs={"score_threshold": score_threshold, "k": k} ) return retriever def build_llm(config): return ChatOpenAI( openai_api_key=config['TEXTGEN_API_KEY'], #model=config['MODEL_NAME'], base_url=config['TEXTGEN_BASE_URL'], #temperature=0.6, streaming=True, model_kwargs={ #"top_p": 0.95, }, ) # ----------------------------- 工具函数 ----------------------------- def format_docs(docs): results = [] for i, doc in enumerate(docs): ctype = doc.metadata.get("content_type", "") if ctype == "qa": results.append(f"{i+1}. QA 回答：{doc.metadata.get('answer', '')}") elif ctype == "pdf": results.append(f"{i+1}. {doc.page_content}") return "\n\n".join(results) # ----------------------------- Chainlit 回调逻辑 ----------------------------- @cl.on_chat_start async def on_chat_start(): config = load_env_config() embeddings = build_embedding(config) connection_string = build_pg_connection_string(config) # 建立两种数据源的 Retriever：QA + PDF retriever_qa = build_pgvector_retriever("labor_qa_dataset@20260717", connection_string, embeddings) retriever_pdf = build_pgvector_retriever("labor_leave_policy_pdf@20260717", connection_string, embeddings) # 组合多个 Retriever，形成单一入口（会加权后整合查询结果） retriever = EnsembleRetriever( retrievers=[retriever_qa, retriever_pdf], weights=[0.5, 0.5] ) # ----- Chain 组装逻辑 ----- # LLM llm = build_llm(config) # Prompt chat_template = ChatPromptTemplate.from_messages( [ ("system", "你是一个智能客服助理，协助回答员工在公司政策、薪酬、福利、休假、调休、补班、加班上的相关问题"), ("human", "请根据以下提供的内容回答问题" "问题: {question}\n内容: {context}\n答案:") ] ) # RAG Chain：retriever → 格式化 → prompt → llm → output rag_chain = ( { "context": retriever | format_docs, # retriever 根据 question 找出文件，并用 format_docs 合并 "question": RunnablePassthrough() } | chat_template | llm | StrOutputParser() ) cl.user_session.set("rag_chain_with_source", rag_chain) await cl.Message(content="Hi，欢迎来到智能客服！请输入您的问题开始对话。\n\n例如：一年有几天事假？", author="rag").send() @cl.on_message async def on_message(message: cl.Message): rag_chain_with_source = cl.user_session.get("rag_chain_with_source") response = await rag_chain_with_source.ainvoke( message.content, config={ "callbacks": [cl.AsyncLangchainCallbackHandler()], }, ) await cl.Message(response, author="rag").send()

运行：

chainlit 2_app.py --host 0.0.0.0 --port 4200

5.7 编写 RAG 问答应用 with Streaming

基于 5.6 的基础架构进一步强化，加入了「Streaming 回答」的能力，提升用户体验。

3_app.py

# ----- Streaming 版 RAG Chatbot ----- # Streaming 允许模型在生成响应时「逐步」传送内容，提升使用者互动体验， # 适合响应较长或需要实时反馈的应用情境。 import sys import asyncio import chainlit as cl from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough from langchain_openai.chat_models import ChatOpenAI from local_llm import TextGenEmbeddings from langchain_postgres import PGVector from langchain.retrievers import EnsembleRetriever from env_config import load_env_config if sys.platform == "win32": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) def build_embedding(config): return TextGenEmbeddings( base_url=config['TEXTGEN_BASE_URL'], api_key=config['TEXTGEN_API_KEY'], ) def build_pg_connection_string(config): return PGVector.connection_string_from_db_params( driver = "psycopg", host = config['PGVECTOR_HOST'], port = int(config['PGVECTOR_PORT']), database = config['PGVECTOR_DATABASE'], user = config['PGVECTOR_USER'], password = config['PGVECTOR_PWD'], ) def build_pgvector_retriever(collection_name, connection_string, embeddings, score_threshold=0.5, k=2): from sqlalchemy.ext.asyncio import create_async_engine vectorstore = PGVector( connection=create_async_engine(connection_string), collection_name=collection_name, embeddings=embeddings, ) retriever = vectorstore.as_retriever( search_type="similarity_score_threshold", search_kwargs={"score_threshold": score_threshold, "k": k} ) return retriever def build_llm(config): return ChatOpenAI( openai_api_key=config['TEXTGEN_API_KEY'], #model=config['MODEL_NAME'], base_url=config['TEXTGEN_BASE_URL'], #temperature=0.6, streaming=True, model_kwargs={ #"top_p": 0.95, }, ) def format_docs(docs): results = [] for i, doc in enumerate(docs): ctype = doc.metadata.get("content_type", "") if ctype == "qa": results.append(f"{i+1}. QA 回答：{doc.metadata.get('answer', '')}") elif ctype == "pdf": results.append(f"{i+1}. {doc.page_content}") return "\n\n".join(results) @cl.on_chat_start async def on_chat_start(): config = load_env_config() embeddings = build_embedding(config) connection_string = build_pg_connection_string(config) # 建立两种数据源的 Retriever：QA + PDF retriever_qa = build_pgvector_retriever("labor_qa_dataset@20260717", connection_string, embeddings) retriever_pdf = build_pgvector_retriever("labor_leave_policy_pdf@20260717", connection_string, embeddings) retriever = EnsembleRetriever( retrievers=[retriever_qa, retriever_pdf], weights=[0.5, 0.5] ) llm = build_llm(config) chat_template = ChatPromptTemplate.from_messages( [ ("system", "你是一个智能客服助理，协助回答员工在公司政策、薪酬、福利、休假、调休、补班、加班上的相关问题"), ("human", "请根据以下提供的内容回答问题" "问题: {question}\n内容: {context}\n答案:") ] ) rag_chain = ( { "context": retriever | format_docs, "question": RunnablePassthrough() } | chat_template | llm | StrOutputParser() ) cl.user_session.set("rag_chain_with_source", rag_chain) await cl.Message(content="Hi，欢迎来到智能客服！请输入您的问题开始对话。\n\n例如：一年有几天事假？", author="rag").send() @cl.on_message async def on_message(message: cl.Message): rag_chain_with_source = cl.user_session.get("rag_chain_with_source") # 建立空讯息容器以实现 LLM 响应过程中逐字符串流 msg = cl.Message(content="", author="rag") await msg.send() # 使用 astream() 串流响应，逐步产出每一段回复内容 async for chunk in rag_chain_with_source.astream( message.content, config={"callbacks": [cl.AsyncLangchainCallbackHandler()]}, ): await msg.stream_token(chunk) await msg.update()

运行：

chainlit run 3_app.py --host 0.0.0.0 --port 4200

6.总结

本文详细介绍了如何利用 Chainlit、TextGen API 和 LangChain 构建一个完整的本地化 RAG（检索增强生成）应用。整个方案的核心优势在于实现了完全本地化的部署，保障了数据隐私和安全，同时提供了灵活可扩展的架构。

6.1 核心要点回顾

1. 技术栈整合：通过 TextGen WebUI 提供统一的 OpenAI 兼容 API，支持本地运行各种 LLM 模型（如 Qwen、Llama 等），结合 Chainlit 快速构建交互式前端界面。

2. 环境配置：使用 uv 作为 Python 包管理器，在 Windows 环境下快速搭建开发环境，确保依赖管理的效率和一致性。

3. RAG 流程构建：

• 文档处理：支持 Excel QA 对和 PDF 文档的加载与切分
• 向量化存储：使用 TextGen Embeddings 生成向量，并存储到 PostgreSQL 的 PGVector 扩展中
• 检索增强：通过 EnsembleRetriever 结合多种数据源，提升检索质量

4. 应用开发：提供了完整的 Chainlit 应用示例，包括基础版和流式响应版，展示了如何将 RAG 系统与用户界面无缝集成。

6.2 关键实践建议

• 模型选择：根据硬件资源选择合适的 LLM 模型，平衡性能与资源消耗
• 分块策略：根据文档类型调整 chunk_size 和 chunk_overlap 参数，优化检索效果
• 向量化质量：选择合适的嵌入模型（如 all-mpnet-base-v2 或 bge-large-zh-v1.5）提升语义理解能力
• 部署优化：使用 systemd 或 supervisor 管理服务进程，确保系统稳定性

6.3 扩展方向

本方案为构建企业级知识助手提供了坚实基础，未来可以进一步扩展：

• 多模态支持：集成图像、表格等非文本数据的处理能力
• 多轮对话：添加对话历史管理，支持上下文连贯的问答
• 权限控制：实现基于角色的访问控制，满足企业安全需求
• 监控分析：集成 Langfuse 等工具，对问答质量和使用情况进行监控分析

通过本文的实践指南，开发者可以快速搭建一个功能完整的本地 RAG 系统，既保证了数据安全性，又获得了与云端服务相当的用户体验。这套方案特别适合对数据隐私有严格要求的企业环境，同时也为个性化定制提供了充分的空间。