python-nlql 0.1.1

A SQL-like query language for unstructured text and RAG retrieval

NLQL (Natural Language Query Language)

为非结构化文本和 RAG 检索设计的类 SQL 查询语言。

NLQL 是一个中间件层，旨在将 SQL 的确定性逻辑与 NLP 的模糊语义能力相结合。它不存储数据，而是充当 查询转换器（Query Translator） 或 内存处理器（In-Memory Processor），允许开发者用统一的语法查询纯文本、向量数据库（如 ChromaDB, FAISS）或混合数据源。

English

---

🌟 核心特性

• 语义与逻辑的融合：在一个语句中同时支持布尔逻辑 (AND, OR, NOT)、关键词匹配 (MATCH) 和 向量语义匹配 (SIMILAR_TO)。
• RAG 原生设计：支持动态粒度（Chunk, Sentence, Span），提供上下文窗口查询，专为检索增强生成（RAG）场景优化。
• 高度可扩展：支持用户重载操作符、注册自定义函数（如 NOW()）、定义新的文本切割策略。

---

🚀 快速开始

from nlql import NLQL
from nlql.adapters import MemoryAdapter

# 1. 初始化 (使用显式适配器模式)
adapter = MemoryAdapter()
adapter.add_text(
    "AI Agent architecture involves planning, memory, and tool use.",
    {"status": "published", "topic": "AI Agents"}
)
nlql = NLQL(adapter=adapter)

# 2. 执行查询
# 目标：查找关于"AI Agent"的、且不在"Draft"状态的段落，
# 并按相关度排序，返回结果及其后的一句话作为上下文。
query = """
SELECT SPAN(SENTENCE, window=1)
WHERE
    SIMILAR_TO("AI Agent architecture") > 0.8
    AND META("status") != 'draft'
    AND (MATCH("planning") OR MATCH("memory"))
ORDER BY
    SIMILARITY DESC
LIMIT 5
"""

results = nlql.execute(query)

for res in results:
    print(f"[{res.metadata['similarity']:.2f}] {res.content}")

---

📖 语法指南

NLQL 的语法基于标准 SQL，但针对文本检索进行了语义增强。

1. SELECT - 查询粒度 (Text Units)

除了返回原始文档，NLQL 支持动态切分和上下文提取。

• SELECT DOCUMENT: 返回完整文档。
• SELECT CHUNK: 返回向量库中存储的基础切片（RAG 标准）。
• SELECT SENTENCE: 从内容中动态提取句子。
• SELECT SPAN(<Unit>, window=N): 返回目标单位及其前后 N 个单位的上下文。
  • 例: SELECT SPAN(SENTENCE, window=1) (返回目标句 + 前一句 + 后一句)。

2. WHERE - 过滤条件

支持复杂的布尔逻辑组合。

基础与语义操作符

操作符/函数	描述	示例
MATCH("text")	精确文本匹配	MATCH("API Error")
SIMILAR_TO("text")	向量语义相似度 (返回 0-1 分数)	SIMILAR_TO("login failed") > 0.75
CONTAINS("text")	子串包含 (大小写不敏感)	CONTAINS("Error 500")
META("key")	元数据字段访问	META("author") == "admin"

内置辅助函数

函数	描述
LENGTH()	内容长度
NOW()	当前时间戳 (用于比较时间)
COUNT("text")	统计某词出现的次数

3. ORDER BY - 排序

支持按语义分数或元数据排序。

• ORDER BY SIMILARITY DESC: 按 SIMILAR_TO 计算出的分数降序排列（默认）。
• ORDER BY META("date") DESC: 按时间排序。
• ORDER BY LENGTH() ASC: 按文本长度排序。

---

🛠️ 扩展与重载 (Extensibility)

NLQL 的核心哲学是让开发者定义逻辑。你可以通过装饰器轻松扩展功能。

NLQL 支持两种注册方式：

• 全局注册：使用 @register_* 装饰器，所有 NLQL 实例共享
• 实例级注册：使用 nlql.register_*() 方法，仅对特定实例生效

1. 自定义 Embedding Provider

你可以使用自己的 Embedding 模型替代默认的 all-MiniLM-L6-v2。

from nlql import register_embedding_provider

