项目简介

Quivr 是一个开源的个人第二大脑（Second Brain）项目，利用生成式人工智能（Generative AI）技术帮助用户构建智能知识助手。该项目旨在为开发者提供一套快速、高效且可定制的检索增强生成（RAG）解决方案，使开发者能够专注于产品开发，而无需深入研究 RAG 技术的底层细节。

资料来源：README.md:1-15

核心定位

Quivr 的核心价值主张在于：

• 开箱即用的 RAG 方案：提供经过优化的检索增强生成工作流，开箱即用
• 多模型支持：兼容 OpenAI、Anthropic、Mistral、Gemma 等主流大语言模型
• 任意文件类型支持：支持 PDF、TXT、Markdown 等常见格式，并允许自定义解析器
• 完全可定制：支持添加互联网搜索、工具扩展等高级功能

资料来源：README.md:25-35

技术架构

整体架构

Quivr 采用模块化架构设计，主要包含以下核心组件：

graph TD
    A[用户查询] --> B[Brain 核心类]
    B --> C[RAG 工作流引擎]
    C --> D[检索模块]
    C --> E[重排序模块]
    C --> F[生成模块]
    D --> G[向量数据库]
    E --> G
    F --> H[LLM 大语言模型]
    G --> D

核心模块说明

资料来源：core/README.md:1-15

快速入门

环境要求

资料来源：README.md:50-55

安装方式

通过 pip 安装 quivr-core 核心包：

pip install quivr-core

资料来源：README.md:30-35

5 行代码创建 RAG

以下示例展示了如何使用 Quivr 构建一个简单的问答系统：

import tempfile
from quivr_core import Brain

if __name__ == "__main__":
    with tempfile.NamedTemporaryFile(mode="w", suffix=".txt") as temp_file:
        temp_file.write("Gold is a liquid of blue-like colour.")
        temp_file.flush()

        brain = Brain.from_files(
            name="test_brain",
            file_paths=[temp_file.name],
        )

        answer = brain.ask("what is gold? answer in french")
        print("answer:", answer)

资料来源：README.md:38-55

Brain 核心类

核心功能

Brain 类是 Quivr 的核心入口点，封装了文件处理、向量存储和问答功能。该类提供了以下关键方法：

资料来源：core/quivr_core/brain/brain.py:1-80

文件处理流程

graph LR
    A[上传文件] --> B[Processor 解析]
    B --> C[分块处理]
    C --> D[元数据提取]
    D --> E[语言检测]
    E --> F[向量数据库存储]

元数据处理

每个文档块在处理时都会附加丰富的元数据信息：

资料来源：core/quivr_core/processor/processor_base.py:15-35

RAG 工作流配置

工作流节点

Quivr 支持通过 YAML 配置文件定义 RAG 工作流，包含以下标准节点：

workflow_config:
  name: "standard RAG"
  nodes:
    - name: "START"
      edges: ["filter_history"]
    - name: "filter_history"
      edges: ["rewrite"]
    - name: "rewrite"
      edges: ["retrieve"]
    - name: "retrieve"
      edges: ["generate_rag"]
    - name: "generate_rag"
      edges: ["END"]

资料来源：README.md:70-85

配置参数

资料来源：README.md:85-100

支持的文件格式

Quivr 通过处理器架构支持多种文件格式：

资料来源：README.md:35-40

集成与扩展

大语言模型集成

Quivr 支持多种大语言模型供应商：

资料来源：README.md:20-25

外部服务集成

• Megaparse：用于大规模文件解析和 ingestion
• 向量数据库：支持可配置的向量存储后端

资料来源：README.md:38-42

应用示例

基础问答机器人

使用 Chainlit 框架构建的文本问答机器人架构：

graph TD
    A[用户上传文件] --> B[Chainlit 前端]
    B --> C[Quivr Brain 处理]
    C --> D[向量检索]
    D --> E[LLM 生成回答]
    E --> B

核心代码示例：

from quivr_core import Brain

brain = Brain.from_files(
    name="my smart brain",
    file_paths=["./my_first_doc.pdf", "./my_second_doc.txt"],
)

answer = brain.ask(question)

资料来源：examples/simple_question/simple_question.py:1-20

语音交互应用

quivr-whisper 示例展示了如何构建语音交互系统：

1. 用户录制语音问题
2. 通过 Whisper 转录为文本
3. 调用 Quivr API 获取回答
4. 使用 TTS 将回答合成语音并播放

资料来源：examples/quivr-whisper/README.md:1-25

项目结构

quivr/
├── README.md              # 主项目文档
├── core/                  # 核心包目录
│   ├── README.md          # quivr-core 说明
│   └── quivr_core/
│       ├── brain/         # Brain 核心类
│       ├── rag/           # RAG 工作流
│       ├── processor/     # 文件处理器
│       └── llm/           # LLM 接口封装
├── examples/              # 示例代码
│   ├── chatbot/           # Chainlit 聊天机器人
│   ├── chatbot_voice/     # 语音聊天机器人
│   ├── simple_question/   # 简单问答示例
│   └── pdf_parsing_tika.py # PDF 解析示例

资料来源：core/README.md:1-15

技术栈总结

下一步

• 访问 官方文档 了解更多高级功能
• 查看 示例代码 学习具体用法
• 参与 贡献指南 加入社区

概述

Quivr-core 是 Quivr 项目的核心 RAG（检索增强生成）引擎，专注于让开发者能够快速构建基于文档的智能问答系统。本指南将帮助您从零开始，在 5 行代码内创建一个功能完整的 RAG 应用。

安装

使用 pip 安装 quivr-core 包：

pip install quivr-core

资料来源：core/README.md

核心概念

Brain（大脑）

Brain 是 Quivr-core 的核心类，负责管理文档处理、向量存储和问答逻辑。

graph TD
    A[用户提问] --> B[Brain.ask]
    B --> C[检索相关文档]
    C --> D[构建提示词]
    D --> E[调用LLM]
    E --> F[返回答案]

基本用法

创建 Brain 并提问

以下示例演示了最基础的 RAG 用法：

import tempfile
from quivr_core import Brain

with tempfile.NamedTemporaryFile(mode="w", suffix=".txt") as temp_file:
    temp_file.write("Gold is a liquid of blue-like colour.")
    temp_file.flush()

    brain = Brain.from_files(
        name="test_brain",
        file_paths=[temp_file.name],
    )

    answer = brain.ask(
        question="what is gold? answer in french",
        run_id=None  # 需要传入UUID
    )
    print("answer:", answer.answer)

资料来源：examples/simple_question/simple_question.py:1-20

异步提问方式

Brain.ask 方法存在同步版本和异步版本：

异步方法签名：

async def ask_streaming(
    self,
    run_id: UUID,
    question: str,
    system_prompt: str | None = None,
    retrieval_config: RetrievalConfig | None = None,
    rag_pipeline: Type[Union[QuivrQARAG, QuivrQARAGLangGraph]] | None = None,
    list_files: list[QuivrKnowledge] | None = None,
    chat_history: ChatHistory | None = None,
) -> AsyncGenerator[ParsedRAGChunkResponse, None]

资料来源：core/quivr_core/brain/brain.py:检索配置

流式响应示例

import asyncio
from uuid import uuid4

async def streaming_example():
    brain = Brain.from_files(
        name="my_brain",
        file_paths=["./documents.pdf"]
    )
    
    async for chunk in brain.ask_streaming(
        run_id=uuid4(),
        question="Summarize the document"
    ):
        if not chunk.last_chunk:
            print(chunk.answer, end="", flush=True)

环境配置

API Key 设置

在使用 LLM 提供商之前，需要配置 API Key：

import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

Quivr 支持的 LLM 提供商：

进阶配置

自定义检索配置

通过 RetrievalConfig 可以自定义检索行为：

from quivr_core.config import RetrievalConfig

config = RetrievalConfig(
    max_history=10,
    prompt="You are a helpful assistant.",
    llm_config=LLMEndpointConfig(
        model="gpt-4",
        temperature=0.7,
        max_input_tokens=4000
    )
)

answer = brain.ask(
    run_id=uuid4(),
    question="your question",
    retrieval_config=config
)

使用 Chainlit 构建对话界面

创建交互式聊天机器人的完整示例：

import chainlit as cl
from uuid import uuid4
from quivr_core import Brain

