基于开源套件构建企业级RAG系统：从上下文工程到工程化实践

1. 项目概述：从“上下文工程”到“工程化工具箱”

最近在跟几个做AI应用落地的朋友聊天，大家普遍有个痛点：大模型（LLM）的能力很强，但想让它稳定、可靠、低成本地解决一个具体业务问题，中间隔着一条巨大的鸿沟。这条鸿沟，就是“上下文工程”。简单来说，上下文工程就是如何高效、精准地把我们想让模型知道的信息（比如公司内部文档、产品手册、私有数据）和希望它遵循的指令（比如回答格式、安全边界、业务流程）组织起来，并“喂”给模型的过程。这听起来简单，做起来却是个系统工程，涉及到数据预处理、向量化、检索、提示词设计、编排、评估等一系列环节。

正是在这个背景下，我注意到了GitHub上的一个开源项目：NeoLabHQ/context-engineering-kit。这个项目名字起得很直白——“上下文工程套件”。它不是一个单一的库，而是一个工具箱，或者说，是一个“工程化”的脚手架。它的目标很明确：为开发者提供一套开箱即用的、模块化的组件和最佳实践，帮助大家把上下文工程这件事，从一个充满不确定性的“艺术”，变成一个可重复、可度量、可优化的“工程”。

我自己花了一周多的时间，把这个套件从里到外研究了一遍，并且基于它搭建了一个内部知识库问答的原型。整个过程下来，我的感受是：它确实抓住了当前LLM应用开发的核心痛点，提供了一套非常务实的解决方案。它不是要替代LangChain或LlamaIndex这类成熟的框架，而是站在一个更贴近工程落地的视角，把那些框架中抽象的概念，用具体的代码、配置和流程给“固化”了下来。接下来，我就结合自己的实践，把这个套件的核心设计、使用方法和踩过的坑，系统地分享给大家。

2. 核心设计理念与架构拆解

2.1 为什么是“套件”而非“框架”？

在深入代码之前，理解这个项目的定位至关重要。市面上已经有了很多优秀的LLM应用开发框架，比如LangChain，它提供了极其丰富的“链”（Chain）和“智能体”（Agent）抽象，功能强大但学习曲线陡峭；再比如LlamaIndex，专注于数据连接和检索，在RAG（检索增强生成）领域非常专业。

context-engineering-kit的出发点不同。它假设你已经了解了RAG、智能体等基本概念，现在需要快速搭建一个可用的系统，并且这个系统要易于维护、监控和迭代。因此，它更像一个“套件”或“样板间”（Boilerplate）。它提供的是：

1. 预置的、经过验证的流水线：比如一个完整的“文档加载 -> 分块 -> 向量化 -> 检索 -> 生成”的RAG流程，每个环节都有默认的、效果不错的实现。
2. 模块化的可替换组件：你不喜欢默认的文本分块策略？可以很容易地换成另一个。想换一个向量数据库？接口已经定义好了。
3. 工程化的配套设施：包括配置管理（用YAML或环境变量）、日志记录、简单的评估工具，甚至是一些部署的示例。

这种设计理念的好处是“上手即用，深度可定制”。你不需要从零开始组装所有零件，可以直接开车上路；当你想改装引擎时，也有清晰的接口和文档指引。

2.2 项目核心模块全景图

套件的代码结构清晰地反映了它的模块化思想。核心目录通常包括：

