agents.rag.branched_chain

Branched RAG using ChainAgent.

RAG system that branches into multiple specialized retrieval paths based on query type, then merges results for comprehensive answers.

Classes

BranchResult

Result from a retrieval branch.

MergedResult

Final merged result.

QueryClassification

Query classification result.

QueryType

Types of queries for branching.

Functions

create_adaptive_branched_rag(documents[, llm_config])

Create an adaptive branched RAG that selects branches based on query type.

create_branched_rag_chain(documents[, llm_config, name])

Create a branched RAG system using ChainAgent.

create_parallel_branched_rag(documents[, llm_config])

Create a parallel branched RAG that runs all branches simultaneously.

get_branched_rag_io_schema()

Get I/O schema for branched RAG.

Module Contents

class agents.rag.branched_chain.BranchResult(/, **data)

Bases: pydantic.BaseModel

Result from a retrieval branch.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class agents.rag.branched_chain.MergedResult(/, **data)

Bases: pydantic.BaseModel

Final merged result.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class agents.rag.branched_chain.QueryClassification(/, **data)

Bases: pydantic.BaseModel

Query classification result.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class agents.rag.branched_chain.QueryType

Bases: str, enum.Enum

Types of queries for branching.

Initialize self. See help(type(self)) for accurate signature.

agents.rag.branched_chain.create_adaptive_branched_rag(documents, llm_config=None)

Create an adaptive branched RAG that selects branches based on query type.

Parameters:

  • documents (list[langchain_core.documents.Document])

  • llm_config (haive.core.models.llm.base.LLMConfig | None)

Return type:

haive.agents.chain.ChainAgent

agents.rag.branched_chain.create_branched_rag_chain(documents, llm_config=None, name='Branched RAG')

Create a branched RAG system using ChainAgent.

Parameters:

  • documents (list[langchain_core.documents.Document])

  • llm_config (haive.core.models.llm.base.LLMConfig | None)

  • name (str)

Return type:

haive.agents.chain.ChainAgent

agents.rag.branched_chain.create_parallel_branched_rag(documents, llm_config=None)

Create a parallel branched RAG that runs all branches simultaneously.

Parameters:

  • documents (list[langchain_core.documents.Document])

  • llm_config (haive.core.models.llm.base.LLMConfig | None)

Return type:

haive.agents.chain.ChainAgent

agents.rag.branched_chain.get_branched_rag_io_schema()

Get I/O schema for branched RAG.

Return type:

dict[str, list[str]]