@cl.on_message
async def main(message: cl.Message):
    # 获取或创建 brain
    brain = await cl.make_async(Brain.from_files)(
        name="chatbot_brain",
        file_paths=["./data.txt"]
    )
    
    msg = cl.Message(content="")
    async for response in brain.ask_streaming(
        run_id=uuid4(),
        question=message.content
    ):
        if not response.last_chunk:
            await msg.stream_token(response.answer)
    
    await msg.send()

资料来源：examples/chatbot/main.py

工作流程

graph LR
    A[文件] --> B[文件处理器]
    B --> C[文本分块]
    C --> D[向量化]
    D --> E[向量数据库]
    E --> F[检索]
    F --> G[LLM生成]
    G --> H[返回答案]

支持的文件格式

Quivr-core 通过内置处理器支持多种文件格式：

下一步

• 查看 Workflow 配置文档 了解高级检索策略
• 参考 LangChain 集成 实现自定义处理流程
• 探索 语音聊天示例 构建多模态应用

概述

Quivr 是一个基于生成式 AI 的个人第二大脑框架，其核心库 quivr-core 提供了一套完整的 RAG（检索增强生成）系统。该系统通过模块化设计，支持多种文件格式处理、灵活的 LLM 配置、可定制的检索策略以及持久化存储能力。资料来源：README.md:1-30

核心架构图

graph TD
    subgraph 用户层
        A[用户查询] --> B[Brain.ask]
    end

    subgraph 核心层
        B --> C[RetrievalConfig]
        C --> D[QuivrQARAGLangGraph]
        D --> E[LLMEndpoint]
        D --> F[VectorStore]
    end

    subgraph 数据层
        F --> G[Chroma/FAISS/Pgvector]
        G --> H[Embeddings]
    end

    subgraph 存储层
        I[LocalStorage] --> J[Brain持久化]
    end

    style 用户层 fill:#e1f5fe
    style 核心层 fill:#fff3e0
    style 数据层 fill:#e8f5e9
    style 存储层 fill:#fce4ec

核心组件

Brain 类

Brain 是整个系统的核心入口类，负责管理文件处理、向量存储、检索配置和 LLM 调用。资料来源：core/quivr_core/brain/brain.py:1-100

classDiagram
    class Brain {
        +id: UUID
        +name: str
        +llm: LLMEndpoint
        +embedder: Embeddings
        +vector_db: VectorStore
        +storage: StorageInterface
        +from_files()
        +afrom_files()
        +from_langchain_documents()
        +ask()
        +asearch()
        +save()
        +load()
    }

Brain 核心属性表：

资料来源：core/quivr_core/brain/brain.py:50-60

RAG 引擎

Quivr 提供两种 RAG 实现：基于 LangChain 的传统 RAG 和基于 LangGraph 的工作流 RAG。资料来源：core/quivr_core/rag/quivr_rag.py:1-50

graph LR
    A[用户问题] --> B[检索阶段]
    B --> C[重写查询]
    C --> D[历史过滤]
    D --> E[向量检索]
    E --> F[生成阶段]
    F --> G[LLM响应]

QuivrQARAGLangGraph 工作流节点：

资料来源：core/quivr_core/rag/quivr_rag_langgraph.py:1-80

配置系统

配置系统通过 RetrievalConfig 类统一管理 RAG 流程的各个环节。资料来源：core/quivr_core/config.py:1-100

retrieval_config = RetrievalConfig.from_yaml("./basic_rag_workflow.yaml")

RetrievalConfig 核心参数：

资料来源：core/quivr_core/config.py:30-50

数据流程

文件处理流程

graph TD
    A[文件上传] --> B[FileValidator]
    B --> C[FileProcessor]
    C --> D[LangChain Documents]
    D --> E[Embedder 向量化]
    E --> F[VectorStore 存储]
    F --> G[Brain 索引完成]

问答流程

sequenceDiagram
    participant U as 用户
    participant B as Brain
    participant R as QuivrQARAGLangGraph
    participant V as VectorStore
    participant L as LLM

    U->>B: ask(question)
    B->>R: answer_astream()
    R->>V: asearch(query)
    V-->>R: 相关文档块
    R->>L: generate(上下文)
    L-->>R: 流式响应
    R-->>B: ParsedRAGChunkResponse
    B-->>U: 返回答案

资料来源：core/quivr_core/rag/quivr_rag.py:80-120

存储系统

LocalStorage 实现

LocalStorage 提供了本地文件系统存储能力，支持文件去重（基于 SHA-1 哈希）。资料来源：core/quivr_core/storage/local_storage.py:1-80

graph TD
    A[上传文件] --> B{计算SHA-1}
    B --> C{文件存在?}
    C -->|是| D{exists_ok?}
    C -->|否| E[复制/链接文件]
    D -->|True| E
    D -->|False| F[抛出FileExistsError]
    E --> G[更新文件列表]

LocalStorage 核心方法：

资料来源：core/quivr_core/storage/local_storage.py:50-70

持久化与加载

Brain 保存与加载

Brain 实例支持通过 save() 和 load() 方法进行持久化。资料来源：examples/save_load_brain.py:1-25

# 保存 Brain
save_path = await brain.save("/home/user/.local/quivr")

# 加载 Brain
brain_loaded = Brain.load(save_path)

保存流程图：

graph LR
    A[Brain实例] --> B[序列化元数据]
    A --> C[序列化VectorDB]
    A --> D[序列化LLM配置]
    B --> E[写入文件系统]
    C --> E
    D --> E
    E --> F[返回保存路径]

LLM 配置

Quivr 支持多种 LLM 提供商，包括 OpenAI、Anthropic、Mistral 以及本地 Ollama 模型。资料来源：core/quivr_core/llm/llm_endpoint.py:1-50

from quivr_core.llm.llm_endpoint import LLMEndpoint
from quivr_core.rag.entities.config import LLMEndpointConfig

llm = LLMEndpoint.from_config(
    config=LLMEndpointConfig(
        model="gpt-4",
        llm_base_url="https://api.openai.com/v1",
        api_key=os.environ["OPENAI_API_KEY"]
    )
)

工作流配置

通过 YAML 配置文件可以自定义 RAG 工作流节点和边。资料来源：README.md:40-80

workflow_config:
  name: "standard RAG"
  nodes:
    - name: "START"
      edges: ["filter_history"]
    - name: "filter_history"
      edges: ["rewrite"]
    - name: "rewrite"
      edges: ["generate"]
    - name: "generate"
      edges: []

模块依赖关系

graph TD
    subgraph quivr_core
        A[Brain] --> B[RAG模块]
        A --> C[存储模块]
        A --> D[LLM模块]
        B --> E[配置模块]
        E --> F[向量存储]
        D --> F
    end

    subgraph 外部依赖
        G[LangChain]
        H[Chroma/FAISS]
        I[OpenAI API]
    end

    F --> H
    D --> G
    B --> G
    G --> I

关键设计决策

总结

Quivr 的系统架构采用分层设计，核心层由 Brain 类统一调度，通过 RetrievalConfig 配置检索策略，借助 LangGraph 工作流实现灵活的 RAG 流程。向量存储层支持多种后端，存储层提供本地持久化能力，整体设计注重可扩展性和模块化，使开发者能够根据需求自由组合各组件。

概述

Quivr 的 RAG（检索增强生成）处理流程是核心功能模块，负责将用户查询与知识库中的文档相结合，生成准确、上下文相关的回答。该流程整合了文档检索、查询重写、历史过滤和流式响应生成等多个环节，形成一个完整的问答管道。

RAG 流程的主要职责包括：从向量数据库中检索相关文档、对聊天历史进行过滤处理、将检索结果与用户查询组合成提示词、调用大语言模型生成回答，以及支持流式输出和元数据返回。资料来源：core/quivr_core/rag/quivr_rag.py:1-50

核心组件架构

组件关系图

graph TD
    A[用户查询] --> B[ChatHistory 聊天历史]
    B --> C[filter_history 过滤历史]
    C --> D[检索配置 RetrievalConfig]
    D --> E[向量存储 VectorStore]
    E --> F[Reranker 文档重排序]
    F --> G[LLMEndpoint 大语言模型]
    G --> H[流式响应 ParsedRAGChunkResponse]
    H --> I[最终回答]
    
    J[Brain 脑实例] --> E
    K[Embeddings 嵌入模型] --> E

核心类说明

资料来源：core/quivr_core/rag/quivr_rag.py:44-60

处理流程详解

1. 聊天历史过滤

