---

功能

🔨 构建可靠且可扩展的生成式AI应用

• 随时切换大模型 – 可通过LiteLLM在100多种大模型之间切换（详情），或运行本地模型（详情）。
• 类型安全的大模型调用 – 使用Python泛型，在模型交互中强制实现严格的类型安全性（详情）。
• 自定义向量存储 – 内置支持连接到Qdrant（详情）、PgVector（详情）等，也可轻松集成其他向量数据库。
• 内置开发者工具 – 可通过命令行管理向量存储（详情）、查询管道，并从终端测试提示词（详情）。
• 模块化安装 – 只需安装所需组件，减少依赖项并提升性能。

📚 快速灵活的RAG处理

• 支持20+种格式 – 可处理PDF、HTML、电子表格、演示文稿等多种文件格式。可使用Docling（GitHub链接）、Unstructured（GitHub链接）进行数据处理，或自定义解析器。
• 处理复杂数据 – 内置视觉语言模型支持，可提取表格、图像及结构化内容。
• 连接任意数据源 – 提供S3、GCS、Azure等预构建连接器，也可自行实现自定义连接。
• 规模化数据摄取 – 利用基于Ray的并行处理技术，快速处理大规模数据集（详情）。

🤖 轻松构建多智能体工作流

• 多智能体协作 – 使用A2A协议创建角色分工明确的智能体团队，实现互操作性（教程）。
• 实时数据集成 – 借助模型上下文协议（MCP），实现对网页、数据库查询及API的实时访问（详情）。
• 对话状态管理 – 通过自动历史记录跟踪功能，保持跨轮次交互的上下文一致性（详情）。

🚀 放心部署与监控

• 实时可观测性 – 使用OpenTelemetry（详情）和CLI洞察工具（详情）追踪系统性能。
• 内置测试工具 – 在部署前使用Promptfoo验证提示词的有效性（详情）。
• 自动优化 – 持续评估并优化模型性能。
• 聊天UI – 提供完整的聊天机器人接口（详情），支持API调用、数据持久化及用户反馈收集。

安装

稳定版

若想快速上手，可安装最新稳定版：

pip install ragbits

夜间构建版

如需体验最新的开发功能，可安装从main分支自动发布的夜间构建版：

pip install ragbits --pre

注意： 夜间构建版包含最新功能和错误修复，但稳定性可能不如正式发布版。其版本号格式为X.Y.Z.devYYYYMMDDHHMM。

包含内容

此入门级软件包包含以下核心组件：

• ragbits-core - 处理提示词、大模型和向量数据库的基础工具。
• ragbits-agents - 用于构建智能体系统的抽象层。
• ragbits-document-search - 用于知识库的检索与数据摄取管道。
• ragbits-evaluate - Ragbits组件的统一评估框架。
• ragbits-guardrails - 用于确保响应安全性和相关性的实用工具。
• ragbits-chat - 构建对话式AI应用的全栈基础设施。
• ragbits-cli - 用于与Ragbits组件交互的命令行工具。

此外，您也可以单独安装各个组件以满足特定需求。

快速入门

基础示例

定义提示词并调用大模型：

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

class QuestionAnswerPromptInput(BaseModel):
    question: str

class QuestionAnswerPrompt(Prompt[QuestionAnswerPromptInput, str]):
    system_prompt = """
    你是一位问答助手。请尽最大努力回答问题。
    """
    user_prompt = """
    问题：{{ question }}
    """

llm = LiteLLM(model_name="gpt-4.1-nano")

async def main() -> None:
    prompt = QuestionAnswerPrompt(QuestionAnswerPromptInput(question="Linux中的高内存和低内存是什么？"))
    response = await llm.generate(prompt)
    print(response)

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

文档搜索

构建并查询一个简单的向量存储索引：

import asyncio
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() -> None:
    await document_search.ingest("web://https://arxiv.org/pdf/1706.03762")
    result = await document_search.search("这篇论文的主要发现是什么？")
    print(result)

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

检索增强生成

构建一个简单的 RAG 流程：

import asyncio
from collections.abc import Iterable
from pydantic import BaseModel
from ragbits.core.embeddings import LiteLLMEmbedder
from ragbits.core.llms import LiteLLM
from ragbits.core.prompt import Prompt
from ragbits.core.vector_stores import InMemoryVectorStore
from ragbits.document_search import DocumentSearch
from ragbits.document_search.documents.element import Element

