一、前言本次项 目完成了认知检索R AG的A gent的开发实现了智能路由分类、分库知识库检索、大模型增强问答的完整流程。项目采用模块化开发构建发票、维修工单两类业务知识库能够自动识别用户问题类型、匹配对应知识库内容并结合检索资料生成精准回答。本次开发重点解决了模块缺失报错、大模型忽略检索上下文两大核心问题最终实现可稳定运行的轻量化RAG系统。二、项目环境准备1. 安装依赖包打开终端执行安装命令pip install zhipuai python-dotenv2.项目目录结构所有文件必须放在同一个文件夹下缺一不可否则会报模块缺失错误所有文件必须放在同一个文件夹下缺一不可否则会报模块缺失错误main.py程序入口、测试用例rag_agent.pyAgent 核心调度类mock_retriever.py模拟向量知识库simple_router.py智能路由分类模块simple_zhipu_client.py智谱大 模型的调用封装.env存放智谱 API 个人密钥ZHIPUAI_API_KEY你的智谱API Key三、运行代码过程1. mock_retriever.py 模拟向量 知识 库采用内存字典模拟向量 数据 库支持分索引存储文档、新增数据和关键词模糊检索适合课程实验演示。class MockVectorStore: def __init__(self): # 存储结构 {索引名: [{id: , text: }]} self.store {} def add(self, index_name, doc_id, text): if index_name not in self.store: self.store[index_name] [] self.store[index_name].append({id: doc_id, text: text}) def search(self, index_name, query): # 简易模糊匹配检索 if index_name not in self.store: return [] res [] for item in self.store[index_name]: if query in item[text]: res.append(item) return res2. simple_router.py 智能路由分类利用大 模型实现语义分类自动将问题划分至发票库、维修工单库或通用库避免全库检索浪费资源。from zhipuai import ZhipuAI class SimpleRouter: def __init__(self, api_key, modelglm-4-flash): self.client ZhipuAI(api_keyapi_key) self.model model def route(self, query): prompt 你是分类器只输出三个值其中一个invoice_index / repair_index / general_index 规则 1. 包含发票、金额 → invoice_index 2. 包含工单、维修、rep → repair_index 3. 其他 → general_index 只返回单词不要多余文字 用户问题{query} resp self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt.format(queryquery)}] ) return resp.choices[0].message.content.strip()3. simple_zhipu_client.py 大 模型调用封装对智谱模型接口进行二次封装并加入强约束提示词解决大模型不读取知识 库、回复无数据的经典问题。from zhipuai import ZhipuAI class SimpleZhipuClient: def __init__(self, api_keyNone, modelglm-4-flash): self.client ZhipuAI(api_keyapi_key) self.model model def ask(self, query, contextNone): # 关键修复添加强约束Prompt解决AI拒绝读取知识库 if context: user_prompt f 你只能根据下面提供的参考文档回答禁止说无法查询、无数据库等话术。 参考文档 {context} 用户问题{query} 给出简短准确答案。 else: user_prompt query response self.client.chat.completions.create( modelself.model, messages[{role: user, content: user_prompt}] ) return response.choices[0].message.content4. rag_agent.py 核心调度中心统筹路由、检索、生成全流程统一对外提供 handle 接口同时记录运行事件与耗时结构清晰可扩展。import time from mock_retriever import MockVectorStore from simple_zhipu_client import SimpleZhipuClient from simple_router import SimpleRouter class SimpleRAGAgent: 简化的RAG Agent - 使用智谱AI def __init__(self, store, api_keyNone, modelglm-4-flash, use_smart_routerTrue): self.store store self.llm SimpleZhipuClient(api_keyapi_key, modelmodel) if use_smart_router: self.router SimpleRouter(api_keyapi_key, modelmodel) else: self.router None def handle(self, query): 处理用户查询主逻辑 result { query: query, events: [] } start time.time() # 步骤1智能路由分发判断问题所属知识库 if self.router: route self.router.route(query) else: route general_index result[events].append({event: route, route: route}) # 步骤2在对应知识库检索相关内容 hits self.store.search(route, query) result[events].append({event: retrieve, hits: len(hits)}) # 步骤3调用大模型生成回答 if hits: # 拼接检索到的前3条文档作为上下文 context \n.join(h[text] for h in hits[:3]) answer self.llm.ask(query, contextcontext) result[answer] answer result[sources] hits else: # 无匹配知识库内容直接裸问答 answer self.llm.ask(query) result[answer] answer result[sources] [] # 统计接口总耗时 result[latency] round(time.time() - start, 2) return result __all__ [SimpleRAGAgent]5. main.py 入口与测试加载环境密钥、初始化知识库、批量执行测试用例直观展示路由效果、检索结果和模型回答。import os from dotenv import load_dotenv # 加载.env文件内存储的智谱API密钥 load_dotenv() from mock_retriever import MockVectorStore from rag_agent import SimpleRAGAgent # 初始化模拟知识库划分两个索引库 store MockVectorStore() # 发票知识库数据 store.add(invoice_index, inv-001, 发票2024-001 金额 ¥1200) store.add(invoice_index, inv-002, 发票2024-002 金额 ¥3500) # 维修工单知识库数据 store.add(repair_index, rep-42, 维修工单42更换电池) store.add(repair_index, rep-43, 维修工单43更换屏幕) # 实例化RAG问答Agent agent SimpleRAGAgent( storestore, api_keyos.getenv(ZHIPUAI_API_KEY), modelglm-4-flash ) # 批量测试用例 print( * 60) print(简化的 RAG 系统测试) print( * 60) tests [ 查询发票2024-001的金额, 报修记录 rep-42, 不存在的条目, ] # 循环执行全部测试问题 for query in tests: print(f\n查询: {query}) result agent.handle(query) print(f路由: {result[events][0][route]}) print(f回答: {result[answer]}) print(f耗时: {result[latency]}s) print(\n * 60) print(✅ 测试完成!)四.踩坑记录运行报错解决方案报错 1ModuleNotFoundError: No module named mock_retriever报错截图解决在同目录新建mock_retriever.py粘贴上方完整代码保证所有自定义模块文件齐全。报错 2大模型无视检索文档一直回复 “无法查询数据库”LLM 调用时没有添加强制约束提示词解决使用本文提供的simple_zhipu_client.py内置强约束 Prompt强制模型基于文档作答。本项目当前是内存M ock知识库可替换 Chroma、FAISS 真实向量 数据库实现持久化向量存储五.项目总结本项目成功搭建了一套基于智谱大模型的模块化认知检索 R AG 问答系统实现问题智能路由、分库检索与上下文增强生成的完整流程。项目通过大模型分类路由优化了检索效率避免全库检索资源浪费并通过自定义约束提示词解决了 RAG 项目中常见的模型忽略检索内容的缺陷。系统结构清晰、低耦合、可拓展当前使用内存模拟向量库后续可替换为 FAISS、Chroma 等真实向量数据库实现持久化存储与正式落地使用。