在处理用户查询前，系统首先对聊天历史进行过滤，移除与当前问题无关的历史消息以提高检索效率。

def filter_history(
    self,
    chat_history: ChatHistory,
):
    """
    Filter out the chat history to only include the messages 
    that are relevant to the current question
    """

该方法接收 ChatHistory 对象作为输入，通过分析历史消息的相关性，返回过滤后的消息列表。资料来源：core/quivr_core/rag/quivr_rag.py:68-76

2. 文档检索配置

检索配置定义了 RAG 流程的行为参数，包括：

资料来源：core/quivr_core/brain/brain.py:80-95

3. 向量存储检索

Brain 类提供了 asearch 异步搜索方法，用于从向量数据库中检索相关文档：

async def asearch(
    self,
    query: str | Document,
    n_results: int = 5,
    filter: Callable | Dict[str, Any] | None = None,
    fetch_n_neighbors: int = 20,
) -> list[SearchResult]:

该方法返回 SearchResult 对象列表，每个结果包含文档内容和元数据。资料来源：core/quivr_core/brain/brain.py:80-95

4. 提示词构建

检索到的文档与用户查询一起被注入到提示词模板中：

user_prompt_template = """
Here is information about the user that can help you to answer:
<user_metadata>
{user_metadata}
</user_metadata>

Here are metadata on the current ticket:
<ticket_metadata>
{ticket_metadata}
</ticket_metadata>

Here are the most relevant similar tickets:
<similar_tickets>
{similar_tickets}
</similar_tickets>

Here is the client question:
<client_query>
{client_query}
</client_query>
"""

提示词模板支持以下占位符替换：

• {user_metadata} - 用户元数据
• {ticket_metadata} - 工单元数据
• {similar_tickets} - 相似工单
• {ticket_history} - 工单历史
• {additional_information} - 附加信息
• {client_query} - 客户端查询

资料来源：core/quivr_core/rag/prompts.py:1-50

5. 流式响应生成

QuivrQARAG 类实现了流式响应生成，通过 answer_astream 方法逐块返回 LLM 输出：

async def answer_astream(
    self,
    question: str,
    chat_history: ChatHistory,
    user_metadata: Dict[str, Any],
    # ... 其他参数
) -> AsyncIterator[ParsedRAGChunkResponse]:

流式响应包含以下数据块结构：

资料来源：core/quivr_core/rag/quivr_rag.py:20-45

完整处理流程图

sequenceDiagram
    participant U as 用户
    participant RAG as QuivrQARAG
    participant VS as VectorStore
    participant LLM as LLMEndpoint
    participant KB as 知识库

    U->>RAG: 发送查询问题
    RAG->>RAG: filter_history() 过滤历史
    
    alt 使用 LangGraph 工作流
        RAG->>RAG: rewrite_question() 重写问题
        RAG->>RAG: select_files() 选择文件
        RAG->>RAG: run_retrieval() 执行检索
    end
    
    RAG->>VS: 相似度搜索
    VS->>KB: 查询向量索引
    KB-->>VS: 返回相关文档
    VS-->>RAG: SearchResult 列表
    
    RAG->>RAG: compress_documents() 文档压缩
    RAG->>RAG: 构建提示词
    
    RAG->>LLM: 发送提示词
    LLM-->>RAG: 流式文本块
    
    loop 每个文本块
        RAG-->>U: ParsedRAGChunkResponse
    end
    
    RAG-->>U: 最终元数据块 (last_chunk=True)

文档处理管道

在文档被摄入系统时，处理器会执行以下转换步骤：

1. 文件解析 - 通过处理器提取文本内容
2. 语言检测 - 使用 detect_language 函数识别文档语言
3. 分块处理 - 将长文档分割为可管理的块
4. 元数据增强 - 添加 chunk_index、quivr_core_version 等信息
5. UTF-8 规范化 - 处理特殊字符和编码问题

doc.metadata = {
    "chunk_index": idx,
    "quivr_core_version": qvr_version,
    "language": detect_language(
        text=doc.page_content.replace("\\n", " ").replace("\n", " "),
        low_memory=True,
    ).value,
    **file.metadata,
    **doc.metadata,
    **self.processor_metadata,
}

资料来源：core/quivr_core/processor/processor_base.py:1-30

检索配置加载

支持从 YAML 配置文件加载检索配置：

retrieval_config = RetrievalConfig.from_yaml(config_file_name)

配置示例：

workflow_config:
  name: "standard RAG"
  nodes:
    - name: "START"
      edges: ["filter_history"]
    - name: "filter_history"
      edges: ["rewrite"]
    - name: "rewrite"
      edges: ["llm"]
  temperature: 0.7

资料来源：README.md:30-60

使用示例

基础 RAG 问答

from quivr_core import Brain

brain = Brain.from_files(
    name="my_brain",
    file_paths=["./document.pdf"],
)

answer = brain.ask("文档的主要内容是什么？")
print(answer.answer)

自定义检索配置

from quivr_core.config import RetrievalConfig

config = RetrievalConfig.from_yaml("./basic_rag_workflow.yaml")
answer = brain.ask(
    "问题内容",
    retrieval_config=config
)

总结

Quivr 的 RAG 处理流程通过模块化设计实现了高效的检索增强生成能力。核心流程包括聊天历史过滤、向量检索、文档重排序、提示词构建和大语言模型生成，最终通过流式接口返回响应结果。该架构支持灵活的配置和扩展，能够适应不同的应用场景需求。

概述

Brain（大脑）是 Quivr 项目的核心抽象概念，它代表了一个基于生成式 AI 的个人知识库助手系统。Brain 将文档处理、向量存储检索和大语言模型问答有机结合，为用户提供与私有文档进行自然语言交互的能力。

Quivr 的 Brain 采用 RAG（检索增强生成）架构，支持多种文件类型、多种 LLM 提供商，并允许用户通过 YAML 配置文件自定义检索策略和工作流程。

核心组件

Brain 类

Brain 类是整个系统的核心，位于 core/quivr_core/brain/brain.py。它封装了大脑的所有功能：

创建 Brain

#### 从文件创建

from quivr_core import Brain

brain = Brain.from_files(
    name="my_smart_brain",
    file_paths=["./my_first_doc.pdf", "./my_second_doc.txt"],
)

#### 异步创建

brain = await Brain.afrom_files(
    name="test_brain",
    file_paths=[temp_file.name]
)

#### 从 LangChain 文档创建

from langchain_core.documents import Document
from quivr_core import Brain

documents = [Document(page_content="Hello, world!")]
brain = await Brain.afrom_langchain_documents(
    name="My Brain",
    langchain_documents=documents
)

核心功能

问答功能 (ask)

Brain 的主要功能是回答用户关于已加载文档的问题：

answer = brain.ask("what is gold? answer in french")
print("answer:", answer)

问答流程如下：

graph TD
    A[用户问题] --> B[检索配置解析]
    B --> C{是否指定LLM}
    C -->|是| D[使用指定LLM]
    C -->|否| E[使用Brain默认LLM]
    D --> F[创建QuivrQARAGLangGraph]
    E --> F
    F --> G[向量数据库检索相关文档]
    G --> H[构建聊天历史]
    H --> I[调用LLM生成答案]
    I --> J[返回Answer对象]

异步流式问答 (ask_streaming)

支持流式输出，适合实时展示生成过程：

async for chunk in brain.ask_streaming("What is the meaning of life?"):
    print(chunk.answer)

文档检索 (asearch)

直接检索向量数据库中的相关文档块：

results = await brain.asearch(
    query="查询文本",
    n_results=5,
    fetch_n_neighbors=20
)

数据模型

Answer 对象

ask() 方法返回 Answer 对象，包含：

• answer: 生成的答案文本
• sources: 答案来源的文档块
• metadata: 额外的元数据信息

SearchResult 对象

asearch() 方法返回 SearchResult 列表，每个结果包含：

• content: 文档内容
• metadata: 文档元数据
• score: 相关性分数

配置管理

RetrievalConfig

检索配置控制 RAG 工作流的行为：

from quivr_core.config import RetrievalConfig

retrieval_config = RetrievalConfig.from_yaml(config_file_name)

answer = brain.ask(
    question,
    retrieval_config=retrieval_config
)

YAML 工作流配置

workflow_config:
  name: "standard RAG"
  nodes:
    - name: "START"
      edges: ["filter_history"]
    - name: "filter_history"
      edges: ["rewrite"]
    - name: "rewrite"
      edges: ["retrieval"]

