🌟 SAG

SQL驱动的RAG引擎 · 查询时自动构建知识图谱

The SQL-Driven Smart Auto Graph Engine

[![zread](https://img.shields.io/badge/Ask_Zread-_.svg?style=flat-square&color=00b0aa&labelColor=000000&logo=data%3Aimage%2Fsvg%2Bxml%3Bbase64%2CPHN2ZyB3aWR0aD0iMTYiIGhlaWdodD0iMTYiIHZpZXdCb3g9IjAgMCAxNiAxNiIgZmlsbD0ibm9uZSIgeG1sbnM9Imh0dHA6Ly93d3cudzMub3JnLzIwMDAvc3ZnIj4KPHBhdGggZD0iTTQuOTYxNTYgMS42MDAxSDIuMjQxNTZDMS44ODgxIDEuNjAwMSAxLjYwMTU2IDEuODg2NjQgMS42MDE1NiAyLjI0MDFWNC45NjAxQzEuNjAxNTYgNS4zMTM1NiAxLjg4ODEgNS42MDAxIDIuMjQxNTYgNS42MDAxSDQuOTYxNTZDNS5zMTUwMiA1LjYwMDEgNS42MDE1NiA1LjMxMzU2IDUuNjAxNTYgNC45NjAxVjIuMjQwMUM1LjYwMTU2IDEuODg2NjQgNS5zMTUwMiAxLjYwMDEgNC45NjE1NiAxLjYwMDFaIiBmaWxsPSIjZmZmIi8%2BCjxwYXRoIGQ9Ik00Ljk2MTU2IDEwLjM5OTlIMi4yNDE1NkMxLjg4ODEgMTAuMzk5OSAxLjYwMTU2IDEwLjY4NjQgMS42MDE1NiAxMS4wMzk5VjEzLjc1OTlDMS42MDE1NiAxNC4xMTM0IDEuODg4MSAxNC4zOTk5IDIuMjQxNTYgMTQuMzk5OUH0NC45NjE1NiM1LjMxNTUwMiAxNC4zOTk5OSA1LjYwMTU2IDEzLjc1OTlVjExLjAzOTlDNS5YjYwMTU2IDEwLjYODY0IDUuMzE1MDIgMTAuMzk5OSA0Ljk6YjE1NiAxMC4zOTOTLaIiBmaWxsPSIjZmZmIi8%2BCjxwYXRoIGQ9Ik0xMy43NTg0IDEuNjAwMUgxMS.0M384C10.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.685IDE.68......

方法二：Python SDK

import asyncio
from sag import SAGEngine
from sag.modules.load.config import LoadBaseConfig
from sag.modules.extract.config import ExtractBaseConfig
from sag.modules.search.config import SearchBaseConfig

async def main():
    # 初始化
    engine = SAGEngine(source_config_id="my-project")
    
    # 加载文档
    await engine.load(LoadBaseConfig(
        type="path",
        origin=["./docs/article.md"],
        background="技术文档"
    ))
    
    # 提取事件
    await engine.extract(ExtractBaseConfig(
        parallel=True,
        background="AI大模型文档"
    ))
    
    # 智能检索
    result = await engine.search(SearchBaseConfig(
        query="如何优化大模型推理速度？",
        depth=2,
        top_k=10
    ))
    
    # 查看结果
    for event in result.events:
        print(f"[{event.score:.2f}] {event.title}")
        print(f"  {event.summary}\n")

asyncio.run(main())

方法三：Web界面

访问 http://localhost:3000

1. 上传文档：拖拽 Markdown、PDF、HTML
2. 自动处理：系统自动加载→提取→索引
3. 智能搜索：输入自然语言查询
4. 查看结果：浏览事件、线索图谱、来源

---

🌐 开源版 vs 完整版

功能对比

功能	开源基础版	完整版
核心引擎	✅ 完整开源	✅ 相同引擎
文档加载	✅ 本地文件	✅ 多种信息源
数据源	✅ 手动上传	✅ 自动更新
内容发布	❌	✅ 一键生成文章/报告
协作	❌ 单用户	✅ 团队 + 权限管理
高级功能	❌	✅ 智能推荐 + 自动摘要
云服务	❌ 需自建	✅ 开箱即用
支持	社区	专业技术团队

为什么开源基础版？

我们相信：

• 🌍 技术共享：核心算法应该被更多人使用和改进
• 🔧 灵活部署：企业可自建私有化部署
• 🤝 社区驱动：开源社区的反馈让产品更好
• 💡 创新激励：开发者可基于SAG构建自己的应用

什么时候用完整版？

• 需要自动网页追踪和信息流管理
• 想接入更多信息源
• 需要团队协作和权限管理
• 希望零部署，开箱即用
• 需要专业技术支持

体验完整版：https://zleap.ai

---

📖 深入学习

这一节是给对算法细节感兴趣的开发者和研究人员的简版说明。

🧠 核心理念：Event & Natural Language Vector

SAG 的底层思想可以用两句话概括：

• 事件原子化（Event Atomization）
  不再按字符/Token 长度“机械切块”，而是将文档转化为一个个 语义完整、彼此独立 的“事件 (Event)”。
• 自然语言向量（Natural Language Vector）
  不只把整段文本编码成向量，而是为每个 Event 抽取多维实体：时间、地点、人物、动作、话题、标签…
  它们组成了一个“由自然语言实体构成的向量”。

关键洞察：

• Event 是 原子知识单元
• Entity 是 事件的实体维度
• 事件之间的关系不提前计算，而是 在查询时动态计算

🧮 三阶段搜索算法（Recall → Expand → Rerank）

1. Recall：实体驱动召回（Entity-Based Recall）

目标：从查询语句出发，找到一批高度相关的 实体 + 事件。

• 步骤概要：
  • LLM 解析查询：抽取结构化实体（TOPIC、ACTION、PERSON…）
  • 向量检索实体：在实体向量空间中搜索
  • 用实体查事件（SQL）：通过实体 ID 反查事件
  • 事件向量检索：直接在 Event 向量上查
  • 交集过滤 + 权重反向传播：兼顾语义相似度与实体匹配

2. Expand：基于 BFS 的多跳扩展

目标：通过“共享实体模式”在事件-实体空间做 多跳搜索，找到更深层的相关信息。

• 做法：

  • 将高权重实体视作当前“前沿层”
  • 用这些实体在 SQL 中查找新事件
  • 对新事件计算相似度和权重，并将权重反向传播给新实体
  • 只保留“新出现”的实体，形成下一跳前沿层
  • 过程中带有 权重衰减 + 去重，无新实体时自动收敛

• 特性：

  • 与“六度空间理论”类似：任意两个事件，往往可以通过少量中间实体连接
  • 深度 2 通常在 精度 / 召回 / 延迟 上达到最优平衡

• 实体权重示意公式：

$$W(k_i) = \sum_{e_j \in E} \left[ W_{e2}(e_j) \times \frac{count(k_i, e_j)}{\ln(1 + step_{ij})} \right]$$

3. Rerank：基于方向性 PageRank 的排序

在 Recall + Expand 得到的事件子图上，SAG 构建隐式图并运行 加权 PageRank：

• 节点：事件 e
• 有向边：共享实体关系，边权由实体权重 + 频次决定：

$$ W(e_i \rightarrow e_j) = \sum_{k \in (e_i \cap e_j)} W_{\text{entity}}(k) \cdot \ln(1 + \text{freq}(k, e_j)) $$

• PageRank 迭代：

$$ \mathrm{PR}(e_j) = \frac{1-d}{N} + d \sum_{e_i \in \mathrm{In}(e_j)} \mathrm{PR}(e_i) \cdot \frac{W(e_i \rightarrow e_j)}{\sum\limits_k W(e_i \rightarrow e_k)} $$

• 最终综合评分（四因子加权）：

$$ S(e) = \alpha \cdot \mathrm{PR}(e) + \beta \cdot \mathrm{Sim}(Q, e) + γ \cdot \mathrm{EntityScore}(e) + δ \cdot \mathrm{TimeDecay}(e) $$

其中典型配置：α=0.4，β=0.3，γ=0.2，δ=0.1。

---

🤝 社区与贡献

加入我们

• 🌐 官网：https://zleap.ai
• 💬 Discord：加入讨论
• 📧 邮箱：contact@zleap.ai
• 🐦 Twitter：@ZleapAI

如何贡献

# 1. Fork并克隆
git clone https://github.com/your-name/SAG.git

# 2. 创建分支
git checkout -b feature/amazing-feature

# 3. 提交更改
git commit -m "feat: add amazing feature"

# 4. 推送
git push origin feature/amazing-feature

# 5. 开启 Pull Request

Commit规范：feat: | fix: | docs: | refactor: | test: | chore:

贡献者墙

---

🙏 致谢

• 感谢所有贡献者
• 特别感谢302.AI的算力支持

---

📄 许可证

本项目采用 Apache-2.0 License

---

⭐ Star History

---

---

git clone https://github.com/Zleap-AI/SAG.git
cd SAG

cp .env.example .env

python scripts/download_nltk_data.py

docker compose up -d

docker compose logs -f

import asyncio
from sag import SAGEngine
from sag.modules.load.config import LoadBaseConfig
from sag.modules.extract.config import ExtractBaseConfig
from sag.modules.search.config import SearchBaseConfig

async def main():
    # 1. 初始化引擎
    engine = SAGEngine(source_config_id="my-first-project")
    
    # 2. 加载本地文档
    await engine.load(LoadBaseConfig(
        type="path",
        origin=["./docs/tech_report.md"], # 替换为您的文件路径
        background="技术分析报告"
    ))
    
    # 3. 提取语义事件 (构建图谱)
    await engine.extract(ExtractBaseConfig(
        parallel=True,
        background="AI 大模型技术文档"
    ))
    
    # 4. 执行智能检索
    result = await engine.search(SearchBaseConfig(
        query="如何优化大模型推理速度？",
        depth=2,      # 多跳扩展深度
        top_k=10      # 返回结果数量
    ))
    
    # 5. 输出结果
    for event in result.events:
        print(f"[{event.score:.2f}] {event.title}")
        print(f"  {event.summary}\n")

if __name__ == "__main__":
    asyncio.run(main())

pip install -e .  # 首次运行需安装依赖
python test_sag.py