# 使用 OpenAI embeddings
@register_embedding_provider
def openai_embedding(texts: list[str]) -> list[list[float]]:
    """Custom embedding using OpenAI API."""
    import openai
    response = openai.Embedding.create(
        input=texts,
        model="text-embedding-ada-002"
    )
    return [item["embedding"] for item in response["data"]]

# 现在所有查询都会使用 OpenAI embeddings

2. 注册自定义操作符

扩展 WHERE 子句中可用的操作符。

from nlql import register_operator
import re

@register_operator("HAS_EMAIL")
def has_email_address(text: str) -> bool:
    """Check if text contains an email address."""
    return bool(re.search(r'[\w\.-]+@[\w\.-]+', text))

@register_operator("HAS_URL")
def has_url(text: str) -> bool:
    """Check if text contains a URL."""
    return bool(re.search(r'https?://[^\s]+', text))

# 使用: WHERE HAS_EMAIL(content) AND HAS_URL(content)

3. 注册自定义函数

扩展可在 WHERE 或 ORDER BY 中使用的函数。

from nlql import register_function
from datetime import datetime, timedelta

@register_function("days_ago")
def days_ago(days: int) -> datetime:
    """Get datetime N days ago."""
    return datetime.now() - timedelta(days=days)

@register_function("word_count")
def word_count(text: str) -> int:
    """Count words in text."""
    return len(text.split())

# 使用: WHERE META("created_at") > days_ago(7) AND word_count(content) > 100

4. 自定义切割器 (Splitter)

适配不同语言或特殊格式的文本切分。

from nlql import register_splitter

@register_splitter("SENTENCE")
def split_german_text(text: str) -> list[str]:
    """Custom sentence splitter for German text."""
    import nltk
    return nltk.sent_tokenize(text, language='german')

# 使用: SELECT SENTENCE WHERE ...

5. 实例级注册（Instance-Level Registration）

为不同的 NLQL 实例注册不同的实现，适用于多租户应用或 A/B 测试。

from nlql import NLQL
from nlql.adapters import MemoryAdapter

# 创建两个独立的 NLQL 实例
nlql1 = NLQL(adapter=MemoryAdapter())
nlql2 = NLQL(adapter=MemoryAdapter())

# 为每个实例注册不同的函数实现
@nlql1.register_function("SCORE")
def score_v1(text: str) -> float:
    return len(text) / 100.0  # 简单评分

@nlql2.register_function("SCORE")
def score_v2(text: str) -> float:
    return len(set(text.split())) / 50.0  # 基于唯一词数评分

# 两个实例使用不同的 SCORE 函数实现
results1 = nlql1.execute("SELECT CHUNK WHERE SCORE(content) > 0.5")
results2 = nlql2.execute("SELECT CHUNK WHERE SCORE(content) > 0.5")

💡 详细文档：查看 docs/user-guide/extensibility.md 了解完整的扩展性功能，或运行 examples/instance_registry_demo.py 查看实例级注册的完整示例。

---

🏗️ 架构原理

NLQL 采用显式适配器模式，执行流程分为以下阶段：

1. 解析 (Parsing): 将 NLQL 语句解析为抽象语法树 (AST)。
2. 数据获取 (Data Retrieval):
   • Adapter 职责: 只负责从数据源获取原始数据（chunks/documents）
   • 不做过滤: Adapter 不执行 WHERE/ORDER BY/LIMIT 逻辑
3. 查询执行 (Query Execution):
   • 语义搜索: 如果查询包含 SIMILAR_TO，计算向量相似度并存储分数到 metadata["similarity"]
   • 粒度转换: 根据 SELECT 的粒度（SENTENCE/SPAN）转换文本单元
   • WHERE 过滤: 在转换后的单元上应用过滤条件
   • ORDER BY 排序: 按指定字段或相似度排序
   • LIMIT 限制: 限制返回结果数量
4. 结果返回: 将处理后的文本单元转换为 Result 对象返回

设计原则：

• 职责分离: Adapter 只负责数据获取，Executor 负责所有查询逻辑
• 一致性: 所有数据源的查询语义完全一致
• 可扩展: 新数据源只需实现简单的 query() 方法

---

📦 安装

pip install python-nlql

🗓️ Roadmap

• 支持 SQL JOIN 操作（多文档集合关联）。
• 集成更多向量数据库适配器 (Milvus, Qdrant)。
• 引入 EXPLAIN 语句，解释查询计划和 Token 消耗。