• ingestion/(数据摄取)：负责从各种来源（本地文件、网页、数据库）加载数据，并进行清洗、分块。这是上下文工程的“原料处理车间”。
• embedding/(向量化)：封装了文本到向量的转换过程。通常会支持OpenAI的text-embedding-ada-002，以及开源的Sentence Transformers模型，方便你在效果和成本之间做权衡。
• vectordb/(向量数据库)：提供了与主流向量数据库（如Chroma, Pinecone, Weaviate, Qdrant）交互的客户端封装。重点是统一接口，让业务逻辑与具体的数据库解耦。
• retrieval/(检索)：这是RAG的核心。不仅实现了基础的基于向量相似度的检索，通常还会包含“重排序”（Re-ranking）模块。重排序可以用一个更精细的模型（如Cohere的rerank模型或开源的BGE-reranker）对初步检索到的Top K个结果进行二次打分和排序，显著提升召回结果的相关性。
• generation/(生成)：与大语言模型交互的部分。封装了提示词模板、对话历史管理、以及调用不同模型API（如OpenAI GPT, Anthropic Claude, 开源Llama via Ollama）的逻辑。这里会强调“提示词工程”的最佳实践。
• orchestration/(编排)：如果说前面是零件，这里就是组装生产线。它定义了整个应用的执行流程，比如一个问答会话如何串联起检索和生成步骤，如何处理多轮对话的上下文。
• evaluation/(评估)：这是很多个人项目会忽略，但工程化不可或缺的一环。提供了对检索效果（查全率、查准率）和生成效果（相关性、事实准确性、有害性）进行量化评估的工具雏形。
• config/与utils/：提供配置管理和通用工具函数，保证项目的可配置性和代码复用。

注意：以上模块划分是我根据项目常见结构和其理念推断的典型组成。实际项目中，模块名称和划分可能略有不同，但核心思想一致：关注点分离。每个模块职责单一，通过清晰的接口进行通信。

2.3 配置驱动与约定优于配置

这个套件非常强调“配置驱动”。几乎所有的核心选择——用哪个嵌入模型、连接哪个向量数据库、检索时返回几条结果——都可以通过一个中心化的配置文件（比如config.yaml）来管理。

# 示例配置结构 embedding: provider: "openai" # 或 "huggingface" model: "text-embedding-3-small" dimensions: 1536 vectordb: provider: "chroma" path: "./data/chroma_db" collection_name: "company_docs" retrieval: top_k: 5 use_reranker: true reranker_model: "BAAI/bge-reranker-large" generation: llm_provider: "openai" model: "gpt-4-turbo-preview" temperature: 0.1 system_prompt: "你是一个专业的助手，请根据提供的上下文回答问题。"

这种做法的好处显而易见：

• 环境隔离：开发、测试、生产环境可以使用不同的配置（如API密钥、模型版本）。
• 非开发者友好：产品经理或运维人员可以在不碰代码的情况下调整一些关键参数。
• 实验可复现：每一次实验的配置都可以保存下来，确保结果可追溯。

“约定优于配置”体现在它提供了一套默认的、合理的配置。你不需要在项目启动时就纠结于每一个参数，可以直接使用默认值快速跑通流程，然后再针对性地调优。

3. 从零搭建一个知识库问答系统：实操全流程

理论说了这么多，我们来点实际的。假设我们要为一个产品团队搭建一个内部技术文档问答机器人。下面我就以context-engineering-kit的思维模式，一步步实现。

3.1 环境准备与项目初始化

首先，我们需要一个干净的环境。强烈建议使用虚拟环境。

# 创建并激活虚拟环境 python -m venv venv_context_kit source venv_context_kit/bin/activate # Linux/Mac # venv_context_kit\Scripts\activate # Windows # 克隆项目（假设我们以这个套件为模板） git clone https://github.com/NeoLabHQ/context-engineering-kit.git cd context-engineering-kit # 安装核心依赖 pip install -r requirements.txt # 通常包括：langchain, openai, chromadb, sentence-transformers, pydantic, pyyaml等

接下来，设置关键的环境变量，特别是API密钥。我习惯用一个.env文件来管理。

# .env 文件 OPENAI_API_KEY=sk-你的密钥 # 如果使用其他服务，如 Pinecone, Cohere # PINECONE_API_KEY=... # COHERE_API_KEY=...

在代码中，使用python-dotenv加载它们。

# config/settings.py from pydantic_settings import BaseSettings from dotenv import load_dotenv load_dotenv() class Settings(BaseSettings): openai_api_key: str # ... 其他配置字段 class Config: env_file = ".env" settings = Settings()

实操心得：不要把API密钥硬编码在代码或配置文件里提交到Git！.env文件必须加入.gitignore。团队协作时，可以使用.env.example文件列出需要的环境变量，供他人参考。

3.2 数据摄取与向量化：构建知识库的基石

我们的数据源是一堆Markdown格式的产品需求文档（PRD）和API设计文档。这一步的目标是把它们变成向量数据库里的一条条记录。

