企业官网建设流程全解析

一、前言

本次项 目完成了认知检索R AG的A gent的开发，实现了智能路由分类、分库知识库检索、大模型增强问答的完整流程。项目采用模块化开发，构建发票、维修工单两类业务知识库，能够自动识别用户问题类型、匹配对应知识库内容，并结合检索资料生成精准回答。本次开发重点解决了模块缺失报错、大模型忽略检索上下文两大核心问题，最终实现可稳定运行的轻量化RAG系统。

二、项目环境准备

1. 安装依赖包

打开终端执行安装命令：pip install zhipuai python-dotenv

2.项目目录结构

所有文件必须放在同一个文件夹下，缺一不可，否则会报模块缺失错误：

所有文件必须放在同一个文件夹下，缺一不可，否则会报模块缺失错误：

• main.py：程序入口、测试用例
• rag_agent.py：Agent 核心调度类
• 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 res

2. simple_router.py 智能路由分类

利用大 模型实现语义分类，自动将问题划分至发票库、维修工单库或通用库，避免全库检索浪费资源。

from zhipuai import ZhipuAI class SimpleRouter: def __init__(self, api_key, model="glm-4-flash"): self.client = ZhipuAI(api_key=api_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( model=self.model, messages=[{"role": "user", "content": prompt.format(query=query)}] ) return resp.choices[0].message.content.strip()

3. simple_zhipu_client.py 大 模型调用封装

对智谱模型接口进行二次封装，并加入强约束提示词，解决大模型不读取知识 库、回复无数据的经典问题。

from zhipuai import ZhipuAI class SimpleZhipuClient: def __init__(self, api_key=None, model="glm-4-flash"): self.client = ZhipuAI(api_key=api_key) self.model = model def ask(self, query, context=None): # 关键修复：添加强约束Prompt，解决AI拒绝读取知识库 if context: user_prompt = f""" 你只能根据下面提供的参考文档回答，禁止说无法查询、无数据库等话术。 参考文档： {context} 用户问题：{query} 给出简短准确答案。 """ else: user_prompt = query response = self.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": user_prompt}] ) return response.choices[0].message.content

4. 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_key=None, model="glm-4-flash", use_smart_router=True): self.store = store self.llm = SimpleZhipuClient(api_key=api_key, model=model) if use_smart_router: self.router = SimpleRouter(api_key=api_key, model=model) 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, context=context) 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( store=store, api_key=os.getenv("ZHIPUAI_API_KEY"), model="glm-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("✅ 测试完成!")

四.踩坑记录：运行报错解决方案

报错 1：ModuleNotFoundError: No module named 'mock_retriever'

报错截图：

解决：在同目录新建mock_retriever.py，粘贴上方完整代码，保证所有自定义模块文件齐全。

报错 2：大模型无视检索文档，一直回复 “无法查询数据库”

LLM 调用时没有添加强制约束提示词

解决：使用本文提供的simple_zhipu_client.py，内置强约束 Prompt，强制模型基于文档作答。

本项目当前是内存M ock知识库，可替换 Chroma、FAISS 真实向量 数据库，实现持久化向量存储

五.项目总结

本项目成功搭建了一套基于智谱大模型的模块化认知检索 R AG 问答系统，实现问题智能路由、分库检索与上下文增强生成的完整流程。项目通过大模型分类路由优化了检索效率，避免全库检索资源浪费，并通过自定义约束提示词解决了 RAG 项目中常见的模型忽略检索内容的缺陷。系统结构清晰、低耦合、可拓展，当前使用内存模拟向量库，后续可替换为 FAISS、Chroma 等真实向量数据库，实现持久化存储与正式落地使用。