持久化与加载

保存 Brain

save_path = await brain.save("/home/user/.local/quivr")

保存机制使用 SHA-1 哈希检测重复文件，避免存储冗余：

# 资料来源：core/quivr_core/storage/local_storage.py:45-52
if file.file_sha1 in self.hashes and not exists_ok:
    raise FileExistsError(f"file {file.original_filename} already uploaded")

加载 Brain

brain_loaded = Brain.load(save_path)
brain_loaded.print_info()

高级配置

自定义 LLM

from quivr_core.rag.entities.config import LLMEndpointConfig
from quivr_core.llm.llm_endpoint import LLMEndpoint
from langchain_core.language_models import FakeListChatModel

brain = Brain.from_files(
    name="test_brain",
    file_paths=["tests/processor/data/dummy.pdf"],
    llm=LLMEndpoint(
        llm=FakeListChatModel(responses=["good"]),
        llm_config=LLMEndpointConfig(model="fake_model", llm_base_url="local"),
    ),
    embedder=DeterministicFakeEmbedding(size=20),
)

自定义解析器

processor_kwargs = {
    "megaparse_config": megaparse_config,
    "splitter_config": assistant_config.ingestion_config.parser_config.splitter_config,
}

brain = await Brain.afrom_files(
    name="test_brain",
    file_paths=file_path,
    processor_kwargs=processor_kwargs,
)

支持的文件类型

Quivr Brain 支持处理多种文件类型：

• PDF 文档
• TXT 文本文件
• Markdown 文件
• 支持自定义解析器扩展

信息查看

查看 Brain 的详细信息：

brain.print_info()

系统架构图

graph TB
    subgraph "用户层"
        A[用户提问]
        B[流式输出]
    end
    
    subgraph "Brain 核心"
        C[Brain.ask]
        D[RetrievalConfig]
        E[QuivrQARAGLangGraph]
        F[RAG 工作流]
    end
    
    subgraph "检索层"
        G[向量数据库]
        H[嵌入模型]
        I[元数据过滤]
    end
    
    subgraph "生成层"
        J[LLM Endpoint]
        K[聊天历史]
        L[系统提示词]
    end
    
    A --> C
    C --> D
    D --> E
    E --> F
    F --> G
    F --> H
    F --> I
    F --> J
    J --> B

最佳实践

1. 选择合适的嵌入模型：嵌入模型直接影响检索质量
2. 配置检索参数：根据文档复杂度调整 n_results 和 fetch_n_neighbors
3. 使用 YAML 配置：通过配置文件管理复杂的工作流，便于调整和版本控制
4. 异步处理大文件：使用 afrom_files() 和 ask_streaming() 提升性能

概述

Quivr 的文件处理系统是构建检索增强生成（RAG）工作流的核心组件，负责将各种格式的文档转换为可供大语言模型（LLM）处理的文本块（chunks）。该系统采用插件化架构，通过处理器注册表动态加载和支持多种文件类型，支持包括 PDF、Word、文本、CSV、HTML、Markdown 等常见文档格式。

系统的设计目标是将非结构化文档内容标准化为统一的 Document 对象格式，同时通过文本分割器将长文档切分为适合嵌入和检索的语义块。这一处理流程确保了后续的向量数据库索引和问答系统能够高效准确地工作。

核心架构

系统组件

graph TD
    A[QuivrFile] --> B[ProcessorRegistry]
    B --> C[ProcessorBase]
    C --> D[process_file]
    D --> E[process_file_inner]
    E --> F[ProcessedDocument]
    F --> G[Document Chunks]
    G --> H[向量数据库]
    I[Language Detection] --> F
    J[Metadata Enrichment] --> F

处理流程

文件处理的核心流程包含以下步骤：

1. 文件验证：检查文件扩展名是否在支持列表中
2. 内容提取：调用具体处理器的 process_file_inner 方法提取文本
3. 语言检测：使用 detect_language 函数自动识别文档语言
4. 元数据丰富：添加处理版本、块索引、语言标识等元数据
5. 文本分割：按配置的 SplitterConfig 将文档分割为块
6. 格式标准化：处理 Unicode 特殊字符和编码问题

ProcessorBase 基类

类定义与接口

ProcessorBase 是所有文件处理器的抽象基类，定义在 processor_base.py 中：

class ProcessorBase(ABC, Generic[R]):
    supported_extensions: list[FileExtension | str]
    
    @property
    @abstractmethod
    def processor_metadata(self) -> dict[str, Any]:
        raise NotImplementedError
    
    async def process_file(self, file: QuivrFile) -> ProcessedDocument[R]:
        # 处理逻辑

ProcessedDocument 数据结构

处理器返回的 ProcessedDocument 包含三个核心字段：

元数据处理逻辑

处理过程中会自动丰富每个文档块的元数据：

doc.metadata = {
    "chunk_index": idx,           # 块在文档中的位置
    "quivr_core_version": qvr_version,  # 版本信息
    "language": detect_language(...).value,  # 检测到的语言
    **file.metadata,            # 文件原始元数据
    **doc.metadata,             # 文档自身元数据
    **self.processor_metadata,  # 处理器特定元数据
}

资料来源：processor_base.py:30-50

处理器注册表

注册机制

registry.py 实现了动态处理器加载机制，支持按优先级回退：

def fetch_processor_class(file_extension: FileExtension) -> Type[ProcessorBase]:
    if file_extension not in registry:
        entries = known_processors[file_extension]
        while entries:
            proc_entry = heappop(entries)
            try:
                register_processor(file_extension, _import_class(proc_entry.cls_mod))
                break
            except ImportError:
                logger.warn(f"{proc_entry.err}. Falling to next...")

优先级回退策略

当某个处理器因缺少依赖无法加载时，系统自动尝试下一个候选处理器：

资料来源：registry.py:30-55

文本分割器配置

SplitterConfig 配置项

分割器通过 SplitterConfig 数据类配置：

资料来源：splitter.py

分割器实现

Quivr 使用 LangChain 的 RecursiveCharacterTextSplitter 作为默认分割器，该分割器按优先级尝试按不同分隔符分割文本，确保语义完整性。

处理器实现

DefaultProcessor

默认处理器基于 LangChain 社区的文档加载器构建，支持多种格式：

处理器类通过动态工厂模式构建，每个类包含支持的扩展名列表和对应的分割器配置。

资料来源：implementations/default.py:15-35

TikaProcessor

Tika 处理器基于 Apache Tika 引擎，提供更强大的 PDF 解析能力：

class TikaProcessor(ProcessorBase):
    supported_extensions = [FileExtension.pdf]

Tika 能够提取 PDF 的文本内容、元数据并处理复杂的文档结构。

MegaparseProcessor

Megaparse 处理器是 Quivr 官方开发的高性能解析器，针对大规模文档处理优化：

class MegaparseProcessor(ProcessorBase):
    supported_extensions = [FileExtension.pdf, FileExtension.txt]
    
    @property
    def processor_metadata(self) -> dict[str, Any]:
        return {
            "processor_cls": "MegaparseProcessor",
            "parser": "megaparse",
        }

使用示例

基本用法

使用 Brain.from_files 方法时，系统自动选择合适的处理器：

from quivr_core import Brain

brain = Brain.from_files(
    name="test_brain",
    file_paths=["./document.pdf", "./notes.txt"],
)
answer = brain.ask("What is the main topic?")

指定 Tika 处理器解析 PDF

from langchain_core.embeddings import DeterministicFakeEmbedding
from langchain_core.language_models import FakeListChatModel
from quivr_core import Brain
from quivr_core.rag.entities.config import LLMEndpointConfig
from quivr_core.llm.llm_endpoint import LLMEndpoint
from quivr_core.processor.implementations.tika_processor import TikaProcessor

brain = Brain.from_files(
    name="pdf_brain",
    file_paths=["document.pdf"],
    processor=TikaProcessor(),
    llm=LLMEndpoint(
        llm=FakeListChatModel(responses=["good"]),
        llm_config=LLMEndpointConfig(model="fake_model", llm_base_url="local"),
    ),
    embedder=DeterministicFakeEmbedding(size=20),
)

资料来源：examples/pdf_parsing_tika.py

指定 Megaparse 处理器

from quivr_core import Brain
from quivr_core.processor.implementations megaparse_processor import MegaparseProcessor

brain = Brain.from_files(
    name="megaparse_brain",
    file_paths=["document.pdf"],
    processor=MegaparseProcessor(),
)