步骤一：文档加载与解析套件一般会提供一个统一的文档加载器。我们需要处理Markdown，保持其层级结构（标题、列表）对于理解文档很重要。

# ingestion/document_loader.py from langchain.document_loaders import DirectoryLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter def load_and_split_documents(data_dir: str): """加载指定目录下的所有Markdown文件并进行智能分块""" loader = DirectoryLoader( data_dir, glob="**/*.md", loader_cls=UnstructuredMarkdownLoader, show_progress=True ) raw_documents = loader.load() print(f"成功加载 {len(raw_documents)} 个文档") # 关键：分块策略 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 每个块的最大字符数 chunk_overlap=200, # 块之间的重叠字符，避免上下文断裂 separators=["\n\n", "\n", "。", "！", "？", " ", ""] # 中文友好的分隔符 ) split_docs = text_splitter.split_documents(raw_documents) print(f"分块后得到 {len(split_docs)} 个文本块") return split_docs

为什么是1000和200？

• chunk_size=1000：这是一个经验值。太小（如200）会丢失上下文，导致检索到的片段信息不完整；太大（如2000）可能包含无关信息，稀释核心内容，且增加模型处理负担。对于技术文档，1000左右能较好平衡。
• chunk_overlap=200：重叠是为了防止一个完整的句子或概念被硬生生切在两块中间。例如，一个问题的描述在一块，答案在下一块的开头，没有重叠就可能丢失关键联系。

步骤二：文本向量化（Embedding）这是将文本转化为数学表示（向量）的过程，相似的文本会有相似的向量。

# embedding/embedding_client.py from langchain.embeddings import OpenAIEmbeddings from langchain.embeddings import HuggingFaceEmbeddings import numpy as np class EmbeddingClient: def __init__(self, provider="openai", model_name=None): self.provider = provider if provider == "openai": # 使用OpenAI的嵌入模型，效果好，但需付费且有延迟 self.client = OpenAIEmbeddings( model="text-embedding-3-small", dimensions=1536 # 可以指定维度，小型号性价比高 ) elif provider == "huggingface": # 使用开源模型，免费，可本地部署，适合数据敏感或成本敏感场景 model_kwargs = {'device': 'cpu'} # 或 'cuda' encode_kwargs = {'normalize_embeddings': True} self.client = HuggingFaceEmbeddings( model_name=model_name or "BAAI/bge-small-zh-v1.5", # 优秀的中文模型 model_kwargs=model_kwargs, encode_kwargs=encode_kwargs ) else: raise ValueError(f"不支持的Embedding提供商: {provider}") def embed_documents(self, texts): return self.client.embed_documents(texts) def embed_query(self, text): return self.client.embed_query(text)

模型选择考量：

• OpenAItext-embedding-3-***：业界标杆，效果稳定，API调用简单，但会产生持续费用，且数据需出境。
• HuggingFace 开源模型（如BGE、M3E）：免费，可私有化部署，数据安全。BAAI/bge-*系列对中文支持很好，是中文项目的首选。需要一定的本地计算资源（GPU更佳）。

踩坑记录：初期我直接用了OpenAI的text-embedding-ada-002，在构建上万条记录的索引时，API调用费用和耗时成了瓶颈。后来切换到BAAI/bge-small-zh-v1.5本地运行，虽然初始化慢一点，但后续零成本，且对于中文技术文档，效果几乎没有损失。建议：先用小规模数据测试开源模型效果，如果可接受，优先考虑开源方案以控制长期成本。

3.3 向量数据库的选型与数据灌入

有了文本块和它们的向量，我们需要一个地方存储并能够快速检索它们。这就是向量数据库。

步骤一：选择与初始化客户端套件通常会抽象一个统一的向量数据库客户端。这里以轻量级的ChromaDB（本地运行）为例。

