autorag package

Subpackages

Submodules

autorag.chunker module

class autorag.chunker.Chunker(raw_df: DataFrame, project_dir: str | None = None)[source]

Bases: object

classmethod from_parquet(parsed_data_path: str, project_dir: str | None = None) → Chunker[source]
start_chunking(yaml_path: str)[source]

autorag.cli module

autorag.dashboard module

autorag.dashboard.find_node_dir(trial_dir: str) → List[str][source]
autorag.dashboard.get_metric_values(node_summary_df: DataFrame) → Dict[source]
autorag.dashboard.make_trial_summary_md(trial_dir)[source]
autorag.dashboard.node_view(node_dir: str)[source]
autorag.dashboard.run(trial_dir: str, port: int = 7690)[source]
autorag.dashboard.yaml_to_markdown(yaml_filepath)[source]

autorag.evaluator module

class autorag.evaluator.Evaluator(qa_data_path: str, corpus_data_path: str, project_dir: str | None = None)[source]

Bases: object

restart_trial(trial_path: str)[source]
start_trial(yaml_path: str, skip_validation: bool = False, full_ingest: bool = True)[source]

Start AutoRAG trial. The trial means one experiment to optimize the RAG pipeline. It consists of ingesting corpus data, running all nodes and modules, evaluating and finding the optimal modules.

Parameters:

  • yaml_path – The config YAML path

  • skip_validation – If True, it skips the validation step. The validation step checks the input config YAML file is well formatted, and there is any problem with the system settings. Default is False.

  • full_ingest – If True, it checks the whole corpus data from corpus.parquet that exists in the Vector DB. If your corpus is huge and don’t want to check the whole vector DB, please set it to False.

Returns:

None

autorag.node_line module

autorag.node_line.make_node_lines(node_line_dict: Dict) → List[Node][source]

This method makes a list of nodes from node line dictionary. :param node_line_dict: Node_line_dict loaded from yaml file, or get from user input. :return: List of Nodes inside this node line.

autorag.node_line.run_node_line(nodes: ~typing.List[~autorag.schema.node.Node], node_line_dir: str, previous_result: ~pandas.core.frame.DataFrame | None = None, progress: ~rich.progress.Progress = None, task_eval: <property object at 0x7f08930f2f70> = None)[source]

Run the whole node line by running each node.

Parameters:

  • nodes – A list of nodes.

  • node_line_dir – This node line’s directory.

  • previous_result – A result of the previous node line. If None, it loads qa data from data/qa.parquet.

  • progress – Rich Progress object.

  • task_eval – Progress task object

Returns:

The final result of the node line.

autorag.parser module

class autorag.parser.Parser(data_path_glob: str, project_dir: str | None = None)[source]

Bases: object

start_parsing(yaml_path: str)[source]

autorag.strategy module

autorag.strategy.avoid_empty_result(return_index: List[int])[source]

Decorator for avoiding empty results from the function. When the func returns an empty result, it will return the origin results. When the func returns a None, it will return the origin results. When the return value is a tuple, it will check all the value or list is empty. If so, it will return the origin results. It keeps parameters at return_index of the function as the origin results.

Parameters:

return_index – The index of the result to be returned when there is no result.

Returns:

The origin results or the results from the function.

autorag.strategy.filter_by_threshold(results, value, threshold, metadatas=None) → Tuple[List, List][source]

Filter results by value’s threshold.

Parameters:

  • results – The result list to be filtered.

  • value – The value list to be filtered. It must have the same length with results.

  • threshold – The threshold value.

  • metadatas – The metadata of each result.

Returns:

Filtered list of results and filtered list of metadatas. Metadatas will be returned even if you did not give input metadatas.

Return type:

Tuple[List, List]

autorag.strategy.measure_speed(func, *args, **kwargs)[source]

Method for measuring execution speed of the function.

