Ragbits：快速构建 GenAI 应用的模块化工具包

什么是 Ragbits？

Ragbits 是一套专为加速生成式 AI 应用开发而设计的模块化工具包。它提供了构建可靠、可扩展 AI 应用所需的核心组件，让开发者能够快速实现以下功能：

• 无缝集成 100+大语言模型
• 构建文档检索增强系统(RAG)
• 创建带用户界面的聊天机器人
• 实现分布式文档处理
• 部署生产级 AI 应用

这个工具包由深度求索团队开发，采用 MIT 开源协议，特别适合需要快速原型验证和生产部署的 AI 项目。

核心功能解析

🔨 构建可靠可扩展的 GenAI 应用

1. 灵活的 LLM 集成

• 通过 LiteLLM 支持 100+主流大语言模型
• 本地模型部署能力
• 类型安全的 LLM 调用接口（Python 泛型支持）

# 类型安全的LLM调用示例
class QuestionAnswerPromptOutput(BaseModel):
    answer: str

llm = LiteLLM(model_name="gpt-4.1-nano", use_structured_output=True)
# 输出自动验证为QuestionAnswerPromptOutput结构

2. 向量数据库自由选择

• 内置 Qdrant、PgVector 等流行向量库支持
• 内存向量存储快速原型验证
• 自定义存储适配器接口

3. 开发者友好工具

• 命令行管理向量存储
• 终端直接测试提示词
• 模块化安装减少依赖

ragbits vector-store manage  # CLI管理向量存储

📚 文档处理与检索增强

1. 多格式文档支持

• 处理 PDF/HTML/表格/演示稿等 20+格式
• 可选 Docling 或 Unstructured 解析引擎
• 支持表格/图片等复杂内容提取

2. 分布式文档处理

• Ray 框架实现并行处理
• 大规模数据集快速导入
• 云存储原生支持(S3/GCS/Azure)

# 分布式文档处理示例
await document_search.ingest("web://https://arxiv.org/pdf/1706.03762")

3. 检索增强生成(RAG)

• 端到端 RAG 流水线
• 上下文感知问答系统
• 检索结果自动注入提示词

🚀 部署与监控方案

1. 全链路可观测性

• OpenTelemetry 集成
• CLI 实时性能监控
• 用户反馈收集机制

2. 生产就绪功能

• 基于 promptfoo 的提示词测试
• 模型性能自动优化
• 带持久化存储的聊天 UI

# 聊天机器人部署示例
RagbitsAPI(MyChat).run()  # 一键启动聊天服务

3. 模块化扩展架构

• 按需安装组件
• 自定义解析器开发接口
• 适配器模式连接外部系统

五分钟快速上手

安装指南

基础安装（推荐初学者）：

pip install ragbits

包含六大核心模块：

1. ragbits-core – LLM/向量库基础操作
2. ragbits-agents – 智能体系统构建
3. ragbits-document-search – 文档检索流水线
4. ragbits-evaluate – 评估框架
5. ragbits-chat – 聊天应用基础设施
6. ragbits-cli – 命令行工具

高级用户可选组件安装：

pip install ragbits-core ragbits-document-search

基础 LLM 调用

import asyncio
from pydantic import BaseModel
from ragbits.core.llms import LiteLLM
from ragbits.core.prompt import Prompt

# 定义数据结构
class QAInput(BaseModel):
    question: str

class QAOutput(BaseModel):
    answer: str

# 创建提示模板
class QAPrompt(Prompt[QAInput, QAOutput]):
    system_prompt = "你是一个问答助手，请准确回答问题"
    user_prompt = "问题：{{ question }}"

# 执行LLM调用
async def main():
    llm = LiteLLM(model_name="gpt-4.1-nano")
    prompt = QAPrompt(QAInput(question="Linux中的high memory和low memory是什么？"))
    response = await llm.generate(prompt)
    print(response.answer)

asyncio.run(main())

文档检索实战

from ragbits.core.embeddings import LiteLLMEmbedder
from ragbits.core.vector_stores import InMemoryVectorStore
from ragbits.document_search import DocumentSearch

# 初始化检索系统
embedder = LiteLLMEmbedder(model_name="text-embedding-3-small")
vector_store = InMemoryVectorStore(embedder=embedder)
document_search = DocumentSearch(vector_store=vector_store)

# 摄入并查询论文
async def run():
    await document_search.ingest("web://https://arxiv.org/pdf/1706.03762")
    result = await document_search.search("论文的核心发现是什么？")
    print(result[0].text_representation[:500])  # 打印前500字符

asyncio.run(run())

完整 RAG 管道