扩展自定义处理器

实现步骤

1. 继承 ProcessorBase 基类
2. 定义 supported_extensions 列表
3. 实现 process_file_inner 异步方法
4. 实现 processor_metadata 属性

示例代码

from quivr_core.processor.processor_base import ProcessedDocument, ProcessorBase
from quivr_core.files.file import QuivrFile
from langchain_core.documents import Document

class CustomProcessor(ProcessorBase):
    supported_extensions = [".custom"]
    
    @property
    def processor_metadata(self) -> dict[str, Any]:
        return {"processor_cls": "CustomProcessor"}
    
    async def process_file_inner(self, file: QuivrFile) -> ProcessedDocument[str]:
        # 实现自定义解析逻辑
        content = await self.extract_content(file)
        return ProcessedDocument(
            chunks=[Document(page_content=content, metadata={})],
            processor_cls="CustomProcessor",
            processor_response=content,
        )

错误处理

常见错误类型

验证机制

处理器在执行前会验证文件类型：

def check_supported(self, file: QuivrFile) -> None:
    if file.file_extension not in self.supported_extensions:
        raise ValueError(f"can't process a file of type {file.file_extension}")

资料来源：processor_base.py:35-38

总结

Quivr 的文件处理系统通过模块化设计和动态加载机制，提供了灵活且可扩展的文档处理能力。处理器注册表配合优先级回退策略确保了系统的鲁棒性，而统一的 ProcessedDocument 接口使得不同处理器的输出可以被后续组件无缝使用。开发者可以通过实现 ProcessorBase 接口轻松添加新的文件格式支持。

概述

Quivr 的存储系统是整个 RAG（检索增强生成）框架的核心组件之一，负责管理文件的存取、元数据管理以及文件到向量数据库的转换流程。存储系统采用了抽象基类设计模式，支持本地存储和透明存储两种实现方式，同时允许开发者通过实现 StorageBase 接口来创建自定义存储后端。

存储系统的主要职责包括：

• 文件的读取与写入操作
• 文件元数据的提取与管理（包括文件大小、SHA1 哈希值、文件扩展名等）
• 支持异步文件操作以提高性能
• 为处理器提供统一的文件访问接口

资料来源：core/quivr_core/brain/brain.py

核心架构

存储系统类层次结构

graph TD
    A[StorageBase] --> B[LocalStorage]
    A --> C[TransparentStorage]
    A --> D[Custom Storage]
    
    E[QuivrFile] --> F[文件元数据]
    G[process_files] --> A
    H[Brain] --> G

核心组件

资料来源：core/quivr_core/storage/storage_base.py

StorageBase 抽象基类

StorageBase 是所有存储实现的基类，定义了存储系统的核心接口规范。

主要接口方法

抽象方法定义

class StorageBase(ABC):
    @abstractmethod
    async def get_files(self) -> AsyncGenerator[QuivrFile, None]:
        raise NotImplementedError
    
    @abstractmethod
    async def add_file(self, file: QuivrFile) -> None:
        raise NotImplementedError

资料来源：core/quivr_core/storage/storage_base.py

QuivrFile 文件元数据类

QuivrFile 是用于表示存储中文件的数据类，使用 __slots__ 优化内存占用。

数据结构

class QuivrFile:
    __slots__ = [
        "id",           # UUID 文件唯一标识
        "brain_id",     # UUID 所属脑部ID
        "path",         # Path 文件路径
        "original_filename",  # str 原始文件名
        "file_size",    # int 文件大小
        "file_extension",     # FileExtension 文件扩展名
        "file_sha1",    # str SHA1哈希值
        "additional_metadata", # dict 额外元数据
    ]

构造函数参数

资料来源：core/quivr_core/files/file.py

异步文件打开

QuivrFile 提供了异步上下文管理器用于安全地打开文件：

@asynccontextmanager
async def open(self) -> AsyncGenerator[AsyncIterable[bytes], None]:
    f = await aiofiles.open(self.path, mode="rb")
    try:
        yield f
    finally:
        await f.close()

资料来源：core/quivr_core/files/file.py

LocalStorage 本地存储实现

LocalStorage 是基于本地文件系统的存储实现，适用于单节点部署场景。

文件处理流程

graph TD
    A[开始处理] --> B{文件是否存在}
    B -->|不存在| C[抛出 ValueError]
    B -->|存在| D[提取文件扩展名]
    D --> E[计算 SHA1 哈希]
    E --> F[计算文件大小]
    F --> G[创建 QuivrFile 对象]
    G --> H[添加到文件列表]
    H --> I[返回 QuivrFile]

核心功能

资料来源：core/quivr_core/storage/local_storage.py

TransparentStorage 透明存储

TransparentStorage 是一个存储包装器，用于在不改变现有接口的情况下扩展存储功能。

使用场景

• 日志记录：透明地记录所有存储操作
• 缓存层：在存储操作前后添加缓存逻辑
• 访问控制：添加权限检查逻辑

资料来源：core/quivr_core/rag/quivr_rag_langgraph.py

文件处理集成

存储系统与处理器系统紧密集成，形成完整的文件处理流水线：

graph LR
    A[StorageBase] -->|get_files| B[process_files]
    B -->|QuivrFile| C[get_processor_class]
    C -->|ProcessorBase| D[process_file]
    D -->|Document| E[向量数据库]

异步处理流程

async def process_files(
    storage: StorageBase, 
    skip_file_error: bool, 
    **processor_kwargs: dict[str, Any]
) -> list[Document]:

处理流程说明：

1. 遍历存储中的所有文件
2. 根据文件扩展名获取对应的处理器
3. 调用处理器提取文档内容
4. 返回 LangChain Document 列表

资料来源：core/quivr_core/brain/brain.py

自定义存储实现

开发者可以通过继承 StorageBase 来实现自定义存储后端。

实现步骤

1. 创建继承自 StorageBase 的新类
2. 实现 get_files() 异步生成器方法
3. 实现 add_file() 异步方法
4. 在 Brain 初始化时传入自定义存储实例

示例结构

class CustomStorage(StorageBase):
    async def get_files(self) -> AsyncGenerator[QuivrFile, None]:
        # 实现自定义文件获取逻辑
        ...
    
    async def add_file(self, file: QuivrFile) -> None:
        # 实现自定义文件添加逻辑
        ...

资料来源：examples/custom_storage.md

配置与使用

在 Brain 中使用存储

from quivr_core import Brain
from quivr_core.storage.local_storage import LocalStorage

# 使用默认本地存储
brain = Brain.from_files(
    name="my_brain",
    file_paths=["./documents/"]
)

# 使用自定义存储
brain = Brain.from_files(
    name="my_brain",
    file_paths=[],
    storage=CustomStorage(...)
)

资料来源：README.md

最佳实践

性能优化

错误处理

存储系统在处理文件时会进行多种校验：

• 文件扩展名识别
• 文件 SHA1 哈希计算
• 文件大小统计
• 路径有效性验证

资料来源：core/quivr_core/files/file.py

总结

Quivr 的存储系统采用模块化设计，通过抽象基类 StorageBase 提供了灵活的可扩展性。开发者可以根据实际需求选择本地存储、透明存储或实现自定义存储后端。存储系统与处理器系统的紧密集成使得文件到向量数据库的转换过程透明且高效。

架构概述

Quivr 的 LLM 集成采用分层架构设计，通过 LLMEndpoint 类封装底层模型调用，提供统一的接口供 RAG 工作流使用。

graph TD
    subgraph "LLM 集成层"
        A[LLMEndpoint] --> B[LLMEndpointConfig]
        A --> C[BaseChatModel]
    end
    
    subgraph "RAG 工作流"
        D[QuivrRAG] --> A
        E[QuivrRAGLangGraph] --> A
    end
    
    subgraph "模型提供商"
        F[OpenAI] --> A
        G[Anthropic] --> A
        H[Mistral] --> A
        I[Ollama 本地模型] --> A
    end

核心组件

LLMEndpoint

LLMEndpoint 是 LLM 集成的核心类，位于 core/quivr_core/llm/llm_endpoint.py。它封装了 LangChain 的聊天模型，提供配置管理和调用接口。

from quivr_core.llm.llm_endpoint import LLMEndpoint
from quivr_core.rag.entities.config import LLMEndpointConfig