autorag.strategy.select_best(results: List[DataFrame], columns: Iterable[str], metadatas: List[Any] | None = None, strategy_name: str = 'mean') → Tuple[DataFrame, Any][source]
autorag.strategy.select_best_average(results: List[DataFrame], columns: Iterable[str], metadatas: List[Any] | None = None) → Tuple[DataFrame, Any][source]

Select the best result by average value among given columns.

Parameters:

  • results – The list of results. Each result must be pd.DataFrame.

  • columns – Column names to be averaged. Standard to select the best result.

  • metadatas – The metadata of each result. It will select one metadata with the best result.

Returns:

The best result and the best metadata. The metadata will be returned even if you did not give input ‘metadatas’ parameter.

Return type:

Tuple[pd.DataFrame, Any]

autorag.strategy.select_best_rr(results: List[DataFrame], columns: Iterable[str], metadatas: List[Any] | None = None) → Tuple[DataFrame, Any][source]
autorag.strategy.select_normalize_mean(results: List[DataFrame], columns: Iterable[str], metadatas: List[Any] | None = None) → Tuple[DataFrame, Any][source]
autorag.strategy.validate_strategy_inputs(results: List[DataFrame], columns: Iterable[str], metadatas: List[Any] | None = None)[source]

autorag.support module

autorag.support.dynamically_find_function(key: str, target_dict: Dict) → Callable[source]
autorag.support.get_support_modules(module_name: str) → Callable[source]
autorag.support.get_support_nodes(node_name: str) → Callable[source]

autorag.validator module

class autorag.validator.Validator(qa_data_path: str, corpus_data_path: str)[source]

Bases: object

validate(yaml_path: str, qa_cnt: int = 5, random_state: int = 42)[source]

autorag.web module

autorag.web.chat_box(runner: Runner)[source]
autorag.web.get_runner(yaml_path: str | None, project_dir: str | None, trial_path: str | None)[source]
autorag.web.set_initial_state()[source]
autorag.web.set_page_config()[source]
autorag.web.set_page_header()[source]

Module contents

class autorag.AutoRAGBedrock(model: str, temperature: float | None = 0.1, max_tokens: int | None = 512, context_size: int | None = None, profile_name: str | None = None, aws_access_key_id: str | None = None, aws_secret_access_key: str | None = None, aws_session_token: str | None = None, region_name: str | None = None, botocore_session: Any | None = None, client: Any | None = None, timeout: float | None = 60.0, max_retries: int | None = 10, botocore_config: Any | None = None, additional_kwargs: Dict[str, Any] | None = None, callback_manager: CallbackManager | None = None, system_prompt: str | None = None, messages_to_prompt: Callable[[Sequence[ChatMessage]], str] | None = None, completion_to_prompt: Callable[[str], str] | None = None, pydantic_program_mode: PydanticProgramMode = PydanticProgramMode.DEFAULT, output_parser: BaseOutputParser | None = None, *, query_wrapper_prompt: BasePromptTemplate | None = None)[source]

Bases: Bedrock

async acomplete(prompt: str, formatted: bool = False, **kwargs: Any) → CompletionResponse[source]

Async completion endpoint for LLM.

If the LLM is a chat model, the prompt is transformed into a single user message.

Args:
prompt (str):

Prompt to send to the LLM.

formatted (bool, optional):

Whether the prompt is already formatted for the LLM, by default False.

kwargs (Any):

Additional keyword arguments to pass to the LLM.

Returns:

CompletionResponse: Completion response from the LLM.

Examples:

