SQLite按字符长度限制分组拼接Purchase字段生成Chunk方案问询

问题背景 #

现有SQLite表结构如下：

CREATE TABLE purchases (
    ID INTEGER PRIMARY KEY,
    Customer TEXT,
    Purchase TEXT
);

数据示例：

需求要求：

• 按Customer分组，将多行Purchase拼接为多个Chunk，每个Chunk总字符数不超过10000
• 最大化每个Chunk的内容，优先填充能利用剩余空间的行以提升利用率
• 不重复处理同一行数据，最终用于LLM输入（超长内容需分多次处理）
• 已知单行Purchase长度均不超过10000字符

方案一：Python 3.12 + SQLite（推荐） #

Python的灵活性更适合处理这种动态长度计算和贪心填充逻辑，代码易维护、易扩展。

实现步骤 #

1. 从SQLite读取数据并按用户分组
2. 对每个用户的采购记录按长度降序排序，便于优先填充大尺寸内容
3. 用贪心算法生成符合字符限制的Chunk，尽可能填满每个Chunk的可用空间

代码示例 #

import sqlite3

# 配置参数
CHUNK_CHAR_LIMIT = 10000
DATABASE_PATH = "your_db_file.db"
SEPARATOR = "\n"  # 拼接分隔符，可根据需求调整
SEPARATOR_LENGTH = len(SEPARATOR)

def fetch_customer_purchases():
    """从SQLite读取所有采购数据，按用户分组"""
    conn = sqlite3.connect(DATABASE_PATH)
    cursor = conn.cursor()
    cursor.execute("""
        SELECT Customer, ID, Purchase, LENGTH(Purchase) AS purchase_length
        FROM purchases
        ORDER BY Customer
    """)
    raw_data = cursor.fetchall()
    conn.close()

    # 按用户整理数据
    customer_groups = {}
    for cust, idx, purch, length in raw_data:
        if cust not in customer_groups:
            customer_groups[cust] = []
        customer_groups[cust].append({
            "id": idx,
            "purchase_text": purch,
            "length": length
        })
    return customer_groups

def build_optimal_chunks(purchase_list, customer_name):
    """为单个用户生成符合限制的Chunk列表"""
    chunks = []
    unprocessed = sorted(purchase_list, key=lambda x: -x["length"])  # 按长度降序，优先处理长内容

    while unprocessed:
        current_chunk = []
        current_total_length = 0

        # 贪心填充当前Chunk
        temp_unprocessed = unprocessed.copy()
        for item in temp_unprocessed:
            # 计算添加当前条目后的总长度（含分隔符）
            required_length = item["length"] + (SEPARATOR_LENGTH if current_chunk else 0)
            if current_total_length + required_length <= CHUNK_CHAR_LIMIT:
                current_total_length += required_length
                current_chunk.append(item)
                unprocessed.remove(item)

        # 生成Chunk结果
        if current_chunk:
            chunks.append({
                "customer": customer_name,
                "included_ids": [item["id"] for item in current_chunk],
                "total_length": current_total_length,
                "chunk_content": SEPARATOR.join([item["purchase_text"] for item in current_chunk])
            })
    return chunks

def main():
    customer_data = fetch_customer_purchases()
    all_chunks = []

    for cust_name, purchases in customer_data.items():
        cust_chunks = build_optimal_chunks(purchases, cust_name)
        all_chunks.extend(cust_chunks)

    # 示例：输出Chunk信息，可替换为存入临时表或直接传入LLM处理
    for chunk in all_chunks:
        print(f"=== Customer: {chunk['customer']} | Chunk Length: {chunk['total_length']} ===")
        print(f"Included IDs: {chunk['included_ids']}")
        print(f"Content:\n{chunk['chunk_content']}\n")

if __name__ == "__main__":
    main()

方案优势 #

• 贪心算法确保每个Chunk的空间利用率最大化
• 可灵活调整分隔符、字符限制等参数
• 支持记录已处理的行ID，避免重复处理（可扩展存入临时表）
• 逻辑清晰，便于调试和后续功能扩展（如LLM调用集成）

方案二：纯SQLite实现（无Python环境场景） #

利用SQLite的递归CTE和窗口函数实现分组拼接，逻辑较复杂，适合只能使用SQL的场景。

实现SQL #

-- 临时表：存储带长度和分组内序号的采购数据（按长度降序排序）
CREATE TEMP TABLE IF NOT EXISTS temp_purchase_data AS
SELECT
    Customer,
    ID,
    Purchase,
    LENGTH(Purchase) AS purchase_length,
    ROW_NUMBER() OVER (PARTITION BY Customer ORDER BY LENGTH(Purchase) DESC) AS row_num
FROM purchases;

-- 递归CTE：生成每个行对应的Chunk编号
WITH RECURSIVE chunk_assignment AS (
    -- 初始化：每个用户的第一行数据
    SELECT
        Customer,
        ID,
        Purchase,
        purchase_length,
        row_num,
        1 AS chunk_id,
        purchase_length AS current_total_length
    FROM temp_purchase_data
    WHERE row_num = 1

    UNION ALL

    -- 递归处理后续行，判断是否加入当前Chunk或新建Chunk
    SELECT
        t.Customer,
        t.ID,
        t.Purchase,
        t.purchase_length,
        t.row_num,
        -- 若添加当前行（含分隔符）超过限制，则新建Chunk
        CASE
            WHEN ca.current_total_length + 1 + t.purchase_length > 10000 THEN ca.chunk_id + 1
            ELSE ca.chunk_id
        END AS chunk_id,
        -- 更新当前Chunk的总长度
        CASE
            WHEN ca.current_total_length + 1 + t.purchase_length > 10000 THEN t.purchase_length
            ELSE ca.current_total_length + 1 + t.purchase_length
        END AS current_total_length
    FROM chunk_assignment ca
    JOIN temp_purchase_data t
        ON t.Customer = ca.Customer AND t.row_num = ca.row_num + 1
)

-- 按用户和Chunk编号拼接内容，输出最终结果
SELECT
    Customer,
    chunk_id,
    GROUP_CONCAT(Purchase, '\n') AS chunk_content,
    -- 计算总长度：所有采购文本长度 + 分隔符数量（行数-1）
    SUM(purchase_length) + (COUNT(*) - 1) AS total_chunk_length
FROM chunk_assignment
GROUP BY Customer, chunk_id
ORDER BY Customer, chunk_id;

说明 #

• 使用GROUP_CONCAT拼接内容，分隔符为换行符（\n），总长度需包含分隔符的字符数
• 递归CTE中通过累计长度判断是否新建Chunk，优先处理长行以提升利用率
• 临时表的排序确保贪心逻辑的实现，若需保留原数据顺序，可调整ORDER BY规则

内容的提问来源于stack exchange，提问作者Darius Dauer