# vectordb/chroma_client.py import chromadb from chromadb.config import Settings from langchain.vectorstores import Chroma class ChromaClient: def __init__(self, persist_directory: str, collection_name: str, embedding_client): # 创建持久化客户端 self.client = chromadb.PersistentClient( path=persist_directory, settings=Settings(anonymized_telemetry=False) # 关闭遥测 ) self.collection_name = collection_name self.embedding_client = embedding_client self.langchain_chroma = None def create_collection(self): """创建或获取集合""" try: collection = self.client.get_collection(name=self.collection_name) print(f"集合 '{self.collection_name}' 已存在，直接使用。") except: collection = self.client.create_collection(name=self.collection_name) print(f"创建新集合: '{self.collection_name}'") return collection def add_documents(self, documents): """将文档添加到向量库""" texts = [doc.page_content for doc in documents] metadatas = [doc.metadata for doc in documents] # 使用LangChain的集成方式，它会自动处理向量化和存储 self.langchain_chroma = Chroma.from_documents( documents=documents, embedding=self.embedding_client.client, persist_directory=self.client._settings.persist_directory, collection_name=self.collection_name ) print(f"已添加 {len(documents)} 个文档到向量数据库。")

步骤二：执行数据灌入流水线现在，我们把前两步串联起来。

# main_ingestion.py from config.settings import settings from ingestion.document_loader import load_and_split_documents from embedding.embedding_client import EmbeddingClient from vectordb.chroma_client import ChromaClient def main(): # 1. 加载并分块文档 data_dir = "./data/company_docs" all_splits = load_and_split_documents(data_dir) # 2. 初始化Embedding客户端 (选择开源模型) embed_client = EmbeddingClient(provider="huggingface", model_name="BAAI/bge-small-zh-v1.5") # 3. 初始化向量数据库客户端 persist_dir = "./data/vector_db" chroma_client = ChromaClient(persist_directory=persist_dir, collection_name="product_docs", embedding_client=embed_client) chroma_client.create_collection() # 4. 灌入数据 chroma_client.add_documents(all_splits) print("知识库构建完成！") if __name__ == "__main__": main()

运行这个脚本，你的本地知识库就建好了。./data/vector_db目录下会保存所有向量数据和元数据。

向量数据库选型对比：

实操心得：对于内部工具或实验性项目，从Chroma开始绝对是最佳选择。它没有外部依赖，一个Python包搞定，让你完全专注于业务逻辑验证。等到数据量和并发请求上来后，再平滑迁移到Qdrant或Pinecone也不迟。套件的价值就在于，通过抽象接口，让这种迁移可能只需要改几行配置。

4. 检索与生成：构建问答引擎

知识库准备好了，接下来是打造问答引擎的核心：检索器（Retriever）和生成器（Generator）。

4.1 实现混合检索与重排序

单纯的向量相似度检索（语义搜索）有时会漏掉一些关键词完全匹配的重要文档。因此，工业级系统通常会采用“混合检索”。

# retrieval/hybrid_retriever.py from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain.vectorstores import Chroma from typing import List, Dict, Any class HybridRetriever: def __init__(self, vectorstore, text_split_docs, k=10): """ Args: vectorstore: 已初始化的向量库对象（如Chroma实例） text_split_docs: 用于构建BM25索引的原始文档分块列表 k: 每种检索器返回的初始结果数量 """ self.k = k # 1. 语义检索器 (向量搜索) self.vector_retriever = vectorstore.as_retriever( search_kwargs={"k": self.k} ) # 2. 关键词检索器 (BM25) self.bm25_retriever = BM25Retriever.from_documents( text_split_docs ) self.bm25_retriever.k = self.k # 3. 集成检索器 self.ensemble_retriever = EnsembleRetriever( retrievers=[self.bm25_retriever, self.vector_retriever], weights=[0.3, 0.7] # 调整权重：更依赖语义搜索 ) def retrieve(self, query: str) -> List[Dict[str, Any]]: """执行混合检索""" docs = self.ensemble_retriever.get_relevant_documents(query) # 去重（因为两个检索器可能返回相同文档） seen_ids = set() unique_docs = [] for doc in docs: # 用一个唯一标识，比如元数据中的ID或内容哈希 doc_id = doc.metadata.get("source", "") + ":" + str(doc.metadata.get("chunk_index", "")) if doc_id not in seen_ids: seen_ids.add(doc_id) unique_docs.append({ "content": doc.page_content, "metadata": doc.metadata, "score": 0.0 # 集成检索器可能不返回分数，需要后续处理 }) return unique_docs[:self.k] # 返回Top K个唯一结果

