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扩展支援向量。uvAn extremely fast Python package and project manager, written in Rust. uv repoChainlit用于构建 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 | iex2.2.2 创建 Chainlit 项目Get started with Overview - Chainlit2.2.3 创建 uv 虚拟环境进行项目文件夹执行以下命令指定 Python 版本uv venv --python 3.122.2.4 启动 uv 虚拟环境.\.venv\Scripts\activate如果启动出错查看执行权限Get-ExecutionPolicy如果返回 Restricted设定 ExecutionPolicySet-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned2.2.5 安装依赖包 (uv sync)成功启动虚拟环境后使用uv sync自动读取pyproject.toml中的依赖并安装。uv syncpyproject.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 [ chainlit2.6.2, fastapi, langchain0.3.27, langchain-openai0.3.28, azure-identity, pandas, langchain-community, langchain-postgres, langchain-core, psycopg[binary], openpyxl, pypdf, langfuse2.60.5, ipykernel7.1.0, grandalf0.8, python-dotenv, uvicorn, ]2.2.6 使用 uv pip 查看已安装的相关 Python 包uv pip list3. 搭建本地 LLM 服务TextGen首先我们需要一个本地运行的 LLM 服务。这里以TextGen为例它支持多种模型格式并提供了兼容 OpenAI 的 API 接口。3.1 启动 TextGen下载发布版本并安装依赖TextGen 安装教程TextGen for Local LLM-CSDN博客。下载你需要的 LLM 模型文件如 Qwen3.5-4B-Q4_K_M.gguf。下载你需要的 Embedding 模型文件 Embedding model 安装教程TextGen 启用 embedding 并安装 all-mpnet-base-v2 模型教程-CSDN博客。启动 WebUI 服务并启用 API 模式export OPENEDAI_EMBEDDING_MODEL/home/ai/textgen-4.9/user_data/models/embedding/all-mpnet-base-v2 export OPENEDAI_EMBEDDING_DEVICEauto # 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, jsondata, headersheaders) print(response.json())4. 构建 RAG 核心流程接下来我们使用 LangChain 构建一个标准的 RAG 流程包括文档加载、文本切分、向量化存储和检索。范例目标将整理好的 QA 数据索引到 PostgreSQL 数据库使用 PGVector 进行向量储存功能概述1. 从 Excel 档案读取 QA 数据2. 将数据转为 LangChain 的 Document 格式包含问题和元数据3. 使用 TextGen Embeddings 将问题转为向量4. 将向量和元数据储存到 PostgreSQL 的 PGVector4.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(indexFalse)) # 使用 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_columnquestion) 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 {i1}) 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 docs4.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(fConnection 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( embeddingembeddings, documentsdocs, collection_namecollection_name, connectionconnection_string, pre_delete_collectionTrue ) print(f✅ 成功索引 {len(docs)} 笔数据) print( * 80 \n) return vectorstore5. 集成 Chainlit 与 TextGen APIChainlit 的核心是定义一个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.pyPGVector / DB 连接资讯 和 TextGen local LLM API URL、API KEYenv_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 实现 Embeddinglocal_llm.pyimport 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: fBearer {self.api_key}, Content-Type: application/json, }, jsonpayload, timeoutself.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.pdflabor_qa_dataset.xlsx 包含的栏位如下labor_qa_datasetquestioncategoryurlcontent_typeanswer问题例如劳工工作年资满一年有5天特别休假这5天是包含满半年的2天吗类别工时(休息、休假、请假)规章制度的出度URLqa回答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(indexFalse)) # 使用 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_columnquestion) 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 {i1}) 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(fConnection 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( embeddingembeddings, documentsdocs, collection_namecollection_name, connectionconnection_string, pre_delete_collectionTrue ) 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_urlconfig[TEXTGEN_BASE_URL], api_keyconfig[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_dataset20260717 index_docs_to_pgvector(docs, collection_name, connection_string, embeddings) if __name__ __main__: main()将 QA Pair 向量化并储存至 PGVectorpython 0_indexing_qapair.py5.5.1 PDF 文档内容转换为向量并存储1_indexing_longtxt.py# ----- 1.Indexing Long Text ----- # 范例目标 # 将 PDF 文件的长文本转为嵌入向量并索引至 PGVector # 功能概述 # 1. 从 PDF 加载内容为 LangChain Document 格式 # 2. 将长内容切割为多段 chunk提升检索效果 # 3. 使用 TextGen Embedding 将文本转为向量 # 4. 储存至 PGVectorPostgreSQL 向量数据库 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_size200, chunk_overlap50): from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter(chunk_sizechunk_size, chunk_overlapchunk_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( embeddingembeddings, documentsdocs, collection_namecollection_name, connectionconnection_string, pre_delete_collectionTrue # 每次执行会先清空 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_urlconfig[TEXTGEN_BASE_URL], api_keyconfig[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_pdf20260717 index_docs_to_pgvector(chunks, collection_name, connection_string, embeddings) if __name__ __main__: main()将长文本切割成 chunks 并向量化并储存至 PGVectorpython 1_indexing_longtxt.py5.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_urlconfig[TEXTGEN_BASE_URL], api_keyconfig[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_threshold0.5, k2): from sqlalchemy.ext.asyncio import create_async_engine vectorstore PGVector( connectioncreate_async_engine(connection_string), collection_namecollection_name, embeddingsembeddings, ) retriever vectorstore.as_retriever( search_typesimilarity_score_threshold, search_kwargs{score_threshold: score_threshold, k: k} ) return retriever def build_llm(config): return ChatOpenAI( openai_api_keyconfig[TEXTGEN_API_KEY], #modelconfig[MODEL_NAME], base_urlconfig[TEXTGEN_BASE_URL], #temperature0.6, streamingTrue, 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{i1}. QA 回答{doc.metadata.get(answer, )}) elif ctype pdf: results.append(f{i1}. {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) # 建立两种数据源的 RetrieverQA PDF retriever_qa build_pgvector_retriever(labor_qa_dataset20260717, connection_string, embeddings) retriever_pdf build_pgvector_retriever(labor_leave_policy_pdf20260717, 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 Chainretriever → 格式化 → 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(contentHi欢迎来到智能客服请输入您的问题开始对话。\n\n例如一年有几天事假, authorrag).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, authorrag).send()运行chainlit 2_app.py --host 0.0.0.0 --port 42005.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_urlconfig[TEXTGEN_BASE_URL], api_keyconfig[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_threshold0.5, k2): from sqlalchemy.ext.asyncio import create_async_engine vectorstore PGVector( connectioncreate_async_engine(connection_string), collection_namecollection_name, embeddingsembeddings, ) retriever vectorstore.as_retriever( search_typesimilarity_score_threshold, search_kwargs{score_threshold: score_threshold, k: k} ) return retriever def build_llm(config): return ChatOpenAI( openai_api_keyconfig[TEXTGEN_API_KEY], #modelconfig[MODEL_NAME], base_urlconfig[TEXTGEN_BASE_URL], #temperature0.6, streamingTrue, 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{i1}. QA 回答{doc.metadata.get(answer, )}) elif ctype pdf: results.append(f{i1}. {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) # 建立两种数据源的 RetrieverQA PDF retriever_qa build_pgvector_retriever(labor_qa_dataset20260717, connection_string, embeddings) retriever_pdf build_pgvector_retriever(labor_leave_policy_pdf20260717, 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(contentHi欢迎来到智能客服请输入您的问题开始对话。\n\n例如一年有几天事假, authorrag).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, authorrag) 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 42006.总结本文详细介绍了如何利用 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 系统既保证了数据安全性又获得了与云端服务相当的用户体验。这套方案特别适合对数据隐私有严格要求的企业环境同时也为个性化定制提供了充分的空间。
