4.1K Star！GitHub 上挖到一个救星级别的 RAG 数据流水线项目！ - 文章 - 开发者社区

很多人一开始做 RAG，注意力全在模型上。真正做过项目后才发现 —— 最折磨人的根本不是模型，而是数据处理。

一般 RAG 项目里最容易失控的部分：

• PDF、Markdown、网页、代码混在一起
• 源文件一改，全量重跑 embedding
• 跑一次向量化就是钱
• 数据清洗、切片、入库脚本散落一地，没人敢动

最终结果往往是：

“RAG 能跑，但谁都不敢维护。”

最近，我在 GitHub 上挖到了一个救星级别的开源项目 CocoIndex 。

picture.image

它不搞花里胡哨的模型包装，而是专注于解决 RAG 最底层的数据流水线问题。

官方号称只需 100 行左右的 Python 代码，就能构建一个企业级、支持增量更新的数据处理流。

项目介绍

CocoIndex 是一个专为 AI 场景打造的高性能数据转换框架。

picture.image

你可以把它理解为 RAG 领域的“超级流水线工厂”，你只需要定义好“原材料”（数据源）和“产品”（向量库），中间的加工过程，它全包了。

核心设计理念是：

把数据处理流程，当成一条“可追踪、可复用、可增量”的索引管道（Index Pipeline）。

内置一整套 RAG 必备组件，不用自己造轮子。PDF 解析、文本分块（Chunking）、Embedding 生成、知识图谱构建、结构化/非结构化数据统一处理等一应俱全。

核心能力

1、卓越的速度

仅需约100行Python代码即可在数据流中声明转换逻辑。


 
 
 
 
   
# import  
data['content'] = flow\_builder.add\_source(...)  
  
# transform  
data['out'] = data['content']  
    .transform(...)  
    .transform(...)  
  
# collect data  
collector.collect(...)  
  
# export to db, vector db, graph db ...  
collector.export(...)

每个转换仅基于输入字段生成新字段，没有隐藏状态和值突变。所有转换前后的数据均可观察，并自带数据血缘追踪。

特别之处在于，开发者无需通过创建、更新和删除操作来显式改变数据，只需为源数据集定义转换规则/公式即可。

2、即插即用构建模块

为不同数据源、目标和转换提供原生内置组件。

picture.image

标准化接口，实现不同组件间的一行代码切换——如同搭积木般简单。

从个人知识库到企业级数据管道，都能平滑覆盖。

3、数据新鲜度

CocoIndex能毫不费力地保持源数据与目标的同步。

它提供开箱即用的增量索引支持：在源数据或逻辑变更时执行最小化重计算，（重新）处理必要部分，尽可能复用缓存。

picture.image

快速开始

如果你是第一次了解到这个项目，建议详细阅览其使用文档。

使用文档：https://cocoindex.io/docs

安装 CocoIndex Python 库


 
 
 
 
   
pip install -U cocoindex

当然，如果你想安装 Claude Code 技能以获得增强的开发体验。在 Claude Code 中运行以下命令：


 
 
 
 
   
/plugin marketplace add cocoindex-io/cocoindex-claude  
/plugin install cocoindex-skills@cocoindex

定义数据流


 
 
 
 
   
@cocoindex.flow\_def(name="TextEmbedding")  
def text\_embedding\_flow(flow\_builder: cocoindex.FlowBuilder, data\_scope: cocoindex.DataScope):  
    # Add a data source to read files from a directory  
    data\_scope["documents"] = flow\_builder.add\_source(cocoindex.sources.LocalFile(path="markdown\_files"))  
  
    # Add a collector for data to be exported to the vector index  
    doc\_embeddings = data\_scope.add\_collector()  
  
    # Transform data of each document  
    with data\_scope["documents"].row() as doc:  
        # Split the document into chunks, put into `chunks` field  
        doc["chunks"] = doc["content"].transform(  
            cocoindex.functions.SplitRecursively(),  
            language="markdown", chunk\_size=2000, chunk\_overlap=500)  
  
        # Transform data of each chunk  
        with doc["chunks"].row() as chunk:  
            # Embed the chunk, put into `embedding` field  
            chunk["embedding"] = chunk["text"].transform(  
                cocoindex.functions.SentenceTransformerEmbed(  
                    model="sentence-transformers/all-MiniLM-L6-v2"))  
  
            # Collect the chunk into the collector.  
            doc\_embeddings.collect(filename=doc["filename"], location=chunk["location"],  
                                   text=chunk["text"], embedding=chunk["embedding"])  
  
    # Export collected data to a vector index.  
    doc\_embeddings.export(  
        "doc\_embeddings",  
        cocoindex.targets.Postgres(),  
        primary\_key\_fields=["filename", "location"],  
        vector\_indexes=[  
            cocoindex.VectorIndexDef(  
                field\_name="embedding",  
                metric=cocoindex.VectorSimilarityMetric.COSINE\_SIMILARITY)])