重排序（Re-ranking）：混合检索返回了K个结果，但它们的顺序可能还不是最优的。重排序使用一个更精细的、专门做“相关性打分”的模型，对这K个结果进行二次排序。

# retrieval/reranker.py # 假设使用Cohere的在线重排序API（效果很好，但需付费） import cohere from config.settings import settings class Reranker: def __init__(self): self.client = cohere.Client(settings.cohere_api_key) # 需要COHERE_API_KEY def rerank(self, query: str, documents: List[Dict], top_n: int = 5): """使用Cohere API对文档进行重排序""" doc_texts = [doc["content"] for doc in documents] try: response = self.client.rerank( model="rerank-english-v2.0", # 或 "rerank-multilingual-v2.0" query=query, documents=doc_texts, top_n=top_n, ) reranked_indices = [result.index for result in response.results] reranked_docs = [documents[i] for i in reranked_indices] # 更新分数 for i, result in enumerate(response.results): reranked_docs[i]["relevance_score"] = result.relevance_score return reranked_docs except Exception as e: print(f"重排序API调用失败: {e}，返回原始顺序。") return documents[:top_n]

注意：重排序虽然能大幅提升效果，但增加了延迟和成本（如果使用云服务）。建议策略：在初步检索返回较多结果（如K=20）时，使用重排序筛选出最相关的少数几个（如top_n=5）送给LLM，性价比最高。对于内部工具，如果对延迟不敏感，强烈建议加上这一步。

4.2 提示词工程与生成器封装

检索到了最相关的上下文，现在需要把它们和用户的问题一起，巧妙地“喂”给大模型。这就是提示词工程。

# generation/prompt_templates.py from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate def build_qa_prompt(contexts: List[str], query: str, chat_history: List[tuple] = None) -> ChatPromptTemplate: """ 构建一个带有上下文和聊天历史的问答提示词。 """ system_template = """你是一个专业的产品技术助手，负责回答关于公司产品和内部技术的问题。 请严格根据以下提供的上下文信息来回答问题。如果上下文信息不足以回答问题，请直接说“根据现有资料，我无法回答这个问题”，不要编造信息。 上下文信息： {context} 当前对话历史： {history} """ human_template = "用户问题：{question}" # 格式化上下文 context_str = "\n\n---\n\n".join([f"[片段{i+1}]: {ctx}" for i, ctx in enumerate(contexts)]) # 格式化历史 history_str = "" if chat_history: for human, ai in chat_history[-3:]: # 只保留最近3轮对话 history_str += f"用户: {human}\n助手: {ai}\n" system_message_prompt = SystemMessagePromptTemplate.from_template(system_template) human_message_prompt = HumanMessagePromptTemplate.from_template(human_template) chat_prompt = ChatPromptTemplate.from_messages( [system_message_prompt, human_message_prompt] ) # 返回格式化后的消息列表，方便直接调用LLM return chat_prompt.format_prompt( context=context_str, history=history_str, question=query ).to_messages()

提示词设计要点：

1. 角色设定：明确告诉模型“你是谁”，引导其输出风格。
2. 指令清晰：“严格根据上下文”是关键，这是RAG避免幻觉的核心。
3. 上下文格式化：用清晰的分隔符（如---）和编号标明不同片段，帮助模型区分。
4. 处理未知：明确指示模型在上下文不足时如何回应，避免胡编乱造。
5. 管理历史：只保留最近几轮对话，避免上下文窗口被占满，同时保持对话连贯性。

接下来是生成器，它负责调用LLM API。

# generation/llm_client.py from langchain.chat_models import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage, AIMessage import logging class LLMClient: def __init__(self, model_name="gpt-4-turbo-preview", temperature=0.1, max_tokens=2000): # 使用LangChain的封装，也可以直接调用OpenAI SDK self.llm = ChatOpenAI( model_name=model_name, temperature=temperature, # 低温度保证输出稳定、事实性强 max_tokens=max_tokens, request_timeout=60 ) self.logger = logging.getLogger(__name__) def generate(self, messages): """发送消息列表并获取回复""" try: response = self.llm(messages) return response.content except Exception as e: self.logger.error(f"LLM调用失败: {e}") return "抱歉，处理您的请求时出现了问题，请稍后再试。"