class QuestionAnswerPromptInput(BaseModel):
    question: str
    context: Iterable[Element]

class QuestionAnswerPrompt(Prompt[QuestionAnswerPromptInput, str]):
    system_prompt = """
    你是一位问答代理。请根据提供的上下文回答问题。
    如果给定的上下文中信息不足，请拒绝回答。
    """
    user_prompt = """
    问题：{{ question }}
    上下文：{% for chunk in context %}{{ chunk.text_representation }}{%- endfor %}
    """

llm = LiteLLM(model_name="gpt-4.1-nano")
embedder = LiteLLMEmbedder(model_name="text-embedding-3-small")
vector_store = InMemoryVectorStore(embedder=embedder)
document_search = DocumentSearch(vector_store=vector_store)

async def run() -> None:
    question = "这篇论文的主要发现是什么？"

    await document_search.ingest("web://https://arxiv.org/pdf/1706.03762")
    chunks = await document_search.search(question)

    prompt = QuestionAnswerPrompt(QuestionAnswerPromptInput(question=question, context=chunks))
    response = await llm.generate(prompt)
    print(response)

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

智能体式 RAG

构建一个智能体式的 RAG 流程：

import asyncio
from ragbits.agents import Agent
from ragbits.core.embeddings import LiteLLMEmbedder
from ragbits.core.llms import LiteLLM
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)

llm = LiteLLM(model_name="gpt-4.1-nano")
agent = Agent(llm=llm, tools=[document_search.search])

async def main() -> None:
    await document_search.ingest("web://https://arxiv.org/pdf/1706.03762")
    response = await agent.run("这篇论文的主要发现是什么？")
    print(response.content)

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

聊天 UI

通过 Ragbits API 公开你的生成式 AI 应用程序：

from collections.abc import AsyncGenerator
from ragbits.agents import Agent, ToolCallResult
from ragbits.chat.api import RagbitsAPI
from ragbits.chat.interface import ChatInterface
from ragbits.chat.interface.types import ChatContext, ChatResponse, LiveUpdateType
from ragbits.core.embeddings import LiteLLMEmbedder
from ragbits.core.llms import LiteLLM, ToolCall
from ragbits.core.prompt import ChatFormat
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)

llm = LiteLLM(model_name="gpt-4.1-nano")
agent = Agent(llm=llm, tools=[document_search.search])

class MyChat(ChatInterface):
    async def setup(self) -> None:
        await document_search.ingest("web://https://arxiv.org/pdf/1706.03762")

    async def chat(
        self,
        message: str,
        history: ChatFormat,
        context: ChatContext,
    ) -> AsyncGenerator[ChatResponse]:
        async for result in agent.run_streaming(message):
            match result:
                case str():
                    yield self.create_live_update(
                        update_id="1",
                        type=LiveUpdateType.START,
                        label="正在回答...",
                    )
                    yield self.create_text_response(result)
                case ToolCall():
                    yield self.create_live_update(
                        update_id="2",
                        type=LiveUpdateType.START,
                        label="正在搜索...",
                    )
                case ToolCallResult():
                    yield self.create_live_update(
                        update_id="2",
                        type=LiveUpdateType.FINISH,
                        label="搜索",
                        description=f"找到了 {len(result.result)} 个相关片段。",
                    )

        yield self.create_live_update(
            update_id="1",
            type=LiveUpdateType.FINISH,
            label="答案",
        )

if __name__ == "__main__":
    api = RagbitsAPI(MyChat)
    api.run()

快速开发

从模板创建 Ragbits 项目：

uvx create-ragbits-app

可在 这里 探索 create-ragbits-app 仓库。如果你有新的模板创意，欢迎贡献！

文档

• 教程 - 几分钟内开始使用 Ragbits
• 操作指南 - 学习如何在你的项目中使用 Ragbits
• CLI - 学习如何在终端中运行 Ragbits
• API 参考 - 探索底层的 Ragbits API

贡献

我们欢迎贡献！请阅读 CONTRIBUTING.md 以获取更多信息。

许可证