`python response = await llm.acomplete("your prompt") print(response.text) `

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'additional_kwargs': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='Additional kwargs for the bedrock invokeModel request.'), 'aws_access_key_id': FieldInfo(annotation=Union[str, NoneType], required=True, description='AWS Access Key ID to use', exclude=True), 'aws_secret_access_key': FieldInfo(annotation=Union[str, NoneType], required=True, description='AWS Secret Access Key to use', exclude=True), 'aws_session_token': FieldInfo(annotation=Union[str, NoneType], required=True, description='AWS Session Token to use', exclude=True), 'botocore_config': FieldInfo(annotation=Union[Any, NoneType], required=True, description='Custom configuration object to use instead of the default generated one.', exclude=True), 'botocore_session': FieldInfo(annotation=Union[Any, NoneType], required=True, description='Use this Botocore session instead of creating a new default one.', exclude=True), 'callback_manager': FieldInfo(annotation=CallbackManager, required=False, default_factory=<lambda>, exclude=True), 'completion_to_prompt': FieldInfo(annotation=Union[CompletionToPromptType, NoneType], required=False, default=None, description='Function to convert a completion to an LLM prompt.', exclude=True, metadata=[WithJsonSchema(json_schema={'type': 'string'}, mode=None)]), 'context_size': FieldInfo(annotation=int, required=False, default='The maximum number of tokens available for input.'), 'max_retries': FieldInfo(annotation=int, required=False, default=10, description='The maximum number of API retries.', metadata=[Gt(gt=0)]), 'max_tokens': FieldInfo(annotation=int, required=True, description='The maximum number of tokens to generate.'), 'messages_to_prompt': FieldInfo(annotation=Union[MessagesToPromptType, NoneType], required=False, default=None, description='Function to convert a list of messages to an LLM prompt.', exclude=True, metadata=[WithJsonSchema(json_schema={'type': 'string'}, mode=None)]), 'model': FieldInfo(annotation=str, required=True, description='The modelId of the Bedrock model to use.'), 'output_parser': FieldInfo(annotation=Union[BaseOutputParser, NoneType], required=False, default=None, description='Output parser to parse, validate, and correct errors programmatically.', exclude=True), 'profile_name': FieldInfo(annotation=Union[str, NoneType], required=True, description='The name of aws profile to use. If not given, then the default profile is used.'), 'pydantic_program_mode': FieldInfo(annotation=PydanticProgramMode, required=False, default=<PydanticProgramMode.DEFAULT: 'default'>), 'query_wrapper_prompt': FieldInfo(annotation=Union[BasePromptTemplate, NoneType], required=False, default=None, description='Query wrapper prompt for LLM calls.', exclude=True), 'region_name': FieldInfo(annotation=Union[str, NoneType], required=True, description='AWS region name to use. Uses region configured in AWS CLI if not passed', exclude=True), 'system_prompt': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='System prompt for LLM calls.'), 'temperature': FieldInfo(annotation=float, required=True, description='The temperature to use for sampling.'), 'timeout': FieldInfo(annotation=float, required=False, default=60.0, description='The timeout for the Bedrock API request in seconds. It will be used for both connect and read timeouts.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

model_post_init(context: Any, /) → None

We need to both initialize private attributes and call the user-defined model_post_init method.

class autorag.LazyInit(factory, *args, **kwargs)[source]

Bases: object

class autorag.MockEmbeddingRandom(embed_dim: int, *, model_name: str = 'unknown', embed_batch_size: Annotated[int, Gt(gt=0), Le(le=2048)] = 10, callback_manager: CallbackManager = None, num_workers: int | None = None)[source]

Bases: MockEmbedding

Mock embedding with random vectors.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'protected_namespaces': ('pydantic_model_',)}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'callback_manager': FieldInfo(annotation=CallbackManager, required=False, default_factory=<lambda>, exclude=True), 'embed_batch_size': FieldInfo(annotation=int, required=False, default=10, description='The batch size for embedding calls.', metadata=[Gt(gt=0), Le(le=2048)]), 'embed_dim': FieldInfo(annotation=int, required=True), 'model_name': FieldInfo(annotation=str, required=False, default='unknown', description='The name of the embedding model.'), 'num_workers': FieldInfo(annotation=Union[int, NoneType], required=False, default=None, description='The number of workers to use for async embedding calls.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

autorag.handle_exception(exc_type, exc_value, exc_traceback)[source]
autorag.random() → x in the interval [0, 1).