4.3 编排层：将所有组件串联起来

最后，我们需要一个“大脑”来协调检索、重排序、提示词构建和生成。

# orchestration/qa_orchestrator.py from retrieval.hybrid_retriever import HybridRetriever from retrieval.reranker import Reranker from generation.prompt_templates import build_qa_prompt from generation.llm_client import LLMClient class QAOrchestrator: def __init__(self, retriever: HybridRetriever, llm_client: LLMClient, use_reranker=True): self.retriever = retriever self.llm_client = llm_client self.use_reranker = use_reranker if use_reranker: self.reranker = Reranker() self.conversation_history = [] # 简单的内存历史记录 def answer_question(self, user_query: str): """处理用户问答的核心流程""" # 1. 检索 retrieved_docs = self.retriever.retrieve(user_query) if not retrieved_docs: return "未找到相关文档，无法回答此问题。" # 2. (可选) 重排序 if self.use_reranker and len(retrieved_docs) > 3: retrieved_docs = self.reranker.rerank(user_query, retrieved_docs, top_n=5) # 3. 构建提示词 contexts = [doc["content"] for doc in retrieved_docs[:5]] # 取前5个上下文 messages = build_qa_prompt(contexts, user_query, self.conversation_history) # 4. 调用LLM生成答案 answer = self.llm_client.generate(messages) # 5. 更新对话历史 self.conversation_history.append((user_query, answer)) # 限制历史长度，防止无限增长 if len(self.conversation_history) > 10: self.conversation_history.pop(0) # 6. (可选) 可以在这里添加引用来源 answer_with_sources = f"{answer}\n\n---\n**参考来源**:\n" for i, doc in enumerate(retrieved_docs[:3]): # 引用前3个来源 source = doc["metadata"].get("source", "未知文档") answer_with_sources += f"{i+1}. {source}\n" return answer_with_sources

现在，一个完整的、具备混合检索、重排序、多轮对话能力的问答引擎就搭建完成了。你可以通过一个简单的循环或Web接口来使用它。

# main.py from vectordb.chroma_client import ChromaClient from embedding.embedding_client import EmbeddingClient from retrieval.hybrid_retriever import HybridRetriever from generation.llm_client import LLMClient from orchestration.qa_orchestrator import QAOrchestrator # ... 初始化各组件 ... def main(): # 初始化所有组件（略，参考前面代码） embed_client = EmbeddingClient(provider="huggingface") chroma_client = ChromaClient(...) vectorstore = chroma_client.get_langchain_vectorstore() # 假设有这个方法 retriever = HybridRetriever(vectorstore, original_docs) llm_client = LLMClient(model_name="gpt-3.5-turbo") # 先用便宜的模型测试 orchestrator = QAOrchestrator(retriever, llm_client, use_reranker=True) print("知识库问答机器人已启动，输入'退出'结束。") while True: query = input("\n请输入您的问题: ") if query.lower() in ['退出', 'exit', 'quit']: break answer = orchestrator.answer_question(query) print(f"\n助手: {answer}") if __name__ == "__main__": main()

5. 评估、监控与迭代：工程化的闭环

一个可用的系统建成了，但一个可靠的系统需要评估和迭代。这是“工程化”与“脚本”的本质区别。

5.1 如何评估RAG系统的效果？

你不能只靠手动问几个问题来判断好坏。需要建立量化的评估体系。套件通常会提供一个评估模块的雏形。

核心评估指标：

1. 检索质量：
   • 命中率（Hit Rate）：对于一组有标准答案的问题，检索到的Top K个文档中包含正确答案的比率。
   • 平均精度均值（Mean Average Precision, MAP）：衡量检索结果排序好坏的指标。
2. 生成质量：
   • 事实一致性（Faithfulness）：模型生成的答案是否严格基于提供的上下文，有没有“幻觉”。
   • 答案相关性（Answer Relevance）：生成的答案是否直接回答了问题。
   • 有用性（Helpfulness）：人工评分，答案是否有用。

简易评估脚本示例：你可以准备一个测试集eval_qa_pairs.jsonl，每行是一个{"question": "...", "ground_truth_answer": "...", "ground_truth_docs": ["doc_id1", ...]}。