Ragbits 采用 MIT 许可证 许可。

Ragbits 快速上手指南

Ragbits 是一个用于快速构建生成式 AI（GenAI）应用的模块化框架，提供从 LLM 调用、RAG 流水线到多智能体协作的全套构建块。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统：Linux, macOS 或 Windows
• Python 版本：3.10 或更高版本
• 包管理器：pip
• 前置依赖：建议创建虚拟环境以避免依赖冲突

  python -m venv venv
  source venv/bin/activate  # Windows 用户使用: venv\Scripts\activate
  

提示：国内开发者若遇到 PyPI 下载缓慢问题，可使用清华或阿里镜像源加速安装。

安装步骤

1. 安装稳定版

安装包含核心功能、代理、文档搜索和聊天界面的完整套件：

pip install ragbits

使用国内镜像源加速安装：

pip install ragbits -i https://pypi.tuna.tsinghua.edu.cn/simple

2. 安装夜间构建版（可选）

如需体验最新开发特性（可能不够稳定）：

pip install ragbits --pre

3. 模块化安装（可选）

Ragbits 支持按需安装子模块，例如仅安装核心功能：

pip install ragbits-core

基本使用

以下是几个核心场景的最小化代码示例，所有示例均需异步运行。

1. 基础 LLM 调用

定义一个类型安全的 Prompt 并调用 LLM：

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

class QuestionAnswerPromptInput(BaseModel):
    question: str

class QuestionAnswerPrompt(Prompt[QuestionAnswerPromptInput, str]):
    system_prompt = """
    You are a question answering agent. Answer the question to the best of your ability.
    """
    user_prompt = """
    Question: {{ question }}
    """

llm = LiteLLM(model_name="gpt-4.1-nano")

async def main() -> None:
    prompt = QuestionAnswerPrompt(QuestionAnswerPromptInput(question="What are high memory and low memory on linux?"))
    response = await llm.generate(prompt)
    print(response)

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

2. 构建简单 RAG 流水线

实现文档摄入、向量检索与生成的完整流程：

import asyncio
from collections.abc import Iterable
from pydantic import BaseModel
from ragbits.core.embeddings import LiteLLMEmbedder
from ragbits.core.llms import LiteLLM
from ragbits.core.prompt import Prompt
from ragbits.core.vector_stores import InMemoryVectorStore
from ragbits.document_search import DocumentSearch
from ragbits.document_search.documents.element import Element

class QuestionAnswerPromptInput(BaseModel):
    question: str
    context: Iterable[Element]

class QuestionAnswerPrompt(Prompt[QuestionAnswerPromptInput, str]):
    system_prompt = """
    You are a question answering agent. Answer the question that will be provided using context.
    If in the given context there is not enough information refuse to answer.
    """
    user_prompt = """
    Question: {{ question }}
    Context: {% for chunk in context %}{{ chunk.text_representation }}{%- endfor %}
    """

llm = LiteLLM(model_name="gpt-4.1-nano")
embedder = LiteLLMEmbedder(model_name="text-embedding-3-small")
vector_store = InMemoryVectorStore(embedder=embedder)
document_search = DocumentSearch(vector_store=vector_store)

async def run() -> None:
    question = "What are the key findings presented in this paper?"

    # 摄入文档 (支持本地路径或 URL)
    await document_search.ingest("web://https://arxiv.org/pdf/1706.03762")
    
    # 检索相关片段
    chunks = await document_search.search(question)

    # 生成回答
    prompt = QuestionAnswerPrompt(QuestionAnswerPromptInput(question=question, context=chunks))
    response = await llm.generate(prompt)
    print(response)

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

3. 智能体 RAG (Agentic RAG)

将文档搜索作为工具赋予智能体，使其自主决定何时检索：

import asyncio
from ragbits.agents import Agent
from ragbits.core.embeddings import LiteLLMEmbedder
from ragbits.core.llms import LiteLLM
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)

llm = LiteLLM(model_name="gpt-4.1-nano")
# 将 search 方法注册为工具
agent = Agent(llm=llm, tools=[document_search.search])

async def main() -> None:
    await document_search.ingest("web://https://arxiv.org/pdf/1706.03762")
    response = await agent.run("What are the key findings presented in this paper?")
    print(response.content)

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