llm = LLMEndpoint(
    llm=chat_model_instance,
    llm_config=LLMEndpointConfig(
        model="gpt-4",
        temperature=0.7,
        max_tokens=2000
    )
)

LLMEndpointConfig 配置参数

资料来源：core/quivr_core/rag/entities/config.py:1-100

配置管理

默认 LLM 配置

Quivr 提供了默认 LLM 工厂函数，简化常见场景下的配置：

from quivr_core.llm import default_llm

# 使用默认配置创建 LLM
llm = default_llm()

环境变量配置

LLM 集成支持通过环境变量配置 API 密钥：

import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-key"
os.environ["MISTRAL_API_KEY"] = "your-mistral-key"

模型支持

Quivr 的 LLM 集成支持多种模型提供商：

支持的提供商

本地模型支持（Ollama）

Quivr 支持使用 Ollama 运行本地大语言模型：

from langchain_ollama import ChatOllama

llm = ChatOllama(
    model="llama2",
    base_url="http://localhost:11434"
)

config = LLMEndpointConfig(
    model="llama2",
    llm_base_url="http://localhost:11434"
)

endpoint = LLMEndpoint(llm=llm, llm_config=config)

RAG 中的 LLM 集成

QuivrRAG 类

QuivrRAG 类是核心 RAG 实现，通过 LLMEndpoint 进行问答生成：

sequenceDiagram
    User->>QuivrRAG: ask(question)
    QuivrRAG->>VectorDB: search(query, n_results)
    VectorDB-->>QuivrRAG: relevant_chunks
    QuivrRAG->>LLMEndpoint: generate(prompt)
    LLMEndpoint->>LLM: chat.completions.create
    LLM-->>LLMEndpoint: response
    LLMEndpoint-->>QuivrRAG: ParsedRAGChunkResponse
    QuivrRAG-->>User: RAGResponse

流式响应处理

LLM 集成支持流式输出，通过 answer_astream 方法实现：

async for chunk in brain.asearch_stream(question):
    print(chunk.answer, end="", flush=True)

流式响应的关键流程：

1. 逐块接收 LLM 返回的响应
2. 实时解析并格式化输出
3. 最后一块包含完整的元数据

资料来源：core/quivr_core/rag/quivr_rag.py:100-200

工作流节点配置

LLM 集成可配置在工作流节点中，通过 YAML 文件定义：

workflow_config:
  name: "standard RAG"
  nodes:
    - name: "START"
      edges: ["filter_history"]
    - name: "filter_history"
      edges: ["rewrite"]
    - name: "rewrite"
      edges: ["retrieval"]

RetrievalConfig 配置

资料来源：core/quivr_core/rag/entities/config.py

提示词模板

Quivr 的 LLM 集成使用结构化的提示词模板系统：

模板类型

自定义提示词

可以通过 custom_prompts 配置自定义提示词：

from quivr_core.rag.prompts import custom_prompts, TemplatePromptName

custom_prompts[TemplatePromptName.DEFAULT_ANSWER] = {
    "prompt": "自定义提示词模板...",
    "guidelines": "自定义指南..."
}

资料来源：core/quivr_core/rag/prompts.py

LangChain 集成

Quivr 的 LLM 集成深度集成 LangChain 生态系统：

LangChain 兼容性

from langchain_core.language_models import FakeListChatModel
from langchain_core.embeddings import DeterministicFakeEmbedding
from quivr_core import Brain

# 使用 Fake LLM 进行测试
brain = Brain.from_files(
    name="test_brain",
    file_paths=["./docs.pdf"],
    llm=LLMEndpoint(
        llm=FakeListChatModel(responses=["测试回答"]),
        llm_config=LLMEndpointConfig(model="fake_model", llm_base_url="local")
    ),
    embedder=DeterministicFakeEmbedding(size=20)
)

Langfuse 追踪集成

LLM 调用支持 Langfuse 追踪，便于性能监控和调试：

from quivr_core.rag.utils import LangfuseService

langfuse_service = LangfuseService()
langfuse_handler = langfuse_service.get_handler()

资料来源：core/quivr_core/rag/quivr_rag_langgraph.py

高级配置

模型参数覆盖

在 Brain.ask() 调用时可以覆盖默认配置：

answer = brain.ask(
    question,
    retrieval_config=RetrievalConfig(
        max_tokens=1000,
        temperature=0.5
    )
)

多模型路由

Quivr 支持在工作流中配置多模型路由：

# 为不同节点指定不同模型
config = RetrievalConfig(
    model_configs={
        "rewrite": {"model": "gpt-3.5-turbo", "temperature": 0.3},
        "answer": {"model": "gpt-4", "temperature": 0.7}
    }
)

错误处理

LLM 集成内置了完整的错误处理机制：

最佳实践

1. 生产环境配置

config = LLMEndpointConfig(
    model="gpt-4-turbo",
    temperature=0.3,  # 降低随机性
    max_tokens=2000,
    streaming=True
)

2. 成本优化

• 使用 gpt-3.5-turbo 处理简单查询
• 启用流式输出减少感知延迟
• 合理设置 max_tokens 避免浪费

3. 安全性

• 使用环境变量存储 API 密钥
• 避免在日志中记录敏感信息
• 实施请求频率限制

总结

Quivr 的 LLM 集成模块提供了灵活、高效的大语言模型集成方案。通过统一的 LLMEndpoint 接口，开发者可以轻松切换不同的模型提供商，同时保持 RAG 工作流的稳定性。深度集成的 LangChain 支持和完善的配置管理使 Quivr 成为构建智能问答系统的理想选择。

概述

Quivr 的工具扩展系统是实现 RAG（检索增强生成）工作流自定义的核心组件，允许用户为 LLM（大语言模型）集成额外的工具能力，如网络搜索、API 调用等。根据项目文档，Quivr 支持"添加工具"（add tools）功能，使用户能够扩展基础 RAG 的能力范围。

资料来源：README.md:1-50

架构设计

工具系统与 RAG 的关系

工具扩展系统与 QuivrQARAG 类紧密集成。根据源码设计，QuivrQARAG 接受 RetrievalConfig 配置，并支持在检索过程中注入重排序（reranker）等扩展组件。这种模块化设计使得工具系统可以无缝嵌入 RAG 工作流。

graph TD
    A[用户查询] --> B[QuivrQARAG]
    B --> C{检索配置}
    C --> D[向量存储检索]
    C --> E[历史过滤]
    C --> F[查询改写]
    D --> G[工具扩展层]
    G --> H[LLM 端点]
    H --> I[生成回答]
    
    F -->|可选| J[网络搜索工具]
    G -->|可扩展| K[自定义工具]

资料来源：core/quivr_core/rag/quivr_rag.py:50-80

核心组件结构

根据项目架构设计，工具扩展系统涉及以下核心组件：

资料来源：core/quivr_core/brain/brain.py:1-100

配置与使用

基础配置流程

用户可以通过 YAML 配置文件定义工具扩展的工作流程：

workflow_config:
  name: "standard RAG"
  nodes:
    - name: "START"
      edges: ["filter_history"]
    - name: "filter_history"
      edges: ["rewrite"]
    - name: "rewrite"
      edges: ["END"]

资料来源：README.md:1-100

LLM 端点配置

工具系统支持多种 LLM 提供商，包括：

• OpenAI：通过 OPENAI_API_KEY 环境变量配置
• Anthropic：支持 Claude 系列模型
• Mistral：支持 Mistral AI 模型
• Ollama：支持本地模型部署

from quivr_core.config import RetrievalConfig

config_file_name = "./basic_rag_workflow.yaml"
retrieval_config = RetrievalConfig.from_yaml(config_file_name)

资料来源：README.md:1-80

工作流程集成

问答流程

工具扩展系统在 Brain.ask() 方法中被调用，完整流程如下：

1. 历史过滤（filter_history）：过滤相关聊天历史
2. 查询改写（rewrite）：优化用户查询
3. 检索（retrieve）：从向量数据库获取相关内容
4. 工具执行（可选）：如网络搜索等扩展工具
5. 生成回答：LLM 基于检索结果和工具输出生成答案

graph LR
    A[用户提问] --> B[filter_history]
    B --> C[rewrite]
    C --> D{是否需要工具}
    D -->|是| E[执行工具]
    D -->|否| F[直接检索]
    E --> F
    F --> G[LLM 生成]
    G --> H[返回答案]

资料来源：core/quivr_core/rag/quivr_rag.py:100-150

异步搜索接口