# evaluation/evaluator.py import json from typing import List, Dict class SimpleRAGEvaluator: def __init__(self, orchestrator): self.orchestrator = orchestrator def evaluate_retrieval(self, test_data: List[Dict], top_k=5): """评估检索效果""" total_hits = 0 for item in test_data: question = item["question"] ground_truth_docs = set(item["ground_truth_docs"]) # 调用检索器（不经过重排序和生成） retrieved_docs = self.orchestrator.retriever.retrieve(question)[:top_k] retrieved_ids = {doc["metadata"].get("doc_id") for doc in retrieved_docs} # 检查是否有命中 if ground_truth_docs & retrieved_ids: total_hits += 1 hit_rate = total_hits / len(test_data) return {"hit_rate@5": hit_rate} def evaluate_generation(self, test_data: List[Dict]): """评估生成效果（简易版，可用LLM作为裁判）""" results = [] for item in test_data: question = item["question"] ground_truth = item["ground_truth_answer"] # 获取系统答案 system_answer = self.orchestrator.answer_question(question) # 这里可以调用另一个LLM（如GPT-4）来对比系统答案和标准答案的相似度/一致性 # 或者进行人工评估 results.append({ "question": question, "ground_truth": ground_truth, "system_answer": system_answer, # "score": llm_judge(question, ground_truth, system_answer) }) return results

定期运行评估脚本，记录指标变化，是判断优化措施（如调整分块大小、换Embedding模型、加重排序）是否有效的唯一科学方法。

5.2 监控与日志

在生产环境中，你需要知道系统运行状况。

• 日志记录：记录每一次问答的查询、检索到的文档ID、生成的答案、耗时、Token使用量。这有助于分析用户真实需求、发现检索盲点、核算成本。
• 异常监控：监控API调用失败、超时、高延迟等情况。
• 反馈循环：提供一个“ thumbs up/down”的反馈按钮，收集用户对答案质量的评价，这些数据是优化模型和检索器的重要依据。

5.3 持续迭代路线图

基于评估和监控，你可以系统地优化系统：

1. 优化检索：
   • 分块策略调优：尝试不同的chunk_size和chunk_overlap，甚至尝试按语义分块（如用LLM划分）。
   • 检索器升级：尝试不同的Embedding模型，加入HyDE（假设性文档嵌入）等技术。
   • 元数据过滤：为文档块添加更多元数据（如文档类型、创建日期、部门），在检索时进行过滤，提升精度。
2. 优化生成：
   • 提示词迭代：根据bad case调整提示词，比如让模型更严格地引用来源。
   • 模型升级：从GPT-3.5-Turbo升级到GPT-4-Turbo，或微调开源模型。
   • 后处理：对模型输出进行格式化、敏感信息过滤等。
3. 优化系统：
   • 缓存：对常见问题的检索结果和生成答案进行缓存，大幅降低成本和延迟。
   • 异步处理：对于耗时的重排序或复杂生成，采用异步任务。
   • 部署优化：将向量数据库、Embedding模型、Reranker模型等服务化，提高整体吞吐量。

6. 常见问题与排查技巧实录

在实际搭建和运行过程中，你一定会遇到各种问题。以下是我总结的一些典型问题及解决方案。

6.1 检索相关问题

问题1：检索结果不相关，总是答非所问。

• 可能原因A：Embedding模型不匹配。你用英文模型处理了中文文档。
  • 排查：检查Embedding模型名称。对于中文，BAAI/bge-*或moka-ai/m3e-*是更好的选择。
  • 解决：更换为合适的中文Embedding模型，并重新构建向量库。
• 可能原因B：分块大小不合适。块太大包含太多噪声，块太小丢失上下文。
  • 排查：查看检索到的文档块内容，是否完整包含问题相关的信息。
  • 解决：调整chunk_size（尝试500, 800, 1000, 1500）和chunk_overlap（建议为chunk_size的10%-20%）。对于技术文档，可以尝试按章节分块。
• 可能原因C：查询本身太短或模糊。
  • 排查：查看用户原始查询。
  • 解决：实现“查询扩展”（Query Expansion）。例如，使用LLM将简短的用户问题重写或扩展成一个更详细的、包含同义词的查询，再进行检索。

