Prithvi4216/nexus-ops-multi-agent-rag

# 多智能体 RAG 平台  *(内部品牌为 **Nexus-Ops Cognitive Platform** — 见下方截图)* 这是一个基于 **LangGraph** 构建的多智能体检索增强生成系统，通过 supervisor → guardrail → specialist-agent 的 pipeline 路由用户查询。该系统基于带有语义缓存、离线 RAGAS 评估以及全执行链路可观测性的本地索引知识库。 该项目完全独立端到端构建 —— 包括摄入 pipeline、向量存储、LangGraph 编排、Streamlit UI、测试套件和评估工具—— 完全运行在一台 **4GB RAM / 无 GPU** 的机器上，并使用免费层级的 API。 ## 功能介绍 提出一个问题 → 系统会决定*如何*回答它，而不仅仅是回答*什么*： - 在运行任何 LLM 调用之前，**semantic cache** 会检查是否已经回答过足够相似的问题。 - **supervisor node** 会对意图进行分类（代码 / 散文 / 对话 / 学习笔记），并将其路由到合适的专业智能体和模型。 - **guardrail** 会在任何查询到达智能体之前，对其进行 prompt-injection / 越狱尝试的筛查。 - 活跃的智能体会针对每次查询决定是否真的需要知识库，并可以根据问题的形态（广泛搜索、精确概念查找或并排对比）调用最多 3 种不同的检索工具。 - 每一轮对话都会被记录、进行成本/延迟评分，并支持通过真实的 LLM 评估器进行离线 **RAGAS** 评估（faithfulness、answer relevancy、context precision）。 ## 架构 **离线 pipeline** (`main.py` — 运行一次，构建知识库)：  **实时 agent graph** (`src/agent.py` — 每次查询对应一个 LangGraph 状态机)：  ``` [START] → cache_lookup ──HIT──→ [END] (cached answer, zero LLM calls) │ MISS ▼ supervisor → guardrail ──BLOCKED──→ [END] │ ALLOWED ▼ { retrieval_agent | code_agent | direct_answer | notes_agent } │ retrieval_agent / notes_agent ⟲ tools (loop on tool_calls) │ observability → [END] ``` ## 知识库 演示实例基于 6 本 O'Reilly AI/ML 教科书进行了索引（总共约 5,400 个 chunks）： - *AI Engineering* - *Applied Machine Learning and AI for Engineers* - *Generative Deep Learning* - *Hands-On Large Language Models* - *Hands-On Generative AI with Transformers and Diffusion* - *NLP with Transformer Models* **本 repo 中不包含这些 PDF**（受版权保护的材料）—— 摄入 pipeline (`src/ingestion.py`) 适用于任何 PDF 集合。将你自己的技术 PDF 放入 `data/` 并运行 `main.py` 即可构建你自己的知识库；具备布局感知的 Markdown 提取、chunk 分类（代码/表格/文本）以及章节层级标记功能适用于任何格式良好的 PDF，而不仅限于这些特定的书籍。 ## 截图 **基于知识库检索并合成的有据可依的回答：**  **同一问题的执行路径，逐个节点重放：**  **确切检索到了哪些 chunks 以及原因（来源、章节、余弦得分）：**  **拒绝幻觉，实事求是 —— 当知识库中没有相关内容时，智能体在基于通用知识回答之前会主动说明：**  **离线 RAGAS 评估 —— 真实的 LLM 评估器打分，而非启发式算法：**   **交互式 tokenizer（与智能体用于上下文计数的 `cl100k_base` 编码相同）：**  ## 关键工程决策 有几个值得一提的地方 —— 这些源于真实的约束条件，而非教程的默认设置： - **为什么特别选择 LangGraph。** `src/agent.py` 中的 graph 并非线性 pipeline —— `cache_lookup` 和 `guardrail` 都可以直接短路直达 `END`，`retrieval_agent`/`notes_agent` 在每轮对话中会通过 `tools` 循环 0 到 N 次，四个不同的专业智能体会汇聚回一个 `observability` 节点，并且所有这些都在读写同一个强类型的 `AgentState`。基于链的工具（LangChain 的 LCEL）和固定的对话模式（CrewAI 的顺序/分层 crews，AutoGen 的点对点聊天循环）无法原生表达基于共享状态的任意条件循环 —— 你最终依然需要在它们之上手动实现那些路由逻辑。LangGraph 将节点、条件边缘和共享状态作为一等公民，因此 `cache_router`、`guardrail_router` 和 `tools_return_router` 只是 graph 定义中的边缘，而不是自定义的控制流。 - **解决 Windows/Python 3.13 原生 DLL 崩溃问题。** `chromadb`（通过 opentelemetry→protobuf）、`fitz`/PyMuPDF 和 `torch`（通过 `langchain_huggingface`）在此技术栈中会相互冲突其原生库。通过严格的导入顺序（`langchain_text_splitters` → `chromadb` → `fitz`）以及完全弃用 `voyageai` 的官方 SDK 解决了该问题——它会在模块级别导入 PIL，这也会导致崩溃——转而使用原生的 `requests` 调用 Voyage AI 的 REST API（零原生 DLL，效果相同）。 - **自适应 RAM 的批处理。** PDF 摄入和向量存储索引都会在启动时读取可用的系统 RAM，并据此调整其批处理大小（预留 30-40% 的操作系统余量），而不是使用固定的批处理大小从而在受限硬件上导致 OOM。这是在 4GB 内存且无 GPU 的机器上开发和测试的——这不是理论上的担忧。 - **多模型路由，而非单一 LLM。** 一个轻量级的关键词分类器（带有可选的 LLM 路由回退机制）会在用于深度推理/RAG 合成的 120B 模型与用于代码/对话回合的 20B 模型之间进行选择——从而避免在处理“hello”这样的请求时付出 120B 模型的延迟代价。 - **三阶段 prompt-injection 防护栏。** 精确匹配黑名单 → 软信号关键词关卡 → 用于伪装/重述尝试的 LLM 分类器回退——经过调优，使得大多数普通查询永远不会触及（更慢、更昂贵的）LLM 阶段。 - **带有非对称 embedding 校正的语义答案缓存。** Voyage 的 embedding 在“文档”和“查询”编码模式之间是有意设定为非对称的；缓存必须强制双方通过相同的编码路径，以便完全相同的重复问题实际上能达到约 0 距离，而不是永久性的假未命中。 - **自动 provider 故障转移。** 如果在对话中途触及了 Groq 的速率限制，智能体会透明地在 Gemini 上重试同一轮次，而不是导致请求失败。 - **离线 RAGAS 评估**，不是生搬硬套的演示指标——faithfulness、answer relevancy 和 context precision 都由真实的 LLM 评估器针对实际记录的交互历史进行评分，并采用 token 预算截断，以确保评估器自身的调用不会突破 Groq 的 TPM 限制。 ## 测试 **涵盖 13 个文件的 81 个测试用例**，覆盖了 graph 拓扑、路由逻辑、guardrail 行为、缓存正确性、token 处理以及并发下的线程安全——而不仅仅是正常路径检查。 | 文件 | 重点 | |---|---| | `test_agent_graph.py` | LangGraph 拓扑健全性检查（节点/边缘连接正确） | | `test_routing_matrix.py` → `test_routing.py` | 完整的意图分类矩阵（12 个测试）—— 每一个覆盖前缀和启发式路径 | | `test_guardrail.py` | 黑名单、软信号关卡，以及针对伪装注入尝试的**实时** LLM 分类器回退 | | `test_semantic_cache.py` | 缓存命中/未命中正确性、升级短语绕过、**并发写入安全性**（8 个线程） | | `test_ingestion.py` | 内容分类（代码/表格/文本）、线程安全去重（20 个并发线程）、章节层级 | | `test_intent_switching.py` | **实时** —— 连续的意图切换 (CODE→PROSE→CONVERSATIONAL) 不会在 Groq 的绑定工具验证中出错 | | `test_extreme_cases.py` | **实时** —— 内存摘要确实会在超过阈值时触发；针对真实的 Groq 进行完整的端到端轮次测试 | | `test_token_guard.py` | 回归测试覆盖了一个真实 bug，即 Gemini 的列表形态响应曾被静默少算 | | `test_database.py`, `test_usage_tracker.py`, `test_interaction_logger.py`, `test_tool_error_handling.py`, `test_app_helpers.py`, `test_edge_cases.py` | Embedding 批处理数学计算、线程安全的使用日志记录、日志 I/O 正确性、结构化工具错误处理、Streamlit 层辅助程序、对抗性/unicode/空输入处理 | 一部分测试（上面标记为 `live` 的部分）会调用真实的 Groq/Voyage API，而不是对它们进行 mocking —— 这是有意为之，旨在捕捉 mocking 会掩盖的集成问题（例如上面提到的 Gemini 内容形态 bug 就是通过这种方式发现的）。 完整的逐项测试细目：见 [`TESTS.md`](TESTS.md)。 ``` pytest # full suite pytest -k "not live" # skip tests that hit real APIs ``` ## 技术栈 | 层级 | 工具 | |---|---| | 编排 | LangGraph (状态机、条件路由、工具循环) | | LLMs | Groq (`openai/gpt-oss-120b`, `openai/gpt-oss-20b`) + Google Gemini (`gemini-2.5-flash`, `gemini-3.1-flash-lite`) 作为免费层级的备用方案 | | Embeddings | Voyage AI `voyage-4-lite` (1024-维, 通过原生 REST 实现 — 见上文) | | 向量存储 | ChromaDB (本地, HNSW 余弦相似度) | | PDF → 结构化文本 | PyMuPDF / `pymupdf4llm` (具备布局感知的 Markdown 提取) | | UI | Streamlit | | 评估 | RAGAS (faithfulness / answer relevancy / context precision) | | 测试 | pytest, 13 个模块, 81 个测试用例 | ## 设置 **要求：** Python 3.13，以及 Groq、Voyage AI 和（可选）Google AI Studio 的免费层级 API 密钥。 ``` git clone https://github.com/Prithvi4216/nexus-ops-multi-agent-rag.git cd nexus-ops-multi-agent-rag # 使用 uv（推荐 — 包含 uv.lock） uv sync # 或使用普通 pip pip install -r Requirements.txt ``` 将 `.env_sample` 复制到 `.env` 并填入你的密钥： ``` GROQ_API_KEY=your_key_here VOYAGE_API_KEY=your_key_here GOOGLE_API_KEY=your_key_here # optional — enables Gemini fallback/manual selection ``` 将任何你想要索引的 PDF 放入 `data/`（见上文的**知识库**）。 ``` python main.py # runs ingestion + indexing if not already done, then launches the CLI agent # 或 streamlit run app.py # full UI — chat, tokenizer, system flow, live evaluation ``` 在聊天会话结束后运行离线 RAGAS 评估： ``` python evaluation.py --last 10 ``` ## 已知局限 - 仅在 **Windows 上开发和测试** —— 上述提到的导入顺序修复是针对 Windows/Python-3.13 的特定问题；在 Linux/Mac 上的行为未经测试（可能完全不需要这些变通方案）。 - 基于并针对**免费层级速率限制**（Groq、Voyage、Gemini）构建和调优 —— 重试/故障转移逻辑正是专门为此而存在的。 - 会话状态 (`data/sessions/`) 和关闭（"exit"/"quit"）命令是为**单本地用户**使用而设计的，而非多用户部署 —— 相关代码中已明确指出了这一点。 ## 为什么开发这个项目 独立构建 —— 没有参加课程，也没有参加训练营，完全通过官方文档和动手实践拼凑而成 —— 专门为了在实践中深入掌握我在工作中无法接触到的 Agentic/RAG 系统。目标是构建一个具备真实生产环境关注点（可观测性、guardrails、评估、资源约束）的项目，而不是一个单文件的教程聊天机器人。 ## 文档 每个核心模块都附有一份深入的文档，涵盖了其设计原理，而不仅仅是 API： | 模块 | 深入解析 | |---|---| | `main.py` | [`main_py.md`](main_py.md) | | `app.py` | [`app_py.md`](app_py.md) | | `evaluation.py` | [`Evaluation_py.md`](Evaluation_py.md) | | `interaction_logger.py` | [`interaction_logger_py.md`](interaction_logger_py.md) | | `usage_tracker.py` | [`usage_tracker_py.md`](usage_tracker_py.md) | | `src/agent.py` (核心 graph) | [`src/agent(core)_py.md`]() | | `src/agent.py` (工具) | [`src/_py_tools.md`](src/agent_py_tools.md) | | `src/database.py` | [`src/database_py.md`](src/database_py.md) | | `src/ingestion.py` | [`src/ingestion_py.md`](src/ingestion_py.md) | | `src/semantic_cache.py` | [`src/semantic_cache_py.md`](src/semantic_cache_py.md) | 完整的测试细目：[`TESTS.md`](TESTS.md)。 ## License 本代码出于作品集和审查目的公开。**保留所有权利** —— 未经许可，请勿将此工作作为您自己的工作重复使用、重新分发或展示。如果您对某个组件的工作原理感到好奇，欢迎讨论具体细节 —— 请与我联系。 ## 联系方式 - LinkedIn: [linkedin.com/in/prithviraj-chouhan](https://www.linkedin.com/in/prithviraj-chouhan/) - GitHub: [github.com/Prithvi4216](https://github.com/Prithvi4216)

标签：Kubernetes, LangGraph, PyRIT, RAGAS评估, 人工智能, 多智能体系统, 大语言模型应用, 安全规则引擎, 检索增强生成, 用户模式Hook绕过, 语义缓存, 逆向工具