摘要：本文基于解析如何从零构建一个支持本地知识库+网络搜索的智能问答系统。项目采用Qwen Agent框架，集成Elasticsearch向量检索，支持BM25+Embedding混合检索，并具备现代化的WebUI界面。
本文基于解析如何从零构建一个支持本地知识库+网络搜索的智能问答系统。项目采用Qwen Agent框架，集成Elasticsearch向量检索，支持BM25+Embedding混合检索，并具备现代化的WebUI界面。
书接上回：【RAG 系统架构：让 AI 学会"查资料"的魔法】
上次提到了通过本地建一个 RAG 知识库，来帮我们更好的对比保险产品，来选择更适合自己的。
接下来又遇到我好队友公司的一个业务痛点：
她公司负责管理供应商研学基地项目，积累了海量 #技术分享的资料文档（十几个 GB 的数据量） ，包括：
供应商资质文件 ：营业执照、资质证书、技术能力证明等️ 项目案例资料 ：历史项目文档、技术方案、实施报告标书模板库 ：各类招标文件、投标书模板、技术规格书市场调研报告 ：行业分析、竞争对手资料、市场趋势合同文档 ：合作协议、服务条款、价格清单等传统工作流程的问题 ：
信息查找困难 ：每次需要写标书或准备资料时，都要人工翻阅大量文档⏰ 效率低下 ：找到相关信息往往需要几个小时甚至几天重复劳动 ：相似的项目需要重复整理相似的资料信息不全面 ：人工检索容易遗漏重要信息基于这个真实的业务场景，我决定构建一个 基于 Qwen Agent 的 RAG 智能知识库系统 ，目标是：
核心功能 ：
支持PDF/Word文档的智能解析和知识提取实现高效的语义检索和关键词匹配集成网络搜索补充知识库不足提供现代化的用户交互界面支持私有化部署和数据安全预期效果 ：
检索效率提升10倍 ：从几十分钟缩短到几分钟信息召回率95%+ ：不会遗漏重要资料标书撰写效率翻倍 ：自动整理相关案例和模板知识更新实时化 ：新资料入库后立即可用技术选型 ：
框架 ：Qwen Agent（阿里通义千问智能体框架） 二次开发向量数据库 ：Elasticsearch（支持dense_vector）检索策略 ：BM25 + Embedding混合检索网络搜索 ：Tavily-mcp集成前端界面 ：Gradio WebUI（知乎风格美化）1. 成熟的智能体框架
class Assistant(FnCallAgent): def _run(self, messages, **kwargs):2. 强大的工具调用能力
原生工具注册 ： @register_tool 装饰器，简化工具开发参数验证 ：自动JSON格式验证和类型检查错误处理 ：内置异常捕获和重试机制并发支持 ：支持多工具并行调用3. 灵活的插件机制
function_list = [ {'name': 'retrieval', 'max_ref_token': 4000}, {'name': 'doc_parser', 'parser_page_size': 500}, 'code_interpreter', CustomTool] 但本身Qwen Agent知识库这块比较一般，就有二次开发的必要性哇1. 先让增强它的检索能力
class Retrieval(BaseTool):class Retrieval(BaseTool): def call(self, params, **kwargs): search_type = params.get('search_type', 'hybrid')2. 存储层重构
原生问题 ：只支持本地文件存储，无法处理GB级数据解决方案 ：集成Elasticsearch，支持分布式存储和高效检索性能提升 ：从文件系统检索升级到向量数据库检索3. 用户体验优化
原生界面 ：基础的Gradio界面，缺乏现代化设计二次开发 ：知乎风格美化，响应式布局，移动端适配交互优化 ：侧边栏功能、弹窗提示、实时反馈| 方案 | 开发周期 | 技术风险 | 功能完整性 | 维护成本 | | ---
| 从零开发 | 长 | 需要重新实现所有功能 | 高 | 低 | | 基于 LangChain | 短 | 功能丰富但复杂 | 中 | 低 | | 基于 Qwen Agent | 短 | 专注业务创新 | 低 | 中 |
1. 专注业务创新
class ESMemory: def hybrid_search(self, query, top_k=5):2. 降低技术风险
框架稳定性 ：Qwen Agent经过阿里大规模验证社区支持 ：活跃的开源社区，问题解决及时版本迭代 ：持续的功能更新和性能优化3. 提升开发效率
工具生态 ：丰富的预置工具，开箱即用文档完善 ：详细的中文文档和示例代码调试友好 ：内置日志和调试工具1. 存储层适配
class Memory(Agent): def __init__(self, memory_type='local'): if memory_type == 'es': self.es_memory = ESMemory else: self.es_memory = None2. 检索工具增强
if memory_type == 'es' and es_memory is not None: return es_memory.hybrid_search(query, top_k=top_k)else: return self.search.call(params={'query': query}, docs=docs)3. 配置管理优化
llm_cfg = { 'model': LLM_MODEL, 'model_server': LLM_BASE_URL, 'api_key': LLM_API_KEY,}1. 框架选择原则
成熟度优先 ：选择经过验证的成熟框架生态丰富 ：优先选择工具生态丰富的框架社区活跃 ：选择有活跃社区支持的框架2. 二次开发策略
渐进式改造 ：保持原有功能，逐步增强向后兼容 ：确保原有代码仍能正常工作模块化设计 ：新增功能独立封装，便于维护3. 技术债务管理
文档完善 ：详细记录修改原因和实现方式测试覆盖 ：确保修改不影响原有功能版本控制 ：清晰的分支管理和版本标签总结 ：选择二次修改 Qwen Agent 是基于"站在巨人肩膀上"的智慧，让我们能够专注业务创新，快速构建企业级 RAG 系统，同时享受成熟框架带来的稳定性和生态优势。
相比之前写的《RAG 系统架构通俗解读》文章里的项目，这个实战项目在技术实现上有了显著提升：
检索能力升级| 特性 | 之前版本 | 现有版本 | 改进效果 | | ---
| 检索策略 | 单一向量检索 | BM25+Embedding 混合检索 | 召回率提升大概10% | | 索引设计 | 单索引存储 | 双索引分离设计 | 性能提升约有3倍 | | 检索精度 | 基础相似度匹配 | 智能去重+重排序 | 准确率提升约有15% |
工程化改进文档解析 ：从基础PDF解析高质量结构化解析（也可通过MinerU来直接进行文档解析）向量化 ：支持多种embedding模型，兼容OpenAI/通义千问部署方式 ：从理论概念到私有化部署，支持企业级应用用户体验 ：从命令行到现代化WebUI，知乎风格界面网络搜索 ：集成Tavily-mcp，解决知识时效性问题多模型支持 ：在线模型API调取，以及也可替换本地模型调取，降低部署成本配置管理 ：统一的config.py配置，支持环境变量监控优化 ：检索效果监控，性能指标追踪数据规模 ：从保险文档扩展到企业级GB级数据应用场景 ：从个人保险咨询到企业标书撰写、内部资料整理等用户群体 ：从个人用户到企业团队协作核心优势总结 ：
企业级特性 ：私有化部署、数据安全、团队协作智能化程度 ：混合检索、网络搜索、智能去重用户体验 ：现代化WebUI、移动端适配、操作简单扩展性强 ：模块化设计、插件机制、多模型支持{ "mappings": { "properties": { "doc_name": {"type": "keyword"}, "content": {"type": "text"}, "chunk_id": {"type": "integer"} } }}{ "mappings": { "properties": { "content_vector": { "type": "dense_vector", "dims": 1024, "index": True, "similarity": "cosine" } } } }设计亮点 ：
双索引分离 ：BM25和向量检索使用不同索引，避免性能冲突混合检索 ：支持单一检索和混合检索模式动态配置 ：通过 memory_type 参数控制使用本地存储还是ES存储class ESMemory: def __init__(self, index=ES_INDEX, embedding_index=ES_EMBEDDING_INDEX): self.es = Elasticsearch(es_host_with_port, basic_auth=(username, password)) def hybrid_search(self, query, top_k=5): """BM25+embedding混合检索，合并去重""" bm25_results = self.search(query, top_k=top_k) embedding_results = self.embedding_search(query, top_k=top_k) return self._merge_and_deduplicate(bm25_results, embedding_results)@register_tool('retrieval')class Retrieval(BaseTool): def call(self, params, **kwargs): search_type = params.get('search_type', 'hybrid') top_k = params.get('top_k', 5) if memory_type == 'es' and es_memory is not None: if search_type == 'bm25': return es_memory.search(query, top_k=top_k) elif search_type == 'embedding': return es_memory.embedding_search(query, top_k=top_k) else: return es_memory.hybrid_search(query, top_k=top_k)检索策略对比 ：
BM25检索 ：基于TF-IDF，适合关键词精确匹配Embedding检索 ：基于语义相似度，适合语义理解Hybrid检索 ：结合两者优势，提升召回率和准确率mcp_tools = [{ "mcpServers": { "tavily-mcp": { "command": "npx", "args": ["-y", "tavily-mcp@0.1.4"], "env": {"TAVILY_API_KEY": TAVILY_API_KEY} } }}]设计思路 ：
当本地知识库无法回答时，自动调用网络搜索搜索结果与本地知识库结果智能融合支持实时信息补充，解决知识时效性问题class Assistant(FnCallAgent): def _prepend_knowledge_prompt(self, messages, knowledge=''): """将检索到的知识注入到对话上下文中""" if not knowledge: *_, last = self.mem.run(messages=messages) knowledge = last[-1][CONTENT] knowledge_prompt = self._format_knowledge(knowledge) messages = self._inject_knowledge(messages, knowledge_prompt) return messages智能体特色 ：
ReAct范式 ：推理(Reason) + 行动(Act)的循环模式工具调用 ：支持检索、解析、网络搜索等多种工具上下文管理 ：智能管理对话历史和知识注入def parse_mineru_json(json_path): """解析MinerU结构化JSON数据""" with open(json_path, 'r', encoding='utf-8') as f: data = json.load(f) texts = if isinstance(data, dict): if 'pages' in data: for page in data['pages']: txt = page.get('text', '') if txt.strip: texts.append(txt) return textsdef split_text(text, chunk_size=500): """智能文本分块，保持语义完整性""" chunks = return chunksdef get_embedding(text: str, client=None) -> list: """生成文本embedding向量""" if client is None: client = OpenAI( api_key=DASHSCOPE_API_KEY, base_url=DASHSCOPE_BASE_URL ) response = client.embeddings.create( model=EMBEDDING_MODEL, input=text, dimensions=EMBEDDING_DIM, encoding_format="float" ) return response.data[0].embedding第三步：智能检索def hybrid_search(self, query, top_k=5): """混合检索实现""" bm25_results = self.search(query, top_k=top_k) query_vector = get_embedding(query) embedding_results = self.embedding_search(query_vector, top_k=top_k) all_results = bm25_results + embedding_results merged = self._merge_and_deduplicate(all_results, top_k) return mergeddef _format_knowledge_to_prompt(self, knowledge_results): """将检索结果格式化为提示词""" snippets = for result in knowledge_results: snippet = KNOWLEDGE_SNIPPET.format( source=result['doc_name'], content=result['content'] ) snippets.append(snippet) return '\n\n'.join(snippets).chat-container { background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%); border-radius: 20px; box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); backdrop-filter: blur(10px);}.message-card { background: rgba(255, 255, 255, 0.9); border-radius: 15px; padding: 20px; margin: 10px 0; box-shadow: 0 4px 15px rgba(0, 0, 0, 0.08); }交互功能with gr.Column(scale=1): search_btn = gr.Button("搜索") kb_btn = gr.Button("知识库") fav_btn = gr.Button("收藏") history_btn = gr.Button("历史") for btn in [search_btn, kb_btn, fav_btn, history_btn]: btn.click(lambda: gr.Info("暂未实现，敬请期待"), outputs=None)| 检索方式 | 召回率 | 准确率 | 响应时间 | | ---
| BM25 | 85% | 78% |
看到项目部署启动后，在好队友实际使用一段时间后，得出的反馈是大大的有帮助，减少了因这事而加班的时间。
她好才是真的好哇
吹了上面这么多，又到固定环节，大佬们可以去瞄一眼哇
来源：墨码行者