Brain 类提供了异步搜索方法，支持灵活的工具扩展集成：

async def asearch(
    self,
    query: str | Document,
    n_results: int = 5,
    filter: Callable | Dict[str, Any] | None = None,
    fetch_n_neighbors: int = 20,
) -> list[SearchResult]

资料来源：core/quivr_core/brain/brain.py:100-120

扩展机制

自定义工具注册

根据项目架构设计，工具系统支持用户注册自定义处理器和加载器：

from quivr_core.processor.processor_base import ProcessorBase
from quivr_core.files.file import FileExtension, QuivrFile

class CustomProcessor(ProcessorBase):
    supported_extensions = [".custom"]
    
    async def process_file_inner(self, file: QuivrFile) -> ProcessedDocument:
        # 自定义处理逻辑
        pass

资料来源：core/quivr_core/processor/implementations/default.py:1-50

支持的文件格式

工具系统通过处理器基类支持多种文件格式：

资料来源：core/quivr_core/processor/implementations/default.py:10-30

版本与兼容性

工具扩展系统随 quivr-core 包发布，版本信息记录在文档元数据中：

qvr_version = version("quivr-core")
doc.metadata = {
    "chunk_index": idx,
    "quivr_core_version": qvr_version,
    "language": detect_language(text=...).value,
    **file.metadata,
    **doc.metadata,
    **self.processor_metadata,
}

资料来源：core/quivr_core/processor/processor_base.py:20-40

快速开始

完整示例

import tempfile
from quivr_core import Brain
from quivr_core.config import RetrievalConfig

# 创建 Brain
brain = Brain.from_files(
    name="my_brain",
    file_paths=["./documents/"],
)

# 配置检索（含工具扩展）
retrieval_config = RetrievalConfig.from_yaml("./config.yaml")

# 提问
answer = brain.ask(
    "您的技术问题",
    retrieval_config=retrieval_config
)

print(answer.answer)

资料来源：README.md:40-70

总结

Quivr 的工具扩展系统通过模块化设计，将 LLM 工具能力与 RAG 工作流深度整合。系统支持自定义工具注册、多 LLM 提供商集成，以及灵活的工作流配置，为构建智能问答应用提供了强大的扩展基础。开发者可以通过继承 ProcessorBase 类或配置 YAML 工作流文件来扩展系统能力。

概述

工作流配置（Workflow Configuration）是 Quivr Core 框架中用于定义 RAG（检索增强生成）流程的核心机制。通过 YAML 格式的配置文件，用户可以灵活地定义检索、生成和对话历史处理的各个环节，实现高度定制化的问答系统。

工作流配置的主要作用包括：

• 定义处理流程节点：指定 RAG 流程中的各个处理阶段（如历史过滤、问题改写、文档检索、答案生成）
• 配置节点间连接：通过边（edges）定义节点之间的执行顺序和数据流向
• 控制对话历史：设置历史消息的过滤策略和最大上下文长度
• 配置重排序器：选择重排序模型以提升检索结果质量
• 调整 LLM 参数：控制生成答案的最大输入令牌数和温度参数

配置结构

工作流配置文件采用 YAML 格式，整体结构如下：

workflow_config:
  name: "工作流名称"
  nodes:
    - name: "节点名称"
      edges: ["下一节点名称"]

max_history: 10

reranker_config:
  supplier: "cohere"
  model: "rerank-multilingual-v3.0"
  top_n: 5

llm_config:
  max_input_tokens: 4000
  temperature: 0.7

核心配置项

workflow_config（工作流配置）

定义 RAG 处理流程的节点拓扑结构。

节点定义

节点名称必须是预定义的系统节点，当前支持的节点包括：

• START：流程起始节点
• filter_history：对话历史过滤节点
• rewrite：问题改写节点
• retrieve：文档检索节点
• generate_rag：RAG 答案生成节点（通常是最后一个节点）
• END：流程结束节点

资料来源：examples/chatbot/basic_rag_workflow.yaml:1-22

max_history（历史消息配置）

控制对话历史中包含的最大迭代次数。

此参数用于限制传入 LLM 的历史消息数量，防止上下文过长导致令牌浪费和性能下降。

reranker_config（重排序配置）

配置文档重排序器以提升检索结果的相关性。

当前支持的重排序供应商包括 cohere，模型推荐使用 rerank-multilingual-v3.0。

资料来源：examples/chatbot/basic_rag_workflow.yaml:24-32

llm_config（语言模型配置）

配置用于答案生成的大语言模型参数。

• temperature 越低，答案越确定性；越高，答案越有创造性
• max_input_tokens 应根据 LLM 的上下文窗口和文档数量合理设置

工作流执行流程

标准 RAG 流程图

graph TD
    START([START]) --> FILTER[filter_history]
    FILTER --> REWRITE[rewrite]
    REWRITE --> RETRIEVE[retrieve]
    RETRIEVE --> GENERATE[generate_rag]
    GENERATE --> END([END])
    
    style START fill:#90EE90
    style END fill:#FFB6C1

各节点功能详解

#### filter_history（历史过滤）

该节点负责过滤和精简对话历史，只保留最近 N 轮对话内容。配置参数为 max_history，用于控制保留的历史对话对数。

资料来源：core/quivr_core/rag/quivr_rag.py:120-140

#### rewrite（问题改写）

将用户当前问题与对话历史结合，改写为独立的、上下文无关的问题，以便更好地进行文档检索。这一步确保即使问题依赖历史上下文，也能准确检索相关文档。

#### retrieve（文档检索）

基于改写后的问题，从向量数据库中检索最相关的文档块。检索结果的数量由 reranker_config.top_n 控制。

#### generate_rag（答案生成）

将检索到的文档块与问题、历史上下文结合，调用 LLM 生成最终答案。最终节点的输出会流式返回给用户。

配置类定义

工作流配置在代码中由以下数据类定义：

class RetrievalConfig(BaseModel):
    max_history: int = 10
    prompt: Optional[str] = None
    reranker_config: Optional[RerankerConfig] = None
    llm_config: Optional[LLMConfig] = None

资料来源：core/quivr_core/rag/entities/config.py

RerankerConfig（重排序配置类）

class RerankerConfig(BaseModel):
    supplier: str
    model: str
    top_n: int

LLMConfig（语言模型配置类）

class LLMConfig(BaseModel):
    max_input_tokens: int
    temperature: float

NodeConfig（节点配置类）

class NodeConfig(BaseModel):
    name: str
    edges: List[str]

使用示例

基本 RAG 工作流配置

创建文件 basic_rag_workflow.yaml：

workflow_config:
  name: "standard RAG"
  nodes:
    - name: "START"
      edges: ["filter_history"]

    - name: "filter_history"
      edges: ["rewrite"]

    - name: "rewrite"
      edges: ["retrieve"]

    - name: "retrieve"
      edges: ["generate_rag"]

    - name: "generate_rag"
      edges: ["END"]

max_history: 10

reranker_config:
  supplier: "cohere"
  model: "rerank-multilingual-v3.0"
  top_n: 5

llm_config:
  max_input_tokens: 4000
  temperature: 0.7

资料来源：examples/chatbot/basic_rag_workflow.yaml:1-32

在代码中使用配置

from quivr_core import Brain
from quivr_core.config import RetrievalConfig

config_file_name = "./basic_rag_workflow.yaml"
retrieval_config = RetrievalConfig.from_yaml(config_file_name)

brain = Brain.from_files(
    name="my_smart_brain",
    file_paths=["./my_first_doc.pdf", "./my_second_doc.txt"],
)

answer = brain.ask(
    "what is gold?",
    retrieval_config=retrieval_config
)

语音聊天工作流配置

语音聊天场景的工作流配置与基本 RAG 类似，主要区别在于支持音频输入和语音合成：

workflow_config:
  name: "voice RAG"
  nodes:
    - name: "START"
      edges: ["filter_history"]

    - name: "filter_history"
      edges: ["rewrite"]

    - name: "rewrite"
      edges: ["retrieve"]

    - name: "retrieve"
      edges: ["generate_rag"]

    - name: "generate_rag"
      edges: ["END"]

max_history: 10

reranker_config:
  supplier: "cohere"
  model: "rerank-multilingual-v3.0"
  top_n: 5

llm_config:
  max_input_tokens: 4000
  temperature: 0.7

资料来源：examples/chatbot_voice/basic_rag_workflow.yaml:1-32

高级配置：自定义提示词

