一、前言

本次项 目完成了认知检索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 等真实向量数据库，实现持久化存储与正式落地使用。