问题2：检索速度慢，尤其是第一次查询。

• 可能原因A：Embedding模型首次加载慢。特别是HuggingFace的大模型。
  • 解决：在服务启动时预加载模型，而不是每次查询时加载。或者使用更轻量的模型。
• 可能原因B：向量数据库未使用索引或数据量大。
  • 排查：Chroma默认会创建索引。如果数据量超过10万，考虑使用Pinecone或Qdrant等支持高性能索引的数据库。
  • 解决：确保在创建集合时正确配置了索引（如HNSW）。对于Chroma，确保persist_directory在SSD上。

6.2 生成相关问题

问题3：模型回答出现“幻觉”，编造不存在的信息。

• 可能原因A：提示词指令不够强硬。
  • 解决：在系统提示词中反复强调“严格根据上下文”，并明确说明“如果上下文没有，就说不知道”。可以增加惩罚性描述，如“任何编造信息都是不可接受的”。
• 可能原因B：检索到的上下文质量差或数量不足。
  • 排查：检查传给模型的上下文内容，是否真的包含了答案。
  • 解决：提升检索质量（见问题1）。或者，在提示词中让模型先判断上下文是否足够回答，如果不够，直接回复无法回答，而不是强行生成。
• 可能原因C：模型温度（Temperature）设置过高。
  • 解决：将temperature参数调低，如设为0.1或0，让模型输出更确定、更保守。

问题4：多轮对话中，模型忘记之前的对话或混淆上下文。

• 可能原因：对话历史管理不当。
  • 排查：检查传递给模型的chat_history字符串，是否正确包含了历史问答对。
  • 解决：
    1. 确保历史被传入：在build_qa_prompt函数中正确拼接历史。
    2. 控制历史长度：只保留最近N轮（如3-5轮），避免超过模型的上下文窗口。
    3. 为历史添加上下文：在每一轮历史问答前，可以附加一个简短的摘要或关键词，帮助模型理解。

6.3 系统与部署问题

问题5：API调用费用飙升。

• 可能原因A：每次问答都重新计算查询的Embedding。
  • 解决：对查询Embedding进行缓存。相同的查询，直接使用缓存的结果。
• 可能原因B：检索返回的上下文块太多、太大。
  • 解决：优化检索，返回更精准、更少的上下文块（如从10个减到5个）。同时，在重排序后只选择最相关的2-3个块送给LLM。LLM的Token费用是大头。
• 可能原因C：使用了昂贵的模型（如GPT-4）处理简单问题。
  • 解决：实现路由策略。例如，先用一个简单的规则或小模型判断问题复杂度，简单问题用GPT-3.5-Turbo，复杂问题再用GPT-4。

问题6：服务响应延迟高。

• 优化点：
  1. 并行化：Embedding查询、向量检索、重排序（如果使用）可以并行执行。
  2. 异步化：将整个问答流程设计为异步，使用像FastAPI这样的异步框架。
  3. 硬件加速：如果使用本地Embedding/重排序模型，确保使用GPU（CUDA）。
  4. 数据库优化：向量数据库使用SSD，并优化索引参数。

问题7：如何管理不同版本的知识库？

• 场景：文档更新了，需要更新向量库，但又不能影响线上服务。
• 解决策略：
  • 版本化集合：每次全量更新时，创建一个新的向量集合（如docs_v2），更新完成后，将应用指向新集合。旧集合保留一段时间供回滚。
  • 增量更新：如果套件或底层向量库支持，只对新文档或修改的文档进行增量Embedding和插入。同时，需要考虑旧文档的删除或失效标记，这通常更复杂。

搭建一个健壮的上下文工程系统，就像搭积木，NeoLabHQ/context-engineering-kit这样的套件提供了预先打磨好的积木块和搭建说明书。它能让你快速起步，但真正让它发挥价值的，是你对自身业务数据的理解，以及基于评估数据进行的持续迭代和调优。从简单的ChromaDB和开源Embedding开始，逐步加入重排序、混合检索、复杂的提示词和评估体系，这个过程中积累的经验，远比单纯调用一个API来得宝贵。