class RAGInput(BaseModel):
    question: str
    context: list[str]

class RAGPrompt(Prompt[RAGInput, str]):
    system_prompt = "基于上下文回答问题，不知道就说不知道"
    user_prompt = "问题：{{ question }}\n上下文：{% for item in context %}{{ item }}{% endfor %}"

# 组合检索+生成
async def rag_pipeline(question: str):
    # 文档检索
    search_results = await document_search.search(question)

    # 构建提示
    rag_input = RAGInput(
        question=question,
        context=[r.text_representation for r in search_results]
    )
    prompt = RAGPrompt(rag_input)

    # 生成答案
    return await llm.generate(prompt)

聊天机器人开发

from ragbits.chat.api import RagbitsAPI
from ragbits.chat.interface import ChatInterface

class MyChatBot(ChatInterface):
    async def setup(self):
        # 初始化检索系统
        self.embedder = LiteLLMEmbedder(model_name="text-embedding-3-small")
        self.vector_store = InMemoryVectorStore(embedder=self.embedder)
        self.document_search = DocumentSearch(vector_store=self.vector_store)
        await self.document_search.ingest("web://https://example.com/knowledge.pdf")

    async def chat(self, message: str, history=None, context=None):
        # 检索相关文档
        results = await self.document_search.search(message)
        context_texts = [r.text_representation for r in results]

        # 流式响应生成
        async for chunk in self.llm.generate_streaming(
            RAGPrompt(RAGInput(question=message, context=context_texts))
        ):
            yield self.create_text_response(chunk)

# 启动服务
RagbitsAPI(MyChatBot).run()

进阶开发技巧

项目脚手架生成

uvx create-ragbits-app

生成包含以下结构的项目：

/my-rag-app
  ├── data_sources/    # 文档存储
  ├── pipelines/       # 处理流水线
  ├── prompts/         # 提示词模板
  ├── evaluators/      # 评估模块
  └── app.py           # 主应用

性能优化建议

1. 嵌入模型选择

   • 小型应用：text-embedding-3-small
   • 高精度需求：text-embedding-3-large

2. 分布式文档处理

# 启用Ray并行处理
from ragbits.document_search.distributed import RayDocumentProcessor
processor = RayDocumentProcessor(num_workers=8)
await processor.process_batch(documents)

3. 混合检索策略

# 结合关键词+向量检索
from ragbits.document_search.retrievers import HybridRetriever
retriever = HybridRetriever(vector_store, keyword_store)

最佳实践指南

提示工程技巧

1. 结构化输出控制

class StructuredOutput(BaseModel):
    key_points: list[str]
    summary: str
    confidence: float

# 强制LLM返回结构化数据
llm = LiteLLM(use_structured_output=True)

2. 动态上下文注入

user_prompt = """
{% if user.role == 'admin' %}
  管理员指令：{{ admin_query }}
{% else %}
  用户问题：{{ question }}
{% endif %}
"""

生产环境部署

1. 监控配置

# opentelemetry.yaml
exporters:
  console:
    verbosity: basic
  jaeger:
    endpoint: "localhost:14250"

2. 聊天界面定制

RagbitsAPI(
    MyChatBot,
    ui_config={
        "title": "企业知识助手",
        "primary_color": "#2563eb"
    }
).run(port=8080)

常见问题解答

技术实现类

Q：Ragbits 支持哪些本地大模型？
A：通过 LiteLLM 兼容所有 HuggingFace 格式的本地模型，支持 GGUF/GGML 等量化格式

Q：如何处理扫描版 PDF 中的表格数据？
A：集成 Unstructured.io 的 OCR 引擎，自动提取表格和图像内容

Q：向量检索的精度如何保证？
A：提供多级优化方案：

1. 分块策略调整（滑动窗口/语义分块）
2. 重排序(re-ranking)支持
3. 混合检索模式

使用场景类

Q：适合构建什么类型的应用？
A：典型应用场景包括：

• 企业知识库问答系统
• 研究论文分析工具
• 客户支持聊天机器人
• 内部文档智能检索

Q：与传统 LangChain 有何不同？
A：核心差异点：

1. 强类型系统保障
2. 原生分布式处理
3. 内置生产监控
4. 模块化依赖管理

性能与扩展

Q：单机支持多大知识库？
A：基准测试数据：

Q：如何扩展处理能力？
A：水平扩展方案：

1. 文档处理层：增加 Ray 工作节点
2. 检索层：向量库分片
3. 生成层：LLM 负载均衡

资源与支持

学习资源

• 官方文档
• API 参考
• 示例项目库

社区支持

• GitHub 讨论区
• 问题追踪
• 贡献指南