1. 项目概述当RAG遇上SKILL智能体如何“精准”思考最近在折腾一个挺有意思的项目核心就一句话让一个AI智能体在面对用户五花八门的问题时能像一位经验丰富的专家一样快速、精准地从它自己的“专属知识库”里找到最相关的信息来回答。听起来是不是有点像现在大火的RAG没错但又不完全是。我们这次玩得更深入一点把RAG检索增强生成和SKILL技能架构给“焊”在了一起。这个项目的标题叫“轻量级RAG与SKILL架构深度融合专属知识库驱动智能体精准知识匹配应用实践”名字有点长但拆开来看就是三个核心轻量级RAG、SKILL架构、以及它们结合后驱动的精准知识匹配。为什么要把这两者结合这是我踩过不少坑后的深刻体会。纯RAG系统就像一个记忆力超群但不太会变通的“书呆子”。你问它问题它能从海量文档里找到相关段落然后原封不动或者稍作改写地吐给你。但现实世界的问题往往更复杂用户可能问得模糊可能需要多步推理可能需要结合不同来源的知识进行判断。这时候单纯的“检索-生成”就显得力不从心了。而SKILL架构恰恰是让智能体“会变通”的关键。它把复杂任务拆解成一个个可执行、可组合的“技能”比如“解析用户意图”、“查询知识库”、“验证信息”、“格式化回答”等。当RAG的检索能力被封装成一个或多个SKILL智能体就能更智能地决定什么时候该去查资料查哪些资料查到资料后该怎么用这个实践的目标就是构建一个不依赖庞大算力、能快速部署、且真正“懂行”的智能体。它特别适合那些有垂直领域知识库的场景比如企业内部的技术支持文档库、某个专业领域的法规库、甚至是个人精心整理的笔记系统比如用Obsidian搭建的Wiki。接下来我就把自己从架构设计、工具选型到代码实现的完整过程以及过程中那些“血泪教训”和“意外惊喜”毫无保留地分享出来。2. 核心架构设计轻量、解耦与精准的三角平衡设计这个系统的起点是明确三个核心原则轻量、解耦、精准。轻量意味着我们不能一上来就搞复杂的微服务集群得让它在单机甚至资源受限的环境下也能跑起来解耦是为了让RAG模块和SKILL模块能独立进化互不干扰精准则是最终目标一切设计都要服务于提升答案的相关性和准确性。2.1 为什么是“轻量级”RAG市面上成熟的RAG框架很多像LangChain、LlamaIndex功能强大但生态复杂有时候给人一种“杀鸡用牛刀”的感觉。对于专属知识库场景尤其是初期验证阶段我们更需要一个聚焦核心流程、依赖少、调试透明的方案。我的选择是自研一个轻量级RAG管道。它的核心流程只有四步加载 - 分块 - 向量化 - 检索。听起来简单但每一步都有讲究。加载与分块我放弃了处理所有格式的幻想优先支持Markdown和纯文本因为这是知识库最常见的形态。分块策略上没有采用简单的固定长度重叠分块而是尝试了基于语义的分割器如semantic-text-splitter它能更好地在句子或段落边界处切割保持语义完整性。一个关键参数是块大小chunk_size和重叠区overlap。经过测试对于技术文档512到1024的token长度配合10%-15%的重叠在召回率和上下文噪音之间取得了不错的平衡。注意重叠区不是越大越好。过大的重叠会导致检索出大量高度相似的冗余片段反而干扰后续的重排序和生成阶段。向量化与检索向量模型我选了BAAI/bge-small-zh-v1.5这个模型在中文语义相似度任务上表现均衡且模型体积小推理速度快。向量数据库则是ChromaDB它轻量、易嵌入、且支持内存和持久化两种模式非常适合轻量级部署。检索环节最基础的当然是余弦相似度但仅仅这样还不够。2.2 SKILL架构如何赋予智能体“行动力”SKILL架构的核心思想是“任务分解”和“工具调用”。你可以把它理解为一个智能体的“技能工具箱”。每个SKILL都是一个独立的函数或模块有明确的输入、输出和职责。智能体的大脑通常是LLM根据用户的问题规划需要调用哪些SKILL并按顺序执行它们。在这个项目中我没有直接用像LangChain Agent那样复杂的Agent执行器而是设计了一个更简单的基于LLM函数调用Function Calling的SKILL调度器。具体来说技能定义我将核心能力定义成几个关键的SKILL。skill_parse_intent: 分析用户问题识别真实意图和关键实体。skill_retrieve_related_info: 这是与RAG对接的核心技能。它接收解析后的意图和实体构造查询语句可能是关键词也可能是改写后的问题调用向量数据库进行检索并返回top-k个相关片段。skill_rerank_and_synthesize: 对检索到的多个片段进行重排序和去重并初步合成一个更连贯的上下文。skill_generate_answer: 利用合成后的上下文和原始问题生成最终答案。skill_ask_for_clarification: 当检索结果置信度太低或意图模糊时向用户提问以澄清。调度逻辑智能体的“大脑”我选用的是通义千问或DeepSeek的API因为它们对函数调用支持良好会根据当前对话状态决定下一步调用哪个SKILL。这个过程是动态的。例如用户问“Python里怎么连接数据库”流程可能是parse_intent-retrieve_related_info查“Python 数据库 连接”-generate_answer。但如果用户接着问“那用异步的方式呢”流程可能变成parse_intent识别出是上一问的细化-retrieve_related_info查“Python 异步 数据库 连接”-rerank_and_synthesize可能需要结合上一轮的上下文-generate_answer。这种架构的好处是透明且可控。每个SKILL都可以单独测试、优化。比如你可以轻易地替换skill_retrieve_related_info内部的检索算法而不影响其他技能。2.3 “深度融合”体现在哪里深度融合不是简单地把RAG作为一个SKILL来调用而是在数据流和控制流层面进行交织。查询改写与扩展在skill_retrieve_related_info内部我不会直接把原始问题扔去检索。而是先用一个小模型或LLM的少量提示对查询进行改写和扩展。比如“怎么报错”可能被改写成“Python程序运行时错误信息处理与调试方法”。这能显著提升检索召回率。迭代检索如果第一轮检索返回的结果质量不高比如相似度分数都低于某个阈值skill_retrieve_related_info可以触发skill_ask_for_clarification向用户询问更多细节然后基于新的信息发起第二轮检索。这就是SKILL架构带来的灵活性。上下文感知的检索skill_retrieve_related_info在构造查询时会参考对话历史作为上下文传入。这使得智能体能进行指代消解比如明白“上面说的那个方法”具体指什么。重排序Reranking的引入这是提升“精准”度的关键一步。初次向量检索称为“召回”追求的是全可能会返回一些相关但并非最相关的片段。我引入了一个轻量级的交叉编码器Cross-Encoder比如BAAI/bge-reranker-base。它的作用是将查询和每一个召回片段进行更精细的深度交互计算给出一个更准确的相关性分数然后根据这个分数对片段进行重排序。实测下来即使只保留重排序后的top-3片段生成答案的质量也常常优于使用top-10的原始向量检索结果。整个架构的流程图在脑海中是这样的用户输入 - 意图解析SKILL - (可能循环) 查询改写 - 向量检索 - 重排序SKILL - 信息合成SKILL - 生成答案SKILL - 输出。每个环节都可以被监控和度量。3. 从零搭建工具链选择与实操步骤理论说再多不如一行代码。下面我就手把手带你过一遍搭建过程。我的环境是Python 3.9追求极简依赖。3.1 环境准备与核心依赖安装首先创建一个干净的虚拟环境然后安装核心包。这里的关键是避免安装那些巨型全家桶。# 创建虚拟环境可选但推荐 python -m venv .venv source .venv/bin/activate # Linux/Mac # .venv\Scripts\activate # Windows # 核心依赖 pip install chromadb # 向量数据库核心 pip install sentence-transformers # 用于加载BGE等嵌入模型 pip install torch # 深度学习框架sentence-transformers依赖 pip install pypdf markdown-it-py # 文档加载器处理PDF和Markdown pip install tiktoken # 用于文本分块时的token计数更准确 pip install openai # 或 dashscope阿里云、zhipuai智谱等用于LLM API调用 # 注意如果你用国产API可能需要安装对应的SDK如 pip install dashscope对于重排序模型我们可以用sentence-transformers直接加载交叉编码器pip install sentence-transformers[cross-encoder]3.2 构建专属知识库向量库这是最基础的一步决定了智能体“知识”的广度和质量。步骤1文档加载我写了一个简单的加载器支持目录递归读取。import os from pathlib import Path def load_documents_from_dir(directory_path, extensions[.md, .txt]): 从目录加载所有指定扩展名的文档 docs [] for ext in extensions: for file_path in Path(directory_path).rglob(f*{ext}): try: with open(file_path, r, encodingutf-8) as f: content f.read() docs.append({ content: content, source: str(file_path.relative_to(directory_path)), type: ext }) except Exception as e: print(fError reading {file_path}: {e}) return docs # 示例加载你的Obsidian知识库目录 my_knowledge_base_path ./my_obsidian_vault raw_documents load_documents_from_dir(my_knowledge_base_path) print(fLoaded {len(raw_documents)} documents.)步骤2文本分块这里我使用了基于语义的分割但为了轻量先实现一个带重叠的固定长度分块作为备选。import tiktoken # 用于精确计算token数特别是对LLM上下文友好 def split_text_fixed_with_overlap(text, chunk_size500, chunk_overlap50, encoding_namecl100k_base): 将文本分割成指定token大小的块带有重叠区。 tokenizer tiktoken.get_encoding(encoding_name) tokens tokenizer.encode(text) chunks [] start 0 while start len(tokens): end start chunk_size chunk_tokens tokens[start:end] chunk_text tokenizer.decode(chunk_tokens) # 记录元数据源文件、起始位置等便于溯源 chunks.append({ text: chunk_text, token_count: len(chunk_tokens), start_idx: start, end_idx: end }) start chunk_size - chunk_overlap # 移动步长为块大小减重叠 return chunks # 对每个文档进行分块 all_chunks [] for doc in raw_documents: chunks split_text_fixed_with_overlap(doc[content], chunk_size600, chunk_overlap80) for chunk in chunks: chunk[source] doc[source] # 保留来源信息 all_chunks.extend(chunks) print(fCreated {len(all_chunks)} text chunks.)步骤3生成向量并存入ChromaDB这是构建检索核心的步骤。from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings # 1. 初始化嵌入模型 # 首次运行会下载模型建议使用国内镜像源加速 embed_model SentenceTransformer(BAAI/bge-small-zh-v1.5, devicecpu) # 小模型用CPU也行 # 2. 准备数据 chunk_texts [chunk[text] for chunk in all_chunks] chunk_metadatas [{source: chunk[source], start_idx: chunk[start_idx]} for chunk in all_chunks] chunk_ids [fchunk_{i} for i in range(len(all_chunks))] # 3. 生成向量 print(Generating embeddings... (this may take a while)) embeddings embed_model.encode(chunk_texts, normalize_embeddingsTrue) # 归一化很重要方便余弦相似度计算 print(fEmbeddings shape: {embeddings.shape}) # 4. 初始化ChromaDB客户端并创建集合 # 持久化到磁盘下次无需重新计算 client chromadb.PersistentClient(path./chroma_db_knowledge) collection client.create_collection( namemy_knowledge_base, metadata{hnsw:space: cosine} # 使用余弦相似度 ) # 5. 批量添加数据 collection.add( embeddingsembeddings.tolist(), # ChromaDB接收list of lists documentschunk_texts, metadataschunk_metadatas, idschunk_ids ) print(Knowledge base vector store created successfully!)实操心得normalize_embeddingsTrue这个参数至关重要。它会把向量归一化为单位长度这样计算余弦相似度就简化为点积速度快并且更符合语义相似度的比较方式。如果不归一化直接计算余弦相似度结果可能会不稳定。3.3 实现核心SKILL模块有了知识库接下来就是打造智能体的“技能”。技能1检索技能 (skill_retrieve_related_info)这是最核心的技能它封装了查询改写、向量检索、重排序全流程。from sentence_transformers import CrossEncoder class RetrievalSkill: def __init__(self, collection, embed_model, rerank_model_nameBAAI/bge-reranker-base): self.collection collection self.embed_model embed_model # 初始化重排序模型 self.reranker CrossEncoder(rerank_model_name, max_length512) def _rewrite_query(self, original_query, conversation_historyNone): 简单的查询改写。生产环境可以用小模型或LLM prompt优化。 # 这里是一个简单示例添加领域相关上下文 rewritten original_query if 错误 in original_query or 报错 in original_query: rewritten f问题排查: {original_query} # 如果有对话历史可以拼接上轮问答作为上下文 if conversation_history: # 简单取最后两轮 context .join([fQ:{h[q]} A:{h[a]} for h in conversation_history[-2:]]) rewritten f{context} 当前问题: {original_query} return rewritten def _retrieve_with_rerank(self, query, top_k_initial10, top_k_final3): 检索并重排序 # 1. 查询改写 rewritten_query self._rewrite_query(query) # 2. 向量检索召回 query_embedding self.embed_model.encode(rewritten_query, normalize_embeddingsTrue) results self.collection.query( query_embeddings[query_embedding.tolist()], n_resultstop_k_initial, include[documents, metadatas, distances] ) retrieved_docs results[documents][0] retrieved_metas results[metadatas][0] retrieved_distances results[distances][0] if not retrieved_docs: return [] # 3. 重排序 # 构造 (query, document) 对 pairs [[rewritten_query, doc] for doc in retrieved_docs] rerank_scores self.reranker.predict(pairs) # 4. 结合重排序分数和原始距离分数可选这里以重排序分数为主 combined_results list(zip(retrieved_docs, retrieved_metas, retrieved_distances, rerank_scores)) # 按重排序分数降序排列 combined_results.sort(keylambda x: x[3], reverseTrue) # 5. 返回top_k_final个结果 final_results [] for doc, meta, dist, score in combined_results[:top_k_final]: final_results.append({ content: doc, source: meta[source], vector_distance: dist, rerank_score: score }) return final_results def execute(self, query, contextNone): 技能执行入口 return self._retrieve_with_rerank(query)技能2生成技能 (skill_generate_answer)这个技能负责整合检索到的信息生成友好、准确的回答。# 假设我们使用OpenAI格式的API如通义千问、DeepSeek等 import openai # 这里作为示例实际请替换为你所用平台的SDK class GenerationSkill: def __init__(self, api_key, base_url, modelqwen-max): # 示例为通义千问 # 配置客户端请根据你使用的平台调整 self.client openai.OpenAI( api_keyapi_key, base_urlbase_url # 例如 https://dashscope.aliyuncs.com/compatible-mode/v1 ) self.model model def _build_prompt(self, query, retrieved_contexts): 构建生成提示词。这里是效果好坏的关键 context_str \n---\n.join([f[来源{ctx[source]}]\n{ctx[content]} for ctx in retrieved_contexts]) prompt f你是一个专业的助手请根据以下提供的参考信息来回答问题。如果信息足够请基于信息给出准确、清晰的回答并注明信息来源。如果信息不足或与问题无关请如实告知并尝试根据你的知识进行回答同时说明这部分并非来自提供的资料。 参考信息 {context_str} 问题{query} 请用中文回答 return prompt def execute(self, query, retrieved_contexts): if not retrieved_contexts: # 如果没有检索到相关信息直接让模型自由发挥或调用其他技能 prompt f问题{query}\n\n请用中文回答 else: prompt self._build_prompt(query, retrieved_contexts) try: response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.2, # 低温度保证答案稳定 max_tokens1024 ) answer response.choices[0].message.content return answer except Exception as e: return f生成答案时出错{e}技能3意图解析技能 (skill_parse_intent)这个技能让智能体学会“听懂话”。class IntentParsingSkill: def __init__(self, llm_client): self.llm llm_client # 可以复用上面的GenerationSkill的client或者用更小的模型 def execute(self, query, historyNone): 解析用户意图返回结构化的信息 prompt f请分析以下用户问题的意图和关键实体。 问题{query} 请以JSON格式输出包含以下字段 - intent (字符串): 概括用户意图如“查询操作方法”、“寻求故障排除”、“请求定义解释”、“比较差异”等。 - entities (列表): 提取的关键词或实体如技术名词、产品名、错误代码等。 - is_clarification (布尔值): 该问题是否是对前一个问题的澄清或细化。 - needs_context (布尔值): 回答此问题是否需要参考之前的对话历史。 # 调用LLM进行解析 # 这里简化为直接调用实际应考虑错误处理 response self.llm.chat.completions.create( modelqwen-plus, # 可以用小一点的模型做解析 messages[{role: user, content: prompt}], temperature0.1, response_format{ type: json_object } # 要求返回JSON ) import json try: parsed json.loads(response.choices[0].message.content) return parsed except: # 解析失败返回默认值 return { intent: general_query, entities: [], is_clarification: False, needs_context: False }3.4 组装智能体简单的调度逻辑有了这些技能我们就可以组装一个简单的智能体了。这里实现一个顺序执行的流程更复杂的可以根据意图解析结果动态规划。class SimpleRAGAgent: def __init__(self, retrieval_skill, generation_skill, intent_skill): self.retrieval retrieval_skill self.generation generation_skill self.intent intent_skill self.conversation_history [] def chat(self, user_query): print(f用户: {user_query}) # 1. 解析意图 intent_result self.intent.execute(user_query, self.conversation_history) print(f解析意图: {intent_result}) # 2. 检索相关信息 # 可以根据意图调整检索策略例如如果是“比较差异”可能需要检索更多片段 retrieved self.retrieval.execute(user_query, self.conversation_history) print(f检索到 {len(retrieved)} 条相关片段) # 3. 生成答案 answer self.generation.execute(user_query, retrieved) # 4. 更新历史 self.conversation_history.append({q: user_query, a: answer[:100]}) # 只存摘要 print(f助手: {answer[:200]}...) # 打印部分回答 return answer, retrieved # 返回答案和检索来源便于调试 # 初始化智能体 # 先初始化各个技能所需的组件 client chromadb.PersistentClient(path./chroma_db_knowledge) collection client.get_collection(my_knowledge_base) embed_model SentenceTransformer(BAAI/bge-small-zh-v1.5) retrieval_skill RetrievalSkill(collection, embed_model) generation_skill GenerationSkill(api_keyyour_api_key, base_urlyour_base_url) intent_skill IntentParsingSkill(generation_skill.client) # 共享client agent SimpleRAGAgent(retrieval_skill, generation_skill, intent_skill) # 开始对话 answer, sources agent.chat(Python中如何读取CSV文件)4. 效果优化与深度调参实战系统跑起来只是第一步要让其真正“精准”还需要精细化的调优。这部分是区分玩具和可用工具的关键。4.1 分块策略的玄学大小、重叠与语义边界分块是RAG的“地基”地基不牢后面检索再强也白搭。块大小Chunk Size这需要在“信息完整性”和“检索噪音”之间权衡。我的经验是事实性问答如“某函数的参数是什么”适合较小的块256-512 tokens目标明确信息集中。概念性解释如“解释一下什么是RAG”需要较大的块1024 tokens或更大因为解释可能跨越多个段落。实操方法如“如何部署一个Django项目”中等块512-768 tokens既包含步骤又不会混入太多无关信息。测试方法准备一组典型问题用不同块大小构建向量库然后看检索到的前3个片段的平均相关性可以人工标注或通过LLM评估。选择召回率和精度综合最好的那个。重叠区Overlap目的是防止一个概念被生硬地切割在两个块中。但重叠不是简单的复制粘贴。我尝试过一种动态重叠策略在分块时如果当前块的结尾是一个句子的中间或者下一个句子的开头明显是承接关系比如“然而”、“此外”、“具体来说”就增加重叠量直到找到一个合适的句子边界。这需要一些简单的规则或小模型来判断但效果比固定重叠好。语义分割这是进阶玩法。我后来引入了semantic-text-splitter库它利用嵌入模型计算句子间的相似度在语义变化大的地方进行切割。这对于结构松散、段落长的文档如会议记录、长篇文章效果显著。4.2 检索环节的“组合拳”从关键词到混合搜索单一的向量搜索并非万能。尤其是在知识库包含大量专有名词、代码、版本号时传统的关键词搜索如BM25往往更准。我实现了混合检索Hybrid Search同时进行向量检索和关键词检索然后融合两者的结果。# 伪代码示例使用ChromaDB的where过滤器进行简单关键词匹配需提前在metadata中存好关键词 def hybrid_retrieve(query, collection, embed_model, alpha0.5): # 1. 向量检索 vec_results collection.query(query_embeddings[embed_model.encode(query)], n_results10) # 2. 关键词检索 (简化版从查询中提取名词作为关键词) # 这里需要更复杂的关键词提取可以用jieba等 keywords extract_keywords(query) keyword_results collection.query( query_texts[query], # ChromaDB也支持文本查询基于TF-IDF等 n_results10, # 或者用 where 过滤器进行元数据过滤 # where{$or: [{metadata_key: {$contains: kw}} for kw in keywords]} ) # 3. 结果融合 ( Reciprocal Rank Fusion, RRF 是一种简单有效的方法) fused_results reciprocal_rank_fusion(vec_results, keyword_results) return fused_resultsReciprocal Rank Fusion (RRF)算法很简单但效果拔群。它不关心分数绝对值只关心排名。公式是score 1 / (rank k)其中k是一个常数通常取60。对每个文档将它在两个结果列表中的这个分数相加得到最终分然后重新排序。这样一个在两个列表中排名都靠前的文档最终分数会很高。4.3 提示工程让LLM成为“信息整合大师”skill_generate_answer中的提示词模板是灵魂。经过无数次调试我总结出几个黄金法则明确指令开头就告诉模型“你是一个XX领域的专家请基于以下参考信息回答问题”。这能有效降低幻觉。清晰的结构用“---”或“###”等符号分隔不同的参考片段并在每个片段前注明来源如文件名。这能帮助模型区分不同来源的信息。强制引用在提示词末尾加上“请在你的回答中引用来源例如[来源1]”。虽然模型不一定完全遵守但能显著提高其参考提供信息的意识。处理“不知道”明确告诉模型“如果参考信息不足以回答问题请如实说明并可以基于你的通用知识进行补充但需指出这部分并非来自资料”。这比让它胡编乱造要好。分步思考Chain-of-Thought对于复杂问题可以要求模型先复述问题然后列出参考信息中的相关点最后进行综合。虽然增加了token消耗但能提升推理的准确性和可解释性。一个优化后的提示词模板如下def build_enhanced_prompt(query, contexts): context_str for i, ctx in enumerate(contexts): context_str f[资料片段{i1}, 来自 {ctx[source]}]:\n{ctx[content]}\n\n prompt f你是一个技术专家你的任务是根据用户问题严格依据下面提供的参考资料来组织答案。请遵循以下步骤 1. 理解问题{query} 2. 仔细阅读以下所有参考资料。 3. 判断哪些资料与问题直接相关。 4. 综合相关部分形成完整、准确的答案。 5. 在答案中用括号标注引用的资料编号例如[1]。 参考资料 {context_str} 请开始你的回答直接给出答案无需重复步骤 return prompt4.4 评估与迭代如何知道系统变好了不能凭感觉优化。我建立了一个简单的评估体系构建测试集QA对从知识库中手动整理或生成50-100个“问题-标准答案”对。答案应能从知识库中明确找到。定义评估指标检索召回率Retrieval Recall标准答案所在的文档块是否出现在检索结果的Top-K中K3,5,10。这是基础。答案相关性Answer Relevance用LLM如GPT-4或人工判断生成的答案与标准答案的语义相关性1-5分。答案忠实度Answer Faithfulness生成的答案是否严格基于提供的参考资料有没有“无中生有”幻觉。这可以通过让LLM判断答案中的陈述是否能在上下文中找到依据来评估。A/B测试每次调整一个参数如分块大小、重排序模型、提示词在测试集上运行对比指标变化。只有数据提升才说明优化有效。5. 避坑指南与常见问题排查这条路我踩过不少坑这里把最常见的“雷区”和解决方法列出来希望能帮你节省大量时间。5.1 检索效果差总是答非所问可能原因1向量模型不匹配。你用了一个通用英文模型去编码中文知识库。解决务必使用与文档语言匹配的模型。中文首选BAAI/bge-*系列或m3e系列。对于混合中英文的文档BAAI/bge-m3是不错的选择。可能原因2分块不合理。块太大包含了无关信息块太小语义不完整。解决回顾4.1节针对你的文档类型调整分块策略。可视化你的块随机抽样一些块看看内容是否自然连贯。可能原因3查询与文档表述差异大。用户问“咋装软体”文档里写“软件安装步骤”。解决强化skill_retrieve_related_info中的查询改写模块。可以用一个轻量级模型如Qwen2.5-1.5B专门做查询扩展和同义词替换。可能原因4没有重排序。向量检索的Top1不一定是最相关的。解决务必加上重排序步骤。即使是小模型如BAAI/bge-reranker-v2-mini也能带来显著提升。5.2 回答出现幻觉Hallucination编造信息可能原因1提示词不够强硬。模型没有被严格限制在参考信息内。解决使用4.3节中的“强制引用”和“分步思考”提示词。明确告知模型“只能使用提供的资料”。可能原因2检索到的上下文质量太低或为空。模型在“无米下锅”时容易瞎编。解决在skill_generate_answer中增加判断。如果检索结果为空或最高分数低于阈值如0.5则直接回复“在现有资料中未找到相关信息”并建议用户换个问法或补充知识库。不要让它自由发挥。可能原因3上下文过长模型“忘记”了指令。当检索到的片段很多拼成的上下文超过模型有效上下文窗口时模型可能会忽略开头的指令。解决严格控制输入模型的上下文长度。在合成上下文时只保留重排序分数最高的前3-5个片段。或者采用“Map-Reduce”策略让模型先对每个片段单独总结再基于总结生成最终答案虽然更耗时。5.3 系统响应速度慢可能原因1嵌入模型太大。使用了像text-embedding-3-large这样的大模型。解决在CPU上bge-small比bge-large快一个数量级而精度损失在可接受范围内。先用小模型跑通再考虑升级。可能原因2每次检索都实时计算查询向量。解决对于常见问题可以做一个简单的缓存。将(query, top_k_results)缓存起来下次相同或相似查询直接返回。相似判断可以用查询向量的余弦相似度。可能原因3LLM生成速度慢。解决考虑使用推理速度更快的模型如DeepSeek Coder或者启用API的流式输出streaming让用户能先看到部分结果。对于简单、事实性问题甚至可以尝试不调用大模型直接从检索到的片段中提取答案基于规则的或用小模型做抽取。5.4 ChromaDB相关报错Collection not found确保创建集合和查询集合时使用的name完全一致包括大小写。使用client.list_collections()检查现有集合。添加数据时内存不足如果文档量极大10万一次性生成所有向量并添加可能导致内存溢出。解决分批处理。每处理1000个块就collection.add一次。查询时距离分数异常如果发现所有距离分数都差不多比如都在0.99以上很可能是嵌入向量没有归一化。解决确保调用embed_model.encode(text, normalize_embeddingsTrue)。6. 进阶思路让智能体更“智能”当基础版本稳定后可以探索一些进阶功能让系统从“好用”变得“聪明”。SKILL的链式与图式调用目前的调度是线性的。更高级的智能体可以根据意图解析的结果动态生成一个技能调用图DAG。例如对于问题“比较一下Django和Flask在ORM方面的优劣”可以并行调用两个检索技能分别查Django ORM和Flask SQLAlchemy然后将结果交给一个“比较分析”技能进行合成。自我反思与修正在生成答案后增加一个skill_self_reflect。让LLM自己检查答案是否回答了问题是否引用了资料是否有矛盾之处如果发现问题可以触发新一轮的检索或生成。知识库的主动更新与评估智能体可以记录那些它无法回答或回答质量差的问题。定期将这些“未解决问题”报告给知识库维护者提示需要补充或更新哪些文档。甚至可以尝试让智能体根据对话自动生成知识库条目的草稿。多模态知识库不仅限于文本。可以将图片、表格、PDF中的图表通过多模态模型如Qwen-VL也编码进向量库。当用户问“请展示一下架构图”智能体可以检索出相关的图片片段。与外部工具集成将SKILL扩展到知识库之外。例如skill_execute_code可以运行代码片段验证答案skill_search_web可以在本地知识库不足时安全地搜索网络信息需谨慎处理。这个项目做到最后给我的感觉不再是简单地拼接工具而是在设计一个“思考流程”。RAG提供了记忆SKILL架构定义了思考的步骤。两者的深度融合让这个智能体在面对专业问题时终于有了一点“老师傅”的味道——知道该去哪里找资料知道怎么把资料组织成答案也知道什么时候该承认自己不会。整个过程里最花时间的往往不是写代码而是反复调整分块、优化提示词、评估效果这些细致活。但每解决一个小问题看到回答的准确度提升一点那种成就感是实实在在的。希望这份详细的实践记录能帮你少走些弯路更快地构建出属于你自己的、那个“懂行”的智能助手。 轻量级RAG与SKILL架构融合：构建精准知识匹配智能体实践
1. 项目概述当RAG遇上SKILL智能体如何“精准”思考最近在折腾一个挺有意思的项目核心就一句话让一个AI智能体在面对用户五花八门的问题时能像一位经验丰富的专家一样快速、精准地从它自己的“专属知识库”里找到最相关的信息来回答。听起来是不是有点像现在大火的RAG没错但又不完全是。我们这次玩得更深入一点把RAG检索增强生成和SKILL技能架构给“焊”在了一起。这个项目的标题叫“轻量级RAG与SKILL架构深度融合专属知识库驱动智能体精准知识匹配应用实践”名字有点长但拆开来看就是三个核心轻量级RAG、SKILL架构、以及它们结合后驱动的精准知识匹配。为什么要把这两者结合这是我踩过不少坑后的深刻体会。纯RAG系统就像一个记忆力超群但不太会变通的“书呆子”。你问它问题它能从海量文档里找到相关段落然后原封不动或者稍作改写地吐给你。但现实世界的问题往往更复杂用户可能问得模糊可能需要多步推理可能需要结合不同来源的知识进行判断。这时候单纯的“检索-生成”就显得力不从心了。而SKILL架构恰恰是让智能体“会变通”的关键。它把复杂任务拆解成一个个可执行、可组合的“技能”比如“解析用户意图”、“查询知识库”、“验证信息”、“格式化回答”等。当RAG的检索能力被封装成一个或多个SKILL智能体就能更智能地决定什么时候该去查资料查哪些资料查到资料后该怎么用这个实践的目标就是构建一个不依赖庞大算力、能快速部署、且真正“懂行”的智能体。它特别适合那些有垂直领域知识库的场景比如企业内部的技术支持文档库、某个专业领域的法规库、甚至是个人精心整理的笔记系统比如用Obsidian搭建的Wiki。接下来我就把自己从架构设计、工具选型到代码实现的完整过程以及过程中那些“血泪教训”和“意外惊喜”毫无保留地分享出来。2. 核心架构设计轻量、解耦与精准的三角平衡设计这个系统的起点是明确三个核心原则轻量、解耦、精准。轻量意味着我们不能一上来就搞复杂的微服务集群得让它在单机甚至资源受限的环境下也能跑起来解耦是为了让RAG模块和SKILL模块能独立进化互不干扰精准则是最终目标一切设计都要服务于提升答案的相关性和准确性。2.1 为什么是“轻量级”RAG市面上成熟的RAG框架很多像LangChain、LlamaIndex功能强大但生态复杂有时候给人一种“杀鸡用牛刀”的感觉。对于专属知识库场景尤其是初期验证阶段我们更需要一个聚焦核心流程、依赖少、调试透明的方案。我的选择是自研一个轻量级RAG管道。它的核心流程只有四步加载 - 分块 - 向量化 - 检索。听起来简单但每一步都有讲究。加载与分块我放弃了处理所有格式的幻想优先支持Markdown和纯文本因为这是知识库最常见的形态。分块策略上没有采用简单的固定长度重叠分块而是尝试了基于语义的分割器如semantic-text-splitter它能更好地在句子或段落边界处切割保持语义完整性。一个关键参数是块大小chunk_size和重叠区overlap。经过测试对于技术文档512到1024的token长度配合10%-15%的重叠在召回率和上下文噪音之间取得了不错的平衡。注意重叠区不是越大越好。过大的重叠会导致检索出大量高度相似的冗余片段反而干扰后续的重排序和生成阶段。向量化与检索向量模型我选了BAAI/bge-small-zh-v1.5这个模型在中文语义相似度任务上表现均衡且模型体积小推理速度快。向量数据库则是ChromaDB它轻量、易嵌入、且支持内存和持久化两种模式非常适合轻量级部署。检索环节最基础的当然是余弦相似度但仅仅这样还不够。2.2 SKILL架构如何赋予智能体“行动力”SKILL架构的核心思想是“任务分解”和“工具调用”。你可以把它理解为一个智能体的“技能工具箱”。每个SKILL都是一个独立的函数或模块有明确的输入、输出和职责。智能体的大脑通常是LLM根据用户的问题规划需要调用哪些SKILL并按顺序执行它们。在这个项目中我没有直接用像LangChain Agent那样复杂的Agent执行器而是设计了一个更简单的基于LLM函数调用Function Calling的SKILL调度器。具体来说技能定义我将核心能力定义成几个关键的SKILL。skill_parse_intent: 分析用户问题识别真实意图和关键实体。skill_retrieve_related_info: 这是与RAG对接的核心技能。它接收解析后的意图和实体构造查询语句可能是关键词也可能是改写后的问题调用向量数据库进行检索并返回top-k个相关片段。skill_rerank_and_synthesize: 对检索到的多个片段进行重排序和去重并初步合成一个更连贯的上下文。skill_generate_answer: 利用合成后的上下文和原始问题生成最终答案。skill_ask_for_clarification: 当检索结果置信度太低或意图模糊时向用户提问以澄清。调度逻辑智能体的“大脑”我选用的是通义千问或DeepSeek的API因为它们对函数调用支持良好会根据当前对话状态决定下一步调用哪个SKILL。这个过程是动态的。例如用户问“Python里怎么连接数据库”流程可能是parse_intent-retrieve_related_info查“Python 数据库 连接”-generate_answer。但如果用户接着问“那用异步的方式呢”流程可能变成parse_intent识别出是上一问的细化-retrieve_related_info查“Python 异步 数据库 连接”-rerank_and_synthesize可能需要结合上一轮的上下文-generate_answer。这种架构的好处是透明且可控。每个SKILL都可以单独测试、优化。比如你可以轻易地替换skill_retrieve_related_info内部的检索算法而不影响其他技能。2.3 “深度融合”体现在哪里深度融合不是简单地把RAG作为一个SKILL来调用而是在数据流和控制流层面进行交织。查询改写与扩展在skill_retrieve_related_info内部我不会直接把原始问题扔去检索。而是先用一个小模型或LLM的少量提示对查询进行改写和扩展。比如“怎么报错”可能被改写成“Python程序运行时错误信息处理与调试方法”。这能显著提升检索召回率。迭代检索如果第一轮检索返回的结果质量不高比如相似度分数都低于某个阈值skill_retrieve_related_info可以触发skill_ask_for_clarification向用户询问更多细节然后基于新的信息发起第二轮检索。这就是SKILL架构带来的灵活性。上下文感知的检索skill_retrieve_related_info在构造查询时会参考对话历史作为上下文传入。这使得智能体能进行指代消解比如明白“上面说的那个方法”具体指什么。重排序Reranking的引入这是提升“精准”度的关键一步。初次向量检索称为“召回”追求的是全可能会返回一些相关但并非最相关的片段。我引入了一个轻量级的交叉编码器Cross-Encoder比如BAAI/bge-reranker-base。它的作用是将查询和每一个召回片段进行更精细的深度交互计算给出一个更准确的相关性分数然后根据这个分数对片段进行重排序。实测下来即使只保留重排序后的top-3片段生成答案的质量也常常优于使用top-10的原始向量检索结果。整个架构的流程图在脑海中是这样的用户输入 - 意图解析SKILL - (可能循环) 查询改写 - 向量检索 - 重排序SKILL - 信息合成SKILL - 生成答案SKILL - 输出。每个环节都可以被监控和度量。3. 从零搭建工具链选择与实操步骤理论说再多不如一行代码。下面我就手把手带你过一遍搭建过程。我的环境是Python 3.9追求极简依赖。3.1 环境准备与核心依赖安装首先创建一个干净的虚拟环境然后安装核心包。这里的关键是避免安装那些巨型全家桶。# 创建虚拟环境可选但推荐 python -m venv .venv source .venv/bin/activate # Linux/Mac # .venv\Scripts\activate # Windows # 核心依赖 pip install chromadb # 向量数据库核心 pip install sentence-transformers # 用于加载BGE等嵌入模型 pip install torch # 深度学习框架sentence-transformers依赖 pip install pypdf markdown-it-py # 文档加载器处理PDF和Markdown pip install tiktoken # 用于文本分块时的token计数更准确 pip install openai # 或 dashscope阿里云、zhipuai智谱等用于LLM API调用 # 注意如果你用国产API可能需要安装对应的SDK如 pip install dashscope对于重排序模型我们可以用sentence-transformers直接加载交叉编码器pip install sentence-transformers[cross-encoder]3.2 构建专属知识库向量库这是最基础的一步决定了智能体“知识”的广度和质量。步骤1文档加载我写了一个简单的加载器支持目录递归读取。import os from pathlib import Path def load_documents_from_dir(directory_path, extensions[.md, .txt]): 从目录加载所有指定扩展名的文档 docs [] for ext in extensions: for file_path in Path(directory_path).rglob(f*{ext}): try: with open(file_path, r, encodingutf-8) as f: content f.read() docs.append({ content: content, source: str(file_path.relative_to(directory_path)), type: ext }) except Exception as e: print(fError reading {file_path}: {e}) return docs # 示例加载你的Obsidian知识库目录 my_knowledge_base_path ./my_obsidian_vault raw_documents load_documents_from_dir(my_knowledge_base_path) print(fLoaded {len(raw_documents)} documents.)步骤2文本分块这里我使用了基于语义的分割但为了轻量先实现一个带重叠的固定长度分块作为备选。import tiktoken # 用于精确计算token数特别是对LLM上下文友好 def split_text_fixed_with_overlap(text, chunk_size500, chunk_overlap50, encoding_namecl100k_base): 将文本分割成指定token大小的块带有重叠区。 tokenizer tiktoken.get_encoding(encoding_name) tokens tokenizer.encode(text) chunks [] start 0 while start len(tokens): end start chunk_size chunk_tokens tokens[start:end] chunk_text tokenizer.decode(chunk_tokens) # 记录元数据源文件、起始位置等便于溯源 chunks.append({ text: chunk_text, token_count: len(chunk_tokens), start_idx: start, end_idx: end }) start chunk_size - chunk_overlap # 移动步长为块大小减重叠 return chunks # 对每个文档进行分块 all_chunks [] for doc in raw_documents: chunks split_text_fixed_with_overlap(doc[content], chunk_size600, chunk_overlap80) for chunk in chunks: chunk[source] doc[source] # 保留来源信息 all_chunks.extend(chunks) print(fCreated {len(all_chunks)} text chunks.)步骤3生成向量并存入ChromaDB这是构建检索核心的步骤。from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings # 1. 初始化嵌入模型 # 首次运行会下载模型建议使用国内镜像源加速 embed_model SentenceTransformer(BAAI/bge-small-zh-v1.5, devicecpu) # 小模型用CPU也行 # 2. 准备数据 chunk_texts [chunk[text] for chunk in all_chunks] chunk_metadatas [{source: chunk[source], start_idx: chunk[start_idx]} for chunk in all_chunks] chunk_ids [fchunk_{i} for i in range(len(all_chunks))] # 3. 生成向量 print(Generating embeddings... (this may take a while)) embeddings embed_model.encode(chunk_texts, normalize_embeddingsTrue) # 归一化很重要方便余弦相似度计算 print(fEmbeddings shape: {embeddings.shape}) # 4. 初始化ChromaDB客户端并创建集合 # 持久化到磁盘下次无需重新计算 client chromadb.PersistentClient(path./chroma_db_knowledge) collection client.create_collection( namemy_knowledge_base, metadata{hnsw:space: cosine} # 使用余弦相似度 ) # 5. 批量添加数据 collection.add( embeddingsembeddings.tolist(), # ChromaDB接收list of lists documentschunk_texts, metadataschunk_metadatas, idschunk_ids ) print(Knowledge base vector store created successfully!)实操心得normalize_embeddingsTrue这个参数至关重要。它会把向量归一化为单位长度这样计算余弦相似度就简化为点积速度快并且更符合语义相似度的比较方式。如果不归一化直接计算余弦相似度结果可能会不稳定。3.3 实现核心SKILL模块有了知识库接下来就是打造智能体的“技能”。技能1检索技能 (skill_retrieve_related_info)这是最核心的技能它封装了查询改写、向量检索、重排序全流程。from sentence_transformers import CrossEncoder class RetrievalSkill: def __init__(self, collection, embed_model, rerank_model_nameBAAI/bge-reranker-base): self.collection collection self.embed_model embed_model # 初始化重排序模型 self.reranker CrossEncoder(rerank_model_name, max_length512) def _rewrite_query(self, original_query, conversation_historyNone): 简单的查询改写。生产环境可以用小模型或LLM prompt优化。 # 这里是一个简单示例添加领域相关上下文 rewritten original_query if 错误 in original_query or 报错 in original_query: rewritten f问题排查: {original_query} # 如果有对话历史可以拼接上轮问答作为上下文 if conversation_history: # 简单取最后两轮 context .join([fQ:{h[q]} A:{h[a]} for h in conversation_history[-2:]]) rewritten f{context} 当前问题: {original_query} return rewritten def _retrieve_with_rerank(self, query, top_k_initial10, top_k_final3): 检索并重排序 # 1. 查询改写 rewritten_query self._rewrite_query(query) # 2. 向量检索召回 query_embedding self.embed_model.encode(rewritten_query, normalize_embeddingsTrue) results self.collection.query( query_embeddings[query_embedding.tolist()], n_resultstop_k_initial, include[documents, metadatas, distances] ) retrieved_docs results[documents][0] retrieved_metas results[metadatas][0] retrieved_distances results[distances][0] if not retrieved_docs: return [] # 3. 重排序 # 构造 (query, document) 对 pairs [[rewritten_query, doc] for doc in retrieved_docs] rerank_scores self.reranker.predict(pairs) # 4. 结合重排序分数和原始距离分数可选这里以重排序分数为主 combined_results list(zip(retrieved_docs, retrieved_metas, retrieved_distances, rerank_scores)) # 按重排序分数降序排列 combined_results.sort(keylambda x: x[3], reverseTrue) # 5. 返回top_k_final个结果 final_results [] for doc, meta, dist, score in combined_results[:top_k_final]: final_results.append({ content: doc, source: meta[source], vector_distance: dist, rerank_score: score }) return final_results def execute(self, query, contextNone): 技能执行入口 return self._retrieve_with_rerank(query)技能2生成技能 (skill_generate_answer)这个技能负责整合检索到的信息生成友好、准确的回答。# 假设我们使用OpenAI格式的API如通义千问、DeepSeek等 import openai # 这里作为示例实际请替换为你所用平台的SDK class GenerationSkill: def __init__(self, api_key, base_url, modelqwen-max): # 示例为通义千问 # 配置客户端请根据你使用的平台调整 self.client openai.OpenAI( api_keyapi_key, base_urlbase_url # 例如 https://dashscope.aliyuncs.com/compatible-mode/v1 ) self.model model def _build_prompt(self, query, retrieved_contexts): 构建生成提示词。这里是效果好坏的关键 context_str \n---\n.join([f[来源{ctx[source]}]\n{ctx[content]} for ctx in retrieved_contexts]) prompt f你是一个专业的助手请根据以下提供的参考信息来回答问题。如果信息足够请基于信息给出准确、清晰的回答并注明信息来源。如果信息不足或与问题无关请如实告知并尝试根据你的知识进行回答同时说明这部分并非来自提供的资料。 参考信息 {context_str} 问题{query} 请用中文回答 return prompt def execute(self, query, retrieved_contexts): if not retrieved_contexts: # 如果没有检索到相关信息直接让模型自由发挥或调用其他技能 prompt f问题{query}\n\n请用中文回答 else: prompt self._build_prompt(query, retrieved_contexts) try: response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.2, # 低温度保证答案稳定 max_tokens1024 ) answer response.choices[0].message.content return answer except Exception as e: return f生成答案时出错{e}技能3意图解析技能 (skill_parse_intent)这个技能让智能体学会“听懂话”。class IntentParsingSkill: def __init__(self, llm_client): self.llm llm_client # 可以复用上面的GenerationSkill的client或者用更小的模型 def execute(self, query, historyNone): 解析用户意图返回结构化的信息 prompt f请分析以下用户问题的意图和关键实体。 问题{query} 请以JSON格式输出包含以下字段 - intent (字符串): 概括用户意图如“查询操作方法”、“寻求故障排除”、“请求定义解释”、“比较差异”等。 - entities (列表): 提取的关键词或实体如技术名词、产品名、错误代码等。 - is_clarification (布尔值): 该问题是否是对前一个问题的澄清或细化。 - needs_context (布尔值): 回答此问题是否需要参考之前的对话历史。 # 调用LLM进行解析 # 这里简化为直接调用实际应考虑错误处理 response self.llm.chat.completions.create( modelqwen-plus, # 可以用小一点的模型做解析 messages[{role: user, content: prompt}], temperature0.1, response_format{ type: json_object } # 要求返回JSON ) import json try: parsed json.loads(response.choices[0].message.content) return parsed except: # 解析失败返回默认值 return { intent: general_query, entities: [], is_clarification: False, needs_context: False }3.4 组装智能体简单的调度逻辑有了这些技能我们就可以组装一个简单的智能体了。这里实现一个顺序执行的流程更复杂的可以根据意图解析结果动态规划。class SimpleRAGAgent: def __init__(self, retrieval_skill, generation_skill, intent_skill): self.retrieval retrieval_skill self.generation generation_skill self.intent intent_skill self.conversation_history [] def chat(self, user_query): print(f用户: {user_query}) # 1. 解析意图 intent_result self.intent.execute(user_query, self.conversation_history) print(f解析意图: {intent_result}) # 2. 检索相关信息 # 可以根据意图调整检索策略例如如果是“比较差异”可能需要检索更多片段 retrieved self.retrieval.execute(user_query, self.conversation_history) print(f检索到 {len(retrieved)} 条相关片段) # 3. 生成答案 answer self.generation.execute(user_query, retrieved) # 4. 更新历史 self.conversation_history.append({q: user_query, a: answer[:100]}) # 只存摘要 print(f助手: {answer[:200]}...) # 打印部分回答 return answer, retrieved # 返回答案和检索来源便于调试 # 初始化智能体 # 先初始化各个技能所需的组件 client chromadb.PersistentClient(path./chroma_db_knowledge) collection client.get_collection(my_knowledge_base) embed_model SentenceTransformer(BAAI/bge-small-zh-v1.5) retrieval_skill RetrievalSkill(collection, embed_model) generation_skill GenerationSkill(api_keyyour_api_key, base_urlyour_base_url) intent_skill IntentParsingSkill(generation_skill.client) # 共享client agent SimpleRAGAgent(retrieval_skill, generation_skill, intent_skill) # 开始对话 answer, sources agent.chat(Python中如何读取CSV文件)4. 效果优化与深度调参实战系统跑起来只是第一步要让其真正“精准”还需要精细化的调优。这部分是区分玩具和可用工具的关键。4.1 分块策略的玄学大小、重叠与语义边界分块是RAG的“地基”地基不牢后面检索再强也白搭。块大小Chunk Size这需要在“信息完整性”和“检索噪音”之间权衡。我的经验是事实性问答如“某函数的参数是什么”适合较小的块256-512 tokens目标明确信息集中。概念性解释如“解释一下什么是RAG”需要较大的块1024 tokens或更大因为解释可能跨越多个段落。实操方法如“如何部署一个Django项目”中等块512-768 tokens既包含步骤又不会混入太多无关信息。测试方法准备一组典型问题用不同块大小构建向量库然后看检索到的前3个片段的平均相关性可以人工标注或通过LLM评估。选择召回率和精度综合最好的那个。重叠区Overlap目的是防止一个概念被生硬地切割在两个块中。但重叠不是简单的复制粘贴。我尝试过一种动态重叠策略在分块时如果当前块的结尾是一个句子的中间或者下一个句子的开头明显是承接关系比如“然而”、“此外”、“具体来说”就增加重叠量直到找到一个合适的句子边界。这需要一些简单的规则或小模型来判断但效果比固定重叠好。语义分割这是进阶玩法。我后来引入了semantic-text-splitter库它利用嵌入模型计算句子间的相似度在语义变化大的地方进行切割。这对于结构松散、段落长的文档如会议记录、长篇文章效果显著。4.2 检索环节的“组合拳”从关键词到混合搜索单一的向量搜索并非万能。尤其是在知识库包含大量专有名词、代码、版本号时传统的关键词搜索如BM25往往更准。我实现了混合检索Hybrid Search同时进行向量检索和关键词检索然后融合两者的结果。# 伪代码示例使用ChromaDB的where过滤器进行简单关键词匹配需提前在metadata中存好关键词 def hybrid_retrieve(query, collection, embed_model, alpha0.5): # 1. 向量检索 vec_results collection.query(query_embeddings[embed_model.encode(query)], n_results10) # 2. 关键词检索 (简化版从查询中提取名词作为关键词) # 这里需要更复杂的关键词提取可以用jieba等 keywords extract_keywords(query) keyword_results collection.query( query_texts[query], # ChromaDB也支持文本查询基于TF-IDF等 n_results10, # 或者用 where 过滤器进行元数据过滤 # where{$or: [{metadata_key: {$contains: kw}} for kw in keywords]} ) # 3. 结果融合 ( Reciprocal Rank Fusion, RRF 是一种简单有效的方法) fused_results reciprocal_rank_fusion(vec_results, keyword_results) return fused_resultsReciprocal Rank Fusion (RRF)算法很简单但效果拔群。它不关心分数绝对值只关心排名。公式是score 1 / (rank k)其中k是一个常数通常取60。对每个文档将它在两个结果列表中的这个分数相加得到最终分然后重新排序。这样一个在两个列表中排名都靠前的文档最终分数会很高。4.3 提示工程让LLM成为“信息整合大师”skill_generate_answer中的提示词模板是灵魂。经过无数次调试我总结出几个黄金法则明确指令开头就告诉模型“你是一个XX领域的专家请基于以下参考信息回答问题”。这能有效降低幻觉。清晰的结构用“---”或“###”等符号分隔不同的参考片段并在每个片段前注明来源如文件名。这能帮助模型区分不同来源的信息。强制引用在提示词末尾加上“请在你的回答中引用来源例如[来源1]”。虽然模型不一定完全遵守但能显著提高其参考提供信息的意识。处理“不知道”明确告诉模型“如果参考信息不足以回答问题请如实说明并可以基于你的通用知识进行补充但需指出这部分并非来自资料”。这比让它胡编乱造要好。分步思考Chain-of-Thought对于复杂问题可以要求模型先复述问题然后列出参考信息中的相关点最后进行综合。虽然增加了token消耗但能提升推理的准确性和可解释性。一个优化后的提示词模板如下def build_enhanced_prompt(query, contexts): context_str for i, ctx in enumerate(contexts): context_str f[资料片段{i1}, 来自 {ctx[source]}]:\n{ctx[content]}\n\n prompt f你是一个技术专家你的任务是根据用户问题严格依据下面提供的参考资料来组织答案。请遵循以下步骤 1. 理解问题{query} 2. 仔细阅读以下所有参考资料。 3. 判断哪些资料与问题直接相关。 4. 综合相关部分形成完整、准确的答案。 5. 在答案中用括号标注引用的资料编号例如[1]。 参考资料 {context_str} 请开始你的回答直接给出答案无需重复步骤 return prompt4.4 评估与迭代如何知道系统变好了不能凭感觉优化。我建立了一个简单的评估体系构建测试集QA对从知识库中手动整理或生成50-100个“问题-标准答案”对。答案应能从知识库中明确找到。定义评估指标检索召回率Retrieval Recall标准答案所在的文档块是否出现在检索结果的Top-K中K3,5,10。这是基础。答案相关性Answer Relevance用LLM如GPT-4或人工判断生成的答案与标准答案的语义相关性1-5分。答案忠实度Answer Faithfulness生成的答案是否严格基于提供的参考资料有没有“无中生有”幻觉。这可以通过让LLM判断答案中的陈述是否能在上下文中找到依据来评估。A/B测试每次调整一个参数如分块大小、重排序模型、提示词在测试集上运行对比指标变化。只有数据提升才说明优化有效。5. 避坑指南与常见问题排查这条路我踩过不少坑这里把最常见的“雷区”和解决方法列出来希望能帮你节省大量时间。5.1 检索效果差总是答非所问可能原因1向量模型不匹配。你用了一个通用英文模型去编码中文知识库。解决务必使用与文档语言匹配的模型。中文首选BAAI/bge-*系列或m3e系列。对于混合中英文的文档BAAI/bge-m3是不错的选择。可能原因2分块不合理。块太大包含了无关信息块太小语义不完整。解决回顾4.1节针对你的文档类型调整分块策略。可视化你的块随机抽样一些块看看内容是否自然连贯。可能原因3查询与文档表述差异大。用户问“咋装软体”文档里写“软件安装步骤”。解决强化skill_retrieve_related_info中的查询改写模块。可以用一个轻量级模型如Qwen2.5-1.5B专门做查询扩展和同义词替换。可能原因4没有重排序。向量检索的Top1不一定是最相关的。解决务必加上重排序步骤。即使是小模型如BAAI/bge-reranker-v2-mini也能带来显著提升。5.2 回答出现幻觉Hallucination编造信息可能原因1提示词不够强硬。模型没有被严格限制在参考信息内。解决使用4.3节中的“强制引用”和“分步思考”提示词。明确告知模型“只能使用提供的资料”。可能原因2检索到的上下文质量太低或为空。模型在“无米下锅”时容易瞎编。解决在skill_generate_answer中增加判断。如果检索结果为空或最高分数低于阈值如0.5则直接回复“在现有资料中未找到相关信息”并建议用户换个问法或补充知识库。不要让它自由发挥。可能原因3上下文过长模型“忘记”了指令。当检索到的片段很多拼成的上下文超过模型有效上下文窗口时模型可能会忽略开头的指令。解决严格控制输入模型的上下文长度。在合成上下文时只保留重排序分数最高的前3-5个片段。或者采用“Map-Reduce”策略让模型先对每个片段单独总结再基于总结生成最终答案虽然更耗时。5.3 系统响应速度慢可能原因1嵌入模型太大。使用了像text-embedding-3-large这样的大模型。解决在CPU上bge-small比bge-large快一个数量级而精度损失在可接受范围内。先用小模型跑通再考虑升级。可能原因2每次检索都实时计算查询向量。解决对于常见问题可以做一个简单的缓存。将(query, top_k_results)缓存起来下次相同或相似查询直接返回。相似判断可以用查询向量的余弦相似度。可能原因3LLM生成速度慢。解决考虑使用推理速度更快的模型如DeepSeek Coder或者启用API的流式输出streaming让用户能先看到部分结果。对于简单、事实性问题甚至可以尝试不调用大模型直接从检索到的片段中提取答案基于规则的或用小模型做抽取。5.4 ChromaDB相关报错Collection not found确保创建集合和查询集合时使用的name完全一致包括大小写。使用client.list_collections()检查现有集合。添加数据时内存不足如果文档量极大10万一次性生成所有向量并添加可能导致内存溢出。解决分批处理。每处理1000个块就collection.add一次。查询时距离分数异常如果发现所有距离分数都差不多比如都在0.99以上很可能是嵌入向量没有归一化。解决确保调用embed_model.encode(text, normalize_embeddingsTrue)。6. 进阶思路让智能体更“智能”当基础版本稳定后可以探索一些进阶功能让系统从“好用”变得“聪明”。SKILL的链式与图式调用目前的调度是线性的。更高级的智能体可以根据意图解析的结果动态生成一个技能调用图DAG。例如对于问题“比较一下Django和Flask在ORM方面的优劣”可以并行调用两个检索技能分别查Django ORM和Flask SQLAlchemy然后将结果交给一个“比较分析”技能进行合成。自我反思与修正在生成答案后增加一个skill_self_reflect。让LLM自己检查答案是否回答了问题是否引用了资料是否有矛盾之处如果发现问题可以触发新一轮的检索或生成。知识库的主动更新与评估智能体可以记录那些它无法回答或回答质量差的问题。定期将这些“未解决问题”报告给知识库维护者提示需要补充或更新哪些文档。甚至可以尝试让智能体根据对话自动生成知识库条目的草稿。多模态知识库不仅限于文本。可以将图片、表格、PDF中的图表通过多模态模型如Qwen-VL也编码进向量库。当用户问“请展示一下架构图”智能体可以检索出相关的图片片段。与外部工具集成将SKILL扩展到知识库之外。例如skill_execute_code可以运行代码片段验证答案skill_search_web可以在本地知识库不足时安全地搜索网络信息需谨慎处理。这个项目做到最后给我的感觉不再是简单地拼接工具而是在设计一个“思考流程”。RAG提供了记忆SKILL架构定义了思考的步骤。两者的深度融合让这个智能体在面对专业问题时终于有了一点“老师傅”的味道——知道该去哪里找资料知道怎么把资料组织成答案也知道什么时候该承认自己不会。整个过程里最花时间的往往不是写代码而是反复调整分块、优化提示词、评估效果这些细致活。但每解决一个小问题看到回答的准确度提升一点那种成就感是实实在在的。希望这份详细的实践记录能帮你少走些弯路更快地构建出属于你自己的、那个“懂行”的智能助手。