|
# """ |
|
# This script provides a simplified interface for interacting with the quivr-core library. |
|
# It offers functions to easily create, query, save, and load Quivr brains, |
|
# making it more user-friendly for both direct Python scripting and agent integrations. |
|
# """ |
|
|
|
import os |
|
from pathlib import Path |
|
from quivr_core import Brain |
|
from quivr_core.config import IngestionConfig, RetrievalConfig |
|
|
|
def create_brain_from_files(name: str, file_paths: list[str], config_yaml: str = None) -> Brain: |
|
""" |
|
Creates a Quivr Brain from a list of file paths. |
|
|
|
Args: |
|
name (str): The name of the Brain. |
|
file_paths (list[str]): A list of file paths to process and add to the Brain. |
|
config_yaml (str, optional): Path to a YAML configuration file for ingestion. |
|
Defaults to None, using default configurations. |
|
|
|
Returns: |
|
Brain: The created Quivr Brain instance. |
|
""" |
|
processor_kwargs = {} |
|
if config_yaml: |
|
ingestion_config = IngestionConfig.from_yaml(config_yaml) |
|
processor_kwargs = { |
|
"megaparse_config": ingestion_config.parser_config.megaparse_config, |
|
"splitter_config": ingestion_config.parser_config.splitter_config, |
|
} |
|
|
|
brain = Brain.from_files( |
|
name=name, |
|
file_paths=file_paths, |
|
processor_kwargs=processor_kwargs |
|
) |
|
return brain |
|
|
|
def query_brain(brain: Brain, question: str, config_yaml: str = None) -> str: |
|
""" |
|
Queries a Quivr Brain with a question and returns the answer. |
|
|
|
Args: |
|
brain (Brain): The Quivr Brain instance to query. |
|
question (str): The question to ask. |
|
config_yaml (str, optional): Path to a YAML configuration file for retrieval. |
|
Defaults to None, using default configurations. |
|
|
|
Returns: |
|
str: The answer from the Brain. |
|
""" |
|
retrieval_config = None |
|
if config_yaml: |
|
retrieval_config = RetrievalConfig.from_yaml(config_yaml) |
|
|
|
answer = brain.ask(question, retrieval_config=retrieval_config) |
|
return answer.answer |
|
|
|
def save_brain(brain: Brain, folder_path: str) -> str: |
|
""" |
|
Saves a Quivr Brain to a specified folder path. |
|
|
|
Args: |
|
brain (Brain): The Quivr Brain instance to save. |
|
folder_path (str): The folder path where the Brain should be saved. |
|
|
|
Returns: |
|
str: The path to the saved Brain folder. |
|
""" |
|
saved_path = brain.save(folder_path) |
|
return saved_path |
|
|
|
def load_brain(folder_path: str) -> Brain: |
|
""" |
|
Loads a Quivr Brain from a specified folder path. |
|
|
|
Args: |
|
folder_path (str): The folder path from which to load the Brain. |
|
|
|
Returns: |
|
Brain: The loaded Quivr Brain instance. |
|
""" |
|
brain = Brain.load(folder_path) |
|
return brain |
|
|
|
# --- Tool-callable functions for agents --- |
|
|
|
def quivr_create_brain(brain_name: str, files: str, config_path: str = None) -> str: |
|
"""Tool: Creates a Quivr Brain from specified files. |
|
|
|
Args: |
|
brain_name (str): Name of the brain to create. |
|
files (str): Comma-separated list of file paths. |
|
config_path (str, optional): Path to ingestion config YAML file. Defaults to None. |
|
|
|
Returns: |
|
str: Message indicating success and brain name. |
|
""" |
|
file_list = files.split(',') |
|
brain = create_brain_from_files(brain_name, file_list, config_path) |
|
return f"Brain '{brain.name}' created successfully." |
|
|
|
def quivr_query(brain_folder: str, query_question: str, config_path: str = None) -> str: |
|
"""Tool: Queries a loaded Quivr Brain with a question. |
|
|
|
Args: |
|
brain_folder (str): Path to the folder containing the saved Brain. |
|
query_question (str): The question to ask the Brain. |
|
config_path (str, optional): Path to retrieval config YAML file. Defaults to None. |
|
|
|
Returns: |
|
str: The answer from the Quivr Brain. |
|
""" |
|
brain = load_brain(brain_folder) |
|
answer = query_brain(brain, query_question, config_path) |
|
return answer |
|
|
|
def quivr_save(brain_folder: str, save_path: str) -> str: |
|
"""Tool: Saves a Quivr Brain to a specified folder. |
|
|
|
Args: |
|
brain_folder (str): Path to the folder containing the loaded Brain. |
|
save_path (str): Path to the folder where the brain should be saved. |
|
|
|
Returns: |
|
str: Message indicating success and save path. |
|
""" |
|
brain = load_brain(brain_folder) |
|
saved_path = save_brain(brain, save_path) |
|
return f"Brain saved to '{saved_path}'." |
|
|
|
|
|
if __name__ == "__main__": |
|
# Example Usage |
|
file_list_example = ["./my_first_doc.pdf", "./my_second_doc.txt"] # Replace with your file paths |
|
brain_name_example = "my_example_brain" |
|
save_folder_example = "./saved_brain" |
|
config_ingestion_example = "./basic_ingestion_workflow.yaml" # Optional config for ingestion |
|
config_retrieval_example = "./basic_rag_workflow.yaml" # Optional config for retrieval |
|
|
|
# Ensure example files exist (create dummy files for testing if needed) |
|
Path("./my_first_doc.pdf").touch(exist_ok=True) |
|
Path("./my_second_doc.txt").touch(exist_ok=True) |
|
Path("./basic_ingestion_workflow.yaml").touch(exist_ok=True) # Create dummy config if needed |
|
Path("./basic_rag_workflow.yaml").touch(exist_ok=True) # Create dummy config if needed |
|
|
|
|
|
# 1. Create a Brain |
|
print("Creating Brain...") |
|
example_brain = create_brain_from_files(brain_name_example, file_list_example, config_ingestion_example) |
|
print(f"Brain '{example_brain.name}' created.") |
|
|
|
# 2. Query the Brain |
|
example_question = "What is Quivr?" |
|
print(f"\nAsking: '{example_question}'") |
|
example_answer = query_brain(example_brain, example_question, config_retrieval_example) |
|
print(f"Answer: {example_answer}") |
|
|
|
# 3. Save the Brain |
|
print(f"\nSaving Brain to '{save_folder_example}'...") |
|
saved_brain_path = save_brain(example_brain, save_folder_example) |
|
print(f"Brain saved at: {saved_brain_path}") |
|
|
|
# 4. Load the Brain |
|
print(f"\nLoading Brain from '{save_folder_example}'...") |
|
loaded_brain = load_brain(save_folder_example) |
|
print(f"Brain '{loaded_brain.name}' loaded.") |
|
|
|
# 5. Tool-callable function examples |
|
print("\n--- Tool Function Examples ---") |
|
create_tool_result = quivr_create_brain("tool_brain", "./my_first_doc.pdf,./my_second_doc.txt", config_ingestion_example) |
|
print(f"Tool Create Brain Result: {create_tool_result}") |
|
|
|
query_tool_answer = quivr_query(save_folder_example, "Is this a test?", config_retrieval_example) |
|
print(f"Tool Query Answer: {query_tool_answer}") |
|
|
|
save_tool_result = quivr_save(save_folder_example, "./saved_brain_tool") |
|
print(f"Tool Save Brain Result: {save_tool_result}") |