除了系统默认的提示词模板外，用户可以通过 llm_config.prompt 参数添加自定义指令：

llm_config:
  max_input_tokens: 4000
  temperature: 0.7
  prompt: |
    请用简洁的语言回答，并在答案开头用一句话总结。

自定义提示词会与系统默认指令合并，系统会在生成答案时综合考虑两者的要求。

LangGraph 高级工作流

Quivr 还支持基于 LangGraph 的高级工作流，通过 quivr_rag_langgraph.py 模块实现：

from quivr_core.rag.quivr_rag_langgraph import QuivrRAGLangGraph

rag = QuivrRAGLangGraph(
    retrieval_config=retrieval_config,
    brain=brain,
)

LangGraph 工作流支持更复杂的条件分支和状态管理，适合需要多步骤推理的应用场景。

资料来源：core/quivr_core/rag/quivr_rag_langgraph.py

最佳实践

1. 合理设置 max_history

• 低值（1-3）：适用于简单问答场景，减少上下文长度
• 中值（5-10）：适用于需要多轮对话的应用
• 高值（15+）：适用于长对话分析，但会增加令牌消耗

2. 优化 top_n 参数

• 文档简短：可设置较高值（10-20）
• 文档冗长：建议设置较低值（3-5），减少上下文干扰

3. 调整 temperature

• 事实性问答：建议 0.1-0.3
• 创意性生成：建议 0.7-0.9
• 平衡场景：建议 0.5 左右

4. 配置加载方式

支持从 YAML 文件或 Python 代码直接配置：

# 从 YAML 加载
config = RetrievalConfig.from_yaml("config.yaml")

# 直接创建
config = RetrievalConfig(
    max_history=5,
    reranker_config=RerankerConfig(
        supplier="cohere",
        model="rerank-multilingual-v3.0",
        top_n=5
    ),
    llm_config=LLMConfig(
        max_input_tokens=4000,
        temperature=0.7
    )
)

相关文件

Pitfall Log / 踩坑日志

项目：QuivrHQ/quivr

摘要：发现 17 个潜在踩坑项，其中 2 个为 high/blocking；最高优先级：安全/权限坑 - 来源证据：EU AI Act Compliance Scan Results — Sharing Findings for Feedback。

1. 安全/权限坑 · 来源证据：EU AI Act Compliance Scan Results — Sharing Findings for Feedback

• 严重度：high
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：EU AI Act Compliance Scan Results — Sharing Findings for Feedback
• 对用户的影响：可能影响授权、密钥配置或安全边界。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_46118108af95480ba852109be6e2c66a | https://github.com/QuivrHQ/quivr/issues/3667 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

2. 安全/权限坑 · 来源证据：[Bug]:

• 严重度：high
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：[Bug]:
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_ff260ca3ab5247679b2ded201ec21e24 | https://github.com/QuivrHQ/quivr/issues/2004 | 来源讨论提到 docker 相关条件，需在安装/试用前复核。

3. 安装坑 · 来源证据：Integration idea: Screenpipe for screen/audio context

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Integration idea: Screenpipe for screen/audio context
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_3a7f345691d04bbea72b03858746645e | https://github.com/QuivrHQ/quivr/issues/3658 | 来源类型 github_issue 暴露的待验证使用条件。

4. 安装坑 · 来源证据：[Bug]: RuntimeError: There is no current event loop in thread 'MainThread' when using Brain.from_files() in script

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：[Bug]: RuntimeError: There is no current event loop in thread 'MainThread' when using Brain.from_files() in script
• 对用户的影响：可能阻塞安装或首次运行。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_e3091b23b6184dffb982e0cc494134fd | https://github.com/QuivrHQ/quivr/issues/3650 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

5. 能力坑 · 能力判断依赖假设

• 严重度：medium
• 证据强度：source_linked
• 发现：README/documentation is current enough for a first validation pass.
• 对用户的影响：假设不成立时，用户拿不到承诺的能力。
• 建议检查：将假设转成下游验证清单。
• 防护动作: 假设必须转成验证项；没有验证结果前不能写成事实。
• 证据：capability.assumptions | github_repo:640079149 | https://github.com/QuivrHQ/quivr | README/documentation is current enough for a first validation pass.

6. 运行坑 · 来源证据：The garbage collector is trying to clean up non-checked-in connection <AdaptedConnection <asyncpg.connection.Connection…

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：The garbage collector is trying to clean up non-checked-in connection <AdaptedConnection <asyncpg.connection.Connection object at 0x7f66a5b06110>>, which will…
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_e6fdde799a2b4f53806121190979025a | https://github.com/QuivrHQ/quivr/issues/3654 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

7. 运行坑 · 来源证据：core: v0.0.25

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：core: v0.0.25
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_7c6096fb5a6442c7b53068a8dd8e2c78 | https://github.com/QuivrHQ/quivr/releases/tag/core-0.0.25 | 来源类型 github_release 暴露的待验证使用条件。

8. 运行坑 · 来源证据：core: v0.0.29

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：core: v0.0.29
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_7eeeeb626a10476e820fcd651727a55a | https://github.com/QuivrHQ/quivr/releases/tag/core-0.0.29 | 来源讨论提到 node 相关条件，需在安装/试用前复核。

9. 运行坑 · 来源证据：core: v0.0.33

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：core: v0.0.33
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_d7a735cb41bc44f2abeb148d689f1320 | https://github.com/QuivrHQ/quivr/releases/tag/core-0.0.33 | 来源类型 github_release 暴露的待验证使用条件。

10. 维护坑 · 来源证据：core: v0.0.24

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题：core: v0.0.24
• 对用户的影响：可能影响升级、迁移或版本选择。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_5272fe70cda24220a5aeb994ad06819e | https://github.com/QuivrHQ/quivr/releases/tag/core-0.0.24 | 来源类型 github_release 暴露的待验证使用条件。

11. 维护坑 · 来源证据：core: v0.0.26

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题：core: v0.0.26
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_7f723b616fd446c3be43298d75fd6e8b | https://github.com/QuivrHQ/quivr/releases/tag/core-0.0.26 | 来源类型 github_release 暴露的待验证使用条件。

12. 维护坑 · 维护活跃度未知

• 严重度：medium
• 证据强度：source_linked
• 发现：未记录 last_activity_observed。
• 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
• 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
• 防护动作: 维护活跃度未知时，推荐强度不能标为高信任。
• 证据：evidence.maintainer_signals | github_repo:640079149 | https://github.com/QuivrHQ/quivr | last_activity_observed missing

13. 安全/权限坑 · 下游验证发现风险项

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：下游已经要求复核，不能在页面中弱化。
• 建议检查：进入安全/权限治理复核队列。
• 防护动作: 下游风险存在时必须保持 review/recommendation 降级。
• 证据：downstream_validation.risk_items | github_repo:640079149 | https://github.com/QuivrHQ/quivr | no_demo; severity=medium

14. 安全/权限坑 · 存在评分风险

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：风险会影响是否适合普通用户安装。
• 建议检查：把风险写入边界卡，并确认是否需要人工复核。
• 防护动作: 评分风险必须进入边界卡，不能只作为内部分数。
• 证据：risks.scoring_risks | github_repo:640079149 | https://github.com/QuivrHQ/quivr | no_demo; severity=medium

15. 安全/权限坑 · 来源证据：core: v0.0.27

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：core: v0.0.27
• 对用户的影响：可能影响授权、密钥配置或安全边界。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_027338ba02764da6b371970615591265 | https://github.com/QuivrHQ/quivr/releases/tag/core-0.0.27 | 来源类型 github_release 暴露的待验证使用条件。

16. 维护坑 · issue/PR 响应质量未知

• 严重度：low
• 证据强度：source_linked
• 发现：issue_or_pr_quality=unknown。
• 对用户的影响：用户无法判断遇到问题后是否有人维护。
• 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
• 防护动作: issue/PR 响应未知时，必须提示维护风险。
• 证据：evidence.maintainer_signals | github_repo:640079149 | https://github.com/QuivrHQ/quivr | issue_or_pr_quality=unknown

17. 维护坑 · 发布节奏不明确

• 严重度：low
• 证据强度：source_linked
• 发现：release_recency=unknown。
• 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
• 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
• 防护动作: 发布节奏未知或过期时，安装说明必须标注可能漂移。
• 证据：evidence.maintainer_signals | github_repo:640079149 | https://github.com/QuivrHQ/quivr | release_recency=unknown